Cyrene_v2/internal/services/backthink/store.go

package backthink

import (
	"context"
	"database/sql"
	"fmt"
	"log/slog"
	"strings"
	"time"

	"git.yeij.top/AskaEth/Cyrene_v2/internal/logger"
)

// defaultListLimit 是调用方传入非正 limit 时，List 返回的
// 最近记录条数。
const defaultListLimit = 50

// 思考记录存储的 DDL。at/created_at 遵循项目时间戳
// 约定（带时区的 RFC3339）；trigger 与 conclusion 是面板渲染的
// 自由文本字段；push_* 列携带分解后的 PushResult，
// 使 List 不必解析 blob；urgency 是该轮**实际生效**的紧急度档位
// （空串 = 未知，见 Record.Urgency）。主键是项目标准的自增 id。
const createRecordsSQL = `
CREATE TABLE IF NOT EXISTS backthink_records (
    id           INTEGER PRIMARY KEY AUTOINCREMENT,
    trigger      TEXT NOT NULL,
    at           TEXT NOT NULL,
    conclusion   TEXT NOT NULL,
    should_push  INTEGER NOT NULL,
    push_pushed  INTEGER NOT NULL DEFAULT 0,
    push_route   TEXT NOT NULL,
    push_detail  TEXT NOT NULL DEFAULT '',
    push_error   TEXT NOT NULL DEFAULT '',
    urgency      TEXT NOT NULL DEFAULT '',
    created_at   TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ','now'))
)`

// alterRecordsAddUrgencySQL 给**既有**库补上 urgency 列：CREATE TABLE IF NOT EXISTS
// 对已经存在的表不加列，缺了它每一条 INSERT 都会以「no such column」失败
// （思考存档整条链路断掉）。它属于本模块自建的表的演进，因此**不开迁移号**、
// 也不动 internal/db 的迁移记账（那是 schema 所有者那边的领域，见包注释与 NewStore）。
const alterRecordsAddUrgencySQL = `
ALTER TABLE backthink_records ADD COLUMN urgency TEXT NOT NULL DEFAULT ''`

const createRecordsIndexSQL = `
CREATE INDEX IF NOT EXISTS idx_backthink_records_at ON backthink_records(at)`

// createRecordsTriggerIndexSQL 支撑按触发源筛选的那条查询
// （`WHERE trigger = ? ORDER BY id DESC`）。索引条目是 (trigger, rowid)，而主键
// `id` 就是 rowid ⇒ SQLite 能直接用一次索引范围扫描（反向）同时满足筛选与排序。
// 它是**本片新增读路径**的配套：没有它，筛选就是全表扫描 + 排序（表按心跳每 15 分钟
// 增长一行，一年就是三万余行）。
const createRecordsTriggerIndexSQL = `
CREATE INDEX IF NOT EXISTS idx_backthink_records_trigger ON backthink_records(trigger)`

const insertRecordSQL = `
INSERT INTO backthink_records
    (trigger, at, conclusion, should_push, push_pushed, push_route, push_detail, push_error, urgency)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)`

const selectRecordsSQL = `
SELECT id, trigger, at, conclusion, should_push, push_pushed, push_route, push_detail, push_error, urgency
FROM backthink_records
ORDER BY id DESC
LIMIT ?`

// selectRecordsByTriggerSQL 与 selectRecordsSQL 逐字同形，只多一个按
// `Record.Trigger` 的**服务端**筛选（参数化，不拼字符串）。
//
// 为什么下推到 SQL 而不是「取一页再在内存里筛」：那样 `limit` 就变成「这一页里
// 恰好匹配的条数」，面板会看到「自我唤醒 0 条」而库里其实有 —— 一个静默的假空值
// （design/12 §N C.6 的服务端筛选要的正是「筛不到 ≠ 没发生」）。
const selectRecordsByTriggerSQL = `
SELECT id, trigger, at, conclusion, should_push, push_pushed, push_route, push_detail, push_error, urgency
FROM backthink_records
WHERE trigger = ?
ORDER BY id DESC
LIMIT ?`

// recordsTableName 是本模块自建表的物理名（DDL 与判列共用一处）。
const recordsTableName = "backthink_records"

// tableInfoSQL 列举一张表的列。pragma_table_info 是 SQLite 3.16+ 的表值函数，
// 因此表名也能走占位符，不必把名字拼进 SQL。
const tableInfoSQL = `SELECT name FROM pragma_table_info(?)`

// Record 是一轮已持久化的思考（design/12 §4），由 Store.List
// 与服务的 List 返回。Trigger 是来源子系统标签，At 是决策
// 时间戳，Conclusion 是运行时的输出，ShouldPush 是打扰决策，
// PushResult 是推送如何（或是否）发出。
type Record struct {
	ID         int64
	Trigger    string
	At         time.Time
	Conclusion string
	ShouldPush bool
	PushResult PushResult
	// Urgency 是**该轮实际生效**的紧急度档位，用 internal/urgency 的词汇字符串
	// （routine / warning / critical）：面板因此能区分 high 与 critical，而此前
	// 推送成功行的 PushResult.Detail 已被投递说明覆盖、档位在存档里丢失。
	//
	// 空串**如实表示「未知」**：本列出现之前落下的旧行，以及从未走到档位判定点的
	// 轮次（思考本身失败）都读作未知 —— 绝不反推、绝不编成 routine（与 P3+P4 的
	// fail-safe 同口径：宁可不显示，也不谎报）。
	//
	// JSON 键名刻意是 `urgency`：WS 实时帧发出去的正是本结构体，前端
	// （wsParse.ts）已按 `r.urgency` 防御性读取，带上它实时行才会显示档位，
	// 而不必等重开页面由历史拉取补齐。omitempty 让未知保持「键缺席」——
	// 与「例常」可区分。
	Urgency string `json:"urgency,omitempty"`
}

// Store 把思考记录持久化到 SQLite。它在
// NewStore 中建表与索引，因此本模块自包含（无需迁移记账），
// 与 channel/session 自建表的做法一致。它凭借底层的 *sql.DB
// 而并发安全。
type Store struct {
	db  *sql.DB
	log *slog.Logger
}

// NewStore 构建 Store，确保 backthink_records 表（含既有库缺的 urgency 列，
// 见 ensureRecordsUrgencyColumn）与索引存在。
// db 为 nil 时返回 ErrNilDB。ctx 用于 DDL。
func NewStore(ctx context.Context, db *sql.DB, log *slog.Logger) (*Store, error) {
	if db == nil {
		return nil, ErrNilDB
	}
	log = logger.WithSource(baseLog(log), "services/backthink")
	if _, err := db.ExecContext(ctx, createRecordsSQL); err != nil {
		return nil, fmt.Errorf("backthink: ensure backthink_records: %w", err)
	}
	if err := ensureRecordsUrgencyColumn(ctx, db, log); err != nil {
		return nil, err
	}
	if _, err := db.ExecContext(ctx, createRecordsIndexSQL); err != nil {
		return nil, fmt.Errorf("backthink: ensure backthink_records index: %w", err)
	}
	if _, err := db.ExecContext(ctx, createRecordsTriggerIndexSQL); err != nil {
		return nil, fmt.Errorf("backthink: ensure backthink_records trigger index: %w", err)
	}
	return &Store{db: db, log: log}, nil
}

// ensureRecordsUrgencyColumn **幂等**地保证 backthink_records 带 urgency 列。
// 既有部署的库是先建好的，CREATE TABLE IF NOT EXISTS 那句不会给它加列，所以这里
// 先按 PRAGMA 判存在、缺了才 ALTER。
//
// 判到「缺列」与真的执行 ALTER 之间可能有另一个进程同时启动（同库双开），那次
// ALTER 会以 duplicate column 失败 —— 那是同一个目标的另一次成功，容忍它；
// **其余任何错误都照实上报**（吞掉会让「列其实没加上」静默过关，随后每条 INSERT
// 都失败，而启动日志里什么也看不到）。
func ensureRecordsUrgencyColumn(ctx context.Context, db *sql.DB, log *slog.Logger) error {
	exists, err := columnExists(ctx, db, recordsTableName, "urgency")
	if err != nil {
		return err
	}
	if exists {
		return nil
	}
	if _, err := db.ExecContext(ctx, alterRecordsAddUrgencySQL); err != nil {
		if isDuplicateColumnErr(err) {
			return nil
		}
		return fmt.Errorf("backthink: add backthink_records.urgency: %w", err)
	}
	if log != nil {
		log.Info("backthink records table upgraded", "column", "urgency", "reason", "existing database predates the urgency column")
	}
	return nil
}

// columnExists 报告 table 是否已有 column。
func columnExists(ctx context.Context, db *sql.DB, table, column string) (bool, error) {
	rows, err := db.QueryContext(ctx, tableInfoSQL, table)
	if err != nil {
		return false, fmt.Errorf("backthink: inspect %s columns: %w", table, err)
	}
	defer rows.Close()
	for rows.Next() {
		var name string
		if err := rows.Scan(&name); err != nil {
			return false, fmt.Errorf("backthink: scan %s column: %w", table, err)
		}
		if name == column {
			return true, nil
		}
	}
	if err := rows.Err(); err != nil {
		return false, fmt.Errorf("backthink: iterate %s columns: %w", table, err)
	}
	return false, nil
}

// isDuplicateColumnErr 报告 err 是否为「这一列已经存在」。SQLite 的措辞是
// "duplicate column name: xxx"，驱动原样带回该串，故按子串判定；判错方向的代价
// 不对称（把真错误当重复会掩盖缺列），因此这里只认这一种措辞。
func isDuplicateColumnErr(err error) bool {
	return err != nil && strings.Contains(strings.ToLower(err.Error()), "duplicate column name")
}

// Append 持久化 r 并返回新记录 id。r.Urgency 原样落库（空串 = 未知，
// 见 Record.Urgency）——存储层不替调用方猜一个档位。
func (s *Store) Append(ctx context.Context, r Record) (int64, error) {
	res, err := s.db.ExecContext(ctx, insertRecordSQL,
		r.Trigger,
		r.At.UTC().Format(time.RFC3339Nano),
		r.Conclusion,
		boolInt(r.ShouldPush),
		boolInt(r.PushResult.Pushed),
		string(r.PushResult.Route),
		r.PushResult.Detail,
		r.PushResult.Err,
		r.Urgency,
	)
	if err != nil {
		return 0, fmt.Errorf("backthink: insert record: %w", err)
	}
	id, err := res.LastInsertId()
	if err != nil {
		return 0, fmt.Errorf("backthink: record id: %w", err)
	}
	if s.log != nil {
		s.log.Debug("backthink record appended", "id", id, "trigger", r.Trigger,
			"should_push", r.ShouldPush, "route", r.PushResult.Route, "urgency", r.Urgency)
	}
	return id, nil
}

// List 返回最近的记录，最新的在前。非正的 limit 选用
// defaultListLimit。urgency 列的空串如实读成「未知」（见
// Record.Urgency），不做任何补齐或推断。
func (s *Store) List(ctx context.Context, limit int) ([]Record, error) {
	limit = normListLimit(limit)
	if err := ctx.Err(); err != nil {
		return nil, err
	}
	rows, err := s.db.QueryContext(ctx, selectRecordsSQL, limit)
	if err != nil {
		return nil, fmt.Errorf("backthink: list records: %w", err)
	}
	defer rows.Close()
	return scanRecords(rows, limit)
}

// ListByTrigger 返回**某一触发源**的最近记录，最新的在前：语义与 List 逐字相同，
// 只多一个参数化的 `trigger = ?`（design/12 §N C.6 的服务端筛选）。
//
// ⚠️ 它按**原始落库值**精确匹配（不模糊、不前缀）：`Record.Trigger` 对事件轮是事件
// 主题字符串，对内置轮是 `heartbeat` / `startup` / `self_wake` —— 合法取值由
// `IsTriggerValue`（trigger_domain.go 的唯一真源）判定，本层不做二次裁剪。
func (s *Store) ListByTrigger(ctx context.Context, trigger string, limit int) ([]Record, error) {
	limit = normListLimit(limit)
	if err := ctx.Err(); err != nil {
		return nil, err
	}
	rows, err := s.db.QueryContext(ctx, selectRecordsByTriggerSQL, trigger, limit)
	if err != nil {
		return nil, fmt.Errorf("backthink: list records by trigger: %w", err)
	}
	defer rows.Close()
	return scanRecords(rows, limit)
}

// normListLimit 把非正的 limit 归一为 defaultListLimit（两条读路径同一口径）。
func normListLimit(limit int) int {
	if limit <= 0 {
		return defaultListLimit
	}
	return limit
}

// scanRecords 是两条读路径共用的行解码（抽出来是为了让「列序 / 空串语义」只有
// 一处真源：再加第三条查询时不会悄悄错位）。
func scanRecords(rows *sql.Rows, capacity int) ([]Record, error) {
	records := make([]Record, 0, capacity)
	for rows.Next() {
		var (
			r          Record
			at         string
			route      string
			pushed     int
			shouldPush int
		)
		if err := rows.Scan(
			&r.ID,
			&r.Trigger,
			&at,
			&r.Conclusion,
			&shouldPush,
			&pushed,
			&route,
			&r.PushResult.Detail,
			&r.PushResult.Err,
			&r.Urgency,
		); err != nil {
			return nil, fmt.Errorf("backthink: scan record: %w", err)
		}
		r.ShouldPush = intBool(shouldPush)
		r.PushResult.Pushed = intBool(pushed)
		r.PushResult.Route = Route(route)
		r.At = parseTime(at)
		records = append(records, r)
	}
	if err := rows.Err(); err != nil {
		return nil, fmt.Errorf("backthink: iterate records: %w", err)
	}
	return records, nil
}

// boolInt 把 bool 映射为 SQLite 的 INTEGER 1/0 表示。
func boolInt(b bool) int {
	if b {
		return 1
	}
	return 0
}

// intBool 把 SQLite 的 INTEGER 映射为 bool。
func intBool(n int) bool { return n != 0 }

// parseTime 解析存储的 RFC3339（带纳秒/时区）时间戳，失败时回退到
// 零值时间，因此格式错乱的行绝不会破坏列表查询。
func parseTime(s string) time.Time {
	for _, layout := range []string{time.RFC3339Nano, time.RFC3339} {
		if t, err := time.Parse(layout, s); err == nil {
			return t
		}
	}
	return time.Time{}
}