Files
Cyrene_v2/internal/runtime/history/store.go
T
Aska 2e044d1d19
CI / Backend (build / vet / test / cross-compile) (push) Has been cancelled
CI / Frontend (tsc + vite build) (push) Has been cancelled
feat(channels): 管理面板消息日志——chat_messages 加 adapter_id + 配置级/平台/总览三粒度端点
后端:
- 迁移 0007:chat_messages 加 adapter_id 列 + 索引
- history.Message 加 AdapterID;persistToMain/persistIsolated 传适配器名(a.Name())
- channelbridge: 4 个持久化点 + registerSilenced 传 a;light.go registerSilenced 签名加 a
- Store 新增 ListAdapterLog(?adapter=, 不排除主会话)/ListAllLog(adapter_id!='', 总览);ListPlatformLog 保留
- 新端点 GET /api/v1/admin/channels/messages:?adapter= 配置级 / ?platform= 平台 / 无参=总览,historyMessage 加 adapterId
- 测试:ListAdapterLog/ListAllLog 过滤用例 + 端点粒度用例 + 桥接持久化断言

前端:渠道管理每配置(账号)「消息日志」按钮 + 页顶「适配器总览日志」入口 + ChannelLogPanel 弹窗(总览/配置级),Miuix 风格,build 绿。
2026-09-07 17:25:23 +08:00

590 lines
24 KiB
Go

// Package history implements the persistent chat-history session model described in
// DevDocs/design/16-会话持久化与主次级会话.md.
//
// It is the storage layer, not the orchestration/context layer: it persists the raw
// message stream (chat_messages) and the persistent session records (chat_sessions)
// so the Web frontend can request + display history, and secondary (sub) sessions can
// be created, listed and deleted. It intentionally does not touch memory (design/16
// §3.2: memory is persona-scoped and lives in runtime/memory), does not build context
// (runtime/loop) and does not schedule turns (gateway/session_orchestrator).
//
// Session model (design/16 §2.1): there is exactly one fixed main session (id "web") and
// any number of sub sessions (id "web:<uuid>"). Main + sub share the same persona and
// therefore the same memory; this package only models sessions + messages.
package history
import (
"context"
"crypto/rand"
"database/sql"
"encoding/hex"
"errors"
"fmt"
"io"
"log/slog"
"time"
"git.yeij.top/AskaEth/Cyrene_v2/internal/logger"
)
// Session roles and types. These are the values persisted in the role / type columns.
const (
// RoleUser, RoleAssistant, RoleSystem classify a message's author.
RoleUser = "user"
RoleAssistant = "assistant"
RoleSystem = "system"
// SessionMain and SessionSub classify a persistent session.
SessionMain = "main"
SessionSub = "sub"
// PlatformWeb is the source-platform tag for messages that originate on the Web
// main session (design/16 §4.4). It is both the column default and the value
// assigned when a caller leaves Message.Platform empty.
PlatformWeb = "web"
)
// MainSessionID is the fixed id of the one main session (design/16 §2.1). Sub session
// ids are MainSessionID + ":" + uuid.
const MainSessionID = "web"
// defaultPageLimit is the page size used when the caller does not pass one.
const defaultPageLimit = 50
// maxPageLimit bounds a client-supplied limit so a single request cannot drain memory.
const maxPageLimit = 200
// Sentinels distinguishing history package failures. Callers wrap them with %w so
// errors.Is works through the chain.
var (
// ErrNilDB is returned when NewStore receives a nil *sql.DB.
ErrNilDB = errors.New("history: nil db")
// ErrImmutableMain is returned when a caller tries to delete the fixed main session.
ErrImmutableMain = errors.New("history: the main session cannot be deleted")
)
// Message is one persisted chat message. CreatedAt is a Unix millisecond timestamp and
// doubles as the pagination cursor (design/16 §3.1): pages are ordered newest-first and
// "before" is the created_at of the oldest message on the previous page.
type Message struct {
ID string
SessionID string
Role string
Content string
Mode string
Mood string
Image string
Thinking string
// Platform tags the source platform of a message (design/16 §4.4). The Web main
// session is shared across platforms: messages that originate on the Web carry
// "web", while IM-owner-private messages carry the channel platform (e.g.
// "telegram", "qq-obv11", "weixin-ilink"). An empty Platform defaults to "web" so
// legacy/explicitly-Web writes stay on the Web lane without a separate assignment.
Platform string
// DeviceID is a short, client-produced identifier for the concrete device/app
// that sent the message (design/16 §4.5) — e.g. "web:chrome:a1b2c3" for a
// browser, "weixin-ilink:conf0:wx_uid_123" for IM. It complements Platform: two
// clients on the same platform share Platform but differ in DeviceID.
DeviceID string
// SenderName is the human-readable display name of the message's sender
// (design/16 §4.6). It is set for third-party messages (group chats and
// stranger private chats) so the agent and the IM adapter log page can label
// who is speaking; it is intentionally blank for the owner's private chat on
// the main session (the owner is never labeled). It is display metadata only —
// identity is never decided from this field.
SenderName string
// AdapterID is the adapter (platform + account) registration name of the IM
// source that produced the message, e.g. "kook:3bef..." for a multi-account
// channels.json registration or a bare platform name (e.g. "kook") when the
// account comes from env. IM adapter-originated messages carry it; the Web main
// session leaves it empty, so it is always '' for Web messages.
AdapterID string
CreatedAt int64
}
// SessionMeta is the wire-facing metadata of one persistent session.
type SessionMeta struct {
ID string
Type string // "main" | "sub"
Title string
CreatedAt int64
}
// HistoryStore abstracts the chat-history store so callers (the gateway) and tests can
// hold an interface rather than a concrete *Store.
type HistoryStore interface {
// Append persists msg, assigning an id and timestamp when the caller omitted them.
Append(ctx context.Context, msg Message) error
// AppendFull persists msg like Append but returns the complete stored message
// (with the assigned id/timestamp) so a caller can broadcast it for cross-end sync.
AppendFull(ctx context.Context, msg Message) (Message, error)
// List returns up to limit messages for sessionID, newest first. before is a
// millisecond timestamp cursor: when nonzero, only messages older than before are
// returned (inclusive-free, matching the "before" bound on the frontend). platform
// is an optional source-platform filter (design/16 §4.4): when non-empty only
// messages carrying that platform are returned; when empty all platforms are
// returned (the Web client requests the full main session, an IM adapter asks for
// its own platform only).
List(ctx context.Context, sessionID string, before int64, limit int, platform string) ([]Message, error)
// Recent returns the most recent n messages for sessionID in oldest-first order,
// for WS replay / context tail usage.
Recent(ctx context.Context, sessionID string, n int) ([]Message, error)
// ListPlatformLog returns up to limit messages for an IM platform across every
// non-main (isolated) session, newest first, for the adapter log page (design/16
// §4.6 §4). It excludes the fixed main web session, so the platform's non-owner
// message stream (groups + stranger private chats) is surfaced without the
// owner-private mirror. before is a millisecond cursor; 0 means newest page.
ListPlatformLog(ctx context.Context, platform string, before int64, limit int) ([]Message, error)
// ListAdapterLog returns up to limit messages for a single IM adapter (platform +
// account registration name, e.g. "kook:3bef..."), newest first, for the
// per-adapter log page. Unlike ListPlatformLog it does NOT exclude the main web
// session: an owner-private IM chat lives in the main session but still carries
// the adapter_id, so a per-adapter log includes the owner mirror. before is a
// millisecond cursor; 0 means the newest page. An empty adapterID returns an
// empty slice.
ListAdapterLog(ctx context.Context, adapterID string, before int64, limit int) ([]Message, error)
// ListAllLog returns up to limit messages across every IM adapter source
// (adapter_id != ''), newest first, for the adapter overview page. before is a
// millisecond cursor; 0 means the newest page.
ListAllLog(ctx context.Context, before int64, limit int) ([]Message, error)
// DeleteSession removes a sub session and all its messages. The main session is
// immutable and returns ErrImmutableMain.
DeleteSession(ctx context.Context, sessionID string) error
// NewSubSession creates a new sub session and returns its id (MainSessionID:uuid).
NewSubSession(ctx context.Context, ownerID string) (string, error)
// ListSessions returns the main session first then every sub session, oldest first.
ListSessions(ctx context.Context) ([]SessionMeta, error)
}
// Store is the SQLite-backed chat-history store. It owns the chat_sessions and
// chat_messages tables. Concurrency is delegated to *sql.DB (safe for concurrent use),
// so Store holds no further lock. Every query is parameterized and reads list columns
// explicitly (no SELECT *); reads take a context.
type Store struct {
db *sql.DB
log *slog.Logger
}
// rowScanner is satisfied by both *sql.Row and *sql.Rows, so a single scanMessage
// helper serves List and Recent.
type rowScanner interface {
Scan(dest ...any) error
}
// NewStore builds a Store on db, ensuring the chat_sessions/chat_messages schema exists
// and that the fixed main session row is present. The DDL mirrors the 0003 migration;
// it is idempotent so the store is self-contained and usable from any *sql.DB, matching
// the memory and session packages. db must be live; nil returns ErrNilDB. log may be nil.
func NewStore(ctx context.Context, db *sql.DB, log *slog.Logger) (*Store, error) {
if db == nil {
return nil, ErrNilDB
}
log = logger.WithSource(discardLogger(log), "history")
for _, ddl := range []string{
createSessionsTable,
createMessagesTable,
createMessagesIndex,
} {
if _, err := db.ExecContext(ctx, ddl); err != nil {
return nil, fmt.Errorf("history: apply schema: %w", err)
}
}
s := &Store{db: db, log: log}
if err := s.ensureMainSession(ctx); err != nil {
return nil, err
}
return s, nil
}
// Append persists msg, assigning a fresh id and a now-ms CreatedAt when the caller left
// either empty/zero. It returns the error wrapped with the operation name. Callers that
// need the generated id/timestamp back should use AppendFull instead.
func (s *Store) Append(ctx context.Context, msg Message) error {
_, err := s.insert(ctx, msg)
return err
}
// AppendFull persists msg like Append but returns the complete stored message — the id
// and CreatedAt the store assigned when the caller left them empty/zero — so a caller can
// broadcast it for cross-end realtime sync (design/16 §4.2) without a second read.
func (s *Store) AppendFull(ctx context.Context, msg Message) (Message, error) {
return s.insert(ctx, msg)
}
// insert is the shared Append/AppendFull body. It assigns a fresh id and a now-ms
// CreatedAt when the caller left either empty/zero, then inserts the row and returns the
// resulting message.
func (s *Store) insert(ctx context.Context, msg Message) (Message, error) {
if msg.ID == "" {
id, err := newID()
if err != nil {
return Message{}, fmt.Errorf("history: append id: %w", err)
}
msg.ID = id
}
if msg.CreatedAt == 0 {
msg.CreatedAt = time.Now().UnixMilli()
}
if msg.Platform == "" {
msg.Platform = PlatformWeb
}
_, err := s.db.ExecContext(ctx, `INSERT INTO chat_messages
(id, session_id, role, content, mode, mood, image, thinking, platform, device_id, sender_name, adapter_id, created_at)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
msg.ID, msg.SessionID, msg.Role, msg.Content, msg.Mode, msg.Mood,
msg.Image, msg.Thinking, msg.Platform, msg.DeviceID, msg.SenderName, msg.AdapterID, msg.CreatedAt)
if err != nil {
return Message{}, fmt.Errorf("history: append %s to %s: %w", msg.Role, msg.SessionID, err)
}
return msg, nil
}
// List returns up to limit messages for sessionID, newest first. When before is nonzero
// it bounds the query to messages strictly older than before, so a client can page upward
// by passing the oldest message's created_at from the previous page. When platform is
// non-empty it further restricts the page to messages carrying that source platform
// (design/16 §4.4: the Web client omits platform for the full main session, an IM
// adapter passes its own platform to see only its lane). The returned slice is non-nil.
// limit is clamped to [1, maxPageLimit].
func (s *Store) List(ctx context.Context, sessionID string, before int64, limit int, platform string) ([]Message, error) {
limit = clampLimit(limit)
query := `SELECT id, session_id, role, content, mode, mood, image, thinking, platform, device_id, sender_name, adapter_id, created_at
FROM chat_messages WHERE session_id = ?`
args := []any{sessionID}
if before > 0 {
query += ` AND created_at < ?`
args = append(args, before)
}
if platform != "" {
query += ` AND platform = ?`
args = append(args, platform)
}
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: list %s: %w", sessionID, 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 list %s: %w", sessionID, err)
}
msgs = append(msgs, m)
}
if err := rows.Err(); err != nil {
return nil, fmt.Errorf("history: iterate list %s: %w", sessionID, err)
}
return msgs, nil
}
// ListPlatformLog returns up to limit messages for platform across every non-main
// (isolated) session, newest first. Main-session messages are excluded so only the
// platform's own message stream (groups + stranger private chats) is surfaced to the
// adapter log page (design/16 §4.6 §4). before is a millisecond cursor; when nonzero
// only strictly older messages are returned. The returned slice is non-nil.
func (s *Store) ListPlatformLog(ctx context.Context, platform string, before int64, limit int) ([]Message, error) {
limit = clampLimit(limit)
query := `SELECT id, session_id, role, content, mode, mood, image, thinking, platform, device_id, sender_name, adapter_id, created_at
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 returns up to limit messages for a single IM adapter (platform +
// account registration name, e.g. "kook:3bef..."), newest first, for the per-adapter
// log page. It does NOT exclude the main web session: an owner-private IM chat lives
// in the main session but still carries the adapter_id, so a per-adapter log includes
// the owner mirror. An empty adapterID returns an empty (non-nil) slice. before is a
// millisecond cursor; when nonzero only strictly older messages are returned. The
// returned slice is non-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 id, session_id, role, content, mode, mood, image, thinking, platform, device_id, sender_name, adapter_id, created_at
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 returns up to limit messages across every IM adapter source
// (adapter_id != ''), newest first, for the adapter overview page. before is a
// millisecond cursor; when nonzero only strictly older messages are returned. The
// returned slice is non-nil.
func (s *Store) ListAllLog(ctx context.Context, before int64, limit int) ([]Message, error) {
limit = clampLimit(limit)
query := `SELECT id, session_id, role, content, mode, mood, image, thinking, platform, device_id, sender_name, adapter_id, created_at
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 returns the most recent n messages for sessionID in oldest-first order (the
// tail-most window), for WS replay / context tail. The returned slice is non-nil.
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 id, session_id, role, content, mode, mood, image, thinking, platform, device_id, sender_name, adapter_id, created_at
FROM chat_messages WHERE session_id = ? ORDER BY created_at DESC LIMIT ?`, sessionID, n)
if err != nil {
return nil, fmt.Errorf("history: recent %s: %w", sessionID, err)
}
defer func() { _ = rows.Close() }()
// Recent returns newest-last so the window is in chronological order; build it and
// reverse in place.
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
}
// DeleteSession removes a sub session and all its messages in one transaction. The fixed
// main session is immutable and returns ErrImmutableMain (design/16 §6).
func (s *Store) DeleteSession(ctx context.Context, sessionID string) error {
if sessionID == MainSessionID {
return ErrImmutableMain
}
tx, err := s.db.BeginTx(ctx, nil)
if err != nil {
return fmt.Errorf("history: begin delete %s: %w", sessionID, err)
}
defer func() { _ = tx.Rollback() }()
if _, err := tx.ExecContext(ctx, `DELETE FROM chat_messages WHERE session_id = ?`, sessionID); err != nil {
return fmt.Errorf("history: delete messages %s: %w", sessionID, err)
}
if _, err := tx.ExecContext(ctx, `DELETE FROM chat_sessions WHERE id = ? AND type = ?`, sessionID, SessionSub); err != nil {
return fmt.Errorf("history: delete session %s: %w", sessionID, err)
}
if err := tx.Commit(); err != nil {
return fmt.Errorf("history: commit delete %s: %w", sessionID, err)
}
return nil
}
// NewSubSession creates a sub session and returns its id (MainSessionID:uuid). ownerID is
// accepted for interface completeness (the Web owner is identified by JWT, not stored per
// session); it is logged at debug and not persisted.
func (s *Store) NewSubSession(ctx context.Context, ownerID string) (string, error) {
id, err := newID()
if err != nil {
return "", fmt.Errorf("history: new sub id: %w", err)
}
sessionID := MainSessionID + ":" + id
now := time.Now().UnixMilli()
if _, err := s.db.ExecContext(ctx, `INSERT INTO chat_sessions (id, type, title, created_at)
VALUES (?, ?, ?, ?)`, sessionID, SessionSub, "", now); err != nil {
return "", fmt.Errorf("history: insert sub session: %w", err)
}
if s.log != nil {
s.log.Debug("sub session created", "session_id", sessionID, "owner_id", ownerID)
}
return sessionID, nil
}
// ListSessions returns the main session first, then every sub session, oldest first. The
// returned slice is non-nil.
func (s *Store) ListSessions(ctx context.Context) ([]SessionMeta, error) {
rows, err := s.db.QueryContext(ctx, `SELECT id, type, title, created_at FROM chat_sessions
ORDER BY (type = ?) DESC, created_at ASC`, SessionMain)
if err != nil {
return nil, fmt.Errorf("history: list sessions: %w", err)
}
defer func() { _ = rows.Close() }()
sessions := make([]SessionMeta, 0)
for rows.Next() {
var (
meta SessionMeta
typeStr string
createdAt int64
)
if err := rows.Scan(&meta.ID, &typeStr, &meta.Title, &createdAt); err != nil {
return nil, fmt.Errorf("history: scan session: %w", err)
}
meta.Type = typeStr
meta.CreatedAt = createdAt
sessions = append(sessions, meta)
}
if err := rows.Err(); err != nil {
return nil, fmt.Errorf("history: iterate sessions: %w", err)
}
return sessions, nil
}
// ensureMainSession inserts the fixed main session row when it is absent, so ListSessions
// always reports the main session even before any message is written.
func (s *Store) ensureMainSession(ctx context.Context) error {
_, err := s.db.ExecContext(ctx, `INSERT OR IGNORE INTO chat_sessions (id, type, title, created_at)
VALUES (?, ?, ?, ?)`, MainSessionID, SessionMain, "", time.Now().UnixMilli())
if err != nil {
return fmt.Errorf("history: ensure main session: %w", err)
}
return nil
}
// scanMessage reads one chat_messages row. message columns are scanned positionally.
func scanMessage(sc rowScanner) (Message, error) {
var m Message
var mode, mood, platform, deviceID, senderName, adapterID string
if err := sc.Scan(&m.ID, &m.SessionID, &m.Role, &m.Content, &mode, &mood, &m.Image, &m.Thinking, &platform, &deviceID, &senderName, &adapterID, &m.CreatedAt); err != nil {
return Message{}, err
}
m.Mode, m.Mood, m.Platform, m.DeviceID, m.SenderName, m.AdapterID = mode, mood, platform, deviceID, senderName, adapterID
return m, nil
}
// clampLimit bounds limit to [1, maxPageLimit]; a caller-supplied 0 or negative falls back
// to defaultPageLimit.
func clampLimit(limit int) int {
if limit < 1 {
return defaultPageLimit
}
if limit > maxPageLimit {
return maxPageLimit
}
return limit
}
// newID returns a 128-bit cryptographically random id, hex-encoded. It avoids a UUID
// dependency so the package stays self-contained.
func newID() (string, error) {
b := make([]byte, 16)
if _, err := rand.Read(b); err != nil {
return "", fmt.Errorf("history: generate id: %w", err)
}
return hex.EncodeToString(b), nil
}
// discardLogger returns log if non-nil, otherwise a logger that writes nowhere.
func discardLogger(log *slog.Logger) *slog.Logger {
if log != nil {
return log
}
return slog.New(slog.NewTextHandler(io.Discard, nil))
}
// Schema DDL. The 0003 migration in internal/db is the authoritative mechanism for
// domain tables, but it is intentionally also re-issued here idempotently so the store is
// self-contained and usable from any *sql.DB (matching the memory and session packages).
const createSessionsTable = `
CREATE TABLE IF NOT EXISTS chat_sessions (
id TEXT PRIMARY KEY,
type TEXT NOT NULL,
title TEXT NOT NULL DEFAULT '',
created_at INTEGER NOT NULL
)`
const createMessagesTable = `
CREATE TABLE IF NOT EXISTS chat_messages (
id TEXT PRIMARY KEY,
session_id TEXT NOT NULL,
role TEXT NOT NULL,
content TEXT NOT NULL,
mode TEXT,
mood TEXT,
image TEXT,
thinking TEXT,
platform TEXT NOT NULL DEFAULT 'web',
device_id TEXT NOT NULL DEFAULT '',
sender_name TEXT NOT NULL DEFAULT '',
adapter_id TEXT NOT NULL DEFAULT '',
created_at INTEGER NOT NULL
)`
const createMessagesIndex = `
CREATE INDEX IF NOT EXISTS idx_chat_msgs_session_created
ON chat_messages (session_id, created_at DESC)`