392 lines
14 KiB
Go
392 lines
14 KiB
Go
// Package store 是持久化层唯一入口（存储档接口，设计 4.1/D5/D16）。
|
||
// 两档实现共用同一方法面：
|
||
// - 嵌入式档 SQLite（默认：单文件 + 本地对象目录，零外部服务）；
|
||
// - PostgreSQL 档（多用户/多实例；DSN 配置，pgx 纯 Go 驱动，静态构建兼容）。
|
||
//
|
||
// 方言差异（占位符、UPSERT、全文检索、备份、咨询锁）收敛在 dialect 与方言文件；
|
||
// 业务层只依赖本包公开方法。schema 演进走版本化迁移（migrate.go，双方言、可回滚）。
|
||
// 租户硬边界：租户域查询一律带 owner 过滤，跨租户访问返回 ErrNotFound。
|
||
package store
|
||
|
||
import (
|
||
"context"
|
||
"database/sql"
|
||
"encoding/json"
|
||
"errors"
|
||
"fmt"
|
||
"os"
|
||
"path/filepath"
|
||
"strconv"
|
||
"strings"
|
||
"sync"
|
||
"time"
|
||
|
||
"agent-system-backend/internal/blobstore"
|
||
|
||
_ "github.com/jackc/pgx/v5/stdlib" // PG 档驱动（注册名 "pgx"）
|
||
_ "modernc.org/sqlite" // 嵌入式档驱动（注册名 "sqlite"，CGO-free）
|
||
)
|
||
|
||
var ErrNotFound = errors.New("store: not found")
|
||
var ErrConflict = errors.New("store: conflict")
|
||
|
||
// ErrDuplicateRequest 表示同一会话的 request_id 已经落库。
|
||
// 调用方应返回已有消息，不要映射成 409：客户端把 409 当成会话忙并重新入队。
|
||
var ErrDuplicateRequest = errors.New("store: duplicate request id")
|
||
|
||
// OwnerDefault 是单租户默认 owner（嵌入式档 / 未启用认证时的隐式身份）。
|
||
const OwnerDefault = "local"
|
||
|
||
// PoolOptions PG 连接预算（协作存储方案 KTD12）。零值走默认（DefaultPGPool）；
|
||
// 预算应覆盖全部 API/Worker/LISTEN/migrator 连接，100 在线用户 ≠ 100 连接。
|
||
type PoolOptions struct {
|
||
MaxOpenConn int // 每进程最大打开连接；0=默认 16
|
||
MaxIdleConn int // 每进程最大空闲连接；0=默认 8
|
||
ConnMaxIdleTimeSec int // 空闲连接回收时长；0=默认 300s
|
||
ConnMaxLifetimeSec int // 连接最长生命周期；0=不限
|
||
QueryTimeoutSec int // PG statement_timeout（秒）；0=不限
|
||
}
|
||
|
||
// DefaultPGPool 与历史行为一致的默认预算（16/8/300s）。
|
||
var DefaultPGPool = PoolOptions{MaxOpenConn: 16, MaxIdleConn: 8, ConnMaxIdleTimeSec: 300}
|
||
|
||
func (p PoolOptions) effective() PoolOptions {
|
||
out := p
|
||
if out.MaxOpenConn <= 0 {
|
||
out.MaxOpenConn = DefaultPGPool.MaxOpenConn
|
||
}
|
||
if out.MaxIdleConn <= 0 {
|
||
out.MaxIdleConn = DefaultPGPool.MaxIdleConn
|
||
}
|
||
if out.ConnMaxIdleTimeSec <= 0 {
|
||
out.ConnMaxIdleTimeSec = DefaultPGPool.ConnMaxIdleTimeSec
|
||
}
|
||
return out
|
||
}
|
||
|
||
type Store struct {
|
||
db *sql.DB
|
||
dialect dialect
|
||
dataDir string // 工件对象目录（备份路径等诊断用）
|
||
blobs blobstore.Store // 工件对象存储（本地档；多实例切共享卷/对象存储）
|
||
instanceID string // 本进程标识；CreateRun 写入 runs.instance_id
|
||
pool PoolOptions // 生效的连接预算（诊断输出用）
|
||
|
||
settleMu sync.Mutex
|
||
settleFault error // 非空时 SettleRun 在写入前失败，用于终态提交失败测试
|
||
|
||
migrateMu sync.Mutex
|
||
migrateFault error // 非空时 migratePass 在「DDL 已执行、版本未登记」处失败（原子性测试）
|
||
accountFault error // 非空时账户事务在提交前失败（半态测试）
|
||
|
||
vecPGOnce sync.Once // pgvector 列探测（U6-3：一次性缓存）
|
||
vecPG bool
|
||
}
|
||
|
||
// OpenSQLite 打开（必要时创建）嵌入式存储并执行迁移。
|
||
func OpenSQLite(path string) (*Store, error) {
|
||
if err := os.MkdirAll(filepath.Dir(path), 0o755); err != nil {
|
||
return nil, err
|
||
}
|
||
return openSQLiteFile(path, true, true)
|
||
}
|
||
|
||
// OpenIdentity 打开既有库做身份 ensure/verify：不创建文件或目录，不执行迁移。
|
||
// 缺少 agent_system_identity 表时返回 ErrIdentityTableMissing。
|
||
func OpenIdentity(storageDSN, dataDir string) (*Store, error) {
|
||
if storageDSN != "" {
|
||
return openIdentityPostgres(storageDSN)
|
||
}
|
||
return openIdentitySQLite(filepath.Join(dataDir, "agent-system.db"))
|
||
}
|
||
|
||
func openIdentitySQLite(path string) (*Store, error) {
|
||
if _, err := os.Stat(path); err != nil {
|
||
if os.IsNotExist(err) {
|
||
return nil, fmt.Errorf("%w: %s", ErrIdentityDBMissing, path)
|
||
}
|
||
return nil, err
|
||
}
|
||
dsn := fmt.Sprintf("file:%s?_pragma=busy_timeout(30000)&_pragma=foreign_keys(1)&_txlock=immediate", path)
|
||
db, err := sql.Open("sqlite", dsn)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
db.SetMaxOpenConns(1)
|
||
if err := requireIdentityTable(db, "sqlite"); err != nil {
|
||
db.Close()
|
||
return nil, err
|
||
}
|
||
return &Store{db: db, dialect: sqliteDialect{}}, nil
|
||
}
|
||
|
||
func openIdentityPostgres(dsn string) (*Store, error) {
|
||
db, err := sql.Open("pgx", dsn)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
db.SetMaxOpenConns(2)
|
||
if err := db.Ping(); err != nil {
|
||
db.Close()
|
||
return nil, err
|
||
}
|
||
if err := requireIdentityTable(db, "postgres"); err != nil {
|
||
db.Close()
|
||
return nil, err
|
||
}
|
||
return &Store{db: db, dialect: postgresDialect{}}, nil
|
||
}
|
||
|
||
func requireIdentityTable(db *sql.DB, kind string) error {
|
||
var name string
|
||
var err error
|
||
switch kind {
|
||
case "postgres":
|
||
err = db.QueryRow(`SELECT COALESCE(to_regclass('public.agent_system_identity')::text, '')`).Scan(&name)
|
||
default:
|
||
err = db.QueryRow(`SELECT COALESCE((SELECT name FROM sqlite_master WHERE type='table' AND name='agent_system_identity'), '')`).Scan(&name)
|
||
}
|
||
if err != nil {
|
||
return err
|
||
}
|
||
if name == "" {
|
||
return fmt.Errorf("%w: 先执行受控迁移（服务启动或 migrate），身份命令不建库、不迁移", ErrIdentityTableMissing)
|
||
}
|
||
return nil
|
||
}
|
||
|
||
// SchemaVersion 读取 schema_migrations 的最大版本。表不存在时返回 0。
|
||
func (s *Store) SchemaVersion() (int, error) {
|
||
var v sql.NullInt64
|
||
err := s.db.QueryRow(`SELECT MAX(version) FROM schema_migrations`).Scan(&v)
|
||
if err != nil {
|
||
msg := strings.ToLower(err.Error())
|
||
if strings.Contains(msg, "no such table") || strings.Contains(msg, "does not exist") {
|
||
return 0, nil
|
||
}
|
||
return 0, err
|
||
}
|
||
if !v.Valid {
|
||
return 0, nil
|
||
}
|
||
return int(v.Int64), nil
|
||
}
|
||
|
||
// OpenExistingSQLite 打开既有嵌入式库：数据文件不存在则失败，不创建空库或父目录，也不迁移。
|
||
// schema 与本二进制不一致时拒绝，并提示使用 migrate 或先启动服务。
|
||
func OpenExistingSQLite(path string) (*Store, error) {
|
||
if _, err := os.Stat(path); err != nil {
|
||
if os.IsNotExist(err) {
|
||
return nil, fmt.Errorf("open existing: sqlite file missing: %s", path)
|
||
}
|
||
return nil, err
|
||
}
|
||
return openSQLiteFile(path, false, false)
|
||
}
|
||
|
||
func openSQLiteFile(path string, createArtifacts, migrateSchema bool) (*Store, error) {
|
||
// _txlock=immediate：写事务在 BEGIN 即取库写锁（而非首个写语句升级），
|
||
// 跨进程迁移/写路径依赖 busy_timeout 排队而不是中途 SQLITE_BUSY（KTD8）。
|
||
// busy_timeout 30s：迁移竞争测试在慢速 CI 盘上 4 个迁移者串行全程可超 5s
|
||
// （实测 job 2285 SQLITE_BUSY）；本地档宁可排队不报错。
|
||
// synchronous(FULL) 是恢复方案 KTD14 的可靠档。打开后在同一连接读回，见 verifyDurability。
|
||
dsn := fmt.Sprintf("file:%s?_pragma=busy_timeout(30000)&_pragma=journal_mode(WAL)&_pragma=synchronous(FULL)&_pragma=foreign_keys(1)&_txlock=immediate", path)
|
||
db, err := sql.Open("sqlite", dsn)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
// SQLite 单写者：串行化连接，语句级原子 + 版本化迁移事务。
|
||
db.SetMaxOpenConns(1)
|
||
return finishOpenMode(db, sqliteDialect{}, filepath.Join(filepath.Dir(path), "artifacts"), nil, PoolOptions{}, createArtifacts, migrateSchema)
|
||
}
|
||
|
||
// OpenPostgres 打开 PostgreSQL 档存储并执行迁移（默认连接预算）。
|
||
// DSN 例：postgres://user:pass@host:5432/agent_system?sslmode=disable
|
||
// artifactsDir 为工件对象目录（多实例部署时指向共享卷；单实例默认本地）。
|
||
func OpenPostgres(dsn, artifactsDir string) (*Store, error) {
|
||
return OpenPostgresWithOptions(dsn, artifactsDir, PoolOptions{}, nil)
|
||
}
|
||
|
||
// OpenPostgresWithBlobs 是注入自定义对象存储的 PG 档入口（多实例共享卷/对象存储）。
|
||
func OpenPostgresWithBlobs(dsn, artifactsDir string, blobs blobstore.Store) (*Store, error) {
|
||
return OpenPostgresWithOptions(dsn, artifactsDir, PoolOptions{}, blobs)
|
||
}
|
||
|
||
// OpenPostgresWithOptions 按 KTD12 连接预算打开 PG 档；QueryTimeoutSec>0 时
|
||
// 以 statement_timeout 运行参数附着到 DSN（对 URL 与键值两种形式均生效）。
|
||
// blobs 为 nil（含带类型的空指针，例如未配置端点时的 (*blobstore.S3)(nil)）时回落本地目录。
|
||
func OpenPostgresWithOptions(dsn, artifactsDir string, pool PoolOptions, blobs blobstore.Store) (*Store, error) {
|
||
return openPostgresWithOptions(dsn, artifactsDir, pool, blobs, true, true)
|
||
}
|
||
|
||
// OpenExistingPostgres 打开既有 PG 库，不创建 artifacts 目录，也不迁移。
|
||
func OpenExistingPostgres(dsn, artifactsDir string) (*Store, error) {
|
||
return openPostgresWithOptions(dsn, artifactsDir, PoolOptions{}, nil, false, false)
|
||
}
|
||
|
||
func openPostgresWithOptions(dsn, artifactsDir string, pool PoolOptions, blobs blobstore.Store, createArtifacts, migrateSchema bool) (*Store, error) {
|
||
pool = pool.effective()
|
||
if pool.QueryTimeoutSec > 0 {
|
||
dsn = withPGRuntimeParam(dsn, "statement_timeout", strconv.Itoa(pool.QueryTimeoutSec*1000))
|
||
}
|
||
db, err := sql.Open("pgx", dsn)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
db.SetMaxOpenConns(pool.MaxOpenConn)
|
||
db.SetMaxIdleConns(pool.MaxIdleConn)
|
||
db.SetConnMaxIdleTime(time.Duration(pool.ConnMaxIdleTimeSec) * time.Second)
|
||
if pool.ConnMaxLifetimeSec > 0 {
|
||
db.SetConnMaxLifetime(time.Duration(pool.ConnMaxLifetimeSec) * time.Second)
|
||
}
|
||
return finishOpenMode(db, postgresDialect{}, artifactsDir, blobs, pool, createArtifacts, migrateSchema)
|
||
}
|
||
|
||
// withPGRuntimeParam 向 DSN 追加运行时参数（libpq 语义：URL 查询串或键值对）。
|
||
// 不改写既有参数；值中不含分隔符时安全（statement_timeout/search_path 均满足）。
|
||
func withPGRuntimeParam(dsn, key, value string) string {
|
||
return AppendPGRuntimeParam(dsn, key, value)
|
||
}
|
||
|
||
// AppendPGRuntimeParam 导出形态：测试与其他包（如 runmgr 的 schema 隔离）
|
||
// 共用同一 DSN 拼接规则，避免各自盲拼 "&key=value" 造出非法库名。
|
||
func AppendPGRuntimeParam(dsn, key, value string) string {
|
||
kv := key + "=" + value
|
||
if !strings.Contains(dsn, "://") {
|
||
if strings.TrimSpace(dsn) == "" {
|
||
return kv
|
||
}
|
||
return dsn + " " + kv
|
||
}
|
||
sep := "?"
|
||
if strings.Contains(dsn, "?") {
|
||
sep = "&"
|
||
}
|
||
return dsn + sep + kv
|
||
}
|
||
|
||
// Preserve the existing initialization helper contract; maintenance opens select
|
||
// nonmigrating mode explicitly through finishOpenMode.
|
||
func finishOpen(db *sql.DB, d dialect, artifactsDir string, blobs blobstore.Store, pool PoolOptions, createArtifacts bool) (*Store, error) {
|
||
return finishOpenMode(db, d, artifactsDir, blobs, pool, createArtifacts, true)
|
||
}
|
||
|
||
func finishOpenMode(db *sql.DB, d dialect, artifactsDir string, blobs blobstore.Store, pool PoolOptions, createArtifacts, migrateSchema bool) (*Store, error) {
|
||
if blobstore.IsNil(blobs) {
|
||
var err error
|
||
if createArtifacts {
|
||
blobs, err = blobstore.NewLocal(artifactsDir)
|
||
} else {
|
||
blobs, err = blobstore.OpenLocalExisting(artifactsDir)
|
||
}
|
||
if err != nil {
|
||
db.Close()
|
||
return nil, err
|
||
}
|
||
}
|
||
s := &Store{db: db, dialect: d, dataDir: artifactsDir, blobs: blobs, pool: pool}
|
||
if migrateSchema {
|
||
if err := s.migrate(); err != nil {
|
||
db.Close()
|
||
return nil, err
|
||
}
|
||
} else if err := s.requireCurrentSchema(); err != nil {
|
||
db.Close()
|
||
return nil, err
|
||
}
|
||
if err := s.verifyDurability(); err != nil {
|
||
db.Close()
|
||
return nil, err
|
||
}
|
||
if createArtifacts {
|
||
if err := os.MkdirAll(artifactsDir, 0o755); err != nil {
|
||
db.Close()
|
||
return nil, err
|
||
}
|
||
}
|
||
return s, nil
|
||
}
|
||
|
||
func (s *Store) Close() error { return s.db.Close() }
|
||
|
||
// SetInstanceID 记录本进程标识。之后 CreateRun 把它写入 runs.instance_id。
|
||
// 只在开始接请求前调用一次。
|
||
func (s *Store) SetInstanceID(id string) { s.instanceID = id }
|
||
|
||
// SetSettleFault 让下一次 SettleRun 在写入前失败。nil 表示恢复正常。
|
||
func (s *Store) SetSettleFault(err error) {
|
||
s.settleMu.Lock()
|
||
s.settleFault = err
|
||
s.settleMu.Unlock()
|
||
}
|
||
|
||
// SetMigrateFault 让下一次 migratePass 在首个待应用迁移的「DDL 已执行、版本
|
||
// 未登记」位置失败一次（nil 恢复）。测试专用：验证 DDL 与账本同事务回滚。
|
||
func (s *Store) SetMigrateFault(err error) {
|
||
s.migrateMu.Lock()
|
||
s.migrateFault = err
|
||
s.migrateMu.Unlock()
|
||
}
|
||
|
||
func (s *Store) takeMigrateFault() error {
|
||
s.migrateMu.Lock()
|
||
defer s.migrateMu.Unlock()
|
||
err := s.migrateFault
|
||
s.migrateFault = nil
|
||
return err
|
||
}
|
||
|
||
// Dialect 返回当前存储档名（sqlite | postgres；诊断/健康检查用）。
|
||
func (s *Store) Dialect() string { return s.dialect.name() }
|
||
|
||
// Pool 返回生效的连接预算（启动诊断；脱敏，不含 DSN）。
|
||
func (s *Store) Pool() PoolOptions { return s.pool }
|
||
|
||
// DB 暴露底层连接（运维工具专用：storage-migrate 的控制表与
|
||
// information_schema 查询）。业务代码一律走本包公开方法。
|
||
func (s *Store) DB() *sql.DB { return s.db }
|
||
|
||
// DataDir 返回工件对象目录。
|
||
func (s *Store) DataDir() string { return s.dataDir }
|
||
|
||
// BlobStore 返回工件对象存储（诊断/运维）。
|
||
func (s *Store) BlobStore() blobstore.Store { return s.blobs }
|
||
|
||
// NotifyPG 发送 PostgreSQL NOTIFY（PG 档事件扇出用；SQLite 档返回未执行）。
|
||
func (s *Store) NotifyPG(ctx context.Context, channel, payload string) (bool, error) {
|
||
if s.dialect.name() != "postgres" {
|
||
return false, nil
|
||
}
|
||
_, err := s.db.ExecContext(ctx, `SELECT pg_notify($1, $2)`, channel, payload)
|
||
return true, err
|
||
}
|
||
|
||
// q 按方言重绑占位符（查询统一用 ? 书写，PG 执行前转 $n）。
|
||
// 本包 SQL 不含字符串字面量 "?"，重绑安全。
|
||
func (s *Store) q(query string) string { return s.dialect.rebind(query) }
|
||
|
||
// Backup 生成一致性备份文件。SQLite 档用 VACUUM INTO（在线安全，含 WAL 内容）；
|
||
// PG 档备份走 pg_dump / WAL 归档（运维路径，见 usage 指南），此处返回指引错误。
|
||
func (s *Store) Backup(destPath string) error {
|
||
if s.dialect.name() != "sqlite" {
|
||
return errors.New("store: PG 档备份走 pg_dump/WAL 归档（见 docs/guides/usage.md 存储档一节）")
|
||
}
|
||
if _, err := s.db.Exec(`VACUUM INTO ?`, destPath); err != nil {
|
||
return fmt.Errorf("store: backup: %w", err)
|
||
}
|
||
return nil
|
||
}
|
||
|
||
func nowISO() string { return time.Now().UTC().Format(time.RFC3339Nano) }
|
||
|
||
func marshalJSON(v any) string {
|
||
b, _ := json.Marshal(v)
|
||
return string(b)
|
||
}
|
||
|
||
func unmarshalJSON[T any](raw string) (T, error) {
|
||
var out T
|
||
err := json.Unmarshal([]byte(raw), &out)
|
||
return out, err
|
||
}
|