@hotbunny/events

Events between HotBunny services on RabbitMQ.

• One topic exchange (hotbunny.events) carries every event, routed by its type (account.updated, game.updated).
• Each service has one durable quorum queue (<service>.events) bound to the types it listens to. Events that arrive while the service is down wait there.
• A handler that throws gets the message again. After maxDeliveries (default 5) it goes to the service's dead-letter queue (<service>.events.dead) to be looked at.
• Producers write events to an outbox table in the same transaction as the change; a relay sends them, so a change is never saved without its event.

Delivery is at least once: a handler can see the same event twice (same id) and events can arrive slightly out of order. Write handlers so that running one twice does no harm, for example by re-reading the current state instead of applying a delta.

Events

interface HotbunnyEvent<T> {
    id: string; // uuid, the same on every delivery
    type: string; // area.what-happened, lower case
    source: string; // the publishing service
    occurredAt: string; // ISO 8601
    data: T;
}

Subscribe

import { createEventBus } from "@hotbunny/events";

const bus = createEventBus({ url: process.env.RABBITMQ_URL!, service: "updater" });
bus.subscribe(["game.updated"], async (event) => {
    await syncGame((event.data as { gameId: number }).gameId);
});
await bus.start(); // reconnects by itself; starts even when the broker is down

Patterns follow AMQP topics: account.* is one word, game.# any number. Every subscribe() before start() adds bindings to the service's queue. A binding a service no longer asks for stays on the broker until it is removed there (management UI).

Publish through the outbox

Once per database:

import { outboxTableSql } from "@hotbunny/events";
await db.query(outboxTableSql()); // event_outbox

In the transaction that makes the change:

import { outboxRow } from "@hotbunny/events";
const row = outboxRow("game.updated", { gameId });
await tx.query("INSERT INTO event_outbox (id, type, data, occurred_at) VALUES ($1, $2, $3, $4)", [row.id, row.type, JSON.stringify(row.data), row.occurredAt]);

And the relay, beside the bus:

import { Pool } from "pg";
import { startOutboxRelay } from "@hotbunny/events";

const relay = startOutboxRelay({ bus, pool: new Pool({ connectionString: DATABASE_URL, max: 2 }) });
// after a commit, to send at once instead of within a second:
void relay.flush();

Rows are sent oldest first and marked sent after the broker confirms them. If sending fails the relay stops there and tries again on the next round. Sent rows are deleted after 7 days.

bus.publish(type, data) sends directly, without the outbox, for events that may be lost.

Tests

npm test runs the unit tests. With HOTBUNNY_EVENTS_TEST_URL=amqp://... (a throwaway vhost) it also runs against a real broker.