Messaging

Events tell, messages ask

In the events lesson, the task service announced task.completed and did not care who reacted. Sometimes you need the opposite. The tasks part of the Task API wants the notifications part to send a reminder email, and it wants to know whether that worked, so it can tell the user.

You could import the notifications code and call it directly. That is fine while the app is small. But as the app grows into separate modules (you will do that in the modular monolith lesson), every module importing every other module turns into a tangle.

A message is a small named object with data that asks for something: "send a reminder for task 1". You hand it to a message bus. The bus finds the handler registered for that message type, runs it, and gives you back what the handler returned. Sender and handler only share the type name.

Both are in-process: they live in the memory of one running program, like the event bus.

Install @zudojs/messaging

In your task-api folder:

$ npm install @zudojs/messaging

added 1 package, and audited 75 packages in 3s
…

Only one package was added, because the packages it builds on, @zudojs/errors, @zudojs/constants and @zudojs/middleware, are already in your project. Every example on this page runs in the browser terminal, and on your computer with npx tsx src/<file>.ts.

Your first message

import { createMessageBus } from "@zudojs/messaging";
import type { Message } from "@zudojs/messaging";

interface SendReminder {
  readonly taskId: number;
  readonly email: string;
}

const bus = createMessageBus();

bus.on("reminder.send", async (message: Message<SendReminder>) => {
  console.log(`sending reminder for task ${message.payload.taskId} to ${message.payload.email}`);
  return { delivered: true };
}, { id: "send-reminder" });

const result = await bus.send({
  type: "reminder.send",
  payload: { taskId: 1, email: "ada@example.com" },
});

console.log("success:", result.success);
console.log("value:", result.value);

sending reminder for task 1 to ada@example.com
success: true
value: { delivered: true }

• bus.on(type, handler, { id }) registers a handler. Give it an id, so errors can name it and you can remove it later with bus.off(id).
• The type on the parameter, Message<SendReminder>, tells TypeScript the shape of message.payload.
• bus.send({ type, payload }) builds the message and dispatches it. The result says whether it worked (success) and carries the handler's return value (value).

What a message is

Like an event, a message is a frozen object with an id, a type, a payload and a timestamp. createMessage builds one yourself:

import { createMessage, toCorrelationId } from "@zudojs/messaging";

const message = createMessage({
  type: "reminder.send",
  payload: { taskId: 1, email: "ada@example.com" },
  source: "tasks",
  correlationId: toCorrelationId("req-7f3a"),
});

console.log(message.type, message.source);
console.log(message.id.startsWith("msg:"), message.timestamp instanceof Date);
console.log(message.correlationId);
console.log(Object.isFrozen(message));

reminder.send tasks
true true
req-7f3a
true

source says which part of the app sent the message. The correlationId is explained below. Notice toCorrelationId("req-7f3a"): the ids in this package are branded strings. At runtime they are plain strings, but TypeScript will not accept any old string where a correlation id belongs, so you cannot pass a user id there by mistake. You brand a string once, with toCorrelationId, toMessageId or toCausationId, and they reject empty strings.

Results and failures

A failing handler does not make send throw. It returns a result with success: false, and the reason in error. That forces you to look at the outcome instead of forgetting a try:

import { createMessageBus, MessageHandlerError } from "@zudojs/messaging";

const bus = createMessageBus();

bus.on("reminder.send", async () => {
  throw new Error("mail server is down");
}, { id: "send-reminder" });

const failed = await bus.send({ type: "reminder.send", payload: { taskId: 1 } });
console.log("success:", failed.success);
console.log("wrapped:", failed.error instanceof MessageHandlerError);
console.log("message:", failed.error?.message);

const nobody = await bus.send({ type: "reminder.snd", payload: { taskId: 1 } });
console.log("typo:", nobody.success, nobody.value);

success: false
wrapped: true
message: Handler "send-reminder" failed: mail server is down
typo: true []

Look at the last line: a message that no handler listens for, here because of a typo in the type, still counts as a success, with an empty list as its value. The bus cannot tell "nobody needed to answer" from "you misspelled the address".

When a message must have exactly one handler, as a request that expects an answer should, create the bus with allowMultipleHandlers: false, and check bus.hasHandlers(type) at start-up:

import { createMessageBus } from "@zudojs/messaging";

const bus = createMessageBus({ allowMultipleHandlers: false });

bus.on("reminder.send", async () => ({ delivered: true }), { id: "send-reminder" });

try {
  bus.on("reminder.send", async () => ({ delivered: false }), { id: "old-reminder" });
} catch (error) {
  if (error instanceof Error) console.log(error.name, "-", error.message);
}

for (const type of ["reminder.send", "reminder.cancel"]) {
  console.log(type, "has a handler:", bus.hasHandlers(type));
}

MessageError - Message type "reminder.send" already has handler "send-reminder" and the registry does not allow multiple handlers.
reminder.send has a handler: true
reminder.cancel has a handler: false

A second handler for the same type is now a mistake that fails at once, when the app starts, rather than a surprise in production. With several handlers allowed (the default), they run one after another, lowest priority number first, and value is an array of their results. The first one that fails stops the rest.

Message middleware

Message middleware works like the event middleware you just saw: a function around every dispatch, with a next() that runs the rest. Whatever next() returns becomes the result's value. Here one middleware logs every message, and another checks the payload with a schema from the validation lesson before any handler can see it:

import { createMessageBus, MessageValidationError } from "@zudojs/messaging";
import { schema } from "@zudojs/schema";

const ReminderSchema = schema.object({
  taskId: schema.number().int().positive(),
  email: schema.string().email(),
});

const bus = createMessageBus();

bus.use(async (ctx, next) => {
  console.log(`-> ${ctx.message.type}`);
  try {
    return await next();
  } finally {
    console.log(`<- ${ctx.message.type}`);
  }
}, { id: "logging", priority: 10 });

bus.use(async (ctx, next) => {
  if (ctx.message.type === "reminder.send" && !ReminderSchema.safeParse(ctx.message.payload).success) {
    throw new MessageValidationError("Invalid reminder.send payload");
  }
  return next();
}, { id: "validation", priority: 20 });

bus.on("reminder.send", async () => ({ delivered: true }), { id: "send-reminder" });

const ok = await bus.send({ type: "reminder.send", payload: { taskId: 1, email: "ada@example.com" } });
console.log(ok.success, ok.value);
const bad = await bus.send({ type: "reminder.send", payload: { taskId: -1, email: "not-an-email" } });
console.log(bad.success, bad.error?.name, "-", bad.error?.message);

-> reminder.send
<- reminder.send
true { delivered: true }
-> reminder.send
<- reminder.send
false MessageValidationError - Invalid reminder.send payload

• Middleware runs in ascending priority order here: 10 before 20, and the default is 100. That is the opposite of the event bus, so always look it up.
• The logging middleware uses try/finally, so its closing line prints even when an inner layer throws.
• When a middleware throws, the result is success: false and error is exactly what it threw. The handler never ran with bad data.

Validating in the handler itself is also fine. Validating in middleware means no handler can forget. A message payload often came from outside (an HTTP body, a queue), so treat it like any other outside input.

Correlation and causation ids

One click in the app can start a chain: an HTTP request, which sends reminder.send, whose handler sends email.deliver. When something in that chain fails at 3 a.m., you need to find every log line that belongs to that one click. Two ids make that possible:

• The correlation id is shared by everything that belongs to one bigger operation, usually one HTTP request. Every message in the chain carries the same one.
• The causation id is the id of the one message that directly caused this one. It lets you rebuild the chain step by step.

createDerivedMessage(parent, input) builds a follow-up message that keeps the parent's correlation id and records the parent as its cause. Every handler can read both ids from its second argument, the context:

import { createDerivedMessage, createMessageBus, toCorrelationId } from "@zudojs/messaging";

const bus = createMessageBus();
const ids = new Map<string, string>();

bus.on("reminder.send", async (message, context) => {
  console.log(`[${context.correlationId}] reminder.send, caused by: ${ids.get(context.causationId)}`);
  ids.set(message.id, "reminder.send");
  const email = createDerivedMessage(message, { type: "email.deliver", payload: { to: "ada@example.com" } });
  const delivered = await bus.dispatch(email);
  return delivered.value;
}, { id: "send-reminder" });

bus.on("email.deliver", async (message, context) => {
  console.log(`[${context.correlationId}] email.deliver, caused by: ${ids.get(context.causationId)}`);
  return "queued";
}, { id: "deliver-email" });

const result = await bus.send({
  type: "reminder.send",
  payload: { taskId: 1 },
  correlationId: toCorrelationId("req-7f3a"),
});
console.log("result:", result.value);

[req-7f3a] reminder.send, caused by: undefined
[req-7f3a] email.deliver, caused by: reminder.send
result: queued

Both handlers logged the same correlation id, req-7f3a. The first message had no parent: its causation id defaults to its own id, which was not in the map yet, so it printed undefined. The second one names reminder.send as its cause. bus.dispatch(message) sends a message you already built.

You can also pass the id as the second argument of send: bus.send(input, { context: { correlationId } }). The bus copies it onto the message it builds (unless the input already has its own), so createDerivedMessage keeps the chain intact either way. The Task API code at the end of this page uses that form.

In the Task API, use the HTTP request id as the correlation id. The observability lesson puts the same id on every log line and trace, so one search finds the whole story.

Timeouts

The mail server hangs. Without a limit, the request that sent reminder.send waits forever, and so does the user. Give the dispatch a timeout in milliseconds:

import { createMessageBus, MessageTimeoutError } from "@zudojs/messaging";

const wait = (ms: number) => new Promise((resolve) => setTimeout(resolve, ms));
const bus = createMessageBus();

bus.on("reminder.send", async (message, context) => {
  for (let attempt = 1; attempt <= 5; attempt++) {
    if (context.signal.aborted) {
      console.log(`handler: gave up before attempt ${attempt}`);
      return "cancelled";
    }
    console.log(`handler: attempt ${attempt} to reach the mail server`);
    await wait(30);
  }
  return "sent";
}, { id: "send-reminder" });

const result = await bus.send({ type: "reminder.send", payload: { taskId: 1 } }, { timeout: 45 });
console.log("success:", result.success, "timeout:", result.error instanceof MessageTimeoutError);
console.log(result.error?.message);

handler: attempt 1 to reach the mail server
handler: attempt 2 to reach the mail server
success: false timeout: true
Message processing exceeded the timeout of 45ms.
handler: gave up before attempt 3

Read the order of the lines carefully:

• After 45 ms, send returned with success: false and a MessageTimeoutError. The caller was not kept waiting.
• The handler was not stopped by force. JavaScript cannot kill a running function. It kept going until its next check of context.signal.aborted, then stopped itself.

That context.signal is an AbortSignal, the standard object that JavaScript uses to say "stop, nobody needs this any more". It has an aborted flag, and many built-in functions accept it: pass it to fetch(url, { signal }) and the request itself is cancelled. A handler that ignores the signal keeps running and wasting resources after its caller has gone. createMessageBus({ defaultTimeout: 5000 }) sets a timeout for every dispatch.

Cancellation with AbortController

Sometimes the caller decides to stop, for example because the user closed the page. An AbortController is the object that owns a signal: you pass controller.signal to the work, and call controller.abort() to cancel it. The bus links that signal to the handler's context.signal:

import { createMessageBus, MessageDispatchAbortedError } from "@zudojs/messaging";

const wait = (ms: number) => new Promise((resolve) => setTimeout(resolve, ms));
const bus = createMessageBus();

bus.on("report.build", async (message, context) => {
  for (let page = 1; page <= 5; page++) {
    if (context.signal.aborted) {
      console.log(`handler: stopped before page ${page}`);
      return "cancelled";
    }
    await wait(50);
    console.log(`built page ${page}`);
  }
  return "report ready";
}, { id: "build-report" });

const controller = new AbortController();
setTimeout(() => controller.abort(new Error("the user closed the page")), 75);
const result = await bus.send({ type: "report.build", payload: null }, { signal: controller.signal });
console.log("success:", result.success);
console.log(result.error?.message, result.error instanceof MessageDispatchAbortedError);

try {
  await bus.send({ type: "report.build", payload: null }, { signal: controller.signal });
} catch (error) {
  console.log("already aborted:", error instanceof MessageDispatchAbortedError);
}

built page 1
success: false
Message dispatch was aborted. true
already aborted: true
built page 2
handler: stopped before page 3

Each page takes 50 ms. The controller aborted after 75 ms, while page 2 was being built. Follow the lines:

• send returned at once, at 75 ms, with success: false and a MessageDispatchAbortedError. The caller does not wait for the handler. The reason you gave to abort() stays on controller.signal.reason if you want to log it.
• The second send used a signal that was already aborted, so it never started: that is one of the few cases where send throws instead of returning a result.
• The handler, still running in the background, finished page 2, saw the aborted signal and stopped before page 3.

The dispatch is reported as aborted even though the handler returned normally with "cancelled", so the caller can never mistake cancelled work for finished work (since @zudojs/messaging 1.2.0). A handler that ignores the signal does no harm to the result, but it wastes work: it would build all five pages for nobody. A shorter way to stop is context.signal.throwIfAborted(), which throws if the signal was aborted and does nothing otherwise.

Put it together: reminders in the Task API

The notifications part of the app owns the mailer and registers the handler. The tasks part only knows the message type and the payload shape. They could live in different folders, or later in different modules, without importing each other:

import type { Message, MessageBus } from "@zudojs/messaging";

export interface SendReminder {
  readonly taskId: number;
  readonly title: string;
  readonly email: string;
}

export function registerNotificationHandlers(bus: MessageBus): void {
  bus.on("reminder.send", async (message: Message<SendReminder>, context) => {
    context.signal.throwIfAborted();
    const { title, email } = message.payload;
    console.log(`[${context.correlationId}] mail to ${email}: don't forget "${title}"`);
    return { delivered: true, at: "2026-10-01T09:00:00Z" };
  }, { id: "send-reminder" });
}

import { createMessageBus, toCorrelationId } from "@zudojs/messaging";
import { registerNotificationHandlers } from "./notifications.js";

const bus = createMessageBus({ allowMultipleHandlers: false, defaultTimeout: 2000 });
registerNotificationHandlers(bus);

async function remind(requestId: string, taskId: number, title: string, email: string) {
  const result = await bus.send(
    { type: "reminder.send", source: "tasks", payload: { taskId, title, email } },
    { context: { correlationId: toCorrelationId(requestId) } },
  );
  if (!result.success) {
    return { status: 502, body: { error: "Could not send the reminder" } };
  }
  return { status: 202, body: result.value };
}

console.log(await remind("req-7f3a", 1, "Buy milk", "ada@example.com"));
bus.dispose();

[req-7f3a] mail to ada@example.com: don't forget "Buy milk"
{ status: 202, body: { delivered: true, at: '2026-10-01T09:00:00Z' } }

The route answers 202 (Accepted) with the handler's result. If the handler fails or times out, the client gets a 502 with a short, safe message. The real reason goes to your logs, never to the client. bus.dispose() releases the bus when the app shuts down. After that, every send throws.

Notice that the reminder here is sent while the user waits. If the mail server is slow, the request is slow, and if the server crashes, the reminder is lost. In the background jobs lesson you move this work onto a queue, so it survives crashes and retries on its own.

Practice

import { createMessageBus } from "@zudojs/messaging";

const bus = createMessageBus();

bus.use(async (ctx, next) => {
  ctx.state.set("startedAt", Date.now());
  try {
    const value = await next();
    console.log(`${ctx.message.type}: ok`);
    return value;
  } catch (error) {
    console.log(`${ctx.message.type}: failed`);
    throw error;
  }
});

bus.on("task.count", async () => 3, { id: "count" });
bus.on("task.explode", async () => {
  throw new Error("boom");
}, { id: "explode" });

console.log((await bus.send({ type: "task.count", payload: null })).value);
console.log((await bus.send({ type: "task.explode", payload: null })).success);

task.count: ok
3
task.explode: failed
false

import { createMessageBus } from "@zudojs/messaging";

const wait = (ms: number) => new Promise((resolve) => setTimeout(resolve, ms));
const bus = createMessageBus();
let steps = 0;

bus.on("slow.job", async (message, context) => {
  for (let i = 0; i < 10; i++) {
    context.signal.throwIfAborted();
    await wait(30);
    steps += 1;
  }
  return "done";
}, { id: "slow-job" });

const result = await bus.send({ type: "slow.job", payload: null }, { timeout: 45 });
await wait(100);
console.log("success:", result.success, "steps done:", steps);

success: false steps done: 2

Recap

• An event says "this happened". A message asks "please do this" and brings the answer back.
• bus.on(type, handler, { id }) registers, bus.send({ type, payload }) dispatches and returns { success, value, error }. A failing handler does not throw.
• A message nobody handles is still a success. Use allowMultipleHandlers: false and hasHandlers for requests that need exactly one answer.
• Middleware wraps every dispatch. It runs in ascending priority order. Validate payloads there.
• The correlation id ties one operation together, and the causation id points to the direct parent. createDerivedMessage keeps both right.
• timeout and an AbortController cancel through context.signal. The send returns at once with a failure, but handlers must check the signal and stop themselves, because nothing stops them by force.