Read one inbox row.

List inbox rows for a source, ordered by received_at. Test helper.

Delete completed inbox rows older than keepFor from now.

Returns the number of rows deleted. The retention window defines the duplicate-detection window: a redelivery that arrives after retention GC has run will be processed again, so the window must exceed the maximum delivery delay tolerated by the operator. The user guide recommends 30 days as a default. See tryInsertCompletedTx for the related concurrent-GC race and its at-least-once consequence.

Count inbox rows in a non-terminal state (backlog gauge source).

Backlog = rows still processing (in flight) or failed (awaiting a retry decision). Completed rows are terminal and excluded.

Mark an inbox row failed inside the same transaction as the handler.

Run handler at most once for each (source, dedupe_key).

Computes the dedupe key from policy and kafka, then in one transaction:

• Inserts the inbox row with status completed.
• If the row already exists, branches on its status: InboxCompleted → InboxDuplicate; InboxProcessing → InboxInProgress; InboxFailed → InboxPreviouslyFailed.
• Otherwise runs handler; the row commits only if the handler succeeds.

The handler is invoked with the decoded IntegrationEvent so it does not need to redecode bytes. On exception or condemn the whole transaction rolls back, including the inbox row insert — the next delivery sees no row and can retry.

Variant of runInboxTransaction that controls success-path envelope persistence.

PersistDedupeOnly keeps enough columns for dedupe and operator correlation but stores an empty payload and omits schema, trace, and attribute columns for successfully processed rows.

Lower-level variant that takes the dedupe key directly.

Use when the policy is not enough to express the identity scheme — for example, when the consumer joins fields from multiple headers or derives the key from the payload itself.

Run handler with opt-in poison-message accounting.

This wrapper behaves like runInboxTransaction for fresh messages, duplicates, and in-flight rows, but changes the behavior for handler exceptions and previously failed rows:

• A synchronous exception from handler rolls back the handler transaction, then records a failed attempt in a second transaction and returns InboxHandlerFailed with the new attempt count.
• A previously failed row with attempt_count < ceiling is retried.
• A previously failed row with attempt_count >= ceiling returns InboxPreviouslyFailed without running the handler. The consumer can commit its offset and move on; the failed inbox row is the dead-letter record for operator review.

condemn is not treated as a handler failure by this wrapper. It keeps the original rollback semantics from runInboxTransaction.

Variant of runInboxTransactionWithRetries that controls success-path persistence.

Lower-level retrying variant that takes the dedupe key directly.

Process a batch of inbox deliveries with a single transactional fast path.

The fast path computes each (source, dedupe_key), suppresses repeated keys within the batch as duplicates, then runs all remaining deliveries in one Postgres transaction. If any handler throws or condemns that transaction, the whole batch rolls back and every original delivery is retried through runInboxTransactionWithRetries. That fallback preserves per-message failure accounting and prevents one poison message from discarding unrelated batch mates.

condemn rolls the transaction back at commit but returns normally, so it cannot be observed from the transaction's return value alone. The batch detects it by re-reading one row it should have committed: every write in the fast path belongs to a delivery classified InboxProcessed (fresh insert as completed or retry promotion to completed), so if the first such row is not completed after the transaction returns, the whole batch was condemned and the per-message fallback runs. A batch with no InboxProcessed rows performed no writes, so a condemned transaction loses nothing.

Count the inbox backlog and record the gauge when metrics are enabled.

The backlog is non-terminal rows: legacy processing rows plus failed rows. Schedule this on its own interval; it is intentionally not part of the per-message intake path.