Skip to main content
Version: 0.4.0

domain

import "github.com/Bugs5382/go-saga-orchestration/domain"

Package domain contains the in-process types go-saga-orchestration exchanges with its HTTP layer, store, and engine. Definitions and runs are the core concepts.

Package domain contains the core domain types for go-saga-orchestration.

Constants​

Dispatch transport values for ActionRegistration.Transport.

const (
    // TransportGRPC is the zero-config default: the worker is reachable over
    // the gRPC ExecuteStep stream. Equivalent to an empty Transport.
    TransportGRPC = "grpc"
    // TransportHTTP dispatches the action by POSTing the payload to Address.
    TransportHTTP = "http"
    // TransportRMQ dispatches the action by publishing the payload to the
    // RabbitMQ queue named by Address.
    TransportRMQ = "rmq"
)

type ActionRegistration​

ActionRegistration describes one action a service exposes. Stored in definitions.action_registry.

type ActionRegistration struct {
    ID               uuid.UUID      `json:"id"`
    Service          string         `json:"service"`
    ActionName       string         `json:"action_name"`
    Version          int            `json:"version"`
    Description      string         `json:"description,omitempty"`
    Category         string         `json:"category,omitempty"`
    Compensable      bool           `json:"compensable"`
    InputSchema      map[string]any `json:"input_schema"`
    OutputSchema     map[string]any `json:"output_schema"`
    ErrorCodes       []string       `json:"error_codes,omitempty"`
    DefaultRetry     *RetryPolicy   `json:"default_retry,omitempty"`
    DefaultTimeoutMS int            `json:"default_timeout_ms,omitempty"`
    Deprecated       bool           `json:"deprecated"`
    RegisteredAt     time.Time      `json:"registered_at"`
    ServiceVersion   string         `json:"service_version,omitempty"`
    DryRunSupported  bool           `json:"dry_run_supported"`
    LicenseGroup     string         `json:"license_group,omitempty"`

    // Transport is the optional dispatch descriptor: how the coordinator
    // reaches the worker that runs this action. One of "grpc", "http" or
    // "rmq". Empty means grpc — the zero-config default where the worker
    // is connected over the gRPC ExecuteStep stream. (issue #59)
    Transport string `json:"transport,omitempty"`
    // Address is the dispatch target for non-gRPC transports: a callback
    // URL for "http" or a queue name for "rmq". Required only when
    // Transport is "http" or "rmq"; ignored for "grpc". (issue #59)
    Address string `json:"address,omitempty"`
}

type Branch​

Branch is one outgoing edge of a decision/parallel step.

type Branch struct {
    Next string `json:"next"`
}

type Compensation​

Compensation describes how to undo a completed step. Nil means "non-compensable; log a warning when rolling back."

type Compensation struct {
    Action string         `json:"action"`
    Inputs map[string]any `json:"inputs,omitempty"`
}

type DecisionTableRow​

DecisionTableRow is one row in a decision_table. The When expression is CEL; on truthy result the Then map becomes the rule's output.

type DecisionTableRow struct {
    When string         `json:"when"`
    Then map[string]any `json:"then"`
}

type EventType​

EventType identifies what changed.

type EventType string

The EventType constants enumerate the audit events recorded for a run.

const (
    EventSagaStarted         EventType = "saga.started"
    EventStepDispatched      EventType = "step.dispatched"
    EventStepStarted         EventType = "step.started"
    EventStepSucceeded       EventType = "step.succeeded"
    EventStepFailed          EventType = "step.failed"
    EventStepSkipped         EventType = "step.skipped"
    EventStepPaused          EventType = "step.paused"
    EventRunSucceeded        EventType = "run.succeeded"
    EventRunFailed           EventType = "run.failed"
    EventRunCancelled        EventType = "run.cancelled"
    EventCompensationStarted EventType = "compensation.started"

    EventLog                 EventType = "log"
    EventMetric              EventType = "metric"
    EventRuleEvaluated       EventType = "rule.evaluated"
    EventLicenseGateRejected EventType = "license.gate.rejected"
)

type HitPolicy​

HitPolicy controls how rows in a decision_table are evaluated. v1 supports only HitPolicyFirst (return the first matched row's output).

type HitPolicy string

HitPolicyFirst returns the first matched row's output; the only policy in v1.

const (
    HitPolicyFirst HitPolicy = "first"
)

type RetryPolicy​

RetryPolicy bounds step-level retry. Defaults applied at engine time if a step omits the field.

type RetryPolicy struct {
    MaxAttempts      int     `json:"max_attempts"`
    InitialBackoffMS int     `json:"initial_backoff_ms"`
    MaxBackoffMS     int     `json:"max_backoff_ms,omitempty"`
    Multiplier       float64 `json:"multiplier,omitempty"`
    Jitter           bool    `json:"jitter,omitempty"`
}

type RuleDefinition​

RuleDefinition is one version of one rule. Stored in definitions.rule_definitions.

type RuleDefinition struct {
    ID        uuid.UUID `json:"id"`
    RuleID    string    `json:"rule_id"`
    Version   int       `json:"version"`
    TenantID  *string   `json:"tenant_id,omitempty"`
    Name      string    `json:"name"`
    RuleType  RuleType  `json:"rule_type"`
    Spec      RuleSpec  `json:"spec"`
    Published bool      `json:"published"`
    CreatedAt time.Time `json:"created_at,omitempty"`
    CreatedBy string    `json:"created_by,omitempty"`
}

func NewRuleDefinition​

func NewRuleDefinition(ruleID string, version int, name string, ruleType RuleType, spec RuleSpec, createdBy string) RuleDefinition

NewRuleDefinition returns a fresh RuleDefinition with a generated ID and current timestamp. Caller fills in the spec.

type RuleSpec​

RuleSpec is the union of rule-type-specific bodies. v1 only fills in DecisionTable; later rule types add their own fields.

type RuleSpec struct {
    HitPolicy     HitPolicy          `json:"hit_policy"`
    Rows          []DecisionTableRow `json:"rows"`
    DefaultOutput map[string]any     `json:"default_output,omitempty"`
}

type RuleType​

RuleType is the discriminant for the union of rule shapes. v1 ships only `decision_table`; v1.5+ adds `script` and `expression_tree`.

type RuleType string

RuleTypeDecisionTable is the only rule type shipped in v1.

const (
    RuleTypeDecisionTable RuleType = "decision_table"
)

type RunState​

RunState is the saga-level state.

type RunState string

The RunState constants enumerate the saga-level lifecycle states.

const (
    RunStatePending      RunState = "pending"
    RunStateRunning      RunState = "running"
    RunStatePaused       RunState = "paused"
    RunStateCompensating RunState = "compensating"
    RunStateSucceeded    RunState = "succeeded"
    RunStateFailed       RunState = "failed"
    RunStateCancelled    RunState = "cancelled"
)

func (RunState) IsTerminal​

func (s RunState) IsTerminal() bool

IsTerminal reports whether the state is final (no further transitions).

type SagaRun​

SagaRun is one running instance of a workflow. Stored in runtime.saga_runs.

type SagaRun struct {
    ID                    uuid.UUID         `json:"id"`
    WorkflowID            string            `json:"workflow_id"`
    DefinitionID          uuid.UUID         `json:"definition_id"`
    TenantID              *uuid.UUID        `json:"tenant_id,omitempty"`
    State                 RunState          `json:"state"`
    CurrentStep           string            `json:"current_step,omitempty"`
    Inputs                map[string]any    `json:"inputs"`
    Variables             map[string]any    `json:"variables"`
    StartedAt             time.Time         `json:"started_at"`
    LastEventAt           time.Time         `json:"last_event_at"`
    TerminalAt            *time.Time        `json:"terminal_at,omitempty"`
    RequiresManualReview  bool              `json:"requires_manual_review"`
    TriggerID             *uuid.UUID        `json:"trigger_id,omitempty"`
    ParentRunID           *uuid.UUID        `json:"parent_run_id,omitempty"`
    ParentStepID          *string           `json:"parent_step_id,omitempty"`
    ParentBranchID        *string           `json:"parent_branch_id,omitempty"`
    TryCatchStack         []TryCatchFrame   `json:"try_catch_stack,omitempty"`
    DryRun                bool              `json:"dry_run,omitempty"`
    WakeupAt              *time.Time        `json:"wakeup_at,omitempty"`
    AwaitedSignal         *string           `json:"awaited_signal,omitempty"`
    AwaitedEventTopic     *string           `json:"awaited_event_topic,omitempty"`
    AwaitedEventHeaders   map[string]string `json:"awaited_event_headers,omitempty"`
    FeatureOverrides      map[string]bool   `json:"feature_overrides,omitempty"`
    AwaitedActionDispatch *string           `json:"awaited_action_dispatch,omitempty"`
    CurrentAttempt        int               `json:"current_attempt,omitempty"`
    // LastError records why a run reached a terminal failed/cancelled
    // state — the failing step's error message, or the cancel reason — so a
    // run is self-describing without diffing its event log. nil while the
    // run is non-terminal or terminated cleanly (succeeded). See issue #80.
    LastError *string `json:"last_error,omitempty"`
}

func NewSagaRun​

func NewSagaRun(workflowID string, definitionID uuid.UUID, tenantID *uuid.UUID, inputs map[string]any) SagaRun

NewSagaRun constructs a fresh run row in pending state. The caller supplies the resolved definition ID + inputs; the engine fills in CurrentStep when it picks the run up.

type SagaRunEvent​

SagaRunEvent is one audit row. Stored in audit.saga_run_events.

type SagaRunEvent struct {
    ID         uuid.UUID      `json:"id"`
    RunID      uuid.UUID      `json:"run_id"`
    StepID     string         `json:"step_id,omitempty"`
    Attempt    int            `json:"attempt"`
    EventType  EventType      `json:"event_type"`
    FromState  string         `json:"from_state,omitempty"`
    ToState    string         `json:"to_state,omitempty"`
    Actor      string         `json:"actor"`
    Metadata   map[string]any `json:"metadata,omitempty"`
    RecordedAt time.Time      `json:"recorded_at"`
}

func NewEvent​

func NewEvent(runID uuid.UUID, stepID string, attempt int, eventType EventType, actor string) SagaRunEvent

NewEvent constructs an audit event with ID + timestamp filled in.

type SagaSignal​

SagaSignal is one row in runtime.saga_signals. External code writes these via POST /sagas/{id}/signal/{name}; the engine consumes them to wake `wait_for_signal` steps.

type SagaSignal struct {
    ID         uuid.UUID      `json:"id"`
    RunID      uuid.UUID      `json:"run_id"`
    SignalName string         `json:"signal_name"`
    Payload    map[string]any `json:"payload,omitempty"`
    ReceivedAt time.Time      `json:"received_at"`
    ConsumedAt *time.Time     `json:"consumed_at,omitempty"`
}

type SagaTrigger​

SagaTrigger persists the binding between an external signal and a workflow definition. Created via REST or seeded via migration.

type SagaTrigger struct {
    ID          uuid.UUID
    TriggerType TriggerType
    WorkflowID  string
    Version     int
    // Config is shape-by-trigger-type. For TriggerRecordTransition:
    //   { "record_type": "order", "from_state": "created",
    //     "to_state": "pending_review", "input_mapping": {...} }
    Config    map[string]any
    Enabled   bool
    TenantID  *uuid.UUID
    CreatedAt time.Time
    CreatedBy string
    // Cron scheduling bookkeeping (cron triggers only).
    NextFireAt  *time.Time `json:"next_fire_at,omitempty"`
    LastFiredAt *time.Time `json:"last_fired_at,omitempty"`
}

type Step​

Step is a single node in the workflow graph.

type Step struct {
    ID           string            `json:"id"`
    Type         StepType          `json:"type"`
    Next         string            `json:"next,omitempty"`
    Action       string            `json:"action,omitempty"`
    Inputs       map[string]any    `json:"inputs,omitempty"`
    Compensation *Compensation     `json:"compensation,omitempty"`
    Retry        *RetryPolicy      `json:"retry,omitempty"`
    Branches     map[string]Branch `json:"branches,omitempty"`
    Extra        map[string]any    `json:"-"`
}

type StepType​

StepType is the discriminant for the union of step shapes.

type StepType string

The StepType constants enumerate every verb the engine can dispatch.

const (
    StepTypeEnd            StepType = "end"
    StepTypeAction         StepType = "action"
    StepTypeDecision       StepType = "decision"
    StepTypeError          StepType = "error"
    StepTypeNoop           StepType = "noop"
    StepTypeSetVar         StepType = "set_var"
    StepTypeTransform      StepType = "transform"
    StepTypeMerge          StepType = "merge"
    StepTypeFilter         StepType = "filter"
    StepTypeMap            StepType = "map"
    StepTypeLog            StepType = "log"
    StepTypeMetricEmit     StepType = "metric_emit"
    StepTypeAssert         StepType = "assert"
    StepTypeHTTPRequest    StepType = "http_request"
    StepTypeWebhookEmit    StepType = "webhook_emit"
    StepTypeWaitDuration   StepType = "wait_duration"
    StepTypeWaitUntil      StepType = "wait_until"
    StepTypeWaitForSignal  StepType = "wait_for_signal"
    StepTypeWaitForEvent   StepType = "wait_for_event"
    StepTypeParallel       StepType = "parallel"
    StepTypeForeach        StepType = "foreach"
    StepTypeWhile          StepType = "while"
    StepTypeTryCatch       StepType = "try_catch"
    StepTypeSubSaga        StepType = "sub_saga"
    StepTypeSpawnSaga      StepType = "spawn_saga"
    StepTypeManualApproval StepType = "manual_approval"
    StepTypeCollectInput   StepType = "collect_input"
    StepTypeSwitch         StepType = "switch"
    StepTypeEmitSignal     StepType = "emit_signal"
    StepTypeCancel         StepType = "cancel"
    StepTypeEmitEvent      StepType = "emit_event"
)

type TriggerFireRow​

TriggerFireRow is a single row in runtime.saga_trigger_fires.

type TriggerFireRow struct {
    ID             uuid.UUID  `json:"id"`
    TriggerID      uuid.UUID  `json:"trigger_id"`
    WorkflowID     string     `json:"workflow_id"`
    FiredAt        time.Time  `json:"fired_at"`
    ResultingRunID *uuid.UUID `json:"resulting_run_id,omitempty"`
    Error          string     `json:"error,omitempty"`
}

type TriggerType​

TriggerType is the dispatch key for matching incoming signals (events, cron ticks, etc.) to workflows.

type TriggerType string

The TriggerType constants enumerate the supported trigger dispatch keys.

const (
    // TriggerRecordTransition fires when a record state-transition event
    // arrives matching the trigger's config.record_type, config.from_state,
    // config.to_state.
    TriggerRecordTransition TriggerType = "record_transition"
    // TriggerCron fires on a schedule defined by a cron expression in
    // config.schedule (standard 5-field cron syntax, UTC).
    TriggerCron TriggerType = "cron"
)

type TryCatchFrame​

TryCatchFrame is one entry in a saga's try_catch stack. When a step inside the try block errors, the coordinator pops the top frame and jumps to its CatchStep instead of failing the saga.

type TryCatchFrame struct {
    StepID    string `json:"step_id"`    // the try_catch step itself
    CatchStep string `json:"catch_step"` // step ID to jump to on error
}

type UserTask​

UserTask is one row in runtime.saga_user_tasks. Created by the manual_approval and collect_input verbs; submitted via POST /api/v1/sagas/{run_id}/user_task/{task_id}/submit.

type UserTask struct {
    ID          uuid.UUID      `json:"id"`
    RunID       uuid.UUID      `json:"run_id"`
    StepID      string         `json:"step_id"`
    Assignee    string         `json:"assignee"`
    DueAt       *time.Time     `json:"due_at,omitempty"`
    FormSchema  map[string]any `json:"form_schema,omitempty"`
    SubmittedAt *time.Time     `json:"submitted_at,omitempty"`
    SubmittedBy string         `json:"submitted_by,omitempty"`
    Result      map[string]any `json:"result,omitempty"`
}

type WorkflowDefinition​

WorkflowDefinition is one version of one workflow. Stored in definitions.workflow_definitions.

type WorkflowDefinition struct {
    ID          string            `json:"id"`
    Version     int               `json:"version"`
    TenantID    *string           `json:"tenant_id,omitempty"`
    Name        string            `json:"name"`
    Description string            `json:"description,omitempty"`
    Start       string            `json:"start"`
    Entrypoints map[string]string `json:"entrypoints,omitempty"` // entry name -> step id; "" / "default" => Start
    Steps       []Step            `json:"steps"`
    Published   bool              `json:"published"`
    CreatedAt   time.Time         `json:"created_at,omitempty"`
    CreatedBy   string            `json:"created_by,omitempty"`
}

func (WorkflowDefinition) ResolveEntry​

func (d WorkflowDefinition) ResolveEntry(entrypoint string) (string, error)

ResolveEntry returns the step id a run should begin at for the named entry point. "" and "default" resolve to Start. An unknown name returns an error.

func (*WorkflowDefinition) StepByID​

func (d *WorkflowDefinition) StepByID(id string) (Step, bool)

StepByID returns the step with the given ID and whether it was found.

Generated by gomarkdoc