Event Bus

Once domain events are persisted via the Outbox pattern, a message relay needs to read them from the database and then use an event bus to deliver events to the rest of your system.

ontologic ships a typed event bus with a publisher and a listener, each backed by a pluggable connector. The connector is the only piece you swap out between environments: the publisher/listener logic stays the same whether you're running tests locally or processing millions of messages in production.

---

Core concepts

Publisher

The publisher publish events to a message broker. It accepts any event that matches your domain's event union type, validates the metadata, and hands off the serialized message to its connector.

Listener

The listener subscribes to incoming messages from the broker and dispatches them to your handlers. You register a handler per event name — or a wildcard handler for everything — before starting the listener. Every delivery ends in exactly one of four ways, decided by the listener rather than by the connector:

Outcome	Message	Reported through
Your handler resolves	ack()	—
No handler registered for the event	ack()	onUnhandled(name)
Envelope has no event.name	ack()	onError(error)
Content is not JSON, or fails the validator	nack()	onError(error)
Your handler throws	nack()	onError(error)

Connector

The connector is the adapter between the event bus and the underlying broker. It is the only thing that knows about Kafka, SQS, RabbitMQ, or whatever you're running. The publisher and listener each depend on a connector interface — swap the connector, keep everything else.

Ordered or not

Whether events are delivered in order is your choice, per consumer. Sometimes order matters (a projection that applies an entity's events one after the other), sometimes it doesn't (sending an email per event).

• The publisher takes a plain publisher connector (kind: "publisher") or an ordered one (kind: "ordered-publisher"). An ordered connector requires an orderingKey on every publish, usually the entity id, which is what the message relay passes.
• The listener takes a plain listener connector (kind: "listener") or an ordered one (kind: "ordered-listener"), which hands a key's events over one after the other.

---

Defining your event union

Start by defining the union of all domain events your service produces or consumes. The event bus is generic over this union, which gives you type-safe handlers.

import { DomainEvent } from "ontologic";

class OrderPlaced extends DomainEvent<
  "ORDER_PLACED",
  1,
  { orderId: string; total: number }
> {
  constructor(entityId: string, payload: { orderId: string; total: number }) {
    super({ name: "ORDER_PLACED", version: 1, entityId, payload });
  }
}

class PaymentReceived extends DomainEvent<
  "PAYMENT_RECEIVED",
  1,
  { amount: number; currency: string }
> {
  constructor(entityId: string, payload: { amount: number; currency: string }) {
    super({ name: "PAYMENT_RECEIVED", version: 1, entityId, payload });
  }
}

type OrderEvents = OrderPlaced | PaymentReceived;

---

Publishing events

import { DomainEventBusPublisher } from "ontologic";

const publisher = new DomainEventBusPublisher<OrderEvents>({
  publisherConnector: myConnector,
});

await publisher.start();

await publisher.publish(
  new OrderPlaced("order-123", { orderId: "order-123", total: 99 }),
  {
    id: "evt-abc",
    createdAt: new Date().toISOString(),
  },
);

await publisher.stop();

The publish method validates the metadata before forwarding. An invalid id or createdAt throws immediately — the event is never sent.

With an ordered publisher connector, pass the key whose events must stay in order, usually the entity id. Publishing to an ordered connector without one throws:

await publisher.publish(event, metadata, { orderingKey: event.entityId });

Event metadata

Every published event carries metadata alongside the event payload:

Field	Type	Required	Description
id	string	Yes	Unique identifier for this event occurrence
createdAt	string	Yes	ISO 8601 datetime with timezone (e.g. 2024-01-15T10:30:00.000Z)
offset	number	No	Position in the event stream, if tracked

---

Listening to events

import { DomainEventBusListener } from "ontologic";

const listener = new DomainEventBusListener<OrderEvents>({
  listenerConnector: myConnector,
  options: { validator: parseOrderEvents },
});

listener.listenTo("ORDER_PLACED", async (event, metadata) => {
  // event is typed as OrderPlaced
  console.log(event.payload.orderId, metadata.id);
});

listener.listenTo("PAYMENT_RECEIVED", async (event, metadata) => {
  // event is typed as PaymentReceived
  console.log(event.payload.amount, event.payload.currency);
});

await listener.start();

You must register at least one handler before calling start() — starting without handlers throws an error.

Wildcard handler

To handle all incoming events with a single callback, use "*" as the event name:

listener.listenTo("*", async (event, metadata) => {
  console.log("received", event.name, metadata.id);
});

The wildcard handler is used as a fallback: if an incoming event has no specific handler registered, it falls through to "*". If there is no specific handler and no wildcard, the message is acknowledged and the name is reported through onUnhandled.

Acknowledged, not nack'd: listenTo is per event name precisely so a consumer can take only what concerns it, and a nack asks the broker to deliver the message again. Nothing about the second attempt would differ — the handler is still not registered — so a nack here buys a redelivery loop, or a dead-letter queue filling with events that were never this consumer's business.

If you want to notice a handler you forgot to wire, register onUnhandled:

listener.onUnhandled((eventName) => {
  logger.debug(`no handler for ${eventName}`);
});

It is deliberately not onError. An unsubscribed event is a routine outcome, and a consumer with selective subscriptions would otherwise see a constant stream of non-errors on its error channel.

---

Connectors

The connector is the only environment-specific piece of the event bus. Everything above — generic types, metadata validation, handler dispatch, ack/nack semantics — is implemented once in the publisher and listener, independent of the broker.

Implementing a connector

To integrate a broker, implement two interfaces. They live in ontologic's shared bus connector contract, which the command bus uses too:

Publisher connector:

interface IPublisherConnector {
  readonly kind: "publisher";
  status: "STARTED" | "STOPPED";
  start(): Promise<void>;
  stop(): Promise<void>;
  publish(message: string, options?: { orderingKey?: string }): Promise<void>;
  onError(handler: (error: Error) => void): void;
  checkConnection(): Promise<boolean>;
}

Listener connector:

interface IListenerConnector {
  readonly kind: "listener";
  status: "STARTED" | "STOPPED";
  start(): Promise<void>;
  stop(): Promise<void>;
  onMessage(handler: (message: ReceivedMessage) => Promise<void>): void;
  topicName(): string;
  onError(handler: (error: Error) => void): void;
  checkConnection(): Promise<boolean>;
}

Received message:

interface ReceivedMessage {
  content: string; // JSON-serialized { event, metadata } envelope
  ack(): Promise<void>;
  nack(): Promise<void>;
}

That's the full contract. The connector never sees the event name: the listener reads it from event.name inside the envelope to route to the right handler. An envelope without an event name is acked and reported through onError, because redelivering it cannot make it routable.

The kind field is a discriminator. Several connector interfaces are otherwise structurally identical, and without it TypeScript would accept a listener where an ordered listener was expected.

The listener calls ack() after your handler resolves successfully, and nack() if it throws. How ack/nack are implemented is entirely up to the connector — requeue, dead-letter, or discard are all valid choices depending on your broker and your requirements.

A failing event blocks its group

An event whose handler throws is nacked. On SqsOrderedListener, nothing behind it with the same ordering key then runs until it succeeds. That is deliberate: skipping it would corrupt everything downstream. Set nackVisibilitySeconds as the retry delay, never set maxDeliveries on an event listener, and use no redrive policy (or a very high maxReceiveCount) on event queues.

Upgrading a connector written for 1.9

1.10 changed the connector contract:

• publish(name, message, options) is now publish(message, options).
• ReceivedMessage.name is gone; routing reads event.name from the envelope.
• Every connector declares a kind and implements checkConnection(). Listeners also implement topicName().
• onError handlers receive an Error.
• The interfaces and in-memory connectors moved from event-bus/connectors to the shared bus connectors. They are still exported from ontologic.

Supported brokers

The connector interfaces are intentionally thin. Any broker that can route named messages and support acknowledgments fits the model:

Broker	Notes
AWS SQS	@ontologics/sqs-connectors: SqsPublisher/SqsListener on a standard queue, or SqsOrderedPublisher/SqsOrderedListener on a FIFO queue for per-entity order
Apache Kafka	Topic per event name, or a single topic with the event name in the key
Google Pub/Sub	One topic per event, subscription per listener
RabbitMQ	Topic exchanges with the event name as the routing key
Redis Streams	@ontologics/redis-connectors: RedisOrderedPublisher/RedisOrderedListener (a stream and a consumer group). Per-key order holds only with a single consumer

---

In-memory connectors

ontologic ships in-memory connectors out of the box:

import { InMemoryConnectors } from "ontologic";

const connectors = new InMemoryConnectors();

const publisher = new DomainEventBusPublisher<OrderEvents>({
  publisherConnector: connectors.publisher,
});
const listener = new DomainEventBusListener<OrderEvents>({
  listenerConnector: connectors.listener,
  options: { validator: parseOrderEvents },
});

The connectors share an internal EventEmitter. Publishing an event immediately delivers it to every registered listener in the same process: in-memory listeners broadcast, there are no competing consumers. connectors.orderedPublisher and connectors.orderedListener are there too, to test against the ordered connector kinds.

These connectors are intended for tests and local prototyping only. They have no persistence, no retry logic, no dead-letter queue, and no cross-process delivery. If the process restarts, unprocessed messages are gone. Ack and nack are no-ops. Do not use them in production.

Using in-memory connectors in tests

import {
  InMemoryConnectors,
  DomainEventBusPublisher,
  DomainEventBusListener,
} from "ontologic";

test("listener receives published events", async () => {
  const connectors = new InMemoryConnectors();

  const publisher = new DomainEventBusPublisher<OrderEvents>({
    publisherConnector: connectors.publisher,
  });
  const listener = new DomainEventBusListener<OrderEvents>({
    listenerConnector: connectors.listener,
    options: { validator: parseOrderEvents },
  });

  const received: OrderPlaced[] = [];

  listener.listenTo("ORDER_PLACED", async (event) => {
    received.push(event as OrderPlaced);
  });

  await listener.start();
  await publisher.start();

  await publisher.publish(
    new OrderPlaced("order-1", { orderId: "order-1", total: 50 }),
    { id: "evt-1", createdAt: new Date().toISOString() },
  );

  expect(received).toHaveLength(1);
  expect(received[0].payload.total).toBe(50);

  await publisher.stop();
  await listener.stop();
});

Each test builds its own new InMemoryConnectors(), so there is no shared state between tests.

---

Validating event shape

A validator is required on the listener.

Events travel over the wire as JSON. When the listener receives a message and deserializes it, the result is a plain JavaScript object — not an instance of your DomainEvent class. Without a validator, your handler receives a structurally correct object but one that is missing the class identity: instanceof checks fail, prototype methods are absent, and any class-level invariants you rely on are silently bypassed.

The validator is the step that reconstructs a real event instance from the raw wire payload:

import { DomainEvent } from "ontologic";

// with a schema library like Zod
const OrderPlacedSchema = z.object({
  entityId: z.string().min(1),
  name: z.literal("ORDER_PLACED"),
  version: z.literal(1),
  payload: z.object({ orderId: z.string(), total: z.number() }),
});

function parseOrderPlaced(raw: unknown): OrderPlaced {
  const data = OrderPlacedSchema.parse(raw);
  return new OrderPlaced(data.entityId, data.payload);
}

Pass the validator when constructing the listener:

const listener = new DomainEventBusListener<OrderEvents>({
  listenerConnector: myConnector,
  options: {
    validator: parseOrderPlaced,
  },
});

The validator runs after deserialization and before your handler is called. If it throws, the message is nack'd, the error is reported through onError, and your handler is never invoked.

---

Error handling

Both the publisher and listener surface connector-level errors through an onError callback. Wire it up before calling start():

publisher.onError((error) => {
  logger.error("event bus publisher error", { error });
});

listener.onError((error) => {
  logger.error("event bus listener error", { error });
});

The handler always receives an Error: anything else that gets thrown is wrapped in one first.

Handler-level errors (thrown inside a listenTo callback) result in an automatic nack() and are reported through onError. Events with no matching handler do not reach onError — they are acknowledged and reported through onUnhandled instead, because an event you did not subscribe to is not a failure.

---

Lifecycle

Both publisher and listener follow the same lifecycle:

1. Configure — pass connectors and options to the constructor
2. Register handlers (listener only) — call listenTo before start
3. Start — call start() to initialize the connector and begin processing
4. Stop — call stop() to stop the connector. Whether messages already received finish first is the connector's behaviour: @ontologics/sqs-connectors listeners wait for them, the in-memory ones don't

await publisher.start();
// ... publish events
await publisher.stop();

await listener.start();
// ... process events
await listener.stop();

Starting a publisher or listener whose connector is already started, or stopping one that is already stopped, has no effect.

Both also expose status() and checkConnection(), which delegates to the connector and throws if its check fails. These are the same lifecycle methods the command bus emitter and receiver expose.