Documentation
¶
Overview ¶
Package worker implements the outbox consumer (ADR-016 §Class 2): a SEPARATE process (cmd/appximo-worker) that drains rows the engine wrote to public.outbox and runs each through a Processor. It is deliberately NOT a goroutine inside the engine — it survives engine restarts and never touches the engine's request hot path (the engine only WRITES to the outbox; the worker only READS).
Delivery model — at-least-once:
LISTEN outbox.NotifyChannel ← wake-up hint (ephemeral; lost while down)
+ periodic poll fallback ← the TABLE is the durable source of truth
→ SELECT … WHERE state='pending' FOR UPDATE SKIP LOCKED LIMIT N (one tx)
→ Processor.Process(row) per row
→ UPDATE … sent_at=now(), state='sent' (success)
UPDATE … attempts+1 [, state='failed'] (failure / exhausted)
→ COMMIT
FOR UPDATE SKIP LOCKED makes concurrent workers claim DISJOINT rows, so a row is processed by at most one worker at a time. A worker that dies mid-batch aborts its tx; the locks release, sent_at stays NULL, and another worker reclaims the row — hence delivery is at-least-once and Processors MUST be idempotent (see Processor).
Index ¶
Constants ¶
const DefaultServiceTokenTTL = 60 * time.Second
DefaultServiceTokenTTL is how long a worker's service JWT is valid. Kept SHORT (60s, minted per operation) so there is no long-lived credential to leak — ADR-016 §Class 2 write-back.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type Beginner ¶
Beginner is any pgx handle that can start a transaction — both *pgx.Conn and *pgxpool.Pool satisfy it. Drain takes a Beginner so tests can pass a pool (to exercise concurrent SKIP LOCKED) while the running worker passes a dedicated *pgx.Conn.
type Config ¶
type Config struct {
Channel string // LISTEN channel; default outbox.NotifyChannel
BatchSize int // rows claimed per Drain; default 50
MaxAttempts int // attempts before a row is parked 'failed'; default 5
PollInterval time.Duration // fallback poll cadence; default 5s
BackoffMin time.Duration // reconnect backoff floor; default 2s
BackoffMax time.Duration // reconnect backoff ceiling; default 30s
}
Config tunes the worker. Zero values fall back to the defaults applied by New.
type Connector ¶
Connector opens a fresh, DEDICATED Postgres connection. The worker calls it for the LISTEN connection and for the drain connection, and again after a drop. It must NOT hand out a pooled connection: a listener needs a permanent connection, and a pool silently breaks LISTEN by rotating the underlying conn (ADR-016 / the outbox investigation).
type DrainResult ¶
type DrainResult struct {
Processed int // rows marked 'sent' this batch
Failed int // rows that errored this batch (incl. those parked 'failed')
}
DrainResult summarizes one Drain (one transaction / one batch).
func Drain ¶
func Drain(ctx context.Context, db Beginner, proc Processor, batchSize, maxAttempts int, log *zerolog.Logger) (DrainResult, error)
Drain claims up to batchSize pending rows with FOR UPDATE SKIP LOCKED, runs each through proc, and records the outcome — all inside ONE transaction. A processing error does NOT abort the batch: that row's attempts is incremented (and its state flips to 'failed' once attempts reaches maxAttempts) while the batch's other rows still commit.
db may be any pgx handle that can Begin a tx. For concurrent draining, call Drain from multiple goroutines on a *pgxpool.Pool (or on separate *pgx.Conn): SKIP LOCKED guarantees each row is claimed by at most one of them. log may be nil. Returns the per-batch tallies.
func (DrainResult) Claimed ¶
func (r DrainResult) Claimed() int
Claimed is the number of rows locked this batch (Processed + Failed).
type EngineClient ¶
type EngineClient struct {
// contains filtered or unexported fields
}
EngineClient lets the worker write BACK to the engine through its HTTP API (not the tenant DB directly) so the write inherits the engine's validation + RBAC + the same path any client takes. For each call it mints a fresh, short-lived, SCOPED service JWT (the shared JWT_SECRET, a tenant-scoped service role — NOT admin) and sets the tenant Host header. The engine resolves the tenant from the Host subdomain and enforces the service role's RBAC.
func NewEngineClient ¶
func NewEngineClient(baseURL, tenantDomain, jwtSecret, role string, ttl time.Duration) *EngineClient
NewEngineClient builds a client. role MUST be a scoped service role defined in the schema RBAC (minimal actions/resources), never an admin role. Zero ttl → DefaultServiceTokenTTL.
func (*EngineClient) Do ¶
func (c *EngineClient) Do(ctx context.Context, tenant, method, path string, body any) (int, []byte, error)
Do performs an authenticated request to the engine API for tenant. body is JSON-encoded when non-nil. It returns the HTTP status and response body; a transport error (not an HTTP error status) returns a non-nil error. The caller decides what a non-2xx means — for the outbox, a non-2xx must NOT mark the row sent (it is retried, at-least-once).
type Processor ¶
Processor runs the business logic for one outbox Row.
Idempotency is MANDATORY. Because delivery is at-least-once (a worker that crashes after the side-effect but before COMMIT releases the lock with sent_at still NULL, so the row is redelivered), Process may be called more than once for the same Row.ID. A real consumer with external side-effects must dedupe on a stable idempotency key — Row.ID is the natural choice: record "event <id> handled" in the consumer's own table (ideally in the SAME tx as a write-back side-effect) and no-op on the second delivery.
The echo consumer is trivially idempotent: it only logs, so a redelivery just logs twice. Returning a non-nil error keeps the row pending (sent_at stays NULL, attempts incremented) for retry until maxAttempts, after which the row is parked in state='failed' and never retried again.
type ProcessorFunc ¶
ProcessorFunc adapts a plain function to the Processor interface.
type Row ¶
type Row struct {
ID int64 // public.outbox.id — also the natural idempotency key
TenantID string // owning tenant, read from the row (not re-derived)
Topic string // e.g. "echo.test"
Payload json.RawMessage // the JSONB payload, copied out of pgx's buffer
Attempts int // delivery attempts BEFORE this one (0 on the first)
}
Row is one claimed outbox event handed to a Processor.
type Worker ¶
type Worker struct {
// contains filtered or unexported fields
}
Worker consumes public.outbox via LISTEN/NOTIFY with a polling fallback.
type WritebackProcessor ¶
type WritebackProcessor struct {
// contains filtered or unexported fields
}
WritebackProcessor is the SERVICE-JWT-V1 demonstration consumer: on a "{resource}.created" outbox event it mints a scoped service JWT and PATCHes the created row's status back through the engine API — proving the authenticated write-back chain (event → worker → mint JWT → engine API → RBAC accepts the scoped role → row marked sent). It is NOT real business logic; a real consumer (XLSX/email) replaces the side-effect and dedups on Row.ID (idempotency key).
Idempotency: setting status to a fixed value is naturally idempotent, which is what at-least-once delivery requires. Any non-2xx (or transport error) returns an error so the row stays pending and is retried — never marked sent on failure.
func NewWritebackProcessor ¶
func NewWritebackProcessor(client *EngineClient, statusValue string, log zerolog.Logger) *WritebackProcessor
NewWritebackProcessor builds the demo consumer. statusValue defaults to "done".
Source Files