@krizaka/outbox

InstallCopy Anchor Link

npm install @krizaka/outbox

Also published as subpath exports: @krizaka/outbox/settings, @krizaka/outbox/testkit.

The transactional outbox of every Krizaka product. An event is written in the caller's transaction, so it exists exactly when the work committed, and is delivered after the commit, at least once, by workers that lease a batch. No broker: PostgreSQL and FOR UPDATE SKIP LOCKED share the work between instances. It replaces Akinaza's mail_outbox (whose flushOutbox nobody called) and the analytics outboxes the products carried; it is the TypeScript twin of krizaka-messaging's krizaka_outbox (the envelope is id, type, occurredAt, payload).

The guarantee is at least once. Consumers must be idempotent on the message id. A claim is exclusive while its lease is valid. A worker that delivers and then crashes before complete lets the lease expire, and the message is delivered again. The outbox never promises « exactly once ».

import { emitMigration, transaction } from "@krizaka/db";
import { consumeOnce, createOutbox, outboxSchema, runWorker } from "@krizaka/outbox";
import { readOutboxSettings } from "@krizaka/outbox/settings";

emitMigration(outboxSchema, 1, { tables: {} }); // commit the SQL once, in your migrations: outbox.message and outbox.processed

const outbox = createOutbox({ clock: { now: () => new Date() }, settings: () => readOutboxSettings((key) => settings.get(key)) });

// 1. in the use case: the message commits or rolls back with the work
await transaction(connect, async (tx) => {
  const redemption = await coupons.redeem(tx, { code, accountId });
  await outbox.append(tx, { topic: "credits", type: "evt.coupon.redeemed.v1", payload: { redemptionId: redemption.id }, id: redemption.eventId });
});

// 2. a worker the product starts (Next.js instrumentation.ts); the package has no timer, you pass `sleep`
const stop = new AbortController();
void runWorker({
  outbox, query: pool.query.bind(pool) /* a pool-level Query */, owner: `${hostname}:${process.pid}`, topic: "credits", signal: stop.signal,
  sleep: (seconds) => new Promise((resolve) => setTimeout(resolve, seconds * 1000)),
  deliver: async (message) => { /* throw to retry */ },
});

// 3. a consumer with an effect in the database applies it once per message id, in the effect's transaction
await transaction(connect, (tx) => consumeOnce(tx, "credits", message.id, clock.now(), () => credits.grant(tx, …)));

• append(tx, { topic, type, payload, id?, availableAt? }) takes the transaction's Query first. A given id makes the append idempotent (the same fact twice is one message). type is a routing key, evt.<module>.<fact>.v1 like Java's.
• claim(query, { owner, topic? }) is UPDATE … WHERE id IN (SELECT … FOR UPDATE SKIP LOCKED) and sets the lease. complete and fail(error) act only for the lease's owner (a worker that lost its lease cannot overwrite the new holder). fail backs off — availableAt = now + backoffSeconds × 2^(attempts−1), capped — and after maxAttempts the message is dead (kept, with its last error; retryDead gives it another life). A message whose last allowed attempt's lease expired with no answer is dead too, not delivered a ninth time.
• runWorker loops drainOnce (claim a batch, deliver in order, record each outcome), sleeps through the sleep you pass when a round found nothing, and ends when the signal is aborted. Run drainOnce from a scheduler instead if you prefer cron to a loop.
• Settings (@krizaka/outbox/settings, needs @krizaka/settings and zod): outbox.batchSize (50, 1–500), outbox.leaseSeconds (60, 10–3600: longer than your slowest delivery), outbox.maxAttempts (8, 1–50), outbox.backoffSeconds (30, 1–3600), outbox.backoffMaxSeconds (3600), outbox.retentionDays (7). purge removes delivered and dead messages older than the retention: call it from the product's job.
• Tables (outboxSchema, Postgres schema outbox): message (id, topic, event_type, payload, created_at, available_at, attempts, leased_until, lease_owner, delivered_at, dead_at, last_error) and processed (consumer, message_id), which consumeOnce uses: the marker and the effect commit in the same transaction, so a crash leaves neither and a redelivery finds the marker.
• outboxStoreContract (from @krizaka/outbox/testkit) is the contract every store runs; memoryOutboxStore() is for unit tests.

ExportsCopy Anchor Link

The 30 names the package exports, read from its types.

@krizaka/outbox/settingsCopy Anchor Link

@krizaka/outbox/testkitCopy Anchor Link

PackageCopy Anchor Link

ChangelogCopy Anchor Link

0.1.3Copy Anchor Link

Patch changes

• The READMEs, comments and published declarations no longer name the products that left the organisation (krizaka/.github#413, rule no-departed-published). The sources were cleaned in #262 with an empty changeset, so the published tarballs still carried the old text; this release replaces them. No behaviour change.
• Updated dependencies [2a7898c]
• Updated dependencies [2272b57]
  • @krizaka/[email protected]
  • @krizaka/[email protected]

0.1.2Copy Anchor Link

Patch changes

• Updated dependencies [0d19b64]
  • @krizaka/[email protected]
  • @krizaka/[email protected]

0.1.1Copy Anchor Link

Patch changes

• Updated dependencies [8210a9a]
  • @krizaka/[email protected]

0.1.0Copy Anchor Link

Minor changes

• New package (#53): the transactional outbox. createOutbox (append(tx, …) in the caller's transaction, claim with a lease over FOR UPDATE SKIP LOCKED, complete, fail with a doubling capped back-off then a dead letter, retryDead, purge, stats), drainOnce and runWorker (no timer of its own: the product passes sleep), consumeOnce (the idempotent consumer, in the effect's transaction), memoryOutboxStore, dbOutboxStore, outboxSchema, @krizaka/outbox/settings (outbox.batchSize, leaseSeconds, maxAttempts, backoffSeconds, backoffMaxSeconds, retentionDays) and outboxStoreContract. At least once: consumers are idempotent on the message id.

Patch changes

• Updated dependencies [da6d980]
  • @krizaka/[email protected]