diff --git a/cmd/storage-migrate/integration_test.go b/cmd/storage-migrate/integration_test.go new file mode 100644 index 0000000..d4997c0 --- /dev/null +++ b/cmd/storage-migrate/integration_test.go @@ -0,0 +1,188 @@ +package main + +import ( + "database/sql" + "fmt" + "os" + "path/filepath" + "strings" + "testing" + + "agent-system-backend/internal/store" +) + +// U4 集成(环境门控 AGENT_PG_TEST_DSN:CI test-pg job / 本机隔离实例): +// 夹具 → 受管迁移 → 校验和验证 → 幂等重跑 → 断点续传 → 非空目标拒绝 → +// 未知 owner 阻塞。每用例独立 PG schema(与 store 测试同隔离约定)。 + +func pgSchemaDSN(t *testing.T) (string, func()) { + t.Helper() + dsn := os.Getenv("AGENT_PG_TEST_DSN") + if dsn == "" { + t.Skip("AGENT_PG_TEST_DSN 未设置:真库集成由 CI 的必跑 job 覆盖") + } + schema := "mig_" + strings.ToLower(store.NewID("t")) + "_" + fmt.Sprint(os.Getpid()) + admin, err := sql.Open("pgx", dsn) + if err != nil { + t.Fatal(err) + } + if _, err := admin.Exec(`CREATE EXTENSION IF NOT EXISTS pg_trgm`); err != nil { + admin.Close() + t.Fatalf("preinstall pg_trgm: %v", err) + } + if _, err := admin.Exec("CREATE SCHEMA " + schema); err != nil { + admin.Close() + t.Fatalf("create schema: %v", err) + } + cleanup := func() { + admin.Exec("DROP SCHEMA IF EXISTS " + schema + " CASCADE") + admin.Close() + } + return store.AppendPGRuntimeParam(dsn, "search_path", schema+",public"), cleanup +} + +func TestMigrateEndToEndAndIdempotent(t *testing.T) { + target, cleanup := pgSchemaDSN(t) + defer cleanup() + fixture := buildFixture(t) + + if err := cmdMigrate([]string{"--source", fixture, "--target", target, "--migration-id", "mig-e2e"}); err != nil { + t.Fatalf("首次迁移:%v", err) + } + // 幂等重跑(同身份):表级 done 对账,零重复行。 + if err := cmdMigrate([]string{"--source", fixture, "--target", target, "--migration-id", "mig-e2e"}); err != nil { + t.Fatalf("幂等重跑:%v", err) + } + // 目标侧关键断言:布尔映射、NULL 保留、Unicode、BLOB、SERIAL 序列推进、 + // 复合主键事件序。 + st, err := store.OpenPostgres(target, filepath.Join(t.TempDir(), "artifacts")) + if err != nil { + t.Fatal(err) + } + defer st.Close() + db := st.DB() + var pinned, archived bool + if err := db.QueryRow(`SELECT pinned, archived FROM conversations WHERE id='c1'`).Scan(&pinned, &archived); err != nil || !pinned || archived { + t.Fatalf("布尔映射:pinned=%v archived=%v err=%v", pinned, archived, err) + } + var title, note string + if err := db.QueryRow(`SELECT title, session_note FROM conversations WHERE id='c1'`).Scan(&title, ¬e); err != nil || !strings.Contains(note, "注") { + t.Fatalf("Unicode:title=%q note=%q err=%v", title, note, err) + } + var vec []byte + if err := db.QueryRow(`SELECT vec FROM knowledge_vectors WHERE doc_id='k1'`).Scan(&vec); err != nil || len(vec) != 12 { + t.Fatalf("BLOB 向量:len=%d err=%v", len(vec), err) + } + var ended sql.NullString + if err := db.QueryRow(`SELECT ended_at FROM runs WHERE id='r1'`).Scan(&ended); err != nil || ended.Valid { + t.Fatalf("NULL 保留:ended=%v err=%v", ended, err) + } + var seqMax int + if err := db.QueryRow(`SELECT MAX(seq) FROM events WHERE conversation_id='c1'`).Scan(&seqMax); err != nil || seqMax != 2 { + t.Fatalf("事件复合主键序:max=%d err=%v", seqMax, err) + } + // 复制的显式 id 之后,应用侧插入不撞序列(command_audit.id=7 已复制)。 + if _, err := db.Exec(`INSERT INTO command_audit (time, command_id, source, entrance, state, operation_id, conversation_id, payload, owner) VALUES ('2026-10-11T03:00:00Z','cmd-2','http','chat','done','op-2','c1','{}','alice')`); err != nil { + t.Fatalf("序列推进失败(复制后插入撞主键):%v", err) + } + // 报告不含秘密值。 + var report string + if err := db.QueryRow(`SELECT report FROM migration_control WHERE migration_id='mig-e2e'`).Scan(&report); err != nil || strings.Contains(report, "sk-secret-value") { + t.Fatalf("报告泄密:err=%v", err) + } +} + +func TestMigrateResumeAfterInterrupt(t *testing.T) { + target, cleanup := pgSchemaDSN(t) + defer cleanup() + fixture := buildFixture(t) + + if err := cmdMigrate([]string{"--source", fixture, "--target", target, "--migration-id", "mig-resume"}); err != nil { + t.Fatal(err) + } + // 模拟断点:messages 已 done 后被部分回退(行删一半、状态回 pending), + // 重跑只重做该表,终态校验仍全绿。 + db, err := sql.Open("pgx", target) + if err != nil { + t.Fatal(err) + } + defer db.Close() + if _, err := db.Exec(`DELETE FROM messages WHERE id='m3'`); err != nil { + t.Fatal(err) + } + if _, err := db.Exec(`UPDATE migration_table_state SET state='pending' WHERE migration_id='mig-resume' AND table_name='messages'`); err != nil { + t.Fatal(err) + } + if err := cmdMigrate([]string{"--source", fixture, "--target", target, "--migration-id", "mig-resume"}); err != nil { + t.Fatalf("断点续跑:%v", err) + } + var n int + if err := db.QueryRow(`SELECT COUNT(*) FROM messages`).Scan(&n); err != nil || n != 3 { + t.Fatalf("续跑后 messages=%d:重做该表应恢复 3 行", n) + } +} + +func TestMigrateRefusesForeignDataAndWrongSnapshot(t *testing.T) { + target, cleanup := pgSchemaDSN(t) + defer cleanup() + fixture := buildFixture(t) + + // 非空目标(人工数据、无迁移身份):拒绝。 + st, err := store.OpenPostgres(target, filepath.Join(t.TempDir(), "artifacts")) + if err != nil { + t.Fatal(err) + } + st.Close() + db, err := sql.Open("pgx", target) + if err != nil { + t.Fatal(err) + } + defer db.Close() + if _, err := db.Exec(`INSERT INTO projects (id, name, settings, created_at, updated_at, owner) VALUES ('foreign', '外来', '{}', '2026-10-11T00:00:00Z', '2026-10-11T00:00:00Z', 'alice')`); err != nil { + t.Fatal(err) + } + err = cmdMigrate([]string{"--source", fixture, "--target", target, "--migration-id", "mig-x"}) + if err == nil || !strings.Contains(err.Error(), "非空目标") { + t.Fatalf("非空目标应拒绝:%v", err) + } + if _, err := db.Exec(`DELETE FROM projects WHERE id='foreign'`); err != nil { + t.Fatal(err) + } + + // 同 migration_id 换快照:拒绝(仅同一冻结快照可续传)。 + if err := cmdMigrate([]string{"--source", fixture, "--target", target, "--migration-id", "mig-snap"}); err != nil { + t.Fatal(err) + } + other := buildFixture(t) // 不同文件 ⇒ 不同 snapshot_id + err = cmdMigrate([]string{"--source", other, "--target", target, "--migration-id", "mig-snap"}) + if err == nil || !strings.Contains(err.Error(), "快照不同") { + t.Fatalf("换快照应拒绝:%v", err) + } + + // 其他 migration_id 已存在:拒绝。 + err = cmdMigrate([]string{"--source", other, "--target", target, "--migration-id", "mig-other"}) + if err == nil || !strings.Contains(err.Error(), "其他迁移记录") { + t.Fatalf("他身份应拒绝:%v", err) + } +} + +func TestMigrateBlocksUnknownOwner(t *testing.T) { + target, cleanup := pgSchemaDSN(t) + defer cleanup() + fixture := buildFixture(t) + // 在夹具里造一个 users 之外的 owner 行(直接写源库)。 + srcPath := fixture + st, err := store.OpenSQLite(srcPath) // 普通打开(夹具可写) + if err != nil { + t.Fatal(err) + } + if _, err := st.DB().Exec(`INSERT INTO knowledge (id, project_id, source, title, ref, chunk, created_at, owner) VALUES + ('k_ghost', 'p1', 'upload', '幽灵', 'g.md', '…', '2026-10-11T01:00:00Z', 'ghost-user')`); err != nil { + t.Fatal(err) + } + st.Close() + err = cmdMigrate([]string{"--source", srcPath, "--target", target, "--migration-id", "mig-ghost"}) + if err == nil || !strings.Contains(err.Error(), "未知 owner") { + t.Fatalf("未知 owner 应阻塞:%v", err) + } +} diff --git a/cmd/storage-migrate/main.go b/cmd/storage-migrate/main.go new file mode 100644 index 0000000..1016faa --- /dev/null +++ b/cmd/storage-migrate/main.go @@ -0,0 +1,852 @@ +// storage-migrate:SQLite → PostgreSQL 受管迁移入口(协作存储方案 U4/R7)。 +// +// 纪律(方案 §5 U4 / §6): +// - 只处理 Backend 库;不执行 SQLite dump 作为 PG SQL(按行复制 + 类型映射); +// - 迁移身份(migration_id + source_snapshot_id + source_schema_version)落 +// 目标库 migration_control:仅同一冻结快照可续传;目标已含其他数据即拒绝; +// - 报告只含计数、校验和与归属检查结论——不含 DSN、密钥与行正文(KTD14)。 +// +// 用法: +// +// storage-migrate inventory --source agent-system.db [--artifacts dir] [--out report.json] +// storage-migrate backup --source agent-system.db --out snapshot.db +// storage-migrate migrate --source snapshot.db --target "postgres://…" \ +// --migration-id <名> [--artifacts dir] [--out report.json] +package main + +import ( + "crypto/sha256" + "database/sql" + "encoding/base64" + "encoding/hex" + "encoding/json" + "flag" + "fmt" + "io" + "os" + "path/filepath" + "sort" + "strings" + "time" + + _ "modernc.org/sqlite" // SQLite 驱动(源,只读打开) + + "agent-system-backend/internal/store" +) + +func main() { + if len(os.Args) < 2 { + usage() + os.Exit(2) + } + var err error + switch os.Args[1] { + case "inventory": + err = cmdInventory(os.Args[2:]) + case "backup": + err = cmdBackup(os.Args[2:]) + case "migrate": + err = cmdMigrate(os.Args[2:]) + default: + usage() + os.Exit(2) + } + if err != nil { + fmt.Fprintln(os.Stderr, "storage-migrate:", err) + os.Exit(1) + } +} + +func usage() { + fmt.Fprintln(os.Stderr, "用法:storage-migrate [flags](-h 看各子命令 flags)") +} + +// ---- 公共小件 ---- + +func openSourceRO(path string) (*sql.DB, error) { + if _, err := os.Stat(path); err != nil { + return nil, fmt.Errorf("source %s: %w", path, err) + } + // query_only:迁移源只读(冻结快照,任何写入都是缺陷)。 + db, err := sql.Open("sqlite", fmt.Sprintf("file:%s?_pragma=query_only(1)", path)) + if err != nil { + return nil, err + } + if err := db.Ping(); err != nil { + db.Close() + return nil, fmt.Errorf("source %s: %w", path, err) + } + return db, nil +} + +func fileSHA256(path string) (string, int64, error) { + f, err := os.Open(path) + if err != nil { + return "", 0, err + } + defer f.Close() + h := sha256.New() + n, err := io.Copy(h, f) + if err != nil { + return "", 0, err + } + return hex.EncodeToString(h.Sum(nil)), n, nil +} + +func nowISO() string { return time.Now().UTC().Format(time.RFC3339Nano) } + +// ---- inventory:只读盘点(§6.1) ---- + +func cmdInventory(args []string) error { + fs := flag.NewFlagSet("inventory", flag.ExitOnError) + source := fs.String("source", "", "SQLite 库路径(活跃库或冻结快照)") + artifacts := fs.String("artifacts", "", "工件对象目录(可选;盘点文件清单与哈希)") + out := fs.String("out", "", "报告输出 JSON 路径(缺省打印到 stdout)") + fs.Parse(args) + if *source == "" { + return fmt.Errorf("--source 必填") + } + db, err := openSourceRO(*source) + if err != nil { + return err + } + defer db.Close() + rep, err := inventoryReport(db, *source, *artifacts) + if err != nil { + return err + } + return writeReport(rep, *out) +} + +func inventoryReport(db *sql.DB, source, artifactsDir string) (map[string]any, error) { + version, err := sourceSchemaVersion(db) + if err != nil { + return nil, err + } + var mode string + if err := db.QueryRow(`PRAGMA journal_mode`).Scan(&mode); err != nil { + return nil, fmt.Errorf("journal_mode: %w", err) + } + var chk string + if err := db.QueryRow(`PRAGMA integrity_check`).Scan(&chk); err != nil || chk != "ok" { + return nil, fmt.Errorf("integrity_check: %v (%q)", err, chk) + } + tables, err := sqliteTableCounts(db) + if err != nil { + return nil, err + } + snap, size, err := fileSHA256(source) + if err != nil { + return nil, err + } + rep := map[string]any{ + "command": "inventory", "source": source, "source_sha256": snap, + "source_bytes": size, "source_schema_version": version, + "journal_mode": mode, "integrity": chk, "table_counts": tables, + "generated_at": nowISO(), + } + if artifactsDir != "" { + manifest, err := artifactManifest(artifactsDir) + if err != nil { + return nil, err + } + rep["artifacts"] = manifest + } + return rep, nil +} + +func sourceSchemaVersion(db *sql.DB) (int, error) { + var v int + err := db.QueryRow(`SELECT COALESCE(MAX(version),0) FROM schema_migrations`).Scan(&v) + if err != nil { + if strings.Contains(err.Error(), "no such table") { + return 0, fmt.Errorf("源库缺 schema_migrations(不是本工具管理的库):%w", err) + } + return 0, err + } + return v, nil +} + +// sqliteTableCounts:仅盘点本工具认识的业务表(见 copyOrder)。 +func sqliteTableCounts(db *sql.DB) (map[string]int64, error) { + out := map[string]int64{} + for _, t := range copyOrder { + var n int64 + if err := db.QueryRow(`SELECT COUNT(*) FROM "` + t + `"`).Scan(&n); err != nil { + if strings.Contains(err.Error(), "no such table") { + continue // 旧库可能没有新表 + } + return nil, fmt.Errorf("count %s: %w", t, err) + } + out[t] = n + } + return out, nil +} + +func artifactManifest(dir string) (map[string]any, error) { + type entry struct { + Name, SHA string + Size int64 + } + var entries []entry + hAll := sha256.New() + err := filepath.Walk(dir, func(p string, info os.FileInfo, err error) error { + if err != nil || info.IsDir() { + return err + } + sum, _, err := fileSHA256(p) + if err != nil { + return err + } + rel, _ := filepath.Rel(dir, p) + entries = append(entries, entry{rel, sum, info.Size()}) + fmt.Fprintf(hAll, "%s %d %s\n", rel, info.Size(), sum) + return nil + }) + if err != nil { + return nil, fmt.Errorf("artifacts %s: %w", dir, err) + } + sort.Slice(entries, func(i, j int) bool { return entries[i].Name < entries[j].Name }) + return map[string]any{ + "dir": dir, "files": entries, "file_count": len(entries), + "manifest_sha256": hex.EncodeToString(hAll.Sum(nil)), + }, nil +} + +// ---- backup:一致性快照(§6.2:VACUUM INTO + 可打开校验 + 计数对账) ---- + +func cmdBackup(args []string) error { + fs := flag.NewFlagSet("backup", flag.ExitOnError) + source := fs.String("source", "", "活跃 SQLite 库路径") + out := fs.String("out", "", "快照输出路径(不得与源同文件;已存在则拒绝)") + fs.Parse(args) + if *source == "" || *out == "" { + return fmt.Errorf("--source 与 --out 必填") + } + if same, _ := filepath.EvalSymlinks(*source); same != "" { + if abs1, _ := filepath.Abs(same); abs1 != "" { + if abs2, _ := filepath.Abs(*out); abs2 == abs1 { + return fmt.Errorf("快照输出不得与源同文件") + } + } + } + if _, err := os.Stat(*out); err == nil { + return fmt.Errorf("快照输出已存在(不覆盖):%s", *out) + } + if err := os.MkdirAll(filepath.Dir(*out), 0o755); err != nil { + return err + } + // VACUUM INTO 在线安全(含 WAL 内容);query_only 会禁写,需普通打开。 + db, err := sql.Open("sqlite", fmt.Sprintf("file:%s?_pragma=busy_timeout(30000)", *source)) + if err != nil { + return err + } + defer db.Close() + if _, err := db.Exec(`VACUUM INTO ?`, *out); err != nil { + return fmt.Errorf("vacuum into: %w", err) + } + // 校验:快照可打开、完整性、逐表计数与源一致。 + snap, err := openSourceRO(*out) + if err != nil { + return err + } + defer snap.Close() + var chk string + if err := snap.QueryRow(`PRAGMA integrity_check`).Scan(&chk); err != nil || chk != "ok" { + return fmt.Errorf("快照完整性校验失败:%v (%q)", err, chk) + } + srcCounts, err := sqliteTableCounts(db) + if err != nil { + return err + } + snapCounts, err := sqliteTableCounts(snap) + if err != nil { + return err + } + for t, n := range srcCounts { + if snapCounts[t] != n { + return fmt.Errorf("快照计数不符:%s 源=%d 快照=%d", t, n, snapCounts[t]) + } + } + sum, size, err := fileSHA256(*out) + if err != nil { + return err + } + fmt.Printf("backup ok: %s(%d 字节,sha256 %s…,%d 表计数一致)\n", *out, size, sum[:12], len(srcCounts)) + return nil +} + +// ---- migrate:受管迁移(§6.3 + U4) ---- + +// copyOrder:按外键依赖排序的业务表全集(projects 最先,引用方在后)。 +// admission_leases 是运行态执行许可(迁移无意义),明确排除。 +var copyOrder = []string{ + "projects", "users", "profiles", "settings", "plugins", + "conversations", "runs", "messages", "events", "artifacts", + "hitl_cards", "plans", "todos", "knowledge", "knowledge_vectors", + "activity_cards", "command_audit", "prompt_modules", + "sessions", "api_keys", + "plugin_grants", "plugin_schema", "plugin_kv", "background_tasks", + "memory_entries", "audit_events", "pending_confirms", "llm_usage_daily", + "secret_keys", "settings_snapshots", "regen_pending", "context_entries", + "user_commands", "pending_notices", "recovery_holds", +} + +const controlDDL = ` +CREATE TABLE IF NOT EXISTS migration_control ( + migration_id TEXT PRIMARY KEY, + source_snapshot_id TEXT NOT NULL, + source_schema_version INTEGER NOT NULL, + started_at TEXT NOT NULL, + finished_at TEXT, + verified_at TEXT, + report TEXT +); +CREATE TABLE IF NOT EXISTS migration_table_state ( + migration_id TEXT NOT NULL, + table_name TEXT NOT NULL, + source_count INTEGER NOT NULL, + target_count INTEGER NOT NULL, + checksum TEXT NOT NULL, + state TEXT NOT NULL, + finished_at TEXT, + PRIMARY KEY (migration_id, table_name) +);` + +func cmdMigrate(args []string) error { + fs := flag.NewFlagSet("migrate", flag.ExitOnError) + source := fs.String("source", "", "冻结快照路径(backup 子命令产物;不接受活跃库)") + target := fs.String("target", "", "目标 PostgreSQL DSN(须为空库或同身份续传库)") + migrationID := fs.String("migration-id", "", "迁移实例名(人工指定,报告与续传身份的一部分)") + artifacts := fs.String("artifacts", "", "工件对象目录(可选;仅盘点清单入报告,不复制)") + out := fs.String("out", "", "报告输出 JSON 路径(缺省打印到 stdout)") + fs.Parse(args) + if *source == "" || *target == "" || *migrationID == "" { + return fmt.Errorf("--source、--target、--migration-id 必填") + } + src, err := openSourceRO(*source) + if err != nil { + return err + } + defer src.Close() + snapID, _, err := fileSHA256(*source) + if err != nil { + return err + } + srcVersion, err := sourceSchemaVersion(src) + if err != nil { + return err + } + + // 目标:先用 store 同款迁移器建 schema/扩展(OpenPostgres 会跑迁移与 + // 未知版本门禁),再落控制表。DSN 不回显、不入报告。 + st, err := store.OpenPostgres(*target, filepath.Join(filepath.Dir(*source), "migrate-artifacts")) + if err != nil { + return fmt.Errorf("open target: %w", err) + } + defer st.Close() + db := st.DB() + if _, err := db.Exec(controlDDL); err != nil { + return fmt.Errorf("control tables: %w", err) + } + if err := guardIdentity(db, *migrationID, snapID, srcVersion); err != nil { + return err + } + if err := ensureControlRow(db, *migrationID, snapID, srcVersion); err != nil { + return err + } + + rep := map[string]any{ + "command": "migrate", "migration_id": *migrationID, + "source_snapshot_id": snapID, "source_schema_version": srcVersion, + "started_at": nowISO(), + } + var tables []map[string]any + for _, t := range copyOrder { + if !sqliteHasTable(src, t) { + continue // 旧库无此表 + } + entry, err := migrateTable(db, src, *migrationID, t) + if err != nil { + return fmt.Errorf("table %s: %w", t, err) + } + tables = append(tables, entry) + } + // 未知 owner 检查(U4:阻塞,不静默)。'' 与 'local' 是单租户默认身份。 + if err := checkOwners(db, src); err != nil { + return err + } + // 校验:目标逐表重算计数与校验和,全部一致才落 verified。 + if err := verifyTables(db, *migrationID, tables); err != nil { + return err + } + var totalSrc, totalDst int64 + for _, e := range tables { + totalSrc += e["source_count"].(int64) + totalDst += e["target_count"].(int64) + } + rep["tables"] = tables + rep["rows_total"] = totalSrc + rep["verified_at"] = nowISO() + if *artifacts != "" { + manifest, err := artifactManifest(*artifacts) + if err != nil { + return err + } + rep["artifacts"] = manifest + } + repJSON, _ := json.Marshal(rep) + if _, err := db.Exec(`UPDATE migration_control SET finished_at=$1, verified_at=$2, report=$3 WHERE migration_id=$4`, + nowISO(), rep["verified_at"], string(repJSON), *migrationID); err != nil { + return err + } + return writeReport(rep, *out) +} + +// guardIdentity:目标非空且身份不同即拒绝(R7:不覆盖他人数据)。 +// - 存在其他 migration_id 的完成记录 → 拒绝; +// - 同 migration_id 但 source_snapshot_id 不同(换了快照续传)→ 拒绝; +// - 业务表已有数据而控制表为空(人工造过数据)→ 拒绝。 +func guardIdentity(db *sql.DB, migrationID, snapID string, srcVersion int) error { + var n int + if err := db.QueryRow(`SELECT COUNT(*) FROM migration_control WHERE migration_id<>$1`, migrationID).Scan(&n); err != nil { + return err + } + if n > 0 { + return fmt.Errorf("目标库已有其他迁移记录(%d 条):拒绝覆盖,换空库或专项回迁方案", n) + } + var snap, started string + var sv int + err := db.QueryRow(`SELECT source_snapshot_id, source_schema_version, started_at FROM migration_control WHERE migration_id=$1`, migrationID).Scan(&snap, &sv, &started) + switch { + case err == sql.ErrNoRows: + // 新迁移:业务表必须为空。 + busy, err := businessRows(db) + if err != nil { + return err + } + if busy > 0 { + return fmt.Errorf("目标库业务表已有 %d 行且无迁移身份:拒绝(非空目标不得作为迁移目标)", busy) + } + case err != nil: + return err + default: + if snap != snapID { + return fmt.Errorf("同 migration_id 但快照不同(库=%s… 本次=%s…):仅同一冻结快照可续传", snap[:12], snapID[:12]) + } + _ = sv + _ = started + } + return nil +} + +func businessRows(db *sql.DB) (int64, error) { + var total int64 + for _, t := range copyOrder { + var n int64 + if err := db.QueryRow(`SELECT COUNT(*) FROM "` + t + `"`).Scan(&n); err != nil { + if strings.Contains(err.Error(), "does not exist") { + continue + } + return 0, err + } + total += n + } + return total, nil +} + +func ensureControlRow(db *sql.DB, migrationID, snapID string, srcVersion int) error { + _, err := db.Exec(`INSERT INTO migration_control (migration_id, source_snapshot_id, source_schema_version, started_at) + VALUES ($1,$2,$3,$4) + ON CONFLICT(migration_id) DO NOTHING`, migrationID, snapID, srcVersion, nowISO()) + return err +} + +func sqliteHasTable(db *sql.DB, name string) bool { + var n int + if err := db.QueryRow(`SELECT COUNT(*) FROM sqlite_master WHERE type='table' AND name=?`, name).Scan(&n); err != nil { + return false + } + return n > 0 +} + +// migrateTable:done 则对账(幂等重跑);pending 则按行复制(布尔映射 + +// 批量 INSERT + 序列推进)后落 done 与源侧校验和。 +func migrateTable(db *sql.DB, src *sql.DB, migrationID, table string) (map[string]any, error) { + cols, pk, err := targetColumns(db, table) + if err != nil { + return nil, err + } + var state string + var srcCount, dstCountStored int64 + var srcChecksum string + err = db.QueryRow(`SELECT state, source_count, target_count, checksum FROM migration_table_state WHERE migration_id=$1 AND table_name=$2`, + migrationID, table).Scan(&state, &srcCount, &dstCountStored, &srcChecksum) + if err != nil && err != sql.ErrNoRows { + return nil, err + } + if err == sql.ErrNoRows { + state = "pending" + } + // 源侧计数与校验和始终重算(冻结快照上必然稳定;差异=快照被换,拒绝)。 + bools, err := boolColumns(db, table, cols) + if err != nil { + return nil, err + } + countNow, sumNow, err := sourceScan(src, table, cols, pk, bools) + if err != nil { + return nil, err + } + if state == "done" { + if countNow != srcCount || sumNow != srcChecksum { + return nil, fmt.Errorf("源与登记不符(疑似换快照):count %d/%d checksum %s…/%s…", countNow, srcCount, sumNow[:12], srcChecksum[:12]) + } + entry := map[string]any{"table": table, "source_count": countNow, "target_count": dstCountStored, "checksum": sumNow, "state": "done(idempotent)"} + return entry, nil + } + if err := copyRows(db, src, table, cols); err != nil { + return nil, err + } + dstCount, _, err := targetScan(db, table, cols, pk, bools) + if err != nil { + return nil, err + } + if dstCount != countNow { + return nil, fmt.Errorf("复制后计数不符:源=%d 目标=%d", countNow, dstCount) + } + if err := advanceSequences(db, table); err != nil { + return nil, err + } + if _, err := db.Exec(`INSERT INTO migration_table_state (migration_id, table_name, source_count, target_count, checksum, state, finished_at) + VALUES ($1,$2,$3,$4,$5,$6,$7) + ON CONFLICT(migration_id, table_name) DO UPDATE SET source_count=excluded.source_count, target_count=excluded.target_count, checksum=excluded.checksum, state=excluded.state, finished_at=excluded.finished_at`, + migrationID, table, countNow, dstCount, sumNow, "done", nowISO()); err != nil { + return nil, err + } + return map[string]any{"table": table, "source_count": countNow, "target_count": dstCount, "checksum": sumNow, "state": "done"}, nil +} + +// targetColumns:目标侧列清单(按名排序)+ 主键列。布尔列集合同取自目标 +// information_schema(类型映射以 PG schema 为准——SQLite 侧 0/1 → bool)。 +func targetColumns(db *sql.DB, table string) (cols []string, pk []string, err error) { + // 生成列(PG GENERATED ALWAYS,如 knowledge.tsv)不复制不校验:PG 从 + // 业务列自动派生,源库也没有该列。 + rows, err := db.Query(` + SELECT column_name, data_type FROM information_schema.columns + WHERE table_schema=current_schema() AND table_name=$1 AND is_generated='NEVER' ORDER BY column_name`, table) + if err != nil { + return nil, nil, err + } + defer rows.Close() + exists := false + for rows.Next() { + var name, dtype string + if err := rows.Scan(&name, &dtype); err != nil { + return nil, nil, err + } + exists = true + cols = append(cols, name) + } + if !exists { + return nil, nil, fmt.Errorf("目标缺表 %s(迁移器未建?)", table) + } + prows, err := db.Query(` + SELECT a.attname FROM pg_index i + JOIN pg_attribute a ON a.attrelid=i.indrelid AND a.attnum=ANY(i.indkey) + WHERE i.indisprimary AND i.indrelid=$1::regclass`, table) + if err != nil { + return nil, nil, err + } + defer prows.Close() + for prows.Next() { + var p string + if err := prows.Scan(&p); err != nil { + return nil, nil, err + } + pk = append(pk, p) + } + if len(pk) == 0 { + pk = cols // 无主键表按全列序校验 + } + return cols, pk, nil +} + +func boolColumns(db *sql.DB, table string, cols []string) (map[string]bool, error) { + out := map[string]bool{} + rows, err := db.Query(` + SELECT column_name FROM information_schema.columns + WHERE table_schema=current_schema() AND table_name=$1 AND data_type='boolean'`, table) + if err != nil { + return nil, err + } + defer rows.Close() + for rows.Next() { + var c string + if err := rows.Scan(&c); err != nil { + return nil, err + } + out[c] = true + } + return out, nil +} + +// canonicalRow:稳定行编码(校验和用):列名排序、值规范化 +// (bool→true/false、[]byte→base64、NULL→null、数值/字符串原样)。 +func canonicalRow(cols []string, vals []any, bools map[string]bool) string { + type kv struct { + K string + V any + } + pairs := make([]kv, len(cols)) + for i := range cols { + pairs[i] = kv{cols[i], normalize(vals[i], bools[cols[i]])} + } + b, _ := json.Marshal(pairs) + return string(b) +} + +// normalize:布尔列按目标 schema 对称归一(源 int64 0/1、目标 bool 编码一致, +// 否则校验和必然不符——真库实测);[]byte→base64。 +func normalize(v any, isBool bool) any { + if isBool { + return sqliteBool(v) + } + switch t := v.(type) { + case []byte: + return base64.StdEncoding.EncodeToString(t) + default: + return v + } +} + +// hashChain:按主键序哈希链(前后衔接,顺序敏感)。 +func hashChain(prev string, rowJSON string) string { + h := sha256.Sum256([]byte(prev + "|" + rowJSON)) + return hex.EncodeToString(h[:]) +} + +// scanChecksum:源/目标共用的扫描对账(列清单一致、主键序、行哈希链)。 +func scanChecksum(rows *sql.Rows, cols []string, bools map[string]bool) (int64, string, error) { + vals := make([]any, len(cols)) + ptrs := make([]any, len(cols)) + for i := range vals { + ptrs[i] = &vals[i] + } + var count int64 + sum := "" + for rows.Next() { + if err := rows.Scan(ptrs...); err != nil { + return 0, "", err + } + sum = hashChain(sum, canonicalRow(cols, vals, bools)) + count++ + } + return count, sum, rows.Err() +} + +func sourceScan(src *sql.DB, table string, cols, pk []string, bools map[string]bool) (int64, string, error) { + orderBy := "ORDER BY " + quoteJoin(pk, `"`) + rows, err := src.Query(`SELECT ` + quoteJoin(cols, `"`) + ` FROM "` + table + `" ` + orderBy) + if err != nil { + return 0, "", err + } + defer rows.Close() + return scanChecksum(rows, cols, bools) +} + +func targetScan(db *sql.DB, table string, cols, pk []string, bools map[string]bool) (int64, string, error) { + rows, err := db.Query(`SELECT ` + quoteJoin(cols, `"`) + ` FROM "` + table + `" ORDER BY ` + quoteJoin(pk, `"`)) + if err != nil { + return 0, "", err + } + defer rows.Close() + return scanChecksum(rows, cols, bools) +} + +func quoteJoin(names []string, q string) string { + out := make([]string, len(names)) + for i, n := range names { + out[i] = q + n + q + } + return strings.Join(out, ",") +} + +// copyRows:按行复制。布尔列 0/1→false/true;其余原样(TEXT/INT/REAL/BLOB)。 +func copyRows(db *sql.DB, src *sql.DB, table string, cols []string) error { + bools, err := boolColumns(db, table, cols) + if err != nil { + return err + } + colList := quoteJoin(cols, `"`) + ph := make([]string, len(cols)) + for i := range ph { + ph[i] = fmt.Sprintf("$%d", i+1) + } + placeholders := strings.Join(ph, ",") + srcRows, err := src.Query(`SELECT ` + colList + ` FROM "` + table + `"`) + if err != nil { + return err + } + defer srcRows.Close() + vals := make([]any, len(cols)) + ptrs := make([]any, len(cols)) + for i := range vals { + ptrs[i] = &vals[i] + } + tx, err := db.Begin() + if err != nil { + return err + } + defer tx.Rollback() + stmt, err := tx.Prepare(fmt.Sprintf(`INSERT INTO "%s" (%s) VALUES (%s) ON CONFLICT DO NOTHING`, table, colList, placeholders)) + if err != nil { + return err + } + defer stmt.Close() + var batch int + for srcRows.Next() { + if err := srcRows.Scan(ptrs...); err != nil { + return err + } + args := make([]any, len(cols)) + for i, c := range cols { + v := vals[i] + if bools[c] { + args[i] = sqliteBool(v) + } else { + args[i] = v + } + } + if _, err := stmt.Exec(args...); err != nil { + return fmt.Errorf("insert: %w", err) + } + batch++ + } + if err := srcRows.Err(); err != nil { + return err + } + return tx.Commit() +} + +func sqliteBool(v any) bool { + switch t := v.(type) { + case int64: + return t != 0 + case bool: + return t + default: + return false + } +} + +// advanceSequences:SERIAL/identity 列复制后推进序列(显式插入不推进, +// 后续应用插入会撞主键)。无序列表为空操作。 +func advanceSequences(db *sql.DB, table string) error { + rows, err := db.Query(` + SELECT pg_get_serial_sequence($1, column_name) + FROM information_schema.columns + WHERE table_schema=current_schema() AND table_name=$1 + AND (column_default LIKE 'nextval%' OR is_identity='YES')`, table) + if err != nil { + return err + } + defer rows.Close() + var seqs []string + for rows.Next() { + var s any + if err := rows.Scan(&s); err != nil { + return err + } + if s != nil { + seqs = append(seqs, s.(string)) + } + } + if err := rows.Err(); err != nil { + return err + } + for _, s := range seqs { + if _, err := db.Exec(fmt.Sprintf(`SELECT setval('%s', COALESCE((SELECT MAX(id) FROM "%s"), 1))`, s, table)); err != nil { + return fmt.Errorf("setval %s: %w", s, err) + } + } + return nil +} + +// checkOwners:业务行引用的 owner 必须存在于 users(单租户默认 ”/'local' +// 除外)。未知 owner 阻塞迁移(U4/R7:不静默带过未知归属)。 +func checkOwners(db *sql.DB, src *sql.DB) error { + ownerTables := []string{"projects", "conversations", "runs", "artifacts", "knowledge", "hitl_cards", "profiles", "activity_cards", "command_audit"} + for _, t := range ownerTables { + if !sqliteHasTable(src, t) { + continue + } + rows, err := src.Query(`SELECT DISTINCT owner FROM "` + t + `" WHERE owner<>'' AND owner<>'local'`) + if err != nil { + return err + } + var unknown []string + for rows.Next() { + var o string + if err := rows.Scan(&o); err != nil { + rows.Close() + return err + } + var n int + if err := db.QueryRow(`SELECT COUNT(*) FROM users WHERE owner=$1`, o).Scan(&n); err != nil { + rows.Close() + return err + } + if n == 0 { + unknown = append(unknown, o) + } + } + rows.Close() + if err := rows.Err(); err != nil { + return err + } + if len(unknown) > 0 { + return fmt.Errorf("表 %s 引用未知 owner(users 无此行):%v——先在目标补建用户或修正归属,再迁移", t, unknown) + } + } + return nil +} + +// verifyTables:目标逐表重算,与登记的源计数/校验和全部一致才通过。 +func verifyTables(db *sql.DB, migrationID string, tables []map[string]any) error { + for _, e := range tables { + table := e["table"].(string) + wantCount := e["source_count"].(int64) + wantSum := e["checksum"].(string) + cols, pk, err := targetColumns(db, table) + if err != nil { + return err + } + bools, err := boolColumns(db, table, cols) + if err != nil { + return err + } + gotCount, gotSum, err := targetScan(db, table, cols, pk, bools) + if err != nil { + return err + } + if gotCount != wantCount || gotSum != wantSum { + return fmt.Errorf("表 %s 校验不符:count %d/%d checksum %s…/%s…", table, gotCount, wantCount, gotSum[:12], wantSum[:12]) + } + } + return nil +} + +func writeReport(rep map[string]any, out string) error { + b, err := json.MarshalIndent(rep, "", " ") + if err != nil { + return err + } + if out == "" { + fmt.Println(string(b)) + return nil + } + if err := os.MkdirAll(filepath.Dir(out), 0o755); err != nil { + return err + } + return os.WriteFile(out, append(b, '\n'), 0o644) +} diff --git a/cmd/storage-migrate/main_test.go b/cmd/storage-migrate/main_test.go new file mode 100644 index 0000000..b327cbd --- /dev/null +++ b/cmd/storage-migrate/main_test.go @@ -0,0 +1,193 @@ +package main + +import ( + "encoding/json" + "path/filepath" + "strings" + "testing" + + "agent-system-backend/internal/store" +) + +// U4 迁移工具测试:夹具覆盖全部当前 schema 的边值(布尔/空值/时间/JSON/ +// Unicode/二进制向量/秘密字段),以及 inventory/backup 的行为面。 +// 真库端到端(幂等/续传/非空拒/未知 owner)在 integration_test.go(PG 门控)。 + +// buildFixture:建一个全 schema 的 SQLite 库并填充边值行,返回路径。 +func buildFixture(t *testing.T) string { + t.Helper() + dir := t.TempDir() + path := filepath.Join(dir, "fixture.db") + st, err := store.OpenSQLite(path) + if err != nil { + t.Fatal(err) + } + defer st.Close() + seedEdgeRows(t, st) + return path +} + +func seedEdgeRows(t *testing.T, st *store.Store) { + t.Helper() + // users(含 is_admin/must_change 布尔)与单租户默认身份。 + mustExec(t, st, `INSERT INTO users (id, owner, name, created_at, password_hash, password_set_at, must_change_password, is_admin) VALUES + ('u1', 'alice', '爱丽丝', '2026-10-11T01:00:00Z', 'sha256:…', '2026-10-11T01:00:00Z', false, true), + ('u2', 'bob', 'Bob 😊', '2026-10-11T01:00:00Z', '', '', true, false)`) + // 项目与会话(布尔 pinned/archived/allow_cross_read/cross_private、空 JSON)。 + mustExec(t, st, `INSERT INTO projects (id, name, settings, created_at, updated_at, owner) VALUES + ('p1', '项目一', '{}', '2026-10-11T01:00:00Z', '2026-10-11T01:00:00Z', 'alice'), + ('p2', 'Emoji 项目 🚀', '{"a":null,"b":[1,2,"三"]}', '2026-10-11T02:00:00Z', '2026-10-11T02:00:00Z', '')`) + mustExec(t, st, `INSERT INTO conversations (id, project_id, title, profile_id, created_at, updated_at, owner, pinned, archived, allow_cross_read, cross_private, permission_mode, session_note) VALUES + ('c1', 'p1', '会话', '', '2026-10-11T01:00:00Z', '2026-10-11T01:00:00Z', 'alice', true, false, false, true, '', 'note—注'), + ('c2', 'p2', '', '', '2026-10-11T02:00:00Z', '2026-10-11T02:00:00Z', '', false, true, true, false, 'strict', '')`) + // 消息(request_id 幂等、parts JSON 含空对象/空数组/Unicode)与 seq 顺序。 + mustExec(t, st, `INSERT INTO messages (id, conversation_id, run_id, role, parts, seq, created_at, request_id) VALUES + ('m1', 'c1', '', 'user', '[{"type":"text","text":"你好"}]', 1, '2026-10-11T01:00:01Z', 'req-1'), + ('m2', 'c1', '', 'assistant', '[{"type":"kv","kv":[]},{"type":"text","text":""}]', 2, '2026-10-11T01:00:02Z', ''), + ('m3', 'c2', '', 'system', '[]', 1, '2026-10-11T02:00:01Z', 'req-2')`) + // runs(NULL ended_at/usage/error)。 + mustExec(t, st, `INSERT INTO runs (id, parent_run_id, conversation_id, project_id, owner, kind, profile_id, status, phase, agent_name, created_at, instance_id) VALUES + ('r1', '', 'c1', 'p1', 'alice', 'conversation', '', 'completed', '', '', '2026-10-11T01:00:00Z', 'host:1'), + ('r2', 'r1', 'c1', 'p1', 'alice', 'subagent', '', 'running', 'thinking', '检索', '2026-10-11T01:00:03Z', 'host:2')`) + // events(复合主键 (conversation_id, seq))。 + mustExec(t, st, `INSERT INTO events (conversation_id, seq, run_id, ts, v, type, payload) VALUES + ('c1', 1, 'r1', '2026-10-11T01:00:01Z', 1, 'run.phase', '{"phase":"thinking"}'), + ('c1', 2, 'r1', '2026-10-11T01:00:02Z', 1, 'message.appended', NULL), + ('c2', 1, '', '2026-10-11T02:00:01Z', 1, 'x', '{}')`) + // 工件(size 0 与大值)。 + mustExec(t, st, `INSERT INTO artifacts (id, project_id, name, kind, mime, size, source_run, content_path, created_at, updated_at, owner) VALUES + ('a1', 'p1', '报告.md', 'doc', 'text/markdown', 0, 'r1', 'art/a1', '2026-10-11T01:00:00Z', '2026-10-11T01:00:00Z', 'alice'), + ('a2', 'p2', '数据.xlsx', 'doc', '', 2147483647, '', '', '2026-10-11T02:00:00Z', '2026-10-11T02:00:00Z', '')`) + // HITL 卡(resolved=0、resolution NULL)。 + mustExec(t, st, `INSERT INTO hitl_cards (id, run_id, conversation_id, payload, created_at, resolved, resolution, owner) VALUES + ('h1', 'r2', 'c1', '{"q":"继续?"}', '2026-10-11T01:00:04Z', 0, NULL, 'alice')`) + // 知识 + 向量(BLOB 二进制:float32 little-endian)。 + mustExec(t, st, `INSERT INTO knowledge (id, project_id, source, title, ref, chunk, created_at, owner) VALUES + ('k1', 'p1', 'upload', '标题', 'doc.md', '正文内容…', '2026-10-11T01:00:00Z', 'alice')`) + mustExec(t, st, `INSERT INTO knowledge_vectors (doc_id, owner, ref, dim, vec, created_at) VALUES + ('k1', 'alice', 'doc.md', 3, x'0000803F0000003F000080BF', '2026-10-11T01:00:00Z')`) + // 活动卡(pinned 布尔)。 + mustExec(t, st, `INSERT INTO activity_cards (id, project_id, conversation_id, ref, owner, payload, status, pinned, created_at, updated_at, version) VALUES + ('ac1', 'p1', 'c1', '', 'alice', '{"title":"卡"}', 'active', 1, '2026-10-11T01:00:00Z', '2026-10-11T01:00:00Z', 2)`) + // 命令审计(SERIAL 主键:显式 id 插入)。 + mustExec(t, st, `INSERT INTO command_audit (id, time, command_id, source, entrance, state, operation_id, conversation_id, payload, owner) VALUES + (7, '2026-10-11T01:00:00Z', 'cmd-1', 'http', 'chat', 'done', 'op-1', 'c1', '{"actor":"alice"}', 'alice')`) + // 设置('' 默认租户 scope)。 + mustExec(t, st, `INSERT INTO settings (owner, scope, "values") VALUES ('', 'runtime', '{"rateLimitPerMin":0}')`) + // 秘密字段(值不外泄:迁移按行复制,报告无正文)。 + mustExec(t, st, `INSERT INTO secret_keys (id, owner, service, label, value, active, created_at) VALUES + ('sk1', 'alice', 'openai', 'main', 'sk-secret-value-不打印', 1, '2026-10-11T01:00:00Z')`) + // 恢复隔离(业务态,应随迁)。 + mustExec(t, st, `INSERT INTO recovery_holds (run_id, owner, conversation_id, reason, created_at) VALUES + ('rh1', 'alice', 'c1', '副作用未知', '2026-10-11T01:00:00Z')`) +} + +func must(t *testing.T, err error) { + t.Helper() + if err != nil { + t.Fatal(err) + } +} + +// mustExec:夹具直接走底层连接写行(业务代码不得如此;见 store.DB 注释)。 +func mustExec(t *testing.T, st *store.Store, q string) { + t.Helper() + if _, err := st.DB().Exec(q); err != nil { + t.Fatal(err) + } +} + +// ---- inventory ---- + +func TestInventoryReport(t *testing.T) { + path := buildFixture(t) + db, err := openSourceRO(path) + if err != nil { + t.Fatal(err) + } + defer db.Close() + rep, err := inventoryReport(db, path, "") + if err != nil { + t.Fatal(err) + } + if rep["source_schema_version"].(int) < 20 { + t.Fatalf("schema version: %v", rep["source_schema_version"]) + } + counts := rep["table_counts"].(map[string]int64) + if counts["messages"] != 3 || counts["events"] != 3 || counts["users"] != 2 { + t.Fatalf("counts: %v", counts) + } + b, _ := json.Marshal(rep) + if strings.Contains(string(b), "sk-secret-value") { + t.Fatal("inventory 报告不得包含秘密值") + } +} + +func TestInventoryMissingFile(t *testing.T) { + if _, err := openSourceRO(filepath.Join(t.TempDir(), "nope.db")); err == nil { + t.Fatal("缺文件应报错") + } +} + +// ---- backup ---- + +func TestBackupVerifiesAndSnapshotsConsistent(t *testing.T) { + path := buildFixture(t) + out := filepath.Join(t.TempDir(), "snap.db") + if err := cmdBackup([]string{"--source", path, "--out", out}); err != nil { + t.Fatal(err) + } + snap, err := openSourceRO(out) + if err != nil { + t.Fatal(err) + } + defer snap.Close() + var n int + if err := snap.QueryRow(`SELECT COUNT(*) FROM messages`).Scan(&n); err != nil || n != 3 { + t.Fatalf("snap messages: %d %v", n, err) + } + // 已存在的输出拒绝覆盖。 + if err := cmdBackup([]string{"--source", path, "--out", out}); err == nil { + t.Fatal("已存在快照应拒绝") + } +} + +// ---- 夹具即文档:确认全部业务表在 copyOrder 内(防新表漏迁) ---- + +func TestCopyOrderCoversAllFixtureTables(t *testing.T) { + path := buildFixture(t) + db, err := openSourceRO(path) + if err != nil { + t.Fatal(err) + } + defer db.Close() + rows, err := db.Query(`SELECT name FROM sqlite_master WHERE type='table' AND name NOT LIKE 'sqlite_%'`) + if err != nil { + t.Fatal(err) + } + defer rows.Close() + known := map[string]bool{} + for _, t := range copyOrder { + known[t] = true + } + known["schema_migrations"] = true // 迁移器管理,不随迁 + known["admission_leases"] = true // 运行态,明确排除(见 copyOrder 注释) + // FTS5 虚表影子表:SQLite 内部结构,PG 侧由 v4 生成列/索引从 knowledge + // 行自动重建,不随迁。 + for _, shadow := range []string{"knowledge_fts", "knowledge_fts_data", "knowledge_fts_idx", "knowledge_fts_docsize", "knowledge_fts_config"} { + known[shadow] = true + } + var missing []string + for rows.Next() { + var name string + if err := rows.Scan(&name); err != nil { + t.Fatal(err) + } + if !known[name] { + missing = append(missing, name) + } + } + if len(missing) > 0 { + t.Fatalf("copyOrder 漏表(新表加入 schema 后须同步工具清单):%v", missing) + } +} diff --git a/internal/store/store.go b/internal/store/store.go index 1afefe4..a1e6bf8 100644 --- a/internal/store/store.go +++ b/internal/store/store.go @@ -209,6 +209,10 @@ 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 } diff --git a/storage-migrate b/storage-migrate new file mode 100755 index 0000000..285283f Binary files /dev/null and b/storage-migrate differ