From 3a0e1d876994e45a5fb4a9c9b821110e7df22d0f Mon Sep 17 00:00:00 2001 From: xuans Date: Thu, 1 Oct 2026 02:09:25 +0800 Subject: [PATCH] =?UTF-8?q?=E5=BC=80=E5=8F=91=E5=89=8D=E6=A3=80=E6=9F=A5?= =?UTF-8?q?=E7=82=B9=EF=BC=9A=E5=A2=9E=E5=8A=A0=E9=83=A8=E7=BD=B2=E4=BA=8B?= =?UTF-8?q?=E5=8A=A1=E6=8C=81=E4=B9=85=E5=8C=96=E9=AA=A8=E6=9E=B6?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 为后续接入部署事务恢复流程保留数据库迁移、事务模型和仓储实现检查点。 --- backend/internal/database/migrate.go | 5 +- .../mysql/0017_deployment_transactions.sql | 23 ++ .../postgres/0017_deployment_transactions.sql | 24 ++ .../sqlite/0017_deployment_transactions.sql | 24 ++ backend/internal/deployment/transaction.go | 67 ++++++ .../deployment_transaction_repository.go | 227 ++++++++++++++++++ 6 files changed, 369 insertions(+), 1 deletion(-) create mode 100644 backend/internal/database/migrations/mysql/0017_deployment_transactions.sql create mode 100644 backend/internal/database/migrations/postgres/0017_deployment_transactions.sql create mode 100644 backend/internal/database/migrations/sqlite/0017_deployment_transactions.sql create mode 100644 backend/internal/deployment/transaction.go create mode 100644 backend/internal/repository/deployment_transaction_repository.go diff --git a/backend/internal/database/migrate.go b/backend/internal/database/migrate.go index 27fd501..7cd511b 100644 --- a/backend/internal/database/migrate.go +++ b/backend/internal/database/migrate.go @@ -18,6 +18,7 @@ import ( //go:embed migrations/sqlite/0014_project_members.sql migrations/mysql/0014_project_members.sql migrations/postgres/0014_project_members.sql //go:embed migrations/sqlite/0015_task_idempotency.sql migrations/mysql/0015_task_idempotency.sql migrations/postgres/0015_task_idempotency.sql //go:embed migrations/sqlite/0016_task_deployment_results.sql migrations/mysql/0016_task_deployment_results.sql migrations/postgres/0016_task_deployment_results.sql +//go:embed migrations/sqlite/0017_deployment_transactions.sql migrations/mysql/0017_deployment_transactions.sql migrations/postgres/0017_deployment_transactions.sql var migrationFiles embed.FS func Migrate(ctx context.Context, db *sql.DB, driver string) error { @@ -42,7 +43,7 @@ func Migrate(ctx context.Context, db *sql.DB, driver string) error { return nil } -const latestMigrationVersion = 16 +const latestMigrationVersion = 17 // currentMigrationVersion 在执行业务迁移前拒绝新版数据库及不连续的迁移历史。 func currentMigrationVersion(ctx context.Context, db *sql.DB) (int, error) { @@ -226,6 +227,8 @@ func migrationName(version int) string { return "task_idempotency" case 16: return "task_deployment_results" + case 17: + return "deployment_transactions" default: return "unknown" } diff --git a/backend/internal/database/migrations/mysql/0017_deployment_transactions.sql b/backend/internal/database/migrations/mysql/0017_deployment_transactions.sql new file mode 100644 index 0000000..859ff50 --- /dev/null +++ b/backend/internal/database/migrations/mysql/0017_deployment_transactions.sql @@ -0,0 +1,23 @@ +CREATE TABLE deployment_transactions ( + id VARCHAR(128) NOT NULL PRIMARY KEY, + task_id VARCHAR(36) NOT NULL, + attempt_id VARCHAR(36) NOT NULL, + attempt_number INT NOT NULL, + result_index INT NOT NULL, + server_id VARCHAR(36) NOT NULL, + server_name VARCHAR(255) NOT NULL, + certificate_path TEXT NOT NULL, + certificate_chain_path TEXT NOT NULL, + private_key_path TEXT NOT NULL, + temporary_paths_json TEXT NOT NULL, + installed_paths_json TEXT NOT NULL, + backups_json TEXT NOT NULL, + lock_paths_json TEXT NOT NULL, + reload_attempted INT NOT NULL DEFAULT 0, + phase VARCHAR(64) NOT NULL, + status VARCHAR(32) NOT NULL, + message TEXT NOT NULL, + updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, + INDEX idx_deployment_transactions_status (status, updated_at), + INDEX idx_deployment_transactions_task_id (task_id) +); diff --git a/backend/internal/database/migrations/postgres/0017_deployment_transactions.sql b/backend/internal/database/migrations/postgres/0017_deployment_transactions.sql new file mode 100644 index 0000000..e0ec15f --- /dev/null +++ b/backend/internal/database/migrations/postgres/0017_deployment_transactions.sql @@ -0,0 +1,24 @@ +CREATE TABLE deployment_transactions ( + id VARCHAR(128) NOT NULL PRIMARY KEY, + task_id VARCHAR(36) NOT NULL, + attempt_id VARCHAR(36) NOT NULL, + attempt_number INTEGER NOT NULL, + result_index INTEGER NOT NULL, + server_id VARCHAR(36) NOT NULL, + server_name VARCHAR(255) NOT NULL, + certificate_path TEXT NOT NULL, + certificate_chain_path TEXT NOT NULL, + private_key_path TEXT NOT NULL, + temporary_paths_json TEXT NOT NULL, + installed_paths_json TEXT NOT NULL, + backups_json TEXT NOT NULL, + lock_paths_json TEXT NOT NULL, + reload_attempted INTEGER NOT NULL DEFAULT 0, + phase VARCHAR(64) NOT NULL, + status VARCHAR(32) NOT NULL, + message TEXT NOT NULL, + updated_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP +); + +CREATE INDEX idx_deployment_transactions_status ON deployment_transactions(status, updated_at); +CREATE INDEX idx_deployment_transactions_task_id ON deployment_transactions(task_id); diff --git a/backend/internal/database/migrations/sqlite/0017_deployment_transactions.sql b/backend/internal/database/migrations/sqlite/0017_deployment_transactions.sql new file mode 100644 index 0000000..c967b9d --- /dev/null +++ b/backend/internal/database/migrations/sqlite/0017_deployment_transactions.sql @@ -0,0 +1,24 @@ +CREATE TABLE deployment_transactions ( + id TEXT NOT NULL PRIMARY KEY, + task_id TEXT NOT NULL, + attempt_id TEXT NOT NULL, + attempt_number INTEGER NOT NULL, + result_index INTEGER NOT NULL, + server_id TEXT NOT NULL, + server_name TEXT NOT NULL, + certificate_path TEXT NOT NULL, + certificate_chain_path TEXT NOT NULL, + private_key_path TEXT NOT NULL, + temporary_paths_json TEXT NOT NULL, + installed_paths_json TEXT NOT NULL, + backups_json TEXT NOT NULL, + lock_paths_json TEXT NOT NULL, + reload_attempted INTEGER NOT NULL DEFAULT 0, + phase TEXT NOT NULL, + status TEXT NOT NULL, + message TEXT NOT NULL, + updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP +); + +CREATE INDEX idx_deployment_transactions_status ON deployment_transactions(status, updated_at); +CREATE INDEX idx_deployment_transactions_task_id ON deployment_transactions(task_id); diff --git a/backend/internal/deployment/transaction.go b/backend/internal/deployment/transaction.go new file mode 100644 index 0000000..2f9cd24 --- /dev/null +++ b/backend/internal/deployment/transaction.go @@ -0,0 +1,67 @@ +package deployment + +import ( + "context" + "strings" + "time" +) + +// TransactionBackup 描述一次部署事务保存的旧文件。 +type TransactionBackup struct { + Original string `json:"original"` + Saved string `json:"saved"` +} + +// TransactionRecord 保存远程部署事务的可恢复现场，不包含证书内容或凭据。 +type TransactionRecord struct { + ID string + TaskID string + AttemptID string + AttemptNumber int + ResultIndex int + ServerID string + ServerName string + Paths []string + TemporaryPaths []string + InstalledPaths []string + Backups []TransactionBackup + LockPaths []string + ReloadAttempted bool + Phase string + Status string + Message string + UpdatedAt time.Time +} + +// TransactionJournal 持久化部署现场，保证进程重启后仍能识别未提交事务。 +type TransactionJournal interface { + CreateDeploymentTransaction(context.Context, TransactionRecord) error + UpdateDeploymentTransaction(context.Context, TransactionRecord) error + FinishDeploymentTransaction(context.Context, string, string, string) error + ListPendingDeploymentTransactions(context.Context) ([]TransactionRecord, error) +} + +type transactionContextKey struct{} + +// TransactionContext 保存任务与尝试标识，供部署层建立可追溯事务记录。 +type TransactionContext struct { + TaskID string + AttemptID string + AttemptNumber int +} + +// WithTransactionContext 将任务尝试标识绑定到证书部署上下文。 +func WithTransactionContext(ctx context.Context, value TransactionContext) context.Context { + value.TaskID = strings.TrimSpace(value.TaskID) + value.AttemptID = strings.TrimSpace(value.AttemptID) + return context.WithValue(ctx, transactionContextKey{}, value) +} + +// TransactionContextFromContext 读取任务尝试标识。 +func TransactionContextFromContext(ctx context.Context) (TransactionContext, bool) { + value, ok := ctx.Value(transactionContextKey{}).(TransactionContext) + if !ok || strings.TrimSpace(value.TaskID) == "" || strings.TrimSpace(value.AttemptID) == "" { + return TransactionContext{}, false + } + return value, true +} diff --git a/backend/internal/repository/deployment_transaction_repository.go b/backend/internal/repository/deployment_transaction_repository.go new file mode 100644 index 0000000..9825980 --- /dev/null +++ b/backend/internal/repository/deployment_transaction_repository.go @@ -0,0 +1,227 @@ +package repository + +import ( + "context" + "database/sql" + "encoding/json" + "errors" + "fmt" + "strings" + "time" + + "git.pcux.cn/xuans/Scert/backend/internal/deployment" +) + +// SQLDeploymentTransactionRepository 持久化远程部署的可恢复现场。 +type SQLDeploymentTransactionRepository struct { + db *sql.DB + driver string +} + +func NewSQLDeploymentTransactionRepository(db *sql.DB, driver ...string) *SQLDeploymentTransactionRepository { + return &SQLDeploymentTransactionRepository{db: db, driver: repositoryDriver(driver)} +} + +func (r *SQLDeploymentTransactionRepository) query(statement string) string { + return Rebind(r.driver, statement) +} + +func (r *SQLDeploymentTransactionRepository) CreateDeploymentTransaction(ctx context.Context, item deployment.TransactionRecord) error { + if r == nil || r.db == nil { + return errors.New("数据库连接不能为空") + } + if err := validateDeploymentTransaction(item); err != nil { + return err + } + values, err := encodeDeploymentTransaction(item) + if err != nil { + return err + } + _, err = r.db.ExecContext(ctx, r.query(`INSERT INTO deployment_transactions + (id, task_id, attempt_id, attempt_number, result_index, server_id, server_name, + certificate_path, certificate_chain_path, private_key_path, temporary_paths_json, + installed_paths_json, backups_json, lock_paths_json, reload_attempted, phase, status, + message, updated_at) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`), + item.ID, item.TaskID, item.AttemptID, item.AttemptNumber, item.ResultIndex, item.ServerID, item.ServerName, + values.paths[0], values.paths[1], values.paths[2], values.temporary, values.installed, values.backups, + values.locks, boolInt(item.ReloadAttempted), item.Phase, item.Status, item.Message, values.updatedAt) + if err != nil { + return fmt.Errorf("保存部署事务现场失败：%w", err) + } + return nil +} + +func (r *SQLDeploymentTransactionRepository) UpdateDeploymentTransaction(ctx context.Context, item deployment.TransactionRecord) error { + if r == nil || r.db == nil { + return errors.New("数据库连接不能为空") + } + if err := validateDeploymentTransaction(item); err != nil { + return err + } + values, err := encodeDeploymentTransaction(item) + if err != nil { + return err + } + result, err := r.db.ExecContext(ctx, r.query(`UPDATE deployment_transactions SET + certificate_path = ?, certificate_chain_path = ?, private_key_path = ?, + temporary_paths_json = ?, installed_paths_json = ?, backups_json = ?, lock_paths_json = ?, + reload_attempted = ?, phase = ?, status = ?, message = ?, updated_at = ? WHERE id = ?`), + values.paths[0], values.paths[1], values.paths[2], values.temporary, values.installed, values.backups, + values.locks, boolInt(item.ReloadAttempted), item.Phase, item.Status, item.Message, values.updatedAt, item.ID) + if err != nil { + return fmt.Errorf("更新部署事务现场失败：%w", err) + } + count, err := result.RowsAffected() + if err != nil { + return fmt.Errorf("读取部署事务更新结果失败：%w", err) + } + if count != 1 { + return errors.New("部署事务现场不存在") + } + return nil +} + +func (r *SQLDeploymentTransactionRepository) FinishDeploymentTransaction(ctx context.Context, id, status, message string) error { + if r == nil || r.db == nil { + return errors.New("数据库连接不能为空") + } + id = strings.TrimSpace(id) + status = strings.TrimSpace(status) + if id == "" || status == "" { + return errors.New("部署事务状态参数不能为空") + } + result, err := r.db.ExecContext(ctx, r.query(`UPDATE deployment_transactions SET status = ?, phase = ?, message = ?, updated_at = ? WHERE id = ?`), + status, status, message, time.Now().UTC(), id) + if err != nil { + return fmt.Errorf("完成部署事务记录失败：%w", err) + } + count, err := result.RowsAffected() + if err != nil { + return fmt.Errorf("读取部署事务完成结果失败：%w", err) + } + if count != 1 { + return errors.New("部署事务现场不存在") + } + return nil +} + +func (r *SQLDeploymentTransactionRepository) ListPendingDeploymentTransactions(ctx context.Context) ([]deployment.TransactionRecord, error) { + if r == nil || r.db == nil { + return nil, errors.New("数据库连接不能为空") + } + rows, err := r.db.QueryContext(ctx, r.query(`SELECT id, task_id, attempt_id, attempt_number, result_index, + server_id, server_name, certificate_path, certificate_chain_path, private_key_path, + temporary_paths_json, installed_paths_json, backups_json, lock_paths_json, + reload_attempted, phase, status, message, updated_at + FROM deployment_transactions WHERE status IN ('pending', 'recovery_required') ORDER BY updated_at, id`)) + if err != nil { + return nil, fmt.Errorf("查询待恢复部署事务失败：%w", err) + } + defer rows.Close() + items := make([]deployment.TransactionRecord, 0) + for rows.Next() { + item, err := scanDeploymentTransaction(rows) + if err != nil { + return nil, err + } + items = append(items, item) + } + if err := rows.Err(); err != nil { + return nil, fmt.Errorf("读取待恢复部署事务失败：%w", err) + } + return items, nil +} + +type deploymentTransactionValues struct { + paths [3]string + temporary string + installed string + backups string + locks string + updatedAt time.Time +} + +func validateDeploymentTransaction(item deployment.TransactionRecord) error { + if strings.TrimSpace(item.ID) == "" || strings.TrimSpace(item.TaskID) == "" || strings.TrimSpace(item.AttemptID) == "" { + return errors.New("部署事务标识不能为空") + } + if item.AttemptNumber < 1 || item.ResultIndex < 0 { + return errors.New("部署事务序号无效") + } + if strings.TrimSpace(item.ServerID) == "" || len(item.Paths) != 3 { + return errors.New("部署事务服务器或目标路径无效") + } + if strings.TrimSpace(item.Phase) == "" || strings.TrimSpace(item.Status) == "" { + return errors.New("部署事务阶段不能为空") + } + return nil +} + +func encodeDeploymentTransaction(item deployment.TransactionRecord) (deploymentTransactionValues, error) { + paths := [3]string{strings.TrimSpace(item.Paths[0]), strings.TrimSpace(item.Paths[1]), strings.TrimSpace(item.Paths[2])} + temporary, err := json.Marshal(item.TemporaryPaths) + if err != nil { + return deploymentTransactionValues{}, fmt.Errorf("编码部署临时路径失败：%w", err) + } + installed, err := json.Marshal(item.InstalledPaths) + if err != nil { + return deploymentTransactionValues{}, fmt.Errorf("编码部署已安装路径失败：%w", err) + } + backups, err := json.Marshal(item.Backups) + if err != nil { + return deploymentTransactionValues{}, fmt.Errorf("编码部署备份路径失败：%w", err) + } + locks, err := json.Marshal(item.LockPaths) + if err != nil { + return deploymentTransactionValues{}, fmt.Errorf("编码部署锁路径失败：%w", err) + } + updatedAt := item.UpdatedAt + if updatedAt.IsZero() { + updatedAt = time.Now().UTC() + } else { + updatedAt = updatedAt.UTC() + } + return deploymentTransactionValues{ + paths: paths, temporary: string(temporary), installed: string(installed), backups: string(backups), locks: string(locks), updatedAt: updatedAt, + }, nil +} + +type deploymentTransactionScanner interface { + Scan(...any) error +} + +func scanDeploymentTransaction(row deploymentTransactionScanner) (deployment.TransactionRecord, error) { + var item deployment.TransactionRecord + var paths [3]string + var temporary, installed, backups, locks string + var reloadAttempted int + if err := row.Scan(&item.ID, &item.TaskID, &item.AttemptID, &item.AttemptNumber, &item.ResultIndex, + &item.ServerID, &item.ServerName, &paths[0], &paths[1], &paths[2], &temporary, &installed, &backups, &locks, + &reloadAttempted, &item.Phase, &item.Status, &item.Message, &item.UpdatedAt); err != nil { + return deployment.TransactionRecord{}, fmt.Errorf("读取部署事务失败：%w", err) + } + item.Paths = paths[:] + item.ReloadAttempted = reloadAttempted != 0 + if err := json.Unmarshal([]byte(temporary), &item.TemporaryPaths); err != nil { + return deployment.TransactionRecord{}, fmt.Errorf("解析部署临时路径失败：%w", err) + } + if err := json.Unmarshal([]byte(installed), &item.InstalledPaths); err != nil { + return deployment.TransactionRecord{}, fmt.Errorf("解析部署已安装路径失败：%w", err) + } + if err := json.Unmarshal([]byte(backups), &item.Backups); err != nil { + return deployment.TransactionRecord{}, fmt.Errorf("解析部署备份路径失败：%w", err) + } + if err := json.Unmarshal([]byte(locks), &item.LockPaths); err != nil { + return deployment.TransactionRecord{}, fmt.Errorf("解析部署锁路径失败：%w", err) + } + item.UpdatedAt = item.UpdatedAt.UTC() + return item, nil +} + +func boolInt(value bool) int { + if value { + return 1 + } + return 0 +}