redis
import "github.com/Bugs5382/go-saga-orchestration/store/redis"
Package redis is a Redis/Valkey-backed store.Store implementation.
Package redis is a Redis/Valkey-backed store.Store implementation.
Package redis is a Redis/Valkey-backed store.Store implementation.
Package redis is a Redis/Valkey-backed store.Store implementation.
Package redis is a Redis/Valkey-backed store.Store implementation.
Package redis is a Redis/Valkey-backed store.Store implementation.
Package redis is a Redis/Valkey-backed store.Store implementation.
Package redis is a Redis/Valkey-backed store.Store implementation.
Package redis is a Redis/Valkey-backed store.Store implementation.
Package redis is a Redis/Valkey-backed store.Store implementation.
Package redis is a Redis/Valkey-backed store.Store implementation.
Package redis is a Redis/Valkey-backed store.Store implementation.
Package redis is a Redis/Valkey-backed store.Store implementation.
type Option
Option configures a Store.
type Option func(*Store)
func WithPrefix
func WithPrefix(p string) Option
WithPrefix overrides the default "saga:" key prefix.
func WithRunTTL
func WithRunTTL(d time.Duration) Option
WithRunTTL sets the terminal-run expiry (0 disables).
type Store
Store is a Redis/Valkey-backed store.Store. Safe for concurrent use.
type Store struct {
// contains filtered or unexported fields
}
func Open
func Open(ctx context.Context, url string, opts ...Option) (*Store, error)
Open dials url (redis:// or rediss://; Redis or Valkey) and verifies it.
func (*Store) AppendEvent
func (s *Store) AppendEvent(ctx context.Context, evt domain.SagaRunEvent) error
AppendEvent appends evt to the per-run event list and stores it under its own key for direct lookup by ID.
func (*Store) AppendSignal
func (s *Store) AppendSignal(ctx context.Context, sig domain.SagaSignal) error
AppendSignal appends sig to the signals:{runID} list.
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 (and the run dropped from the active indexes). Idempotent — a no-op (and no event) once the run is terminal. 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() using a Redis WATCH/MULTI transaction. Returns true iff this caller won the compare-and-swap.
func (*Store) ClearPause
func (s *Store) ClearPause(ctx context.Context, runID uuid.UUID) error
ClearPause transitions the run to running, clears all await markers and wakeup_at, and removes the run from idx:wakeup and idx:awaitevent:{topic}.
func (*Store) Close
func (s *Store) Close() error
Close releases the client.
func (*Store) CompleteAction
func (s *Store) CompleteAction(ctx context.Context, runID uuid.UUID, attempt int, result map[string]any) error
CompleteAction clears the dispatch marker, sets WakeupAt=now, and merges result into Variables. No-op when attempt != CurrentAttempt (late delivery).
func (*Store) CountRuns
func (s *Store) CountRuns(ctx context.Context, filter store.RunFilter) (int, error)
CountRuns returns the total count matching filter (ignoring Limit/Offset).
func (*Store) CreateRun
func (s *Store) CreateRun(ctx context.Context, run domain.SagaRun) error
CreateRun stores the run blob and adds it to the two index structures.
func (*Store) CreateUserTask
func (s *Store) CreateUserTask(ctx context.Context, task domain.UserTask) error
CreateUserTask stores a new UserTask and adds it to the per-run index.
func (*Store) DeleteTrigger
func (s *Store) DeleteTrigger(ctx context.Context, id uuid.UUID) error
DeleteTrigger removes the trigger. 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 AFTER the tx. No-op when attempt != CurrentAttempt (late delivery).
func (*Store) FindRunsByAwaitedEvent
func (s *Store) FindRunsByAwaitedEvent(ctx context.Context, topic string) ([]domain.SagaRun, error)
FindRunsByAwaitedEvent returns all paused runs awaiting an event on topic by loading members of idx:awaitevent:{topic} and returning those whose run blob is still paused and awaiting that topic (defensive filter).
func (*Store) FindRunsByDueWakeup
func (s *Store) FindRunsByDueWakeup(ctx context.Context, now time.Time, limit int) ([]uuid.UUID, error)
FindRunsByDueWakeup returns up to limit run IDs of paused runs whose wakeup_at is at or before now. It queries idx:wakeup via ZRANGEBYSCORE then loads the candidate run blobs and retains only those with State==RunStatePaused, matching the memory-store oracle (store/memory/store.go). The limit bounds the result count, not the ZSET scan range.
func (*Store) GetAction
func (s *Store) GetAction(ctx 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(ctx context.Context, id uuid.UUID) (domain.SagaRunEvent, error)
GetEventByID returns the event with the given ID, or ErrNotFound.
func (*Store) GetPublishedRuleByID
func (s *Store) GetPublishedRuleByID(ctx context.Context, ruleID string, _ *uuid.UUID) (domain.RuleDefinition, error)
GetPublishedRuleByID returns the newest published version of ruleID, falling back to the most recent version when none is published. Returns ErrNotFound when ruleID has never been upserted.
func (*Store) GetPublishedWorkflowByID
func (s *Store) GetPublishedWorkflowByID(ctx context.Context, workflowID string, _ *uuid.UUID) (domain.WorkflowDefinition, error)
GetPublishedWorkflowByID returns the newest published version of workflowID, falling back to the most recent version when none is published. Returns ErrNotFound when the workflow ID has never been upserted.
func (*Store) GetRun
func (s *Store) GetRun(ctx 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(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 UserTask or ErrNotFound.
func (*Store) GetWorkflowDefinition
func (s *Store) GetWorkflowDefinition(ctx context.Context, id uuid.UUID) (domain.WorkflowDefinition, error)
GetWorkflowDefinition returns the definition stored at 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 registrations matching the optional filter fields.
func (*Store) ListChildrenByParent
func (s *Store) ListChildrenByParent(ctx context.Context, parentID uuid.UUID, parentStepID string) ([]domain.SagaRun, error)
ListChildrenByParent returns all child runs for parentID/parentStepID.
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, sorted oldest-first, capped at limit.
func (*Store) ListEventsByRun
func (s *Store) ListEventsByRun(ctx context.Context, runID uuid.UUID) ([]domain.SagaRunEvent, error)
ListEventsByRun returns all events for runID in append order. An empty slice (not an error) is returned when no events have been appended.
func (*Store) ListRuns
func (s *Store) ListRuns(ctx context.Context, filter store.RunFilter) ([]domain.SagaRun, error)
ListRuns returns saga runs matching filter, sorted StartedAt DESC, paginated.
func (*Store) ListTriggers
func (s *Store) ListTriggers(ctx context.Context, filter store.TriggerFilter) ([]domain.SagaTrigger, error)
ListTriggers returns triggers matching the optional filter fields.
func (*Store) ListUserTasksByRun
func (s *Store) ListUserTasksByRun(ctx context.Context, runID uuid.UUID) ([]domain.UserTask, error)
ListUserTasksByRun returns all user tasks for runID sorted by ID bytes. This matches the memory store's uuidLess ordering.
func (*Store) MarkAwaitingAction
func (s *Store) MarkAwaitingAction(ctx context.Context, runID uuid.UUID, dispatch string, attempt int) error
MarkAwaitingAction sets state=paused and records the dispatch key + attempt. Idempotent on (attempt, dispatch): same pair returns without writing.
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 on terminal runs. See issue #80.
func (*Store) PopTryCatch
func (s *Store) PopTryCatch(ctx 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. Uses a closure variable to surface the popped value.
func (*Store) PushTryCatch
func (s *Store) PushTryCatch(ctx 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, _ uuid.UUID, _ string, _ *uuid.UUID, _ string) error
RecordTriggerFire is a no-op: Redis has no persistent audit table.
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. The run is added to idx:awaitevent:{topic}.
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 which is also recorded in idx:wakeup.
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 deadline added to idx:wakeup.
func (*Store) SetPausedWithWakeup
func (s *Store) SetPausedWithWakeup(ctx context.Context, runID uuid.UUID, wakeupAt time.Time) error
SetPausedWithWakeup marks the run paused with a wakeup time and registers it in the idx:wakeup ZSET.
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 beginning at the child 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 creates a child run linked to parentID/parentStepID/branchKey, beginning at startStep (empty string means the child definition's default start). The whole operation — def upsert and run/index writes — is performed inside a single WATCH/MULTI/EXEC optimistic-transaction retry loop so it is atomic.
func (*Store) StatsForWorkflow
func (s *Store) StatsForWorkflow(ctx context.Context, workflowID string) (store.WorkflowStats, error)
StatsForWorkflow computes aggregate metrics for workflowID.
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. Returns ErrNotFound if it does not exist.
func (*Store) TryConsumeAwaitedSignal
func (s *Store) TryConsumeAwaitedSignal(ctx context.Context, runID uuid.UUID, signalName string) (bool, error)
TryConsumeAwaitedSignal attempts to consume the awaited signal on the run. Returns (false, nil) when the run is missing, not paused, or the awaited signal name does not match. On match it clears all await markers and wakeup_at, marks the first unconsumed matching signal as consumed, and returns (true, nil).
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 via a WATCH/MULTI transaction. When the new state is terminal, the run is removed from the active idx:wakeup and idx:awaitevent indexes.
func (*Store) UpdateRunVariables
func (s *Store) UpdateRunVariables(ctx context.Context, id uuid.UUID, merge map[string]any) error
UpdateRunVariables merges entries of merge into the run's Variables using dotted-key path semantics, via a WATCH/MULTI transaction.
func (*Store) UpsertActionRegistration
func (s *Store) UpsertActionRegistration(ctx context.Context, reg domain.ActionRegistration) error
UpsertActionRegistration stores or replaces a registration and adds it to idx:actions.
func (*Store) UpsertRuleDefinition
func (s *Store) UpsertRuleDefinition(ctx context.Context, def domain.RuleDefinition) (uuid.UUID, error)
UpsertRuleDefinition stores def atomically using a WATCH/MULTI optimistic transaction retry loop. If def.ID is the zero UUID a new one is generated. When re-upserting an existing storage ID the old version-list entry is removed inside the same transaction to prevent duplicates under concurrency.
func (*Store) UpsertTrigger
func (s *Store) UpsertTrigger(ctx context.Context, trigger domain.SagaTrigger) (uuid.UUID, error)
UpsertTrigger inserts or replaces a SagaTrigger. Generates a new ID when trigger.ID == uuid.Nil. Defaults CreatedAt to now when zero.
func (*Store) UpsertWorkflowDefinition
func (s *Store) UpsertWorkflowDefinition(ctx context.Context, def domain.WorkflowDefinition) (uuid.UUID, error)
UpsertWorkflowDefinition stores def under a fresh storage ID and appends that ID to the def:byname:{workflowID} list (oldest→newest order).
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 state as paused, then removes it from idx:wakeup / idx:awaitevent.
Generated by gomarkdoc