Cyrene_v2/internal/services/gateway/observability_audit.go

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
}