Reliability & governance helpers

Beyond produce/consume, @babelqueue/core ships optional, zero-dependency helpers that mechanise the reliability and governance contracts the wire spec defines. None touch the frozen envelope (schema_version: 1); each is a tooling layer over the codec. They are the Node face of the cross-SDK spec — follow the linked spec page for the full contract.

Idempotency

Wrap(store, handler) makes a handler run at most once per meta.id, even on at-least-once redelivery. The Store methods may be sync or async, so a Redis/DB backend drops straight in.

import { Wrap, InMemoryStore } from "@babelqueue/core";

const store = new InMemoryStore(); // tests / single process
const handler = Wrap(store, async (data, meta) => { await chargeOnce(data); });

The in-memory InMemoryStore is the reference; for a fleet, implement the Store interface (seen / remember / forget) over a shared backend. See Idempotency and the store deep-dive.

Transactional outbox

Outbox / OutboxRelay remove the producer dual write: persist the encoded envelope in the same DB transaction as your business row, then a relay publishes the durable rows verbatim.

import { Outbox, OutboxRelay, InMemoryOutboxStore } from "@babelqueue/core";

const store = new InMemoryOutboxStore(); // production: a DB-backed OutboxStore
const outbox = new Outbox(store);

// inside YOUR DB transaction, beside the business write — no commit of its own:
const id = await outbox.write(envelope);  // encodes via EnvelopeCodec, calls store.save()

// later, a relay drains the durable rows to the broker:
const res = await new OutboxRelay(transport, store).drain(); // res.published / res.failed

OutboxStore is the four-method async contract (save / fetchUnpublished / markPublished / markFailed); the relay forwards through an OutboxTransport. The transaction boundary is yours. See Transactional Outbox.

DLQ redrive & replay-bypass

resetForRedrive(envelope) resets a dead-lettered envelope for re-publish (dead_letter removed, attempts to 0, everything else preserved); the replay guard skips effects that already fired on a deliberate replay. The delivered headers are passed explicitly (the same seam the OTel wrapHandler uses):

import { isReplay, bypassExternalEffects } from "@babelqueue/core";

async function handler(data, meta, headers) {
  saveOrder(data);                                            // idempotent core
  await bypassExternalEffects(headers, () => sendEmail(data)); // skipped on replay
}

RedriveIO.publishWithHeaders stamps the bq-replay-bypass marker on a header-carrying transport. See DLQ redrive & replay-bypass.

GDPR field encryption

protect / unprotect encrypt only the data leaves a schema marks x-gdpr-sensitive, in place. Cipher is caller-bound and AesGcmCipher (on the built-in node:crypto) is the reference. Validate cleartext — protect after validation on produce, unprotect before validation on consume.

import { protect, unprotect, AesGcmCipher } from "@babelqueue/core";

const cipher = new AesGcmCipher(key); // or implement Cipher over a KMS

protect(data, schema, cipher);        // producer: encrypt marked leaves
try {
  unprotect(data, schema, cipher);    // consumer: inverse
} catch (e) {
  // DecryptError → wrong key / tampered → retry / dead-letter
}

See GDPR field encryption.

OpenTelemetry (traceparent)

The @babelqueue/core/otel subpath emits publish <urn> / process <urn> spans and, on a header-carrying transport, injects/extracts the W3C traceparent so a consumer span is a true child of the producer span — degrading to v0.1 trace_id correlation otherwise. Every babelqueue-node-adapters transport now carries the header on its native metadata channel.

import { wrapHandler, publish } from "@babelqueue/core/otel";

@opentelemetry/api is the only optional dependency. See Observability.

Per-URN schema validation

validateSchema (producer guard) / wrap (consumer safety net) validate a message’s data against the JSON Schema registered for its URN; MapProvider embeds schemas in code. See Per-URN schema validation.