Inbox

Transactional dedupe-on-receive for inbound message handlers. Companion to b.outbox: where outbox guarantees at-least-once delivery, inbox lets the receiver guarantee exactly-once handling by recording every (source, messageId) pair in the same database transaction as the business state change. A duplicate redelivery (network retry, replay, broker re-dispatch on consumer failure) collides with the primary-key constraint and the second handler short-circuits cleanly.

Schema (declared via declareSchema(externalDb)): message_id TEXT, source TEXT, received_at TIMESTAMP, processed_at TIMESTAMP NULL, metadata_json JSONB|TEXT, with PRIMARY KEY (source, message_id). Postgres uses ON CONFLICT … DO NOTHING RETURNING to decide fresh-vs-duplicate in one round-trip; SQLite 3.35+ uses INSERT OR IGNORE … RETURNING 1 to avoid the changes() race when callers issue intervening statements on the same transaction handle.

Defenses on the input side: messageId and source are bounded in length (default 256 chars each) and rejected for NUL / C0 / DEL control characters before they reach the primary key — Postgres TEXT may truncate at NUL, opening a dedupe-collision attack where "abc\\0attacker" and "abc" collide. metadata is JSON- serialized through safeJson and capped at maxPayloadBytes (default 64 KiB).

Two APIs: high-level handle(opts, handler) opens a transaction, records receive, runs the handler exactly once when fresh, marks processed, commits — recommended for most callers. Low-level recordReceive(opts, txn) lets operators manage the transaction directly when they need fine-grained control over what runs in the dedupe envelope.

b.inbox.create(opts) #

{
  externalDb:      Object,   // b.externalDb instance (transaction()-shaped)
  table:           string,   // SQL identifier; required
  retentionDays:   number,   // sweep horizon (default 30); unprocessed rows kept 2x as long
  audit:           boolean,  // emit inbox.* audit events (default true)
  maxPayloadBytes: number,   // metadata serialized cap (default 64 KiB)
  messageIdMaxLen: number,   // chars (default 256)
  sourceMaxLen:    number,   // chars (default 256)
}

Build an inbox dedupe-store. Returns { declareSchema, recordReceive, markProcessed, handle, sweep, isFresh, getStats, table, retentionDays }. Operators call declareSchema once at boot, handle per inbound message, and sweep periodically (under their own scheduler) to age out processed rows past retention.

var inbox = b.inbox.create({
  externalDb:    externalDbInstance,
  table:         "inbox_events",
  retentionDays: 30,
});

await inbox.declareSchema(externalDbInstance);

var outcome = await inbox.handle({
  messageId: "evt-9f3c4d",
  source:    "kafka:orders.created.v1",
}, async function (xdb) {
  await xdb.query("INSERT INTO orders (id) VALUES ($1)", ["o-42"]);
  return { orderId: "o-42" };
});
outcome.fresh;             // → true on first delivery, false on replay
outcome.result.orderId;    // → "o-42"

// Replay short-circuits:
var replay = await inbox.handle({
  messageId: "evt-9f3c4d", source: "kafka:orders.created.v1",
}, async function () { return { orderId: "should-not-run" }; });
replay.fresh;              // → false
replay.result;             // → null

var stats = await inbox.getStats({ source: "kafka:orders.created.v1" });
stats.total;               // → 1
stats.processed;           // → 1

var deleted = await inbox.sweep();   // age out beyond retention

Last updated 2026-08-08T16:39:49.652Z by seeder.