api

Documentation

Overview

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.

Index

• Constants
• func HealthLive(w http.ResponseWriter, _ *http.Request)
• func HealthReady(w http.ResponseWriter, _ *http.Request)
• func NewRouter(_ store.Store, sagas *SagaHandler, signals *SignalHandler, ...) *chi.Mux
• func WriteError(w http.ResponseWriter, status int, code, message string)
• func WriteJSON(w http.ResponseWriter, status int, v any)
• type AdvancePublisher
• type RegistryHandler
  • func NewRegistryHandler(s store.Store) *RegistryHandler
  • func (h *RegistryHandler) List(w http.ResponseWriter, r *http.Request)
  • func (h *RegistryHandler) Register(w http.ResponseWriter, r *http.Request)
• type RulesHandler
  • func NewRulesHandler(s store.Store) *RulesHandler
  • func (h *RulesHandler) Evaluate(w http.ResponseWriter, r *http.Request)
• type SagaHandler
  • func NewSagaHandler(s store.Store, p AdvancePublisher, providers ...engine.StartupVariableProvider) *SagaHandler
  • func (h *SagaHandler) Get(w http.ResponseWriter, r *http.Request)
  • func (h *SagaHandler) List(w http.ResponseWriter, r *http.Request)
  • func (h *SagaHandler) Start(w http.ResponseWriter, r *http.Request)
• type SagaStreamHandler
  • func NewSagaStreamHandler(s store.Store, pool *pgxpool.Pool) *SagaStreamHandler
  • func (h *SagaStreamHandler) Stream(w http.ResponseWriter, r *http.Request)
• type SignalHandler
  • func NewSignalHandler(s store.Store, p AdvancePublisher) *SignalHandler
  • func (h *SignalHandler) Post(w http.ResponseWriter, r *http.Request)
• type TriggerHandler
  • func NewTriggerHandler(s store.Store) *TriggerHandler
  • func (h *TriggerHandler) Create(w http.ResponseWriter, r *http.Request)
  • func (h *TriggerHandler) Delete(w http.ResponseWriter, r *http.Request)
  • func (h *TriggerHandler) Get(w http.ResponseWriter, r *http.Request)
  • func (h *TriggerHandler) List(w http.ResponseWriter, r *http.Request)
• type UserTaskHandler
  • func NewUserTaskHandler(s store.Store, p AdvancePublisher) *UserTaskHandler
  • func (h *UserTaskHandler) Submit(w http.ResponseWriter, r *http.Request)
• type WorkflowHandler
  • func NewWorkflowHandler(s store.Store) *WorkflowHandler
  • func (h *WorkflowHandler) Stats(w http.ResponseWriter, r *http.Request)

Constants

const (
	CodeBadRequest    = "bad_request"
	CodeNotFound      = "not_found"
	CodeInternal      = "internal"
	CodeInvalidConfig = "invalid_config"
	CodePublishFailed = "publish_failed"
	CodeUnprocessable = "unprocessable"
	CodeConflict      = "conflict"
)

Error codes used in the JSON error envelope (the `error` field written by WriteError). Stable, machine-readable identifiers shared across handlers.

Functions

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) *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.

Types

type AdvancePublisher

type AdvancePublisher interface {
	PublishSagaAdvance(ctx context.Context, runID string) error
}

AdvancePublisher abstracts the RabbitMQ publisher so handlers can be unit-tested without a broker.

type RegistryHandler

type RegistryHandler struct {
	S store.Store
}

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.

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

type RulesHandler struct {
	S store.Store
}

RulesHandler exposes the rules-package evaluator as a sync REST API.

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

type SagaHandler struct {
	// contains filtered or unexported fields
}

SagaHandler owns the /api/v1/sagas/* routes.

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) 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.

type SagaStreamHandler

type SagaStreamHandler struct {
	S       store.Store
	Pool    *pgxpool.Pool // for LISTEN — acquire a dedicated conn per stream
	Upgrade websocket.Upgrader
}

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.

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

type SignalHandler struct {
	S         store.Store
	Publisher AdvancePublisher
}

SignalHandler accepts external signals delivered to a saga.

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

type TriggerHandler struct {
	S store.Store
}

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

func NewTriggerHandler

func NewTriggerHandler(s store.Store) *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

type UserTaskHandler struct {
	S         store.Store
	Publisher AdvancePublisher
}

UserTaskHandler accepts task submissions from assignees. On submit:

1. Persist the task's submitted_at / submitted_by / result.
2. Append a saga_signal of name `user_task.{task_id}.submitted` to the run, carrying the result as the signal payload.
3. Try to consume the awaited signal; if it matches, publish saga.advance.

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

type WorkflowHandler struct {
	S store.Store
}

WorkflowHandler owns workflow-level aggregate routes.

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.