ZodBus

ZodBus is a typed, in-process event bus powered by Zod, with nested namespaces, typed wildcards, validated payloads, and subscription lifecycles.

setup.mp4

Installation

ZodBus is published on the NPM Registry: https://npmjs.com/package/zodbus

Use Zod 4.1.11 or later within Zod 4. The package supports Node.js 20+ and Bun 1.1+ and provides ESM and CommonJS exports.

npm install -D zod zodbus
pnpm install -D zod zodbus
yarn add -D zod zodbus

Usage

In the examples below, ... stands for a handler.

import { z } from "zod";
import { create } from "zodbus";

// This is how you define your event schema.
//  By traversing the schema, event names are built by concatenating the parent keys with dots,
//  and when a ZodType is encountered it means we have reached the end of the key definition,
//  and the ZodType describes the payload signature for the current event.
const schema = {
  foo: {
    bar: {
      baz: z.object({    // this creates the event "foo.bar.baz",
        id: z.string(),  // with the { id: string; val: number } payload type
        val: z.number()
      })
    }
  },
  zip: {
    za: z.string(),      // this creates the event "zip.za" with a string payload
    zb: z.number(),      // this creates the event "zip.zb" with a number payload
  }
};

const bus = create({ schema, validate: true });

// Non-wildcard subscription signatures are fully-typed.
bus.subscribe("foo.bar.baz", (data: { id: string; val: number; }, event: "foo.bar.baz") => {});
bus.subscribe("zip.za", (data: string, event: "zip.za") => {});

// Wildcard event signatures infer their payload and event types from the events they match.
//  The "*" wildcard is special, as it refers to all events, regardless of the nesting depth.
//  All other wildcard patterns refer to a specific event shape.
bus.subscribe("*", (data: unknown, event: string) => {}); // will be called for all events
bus.subscribe("foo.*.baz", ...); // will be called when publishing a "foo.bar.baz" event
bus.subscribe("foo.bar.*", ...); // same
bus.subscribe("*.bar.*", ...); // same
bus.subscribe("*.*.*", ...); // same
bus.subscribe("*.*", ...); // will be called for "zip.za" and "zip.zb"
bus.subscribeOnce("*.*", ...); // you can also subscribe to an event only once

// You can unsubscribe one or all handlers from one or multiple events.
bus.unsubscribe("zip.za", handler); // unsubscribe handler from "zip.za"
bus.unsubscribe("zip.*", handler); // unsubscribe handler from "zip.za" and "zip.zb"
bus.unsubscribe("zip.*"); // unsubscribe all handlers from "zip.za" and "zip.zb"
bus.unsubscribe("*", handler); // unsubscribe handler from all events
bus.unsubscribe("*"); // unsubscribe all handlers from all events

// Publishing is only allowed on fully-qualified event names.
bus.publish("foo.bar.baz", { id: "uwu", val: 4815162342 });

// Utilities
bus.getEventNames(); // ["foo.bar.baz", "zip.za", "zip.zb"]
bus.getListeners("zip.*"); // gets the listeners for "zip.*"
bus.getListeners(); // gets all listeners
const data = await bus.waitFor("zip.zb"); // typed waitFor with timeout and filter support

Event names join namespace keys with dots. Only schema leaves are publishable: this schema defines foo.bar.baz, zip.za, and zip.zb.

Payloads and validation

Validation is enabled by default. publish accepts the schema's input type, parses once, and delivers its output type to listeners and waitFor. Transforms, defaults, and object-key stripping apply.

const counts = create({
  schema: { count: z.string().transform(Number) },
  onError: (error) => console.error(error),
});

counts.subscribe("count", (value) => console.log(value.toFixed(0)));
counts.publish("count", "42");

All recipients of a publication receive the same parsed value. ZodBus does not freeze or clone it for each listener, so mutations by one listener are visible to later listeners.

With create({ schema, validate: false }), payloads pass through unchanged. Use this only when input and output types are equal and callers already supply valid values. Skipping validation skips transforms, defaults, and key stripping too.

Invalid payloads throw Zod errors synchronously, even with no listeners. Unknown or non-publishable event names throw ValidationError whether validation is enabled or disabled. Validation errors do not go to onError.

Wildcards and types

* alone matches every event at any depth. Within another pattern, each * matches one segment at that exact depth. For the schema above, foo.*.baz, foo.bar.*, *.bar.*, and *.*.* match foo.bar.baz; *.* matches zip.za and zip.zb. A shorter pattern such as foo.* does not match foo.bar.baz.

Wildcard listener arguments are a union of matching [data, event] tuples. Checking event narrows data:

bus.subscribe("zip.*", (data, event) => {
  if (event === "zip.za") {
    console.log(data.toUpperCase());
  } else {
    console.log(data.toFixed(0));
  }
});

Dependent-parameter narrowing requires TypeScript 4.6 or later. A listener annotated (data: unknown, event: string) => void remains assignable.

The root exports InferBusType, InferPublishHandler, InferSubscribeHandler, InferSubscriptionKey, InferPublishKey, and InferMatchedPublishKeys for wrappers:

import type { InferMatchedPublishKeys } from "zodbus";

type ZipEvents = InferMatchedPublishKeys<typeof schema, "zip.*">;

ZipEvents is "zip.za" | "zip.zb". An exact event matches itself, * matches all publish keys, and an undefined subscription key resolves to never.

Subscriptions and delivery order

subscribe(event, listener, options?) returns { event, listener, unsubscribe }. Each call creates an independent subscription. Its unsubscribe() ends only that subscription, so overlapping subscriptions keep working. A listener runs at most once per publication, regardless of how many active subscriptions match it.

subscribeOnce has the same arguments and return shape. Its subscription ends before the listener runs, even if the listener throws or republishes the event.

bus.unsubscribe(pattern, listener?) removes delivery of matching publish paths across subscriptions. Without a listener, it applies to every listener. Other paths remain active. Removed paths are delivered again only after a later subscription covers them.

const handler = (data: unknown, event: string) => console.log(event, data);
bus.subscribe("zip.*", handler);

bus.unsubscribe("zip.za", handler);
bus.unsubscribe("zip.*", handler);
bus.unsubscribe("zip.*");
bus.unsubscribe("*", handler);
bus.unsubscribe("*");

Here handler is the function originally passed to subscribe or subscribeOnce.

Recipients are captured when publish is called. A subscription added afterward does not receive that publication. If all of a recipient's captured matching subscriptions end before its turn, it is skipped. Within a publication, listeners run in their first-registration order for that publish path.

Nested publications are queued in publication order. Each publication finishes invoking its recipients before the next begins. A nested publish still validates synchronously; the outermost publish returns after the queue drains. It returns void and never waits for async listener work. Publishing after an await starts delivery immediately if no other delivery is active.

There is no queue cap or cycle guard. Listeners that unconditionally republish in a cycle can keep publish from returning.

Listener errors and async work

One listener's synchronous throw or returned thenable's rejection does not prevent other listeners from running. onError(error, { event, data, listener }) receives the original error, delivered payload, and listener.

Use a synchronous onError handler. ZodBus ignores its return value: returned promises are not tracked by settled() or dispose(), and their rejections are not captured. If onError starts asynchronous reporting, that work must handle its own failures.

Without onError, each failure is rethrown in its own queueMicrotask. It becomes an uncaught exception in Node.js and Bun, or reaches window.onerror in a browser. A synchronous throw from onError is also rethrown in a microtask without interrupting delivery. Use onError rather than catching listener errors around publish.

A nested publish validation error still reaches its calling listener. If that listener does not catch it, it is reported as a listener failure.

Async listeners run concurrently with later publications. await bus.settled() waits until all tracked listener promises finish, including work started while it waits. It never rejects: listener errors use the reporting path above. A thenable that never settles leaves settled() pending. Work that a listener starts but does not return is not tracked.

Waiting and cancellation

waitFor(event, { timeout?, filter?, signal? }) resolves with the next accepted parsed payload. Wildcard waits resolve with the union of their matching output types.

const controller = new AbortController();
const nextValue = bus.waitFor("zip.zb", {
  timeout: 5000,
  filter: (value) => value > 0,
  signal: controller.signal,
});

bus.publish("zip.zb", 42);
console.log(await nextValue);

The default timeout is 5000 ms; timeout: 0 disables it. A timeout rejects with RuntimeError. A throwing filter rejects that wait, not onError. Aborting rejects with signal.reason, including when the signal is already aborted. Resolution, rejection, cancellation, and disposal remove the wait's subscription, abort listener, and timer.

subscribe and subscribeOnce also accept { signal }. Aborting ends only that subscription. An already-aborted signal returns an already-ended subscription. Ending a subscription another way removes its abort listener. Aborting during delivery skips later recipients whose matching subscriptions have ended.

Disposal and inspection

await bus.dispose() ends every subscription, rejects pending waits with RuntimeError("Bus disposed"), drops queued publications, and waits for tracked listener promises. Repeated calls return the same promise. Disposal does not cancel listener work already running; a never-settling thenable also keeps disposal pending. Avoid awaiting disposal from a listener whose returned promise disposal itself must await.

Disposing inside a listener skips the current publication's remaining recipients and drops nested queued publications. The outer publish returns normally.

After disposal, publish, subscribe, and subscribeOnce throw RuntimeError; waitFor returns a rejected promise. unsubscribe does nothing, and getListeners returns []. settled and getEventNames remain usable.

Before disposal, getListeners(pattern?) returns listeners for the matching paths, or all paths when omitted. Like subscribe, subscribeOnce, unsubscribe, and waitFor, it rejects undefined subscription keys with ValidationError. waitFor reports these failures through its promise. getEventNames() returns the schema's publishable event names.

ValidationError and RuntimeError are exported from zodbus for instanceof checks. Their messages start with [zodbus] .