yms-daemon/internal/transaction/node_ssr_container_deployment.go

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)
	}
	fields := []struct {
		name  string
		value string
	}{
		{name: "container name", value: deployment.ContainerName},
		{name: "image digest", value: deployment.ImageDigest},
		{name: "container ID", value: deployment.ContainerID},
	}
	for _, field := range fields {
		if field.value == "" || strings.TrimSpace(field.value) != field.value {
			return fmt.Errorf("exact Node SSR container %s is required", field.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
}