saga
import "github.com/Bugs5382/go-saga-orchestration/saga"
Package saga is the embedding entrypoint: construct an in-process saga engine, register workflows and custom verbs, and drive runs — without running the engine binaries.
type InProcessEventEmitter
InProcessEventEmitter delivers an emitted event in-process, with no broker, mirroring service mode where the same event feeds both the event subscriber (which wakes awaiting runs) and the trigger dispatcher (which starts new runs):
- wakes runs awaiting the topic, applying the same header-subset match as engine.EventSubscriber, then
- runs the trigger dispatcher so matching triggers start new runs.
type InProcessEventEmitter struct {
// contains filtered or unexported fields
}
func (*InProcessEventEmitter) EmitEvent
func (e *InProcessEventEmitter) EmitEvent(ctx context.Context, topic string, headers map[string]string, payload map[string]any) error
EmitEvent wakes paused runs awaiting topic (header-subset match) and then runs the trigger dispatcher against the event so matching triggers start new runs.
type InProcessPublisher
InProcessPublisher satisfies engine.Publisher (and verbs.ActionDispatchPublisher) for embedded use with no message broker. PublishSagaAdvance runs the coordinator's Advance in a background goroutine bound to the Saga's context and tracked on its WaitGroup, so Saga.Shutdown can cancel and drain in-flight work. Action dispatch is unsupported in-process (use a worker / the service mode).
type InProcessPublisher struct {
// contains filtered or unexported fields
}
func (*InProcessPublisher) PublishActionDispatch
func (p *InProcessPublisher) PublishActionDispatch(_ context.Context, _ string, _ []byte) error
func (*InProcessPublisher) PublishSagaAdvance
func (p *InProcessPublisher) PublishSagaAdvance(_ context.Context, runID string) error
PublishSagaAdvance advances the run in a background goroutine.
Semantics for embedders: this is only reached by workflows that spawn child runs (parallel / foreach / spawn_saga); linear workflows advance synchronously inside Start and never use the publisher. The advance runs on the Saga's context (derived from Options.Context) and is registered on the Saga's WaitGroup, so Saga.Shutdown cancels it (Advance stops between steps) and waits for it to drain. Errors cannot be returned to the caller; set Options.Logger to observe them.
type Options
Options configures a Saga engine instance.
type Options struct {
Store store.Store
Clock clock.Clock
Licensing licensing.Resolver
Secrets secrets.Resolver
Publisher engine.Publisher
StartupProviders []engine.StartupVariableProvider
Logger *zerolog.Logger // optional; nil = no logging
// Context is the base context for background work (parallel/foreach/spawn
// child advances run on a cancellable context derived from it). Defaults to
// context.Background(). Shutdown cancels the derived context.
Context context.Context
}
type Saga
Saga is the embedding facade around the coordinator and store.
type Saga struct {
// contains filtered or unexported fields
}
func InMemory
func InMemory() *Saga
InMemory returns a Saga backed by an in-memory store with all defaults. Convenient for tests and examples.
func New
func New(opts Options) (*Saga, error)
New constructs a Saga from opts. opts.Store is required; all other fields have sensible defaults (SystemClock, in-memory secrets, StubAllowAll licensing, in-process publisher).
func (*Saga) Cancel
func (s *Saga) Cancel(ctx context.Context, runID uuid.UUID, reason string) error
Cancel terminates an in-flight run from outside the run — e.g. an approval policy withdrawing or re-submitting while a run is paused at a manual_approval. The run transitions to terminal cancelled, its open user tasks are closed (so none linger pending), and any awaited signal/event or pending wakeup is cleared so a stray advance cannot resurrect it. reason is recorded on the run's last_error. Idempotent: a no-op when the run is already terminal. See issue #80.
func (*Saga) Coordinator
func (s *Saga) Coordinator() *engine.Coordinator
Coordinator returns the underlying engine.Coordinator for advanced use.
func (*Saga) Get
func (s *Saga) Get(ctx context.Context, runID uuid.UUID) (domain.SagaRun, error)
Get returns the current state of a run by ID.
func (*Saga) Register
func (s *Saga) Register(def domain.WorkflowDefinition) error
Register upserts a workflow definition into the store so it can be started.
func (*Saga) RegisterVerb
func (s *Saga) RegisterVerb(stepType string, licenseGroup string, h verbs.Handler)
RegisterVerb adds or replaces a verb handler identified by stepType in the coordinator's registry.
func (*Saga) Shutdown
func (s *Saga) Shutdown(ctx context.Context) error
Shutdown cancels the Saga's background context (so in-flight background advances stop between steps) and waits for them to drain, bounded by ctx. Returns ctx.Err() if the drain does not complete before ctx is done. After Shutdown the Saga should not be reused.
func (*Saga) Signal
func (s *Saga) Signal(ctx context.Context, runID uuid.UUID, name string, payload map[string]any) error
Signal delivers an external signal to a run. If the run was paused awaiting exactly this signal name, it is consumed and the run advances.
func (*Saga) Start
func (s *Saga) Start(ctx context.Context, workflowID string, inputs map[string]any) (uuid.UUID, error)
func (*Saga) StartAt
func (s *Saga) StartAt(ctx context.Context, workflowID, entrypoint string, inputs map[string]any) (uuid.UUID, error)
StartAt creates a run beginning at the named entry point ("" => default/Start) and advances it once (synchronously to the first pause or terminal state).
func (*Saga) Store
func (s *Saga) Store() store.Store
Store returns the underlying store.Store for direct access.
Generated by gomarkdoc