memory
import "github.com/Bugs5382/go-saga-orchestration/store/memory"
Package memory is an in-process store used by unit tests. Production code uses store/postgres.
type Store
Store is the in-memory store. Safe for concurrent use.
type Store struct {
// contains filtered or unexported fields
}
func New
func New() *Store
New returns an empty Store. Compile-time check confirms it satisfies the interface.
func (*Store) AppendEvent
func (s *Store) AppendEvent(_ context.Context, evt domain.SagaRunEvent) error
AppendEvent appends evt to the event list for evt.RunID.
func (*Store) AppendSignal
func (s *Store) AppendSignal(_ context.Context, sig domain.SagaSignal) error
AppendSignal appends sig to the signal list for sig.RunID.
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. Idempotent — no-op (and no event) once the run is terminal. See issue #80.
func (*Store) ClaimCronFire
func (s *Store) ClaimCronFire(_ 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. Returns true iff this caller won (the CAS matched). Returns false without error when the expected value no longer matches — the caller lost the race.
func (*Store) ClearPause
func (s *Store) ClearPause(_ context.Context, runID uuid.UUID) error
ClearPause returns the run to the running state and clears all wakeup and await markers, or returns ErrNotFound.
func (*Store) CompleteAction
func (s *Store) CompleteAction(ctx context.Context, runID uuid.UUID, attempt int, result map[string]any) error
CompleteAction merges result into Variables and sets wakeup_at=now() so the Advance paused-handling loop resumes the saga. If attempt != current_attempt it is a late/duplicate delivery and is silently ignored.
func (*Store) CountRuns
func (s *Store) CountRuns(_ context.Context, filter store.RunFilter) (int, error)
CountRuns returns the total count matching filter (ignoring Limit/Offset).
func (*Store) CreateRun
func (s *Store) CreateRun(_ context.Context, run domain.SagaRun) error
CreateRun stores run keyed by its ID.
func (*Store) CreateUserTask
func (s *Store) CreateUserTask(_ context.Context, task domain.UserTask) error
CreateUserTask stores a new UserTask keyed by its ID.
func (*Store) DeleteTrigger
func (s *Store) DeleteTrigger(_ context.Context, id uuid.UUID) error
DeleteTrigger removes the trigger from the store. Returns ErrNotFound if it does not exist.
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 != current_attempt it is a late delivery and is silently ignored.
func (*Store) FindRunsByAwaitedEvent
func (s *Store) FindRunsByAwaitedEvent(_ context.Context, topic string) ([]domain.SagaRun, error)
FindRunsByAwaitedEvent returns paused runs awaiting an event on topic.
func (*Store) FindRunsByDueWakeup
func (s *Store) FindRunsByDueWakeup(_ 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.
func (*Store) GetAction
func (s *Store) GetAction(_ context.Context, service, name string, version int) (domain.ActionRegistration, error)
GetAction returns the registration for service+name+version, or ErrNotFound.
func (*Store) GetEventByID
func (s *Store) GetEventByID(_ context.Context, id uuid.UUID) (domain.SagaRunEvent, error)
GetEventByID returns the first event whose ID matches, or ErrNotFound.
func (*Store) GetPublishedRuleByID
func (s *Store) GetPublishedRuleByID(_ context.Context, ruleID string, _ *uuid.UUID) (domain.RuleDefinition, error)
GetPublishedRuleByID returns the newest published version of ruleID, falling back to the most recent version if none is published, or ErrNotFound.
func (*Store) GetPublishedWorkflowByID
func (s *Store) GetPublishedWorkflowByID(_ context.Context, workflowID string, _ *uuid.UUID) (domain.WorkflowDefinition, error)
GetPublishedWorkflowByID returns the newest published version of workflowID, falling back to the most recent version if none is published, or ErrNotFound.
func (*Store) GetRun
func (s *Store) GetRun(_ context.Context, id uuid.UUID) (domain.SagaRun, error)
GetRun returns the run with the given ID, or ErrNotFound.
func (*Store) GetTrigger
func (s *Store) GetTrigger(_ context.Context, id uuid.UUID) (domain.SagaTrigger, error)
GetTrigger returns the SagaTrigger for id, or ErrNotFound.
func (*Store) GetUserTask
func (s *Store) GetUserTask(_ context.Context, taskID uuid.UUID) (domain.UserTask, error)
GetUserTask returns the task or ErrNotFound.
func (*Store) GetWorkflowDefinition
func (s *Store) GetWorkflowDefinition(_ 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(_ context.Context, filter store.ActionFilter) ([]domain.ActionRegistration, error)
ListActions returns all registrations matching the optional filter fields.
func (*Store) ListChildrenByParent
func (s *Store) ListChildrenByParent(_ context.Context, parentID uuid.UUID, parentStepID string) ([]domain.SagaRun, error)
ListChildrenByParent returns all runs whose ParentRunID == parentID and ParentStepID == parentStepID.
func (*Store) ListDueCronTriggers
func (s *Store) ListDueCronTriggers(_ context.Context, now time.Time, limit int) ([]domain.SagaTrigger, error)
ListDueCronTriggers returns enabled cron triggers whose next_fire_at is at or before now, sorted oldest-first, capped at limit.
func (*Store) ListEventsByRun
func (s *Store) ListEventsByRun(_ context.Context, runID uuid.UUID) ([]domain.SagaRunEvent, error)
ListEventsByRun returns a copy of the events recorded for runID.
func (*Store) ListRuns
func (s *Store) ListRuns(_ context.Context, filter store.RunFilter) ([]domain.SagaRun, error)
ListRuns returns saga runs matching filter, sorted by StartedAt DESC. TriggerType filter: iterates s.triggers to build an id→type map, then checks each run's TriggerID.
func (*Store) ListTriggers
func (s *Store) ListTriggers(_ context.Context, filter store.TriggerFilter) ([]domain.SagaTrigger, error)
ListTriggers returns triggers matching the optional filter.
func (*Store) ListUserTasksByRun
func (s *Store) ListUserTasksByRun(_ context.Context, runID uuid.UUID) ([]domain.UserTask, error)
ListUserTasksByRun returns all user tasks whose RunID matches runID. Returned in insertion order by task.ID (domain.UserTask has no CreatedAt field, so ID order is used as a stable proxy for creation order).
func (*Store) MarkAwaitingAction
func (s *Store) MarkAwaitingAction(_ context.Context, runID uuid.UUID, dispatch string, attempt int) error
MarkAwaitingAction sets state=paused and records the dispatch key + attempt. Idempotent on (runID, attempt): if the current_attempt already equals attempt and the dispatch key is the same, the call is a no-op.
func (*Store) MarkRunFailed
func (s *Store) MarkRunFailed(_ 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 on terminal runs. See issue #80.
func (*Store) PopTryCatch
func (s *Store) PopTryCatch(_ context.Context, runID uuid.UUID) (domain.TryCatchFrame, bool, error)
PopTryCatch removes and returns the top TryCatchFrame. Returns (zero, false, nil) when the stack is empty.
func (*Store) PushTryCatch
func (s *Store) PushTryCatch(_ context.Context, runID uuid.UUID, frame domain.TryCatchFrame) error
PushTryCatch appends frame to the run's TryCatchStack. Returns an error if the stack is already at maximum depth (3).
func (*Store) RecordTriggerFire
func (s *Store) RecordTriggerFire(_ context.Context, triggerID uuid.UUID, workflowID string, runID *uuid.UUID, fireErr string) error
RecordTriggerFire appends a TriggerFireRow to the in-memory audit log.
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 matching topic and the given header filter, or returns ErrNotFound.
func (*Store) SetPausedAwaitingEventWithDeadline
func (s *Store) SetPausedAwaitingEventWithDeadline(_ 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).
func (*Store) SetPausedAwaitingSignal
func (s *Store) SetPausedAwaitingSignal(_ context.Context, runID uuid.UUID, signalName string, deadline *time.Time) error
SetPausedAwaitingSignal marks the run paused awaiting signalName, setting an optional wakeup deadline, or returns ErrNotFound.
func (*Store) SetPausedWithWakeup
func (s *Store) SetPausedWithWakeup(_ context.Context, runID uuid.UUID, wakeupAt time.Time) error
SetPausedWithWakeup marks the run paused and sets its wakeup_at, or returns ErrNotFound.
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 creates a child run linked to parentID / parentStepID / branchKey, beginning at the child definition's default Start step.
func (*Store) SpawnChildRunAt
func (s *Store) SpawnChildRunAt(_ context.Context, parentID uuid.UUID, parentStepID, branchKey string, def domain.WorkflowDefinition, inputs map[string]any, startStep string) (uuid.UUID, error)
SpawnChildRunAt creates a child run linked to parentID / parentStepID / branchKey, beginning at startStep (empty string means the child definition's Start field).
func (*Store) StatsForWorkflow
func (s *Store) StatsForWorkflow(_ context.Context, workflowID string) (store.WorkflowStats, error)
StatsForWorkflow computes aggregate metrics for workflowID by iterating s.runs.
func (*Store) SubmitUserTask
func (s *Store) SubmitUserTask(_ context.Context, taskID uuid.UUID, submittedBy string, result map[string]any) error
SubmitUserTask marks the task as submitted. Idempotent (re-writes on repeated calls). Returns ErrNotFound if the task does not exist.
func (*Store) TriggerFires
func (s *Store) TriggerFires() []domain.TriggerFireRow
TriggerFires returns a copy of all recorded trigger-fire rows. Intended for test assertions only.
func (*Store) TryConsumeAwaitedSignal
func (s *Store) TryConsumeAwaitedSignal(_ context.Context, runID uuid.UUID, signalName string) (bool, error)
TryConsumeAwaitedSignal clears the await markers and marks any matching unconsumed signal consumed when the run is paused awaiting signalName, reporting whether it did so.
func (*Store) UpdateRunState
func (s *Store) UpdateRunState(_ context.Context, id uuid.UUID, state domain.RunState, currentStep string) error
UpdateRunState sets the run's state and current step, or returns ErrNotFound.
func (*Store) UpdateRunVariables
func (s *Store) UpdateRunVariables(_ context.Context, runID uuid.UUID, merge map[string]any) error
UpdateRunVariables merges the entries of merge into the run's variables, honouring dotted keys for nested writes, or returns ErrNotFound.
func (*Store) UpsertActionRegistration
func (s *Store) UpsertActionRegistration(_ context.Context, reg domain.ActionRegistration) error
UpsertActionRegistration stores or replaces the registration keyed by service+name+version.
func (*Store) UpsertRuleDefinition
func (s *Store) UpsertRuleDefinition(_ context.Context, def domain.RuleDefinition) (uuid.UUID, error)
UpsertRuleDefinition stores def, preserving a caller-supplied ID (or generating one) and replacing any prior entry with the same ID.
func (*Store) UpsertTrigger
func (s *Store) UpsertTrigger(_ context.Context, trigger domain.SagaTrigger) (uuid.UUID, error)
UpsertTrigger inserts or replaces a SagaTrigger. If trigger.ID == uuid.Nil a new ID is generated. If trigger.ID is set and a row already exists it is replaced in full.
func (*Store) UpsertWorkflowDefinition
func (s *Store) UpsertWorkflowDefinition(_ context.Context, def domain.WorkflowDefinition) (uuid.UUID, error)
UpsertWorkflowDefinition stores def under a fresh ID and records it under the workflow ID's version list, returning the new storage ID.
func (*Store) WakeFromExternal
func (s *Store) WakeFromExternal(_ 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, or returns ErrNotFound.
Generated by gomarkdoc