yms-daemon/internal/transaction/store.go

package transaction

import (
	"bytes"
	"context"
	"crypto/rand"
	"database/sql"
	"encoding/json"
	"errors"
	"fmt"
	"net/url"
	"os"
	"path/filepath"
	"strings"
	"time"

	"github.com/ncruces/go-sqlite3/driver"
)

const schemaVersion = 2

// schemaV1 SQLite 数据库的首版 schema，定义事务、步骤与事件三张核心表。
//
// transactions 表保存事务快照并通过 CHECK 约束限定合法状态；唯一部分索引
// one_unfinished_transaction 保证任一时刻最多只有一条未结束事务；transaction_steps
// 表保存步骤意图与结果；transaction_events 表保存可增量拉取的有序事件。脚本末尾
// 将 user_version 置为 1，供 migrate 判断已应用的版本。
const schemaV1 = `
CREATE TABLE transactions (
    id              TEXT PRIMARY KEY,
    idempotency_key TEXT NOT NULL UNIQUE,
    source          TEXT NOT NULL,
    service         TEXT NOT NULL,
    request_json    TEXT NOT NULL,
    state           TEXT NOT NULL CHECK (state IN (
        'CREATED', 'VALIDATING', 'PREPARED', 'STARTING', 'SWITCHING',
        'VERIFYING', 'DRAINING', 'COMMITTED', 'ROLLING_BACK',
        'ROLLED_BACK', 'FAILED'
    )),
    version         INTEGER NOT NULL,
    created_at      TEXT NOT NULL,
    updated_at      TEXT NOT NULL
) STRICT;

CREATE UNIQUE INDEX one_unfinished_transaction
    ON transactions ((1))
    WHERE state NOT IN ('COMMITTED', 'ROLLED_BACK', 'FAILED');

CREATE TABLE transaction_steps (
    transaction_id TEXT NOT NULL REFERENCES transactions(id) ON DELETE CASCADE,
    step_key       TEXT NOT NULL,
    name           TEXT NOT NULL,
    status         TEXT NOT NULL CHECK (status IN ('INTENT_RECORDED', 'SUCCEEDED', 'FAILED')),
    intent_json    TEXT NOT NULL,
    result_json    TEXT,
    error_message  TEXT NOT NULL DEFAULT '',
    created_at     TEXT NOT NULL,
    updated_at     TEXT NOT NULL,
    PRIMARY KEY (transaction_id, step_key)
) STRICT;

CREATE TABLE transaction_events (
    sequence       INTEGER PRIMARY KEY AUTOINCREMENT,
    transaction_id TEXT NOT NULL REFERENCES transactions(id) ON DELETE CASCADE,
    step_key       TEXT NOT NULL DEFAULT '',
    kind           TEXT NOT NULL,
    from_state     TEXT NOT NULL DEFAULT '',
    to_state       TEXT NOT NULL DEFAULT '',
    message        TEXT NOT NULL DEFAULT '',
    created_at     TEXT NOT NULL
) STRICT;

CREATE INDEX transaction_events_by_transaction
    ON transaction_events (transaction_id, sequence);

PRAGMA user_version = 1;
`

// schemaV2 第二版迁移脚本，新增 backend_container_deployment 单例表。
//
// 该表只允许存在一行（singleton_id 固定为 1），记录 backend 容器最近一次提交
// 使用的端口、容器与镜像信息，并把 active_port 限定在 8080 或 8081。脚本末尾
// 将 user_version 置为 2。
const schemaV2 = `
CREATE TABLE backend_container_deployment (
    singleton_id   INTEGER PRIMARY KEY CHECK (singleton_id = 1),
    active_port    INTEGER NOT NULL CHECK (active_port IN (8080, 8081)),
    container_name TEXT NOT NULL,
    image_digest   TEXT NOT NULL,
    container_id   TEXT NOT NULL,
    transaction_id TEXT NOT NULL REFERENCES transactions(id),
    updated_at     TEXT NOT NULL
) STRICT;

PRAGMA user_version = 2;
`

// Store 服务端 SQLite 事务记录。一个进程只应创建一个 Store。
//
// 它封装了底层 *sql.DB，并把单连接访问（MaxOpenConns/MaxIdleConns 均为 1）作为
// 事务串行化的一部分，同时保证连接级 PRAGMA 始终生效。now 字段用于注入时间，
// 便于测试构造确定性时间戳；生产环境为 time.Now。
type Store struct {
	db  *sql.DB          // 底层 SQLite 数据库连接池，限定为单连接。
	now func() time.Time // 当前时间来源，测试可注入固定时钟。
}

// OpenStore 打开本地 SQLite，并强制校验持久化参数和 schema 版本。
//
// 参数 path 是 SQLite 数据库文件路径，可为相对路径；ctx 用于取消连接建立与
// 校验过程。函数会先解析绝对路径并创建父目录，再以 _txlock=immediate 的连接
// 参数打开数据库，随后依次 ping、配置 SQLite（journal_mode=DELETE、synchronous=
// EXTRA、foreign_keys=ON）、执行迁移、收紧文件权限为 0600。任一步骤失败都会
// 关闭连接并返回错误；成功返回可供使用的 Store。
func OpenStore(ctx context.Context, path string) (*Store, error) {
	if path == "" {
		return nil, errors.New("sqlite path is required")
	}
	absPath, err := filepath.Abs(path)
	if err != nil {
		return nil, fmt.Errorf("resolve sqlite path: %w", err)
	}
	if err := os.MkdirAll(filepath.Dir(absPath), 0o750); err != nil {
		return nil, fmt.Errorf("create sqlite directory: %w", err)
	}

	dsn := (&url.URL{
		Scheme:   "file",
		Path:     absPath,
		RawQuery: url.Values{"_txlock": {"immediate"}}.Encode(),
	}).String()
	db, err := driver.Open(dsn)
	if err != nil {
		return nil, fmt.Errorf("open sqlite driver: %w", err)
	}
	// 单连接是服务端事务串行化的一部分，也保证连接级 PRAGMA 始终生效。
	db.SetMaxOpenConns(1)
	db.SetMaxIdleConns(1)

	closeOnError := func(cause error) (*Store, error) {
		_ = db.Close()
		return nil, cause
	}
	if err := db.PingContext(ctx); err != nil {
		return closeOnError(fmt.Errorf("ping sqlite: %w", err))
	}
	if err := configureSQLite(ctx, db); err != nil {
		return closeOnError(err)
	}
	if err := migrate(ctx, db); err != nil {
		return closeOnError(err)
	}
	if err := os.Chmod(absPath, 0o600); err != nil {
		return closeOnError(fmt.Errorf("set sqlite permissions: %w", err))
	}
	return &Store{db: db, now: time.Now}, nil
}

// configureSQLite 校验并强制设置 SQLite 的持久化与约束参数。
//
// 依次设置并回读验证：journal_mode 必须为 delete、synchronous 必须为 EXTRA（取值
// 3）、foreign_keys 必须为 ON。任何一项设置失败或回读值不符都会返回错误，以保证
// 后续所有事务都在预期的持久性与引用完整性约束下运行。
func configureSQLite(ctx context.Context, db *sql.DB) error {
	var journalMode string
	if err := db.QueryRowContext(ctx, "PRAGMA journal_mode = DELETE").Scan(&journalMode); err != nil {
		return fmt.Errorf("set sqlite journal mode: %w", err)
	}
	if journalMode != "delete" {
		return fmt.Errorf("sqlite journal mode mismatch: got %q, want %q", journalMode, "delete")
	}
	if _, err := db.ExecContext(ctx, "PRAGMA synchronous = EXTRA"); err != nil {
		return fmt.Errorf("set sqlite synchronous: %w", err)
	}
	var synchronous int
	if err := db.QueryRowContext(ctx, "PRAGMA synchronous").Scan(&synchronous); err != nil {
		return fmt.Errorf("read sqlite synchronous: %w", err)
	}
	if synchronous != 3 {
		return fmt.Errorf("sqlite synchronous mismatch: got %d, want 3 (EXTRA)", synchronous)
	}
	if _, err := db.ExecContext(ctx, "PRAGMA foreign_keys = ON"); err != nil {
		return fmt.Errorf("enable sqlite foreign keys: %w", err)
	}
	var foreignKeys int
	if err := db.QueryRowContext(ctx, "PRAGMA foreign_keys").Scan(&foreignKeys); err != nil {
		return fmt.Errorf("read sqlite foreign_keys: %w", err)
	}
	if foreignKeys != 1 {
		return errors.New("sqlite foreign_keys is not enabled")
	}
	return nil
}

// migrate 将 SQLite schema 从当前版本逐步升级到 schemaVersion。
//
// 先读取 PRAGMA user_version 判断当前版本：若高于支持的版本则报错；若已等于目标
// 版本则直接返回；否则在一个 Serializable 事务内按版本号递增顺序执行对应迁移脚本。
// 每个脚本内部自行设置新的 user_version，最后统一提交。任一脚本缺失或执行失败
// 都会回滚并返回错误。
func migrate(ctx context.Context, db *sql.DB) error {
	var version int
	if err := db.QueryRowContext(ctx, "PRAGMA user_version").Scan(&version); err != nil {
		return fmt.Errorf("read sqlite schema version: %w", err)
	}
	if version > schemaVersion {
		return fmt.Errorf("sqlite schema version %d is newer than supported version %d", version, schemaVersion)
	}
	if version == schemaVersion {
		return nil
	}
	tx, err := db.BeginTx(ctx, &sql.TxOptions{Isolation: sql.LevelSerializable})
	if err != nil {
		return fmt.Errorf("begin sqlite migration: %w", err)
	}
	defer tx.Rollback()
	for version < schemaVersion {
		nextVersion := version + 1
		var script string
		switch nextVersion {
		case 1:
			script = schemaV1
		case 2:
			script = schemaV2
		default:
			return fmt.Errorf("sqlite migration script is missing for version %d", nextVersion)
		}
		if _, err := tx.ExecContext(ctx, script); err != nil {
			return fmt.Errorf("apply sqlite schema version %d: %w", nextVersion, err)
		}
		version = nextVersion
	}
	if err := tx.Commit(); err != nil {
		return fmt.Errorf("commit sqlite migration: %w", err)
	}
	return nil
}

// Close 关闭服务端 SQLite。
//
// 释放底层数据库连接，返回底层 Close 的错误。关闭后 Store 不应再被使用。
func (s *Store) Close() error {
	return s.db.Close()
}

// CreateTransaction 原子处理幂等重试和单活动事务约束。created=false 表示返回已有幂等事务。
//
// 参数 request 为创建请求，其中 ID 留空时由存储层生成随机 ID；ctx 用于取消操作。
// 返回值含义如下：record 为最终的事务快照；created 为 true 表示新建了事务，false
// 表示命中了幂等键返回已存在事务；err 非空表示操作失败。处理逻辑：先校验请求；
// 再在 Serializable 事务内按幂等键查找已有事务，若存在且非终态则直接幂等返回，若
// 已处于 FAILED/ROLLED_BACK 则归档其幂等键并继续；随后校验无未结束事务（否则返回
// ActiveTransactionError），最后插入新事务与创建事件并提交。
func (s *Store) CreateTransaction(ctx context.Context, request CreateRequest) (record Transaction, created bool, err error) {
	if err := validateCreateRequest(&request); err != nil {
		return Transaction{}, false, err
	}
	if request.ID == "" {
		request.ID = rand.Text()
	}
	now := s.now().UTC()
	tx, err := s.db.BeginTx(ctx, &sql.TxOptions{Isolation: sql.LevelSerializable})
	if err != nil {
		return Transaction{}, false, fmt.Errorf("begin create transaction: %w", err)
	}
	defer tx.Rollback()

	existing, err := getTransactionByIdempotencyKey(ctx, tx, request.IdempotencyKey)
	if err == nil {
		if existing.State != StateFailed && existing.State != StateRolledBack {
			return existing, false, nil
		}
		archivedKey := existing.IdempotencyKey + ":terminal:" + existing.ID
		result, updateErr := tx.ExecContext(ctx, `
            UPDATE transactions
               SET idempotency_key = ?, version = version + 1, updated_at = ?
             WHERE id = ? AND version = ? AND idempotency_key = ?
               AND state IN (?, ?)`,
			archivedKey,
			formatTime(now),
			existing.ID,
			existing.Version,
			existing.IdempotencyKey,
			StateFailed,
			StateRolledBack,
		)
		if updateErr != nil {
			return Transaction{}, false, fmt.Errorf("archive terminal transaction idempotency key: %w", updateErr)
		}
		rows, rowsErr := result.RowsAffected()
		if rowsErr != nil {
			return Transaction{}, false, fmt.Errorf("read archived transaction rows: %w", rowsErr)
		}
		if rows != 1 {
			return Transaction{}, false, errors.New("terminal transaction changed concurrently")
		}
		if eventErr := insertEvent(ctx, tx, existing.ID, "", "TRANSACTION_RETRY_RELEASED", existing.State, existing.State, "terminal transaction idempotency key archived for manual retry", now); eventErr != nil {
			return Transaction{}, false, eventErr
		}
		err = ErrNotFound
	}
	if !errors.Is(err, ErrNotFound) {
		return Transaction{}, false, err
	}
	active, err := getActiveTransaction(ctx, tx)
	if err == nil {
		return Transaction{}, false, &ActiveTransactionError{TransactionID: active.ID}
	}
	if !errors.Is(err, ErrNotFound) {
		return Transaction{}, false, err
	}

	requestJSON := string(request.Request)
	_, err = tx.ExecContext(ctx, `
        INSERT INTO transactions (
            id, idempotency_key, source, service, request_json,
            state, version, created_at, updated_at
        ) VALUES (?, ?, ?, ?, ?, ?, 1, ?, ?)`,
		request.ID,
		request.IdempotencyKey,
		request.Source,
		request.Service,
		requestJSON,
		StateCreated,
		formatTime(now),
		formatTime(now),
	)
	if err != nil {
		return Transaction{}, false, fmt.Errorf("insert transaction: %w", err)
	}
	if err := insertEvent(ctx, tx, request.ID, "", "TRANSACTION_CREATED", "", StateCreated, "", now); err != nil {
		return Transaction{}, false, err
	}
	if err := tx.Commit(); err != nil {
		return Transaction{}, false, fmt.Errorf("commit create transaction: %w", err)
	}
	return Transaction{
		ID:             request.ID,
		IdempotencyKey: request.IdempotencyKey,
		Source:         request.Source,
		Service:        request.Service,
		Request:        cloneJSON(request.Request),
		State:          StateCreated,
		Version:        1,
		CreatedAt:      now,
		UpdatedAt:      now,
	}, true, nil
}

// validateCreateRequest 校验创建事务请求的必填字段与 JSON 合法性。
//
// IdempotencyKey、Source、Service 均不能为空白；Request 为空时规范化为空对象 {}，
// 非空时必须是合法 JSON。校验通过返回 nil，否则返回描述具体问题的错误。
func validateCreateRequest(request *CreateRequest) error {
	if strings.TrimSpace(request.IdempotencyKey) == "" {
		return errors.New("idempotency key is required")
	}
	if strings.TrimSpace(request.Source) == "" {
		return errors.New("transaction source is required")
	}
	if strings.TrimSpace(request.Service) == "" {
		return errors.New("transaction service is required")
	}
	if len(request.Request) == 0 {
		request.Request = json.RawMessage(`{}`)
	}
	if !json.Valid(request.Request) {
		return errors.New("transaction request is not valid JSON")
	}
	return nil
}

// Transaction 返回指定事务的最新持久化快照。
//
// 参数 id 为目标事务 ID，ctx 用于取消查询。若不存在对应事务则返回 ErrNotFound。
func (s *Store) Transaction(ctx context.Context, id string) (Transaction, error) {
	return getTransactionByID(ctx, s.db, id)
}

// ActiveTransaction 返回当前唯一未结束事务。
//
// 参数 ctx 用于取消查询。存在未结束事务时返回该事务快照，否则返回 ErrNotFound。
func (s *Store) ActiveTransaction(ctx context.Context) (Transaction, error) {
	return getActiveTransaction(ctx, s.db)
}

// ListRecent 按创建时间倒序返回事务历史，覆盖已提交、已回滚、已失败以及进行中的
// 事务。参数 ctx 用于取消查询；filter 提供分页与过滤条件，其中 Limit 必须在 1 到
// 1000 之间，Service 与 State 均为精确匹配且永不做推断。返回匹配的事务切片；参数
// 非法或查询失败时返回错误。
func (s *Store) ListRecent(ctx context.Context, filter ListFilter) ([]Transaction, error) {
	if filter.Limit <= 0 || filter.Limit > 1000 {
		return nil, errors.New("transaction history limit must be between 1 and 1000")
	}
	query := `SELECT id, idempotency_key, source, service, request_json, state, version, created_at, updated_at FROM transactions WHERE 1=1`
	args := make([]any, 0, 3)
	if filter.Service != "" {
		query += " AND service = ?"
		args = append(args, filter.Service)
	}
	if filter.State != "" {
		query += " AND state = ?"
		args = append(args, string(filter.State))
	}
	query += " ORDER BY created_at DESC, id DESC LIMIT ?"
	args = append(args, filter.Limit)
	rows, err := s.db.QueryContext(ctx, query, args...)
	if err != nil {
		return nil, fmt.Errorf("list transaction history: %w", err)
	}
	defer rows.Close()
	result := make([]Transaction, 0, filter.Limit)
	for rows.Next() {
		var item Transaction
		var state string
		var requestJSON string
		var createdAt, updatedAt string
		if err := rows.Scan(&item.ID, &item.IdempotencyKey, &item.Source, &item.Service, &requestJSON, &state, &item.Version, &createdAt, &updatedAt); err != nil {
			return nil, fmt.Errorf("scan transaction history: %w", err)
		}
		item.Request = json.RawMessage(requestJSON)
		item.State = State(state)
		item.CreatedAt, err = time.Parse(time.RFC3339Nano, createdAt)
		if err != nil {
			return nil, fmt.Errorf("parse transaction history creation time: %w", err)
		}
		item.UpdatedAt, err = time.Parse(time.RFC3339Nano, updatedAt)
		if err != nil {
			return nil, fmt.Errorf("parse transaction history update time: %w", err)
		}
		result = append(result, item)
	}
	if err := rows.Err(); err != nil {
		return nil, fmt.Errorf("iterate transaction history: %w", err)
	}
	return result, nil
}

// Transition 校验并原子提交状态变化及其恢复事件。
//
// 参数 id 为目标事务，next 为期望推进到的状态，message 为随事件记录的人类可读
// 信息；ctx 用于取消操作。返回值为更新后的最新事务快照。处理逻辑：若 next 非法
// 直接报错；在 Serializable 事务内读取记录，若已处于 next 则幂等返回；否则调用
// CanTransitionTo 校验，非法转换返回 TransitionError；随后以版本号为条件原子更新
// 状态并写入 TRANSACTION_STATE_CHANGED 事件，最后提交。并发变更导致受影响行数
// 不为 1 时返回“transaction changed concurrently”错误。
func (s *Store) Transition(ctx context.Context, id string, next State, message string) (Transaction, error) {
	if !next.Valid() {
		return Transaction{}, fmt.Errorf("unknown transaction state: %q", next)
	}
	tx, err := s.db.BeginTx(ctx, &sql.TxOptions{Isolation: sql.LevelSerializable})
	if err != nil {
		return Transaction{}, fmt.Errorf("begin state transition: %w", err)
	}
	defer tx.Rollback()
	record, err := getTransactionByID(ctx, tx, id)
	if err != nil {
		return Transaction{}, err
	}
	if record.State == next {
		return record, nil
	}
	if !record.State.CanTransitionTo(next) {
		return Transaction{}, &TransitionError{From: record.State, To: next}
	}
	now := s.now().UTC()
	result, err := tx.ExecContext(ctx, `
        UPDATE transactions
           SET state = ?, version = version + 1, updated_at = ?
         WHERE id = ? AND version = ?`,
		next, formatTime(now), id, record.Version,
	)
	if err != nil {
		return Transaction{}, fmt.Errorf("update transaction state: %w", err)
	}
	rows, err := result.RowsAffected()
	if err != nil {
		return Transaction{}, fmt.Errorf("read updated transaction rows: %w", err)
	}
	if rows != 1 {
		return Transaction{}, errors.New("transaction changed concurrently")
	}
	if err := insertEvent(ctx, tx, id, "", "TRANSACTION_STATE_CHANGED", record.State, next, message, now); err != nil {
		return Transaction{}, err
	}
	if err := tx.Commit(); err != nil {
		return Transaction{}, fmt.Errorf("commit state transition: %w", err)
	}
	record.State = next
	record.Version++
	record.UpdatedAt = now
	return record, nil
}

// RecordStepIntent 先于外部副作用持久化步骤意图。created=false 表示相同意图已经存在。
//
// 参数 transactionID 为所属事务，intent 描述要执行的步骤，ctx 用于取消操作。返回
// 值 record 为步骤快照；created 为 true 表示新建意图，false 表示同 key 意图已存在
// 而幂等返回。处理逻辑：校验 intent 的 Key、Name 与 Intent JSON 合法性；在
// Serializable 事务内确认事务存在且非终态；若同 key 步骤已存在，则校验名称与意图
// 完全一致（否则返回 ErrStepConflict），一致则幂等返回；否则插入 INTENT_RECORDED
// 步骤并写入 STEP_INTENT_RECORDED 事件后提交。
func (s *Store) RecordStepIntent(ctx context.Context, transactionID string, intent StepIntent) (record Step, created bool, err error) {
	if strings.TrimSpace(intent.Key) == "" {
		return Step{}, false, errors.New("step key is required")
	}
	if strings.TrimSpace(intent.Name) == "" {
		return Step{}, false, errors.New("step name is required")
	}
	if len(intent.Intent) == 0 {
		intent.Intent = json.RawMessage(`{}`)
	}
	if !json.Valid(intent.Intent) {
		return Step{}, false, errors.New("step intent is not valid JSON")
	}

	tx, err := s.db.BeginTx(ctx, &sql.TxOptions{Isolation: sql.LevelSerializable})
	if err != nil {
		return Step{}, false, fmt.Errorf("begin record step intent: %w", err)
	}
	defer tx.Rollback()
	transactionRecord, err := getTransactionByID(ctx, tx, transactionID)
	if err != nil {
		return Step{}, false, err
	}
	if transactionRecord.State.Terminal() {
		return Step{}, false, fmt.Errorf("cannot record a step for terminal transaction %s", transactionID)
	}
	existing, err := getStep(ctx, tx, transactionID, intent.Key)
	if err == nil {
		if existing.Name != intent.Name || !bytes.Equal(existing.Intent, intent.Intent) {
			return Step{}, false, ErrStepConflict
		}
		return existing, false, nil
	}
	if !errors.Is(err, ErrNotFound) {
		return Step{}, false, err
	}

	now := s.now().UTC()
	_, err = tx.ExecContext(ctx, `
        INSERT INTO transaction_steps (
            transaction_id, step_key, name, status, intent_json,
            result_json, error_message, created_at, updated_at
        ) VALUES (?, ?, ?, ?, ?, NULL, '', ?, ?)`,
		transactionID,
		intent.Key,
		intent.Name,
		StepIntentRecorded,
		string(intent.Intent),
		formatTime(now),
		formatTime(now),
	)
	if err != nil {
		return Step{}, false, fmt.Errorf("insert step intent: %w", err)
	}
	if err := insertEvent(ctx, tx, transactionID, intent.Key, "STEP_INTENT_RECORDED", "", "", intent.Name, now); err != nil {
		return Step{}, false, err
	}
	if err := tx.Commit(); err != nil {
		return Step{}, false, fmt.Errorf("commit step intent: %w", err)
	}
	return Step{
		TransactionID: transactionID,
		Key:           intent.Key,
		Name:          intent.Name,
		Status:        StepIntentRecorded,
		Intent:        cloneJSON(intent.Intent),
		CreatedAt:     now,
		UpdatedAt:     now,
	}, true, nil
}

// CompleteStep 原子记录外部状态核对后的最终结果。
//
// 参数 transactionID 与 stepKey 定位步骤，status 只能是 StepSucceeded 或
// StepFailed，result 为核对结果载荷（可为空，非空必须合法 JSON），errorMessage
// 为失败描述（成功时传空字符串）；ctx 用于取消操作。返回更新后的步骤快照。处理
// 逻辑：在 Serializable 事务内读取步骤，若已处于目标状态且结果一致则幂等返回，
// 若结果不一致则返回 ErrStepConflict；若步骤不处于 INTENT_RECORDED 则返回
// ErrStepNotPending；否则以 status 为条件原子更新结果并写入 STEP_SUCCEEDED 或
// STEP_FAILED 事件后提交。
func (s *Store) CompleteStep(ctx context.Context, transactionID, stepKey string, status StepStatus, result json.RawMessage, errorMessage string) (Step, error) {
	if status != StepSucceeded && status != StepFailed {
		return Step{}, fmt.Errorf("invalid final step status: %q", status)
	}
	if len(result) > 0 && !json.Valid(result) {
		return Step{}, errors.New("step result is not valid JSON")
	}
	tx, err := s.db.BeginTx(ctx, &sql.TxOptions{Isolation: sql.LevelSerializable})
	if err != nil {
		return Step{}, fmt.Errorf("begin complete step: %w", err)
	}
	defer tx.Rollback()
	record, err := getStep(ctx, tx, transactionID, stepKey)
	if err != nil {
		return Step{}, err
	}
	if record.Status == status {
		if !bytes.Equal(record.Result, result) || record.Error != errorMessage {
			return Step{}, ErrStepConflict
		}
		return record, nil
	}
	if record.Status != StepIntentRecorded {
		return Step{}, ErrStepNotPending
	}
	now := s.now().UTC()
	var resultValue any
	if len(result) > 0 {
		resultValue = string(result)
	}
	updateResult, err := tx.ExecContext(ctx, `
        UPDATE transaction_steps
           SET status = ?, result_json = ?, error_message = ?, updated_at = ?
         WHERE transaction_id = ? AND step_key = ? AND status = ?`,
		status,
		resultValue,
		errorMessage,
		formatTime(now),
		transactionID,
		stepKey,
		StepIntentRecorded,
	)
	if err != nil {
		return Step{}, fmt.Errorf("update step result: %w", err)
	}
	updatedRows, err := updateResult.RowsAffected()
	if err != nil {
		return Step{}, fmt.Errorf("read updated step rows: %w", err)
	}
	if updatedRows != 1 {
		return Step{}, errors.New("step changed concurrently")
	}
	eventKind := "STEP_SUCCEEDED"
	if status == StepFailed {
		eventKind = "STEP_FAILED"
	}
	if err := insertEvent(ctx, tx, transactionID, stepKey, eventKind, "", "", errorMessage, now); err != nil {
		return Step{}, err
	}
	if err := tx.Commit(); err != nil {
		return Step{}, fmt.Errorf("commit step result: %w", err)
	}
	record.Status = status
	record.Result = cloneJSON(result)
	record.Error = errorMessage
	record.UpdatedAt = now
	return record, nil
}

// ReopenFailedStep makes one manually resumed external step pending again.
//
// ReopenFailedStep 把一个已失败的外部步骤重新置为待执行（INTENT_RECORDED），供人工
// 恢复后重新 Apply。参数 transactionID 与 stepKey 定位步骤，message 为随
// STEP_REOPENED 事件记录的原因；ctx 用于取消操作。Coordinator 只有在 Inspect 明确
// 得到 APPLIED 或 NOT_APPLIED 结论后才会调用本方法；UNKNOWN 永远不会重新打开失败
// 步骤。处理逻辑：在 Serializable 事务内读取步骤，若已是待定状态则幂等返回；若
// 不是 FAILED 则返回 ErrStepNotPending；否则以 FAILED 为条件原子清空结果与错误
// 信息、重置状态为 INTENT_RECORDED，并写入 STEP_REOPENED 事件后提交。
func (s *Store) ReopenFailedStep(ctx context.Context, transactionID, stepKey, message string) (Step, error) {
	tx, err := s.db.BeginTx(ctx, &sql.TxOptions{Isolation: sql.LevelSerializable})
	if err != nil {
		return Step{}, fmt.Errorf("begin reopen failed step: %w", err)
	}
	defer tx.Rollback()
	record, err := getStep(ctx, tx, transactionID, stepKey)
	if err != nil {
		return Step{}, err
	}
	if record.Status == StepIntentRecorded {
		return record, nil
	}
	if record.Status != StepFailed {
		return Step{}, ErrStepNotPending
	}
	now := s.now().UTC()
	result, err := tx.ExecContext(ctx, `
        UPDATE transaction_steps
           SET status = ?, result_json = NULL, error_message = '', updated_at = ?
         WHERE transaction_id = ? AND step_key = ? AND status = ?`,
		StepIntentRecorded,
		formatTime(now),
		transactionID,
		stepKey,
		StepFailed,
	)
	if err != nil {
		return Step{}, fmt.Errorf("reopen failed step: %w", err)
	}
	rows, err := result.RowsAffected()
	if err != nil {
		return Step{}, fmt.Errorf("read reopened step rows: %w", err)
	}
	if rows != 1 {
		return Step{}, errors.New("failed step changed concurrently")
	}
	if err := insertEvent(ctx, tx, transactionID, stepKey, "STEP_REOPENED", "", "", message, now); err != nil {
		return Step{}, err
	}
	if err := tx.Commit(); err != nil {
		return Step{}, fmt.Errorf("commit reopened step: %w", err)
	}
	record.Status = StepIntentRecorded
	record.Result = nil
	record.Error = ""
	record.UpdatedAt = now
	return record, nil
}

// PendingSteps 返回重启后必须先核对实际外部状态的步骤。
//
// 参数 transactionID 为目标事务，ctx 用于取消查询。返回该事务中所有仍处于
// INTENT_RECORDED 状态的步骤，按创建时间与 step key 排序。进程重启后调用方据此
// 逐一对这些步骤重新 Inspect，以确定其外部副作用是否真实发生。
func (s *Store) PendingSteps(ctx context.Context, transactionID string) ([]Step, error) {
	rows, err := s.db.QueryContext(ctx, `
        SELECT transaction_id, step_key, name, status, intent_json,
               result_json, error_message, created_at, updated_at
          FROM transaction_steps
         WHERE transaction_id = ? AND status = ?
         ORDER BY created_at, step_key`, transactionID, StepIntentRecorded)
	if err != nil {
		return nil, fmt.Errorf("query pending steps: %w", err)
	}
	defer rows.Close()
	var records []Step
	for rows.Next() {
		record, err := scanStep(rows)
		if err != nil {
			return nil, err
		}
		records = append(records, record)
	}
	if err := rows.Err(); err != nil {
		return nil, fmt.Errorf("iterate pending steps: %w", err)
	}
	return records, nil
}

// EventsAfter 返回指定顺序位置之后的事务事件。
//
// 参数 transactionID 为目标事务，afterSequence 为起始顺序号（返回顺序号严格大于
// 该值的事件），limit 为返回条数上限且必须在 1 到 1000 之间；ctx 用于取消查询。
// 返回按 sequence 升序排列的事件切片，供客户端断线后增量拉取遗漏事件。
func (s *Store) EventsAfter(ctx context.Context, transactionID string, afterSequence int64, limit int) ([]Event, error) {
	if afterSequence < 0 {
		return nil, errors.New("event sequence must not be negative")
	}
	if limit < 1 || limit > 1000 {
		return nil, errors.New("event limit must be between 1 and 1000")
	}
	rows, err := s.db.QueryContext(ctx, `
        SELECT sequence, transaction_id, step_key, kind,
               from_state, to_state, message, created_at
          FROM transaction_events
         WHERE transaction_id = ? AND sequence > ?
         ORDER BY sequence
         LIMIT ?`, transactionID, afterSequence, limit)
	if err != nil {
		return nil, fmt.Errorf("query transaction events: %w", err)
	}
	defer rows.Close()
	var events []Event
	for rows.Next() {
		var event Event
		var fromState, toState, createdAt string
		if err := rows.Scan(
			&event.Sequence,
			&event.TransactionID,
			&event.StepKey,
			&event.Kind,
			&fromState,
			&toState,
			&event.Message,
			&createdAt,
		); err != nil {
			return nil, fmt.Errorf("scan transaction event: %w", err)
		}
		event.FromState = State(fromState)
		event.ToState = State(toState)
		event.CreatedAt, err = parseTime(createdAt)
		if err != nil {
			return nil, err
		}
		events = append(events, event)
	}
	if err := rows.Err(); err != nil {
		return nil, fmt.Errorf("iterate transaction events: %w", err)
	}
	return events, nil
}

// queryRower 抽象了可执行单行查询并返回 *sql.Row 的查询能力。
//
// 该接口由 *sql.DB、*sql.Tx 与 *sql.Conn 等共同满足，使查询辅助函数（如
// getTransactionByID）既能在普通连接上执行，也能在事务内执行，避免重复实现。
type queryRower interface {
	QueryRowContext(context.Context, string, ...any) *sql.Row
}

// rowScanner 抽象了可从一行读取列值的扫描能力。
//
// 该接口由 *sql.Row 与 *sql.Rows 共同满足，使 scanTransaction 与 scanStep 等
// 反序列化辅助函数既能处理单行也能处理多行结果集中的当前行。
type rowScanner interface {
	Scan(...any) error
}

// getTransactionByID 按主键查询单条事务记录并反序列化为 Transaction。
//
// 参数 query 提供查询能力（可为 *sql.DB 或 *sql.Tx），id 为事务主键，ctx 用于
// 取消查询。未命中时返回 ErrNotFound。
func getTransactionByID(ctx context.Context, query queryRower, id string) (Transaction, error) {
	return scanTransaction(query.QueryRowContext(ctx, `
        SELECT id, idempotency_key, source, service, request_json,
               state, version, created_at, updated_at
          FROM transactions
         WHERE id = ?`, id))
}

// getTransactionByIdempotencyKey 按幂等键查询单条事务记录。
//
// 参数 query 提供查询能力，key 为幂等键，ctx 用于取消查询。未命中时返回
// ErrNotFound。
func getTransactionByIdempotencyKey(ctx context.Context, query queryRower, key string) (Transaction, error) {
	return scanTransaction(query.QueryRowContext(ctx, `
        SELECT id, idempotency_key, source, service, request_json,
               state, version, created_at, updated_at
          FROM transactions
         WHERE idempotency_key = ?`, key))
}

// getActiveTransaction 查询当前唯一未结束事务。
//
// 参数 query 提供查询能力，ctx 用于取消查询。返回状态不是 COMMITTED、ROLLED_BACK、
// FAILED 的任一条事务；若无此类事务则返回 ErrNotFound。
func getActiveTransaction(ctx context.Context, query queryRower) (Transaction, error) {
	return scanTransaction(query.QueryRowContext(ctx, `
        SELECT id, idempotency_key, source, service, request_json,
               state, version, created_at, updated_at
          FROM transactions
         WHERE state NOT IN (?, ?, ?)
         LIMIT 1`, StateCommitted, StateRolledBack, StateFailed))
}

// scanTransaction 从单行结果反序列化一条事务记录。
//
// 参数 row 提供扫描能力，返回反序列化后的 Transaction。将持久化的字符串形式
// request_json 还原为 json.RawMessage、state 还原为 State、时间文本解析为
// time.Time。未命中返回 ErrNotFound，其他解析失败返回包装后的错误。
func scanTransaction(row rowScanner) (Transaction, error) {
	var record Transaction
	var requestJSON, state, createdAt, updatedAt string
	if err := row.Scan(
		&record.ID,
		&record.IdempotencyKey,
		&record.Source,
		&record.Service,
		&requestJSON,
		&state,
		&record.Version,
		&createdAt,
		&updatedAt,
	); err != nil {
		if errors.Is(err, sql.ErrNoRows) {
			return Transaction{}, ErrNotFound
		}
		return Transaction{}, fmt.Errorf("scan transaction: %w", err)
	}
	record.Request = json.RawMessage(requestJSON)
	record.State = State(state)
	var err error
	record.CreatedAt, err = parseTime(createdAt)
	if err != nil {
		return Transaction{}, err
	}
	record.UpdatedAt, err = parseTime(updatedAt)
	if err != nil {
		return Transaction{}, err
	}
	return record, nil
}

// getStep 按事务与步骤键查询单条步骤记录。
//
// 参数 query 提供查询能力，transactionID 与 stepKey 联合定位步骤，ctx 用于取消
// 查询。未命中时返回 ErrNotFound。
func getStep(ctx context.Context, query queryRower, transactionID, stepKey string) (Step, error) {
	return scanStep(query.QueryRowContext(ctx, `
        SELECT transaction_id, step_key, name, status, intent_json,
               result_json, error_message, created_at, updated_at
          FROM transaction_steps
         WHERE transaction_id = ? AND step_key = ?`, transactionID, stepKey))
}

// scanStep 从单行结果反序列化一条步骤记录。
//
// 参数 row 提供扫描能力，返回反序列化后的 Step。result_json 允许为 NULL，仅在
// 有效时还原为 json.RawMessage；status 还原为 StepStatus，时间文本解析为
// time.Time。未命中返回 ErrNotFound，其他解析失败返回包装后的错误。
func scanStep(row rowScanner) (Step, error) {
	var record Step
	var status, intentJSON, createdAt, updatedAt string
	var resultJSON sql.NullString
	if err := row.Scan(
		&record.TransactionID,
		&record.Key,
		&record.Name,
		&status,
		&intentJSON,
		&resultJSON,
		&record.Error,
		&createdAt,
		&updatedAt,
	); err != nil {
		if errors.Is(err, sql.ErrNoRows) {
			return Step{}, ErrNotFound
		}
		return Step{}, fmt.Errorf("scan transaction step: %w", err)
	}
	record.Status = StepStatus(status)
	record.Intent = json.RawMessage(intentJSON)
	if resultJSON.Valid {
		record.Result = json.RawMessage(resultJSON.String)
	}
	var err error
	record.CreatedAt, err = parseTime(createdAt)
	if err != nil {
		return Step{}, err
	}
	record.UpdatedAt, err = parseTime(updatedAt)
	if err != nil {
		return Step{}, err
	}
	return record, nil
}

// insertEvent 在事务内插入一条事务事件。
//
// 参数 ctx 用于取消操作，tx 为目标数据库事务；transactionID、stepKey、kind 分别
// 描述事件归属、关联步骤键与事件类型；fromState、toState 记录状态变化（无变化时
// 传空）；message 为附加信息；createdAt 为事件时间。插入失败返回包装后的错误。
func insertEvent(
	ctx context.Context,
	tx *sql.Tx,
	transactionID string,
	stepKey string,
	kind string,
	fromState State,
	toState State,
	message string,
	createdAt time.Time,
) error {
	_, err := tx.ExecContext(ctx, `
        INSERT INTO transaction_events (
            transaction_id, step_key, kind, from_state, to_state, message, created_at
        ) VALUES (?, ?, ?, ?, ?, ?, ?)`,
		transactionID,
		stepKey,
		kind,
		fromState,
		toState,
		message,
		formatTime(createdAt),
	)
	if err != nil {
		return fmt.Errorf("insert transaction event: %w", err)
	}
	return nil
}

// formatTime 将时间格式化为 UTC 的 RFC3339Nano 文本，用于持久化。
//
// 参数 value 为待格式化时间，返回其 UTC 表示。所有持久化时间统一经此函数归一，
// 保证读取端可用 parseTime 精确还原。
func formatTime(value time.Time) string {
	return value.UTC().Format(time.RFC3339Nano)
}

// parseTime 解析持久化的 RFC3339Nano 时间文本。
//
// 参数 value 为待解析文本，返回对应 time.Time。解析失败返回包装后的错误。
func parseTime(value string) (time.Time, error) {
	parsed, err := time.Parse(time.RFC3339Nano, value)
	if err != nil {
		return time.Time{}, fmt.Errorf("parse persisted time %q: %w", value, err)
	}
	return parsed, nil
}

// cloneJSON 深拷贝一段 JSON 载荷，避免对外暴露底层可变字节切片。
//
// 参数 value 为待拷贝的 json.RawMessage，空值返回 nil。用于把从数据库读取的
// Request/Intent/Result 等载荷安全地返回给调用方，防止调用方修改影响后续读取。
func cloneJSON(value json.RawMessage) json.RawMessage {
	if len(value) == 0 {
		return nil
	}
	return bytes.Clone(value)
}