yms-daemon/internal/transaction/store_test.go

package transaction

import (
	"context"
	"encoding/json"
	"errors"
	"net/url"
	"path/filepath"
	"sync"
	"testing"
	"time"

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

// TestStoreMigratesVersionOneAndCommitsBackendContainerDeployment 验证从 v1 schema
// 迁移到最新版本，以及 backend 容器部署提交的完整链路：迁移前无部署记录、非容器类
// backend 事务不计入容器历史、未完成容器事务不计入、提交后单例记录与历史证据正确。
func TestStoreMigratesVersionOneAndCommitsBackendContainerDeployment(t *testing.T) {
	t.Parallel()
	ctx := context.Background()
	databasePath := filepath.Join(t.TempDir(), "transaction.db")
	dsn := (&url.URL{Scheme: "file", Path: databasePath}).String()
	database, err := driver.Open(dsn)
	if err != nil {
		t.Fatalf("open version one database: %v", err)
	}
	if _, err := database.ExecContext(ctx, schemaV1); err != nil {
		_ = database.Close()
		t.Fatalf("create version one database: %v", err)
	}
	if err := database.Close(); err != nil {
		t.Fatalf("close version one database: %v", err)
	}

	store, err := OpenStore(ctx, databasePath)
	if err != nil {
		t.Fatalf("migrate transaction store: %v", err)
	}
	t.Cleanup(func() { _ = store.Close() })

	if _, err := store.BackendContainerDeployment(ctx); !errors.Is(err, ErrNotFound) {
		t.Fatalf("unexpected deployment before commit: %v", err)
	}
	hasHistory, err := store.HasCommittedBackendContainerTransactionHistory(ctx)
	if err != nil || hasHistory {
		t.Fatalf("unexpected empty backend history: found=%t err=%v", hasHistory, err)
	}
	nativeRecord, _, err := store.CreateTransaction(ctx, CreateRequest{
		ID:             "native-backend-transaction",
		IdempotencyKey: "backend:native-request",
		Source:         "test",
		Service:        "backend",
	})
	if err != nil {
		t.Fatalf("create native backend transaction: %v", err)
	}
	if _, err := store.Transition(ctx, nativeRecord.ID, StateFailed, "finish native backend transaction"); err != nil {
		t.Fatalf("finish native backend transaction: %v", err)
	}
	hasHistory, err = store.HasCommittedBackendContainerTransactionHistory(ctx)
	if err != nil || hasHistory {
		t.Fatalf("native backend history was treated as container history: found=%t err=%v", hasHistory, err)
	}

	record, created, err := store.CreateTransaction(ctx, CreateRequest{
		ID:             "container-deployment-transaction",
		IdempotencyKey: "backend:container:deployment-request",
		Source:         "test",
		Service:        "backend",
	})
	if err != nil || !created {
		t.Fatalf("create backend container transaction: record=%+v created=%t err=%v", record, created, err)
	}
	for _, state := range []State{StateValidating, StatePrepared, StateStarting, StateSwitching, StateVerifying, StateDraining} {
		if _, err := store.Transition(ctx, record.ID, state, "test transition"); err != nil {
			t.Fatalf("transition backend container transaction to %s: %v", state, err)
		}
	}
	hasHistory, err = store.HasCommittedBackendContainerTransactionHistory(ctx)
	if err != nil || hasHistory {
		t.Fatalf("unfinished backend container transaction was treated as committed history: found=%t err=%v", hasHistory, err)
	}

	committed, err := store.CommitBackendContainerDeployment(ctx, record.ID, BackendContainerDeployment{
		ActivePort:    8081,
		ContainerName: "backend-8081",
		ImageDigest:   "sha256:aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
		ContainerID:   "container-backend-8081",
	}, "container deployment committed")
	if err != nil || committed.State != StateCommitted {
		t.Fatalf("commit backend container deployment: record=%+v err=%v", committed, err)
	}
	hasHistory, err = store.HasCommittedBackendContainerTransactionHistory(ctx)
	if err != nil || !hasHistory {
		t.Fatalf("committed backend container history was not recorded: found=%t err=%v", hasHistory, err)
	}
	deployment, err := store.BackendContainerDeployment(ctx)
	if err != nil {
		t.Fatalf("read backend container deployment: %v", err)
	}
	if deployment.ActivePort != 8081 || deployment.ContainerName != "backend-8081" || deployment.ContainerID != "container-backend-8081" || deployment.TransactionID != record.ID {
		t.Fatalf("unexpected backend container deployment: %+v", deployment)
	}
	var version int
	if err := store.db.QueryRowContext(ctx, "PRAGMA user_version").Scan(&version); err != nil || version != schemaVersion {
		t.Fatalf("unexpected migrated schema version: version=%d err=%v", version, err)
	}
}

// TestListRecentScansSQLiteRequestTextAsRawJSON 验证 ListRecent 能正确把 SQLite 中
// 存储的 request_json 文本还原为原始 JSON 字节，不丢失也不转义。
func TestListRecentScansSQLiteRequestTextAsRawJSON(t *testing.T) {
	t.Parallel()
	ctx := context.Background()
	store := openTestStore(t)
	record, _, err := store.CreateTransaction(ctx, CreateRequest{
		ID: "list-transaction", IdempotencyKey: "list-idempotency", Source: "test", Service: "backend",
		Request: json.RawMessage(`{"inputType":"native-jar"}`),
	})
	if err != nil {
		t.Fatalf("create list transaction: %v", err)
	}
	items, err := store.ListRecent(ctx, ListFilter{Limit: 20})
	if err != nil {
		t.Fatalf("list transactions: %v", err)
	}
	if len(items) != 1 || items[0].ID != record.ID || string(items[0].Request) != `{"inputType":"native-jar"}` {
		t.Fatalf("unexpected listed transaction: %+v", items)
	}
}

// TestBackendContainerDeploymentCommitRequiresDrainingTransaction 验证 backend 容器
// 部署提交要求事务必须处于 DRAINING 状态，否则返回 TransitionError 且不写入部署行。
func TestBackendContainerDeploymentCommitRequiresDrainingTransaction(t *testing.T) {
	t.Parallel()
	ctx := context.Background()
	store := openTestStore(t)
	record, _, err := store.CreateTransaction(ctx, CreateRequest{
		ID:             "container-deployment-wrong-state",
		IdempotencyKey: "container-deployment-wrong-state-request",
		Source:         "test",
		Service:        "backend",
	})
	if err != nil {
		t.Fatalf("create backend transaction: %v", err)
	}
	_, err = store.CommitBackendContainerDeployment(ctx, record.ID, BackendContainerDeployment{
		ActivePort:    8080,
		ContainerName: "backend-8080",
		ImageDigest:   "sha256:bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb",
		ContainerID:   "container-backend-8080",
	}, "must fail")
	var transitionErr *TransitionError
	if !errors.As(err, &transitionErr) {
		t.Fatalf("expected deployment commit transition error, got %v", err)
	}
	if _, err := store.BackendContainerDeployment(ctx); !errors.Is(err, ErrNotFound) {
		t.Fatalf("failed commit wrote backend deployment: %v", err)
	}
}

// TestStoreCreateIsIdempotentAndAllowsOnlyOneActiveTransaction 验证 CreateTransaction
// 的幂等性与单活动事务约束：同幂等键重试返回同一事务；活动事务未结束前新事务被
// ActiveTransactionError 拒绝；失败事务的幂等键被归档后可用新 ID 重试，且原事务的
// 幂等键被改写为带 :terminal: 前缀的归档形式。
func TestStoreCreateIsIdempotentAndAllowsOnlyOneActiveTransaction(t *testing.T) {
	t.Parallel()
	ctx := context.Background()
	store := openTestStore(t)
	request := CreateRequest{
		ID:             "transaction-1",
		IdempotencyKey: "request-1",
		Source:         "test",
		Service:        "backend",
		Request:        json.RawMessage(`{"service":"backend"}`),
	}
	created, isNew, err := store.CreateTransaction(ctx, request)
	if err != nil {
		t.Fatalf("create transaction: %v", err)
	}
	if !isNew || created.State != StateCreated {
		t.Fatalf("unexpected created transaction: %+v new=%v", created, isNew)
	}

	retried, isNew, err := store.CreateTransaction(ctx, request)
	if err != nil {
		t.Fatalf("retry transaction: %v", err)
	}
	if isNew || retried.ID != created.ID {
		t.Fatalf("idempotent retry created another transaction: %+v", retried)
	}

	_, _, err = store.CreateTransaction(ctx, CreateRequest{
		ID:             "transaction-2",
		IdempotencyKey: "request-2",
		Source:         "test",
		Service:        "frontend",
	})
	var activeErr *ActiveTransactionError
	if !errors.As(err, &activeErr) || activeErr.TransactionID != created.ID {
		t.Fatalf("expected active transaction error, got %v", err)
	}

	if _, err := store.Transition(ctx, created.ID, StateFailed, "test terminal state"); err != nil {
		t.Fatalf("finish first transaction: %v", err)
	}
	retriedAfterFailure, isNew, err := store.CreateTransaction(ctx, CreateRequest{
		ID:             "transaction-1-retry",
		IdempotencyKey: request.IdempotencyKey,
		Source:         request.Source,
		Service:        request.Service,
		Request:        request.Request,
	})
	if err != nil || !isNew || retriedAfterFailure.ID != "transaction-1-retry" {
		t.Fatalf("retry failed transaction: record=%+v new=%v err=%v", retriedAfterFailure, isNew, err)
	}
	archived, err := store.Transaction(ctx, created.ID)
	if err != nil {
		t.Fatalf("read archived failed transaction: %v", err)
	}
	if archived.IdempotencyKey != request.IdempotencyKey+":terminal:"+created.ID {
		t.Fatalf("unexpected archived idempotency key: %q", archived.IdempotencyKey)
	}
	if _, err := store.Transition(ctx, retriedAfterFailure.ID, StateFailed, "finish retried transaction"); err != nil {
		t.Fatalf("finish retried transaction: %v", err)
	}
	second, isNew, err := store.CreateTransaction(ctx, CreateRequest{
		ID:             "transaction-2",
		IdempotencyKey: "request-2",
		Source:         "test",
		Service:        "frontend",
	})
	if err != nil || !isNew || second.ID != "transaction-2" {
		t.Fatalf("create transaction after terminal state: record=%+v new=%v err=%v", second, isNew, err)
	}
}

// TestStoreTransitionStepAndEventPersistence 验证状态、步骤与事件在关闭并重开数据库
// 后仍能正确恢复：事务状态与版本持久化、待定步骤可查询、步骤完成落库、事件有序递增。
func TestStoreTransitionStepAndEventPersistence(t *testing.T) {
	t.Parallel()
	ctx := context.Background()
	databasePath := filepath.Join(t.TempDir(), "transaction.db")
	store, err := OpenStore(ctx, databasePath)
	if err != nil {
		t.Fatalf("open store: %v", err)
	}
	fixedTime := time.Date(2026, time.August, 15, 10, 0, 0, 0, time.UTC)
	store.now = func() time.Time { return fixedTime }
	record, _, err := store.CreateTransaction(ctx, CreateRequest{
		ID:             "transaction-persisted",
		IdempotencyKey: "request-persisted",
		Source:         "test",
		Service:        "all",
	})
	if err != nil {
		t.Fatalf("create transaction: %v", err)
	}
	if _, err := store.Transition(ctx, record.ID, StateValidating, "validation started"); err != nil {
		t.Fatalf("transition transaction: %v", err)
	}
	if _, err := store.Transition(ctx, record.ID, StatePrepared, "validation completed"); err != nil {
		t.Fatalf("transition transaction: %v", err)
	}
	step, isNew, err := store.RecordStepIntent(ctx, record.ID, StepIntent{
		Key:    "prepare-files",
		Name:   "prepare immutable files",
		Intent: json.RawMessage(`{"sha256":"abc"}`),
	})
	if err != nil || !isNew || step.Status != StepIntentRecorded {
		t.Fatalf("record step intent: step=%+v new=%v err=%v", step, isNew, err)
	}
	if err := store.Close(); err != nil {
		t.Fatalf("close store: %v", err)
	}

	reopened, err := OpenStore(ctx, databasePath)
	if err != nil {
		t.Fatalf("reopen store: %v", err)
	}
	t.Cleanup(func() { _ = reopened.Close() })
	persisted, err := reopened.Transaction(ctx, record.ID)
	if err != nil {
		t.Fatalf("read persisted transaction: %v", err)
	}
	if persisted.State != StatePrepared || persisted.Version != 3 {
		t.Fatalf("unexpected persisted transaction: %+v", persisted)
	}
	pending, err := reopened.PendingSteps(ctx, record.ID)
	if err != nil {
		t.Fatalf("read pending steps: %v", err)
	}
	if len(pending) != 1 || pending[0].Key != step.Key {
		t.Fatalf("unexpected pending steps: %+v", pending)
	}
	completed, err := reopened.CompleteStep(ctx, record.ID, step.Key, StepSucceeded, json.RawMessage(`{"installed":true}`), "")
	if err != nil || completed.Status != StepSucceeded {
		t.Fatalf("complete step: step=%+v err=%v", completed, err)
	}
	events, err := reopened.EventsAfter(ctx, record.ID, 0, 100)
	if err != nil {
		t.Fatalf("read events: %v", err)
	}
	if len(events) != 5 {
		t.Fatalf("unexpected event count: got %d events=%+v", len(events), events)
	}
	for index := 1; index < len(events); index++ {
		if events[index].Sequence <= events[index-1].Sequence {
			t.Fatalf("events are not ordered: %+v", events)
		}
	}
}

// TestStoreRejectsInvalidTransitionAndConflictingStepIntent 验证非法状态转换返回
// TransitionError，以及同一步骤键以不同意图重复记录时返回 ErrStepConflict。
func TestStoreRejectsInvalidTransitionAndConflictingStepIntent(t *testing.T) {
	t.Parallel()
	ctx := context.Background()
	store := openTestStore(t)
	record, _, err := store.CreateTransaction(ctx, CreateRequest{
		ID:             "transaction-conflict",
		IdempotencyKey: "request-conflict",
		Source:         "test",
		Service:        "backend",
	})
	if err != nil {
		t.Fatalf("create transaction: %v", err)
	}
	_, err = store.Transition(ctx, record.ID, StatePrepared, "skip validation")
	var transitionErr *TransitionError
	if !errors.As(err, &transitionErr) {
		t.Fatalf("expected transition error, got %v", err)
	}
	intent := StepIntent{Key: "same-key", Name: "first", Intent: json.RawMessage(`{"value":1}`)}
	if _, _, err := store.RecordStepIntent(ctx, record.ID, intent); err != nil {
		t.Fatalf("record first step intent: %v", err)
	}
	_, _, err = store.RecordStepIntent(ctx, record.ID, StepIntent{
		Key:    intent.Key,
		Name:   "different",
		Intent: intent.Intent,
	})
	if !errors.Is(err, ErrStepConflict) {
		t.Fatalf("expected step conflict, got %v", err)
	}
}

// TestStoreSerializesConcurrentCreates 验证并发创建事务时单活动事务约束生效：多个
// 并发创建请求中恰好一个成功，其余均以 ErrActiveExists 拒绝。
func TestStoreSerializesConcurrentCreates(t *testing.T) {
	t.Parallel()
	ctx := context.Background()
	store := openTestStore(t)
	const workers = 12
	var wait sync.WaitGroup
	wait.Add(workers)
	results := make(chan error, workers)
	for index := 0; index < workers; index++ {
		go func(index int) {
			defer wait.Done()
			_, _, err := store.CreateTransaction(ctx, CreateRequest{
				IdempotencyKey: "concurrent-" + string(rune('A'+index)),
				Source:         "test",
				Service:        "backend",
			})
			results <- err
		}(index)
	}
	wait.Wait()
	close(results)
	var created, rejected int
	for err := range results {
		switch {
		case err == nil:
			created++
		case errors.Is(err, ErrActiveExists):
			rejected++
		default:
			t.Fatalf("unexpected create error: %v", err)
		}
	}
	if created != 1 || rejected != workers-1 {
		t.Fatalf("unexpected concurrent result: created=%d rejected=%d", created, rejected)
	}
}

// openTestStore 在临时目录打开一个测试用 Store，并通过 t.Cleanup 确保测试结束时关闭。
//
// 参数 t 用于报告错误与注册清理函数。打开失败时以 t.Fatalf 终止测试。
func openTestStore(t *testing.T) *Store {
	t.Helper()
	store, err := OpenStore(context.Background(), filepath.Join(t.TempDir(), "transaction.db"))
	if err != nil {
		t.Fatalf("open test store: %v", err)
	}
	t.Cleanup(func() {
		if err := store.Close(); err != nil {
			t.Errorf("close test store: %v", err)
		}
	})
	return store
}