@easyweb/events

Shared RabbitMQ event layer for Easyweb microservices — a resilient connection manager, typed publisher/subscriber base classes, and the exchange/routing-key registry that keeps event names consistent across services.

Installation

amqplib and amqp-connection-manager are peer dependencies — the consuming service owns them, so only one copy exists in the tree:

npm install @easyweb/events amqplib amqp-connection-manager

Two copies of amqp-connection-manager would mean two TCP connections per service, two heartbeat timers, and two uncoordinated reconnect loops. Keep them at the top level.

Connection

createRabbitMQConnection owns one connection per process. Call it once, in a module the rest of the service imports:

// src/lib/rabbitmq.ts
import { createRabbitMQConnection } from "@easyweb/events";
import config from "../config";

const { host, port, user, pass, vhost } = config.rabbitmq;

const rabbitmq = createRabbitMQConnection({
  urls: [{ protocol: "amqp", hostname: host, port, username: user, password: pass, vhost }],
  connectionName: "auth-service",
});

export const connectRabbitMQ = rabbitmq.connectRabbitMQ;
export const getChannel = rabbitmq.getChannel;
export const isRabbitMQConnected = rabbitmq.isConnected;
export const onRabbitMQReady = rabbitmq.onReady;
export const closeRabbitMQ = rabbitmq.close;

Options

| Option | Default | Notes | |---|---|---| | urls | — | Passed straight to amqp-connection-manager. Strings or Options.Connect objects. | | connectionName | none | Surfaces in the RabbitMQ management UI under Connections. Set it — it is how you tell services apart. | | heartbeatIntervalInSeconds | 120 | | | onChannelSetup | none | Re-runs on every (re)connect. Declare exchanges, queues, and bindings here so topology survives a channel recreation. |

Handle contract

| Method | Behaviour | |---|---| | connectRabbitMQ() | Resolves once, on first successful connect. Never rejects — but never resolves while the broker is down either. Do not await it ahead of server.listen(), or a broker outage holds your service off its port. Fire it with void after listening. | | getChannel() | Throws "RabbitMQ not initialized" if called before the first connect. | | isConnected() | Current state. Useful for health endpoints. | | onReady(cb) | Fires on first connect and again after every reconnect. Fires immediately if already connected. Use this to rebind subscribers. | | close() | Closes channel then connection. |

Reconnects use exponential backoff with full jitter, capped at 30s, injected via the findServers hook. The jitter is deliberate: it stops every service reconnecting in lockstep after a broker restart.

Publishing

Define the event shape once, then a Publisher per event:

import { Publisher, Exchanges, RoutingKeys, UserCreatedEvent } from "@easyweb/events";
import { getChannel } from "../lib/rabbitmq";

class UserCreatedPublisher extends Publisher<UserCreatedEvent> {
  exchange = Exchanges.AuthService;
  routingKey = RoutingKeys.USER_CREATED;
}

await new UserCreatedPublisher(getChannel()).publish(
  { userId, name, email },
  requestId, // optional — travels as the x-request-id header
);

publish asserts the exchange (durable, topic by default), sends persistent messages, and throws if the channel buffer is full.

Subscribing

import { Subscriber, Exchanges, RoutingKeys, UserCreatedEvent } from "@easyweb/events";
import { onRabbitMQReady } from "../lib/rabbitmq";

class UserCreatedSubscriber extends Subscriber<UserCreatedEvent> {
  exchange = Exchanges.AuthService;
  routingKey = RoutingKeys.USER_CREATED;

  async onMessage(data: UserCreatedEvent["data"], _msg, requestId: string) {
    await createProfile(data.userId, requestId);
  }
}

// Re-subscribe on every reconnect, not just at boot.
onRabbitMQReady((channel) => {
  void new UserCreatedSubscriber(channel).subscribe();
});

subscribe() gives you, per subscriber:

• a durable queue named <exchange>.<routingKey>, overridable via protected queueName
• a dead-letter exchange <exchange>.dlx and queue <queueName>.dead
• a delay queue <queueName>.retry, which nothing consumes
• prefetch(1)
• retry on failure by parking the message in <queueName>.retry with an incremented x-retry-count and an expiration of EVENT_RETRY_DELAY_MS (default 2000). It expires, dead-letters back into <queueName> alone, and is tried again — up to 3 attempts, then the message is acked and dropped to stop poison messages looping.

A retry goes back to this queue, never to the exchange. Five services bind user.account.provisioned; republishing to the exchange re-ran the handler in the four that had already succeeded. See docs/platform/ADR-0005.

"Event handler failed" is logged at warn while retries remain and at error on the delivery that discards the message. The message string is fixed — a Loki recording rule keys on it.

requestId is read from the x-request-id header, falling back to a fresh UUID so every message is traceable.

Event registry

Exchanges and routing keys live in this package so no service invents its own string.

enum Exchanges {
  AuthService = "auth",
  UserService = "user",
  SubscriptionService = "subscription",
  BillingService = "billing",
  PaymentService = "payment",
}

enum RoutingKeys {
  USER_CREATED = "auth.user.created",
  USER_ACCOUNT_PROVISIONED = "user.account.provisioned",
  USER_PROFILE_UPDATED = "user.profile.updated",
  SUBSCRIPTION_PLAN_PUBLISHED = "subscription.plan.published",
  BILLING_INVOICE_ISSUED = "billing.invoice.issued",
  PAYMENT_CHARGE_SUCCEEDED = "payment.charge.succeeded",
}

Routing keys follow <service>.<entity>.<action>. Adding an event means adding the routing key, the event interface under src/events/types/, and publishing a new version of this package.

One exchange per owning service, and only that service publishes on it. RoutingKeys is a single flat enum shared by all of them, so member names carry the owner as a prefix — USER_CREATED is auth's auth.user.created, while user-service's keys are USER_PROFILE_*, USER_ADDRESS_* and so on.

PaymentService has no publisher yet; payment-service is not built. Its keys are pinned here so billing can name them at compile time and so the strings cannot drift when it does land.

Money on the wire

Money is a decimal string of minor units, because the columns behind it are BigInt and JSON.stringify throws on one. Parse to BigInt, not Number. The minor unit depends on the currency, which is always carried in a field beside the amount:

| Currency | Minor unit | "199000" means | |---|---|---| | IDR | Rp 1 (zero-decimal) | Rp 199.000 | | USD | one cent | $1,990.00 |

IDR is zero-decimal because that is what Xendit and Stripe both settle in. Carrying sen would mean rounding every invoice total to a rupiah before it could be charged, and the remainder would have to be stored somewhere and reconciled.

Changes in 2.8.0

subscription.subscription.activated gained required priceAmount, currency and billingInterval.

Additive for every consumer, and a minor rather than a major for that reason — but it is required at the publish site, which is safe only because subscription.activated has exactly one publisher in the platform: subscription-service's trial-conversion sweep.

It was a bug that they were ever absent. Billing's subscription mirror merges partial payloads, so an activated naming no price left the row holding what subscription.created wrote when the account was provisioned on free — a zero price on a LIFETIME interval. Both of billing's cycle-planner queries exclude precisely that shape, so a converted trial was never planned, never invoiced and never charged, and the customer heard nothing about it because notification stays silent on a converted trial in the expectation that the invoice speaks for it.

subscription.trial-started deliberately does not carry them, and must not be made to. Billing's planner reads no trial column; the stale zero price is the only thing keeping it from invoicing a trial that is still running.

Breaking changes in 2.0.0

• IDR amounts are zero-decimal. They were sen (exponent 2) in 1.x, where "9900000" meant Rp 99.000.
• subscription.subscription.{renewed,past-due,paused,resumed} moved to the billing exchange as billing.subscription.*. Billing owns the clock, so only billing knows that a period was paid for and rolled over, or that dunning ran out. The four payloads lost entitlementsVersion — billing cannot know it, and subscription-service computes it when it applies the event — and gained invoiceId, which is what makes each transition traceable to the document that caused it.
• The nine remaining subscription.subscription.* payloads gained version, the subscription row's version, so mirrors can guard them like every other replicated row.

Row versioning

Every payload for a row that crosses a service boundary carries a version. It starts at 1 and advances by exactly 1 per published change, so a mirror can tell a stale redelivery from the expected next one. The full rule, and the VersionGuardOutcome type consumers apply it with, live in src/events/types/version-contract.ts.

This matters because subscribe() retries by parking the message in a delay queue and letting it expire back into the tail of its own queue (see above) — out-of-order delivery is normal operation, not an anomaly.

Publishers must not advance version for writes that touch only unreplicated columns. A login-timestamp bump that increments the version puts a permanent gap in the sequence, and every consumer stalls behind it.

A row's first version is announced by whichever event creates it — user.account.provisioned for the profile, preference and notification rows, user.address.created for an address — so a mirror starts at version 1 and the first edit applies cleanly as 1 → 2. Consumers should still tolerate a snapshot arriving for a row they have never seen: publishing is best-effort, and a creation event lost to a broker outage would otherwise strand the row forever.

Development

npm install
npm run build       # cleans dist, then tsc
npm pack --dry-run  # verify tarball contents before publishing
npm publish