// 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 // 非空时账户事务在提交前失败（半态测试）

	documentDeliveryMu         sync.Mutex
	documentDeliveryFaultStage string
	documentDeliveryFault      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
}