Cyrene_v2/internal/runtime/history/log_reads.go

package history

import (
	"context"
	"fmt"
)

// 本文件归属 chat_messages 的**日志与窗口读取**：适配器日志页的三个粒度
// （单适配器 / 单平台 / 全适配器）与 WS 回放/上下文尾部用的 Recent。它们与
// store.go 的 List/Search 分开按职责分文件，并共用 store.go 的 messageColumns
// 列清单与 scanMessage，因此「一列都不许漏」的口径仍然只有一处。

// ListPlatformLog 按最新优先返回 platform 跨所有非主（隔离）会话的至多 limit
// 条消息。主会话消息被排除，因此只有该平台自己的消息流（群聊 + 陌生人私聊）
// 浮现到适配器日志页（design/16 §4.6 §4）。before 是毫秒游标；非零时只返回
// 严格更旧的消息。返回切片非 nil。
func (s *Store) ListPlatformLog(ctx context.Context, platform string, before int64, limit int) ([]Message, error) {
	limit = clampLimit(limit)
	query := `SELECT ` + messageColumns + `
		FROM chat_messages WHERE platform = ? AND session_id != ?`
	args := []any{platform, MainSessionID}
	if before > 0 {
		query += ` AND created_at < ?`
		args = append(args, before)
	}
	query += ` ORDER BY created_at DESC LIMIT ?`
	args = append(args, limit)

	rows, err := s.db.QueryContext(ctx, query, args...)
	if err != nil {
		return nil, fmt.Errorf("history: platform log %s: %w", platform, err)
	}
	defer func() { _ = rows.Close() }()

	msgs := make([]Message, 0)
	for rows.Next() {
		m, err := scanMessage(rows)
		if err != nil {
			return nil, fmt.Errorf("history: scan platform log %s: %w", platform, err)
		}
		msgs = append(msgs, m)
	}
	if err := rows.Err(); err != nil {
		return nil, fmt.Errorf("history: iterate platform log %s: %w", platform, err)
	}
	return msgs, nil
}

// ListAdapterLog 按最新优先返回单个 IM 适配器（平台 + 账号注册名，例如
// "kook:3bef..."）的至多 limit 条消息，供按适配器的日志页使用。它不排除主 Web
// 会话：主人的 IM 私聊住在主会话里但仍携带 adapter_id，因此按适配器的日志包含
// 主人镜像。adapterID 为空时返回空（非 nil）切片。before 是毫秒游标；非零时只
// 返回严格更旧的消息。返回切片非 nil。
func (s *Store) ListAdapterLog(ctx context.Context, adapterID string, before int64, limit int) ([]Message, error) {
	limit = clampLimit(limit)
	if adapterID == "" {
		return make([]Message, 0), nil
	}
	query := `SELECT ` + messageColumns + `
		FROM chat_messages WHERE adapter_id = ?`
	args := []any{adapterID}
	if before > 0 {
		query += ` AND created_at < ?`
		args = append(args, before)
	}
	query += ` ORDER BY created_at DESC LIMIT ?`
	args = append(args, limit)

	rows, err := s.db.QueryContext(ctx, query, args...)
	if err != nil {
		return nil, fmt.Errorf("history: adapter log %s: %w", adapterID, err)
	}
	defer func() { _ = rows.Close() }()

	msgs := make([]Message, 0)
	for rows.Next() {
		m, err := scanMessage(rows)
		if err != nil {
			return nil, fmt.Errorf("history: scan adapter log %s: %w", adapterID, err)
		}
		msgs = append(msgs, m)
	}
	if err := rows.Err(); err != nil {
		return nil, fmt.Errorf("history: iterate adapter log %s: %w", adapterID, err)
	}
	return msgs, nil
}

// ListAllLog 按最新优先返回跨所有 IM 适配器来源（adapter_id 非空）的至多 limit
// 条消息，供适配器总览页使用。before 是毫秒游标；非零时只返回严格更旧的消息。
// 返回切片非 nil。
func (s *Store) ListAllLog(ctx context.Context, before int64, limit int) ([]Message, error) {
	limit = clampLimit(limit)
	query := `SELECT ` + messageColumns + `
		FROM chat_messages WHERE adapter_id != ''`
	var args []any
	if before > 0 {
		query += ` AND created_at < ?`
		args = append(args, before)
	}
	query += ` ORDER BY created_at DESC LIMIT ?`
	args = append(args, limit)

	rows, err := s.db.QueryContext(ctx, query, args...)
	if err != nil {
		return nil, fmt.Errorf("history: all adapter log: %w", err)
	}
	defer func() { _ = rows.Close() }()

	msgs := make([]Message, 0)
	for rows.Next() {
		m, err := scanMessage(rows)
		if err != nil {
			return nil, fmt.Errorf("history: scan all adapter log: %w", err)
		}
		msgs = append(msgs, m)
	}
	if err := rows.Err(); err != nil {
		return nil, fmt.Errorf("history: iterate all adapter log: %w", err)
	}
	return msgs, nil
}

// Recent 按最旧优先返回 sessionID 最近的 n 条消息（最靠尾部的那段窗口），供
// WS 回放 / 上下文尾部使用。返回切片非 nil。
//
// 并列打破（裁决 13.4.5 的上下文「最新性/一致性」）：同一毫秒写入的多条消息
// 在 created_at 上并列，而 Windows 上 time.Now() 的步长本就约 0.44ms —— 用户
// 连发多条（design/15 §2.1 的忙碌续接场景）时并列是常态。`rowid DESC` 让 LIMIT
// 取到的是**最后插入**的那几条（否则窗口会留下更旧的、丢掉刚发出的），而反转后
// 同毫秒内即回到插入顺序，使续接窗口的时序与真实对话一致。该口径与 ListAfter
// 的「rowid 打破 created_at 的并列」相同。
func (s *Store) Recent(ctx context.Context, sessionID string, n int) ([]Message, error) {
	if n < 1 {
		n = defaultPageLimit
	}
	rows, err := s.db.QueryContext(ctx, `SELECT `+messageColumns+`
		FROM chat_messages WHERE session_id = ? ORDER BY created_at DESC, rowid DESC LIMIT ?`, sessionID, n)
	if err != nil {
		return nil, fmt.Errorf("history: recent %s: %w", sessionID, err)
	}
	defer func() { _ = rows.Close() }()

	// Recent 让最新的排在最后，使窗口处于时间顺序；先按查询顺序构建再原地反转。
	msgs := make([]Message, 0)
	for rows.Next() {
		m, err := scanMessage(rows)
		if err != nil {
			return nil, fmt.Errorf("history: scan recent %s: %w", sessionID, err)
		}
		msgs = append(msgs, m)
	}
	if err := rows.Err(); err != nil {
		return nil, fmt.Errorf("history: iterate recent %s: %w", sessionID, err)
	}
	for i, j := 0, len(msgs)-1; i < j; i, j = i+1, j-1 {
		msgs[i], msgs[j] = msgs[j], msgs[i]
	}
	return msgs, nil
}