postgres
import "github.com/Bugs5382/go-saga-orchestration/store/postgres"
Package postgres is the production Store implementation. It uses pgx for connection pooling and golang-migrate to apply schema migrations. Both cmd/api and cmd/engine call postgres.Migrate(dsn) immediately after postgres.Open succeeds to apply any pending migrations at boot.
func AcquireAdvisoryLock
func AcquireAdvisoryLock(ctx context.Context, pool *pgxpool.Pool, lockID int64) (release func(), err error)
AcquireAdvisoryLock takes a session-level Postgres advisory lock on a dedicated pooled connection and holds it until the returned release func is called. It is the single-leader election primitive used to elect one engine replica to run the timer / cron dispatcher, so multi-replica deployments do not fire duplicate wakeups.
The call blocks until the lock is acquired (pg_advisory_lock waits rather than failing when another session holds it). release runs pg_advisory_unlock on a fresh short-timeout context and then returns the connection to the pool; it is safe to call once, typically via defer on the goroutine that owns the leader role.
func Migrate
func Migrate(dsn string) error
Migrate applies every up-migration in store/postgres/migrations that has not yet been recorded in the schema_migrations table. Safe to call on every boot — no-ops if the schema is already at Head.
The migrations are embedded in the binary at build time so deploys do not need a sidecar migration job or an out-of-band run. cmd/api and cmd/engine each call this immediately after postgres.Open succeeds.
DSN must use the pgx5 form, e.g. "postgres://user:pass@host:5432/db?sslmode=disable".
type Store
Store wraps a pgxpool.Pool and implements store.Store.
type Store struct {
// contains filtered or unexported fields
}
func Open
func Open(ctx context.Context, dsn string) (*Store, error)
Open dials Postgres using the supplied DSN. Caller MUST defer Close().
func (*Store) AppendEvent
func (s *Store) AppendEvent(ctx context.Context, evt domain.SagaRunEvent) error
AppendEvent inserts the audit event, ignoring duplicates on (run_id, step_id, attempt, event_type).
func (*Store) AppendSignal
func (s *Store) AppendSignal(ctx context.Context, sig domain.SagaSignal) error
AppendSignal inserts a received signal row for the run.
func (*Store) Cancel
func (s *Store) Cancel(ctx context.Context, runID uuid.UUID, reason string) error
Cancel terminates an in-flight run: terminal cancelled + terminal_at, reason in last_error, open user tasks closed, await markers / wakeup cleared — all in one transaction. Idempotent: when the run is already terminal the guard updates no row, so user tasks are untouched and no event is emitted. See issue #80.
func (*Store) ClaimCronFire
func (s *Store) ClaimCronFire(ctx context.Context, id uuid.UUID, expectedNextFire, newNextFire time.Time) (bool, error)
ClaimCronFire atomically advances next_fire_at from expectedNextFire to newNextFire and stamps last_fired_at=now(). Returns true iff this caller won the compare-and-swap (false means another pod already claimed this fire).
func (*Store) ClearPause
func (s *Store) ClearPause(ctx context.Context, runID uuid.UUID) error
ClearPause returns the run to the running state and clears all wakeup and await markers.
func (*Store) Close
func (s *Store) Close()
Close releases the pool. Safe to call once.
func (*Store) CompleteAction
func (s *Store) CompleteAction(ctx context.Context, runID uuid.UUID, attempt int, result map[string]any) error
CompleteAction clears the await marker, merges result into variables, and sets wakeup_at=now() so the Advance paused-handling loop resumes the saga. If attempt does not match current_attempt the row is unchanged (late delivery).
func (*Store) CountRuns
func (s *Store) CountRuns(ctx context.Context, filter store.RunFilter) (int, error)
CountRuns returns the total count of runs matching filter (ignoring Limit/Offset).
func (*Store) CreateRun
func (s *Store) CreateRun(ctx context.Context, run domain.SagaRun) error
CreateRun inserts a new saga run row.
func (*Store) CreateUserTask
func (s *Store) CreateUserTask(ctx context.Context, task domain.UserTask) error
CreateUserTask inserts a new user task row into runtime.saga_user_tasks.
func (*Store) DeleteTrigger
func (s *Store) DeleteTrigger(ctx context.Context, id uuid.UUID) error
DeleteTrigger removes the trigger by id. Returns ErrNotFound if absent.
func (*Store) FailAction
func (s *Store) FailAction(ctx context.Context, runID uuid.UUID, attempt int, code, message string, retryable bool) error
FailAction transitions the run to failed and appends an audit event. If attempt does not match current_attempt the call is a no-op (late delivery).
func (*Store) FindRunsByAwaitedEvent
func (s *Store) FindRunsByAwaitedEvent(ctx context.Context, topic string) ([]domain.SagaRun, error)
FindRunsByAwaitedEvent returns paused runs awaiting an event on topic.
func (*Store) FindRunsByDueWakeup
func (s *Store) FindRunsByDueWakeup(ctx context.Context, now time.Time, limit int) ([]uuid.UUID, error)
FindRunsByDueWakeup returns up to limit IDs of paused runs whose wakeup_at is at or before now, ordered by wakeup_at.
func (*Store) GetAction
func (s *Store) GetAction(ctx context.Context, service, name string, version int) (domain.ActionRegistration, error)
GetAction returns the action registration for the given service/name/version.
func (*Store) GetEventByID
func (s *Store) GetEventByID(ctx context.Context, id uuid.UUID) (domain.SagaRunEvent, error)
GetEventByID returns a single audit event by its UUID, or ErrNotFound.
func (*Store) GetPublishedRuleByID
func (s *Store) GetPublishedRuleByID(ctx context.Context, ruleID string, tenantID *uuid.UUID) (domain.RuleDefinition, error)
GetPublishedRuleByID returns the most recent published version of a rule.
func (*Store) GetPublishedWorkflowByID
func (s *Store) GetPublishedWorkflowByID(ctx context.Context, workflowID string, tenantID *uuid.UUID) (domain.WorkflowDefinition, error)
GetPublishedWorkflowByID returns the highest-version published definition for workflowID scoped to tenantID (nil = platform), or ErrNotFound.
func (*Store) GetRun
func (s *Store) GetRun(ctx context.Context, id uuid.UUID) (domain.SagaRun, error)
GetRun loads the run with the given ID, or returns ErrNotFound.
func (*Store) GetTrigger
func (s *Store) GetTrigger(ctx context.Context, id uuid.UUID) (domain.SagaTrigger, error)
GetTrigger returns the SagaTrigger for id, or ErrNotFound.
func (*Store) GetUserTask
func (s *Store) GetUserTask(ctx context.Context, taskID uuid.UUID) (domain.UserTask, error)
GetUserTask returns the user task by ID, or ErrNotFound.
func (*Store) GetWorkflowDefinition
func (s *Store) GetWorkflowDefinition(ctx context.Context, id uuid.UUID) (domain.WorkflowDefinition, error)
GetWorkflowDefinition returns the definition with the given storage ID, or ErrNotFound.
func (*Store) ListActions
func (s *Store) ListActions(ctx context.Context, filter store.ActionFilter) ([]domain.ActionRegistration, error)
ListActions returns all action registrations matching the optional filter.
func (*Store) ListChildrenByParent
func (s *Store) ListChildrenByParent(ctx context.Context, parentID uuid.UUID, parentStepID string) ([]domain.SagaRun, error)
ListChildrenByParent returns all runs with parent_run_id=$1 AND parent_step_id=$2.
func (*Store) ListDueCronTriggers
func (s *Store) ListDueCronTriggers(ctx context.Context, now time.Time, limit int) ([]domain.SagaTrigger, error)
ListDueCronTriggers returns enabled cron triggers whose next_fire_at is at or before now, ordered oldest-first, capped at limit.
func (*Store) ListEventsByRun
func (s *Store) ListEventsByRun(ctx context.Context, runID uuid.UUID) ([]domain.SagaRunEvent, error)
ListEventsByRun returns the events recorded for runID ordered by recorded_at.
func (*Store) ListRuns
func (s *Store) ListRuns(ctx context.Context, filter store.RunFilter) ([]domain.SagaRun, error)
ListRuns returns saga runs matching filter, sorted by started_at DESC. For TriggerType filtering a LEFT JOIN on saga_triggers is added. Limit defaults to 50 when 0; hard max of 500 is enforced by the handler before this is called.
func (*Store) ListTriggers
func (s *Store) ListTriggers(ctx context.Context, filter store.TriggerFilter) ([]domain.SagaTrigger, error)
ListTriggers returns triggers matching the optional filter.
func (*Store) ListUserTasksByRun
func (s *Store) ListUserTasksByRun(ctx context.Context, runID uuid.UUID) ([]domain.UserTask, error)
ListUserTasksByRun returns all user tasks for a given run, ordered by creation time (id order is used as a stable proxy since saga_user_tasks has no separate created_at column beyond the implicit id ordering).
func (*Store) MarkAwaitingAction
func (s *Store) MarkAwaitingAction(ctx context.Context, runID uuid.UUID, dispatch string, attempt int) error
MarkAwaitingAction sets state=paused, awaited_action_dispatch, and current_attempt for the given run. Idempotent on (runID, attempt, dispatch).
func (*Store) MarkRunFailed
func (s *Store) MarkRunFailed(ctx context.Context, runID uuid.UUID, currentStep, lastError string) error
MarkRunFailed transitions a run to terminal failed, stamps terminal_at, and persists lastError on the run. Idempotent: the terminal guard means an already-terminal run is left untouched. See issue #80.
func (*Store) Pool
func (s *Store) Pool() *pgxpool.Pool
Pool exposes the underlying pgxpool for migrations + tests.
func (*Store) PopTryCatch
func (s *Store) PopTryCatch(ctx context.Context, runID uuid.UUID) (domain.TryCatchFrame, bool, error)
PopTryCatch removes and returns the top TryCatchFrame atomically. Returns (zero, false, nil) if the stack is empty.
func (*Store) PushTryCatch
func (s *Store) PushTryCatch(ctx context.Context, runID uuid.UUID, frame domain.TryCatchFrame) error
PushTryCatch reads the try_catch_stack JSONB, appends frame (enforcing max depth 3), and writes it back within a transaction.
func (*Store) RecordTriggerFire
func (s *Store) RecordTriggerFire(ctx context.Context, triggerID uuid.UUID, workflowID string, runID *uuid.UUID, fireErr string) error
RecordTriggerFire inserts a row into runtime.saga_trigger_fires. Best-effort: callers log errors but must not abort the triggering run on failure.
func (*Store) SetPausedAwaitingEvent
func (s *Store) SetPausedAwaitingEvent(ctx context.Context, runID uuid.UUID, topic string, headers map[string]string) error
SetPausedAwaitingEvent marks the run paused awaiting an event on topic with the given header filter.
func (*Store) SetPausedAwaitingEventWithDeadline
func (s *Store) SetPausedAwaitingEventWithDeadline(ctx context.Context, runID uuid.UUID, topic string, headers map[string]string, deadline *time.Time) error
SetPausedAwaitingEventWithDeadline is SetPausedAwaitingEvent plus an optional wakeup deadline (nil = wait indefinitely). The deadline is stored as wakeup_at so the timer dispatcher wakes the run if no matching event arrives in time.
func (*Store) SetPausedAwaitingSignal
func (s *Store) SetPausedAwaitingSignal(ctx context.Context, runID uuid.UUID, signalName string, deadline *time.Time) error
SetPausedAwaitingSignal marks the run paused awaiting signalName, with an optional wakeup deadline.
func (*Store) SetPausedWithWakeup
func (s *Store) SetPausedWithWakeup(ctx context.Context, runID uuid.UUID, wakeupAt time.Time) error
SetPausedWithWakeup marks the run paused and records wakeup_at.
func (*Store) SpawnChildRun
func (s *Store) SpawnChildRun(ctx context.Context, parentID uuid.UUID, parentStepID, branchKey string, def domain.WorkflowDefinition, inputs map[string]any) (uuid.UUID, error)
SpawnChildRun looks up the definition_id for def, constructs a child SagaRun with the parent fields set, and inserts it via CreateRun. The child begins at the definition's default Start step.
func (*Store) SpawnChildRunAt
func (s *Store) SpawnChildRunAt(ctx context.Context, parentID uuid.UUID, parentStepID, branchKey string, def domain.WorkflowDefinition, inputs map[string]any, startStep string) (uuid.UUID, error)
SpawnChildRunAt looks up the definition_id for def, constructs a child SagaRun with the parent fields set, overrides CurrentStep with startStep when non-empty, and inserts it via CreateRun.
func (*Store) StatsForWorkflow
func (s *Store) StatsForWorkflow(ctx context.Context, workflowID string) (store.WorkflowStats, error)
StatsForWorkflow computes aggregate metrics for a single workflow using two aggregate queries: one for success_rate_24h + last_run_at, one for in_flight.
func (*Store) SubmitUserTask
func (s *Store) SubmitUserTask(ctx context.Context, taskID uuid.UUID, submittedBy string, result map[string]any) error
SubmitUserTask marks the task submitted with the given actor and result. Idempotent: re-submitting overwrites the previous submission fields. Returns ErrNotFound if the task does not exist.
func (*Store) TryConsumeAwaitedSignal
func (s *Store) TryConsumeAwaitedSignal(ctx context.Context, runID uuid.UUID, signalName string) (bool, error)
TryConsumeAwaitedSignal atomically clears the await markers and marks matching unconsumed signal rows consumed when the run is paused awaiting signalName, reporting whether it did so.
func (*Store) UpdateRunState
func (s *Store) UpdateRunState(ctx context.Context, id uuid.UUID, state domain.RunState, currentStep string) error
UpdateRunState sets the run's state and current step, stamping terminal_at when the new state is terminal.
func (*Store) UpdateRunVariables
func (s *Store) UpdateRunVariables(ctx context.Context, runID uuid.UUID, merge map[string]any) error
UpdateRunVariables merges merge into saga_runs.variables using jsonb_set per top-level key. Dotted-key writes are flattened into jsonb_set path expressions; nested merges go through the JSONB operator '||' for shallow object combine, then jsonb_set for the dotted writes.
func (*Store) UpsertActionRegistration
func (s *Store) UpsertActionRegistration(ctx context.Context, reg domain.ActionRegistration) error
UpsertActionRegistration inserts or updates an action_registry row keyed by (service, action_name, version).
func (*Store) UpsertRuleDefinition
func (s *Store) UpsertRuleDefinition(ctx context.Context, def domain.RuleDefinition) (uuid.UUID, error)
UpsertRuleDefinition inserts (or upserts on (rule_id, version)) a rule.
func (*Store) UpsertTrigger
func (s *Store) UpsertTrigger(ctx context.Context, trigger domain.SagaTrigger) (uuid.UUID, error)
UpsertTrigger inserts or replaces a row in runtime.saga_triggers keyed by id. If trigger.ID == uuid.Nil a new ID is generated. If the ID is set and a row already exists, all mutable columns are replaced (upsert on conflict).
func (*Store) UpsertWorkflowDefinition
func (s *Store) UpsertWorkflowDefinition(ctx context.Context, def domain.WorkflowDefinition) (uuid.UUID, error)
UpsertWorkflowDefinition inserts def, or updates the existing row on a (workflow_id, version) conflict, returning the row's storage ID.
func (*Store) WakeFromExternal
func (s *Store) WakeFromExternal(ctx context.Context, runID uuid.UUID) error
WakeFromExternal clears all await markers and wakeup_at while leaving the run paused, so the Advance loop resumes it.
Generated by gomarkdoc