705 lines
21 KiB
Go
705 lines
21 KiB
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 = 1
|
|
|
|
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;
|
|
`
|
|
|
|
// Store 是服务端 SQLite 事务记录。一个进程只应创建一个 Store。
|
|
type Store struct {
|
|
db *sql.DB
|
|
now func() time.Time
|
|
}
|
|
|
|
// OpenStore 打开本地 SQLite，并强制校验第一版持久化参数。
|
|
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
|
|
}
|
|
|
|
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
|
|
}
|
|
|
|
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()
|
|
if _, err := tx.ExecContext(ctx, schemaV1); err != nil {
|
|
return fmt.Errorf("apply sqlite schema version 1: %w", err)
|
|
}
|
|
if err := tx.Commit(); err != nil {
|
|
return fmt.Errorf("commit sqlite migration: %w", err)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// Close 关闭服务端 SQLite。
|
|
func (s *Store) Close() error {
|
|
return s.db.Close()
|
|
}
|
|
|
|
// CreateTransaction 原子处理幂等重试和单活动事务约束。created=false 表示返回已有幂等事务。
|
|
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 {
|
|
return existing, false, nil
|
|
}
|
|
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
|
|
}
|
|
|
|
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 返回指定事务的最新持久化快照。
|
|
func (s *Store) Transaction(ctx context.Context, id string) (Transaction, error) {
|
|
return getTransactionByID(ctx, s.db, id)
|
|
}
|
|
|
|
// ActiveTransaction 返回当前唯一未结束事务。
|
|
func (s *Store) ActiveTransaction(ctx context.Context) (Transaction, error) {
|
|
return getActiveTransaction(ctx, s.db)
|
|
}
|
|
|
|
// Transition 校验并原子提交状态变化及其恢复事件。
|
|
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 表示相同意图已经存在。
|
|
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 原子记录外部状态核对后的最终结果。
|
|
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
|
|
}
|
|
|
|
// PendingSteps 返回重启后必须先核对实际外部状态的步骤。
|
|
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 返回指定顺序位置之后的事务事件。
|
|
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
|
|
}
|
|
|
|
type queryRower interface {
|
|
QueryRowContext(context.Context, string, ...any) *sql.Row
|
|
}
|
|
|
|
type rowScanner interface {
|
|
Scan(...any) error
|
|
}
|
|
|
|
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))
|
|
}
|
|
|
|
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))
|
|
}
|
|
|
|
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))
|
|
}
|
|
|
|
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
|
|
}
|
|
|
|
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))
|
|
}
|
|
|
|
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
|
|
}
|
|
|
|
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
|
|
}
|
|
|
|
func formatTime(value time.Time) string {
|
|
return value.UTC().Format(time.RFC3339Nano)
|
|
}
|
|
|
|
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
|
|
}
|
|
|
|
func cloneJSON(value json.RawMessage) json.RawMessage {
|
|
if len(value) == 0 {
|
|
return nil
|
|
}
|
|
return bytes.Clone(value)
|
|
}
|