package gateway import ( "context" "database/sql" "fmt" "log/slog" "sync" "time" ) // AuditRecord 是一条持久化的管理审计条目(who/what/when/result)。 type AuditRecord struct { ID int64 `json:"id"` Who string `json:"who"` What string `json:"what"` When string `json:"when"` Result string `json:"result"` } // AuditStore 持久化管理审计链。实现必须对 // 并发使用安全。*SQLiteAuditStore 是生产实现(由 // 组合根接到 a.db 上);未注入任何实现时网关回退到 // 内存存储,因此端点总能返回数据。 type AuditStore interface { // Append 在 ts 持久化一条审计条目。 Append(ctx context.Context, who, what, result string, ts time.Time) error // List 返回至多 limit 条最近条目,最新优先。非正的 // limit 选用默认值。 List(ctx context.Context, limit int) ([]AuditRecord, error) } // memoryAuditStore 是未注入时的回退:它把审计链放在 // 进程内存中,使管理端点在组合根注入 SQLite 存储 // 之前也能工作。它对并发安全。 type memoryAuditStore struct { mu sync.Mutex records []AuditRecord seq int64 } // newMemoryAuditStore 构建一个空的内存审计存储。 func newMemoryAuditStore() *memoryAuditStore { return &memoryAuditStore{records: make([]AuditRecord, 0, 64)} } // Append 用单调递增的 id 存一条条目。 func (m *memoryAuditStore) Append(ctx context.Context, who, what, result string, ts time.Time) error { if err := ctx.Err(); err != nil { return err } m.mu.Lock() defer m.mu.Unlock() m.seq++ m.records = append(m.records, AuditRecord{ ID: m.seq, Who: who, What: what, When: ts.UTC().Format(time.RFC3339), Result: result, }) return nil } // List 返回最近的条目,最新优先。 func (m *memoryAuditStore) List(ctx context.Context, limit int) ([]AuditRecord, error) { if err := ctx.Err(); err != nil { return nil, err } if limit <= 0 { limit = defaultObsLimit } if limit > maxObsLimit { limit = maxObsLimit } m.mu.Lock() defer m.mu.Unlock() n := len(m.records) start := n - limit if start < 0 { start = 0 } out := make([]AuditRecord, n-start) copy(out, m.records[start:]) for i, j := 0, len(out)-1; i < j; i, j = i+1, j-1 { out[i], out[j] = out[j], out[i] } return out, nil } // 自包含 SQLite 存储的审计 DDL(无需迁移记账),与 // channel/session 与 backthink 所用模式相同。recorded_at 是被审计 // 动作的 UTC RFC3339 墙钟时间。 const ( createAuditSQL = ` CREATE TABLE IF NOT EXISTS audit_logs ( id INTEGER PRIMARY KEY AUTOINCREMENT, who TEXT NOT NULL, what TEXT NOT NULL, result TEXT NOT NULL, recorded_at TEXT NOT NULL, created_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ','now')) )` createAuditIndexSQL = `CREATE INDEX IF NOT EXISTS idx_audit_logs_recorded_at ON audit_logs(recorded_at)` insertAuditSQL = `INSERT INTO audit_logs (who, what, result, recorded_at) VALUES (?, ?, ?, ?)` selectAuditSQL = `SELECT id, who, what, result, recorded_at FROM audit_logs ORDER BY id DESC LIMIT ?` ) // SQLiteAuditStore 把管理审计链持久化在 SQLite 中。凭借底层的 // *sql.DB,它对并发使用安全。 type SQLiteAuditStore struct { db *sql.DB log *slog.Logger } // NewSQLiteAuditStore 在 db 上构建 SQLite 审计存储,保证 // audit_logs 表与索引存在。db 为 nil 或 DDL 失败时 // 返回错误。组合根把它接到主 *sql.DB(a.db)上。 func NewSQLiteAuditStore(ctx context.Context, db *sql.DB, log *slog.Logger) (*SQLiteAuditStore, error) { if db == nil { return nil, fmt.Errorf("gateway: audit store: nil database") } if _, err := db.ExecContext(ctx, createAuditSQL); err != nil { return nil, fmt.Errorf("gateway: ensure audit_logs: %w", err) } if _, err := db.ExecContext(ctx, createAuditIndexSQL); err != nil { return nil, fmt.Errorf("gateway: ensure audit_logs index: %w", err) } return &SQLiteAuditStore{db: db, log: sourcedLog(log)}, nil } // Append 在 ts 持久化一条审计条目。 func (s *SQLiteAuditStore) Append(ctx context.Context, who, what, result string, ts time.Time) error { if _, err := s.db.ExecContext(ctx, insertAuditSQL, who, what, result, ts.UTC().Format(time.RFC3339Nano)); err != nil { return fmt.Errorf("gateway: insert audit: %w", err) } return nil } // List 返回最近的条目,最新优先。 func (s *SQLiteAuditStore) List(ctx context.Context, limit int) ([]AuditRecord, error) { if limit <= 0 { limit = defaultObsLimit } if limit > maxObsLimit { limit = maxObsLimit } if err := ctx.Err(); err != nil { return nil, err } rows, err := s.db.QueryContext(ctx, selectAuditSQL, limit) if err != nil { return nil, fmt.Errorf("gateway: list audit: %w", err) } defer rows.Close() records := make([]AuditRecord, 0, limit) for rows.Next() { var r AuditRecord if err := rows.Scan(&r.ID, &r.Who, &r.What, &r.Result, &r.When); err != nil { return nil, fmt.Errorf("gateway: scan audit: %w", err) } records = append(records, r) } if err := rows.Err(); err != nil { return nil, fmt.Errorf("gateway: iterate audit: %w", err) } return records, nil }