api
import "github.com/Bugs5382/go-saga-orchestration/api"
Package api — WebSocket stream handler for the run inspector.
Auth is intentionally not enforced here; real authentication middleware should be added before production deployment. See the route comment in router.go.
Package api wires the go-saga-orchestration REST API. Routes: health probes + /api/v1/sagas.
Constants
Error codes used in the JSON error envelope (the `error` field written by WriteError). Stable, machine-readable identifiers shared across handlers.
const (
CodeBadRequest = "bad_request"
CodeNotFound = "not_found"
CodeForbidden = "forbidden"
CodeInternal = "internal"
CodeInvalidConfig = "invalid_config"
CodePublishFailed = "publish_failed"
CodeUnprocessable = "unprocessable"
CodeConflict = "conflict"
)
func HealthLive
func HealthLive(w http.ResponseWriter, _ *http.Request)
HealthLive returns 200 if the process is running.
func HealthReady
func HealthReady(w http.ResponseWriter, _ *http.Request)
HealthReady returns 200 once the service is ready to serve. v1 simple: always ready after process start. Future: gate on Postgres ping + a "registry warm" signal.
func NewRouter
func NewRouter(_ store.Store, sagas *SagaHandler, signals *SignalHandler, userTasks *UserTaskHandler, registryHandler *RegistryHandler, rulesHandler *RulesHandler, triggersHandler *TriggerHandler, streamHandler *SagaStreamHandler, workflows *WorkflowHandler, actionResults *ActionResultHandler) *chi.Mux
NewRouter builds the chi router. Saga routes attach via SagaHandler; registry routes attach via RegistryHandler; rule evaluation via RulesHandler; trigger CRUD via TriggerHandler; live run inspector via SagaStreamHandler; workflow stats via WorkflowHandler.
func WriteError
func WriteError(w http.ResponseWriter, status int, code, message string)
WriteError sends a structured error envelope.
func WriteJSON
func WriteJSON(w http.ResponseWriter, status int, v any)
WriteJSON serialises v as JSON with the given status. Errors during encoding are written to the response but cannot change the already-sent header.
type ActionResultHandler
ActionResultHandler accepts a transport-agnostic, asynchronous result callback for an action step. gRPC workers reply over the ExecuteStep stream; http and rmq workers have no return stream, so they report their result here. Both paths converge on the same CompleteAction / FailAction store hooks the gRPC server uses, preserving attempt handling and idempotency. (issue #59)
type ActionResultHandler struct {
S store.Store
Publisher AdvancePublisher
}
func NewActionResultHandler
func NewActionResultHandler(s store.Store, p AdvancePublisher) *ActionResultHandler
NewActionResultHandler returns an ActionResultHandler backed by the given store and advance publisher.
func (*ActionResultHandler) Post
func (h *ActionResultHandler) Post(w http.ResponseWriter, r *http.Request)
Post handles POST /api/v1/sagas/{run_id}/actions/{step_id}/result.
Body (exactly one of):
{ "result": { ... } } -> CompleteAction
{ "error": { "code", "message", "retryable" } } -> FailAction
An optional "attempt" pins the report to a specific dispatch attempt; omitted, it uses the run's current attempt. Mirrors the gRPC Complete/Error semantics (attempt handling + idempotency are enforced by the store hooks: a stale attempt is a no-op).
Responses:
202 — result recorded (success completed, or failure transitioned).
400 — bad run_id or malformed body.
500 — internal error.
type AdvancePublisher
AdvancePublisher abstracts the RabbitMQ publisher so handlers can be unit-tested without a broker.
type AdvancePublisher interface {
PublishSagaAdvance(ctx context.Context, runID string) error
}
type Canceller
Canceller cancels an in-flight run from outside the run. It is satisfied by *engine.Coordinator; handlers depend on the interface so they can be unit-tested with a fake. See engine.Coordinator.Cancel.
type Canceller interface {
Cancel(ctx context.Context, runID uuid.UUID, reason string) error
}
type RegistryHandler
RegistryHandler serves the action-registry REST surface.
POST /api/v1/registry/register — services call this on startup to
register or refresh their action declarations.
GET /api/v1/registry/actions — read-only browser feed for admin/builder UIs.
type RegistryHandler struct {
S store.Store
}
func NewRegistryHandler
func NewRegistryHandler(s store.Store) *RegistryHandler
NewRegistryHandler returns a RegistryHandler backed by the given store.
func (*RegistryHandler) List
func (h *RegistryHandler) List(w http.ResponseWriter, r *http.Request)
List returns the registered actions filtered by service/category/search.
GET /api/v1/registry/actions?service=&category=&search=
func (*RegistryHandler) Register
func (h *RegistryHandler) Register(w http.ResponseWriter, r *http.Request)
Register accepts a service's startup payload. Upserts each action by (service, action_name, version). Idempotent — services may resend on every restart.
type RulesHandler
RulesHandler exposes the rules-package evaluator as a sync REST API.
type RulesHandler struct {
S store.Store
}
func NewRulesHandler
func NewRulesHandler(s store.Store) *RulesHandler
NewRulesHandler returns a RulesHandler backed by the given store.
func (*RulesHandler) Evaluate
func (h *RulesHandler) Evaluate(w http.ResponseWriter, r *http.Request)
Evaluate handles POST /api/v1/rules/{rule_id}/evaluate. Responses:
200 — rule found + evaluated; body {output, audit}.
404 — rule_id not found.
400 — bad request body.
422 — rule evaluation failed (CEL compile/eval error, no_decision_row_matched).
500 — internal error.
type SagaHandler
SagaHandler owns the /api/v1/sagas/* routes.
type SagaHandler struct {
// contains filtered or unexported fields
}
func NewSagaHandler
func NewSagaHandler(s store.Store, p AdvancePublisher, providers ...engine.StartupVariableProvider) *SagaHandler
NewSagaHandler constructs the handler. Optional StartupVariableProviders are invoked at saga start to inject per-tenant "magic" variables.
func (*SagaHandler) Cancel
func (h *SagaHandler) Cancel(w http.ResponseWriter, r *http.Request)
Cancel handles POST /api/v1/sagas/{id}/cancel. It routes to the coordinator's idempotent run-level cancel: a terminal run is a no-op and a cancelled child re-evaluates its parent's join. Returns 202 on success, 400 on a bad id, 404 when the run does not exist, and 501 when no canceller is wired.
func (*SagaHandler) Get
func (h *SagaHandler) Get(w http.ResponseWriter, r *http.Request)
Get handles GET /api/v1/sagas/{id}.
func (*SagaHandler) List
func (h *SagaHandler) List(w http.ResponseWriter, r *http.Request)
List handles GET /api/v1/sagas. Parses optional filter query params, calls store.ListRuns + store.CountRuns, returns paginated JSON.
func (*SagaHandler) Start
func (h *SagaHandler) Start(w http.ResponseWriter, r *http.Request)
Start handles POST /api/v1/sagas/start. Resolves the published definition, inserts a saga_runs row, publishes saga.advance, returns 202 with the run ID.
func (*SagaHandler) WithCanceller
func (h *SagaHandler) WithCanceller(c Canceller) *SagaHandler
WithCanceller attaches the Canceller used by the Cancel handler and returns the receiver for chaining. When unset, POST /sagas/{id}/cancel returns 501.
type SagaStreamHandler
SagaStreamHandler upgrades GET /api/v1/sagas/{run_id}/stream to a WebSocket. On connect: sends the current SagaRun snapshot, then tails audit.saga_run_events via Postgres LISTEN/NOTIFY (channel "saga_event_<run_id_no_dashes>"). One LISTEN per connection — the per-connection pgx Conn is acquired from a dedicated channel pool the handler holds.
Auth is intentionally not enforced today; wire real auth middleware before exposing this endpoint in production.
type SagaStreamHandler struct {
S store.Store
Pool *pgxpool.Pool // for LISTEN — acquire a dedicated conn per stream
Upgrade websocket.Upgrader
}
func NewSagaStreamHandler
func NewSagaStreamHandler(s store.Store, pool *pgxpool.Pool) *SagaStreamHandler
NewSagaStreamHandler constructs the handler.
func (*SagaStreamHandler) Stream
func (h *SagaStreamHandler) Stream(w http.ResponseWriter, r *http.Request)
Stream handles GET /api/v1/sagas/{run_id}/stream.
type SignalHandler
SignalHandler accepts external signals delivered to a saga.
type SignalHandler struct {
S store.Store
Publisher AdvancePublisher
}
func NewSignalHandler
func NewSignalHandler(s store.Store, p AdvancePublisher) *SignalHandler
NewSignalHandler returns a SignalHandler backed by the given store and advance publisher.
func (*SignalHandler) Post
func (h *SignalHandler) Post(w http.ResponseWriter, r *http.Request)
Post handles POST /api/v1/sagas/{run_id}/signal/{name}. Responses:
202 — signal recorded AND matched a paused saga awaiting it (advance published).
409 — signal recorded but the run wasn't paused-and-awaiting this name.
400 — bad run_id.
404 — run not found (only when AppendSignal returns a not-found error).
500 — internal error.
type TriggerHandler
TriggerHandler serves the saga-trigger REST surface.
POST /api/v1/triggers — create
GET /api/v1/triggers — list (optional ?type= ?enabled=)
GET /api/v1/triggers/{id} — get one
DELETE /api/v1/triggers/{id} — remove
type TriggerHandler struct {
S store.Store
Licensing licensing.Resolver
Clock clock.Clock
}
func NewTriggerHandler
func NewTriggerHandler(s store.Store, lr licensing.Resolver, clk clock.Clock) *TriggerHandler
NewTriggerHandler constructs the handler.
func (*TriggerHandler) Create
func (h *TriggerHandler) Create(w http.ResponseWriter, r *http.Request)
Create handles POST /api/v1/triggers.
func (*TriggerHandler) Delete
func (h *TriggerHandler) Delete(w http.ResponseWriter, r *http.Request)
Delete handles DELETE /api/v1/triggers/{id}. Returns 204 on success, 404 if no row exists (mirrors the non-idempotent pattern — store.DeleteTrigger returns ErrNotFound for missing rows).
func (*TriggerHandler) Get
func (h *TriggerHandler) Get(w http.ResponseWriter, r *http.Request)
Get handles GET /api/v1/triggers/{id}.
func (*TriggerHandler) List
func (h *TriggerHandler) List(w http.ResponseWriter, r *http.Request)
List handles GET /api/v1/triggers. Optional query params: ?type= ?enabled=true|false
type UserTaskHandler
UserTaskHandler accepts task submissions from assignees. On submit:
- Persist the task's submitted_at / submitted_by / result.
- Append a saga_signal of name `user_task.{task_id}.submitted` to the run, carrying the result as the signal payload.
- Try to consume the awaited signal; if it matches, publish saga.advance.
type UserTaskHandler struct {
S store.Store
Publisher AdvancePublisher
}
func NewUserTaskHandler
func NewUserTaskHandler(s store.Store, p AdvancePublisher) *UserTaskHandler
NewUserTaskHandler constructs the handler.
func (*UserTaskHandler) Submit
func (h *UserTaskHandler) Submit(w http.ResponseWriter, r *http.Request)
Submit handles POST /api/v1/sagas/{run_id}/user_task/{task_id}/submit.
type WorkflowHandler
WorkflowHandler owns workflow-level aggregate routes.
type WorkflowHandler struct {
S store.Store
}
func NewWorkflowHandler
func NewWorkflowHandler(s store.Store) *WorkflowHandler
NewWorkflowHandler constructs a WorkflowHandler.
func (*WorkflowHandler) Stats
func (h *WorkflowHandler) Stats(w http.ResponseWriter, r *http.Request)
Stats handles GET /api/v1/workflows/{wf_id}/stats. Returns aggregate metrics: success_rate_24h, last_run_at, in_flight.
Generated by gomarkdoc