Enqueue one integration event inside an existing transaction.

The (source, message_id) unique constraint catches duplicate retries from a saga/process-manager. Callers that mint a fresh messageId per attempt should also mint a fresh outboxId; callers that want idempotent retries should reuse both.

The row's created_at value is the PostgreSQL transaction-start time. The per-key and per-source publisher policies therefore provide only best-effort ordering when concurrent transactions enqueue the same key/source and commit in the opposite order. Serialize those transactions when strict order matters; the canonical producer subscription already does so.

Claim up to limit rows ready for publish.

Rows in pending or failed status whose next_attempt_at has passed become candidates. The selection is filtered by OrderingPolicy:

• PerKeyHeadOfLine — a row is claimed only if every earlier non-terminal row with the same (source, message_key) is also claimed by the same statement. Rows with message_key IS NULL bypass the per-key check.
• PerSourceStream — a row is claimed only if every earlier non-terminal row in the same source is also claimed by the same statement, regardless of key.
• StopTheLine — same as PerKeyHeadOfLine at claim time; the worker halts on the first failure (decided at the worker level).
• BestEffort — no head-of-line predicate.

The returned list preserves (created_at, outbox_id) order. Per-key and per-source subsequences are therefore gapless ordered runs. The LIMIT applies to the locked candidate set before the post-filter; under concurrent claimers or a limit cut through the middle of a run, a pass can return fewer than limit rows even when more rows are ready.

Claimed rows are transitioned to publishing and have their attempt_count incremented atomically.

Reclaim rows stranded in publishing longer than olderThan.

Rows whose claim already consumed the attempt budget are dead-lettered; the rest return to failed so the regular claim query can retry them. Returns (requeued, deadLettered).

Mark a row as successfully published. Sets published_at and clears last_error. Returns False if the row left publishing before the mark, for example because a stale-row sweeper or operator changed it while the transport publish was in flight. The publish may still have happened; callers must treat this as at-least-once delivery.

Read a single outbox row by id. Used by tests and inspection tooling.

List outbox rows for a source, ordered by created_at. Used by tests; not intended for application traffic.

List rows a stale-publisher recovery pass would consider, oldest first. This is a read-only operator preview for requeueStuckOutbox.

List sent rows a retention pass would delete, oldest first. This is a read-only operator preview for garbageCollectSent.

Count outbox rows awaiting publish (backlog gauge source).

Backlog = rows in a claimable, non-terminal state. Mirrors the claim query's status IN (pending,failed) predicate so the gauge measures exactly the rows a publisher still has to drain (rows held mid-pass in publishing, and the terminal sent/dead rows, are excluded).

Delete sent rows whose published_at is older than keepFor before now.

Returns the number of rows deleted. dead rows are never deleted: they are operator action items proving an event was not published. The retention window only bounds how long successful publish history remains queryable; consumer dedupe lives in the inbox, not here.

Mint a fresh time-ordered UUIDv7 for use as an OutboxId.

Enqueue an IntegrationEvent from a saga or process manager that is already running inside a runCommandWithSqlEvents transaction. The caller supplies a stable OutboxId so retried command attempts coalesce on the (source, message_id) unique constraint.

Ordering caveat: under PerKeyHeadOfLine and PerSourceStream, the publisher orders rows by created_at, which PostgreSQL sets to transaction-start time. If two concurrent transactions enqueue the same key/source and commit in the opposite order, a publisher can observe that order. Serialize same-key enqueues when strict order matters.

Configuration for the canonical producer subscription.

A service running IntegrationProducer reads its private event stream, decodes each event with a Codec, calls mapEvent, and for each Just result writes one keiro_outbox row. The helper mints messageId on each insert so the id is stable across publish retries.

• name — subscription name used to checkpoint the producer's cursor in the subscriptions table.
• source — value written into keiro_outbox.source; identifies the producing bounded context.
• messageIdPrefix — TypeID prefix used when minting messageId. Must be 1-63 lowercase Latin letters (e.g. "msg", "order"). Prefer constructing producers with mkIntegrationProducer; an invalid prefix passed directly to IntegrationProducer raises when the first message id is minted.
• mapEvent — pure mapper from a private RecordedEvent and its decoded payload to an IntegrationEventDraft. Returning Nothing skips the event without enqueuing a row.

Everything in IntegrationEvent except messageId and source — those are filled in by mintIntegrationEvent from the producer configuration and the freshly minted TypeID.

sourceEventId and sourceGlobalPosition default to the values on the underlying RecordedEvent (see mintIntegrationEvent); a mapper that needs to override them can replace the draft fields directly.

Validate an integration producer before starting its subscription.

Mint a fresh messageId (TypeID with the producer's prefix) and build the full IntegrationEvent from the draft. Lives in IO because TypeID generation reads the global UUIDv7 sequence counter.

Build an IntegrationEvent from a source, a minted message id, and a draft.

Enqueue one drafted producer event inside an existing transaction.

This is the primitive a subscription worker calls per event. It mints a fresh messageId (TypeID), constructs the full envelope, and inserts the row. The caller supplies the OutboxId so retries from a known subscription cursor coalesce on (source, message_id).

The TypeID is minted before the insert; if the transaction rolls back the message id is discarded (no observable effect) and the next attempt mints a different id. Idempotency at the row level relies on a stable OutboxId, not the minted message id.

Ordering caveat: created_at records transaction-start time. Under PerKeyHeadOfLine or PerSourceStream, concurrent transactions for the same key/source can commit in the opposite order and are therefore best-effort unless the caller serializes them. The canonical producer subscription does serialize same-key enqueues.

Result of one publish attempt as reported by the transport-specific publisher.

Drain claimed outbox rows by handing the claimed batch to publish and reflecting the outcomes back into row statuses.

Claims rows in batches of batchSize under the active OrderingPolicy, calls publish with the claimed rows in claim order, and marks every row sent or — using markOutboxFailedTx — failed/dead. The publish result must contain one outcome per input row; a missing outcome is treated as PublishFailed "publisher returned no outcome". If the publisher throws, every row in that call is treated as failed with the exception text.

For ordered policies, if a row fails then later rows in the same ordered group are skipped and returned to failed without consuming an attempt; these skipped rows count as retried. A real Kafka transport must not successfully deliver a later same-key record after reporting an earlier same-key failure from the same call. On StopTheLine, the worker calls publish with singleton batches and halts after the first failed row, recording the offending OutboxId in haltedOn.

Returns when one of:

• No rows are claimable.
• The active policy is StopTheLine and a publish failed.

The worker does not loop indefinitely; the application is expected to schedule it repeatedly (e.g. once per process-compose tick).

Reclaim crashed publisher rows and record the outbox backlog gauge.

Schedule this pass independently from publishClaimedOutbox, typically on a slower timer. It is the only library worker path that reclaims rows stranded in publishing.

Count publishable rows and record the outbox backlog gauge when metrics are enabled.