Agent-System-Backend/internal/store/migrate_atomic_test.go

package store

import (
	"database/sql"
	"errors"
	"fmt"
	"os"
	"path/filepath"
	"strings"
	"sync"
	"testing"
)

// 协作存储方案 U2（KTD8）：迁移原子性与并发正确性。
//   - DDL 与版本登记同事务：注入「DDL 已执行、版本未登记」处故障 → 整体回滚；
//   - 并发迁移者竞争（SQLite _txlock=immediate / PG 咨询锁）：账本恰好一套；
//   - 旧 schema（仅部分版本）打开后自动升级到最新；
//   - team 档 PG 不可达：打开失败、无任何回落（R2 fail-closed 在 store 层的基线）。

// sqliteAt 用直接 SQL 预置一个「旧部署」库：只应用 v1 基线并登记账本。
func sqliteAt(t *testing.T, path string) {
	t.Helper()
	st, err := OpenSQLite(path)
	if err != nil {
		t.Fatalf("seed open: %v", err)
	}
	if err := st.MigrateDown(1); //nolint:errcheck // 回滚到 v1 失败说明 down 链路破损，由下方状态断言兜住
	err != nil {
		t.Logf("seed migrate-down: %v", err)
	}
	if vers, err := st.MigrationStatus(); err != nil || len(vers) == 0 {
		t.Fatalf("seed status: %v %v", vers, err)
	} else if vers[len(vers)-1] != 1 {
		t.Fatalf("seed should stop at v1, got %v", vers)
	}
	st.Close()
}

func openSQLiteAt(t *testing.T, path string) *Store {
	t.Helper()
	st, err := OpenSQLite(path)
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { st.Close() })
	return st
}

// 旧 schema 升级：只有 v1 账本的库重新打开后升级到最新（双档）。
func TestMigrateUpgradesOldSchema(t *testing.T) {
	if os.Getenv("AGENT_PG_TEST_DSN") != "" {
		t.Run("postgres", func(t *testing.T) {
			st := openTestStore(t)
			if st.Dialect() != "postgres" {
				t.Skip("非 PG 档")
			}
			if err := st.MigrateDown(1); err != nil {
				t.Fatalf("migrate-down to v1: %v", err)
			}
			if err := st.migrate(); err != nil {
				t.Fatalf("re-migrate: %v", err)
			}
			assertFullLedger(t, st)
		})
	}
	t.Run("sqlite", func(t *testing.T) {
		path := filepath.Join(t.TempDir(), "old.db")
		sqliteAt(t, path)
		st := openSQLiteAt(t, path)
		assertFullLedger(t, st)
	})
}

func assertFullLedger(t *testing.T, st *Store) {
	t.Helper()
	vers, err := st.MigrationStatus()
	if err != nil {
		t.Fatal(err)
	}
	if len(vers) != len(migrations) {
		t.Fatalf("ledger %v, want %d versions", vers, len(migrations))
	}
	for i, m := range migrations {
		if vers[i] != m.version {
			t.Fatalf("ledger[%d]=%d, want %d", i, vers[i], m.version)
		}
	}
}

// Separate feature candidates reused version 31 before current main assigned
// that number to account_audit. A version match alone must not silently treat
// the candidate's different schema as an applied release migration.
func TestMigrateRejectsLedgerNameMismatch(t *testing.T) {
	path := filepath.Join(t.TempDir(), "name-mismatch.db")
	st := openSQLiteAt(t, path)
	if _, err := st.db.Exec(`UPDATE schema_migrations SET name='work_task_bindings' WHERE version=31`); err != nil {
		t.Fatal(err)
	}
	st.Close()
	if reopened, err := OpenSQLite(path); err == nil {
		reopened.Close()
		t.Fatal("reused migration version with different name was accepted")
	} else if !strings.Contains(err.Error(), "migration identity mismatch") {
		t.Fatalf("missing fail-closed migration identity error: %v", err)
	}
}

func TestMigrateAdoptedExecEnvHistoryUpgradesWithoutReinterpretingV35(t *testing.T) {
	st := openTestStore(t)
	if err := st.MigrateDown(35); err != nil {
		t.Fatalf("migrate down to adopted exec env: %v", err)
	}
	request := ExecEnvRequest{
		ID: "exec-request-1", ConversationID: "conversation-1", Owner: "alice",
		Spec: "python-docx", Status: "approved", CreatedAt: "2026-10-15T00:00:00Z", UpdatedAt: "2026-10-15T00:00:00Z",
	}
	if err := st.InsertExecEnvRequest(request); err != nil {
		t.Fatalf("insert v35 exec env row: %v", err)
	}
	if _, err := st.db.Exec(`SELECT COUNT(*) FROM work_task_bindings`); err == nil {
		t.Fatal("v35 exec env history was silently treated as work_task_bindings")
	}
	if err := st.migrate(); err != nil {
		t.Fatalf("upgrade adopted exec env history: %v", err)
	}
	if got, err := st.GetExecEnvRequest(request.ID); err != nil || got.Spec != request.Spec {
		t.Fatalf("exec env row did not survive feature migrations: got=%+v err=%v", got, err)
	}
	if _, err := st.db.Exec(`SELECT COUNT(*) FROM work_task_bindings`); err != nil {
		t.Fatalf("feature migration did not create work_task_bindings after v35: %v", err)
	}
}

func TestMigrateRejectsLegacyFeatureNumberingBeforeApplyingCurrentTail(t *testing.T) {
	st := openTestStore(t)
	if err := st.MigrateDown(35); err != nil {
		t.Fatalf("migrate down to v35: %v", err)
	}
	for _, query := range []string{
		`DROP TABLE exec_env_bindings`,
		`DROP TABLE exec_env_requests`,
		`CREATE TABLE work_task_bindings(task_id TEXT PRIMARY KEY,owner TEXT NOT NULL,plugin_id TEXT NOT NULL,conversation_id TEXT NOT NULL,project_id TEXT NOT NULL,created_at TEXT NOT NULL,FOREIGN KEY(task_id) REFERENCES background_tasks(id) ON DELETE CASCADE)`,
		`UPDATE schema_migrations SET name='work_task_bindings' WHERE version=35`,
	} {
		if _, err := st.db.Exec(query); err != nil {
			t.Fatalf("seed legacy feature v35: %v", err)
		}
	}
	err := st.migrate()
	if err == nil || !strings.Contains(err.Error(), "legacy feature migration numbering") || !strings.Contains(err.Error(), "v35") {
		t.Fatalf("legacy feature v35 was not rejected explicitly: %v", err)
	}
	if _, err := st.db.Exec(`SELECT COUNT(*) FROM exec_env_requests`); err == nil {
		t.Fatal("rejected legacy history still applied current exec_env identity")
	}
	if _, err := st.db.Exec(`SELECT COUNT(*) FROM work_task_bindings`); err != nil {
		t.Fatalf("rejected legacy history changed its work_task_bindings table: %v", err)
	}
}

func TestMigrateDownRejectsLedgerNameDriftBeforeChangingSchema(t *testing.T) {
	path := filepath.Join(t.TempDir(), "down-name-drift.db")
	st := openSQLiteAt(t, path)
	if _, err := st.db.Exec(`UPDATE schema_migrations SET name='wrong-candidate' WHERE version=45`); err != nil {
		t.Fatal(err)
	}
	err := st.MigrateDown(44)
	if err == nil || !strings.Contains(err.Error(), "migration identity mismatch") {
		t.Fatalf("rollback accepted a changed migration identity: %v", err)
	}
	var remain int
	if err := st.db.QueryRow(`SELECT COUNT(*) FROM schema_migrations WHERE version=45`).Scan(&remain); err != nil || remain != 1 {
		t.Fatalf("rollback changed ledger before refusing: remaining=%d err=%v", remain, err)
	}
}

// DDL 与账本同事务：在「DDL 已执行、版本未登记」处注入故障 → 全部回滚，
// 不留下 v6 的表，也不留下 v6 的账本行（历史两步实现的中间态缺口）。
func TestMigrateAtomicDDLAndLedger(t *testing.T) {
	path := filepath.Join(t.TempDir(), "atomic.db")
	st := openSQLiteAt(t, path)
	if err := st.MigrateDown(5); err != nil {
		t.Fatalf("migrate-down to v5: %v", err)
	}
	vers, _ := st.MigrationStatus()
	if vers[len(vers)-1] != 5 {
		t.Fatalf("want v5 boundary, got %v", vers)
	}
	boom := errors.New("crash between DDL and ledger")
	st.SetMigrateFault(boom)
	err := st.migrate()
	if err == nil || !strings.Contains(err.Error(), "注入故障") {
		t.Fatalf("want injected fault, got %v", err)
	}
	// 账本不含 v6；v6 的表必须不存在（同事务回滚）。
	if vers, _ = st.MigrationStatus(); vers[len(vers)-1] != 5 {
		t.Fatalf("ledger advanced despite fault: %v", vers)
	}
	var n int
	if err := st.db.QueryRow(`SELECT COUNT(*) FROM sqlite_master WHERE type='table' AND name='activity_cards'`).Scan(&n); err != nil {
		t.Fatal(err)
	}
	if n != 0 {
		t.Fatal("activity_cards exists although the version insert faulted: DDL leaked outside the transaction")
	}
	// 恢复后重跑：升级完成。
	if err := st.migrate(); err != nil {
		t.Fatalf("retry migrate: %v", err)
	}
	assertFullLedger(t, st)
}

// 并发迁移者竞争同一 SQLite 文件：_txlock=immediate + busy_timeout 串行化，
// 两边都成功，账本恰好一套（不因竞争出现重复或部分版本）。
func TestMigrateRacingSQLite(t *testing.T) {
	path := filepath.Join(t.TempDir(), "race.db")
	// 先物化文件与 WAL 模式（空库、无 schema）：并发首开的建库/journal 阶段
	// 可即时 SQLITE_BUSY 且不受 busy_timeout 保护（慢速盘 CI 实测），属连接
	// 建立竞争而非迁移互斥；KTD8 要验证的是「已部署库上的并发迁移者」。
	pre, err := sql.Open("sqlite", fmt.Sprintf("file:%s?_pragma=busy_timeout(30000)&_pragma=journal_mode(WAL)", path))
	if err != nil {
		t.Fatal(err)
	}
	if _, err := pre.Exec(`SELECT 1`); err != nil {
		pre.Close()
		t.Fatalf("materialize wal: %v", err)
	}
	pre.Close()
	var wg sync.WaitGroup
	errs := make([]error, 4)
	for i := range errs {
		wg.Add(1)
		go func(i int) {
			defer wg.Done()
			st, err := OpenSQLite(path)
			if err != nil {
				errs[i] = err
				return
			}
			defer st.Close()
			if vers, verr := st.MigrationStatus(); verr != nil {
				errs[i] = verr
			} else if len(vers) != len(migrations) {
				errs[i] = fmt.Errorf("ledger=%v, want %d versions", vers, len(migrations))
			}
		}(i)
	}
	wg.Wait()
	for i, err := range errs {
		if err != nil {
			t.Fatalf("migrator %d: %v", i, err)
		}
	}
	st := openSQLiteAt(t, path)
	assertFullLedger(t, st)
}

// 并发迁移者竞争同一 PG schema：会话级咨询锁串行化，账本恰好一套。
func TestMigrateRacingPostgres(t *testing.T) {
	dsn := os.Getenv("AGENT_PG_TEST_DSN")
	if dsn == "" {
		t.Skip("AGENT_PG_TEST_DSN 未设置：真实 PG 门禁由 CI 的必跑 job 覆盖")
	}
	schema := "t_race_" + strings.ToLower(newID("m")) + "_" + fmt.Sprint(os.Getpid())
	admin, err := sql.Open("pgx", dsn)
	if err != nil {
		t.Fatal(err)
	}
	if _, err := admin.Exec("CREATE SCHEMA " + schema); err != nil {
		admin.Close()
		t.Fatalf("create schema: %v", err)
	}
	t.Cleanup(func() {
		admin.Exec("DROP SCHEMA IF EXISTS " + schema + " CASCADE")
		admin.Close()
	})
	schemaDSN := withTestSearchPath(dsn, schema)
	var wg sync.WaitGroup
	errs := make([]error, 3)
	for i := range errs {
		wg.Add(1)
		go func(i int) {
			defer wg.Done()
			st, err := OpenPostgres(schemaDSN, filepath.Join(t.TempDir(), "artifacts"))
			if err != nil {
				errs[i] = err
				return
			}
			defer st.Close()
			if vers, verr := st.MigrationStatus(); verr != nil {
				errs[i] = verr
			} else if len(vers) != len(migrations) {
				errs[i] = fmt.Errorf("ledger=%v, want %d versions", vers, len(migrations))
			}
		}(i)
	}
	wg.Wait()
	for i, err := range errs {
		if err != nil {
			t.Fatalf("migrator %d: %v", i, err)
		}
	}
	// 等锁者事后看到完整账本（不因竞争出现重复/部分版本）。
	st, err := OpenPostgres(schemaDSN, filepath.Join(t.TempDir(), "artifacts"))
	if err != nil {
		t.Fatal(err)
	}
	defer st.Close()
	assertFullLedger(t, st)
}

// PG 不可达：打开失败（team 档 R2 的 store 层基线——不存在任何回落路径）。
func TestOpenPostgresUnreachableFails(t *testing.T) {
	dsn := "postgres://u:p@127.0.0.1:1/none?sslmode=disable&connect_timeout=1"
	if _, err := OpenPostgres(dsn, filepath.Join(t.TempDir(), "artifacts")); err == nil {
		t.Fatal("unreachable PG must fail to open")
	}
}

// 运行参数附着：URL 与键值两种 DSN 形式都正确追加 statement_timeout。
func TestWithPGRuntimeParam(t *testing.T) {
	url := withPGRuntimeParam("postgres://u:p@h:5432/db?sslmode=disable", "statement_timeout", "5000")
	if want := "postgres://u:p@h:5432/db?sslmode=disable&statement_timeout=5000"; url != want {
		t.Fatalf("url form: %q, want %q", url, want)
	}
	kv := withPGRuntimeParam("host=h port=5432 dbname=db", "statement_timeout", "5000")
	if want := "host=h port=5432 dbname=db statement_timeout=5000"; kv != want {
		t.Fatalf("kv form: %q, want %q", kv, want)
	}
	empty := withPGRuntimeParam("", "statement_timeout", "5000")
	if want := "statement_timeout=5000"; empty != want {
		t.Fatalf("empty dsn: %q, want %q", empty, want)
	}
}

// 连接预算有效值：零值落默认，显式值生效。
func TestPoolOptionsEffective(t *testing.T) {
	got := PoolOptions{}.effective()
	if got != DefaultPGPool {
		t.Fatalf("zero pool should fall back to default, got %+v", got)
	}
	custom := PoolOptions{MaxOpenConn: 4, MaxIdleConn: 2, ConnMaxIdleTimeSec: 60, ConnMaxLifetimeSec: 3600, QueryTimeoutSec: 5}.effective()
	if custom.MaxOpenConn != 4 || custom.MaxIdleConn != 2 || custom.ConnMaxIdleTimeSec != 60 || custom.ConnMaxLifetimeSec != 3600 || custom.QueryTimeoutSec != 5 {
		t.Fatalf("custom pool mangled: %+v", custom)
	}
}