@betterinternship/broker

The transport layer. Two standard RabbitMQ shapes, and deliberately nothing else — no DAGs, no orchestration state, no workflow engine.

| Plane | Shape | Use it when | | ----------- | ------------------------- | --------------------------------------------------- | | Command | work queue | exactly one handler; losing the message is a bug | | Event | topic exchange (events) | zero-to-many consumers, each with its own queue |

The litmus test: a second consumer someday would be a feature → event. Two consumers reacting would be a bug → command.

Local development

docker compose -f docker-compose.rabbit.yml up -d
# RABBITMQ_URL=amqp://bi:bi@localhost:5672/bi-dev

Dev and prod are separated by vhost (bi-dev / bi-prod), never by name prefixes. Every queue, exchange and routing key is spelled identically in both.

Publishing

import { broker, EmailSend } from '@betterinternship/broker';

await broker.send(EmailSend, {
  to: signatory.email,
  subject: '✅ Your form is ready',
  html: renderedHtml,
  alias: 'sign',
});

send and emit await a publisher confirm, so they reject on a nack, on a timeout, and — for commands, which publish mandatory — when no queue exists to take the message. A publish never silently disappears.

There is no RABBITMQ_ENABLED flag and no inline-execution fallback: the broker is tier-1 infrastructure like Postgres. Servers still boot and serve HTTP with it down; publishes just fail fast with a clear error.

Consuming

import { broker, EmailSend, FilloutCompleted, isEvent } from '@betterinternship/broker';

// Command: one handler, prefetch 5.
await broker.handle(EmailSend, async (msg, { attempt }) => {
  await deliver(msg.payload, { idempotencyKey: msg.payload.messageId ?? msg.id });
}, { prefetch: 5 });

// Event: this consumer's own queue, bound to the keys it cares about.
await broker.subscribe('delivery.discord', [FilloutCompleted], async (msg) => {
  if (isEvent(msg, FilloutCompleted)) await postPrefillEmbed(msg.payload);
}, { prefetch: 1 });

Resolve to ack. Throw to nack — the broker counts deliveries and parks the message in <queue>.dlq once x-delivery-limit (3) is reached. Backoff for a transient blip belongs inside the handler:

await retryTransient(() => renderPdf(input), { label: 'pdf render' });

Handlers must be idempotent: a crash, a deploy or a DLQ shovel all replay the same message.

Topology

Declaration is consumer-owned and idempotent — a consumer asserts its queue, its DLQ and its bindings on every boot. Publishers assert exchanges only, so there is no whoever-connects-first topology. Every queue is a quorum queue with x-delivery-limit: 3 and x-dead-letter-exchange: dlx (routing key = queue name).

Shutdown

process.on('SIGTERM', () => void broker.shutdown());

Stops consuming, lets in-flight handlers finish and ack, then closes. Wire it up — a deploy that skips this abandons unacked work to redelivery.

Contracts

Contracts live with their owner. A contract whose producer and consumer are the same repo (docs.fillout) is declared in that repo, so adding a domain job needs no package publish. Only cross-repo contracts live here:

• EmailSend — email.send, consumed by the delivery worker
• FilloutCompleted — docs.fillout.completed

Publishing this package

Shared package: the owner creates and publishes it. Do not bump version here.