From fd2810bf03f4e4f7b4dd25f36d28848aca185d22 Mon Sep 17 00:00:00 2001 From: xuans Date: Mon, 28 Sep 2026 01:52:37 +0800 Subject: [PATCH] =?UTF-8?q?=E4=BF=AE=E5=A4=8DSQLite=E5=8E=86=E5=8F=B2?= =?UTF-8?q?=E4=B8=9A=E5=8A=A1=E6=95=B0=E6=8D=AE=E5=8D=87=E7=BA=A7=E5=B9=B6?= =?UTF-8?q?=E5=A2=9E=E5=8A=A0=E7=8B=AC=E7=AB=8B=E7=A8=8B=E5=BA=8F=E5=9B=9E?= =?UTF-8?q?=E5=BD=92?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../cli/sqlite_business_fixture_test.go | 18 +- .../internal/cli/sqlite_upgrade_cli_test.go | 229 ++++++++++++++++++ backend/internal/database/migrate.go | 3 + .../database/migration_history_test.go | 6 +- .../database/sqlite_business_upgrade_test.go | 207 ++++++++++++++++ .../database/sqlite_task_migration.go | 183 ++++++++++++++ .../database/sqlite_task_migration_test.go | 122 ++++++++++ 7 files changed, 762 insertions(+), 6 deletions(-) create mode 100644 backend/internal/cli/sqlite_upgrade_cli_test.go create mode 100644 backend/internal/database/sqlite_business_upgrade_test.go create mode 100644 backend/internal/database/sqlite_task_migration.go create mode 100644 backend/internal/database/sqlite_task_migration_test.go diff --git a/backend/internal/cli/sqlite_business_fixture_test.go b/backend/internal/cli/sqlite_business_fixture_test.go index bd2b1e4..bbbfbc8 100644 --- a/backend/internal/cli/sqlite_business_fixture_test.go +++ b/backend/internal/cli/sqlite_business_fixture_test.go @@ -31,10 +31,17 @@ type sqliteBusinessFixture struct { certificate certificate.SavedResource } -func newSQLiteBusinessFixture(t *testing.T) sqliteBusinessFixture { +func newSQLiteBusinessAssets(t *testing.T) sqliteBusinessFixture { t.Helper() - fixture := sqliteBusinessFixture{config: config.Default(t.TempDir()), password: rand.Text()} + root := t.TempDir() + fixture := sqliteBusinessFixture{config: config.Default(root), password: rand.Text()} cfg := &fixture.config + // 明确覆盖全部路径，Linux 测试也不能读写正式安装目录。 + cfg.Paths = config.PathsConfig{ + Install: root, Config: filepath.Join(root, "conf"), Data: filepath.Join(root, "data"), + Certificate: filepath.Join(root, "ssl"), Log: filepath.Join(root, "log"), Backup: filepath.Join(root, "backup"), + } + cfg.Database.DSN = filepath.Join(cfg.Paths.Data, "scert.db") for _, directory := range []string{cfg.Paths.Config, cfg.Paths.Data, cfg.Paths.Certificate, cfg.Paths.Backup} { if err := os.MkdirAll(directory, 0o700); err != nil { t.Fatal(err) @@ -69,6 +76,13 @@ func newSQLiteBusinessFixture(t *testing.T) sqliteBusinessFixture { } } fixture.certificate = saveSQLiteBusinessCertificate(t, *cfg) + return fixture +} + +func newSQLiteBusinessFixture(t *testing.T) sqliteBusinessFixture { + t.Helper() + fixture := newSQLiteBusinessAssets(t) + cfg := &fixture.config hash, err := bcrypt.GenerateFromPassword([]byte(fixture.password), bcrypt.MinCost) if err != nil { t.Fatal(err) diff --git a/backend/internal/cli/sqlite_upgrade_cli_test.go b/backend/internal/cli/sqlite_upgrade_cli_test.go new file mode 100644 index 0000000..2bb31ef --- /dev/null +++ b/backend/internal/cli/sqlite_upgrade_cli_test.go @@ -0,0 +1,229 @@ +package cli + +import ( + "context" + "crypto/tls" + "os" + "os/exec" + "path/filepath" + "reflect" + "strings" + "testing" + "time" + + "git.pcux.cn/xuans/Scert/backend/internal/config" + "git.pcux.cn/xuans/Scert/backend/internal/database" + "git.pcux.cn/xuans/Scert/backend/internal/repository" + "git.pcux.cn/xuans/Scert/backend/internal/secretstore" + "golang.org/x/crypto/bcrypt" +) + +// 显式指定当批构建的产品程序；不能用测试子进程或辅助脚本冒充独立 CLI。 +func TestSQLiteBusinessUpgradeStandaloneCLI(t *testing.T) { + binary := os.Getenv("SCERT_SQLITE_TEST_CLI") + if binary == "" { + t.Skip("未指定本批构建的 Scert 独立程序") + } + if !filepath.IsAbs(binary) || !isRegularFile(binary) { + t.Fatal("独立 CLI 测试要求有效的产品程序绝对路径") + } + fixture := newSQLiteBusinessAssets(t) + cfg := &fixture.config + hash, err := bcrypt.GenerateFromPassword([]byte(fixture.password), bcrypt.MinCost) + if err != nil { + t.Fatal(err) + } + cfg.SetupReady = true + cfg.Admin = config.AdminConfig{Username: "upgrade-owner", Email: "owner@example.test", PasswordHash: string(hash)} + cfg.DNS.OwnerUserID = "primary-super-admin" + // 仅构造隔离历史配置，不把预置状态当成真实 DNS 或 Web 初始化证据。 + cfg.DNS.VerificationState = "verified" + cfg.Web.Hostname = "example.test" + if err := cfg.Save(fixture.path); err != nil { + t.Fatal(err) + } + db, err := database.Open(t.Context(), cfg.Database) + if err != nil { + t.Fatal(err) + } + defer db.Close() + if _, err := db.DB.ExecContext(t.Context(), "CREATE TABLE schema_migrations (version INTEGER PRIMARY KEY, applied_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP)"); err != nil { + t.Fatal(err) + } + for index, name := range []string{"0001_initial.sql", "0002_domain_options.sql", "0003_deployment.sql"} { + content, err := os.ReadFile(filepath.Join("..", "database", "migrations", "sqlite", name)) + if err != nil { + t.Fatal(err) + } + if _, err := db.DB.ExecContext(t.Context(), strings.TrimPrefix(string(content), "\ufeff")); err != nil { + t.Fatal(err) + } + if _, err := db.DB.ExecContext(t.Context(), "INSERT INTO schema_migrations (version) VALUES (?)", index+1); err != nil { + t.Fatal(err) + } + } + for _, statement := range []string{ + `INSERT INTO settings (key, value) VALUES ('historical', '历史设置')`, + `INSERT INTO domains (id, name, certificate_type, sans_json, status, dns_provider) VALUES ('domain', 'example.test', 'single', '["example.test"]', 'issued', 'alidns')`, + `INSERT INTO tasks (id, domain_id, task_type, status, message) VALUES ('task', 'domain', 'deploy', 'success', '历史任务')`, + `INSERT INTO servers (id, name, address, port, username, private_key_ref, host_key_fingerprint, certificate_path, private_key_path, certificate_chain_path, web_server, reload_command_json) VALUES ('server', '历史服务器', '192.0.2.10', 22, 'deploy', 'secret://ssh/business', 'test-only', '/test/cert.pem', '/test/key.pem', '/test/chain.pem', 'nginx', '["nginx","-s","reload"]')`, + `INSERT INTO domain_servers (domain_id, server_id) VALUES ('domain', 'server')`, + `INSERT INTO audit_logs (action, resource_type, resource_id, status, message) VALUES ('deploy', 'domain', 'domain', 'success', '历史审计')`, + } { + if _, err := db.DB.ExecContext(t.Context(), statement); err != nil { + t.Fatal(err) + } + } + if err := db.Close(); err != nil { + t.Fatal(err) + } + historical := snapshotSQLiteCLIState(t, *cfg) + archive := filepath.Join(t.TempDir(), "historical-backup") + runSQLiteProductCLI(t, binary, fixture, "backup", "--output", archive) + runSQLiteProductCLI(t, binary, fixture, "certificate", "list") + assertSQLiteCLIUpgradeUsable(t, fixture) + upgraded := snapshotSQLiteCLIState(t, *cfg) + currentArchive := filepath.Join(t.TempDir(), "upgraded-backup") + runSQLiteProductCLI(t, binary, fixture, "backup", "--output", currentArchive) + + for _, scenario := range []struct { + name string + archive string + want map[string]string + }{ + {"historical_backup", archive, historical}, + {"upgraded_backup", currentArchive, upgraded}, + } { + t.Run(scenario.name, func(t *testing.T) { + current, err := database.Open(t.Context(), cfg.Database) + if err != nil { + t.Fatal(err) + } + if _, err := current.DB.ExecContext(t.Context(), "UPDATE tasks SET message = '不应保留'; DELETE FROM domain_servers"); err != nil { + _ = current.Close() + t.Fatal(err) + } + if err := current.Close(); err != nil { + t.Fatal(err) + } + if err := os.WriteFile(fixture.certificate.CertificatePath, []byte("待恢复证书"), 0o600); err != nil { + t.Fatal(err) + } + store, err := secretstore.Open(cfg.Paths.Config) + if err != nil { + t.Fatal(err) + } + if err := store.Put(t.Context(), cfg.DNS.CredentialRef, map[string]string{"changed": "待恢复"}); err != nil { + t.Fatal(err) + } + runSQLiteProductCLI(t, binary, fixture, "restore", "--input", scenario.archive) + if actual := snapshotSQLiteCLIState(t, *cfg); !reflect.DeepEqual(actual, scenario.want) { + t.Fatal("独立 CLI 未恢复全部原始模式和表内容") + } + // 从旧备份恢复后再次走正式应用入口升级，不由测试直接调用 Migrate。 + runSQLiteProductCLI(t, binary, fixture, "certificate", "list") + assertSQLiteCLIUpgradeUsable(t, fixture) + if err := validateBackupManifest(scenario.archive); err != nil { + t.Fatal(err) + } + }) + } +} + +func runSQLiteProductCLI(t *testing.T, binary string, fixture sqliteBusinessFixture, args ...string) { + t.Helper() + ctx, cancel := context.WithTimeout(t.Context(), 30*time.Second) + defer cancel() + cmd := exec.CommandContext(ctx, binary, append([]string{"--config", fixture.path}, args...)...) + cmd.Dir = fixture.config.Paths.Install + cmd.Env = make([]string, 0) + for _, entry := range os.Environ() { + if !strings.HasPrefix(strings.ToUpper(entry), "SCERT_") { + cmd.Env = append(cmd.Env, entry) + } + } + cmd.WaitDelay = 2 * time.Second + output, err := cmd.CombinedOutput() + for _, values := range fixture.secrets { + for _, value := range values { + if strings.Contains(string(output), value) { + t.Fatal("独立 CLI 输出包含敏感凭据") + } + } + } + if strings.Contains(string(output), fixture.password) { + t.Fatal("独立 CLI 输出包含用户密码") + } + if err != nil { + t.Fatalf("独立 CLI %s 失败：%v：%s", args[0], err, output) + } +} + +func snapshotSQLiteCLIState(t *testing.T, cfg config.Config) map[string]string { + t.Helper() + db, err := database.Open(t.Context(), cfg.Database) + if err != nil { + t.Fatal(err) + } + defer db.Close() + result := map[string]string{"schema": snapshotSQLiteRows(t, db.DB, "SELECT type, name, tbl_name, sql FROM sqlite_master ORDER BY type, name")} + rows, err := db.DB.QueryContext(t.Context(), "SELECT name FROM sqlite_master WHERE type = 'table' ORDER BY name") + if err != nil { + t.Fatal(err) + } + var tables []string + for rows.Next() { + var name string + if err := rows.Scan(&name); err != nil { + t.Fatal(err) + } + tables = append(tables, name) + } + if err := rows.Err(); err != nil { + t.Fatal(err) + } + if err := rows.Close(); err != nil { + t.Fatal(err) + } + for _, name := range tables { + result[name] = snapshotSQLiteRows(t, db.DB, `SELECT * FROM "`+strings.ReplaceAll(name, `"`, `""`)+`" ORDER BY 1`) + } + return result +} + +func assertSQLiteCLIUpgradeUsable(t *testing.T, fixture sqliteBusinessFixture) { + t.Helper() + db, err := database.Open(t.Context(), fixture.config.Database) + if err != nil { + t.Fatal(err) + } + defer db.Close() + var version, bindings int + if err := db.DB.QueryRowContext(t.Context(), "SELECT MAX(version) FROM schema_migrations").Scan(&version); err != nil || version != 9 { + t.Fatalf("正式应用入口未完成历史升级：%d，%v", version, err) + } + task, err := repository.NewSQLTaskRepository(db.DB).GetTask(t.Context(), "task") + if err != nil || task.Message != "历史任务" || task.DomainID != "domain" || task.Status != "success" { + t.Fatalf("历史任务不可用：%v", err) + } + if err := db.DB.QueryRowContext(t.Context(), "SELECT COUNT(*) FROM domain_servers WHERE domain_id = 'domain' AND server_id = 'server'").Scan(&bindings); err != nil || bindings != 1 { + t.Fatalf("历史部署绑定丢失：%v", err) + } + user, err := repository.NewSQLUserRepository(db.DB).FindByIdentifier(t.Context(), "upgrade-owner") + if err != nil || bcrypt.CompareHashAndPassword([]byte(user.PasswordHash), []byte(fixture.password)) != nil { + t.Fatalf("升级后用户密码不可用：%v", err) + } + store, err := secretstore.Open(fixture.config.Paths.Config) + if err != nil { + t.Fatal(err) + } + for reference, want := range fixture.secrets { + actual, err := store.Get(t.Context(), reference) + if err != nil || !reflect.DeepEqual(actual, want) { + t.Fatalf("升级恢复后加密凭据不一致：%v", err) + } + } + if _, err := tls.LoadX509KeyPair(fixture.certificate.CertificatePath, fixture.certificate.PrivateKeyPath); err != nil { + t.Fatalf("升级恢复后证书与私钥不可用：%v", err) + } +} diff --git a/backend/internal/database/migrate.go b/backend/internal/database/migrate.go index cec56fc..b972af2 100644 --- a/backend/internal/database/migrate.go +++ b/backend/internal/database/migrate.go @@ -71,6 +71,9 @@ func applyMigration(ctx context.Context, db *sql.DB, driver string, version int) return fmt.Errorf("读取迁移文件失败：%w", err) } statements := migrationStatements(content) + if driver == "sqlite" && version == 4 { + return applySQLiteTaskMigration(ctx, db, statements, version) + } if driver == "mysql" { for _, statement := range statements { skip, err := shouldSkipMySQLStatement(ctx, db, statement) diff --git a/backend/internal/database/migration_history_test.go b/backend/internal/database/migration_history_test.go index 33052b7..56bc157 100644 --- a/backend/internal/database/migration_history_test.go +++ b/backend/internal/database/migration_history_test.go @@ -4,8 +4,6 @@ import ( "fmt" "strings" "testing" - - "git.pcux.cn/xuans/Scert/backend/internal/config" ) func TestMigrateRejectsInvalidHistoryWithoutChangingDatabase(t *testing.T) { @@ -26,7 +24,7 @@ func TestMigrateRejectsInvalidHistoryWithoutChangingDatabase(t *testing.T) { } for _, tc := range cases { t.Run(tc.name, func(t *testing.T) { - db, err := Open(t.Context(), config.Default(t.TempDir()).Database) + db, err := Open(t.Context(), sqliteBusinessUpgradeConfig(t)) if err != nil { t.Fatal(err) } @@ -73,7 +71,7 @@ func TestMigrateRejectsInvalidHistoryWithoutChangingDatabase(t *testing.T) { func TestSQLiteUpgradeFromEverySupportedVersion(t *testing.T) { for version := 0; version <= latestMigrationVersion; version++ { t.Run(fmt.Sprintf("version_%d", version), func(t *testing.T) { - db, err := Open(t.Context(), config.Default(t.TempDir()).Database) + db, err := Open(t.Context(), sqliteBusinessUpgradeConfig(t)) if err != nil { t.Fatal(err) } diff --git a/backend/internal/database/sqlite_business_upgrade_test.go b/backend/internal/database/sqlite_business_upgrade_test.go new file mode 100644 index 0000000..28e8467 --- /dev/null +++ b/backend/internal/database/sqlite_business_upgrade_test.go @@ -0,0 +1,207 @@ +package database + +import ( + "database/sql" + "encoding/json" + "fmt" + "path/filepath" + "reflect" + "strings" + "testing" + "time" + + "git.pcux.cn/xuans/Scert/backend/internal/config" +) + +// 历史版本必须先写入业务数据；空表迁移不能证明真实升级可用。 +func TestSQLiteBusinessUpgradeFromEveryVersion(t *testing.T) { + for version := 1; version <= latestMigrationVersion; version++ { + t.Run(fmt.Sprintf("version_%d", version), func(t *testing.T) { + cfg := sqliteBusinessUpgradeConfig(t) + db, err := Open(t.Context(), cfg) + if err != nil { + t.Fatal(err) + } + defer db.Close() + createSQLiteHistoricalBusiness(t, db.DB, version) + before := snapshotSQLiteHistoricalBusiness(t, db.DB) + if err := db.Close(); err != nil { + t.Fatal(err) + } + db, err = Open(t.Context(), cfg) + if err != nil { + t.Fatal(err) + } + defer db.Close() + for round := 0; round < 2; round++ { + if err := Migrate(t.Context(), db.DB, "sqlite"); err != nil { + t.Fatalf("历史版本 %d 第 %d 次升级失败：%v", version, round+1, err) + } + assertSQLiteHistoricalBusiness(t, db.DB, before) + } + var history int + if err := db.DB.QueryRowContext(t.Context(), "SELECT COUNT(*) FROM schema_migrations").Scan(&history); err != nil || history != latestMigrationVersion { + t.Fatalf("升级后的迁移历史不完整：%d，%v", history, err) + } + var timestamp time.Time + if err := db.DB.QueryRowContext(t.Context(), "SELECT updated_at FROM tasks WHERE id = 'task'").Scan(×tamp); err != nil || timestamp.IsZero() { + t.Fatalf("历史任务更新时间不可用：%v", err) + } + if _, err := db.DB.ExecContext(t.Context(), `INSERT INTO tasks (id, domain_id, task_type, status, message) VALUES ('new-task', 'domain', 'deploy', 'queued', '升级后任务')`); err != nil { + t.Fatal(err) + } + if err := db.DB.QueryRowContext(t.Context(), "SELECT updated_at FROM tasks WHERE id = 'new-task'").Scan(×tamp); err != nil || time.Since(timestamp) < -time.Second || time.Since(timestamp) > time.Minute { + t.Fatalf("新任务必须保留当前时间默认值：%v", err) + } + result, err := db.DB.ExecContext(t.Context(), `INSERT INTO audit_logs (action, resource_type, resource_id, status, message) VALUES ('upgrade', 'task', 'new-task', 'success', '升级后审计')`) + if err != nil { + t.Fatal(err) + } + if id, err := result.LastInsertId(); err != nil || id != 18 { + t.Fatalf("审计自增序列错误：%d，%v", id, err) + } + var check string + if err := db.DB.QueryRowContext(t.Context(), "PRAGMA integrity_check").Scan(&check); err != nil || check != "ok" { + t.Fatalf("升级后完整性检查失败：%s，%v", check, err) + } + }) + } +} + +func sqliteBusinessUpgradeConfig(t *testing.T) config.DatabaseConfig { + t.Helper() + cfg := config.Default(t.TempDir()).Database + // Linux 默认安装路径不随 baseDir 改变，测试必须显式指定临时数据库。 + cfg.DSN = filepath.Join(t.TempDir(), "historical.db") + return cfg +} + +func createSQLiteHistoricalBusiness(t *testing.T, db *sql.DB, version int) { + t.Helper() + if _, err := db.ExecContext(t.Context(), migrationTableSQL("sqlite")); err != nil { + t.Fatal(err) + } + for next := 1; next <= version; next++ { + if err := applyMigration(t.Context(), db, "sqlite", next); err != nil { + t.Fatal(err) + } + } + statements := []string{ + `INSERT INTO settings (key, value) VALUES ('historical', '历史设置：测试')`, + `INSERT INTO domains (id, name, certificate_type, sans_json, status, expires_at) VALUES ('domain', 'example.test', 'single', '["example.test"]', 'issued', '2030-01-01 00:00:00')`, + `INSERT INTO tasks (id, domain_id, task_type, status, message, started_at, finished_at, created_at) VALUES ('task', 'domain', 'deploy', 'success', '历史部署任务', '2020-01-01 01:00:00', '2020-01-01 02:00:00', '2020-01-01 00:00:00')`, + `INSERT INTO tasks (id, domain_id, task_type, status, message) VALUES ('pending', 'domain', 'renew', 'queued', '历史待执行任务')`, + `INSERT INTO audit_logs (id, action, resource_type, resource_id, status, message) VALUES (17, 'deploy', 'domain', 'domain', 'success', '历史审计记录')`, + } + if version >= 2 { + statements = append(statements, `UPDATE domains SET auto_renew = 0, acme_environment = 'production', dns_provider = 'alidns'`) + } + if version >= 3 { + statements = append(statements, + `INSERT INTO servers (id, name, address, port, username, private_key_ref, host_key_fingerprint, certificate_path, private_key_path, certificate_chain_path, web_server, reload_command_json) VALUES ('server', '历史部署服务器', '192.0.2.10', 22, 'deploy', 'secret://ssh/historical', 'test-only', '/test/cert.pem', '/test/key.pem', '/test/chain.pem', 'nginx', '["nginx","-s","reload"]')`, + `INSERT INTO domain_servers (domain_id, server_id) VALUES ('domain', 'server')`) + } + if version >= 4 { + statements = append(statements, + `UPDATE tasks SET stage = 'completed', operator_id = 'owner', certificate_version_id = 'historical-certificate', idempotency_key = 'historical-operation' WHERE id = 'task'`, + `INSERT INTO task_attempts (id, task_id, attempt_number, stage, status, message) VALUES ('attempt', 'task', 1, 'deploy', 'success', '历史部署尝试')`) + } + if version >= 5 { + statements = append(statements, `INSERT INTO users (id, username, email, password_hash, role) VALUES ('owner', 'historical-owner', 'owner@example.test', 'historical-hash', 'super_admin')`) + } + if version >= 6 { + statements = append(statements, `UPDATE servers SET validation_command_json = '["nginx","-t"]'`) + } + if version >= 7 { + statements = append(statements, `UPDATE servers SET auth_method = 'password', password_ref = 'secret://ssh/password'`) + } + if version >= 8 { + statements = append(statements, `UPDATE users SET groups_json = '["历史分组"]'`) + } + if version >= 9 { + statements = append(statements, `UPDATE servers SET ssh_key_status = 'deployed', ssh_key_fingerprint = 'test-fingerprint', ssh_key_deployed_at = '2020-01-01T00:00:00Z'`) + } + for _, statement := range statements { + if _, err := db.ExecContext(t.Context(), statement); err != nil { + t.Fatal(err) + } + } +} + +type sqliteHistoricalTable struct { + query string + rows string +} + +func snapshotSQLiteHistoricalBusiness(t *testing.T, db *sql.DB) map[string]sqliteHistoricalTable { + t.Helper() + result := make(map[string]sqliteHistoricalTable) + for _, name := range []string{"settings", "domains", "tasks", "audit_logs", "servers", "domain_servers", "task_attempts", "users", "sqlite_sequence"} { + var exists int + if err := db.QueryRowContext(t.Context(), "SELECT COUNT(*) FROM sqlite_master WHERE type = 'table' AND name = ?", name).Scan(&exists); err != nil { + t.Fatal(err) + } + if exists == 0 { + continue + } + rows, err := db.QueryContext(t.Context(), `SELECT * FROM "`+name+`" LIMIT 0`) + if err != nil { + t.Fatal(err) + } + columns, err := rows.Columns() + if err != nil { + t.Fatal(err) + } + if err := rows.Close(); err != nil { + t.Fatal(err) + } + for index, column := range columns { + columns[index] = `"` + strings.ReplaceAll(column, `"`, `""`) + `"` + } + query := "SELECT " + strings.Join(columns, ",") + ` FROM "` + name + `" ORDER BY 1` + result[name] = sqliteHistoricalTable{query: query, rows: snapshotSQLiteUpgradeQuery(t, db, query)} + } + return result +} + +func snapshotSQLiteUpgradeQuery(t *testing.T, db *sql.DB, query string) string { + t.Helper() + rows, err := db.QueryContext(t.Context(), query) + if err != nil { + t.Fatal(err) + } + defer rows.Close() + columns, err := rows.Columns() + if err != nil { + t.Fatal(err) + } + var result [][]any + for rows.Next() { + values := make([]any, len(columns)) + destinations := make([]any, len(columns)) + for index := range values { + destinations[index] = &values[index] + } + if err := rows.Scan(destinations...); err != nil { + t.Fatal(err) + } + result = append(result, values) + } + if err := rows.Err(); err != nil { + t.Fatal(err) + } + encoded, err := json.Marshal(result) + if err != nil { + t.Fatal(err) + } + return string(encoded) +} + +func assertSQLiteHistoricalBusiness(t *testing.T, db *sql.DB, expected map[string]sqliteHistoricalTable) { + t.Helper() + for table, snapshot := range expected { + if actual := snapshotSQLiteUpgradeQuery(t, db, snapshot.query); !reflect.DeepEqual(actual, snapshot.rows) { + t.Errorf("历史表 %s 的原有业务字段发生变化", table) + } + } +} diff --git a/backend/internal/database/sqlite_task_migration.go b/backend/internal/database/sqlite_task_migration.go new file mode 100644 index 0000000..5cad01c --- /dev/null +++ b/backend/internal/database/sqlite_task_migration.go @@ -0,0 +1,183 @@ +package database + +import ( + "context" + "database/sql" + "database/sql/driver" + "errors" + "fmt" + "regexp" + "strings" + "time" +) + +// 固定物理连接后按 SQLite 表重建流程处理迁移，外键设置不能泄漏到连接池。 +func applySQLiteTaskMigration(ctx context.Context, db *sql.DB, statements []string, version int) (err error) { + conn, err := db.Conn(ctx) + if err != nil { + return fmt.Errorf("获取 SQLite 迁移连接失败：%w", err) + } + defer conn.Close() + var foreignKeys int + if err := conn.QueryRowContext(ctx, "PRAGMA foreign_keys").Scan(&foreignKeys); err != nil { + return fmt.Errorf("读取 SQLite 外键设置失败：%w", err) + } + defer func() { + cleanup, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + _, resetErr := conn.ExecContext(cleanup, fmt.Sprintf("PRAGMA foreign_keys = %d", foreignKeys)) + if resetErr != nil { + // 不能把外键状态未知的连接继续交给后续业务使用。 + _ = conn.Raw(func(any) error { return driver.ErrBadConn }) + err = errors.Join(err, fmt.Errorf("恢复 SQLite 外键设置失败：%w", resetErr)) + } + }() + if _, err := conn.ExecContext(ctx, "PRAGMA foreign_keys = OFF"); err != nil { + return fmt.Errorf("设置 SQLite 迁移连接失败：%w", err) + } + tx, err := conn.BeginTx(ctx, nil) + if err != nil { + return fmt.Errorf("开始 SQLite 迁移事务失败：%w", err) + } + defer tx.Rollback() + for _, statement := range statements { + if statement == "ALTER TABLE tasks ADD COLUMN updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP" { + err = rebuildSQLiteTasks(ctx, tx) + } else { + _, err = tx.ExecContext(ctx, statement) + } + if err != nil { + return fmt.Errorf("执行 SQLite 任务迁移失败：%w", err) + } + } + rows, err := tx.QueryContext(ctx, "PRAGMA foreign_key_check") + if err != nil { + return fmt.Errorf("检查 SQLite 迁移外键失败：%w", err) + } + invalid := rows.Next() + rowErr := errors.Join(rows.Err(), rows.Close()) + if rowErr != nil { + return fmt.Errorf("读取 SQLite 迁移外键结果失败：%w", rowErr) + } + if invalid { + return errors.New("SQLite 任务迁移存在外键冲突，已拒绝提交") + } + if _, err := tx.ExecContext(ctx, "INSERT INTO schema_migrations (version) VALUES (?)", version); err != nil { + return fmt.Errorf("记录 SQLite 迁移版本失败：%w", err) + } + if err := tx.Commit(); err != nil { + return fmt.Errorf("提交 SQLite 迁移事务失败：%w", err) + } + return nil +} + +var sqliteTaskTableHeader = regexp.MustCompile(`(?is)^CREATE\s+TABLE\s+(?:IF\s+NOT\s+EXISTS\s+)?(?:tasks|"tasks"|` + "`tasks`" + `|\[tasks\])\s*\(`) + +type sqliteTaskSchemaObject struct { + kind string + name string + sql string +} + +func rebuildSQLiteTasks(ctx context.Context, tx *sql.Tx) error { + var definition string + if err := tx.QueryRowContext(ctx, "SELECT sql FROM sqlite_master WHERE type = 'table' AND name = 'tasks'").Scan(&definition); err != nil { + return err + } + header := sqliteTaskTableHeader.FindStringIndex(definition) + end := strings.LastIndex(definition, ")") + if header == nil || end < header[1] { + return errors.New("无法安全识别 SQLite 任务表定义，拒绝重建") + } + // 保留原始列、约束和表选项，不用硬编码的旧表字段覆盖扩展结构。 + create := `CREATE TABLE "scert_tasks_migration_4" (` + definition[header[1]:end] + ", updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP" + definition[end:] + columns, err := sqliteTaskCopyColumns(ctx, tx) + if err != nil { + return err + } + rows, err := tx.QueryContext(ctx, `SELECT type, name, sql FROM sqlite_master WHERE sql IS NOT NULL AND (type IN ('view', 'trigger') OR (type = 'index' AND tbl_name = 'tasks')) ORDER BY type, name`) + if err != nil { + return err + } + var objects []sqliteTaskSchemaObject + for rows.Next() { + var object sqliteTaskSchemaObject + if err := rows.Scan(&object.kind, &object.name, &object.sql); err != nil { + _ = rows.Close() + return err + } + objects = append(objects, object) + } + if err := errors.Join(rows.Err(), rows.Close()); err != nil { + return err + } + // 名字冲突时直接失败，不覆盖或删除已有对象。 + if _, err := tx.ExecContext(ctx, create); err != nil { + return err + } + if _, err := tx.ExecContext(ctx, `INSERT INTO "scert_tasks_migration_4" (`+columns+`) SELECT `+columns+` FROM tasks`); err != nil { + return err + } + // 重命名会校验整个模式，临时移除依赖视图和触发器后按原定义恢复。 + for _, object := range objects { + if object.kind == "view" || object.kind == "trigger" { + if _, err := tx.ExecContext(ctx, "DROP "+strings.ToUpper(object.kind)+" "+quoteSQLiteMigrationName(object.name)); err != nil { + return err + } + } + } + for _, statement := range []string{"DROP TABLE tasks", `ALTER TABLE "scert_tasks_migration_4" RENAME TO tasks`} { + if _, err := tx.ExecContext(ctx, statement); err != nil { + return err + } + } + // 先恢复视图，最后恢复触发器，兼容视图上的 INSTEAD OF 触发器。 + for _, kind := range []string{"index", "view", "trigger"} { + for _, object := range objects { + if object.kind == kind { + if _, err := tx.ExecContext(ctx, object.sql); err != nil { + return err + } + } + } + } + return nil +} + +func sqliteTaskCopyColumns(ctx context.Context, tx *sql.Tx) (string, error) { + rows, err := tx.QueryContext(ctx, "SELECT name FROM pragma_table_xinfo('tasks') WHERE hidden = 0 ORDER BY cid") + if err != nil { + return "", err + } + var columns []string + names := make(map[string]bool) + for rows.Next() { + var name string + if err := rows.Scan(&name); err != nil { + _ = rows.Close() + return "", err + } + names[strings.ToLower(name)] = true + columns = append(columns, quoteSQLiteMigrationName(name)) + } + if err := errors.Join(rows.Err(), rows.Close()); err != nil { + return "", err + } + var withoutRowID int + if err := tx.QueryRowContext(ctx, "SELECT wr FROM pragma_table_list WHERE schema = 'main' AND name = 'tasks'").Scan(&withoutRowID); err != nil { + return "", err + } + if withoutRowID == 0 { + for _, alias := range []string{"rowid", "_rowid_", "oid"} { + if !names[alias] { + columns = append([]string{quoteSQLiteMigrationName(alias)}, columns...) + break + } + } + } + return strings.Join(columns, ","), nil +} + +func quoteSQLiteMigrationName(name string) string { + return `"` + strings.ReplaceAll(name, `"`, `""`) + `"` +} diff --git a/backend/internal/database/sqlite_task_migration_test.go b/backend/internal/database/sqlite_task_migration_test.go new file mode 100644 index 0000000..7fccc70 --- /dev/null +++ b/backend/internal/database/sqlite_task_migration_test.go @@ -0,0 +1,122 @@ +package database + +import ( + "fmt" + "strings" + "testing" +) + +func TestSQLiteBusinessUpgradePreservesDependentObjects(t *testing.T) { + for _, foreignKeys := range []int{0, 1} { + t.Run(fmt.Sprintf("foreign_keys_%d", foreignKeys), func(t *testing.T) { + db, err := Open(t.Context(), sqliteBusinessUpgradeConfig(t)) + if err != nil { + t.Fatal(err) + } + defer db.Close() + db.DB.SetMaxOpenConns(1) + createSQLiteHistoricalBusiness(t, db.DB, 3) + for _, statement := range []string{ + fmt.Sprintf("PRAGMA foreign_keys = %d", foreignKeys), + `ALTER TABLE tasks ADD COLUMN extra TEXT NOT NULL DEFAULT '扩展值)'`, + `ALTER TABLE tasks ADD COLUMN generated TEXT GENERATED ALWAYS AS (message || extra) VIRTUAL`, + `UPDATE tasks SET rowid = 77 WHERE id = 'task'`, + `CREATE UNIQUE INDEX "历史索引" ON tasks(extra, id) WHERE status = 'success'`, + `CREATE TABLE task_links (task_id TEXT PRIMARY KEY REFERENCES tasks(id) ON DELETE CASCADE)`, + `INSERT INTO task_links VALUES ('task')`, + `CREATE TABLE trigger_events (message TEXT)`, + `CREATE TRIGGER "历史任务触发器" AFTER UPDATE ON tasks BEGIN INSERT INTO trigger_events VALUES (new.message); END`, + `CREATE VIEW z_task_view AS SELECT id, message FROM tasks`, + `CREATE VIEW a_task_view AS SELECT * FROM z_task_view`, + `CREATE TRIGGER view_insert INSTEAD OF INSERT ON a_task_view BEGIN UPDATE tasks SET message = new.message WHERE id = new.id; END`, + `CREATE TRIGGER external_insert AFTER INSERT ON settings BEGIN UPDATE tasks SET message = new.value WHERE id = 'pending'; END`, + } { + if _, err := db.DB.ExecContext(t.Context(), statement); err != nil { + t.Fatal(err) + } + } + before := snapshotSQLiteHistoricalBusiness(t, db.DB) + objectsQuery := `SELECT type, name, sql FROM sqlite_master WHERE name IN ('历史索引', '历史任务触发器', 'z_task_view', 'a_task_view', 'view_insert', 'external_insert', 'task_links') ORDER BY name` + objects := snapshotSQLiteUpgradeQuery(t, db.DB, objectsQuery) + generated := snapshotSQLiteUpgradeQuery(t, db.DB, "SELECT rowid, id, generated FROM tasks ORDER BY id") + if err := Migrate(t.Context(), db.DB, "sqlite"); err != nil { + t.Fatal(err) + } + assertSQLiteHistoricalBusiness(t, db.DB, before) + if actual := snapshotSQLiteUpgradeQuery(t, db.DB, objectsQuery); actual != objects { + t.Fatal("依赖对象定义未完整保留") + } + if actual := snapshotSQLiteUpgradeQuery(t, db.DB, "SELECT rowid, id, generated FROM tasks ORDER BY id"); actual != generated { + t.Fatal("行号或生成列发生变化") + } + var count, actualForeignKeys int + if err := db.DB.QueryRowContext(t.Context(), "SELECT COUNT(*) FROM task_links WHERE task_id = 'task'").Scan(&count); err != nil || count != 1 { + t.Fatalf("任务表重建触发了级联删除：%v", err) + } + if err := db.DB.QueryRowContext(t.Context(), "PRAGMA foreign_keys").Scan(&actualForeignKeys); err != nil || actualForeignKeys != foreignKeys { + t.Fatalf("连接的外键设置未恢复：%d，%v", actualForeignKeys, err) + } + if _, err := db.DB.ExecContext(t.Context(), `INSERT INTO a_task_view VALUES ('task', '升级后视图写入'); INSERT INTO settings (key, value) VALUES ('after', '升级后外部触发器')`); err != nil { + t.Fatal(err) + } + if err := db.DB.QueryRowContext(t.Context(), "SELECT COUNT(*) FROM trigger_events").Scan(&count); err != nil || count != 2 { + t.Fatalf("触发器执行结果错误：%d，%v", count, err) + } + }) + } +} + +func TestSQLiteBusinessUpgradeFailureRollsBackAndRetries(t *testing.T) { + cases := []struct { + name string + prepare string + cleanup string + message string + }{ + {"version_record", `CREATE TRIGGER reject_version BEFORE INSERT ON schema_migrations WHEN new.version = 4 BEGIN SELECT RAISE(ABORT, 'blocked-upgrade'); END`, "DROP TRIGGER reject_version", "blocked-upgrade"}, + {"temporary_name", `CREATE TABLE scert_tasks_migration_4 (value TEXT); INSERT INTO scert_tasks_migration_4 VALUES ('保留内容')`, "DROP TABLE scert_tasks_migration_4", "already exists"}, + {"foreign_key", `CREATE TABLE broken_links (id TEXT REFERENCES tasks(id)); INSERT INTO broken_links VALUES ('missing-task')`, "DELETE FROM broken_links", "外键冲突"}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + db, err := Open(t.Context(), sqliteBusinessUpgradeConfig(t)) + if err != nil { + t.Fatal(err) + } + defer db.Close() + db.DB.SetMaxOpenConns(1) + createSQLiteHistoricalBusiness(t, db.DB, 3) + if _, err := db.DB.ExecContext(t.Context(), tc.prepare); err != nil { + t.Fatal(err) + } + if _, err := db.DB.ExecContext(t.Context(), "PRAGMA foreign_keys = ON"); err != nil { + t.Fatal(err) + } + before := snapshotSQLiteHistoricalBusiness(t, db.DB) + schemaQuery := "SELECT type, name, tbl_name, sql FROM sqlite_master ORDER BY type, name" + schema := snapshotSQLiteUpgradeQuery(t, db.DB, schemaQuery) + history := snapshotSQLiteUpgradeQuery(t, db.DB, "SELECT * FROM schema_migrations ORDER BY version") + if err := Migrate(t.Context(), db.DB, "sqlite"); err == nil || !strings.Contains(err.Error(), tc.message) { + t.Fatalf("未拒绝有阻塞项的迁移：%v", err) + } + assertSQLiteHistoricalBusiness(t, db.DB, before) + if actual := snapshotSQLiteUpgradeQuery(t, db.DB, schemaQuery); actual != schema { + t.Fatal("迁移失败后表、索引或触发器定义被改变") + } + if actual := snapshotSQLiteUpgradeQuery(t, db.DB, "SELECT * FROM schema_migrations ORDER BY version"); actual != history { + t.Fatal("迁移失败后版本被错误推进") + } + var foreignKeys int + if err := db.DB.QueryRowContext(t.Context(), "PRAGMA foreign_keys").Scan(&foreignKeys); err != nil || foreignKeys != 1 { + t.Fatalf("失败路径未恢复外键设置：%d，%v", foreignKeys, err) + } + if _, err := db.DB.ExecContext(t.Context(), tc.cleanup); err != nil { + t.Fatal(err) + } + if err := Migrate(t.Context(), db.DB, "sqlite"); err != nil { + t.Fatalf("排除阻塞后无法重新升级：%v", err) + } + assertSQLiteHistoricalBusiness(t, db.DB, before) + }) + } +}