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{} }