feat: add Node SSR deployment transaction record

internal/transaction/node_ssr_container_deployment.go

@@ -0,0 +1,114 @@
+package transaction
+
+import (
+	"context"
+	"database/sql"
+	"errors"
+	"fmt"
+	"strings"
+	"time"
+)
+
+const nodeSsrService = "nodeSsr"
+
+// NodeSsrContainerDeployment 保存 Node SSR 最近一次提交的活动槽位。
+type NodeSsrContainerDeployment struct {
+	ActivePort    int
+	ContainerName string
+	ImageDigest   string
+	ContainerID   string
+	TransactionID string
+	UpdatedAt     time.Time
+}
+
+// NodeSsrContainerDeployment 返回当前 Node SSR 部署记录。
+func (s *Store) NodeSsrContainerDeployment(ctx context.Context) (NodeSsrContainerDeployment, error) {
+	return scanNodeSsrContainerDeployment(s.db.QueryRowContext(ctx, `
+        SELECT active_port, container_name, image_digest, container_id, transaction_id, updated_at
+          FROM node_ssr_container_deployment WHERE singleton_id = 1`))
+}
+
+// CommitNodeSsrContainerDeployment 原子写入 Node SSR 部署记录并提交事务。
+func (s *Store) CommitNodeSsrContainerDeployment(ctx context.Context, transactionID string, deployment NodeSsrContainerDeployment, message string) (Transaction, error) {
+	if transactionID == "" || strings.TrimSpace(transactionID) != transactionID {
+		return Transaction{}, errors.New("exact transaction ID is required")
+	}
+	if err := validateNodeSsrContainerDeployment(deployment); err != nil {
+		return Transaction{}, err
+	}
+	tx, err := s.db.BeginTx(ctx, &sql.TxOptions{Isolation: sql.LevelSerializable})
+	if err != nil {
+		return Transaction{}, fmt.Errorf("begin Node SSR container deployment commit: %w", err)
+	}
+	defer tx.Rollback()
+	record, err := getTransactionByID(ctx, tx, transactionID)
+	if err != nil {
+		return Transaction{}, err
+	}
+	if record.Service != nodeSsrService {
+		return Transaction{}, fmt.Errorf("transaction %s service must be %s", transactionID, nodeSsrService)
+	}
+	if record.State != StateDraining {
+		return Transaction{}, &TransitionError{From: record.State, To: StateCommitted}
+	}
+	now := s.now().UTC()
+	_, err = tx.ExecContext(ctx, `
+        INSERT INTO node_ssr_container_deployment
+            (singleton_id, active_port, container_name, image_digest, container_id, transaction_id, updated_at)
+        VALUES (1, ?, ?, ?, ?, ?, ?)
+        ON CONFLICT(singleton_id) DO UPDATE SET
+            active_port=excluded.active_port, container_name=excluded.container_name,
+            image_digest=excluded.image_digest, container_id=excluded.container_id,
+            transaction_id=excluded.transaction_id, updated_at=excluded.updated_at`,
+		deployment.ActivePort, deployment.ContainerName, deployment.ImageDigest, deployment.ContainerID, transactionID, formatTime(now))
+	if err != nil {
+		return Transaction{}, fmt.Errorf("write Node SSR container deployment: %w", err)
+	}
+	result, err := tx.ExecContext(ctx, `UPDATE transactions SET state=?, version=version+1, updated_at=? WHERE id=? AND version=? AND state=?`, StateCommitted, formatTime(now), transactionID, record.Version, StateDraining)
+	if err != nil {
+		return Transaction{}, fmt.Errorf("commit Node SSR transaction state: %w", err)
+	}
+	rows, err := result.RowsAffected()
+	if err != nil || rows != 1 {
+		return Transaction{}, errors.New("Node SSR transaction changed concurrently")
+	}
+	if err := insertEvent(ctx, tx, transactionID, "", "TRANSACTION_STATE_CHANGED", record.State, StateCommitted, message, now); err != nil {
+		return Transaction{}, err
+	}
+	if err := tx.Commit(); err != nil {
+		return Transaction{}, fmt.Errorf("commit Node SSR container deployment: %w", err)
+	}
+	record.State = StateCommitted
+	record.Version++
+	record.UpdatedAt = now
+	return record, nil
+}
+
+func validateNodeSsrContainerDeployment(deployment NodeSsrContainerDeployment) error {
+	if deployment.ActivePort != 18910 && deployment.ActivePort != 28910 {
+		return fmt.Errorf("Node SSR container deployment port must be 18910 or 28910: %d", deployment.ActivePort)
+	}
+	for name, value := range map[string]string{"container name": deployment.ContainerName, "image digest": deployment.ImageDigest, "container ID": deployment.ContainerID} {
+		if value == "" || strings.TrimSpace(value) != value {
+			return fmt.Errorf("exact Node SSR container %s is required", name)
+		}
+	}
+	return nil
+}
+
+func scanNodeSsrContainerDeployment(row rowScanner) (NodeSsrContainerDeployment, error) {
+	var deployment NodeSsrContainerDeployment
+	var updatedAt string
+	if err := row.Scan(&deployment.ActivePort, &deployment.ContainerName, &deployment.ImageDigest, &deployment.ContainerID, &deployment.TransactionID, &updatedAt); err != nil {
+		if errors.Is(err, sql.ErrNoRows) {
+			return NodeSsrContainerDeployment{}, ErrNotFound
+		}
+		return NodeSsrContainerDeployment{}, fmt.Errorf("scan Node SSR container deployment: %w", err)
+	}
+	parsed, err := parseTime(updatedAt)
+	if err != nil {
+		return NodeSsrContainerDeployment{}, fmt.Errorf("parse Node SSR deployment update time: %w", err)
+	}
+	deployment.UpdatedAt = parsed
+	return deployment, nil
+}