Event-driven architecture in Node.js: pub/sub with NATS, Kafka and Redis — and when an event is not enough

Request/response is how a microservice system starts, and events are how it stops being a distributed monolith. The moment orders has to call mailer, analytics and inventory after every checkout, orders knows about three services it should not care about, is as slow as the slowest of them, and fails when any of them is down. Flip it around — orders announces "order created", and whoever cares subscribes — and the coupling goes away.

Then two questions show up that every pub/sub design has to answer, and that most tutorials skip: who exactly gets the event when a subscriber runs in three copies, and what happens to it when nobody is listening. This post answers both with running code, on three different message brokers, and ends with the part where a plain event is the wrong tool and you need something with an acknowledgement.

Every snippet was executed; the logs are pasted, not typed. The code is in the example repo (run.sh reproduces all five demos).

If you are new to Moleculer: a service is a plain object with actions (request/response) and events (pub/sub handlers); a broker runs services and connects nodes over a transporter — the message broker in the middle, which can be NATS, Redis, Kafka, MQTT, AMQP or plain TCP. broker.emit("order.created", payload) is the publish side; an events: { "order.created"(ctx) {} } block is the subscribe side. The routing logic — which subscriber instance gets what — is Moleculer's own and is the same on every transporter, which is what makes the "same code, three brokers" section below possible.

The setup: one producer, two kinds of consumers

orders creates orders and announces them. Two subscribers: mailer (which we will run twice, to see load balancing) and analytics (which listens with a wildcard).

// services/orders.service.js
module.exports = {
  name: "orders",
  created() { this.seq = 0; },

  actions: {
    create: {
      params: { sku: "string", qty: { type: "number", default: 1 } },
      async handler(ctx) {
        const order = { id: ++this.seq, sku: ctx.params.sku, qty: ctx.params.qty, total: ctx.params.qty * 10 };
        log(this, `orders.create #${order.id} (${order.sku}) — requestID ${ctx.requestID.slice(0, 8)}, level ${ctx.level}`);
        // event: delivered to ONE instance of every subscriber service that is online right now
        await ctx.emit("order.created", order);
        return order;
      },
    },
    cancel: {
      params: { id: "number" },
      async handler(ctx) {
        // broadcast: delivered to EVERY instance of every subscriber service
        await ctx.broadcast("order.cancelled", { id: ctx.params.id });
        return { id: ctx.params.id, cancelled: true };
      },
    },
  },
};

// services/mailer.service.js
module.exports = {
  name: "mailer",
  events: {
    "order.created": {
      // event payloads can be validated like action params (same fastest-validator schema)
      params: { id: "number", sku: "string", total: "number" },
      async handler(ctx) {
        const { id, sku } = ctx.params;
        log(this, `order.created #${id} (${sku}) — requestID ${ctx.requestID.slice(0, 8)}, meta.user ${ctx.meta.user}, level ${ctx.level}`);
        // events can chain: the context carries the same requestID one level deeper
        await ctx.emit("email.sent", { orderId: id, to: `${ctx.meta.user}@example.com` });
      },
    },
    "order.cancelled"(ctx) {
      log(this, `order.cancelled #${ctx.params.id} — dropping queued emails for it`);
    },
  },
};

// services/analytics.service.js
module.exports = {
  name: "analytics",
  events: {
    "order.*"(ctx) {
      log(this, `${ctx.eventName} #${ctx.params.id} — ${ctx.eventType} from ${ctx.nodeID}, level ${ctx.level}`);
    },
    "email.sent"(ctx) {
      log(this, `email.sent for #${ctx.params.orderId} to ${ctx.params.to} — requestID ${ctx.requestID.slice(0, 8)}, level ${ctx.level}`);
    },
  },
};

(log prefixes the line with a timestamp and the node ID so the logs of four processes can be merged in order — that is all it does.)

The config is two lines that matter. The transporter is an environment variable because we are going to swap it:

// moleculer.config.js
module.exports = {
  nodeID: process.env.NODE_NAME || `node-${process.pid}`,
  transporter: process.env.TRANSPORTER || "nats://localhost:4222",
  // "redis://localhost:6379"  or  "kafka://localhost:9092" — nothing else changes
};

Four nodes, each started with moleculer-runner and a SERVICES= filter: app-1 (orders), mailer-1, mailer-2, analytics-1. A fifth throwaway broker (client.js) calls orders.create four times with meta: { user: "alice" }, then orders.cancel once.

1. emit is balanced, broadcast is not

--- 4× orders.create (each one emits order.created)
[app-1] orders.create #1 (SKU-1) — requestID 3b5d9c00, level 1
[analytics-1] order.created #1 — emit from app-1, level 2
[mailer-2] order.created #1 (SKU-1) — requestID 3b5d9c00, meta.user alice, level 2
[analytics-1] email.sent for #1 to [e-mail] — requestID 3b5d9c00, level 3
[app-1] orders.create #2 (SKU-2) — requestID 128071b0, level 1
[analytics-1] order.created #2 — emit from app-1, level 2
[mailer-1] order.created #2 (SKU-2) — requestID 128071b0, meta.user alice, level 2
[analytics-1] email.sent for #2 to [e-mail] — requestID 128071b0, level 3
[app-1] orders.create #3 (SKU-3) — requestID ab6b4a4f, level 1
[analytics-1] order.created #3 — emit from app-1, level 2
[mailer-2] order.created #3 (SKU-3) — requestID ab6b4a4f, meta.user alice, level 2
[analytics-1] email.sent for #3 to [e-mail] — requestID ab6b4a4f, level 3
[app-1] orders.create #4 (SKU-4) — requestID e1966cc6, level 1
[mailer-1] order.created #4 (SKU-4) — requestID e1966cc6, meta.user alice, level 2
[analytics-1] order.created #4 — emit from app-1, level 2
[analytics-1] email.sent for #4 to [e-mail] — requestID e1966cc6, level 3
--- orders.cancel #2 (broadcasts order.cancelled)
[mailer-2] order.cancelled #2 — dropping queued emails for it
[analytics-1] order.cancelled #2 — broadcast from app-1, level 2
[mailer-1] order.cancelled #2 — dropping queued emails for it

Three things in this log, in order of how often they surprise people.

Each order.created reached one mailer, not two. emit is a balanced event: every subscribing service gets the event once, and if that service runs in several instances, the emitter's registry picks one — mailer-2, mailer-1, mailer-2, mailer-1: round-robin, the same strategy the load balancer uses for action calls. analytics is a different service, so it gets its own copy of every event. In broker vocabulary each service name is an event group, and emit delivers one copy per group. This is what a queue-group or consumer-group gives you in NATS or Kafka, except it works identically on all transporters because the choice is made by the emitting broker, not by the message broker. (One consequence: if the emitting node also hosts an instance of the subscriber, that local instance always wins — registry.preferLocal applies to events too.)

order.cancelled reached both mailers. broadcast goes to every instance of every subscriber. That is the right verb for "flush your cache", "reload config", "a node joined" — anything where each process has to react. For "an order was created, someone send the email", it would send two emails. There is also broadcastLocal, which only reaches services in the same process, and it is what the broker's own $node.connected/$services.changed events use.

The request ID survived two hops and the user survived with it. orders.create ran at level 1 with request 3b5d9c00; the mailer's handler ran at level 2 with the same request ID and the same meta.user that the client put on the action call; the email.sent event that the mailer emitted from inside its handler arrived at analytics at level 3, still 3b5d9c00. Event handlers receive a full Context — ctx.params, ctx.meta, ctx.requestID, ctx.call and ctx.emit — so an event is not a dead end for tracing or for tenant/user information. Enable the tracing middleware and this chain shows up as one trace with three spans, without passing anything by hand.

2. Groups and wildcards

analytics subscribed to order.* and received both order.created and order.cancelled, with the actual name in ctx.eventName. Wildcards are * for one segment and ** for any number — ** is how you write an audit service that sees everything.

The group can also be targeted from the emitting side. This emits order.created to the mailer group only:

await broker.emit("order.created", { id: 99, sku: "SKU-99", total: 0 }, ["mailer"]);

=== 2. Event groups: emit only to the mailer group ===
[mailer-1] order.created #99 (SKU-99) — requestID dccf0532, meta.user undefined, level 1
[analytics-1] email.sent for #99 to [e-mail] — requestID dccf0532, level 2

analytics never saw order.created #99 — it only saw the email.sent that the mailer emitted afterwards, which had no group filter. This is handy for replays ("re-send order 99 to the mailer, but do not count it again") and for the reverse case, where a service wants to join another service's group: an event definition can say group: "mailer", and then that handler and the real mailer share one round-robin — only one of them gets each event.

3. Validation happens at the subscriber

The mailer declared a params schema on its handler. What happens when a producer sends garbage?

await broker.emit("order.created", { id: "oops" });

=== 3. Event payload validation ===
[analytics-1] order.created #oops — emit from client-3557966, level 1
--- and on the mailer node that rejected it:
[19:39:50.367Z] ERROR TRANSIT: ValidationError: Parameters validation error!
    at processCheckResponse (…/node_modules/moleculer/src/validators/base.js:103:6)
  data: [ { type: 'number', message: "The 'id' field must be a number.", field: 'id', actual: 'oops', nodeID: 'client-3557966', event: 'order.created' }, { type: 'required', message: "The 'sku' field is required.", field: 'sku', … }, { type: 'required', message: "The 'total' field is required.", field: 'total', … } ]

Two points. The schema is enforced per handler: analytics had no schema on its wildcard handler, so it happily logged #oops. And the error was logged on the mailer node and went nowhere else — broker.emit had already resolved on the client. That is the contract of an event: emit resolves when the event has been handed to the transporter, not when it has been handled. A handler that throws — validation or otherwise — is logged by the receiving broker and that is the end of it. If the producer needs to know the outcome, it wanted an action call, not an event. Keep that sentence in mind for section 5.

4. Same code, three brokers

Nothing in the services above mentions NATS. So let's run the same bench pair — one node emitting 2 000 small events, one node subscribed, counting arrival latency — on all three transporters, changing only TRANSPORTER=:

=== 4. Same code, three transporters: 2 000 events, one emitter node → one subscriber node ===
--- TRANSPORTER=nats
  awaited each emit: 2000 events in 346 ms (5780 events/s) — received 2000, latency p50 2 ms, p99 8 ms, max 9 ms
  fire-and-forget: 2000 events in 305 ms (6557 events/s) — received 2000, latency p50 29 ms, p99 72 ms, max 72 ms
--- TRANSPORTER=redis
  awaited each emit: 2000 events in 287 ms (6969 events/s) — received 2000, latency p50 2 ms, p99 14 ms, max 16 ms
  fire-and-forget: 2000 events in 249 ms (8032 events/s) — received 2000, latency p50 3 ms, p99 20 ms, max 20 ms
--- TRANSPORTER=kafka
  awaited each emit: 2000 events in 5983 ms (334 events/s) — received 2000, latency p50 4 ms, p99 19 ms, max 96 ms
  fire-and-forget: 2000 events in 2105 ms (950 events/s) — received 2000, latency p50 1186 ms, p99 1936 ms, max 1955 ms

("Awaited each emit" is a loop of await broker.emit(...); "fire-and-forget" fires all 2 000 and then awaits the promises, so its latency numbers include queueing behind the previous 1 999. Single machine, brokers in Docker, one run — treat these as shapes, not benchmarks; the NATS and Redis numbers moved by ±50 % between runs.)

The shapes are the story. NATS and Redis are in the same class: a few thousand events per second from a single emitter loop, single-digit-millisecond delivery. Kafka is 10–20× slower on this path — and this is not Kafka being slow, it is Kafka being used for something it is not built for. Here is what each transporter actually does with an event, which explains both the numbers and the durability question coming next:

• NATS (nats://) — core NATS pub/sub. Each node subscribes to its own subject (MOL.EVENT.<nodeID>); the emitter's registry decides the target node and publishes there. At-most-once, no persistence, and the fastest of the three. NATS also has queue groups, and if you set disableBalancer: true Moleculer stops picking the node itself and lets NATS balance within the group instead — the only one of the three transporters that supports this.
• Redis (redis://) — Redis Pub/Sub (PUBLISH/SUBSCRIBE), one channel per node. Same semantics as NATS: a message published while no subscriber is connected is dropped; there is no queue behind it. Nearly as fast, and often already in the stack.
• Kafka (kafka://) — one topic per node (MOL.EVENT.<nodeID>, auto-created), every message to partition 0, and every Moleculer process is its own consumer group (groupId = the broker's unique instance ID). Kafka's own consumer-group balancing and offset replay are deliberately not in play — Moleculer already balanced the event in its registry and targeted a node, so the topic is just a pipe to that node. Each emit is a producer round trip that waits for the broker's ack, which is where the ~3 ms per event goes.

So if you were about to pick Kafka as the transporter because "we need durability", stop: the transporter is a transport. It gets a packet from node A to node B; the registry, the load balancing and the event routing sit above it and behave the same on all of them. Pick the transporter for operational reasons — what you already run, what your ops team can keep alive, latency — and get durability from the layer that is built for it. Which brings us to the demo that matters most.

5. What happens when the subscriber is down

The scenario every event-driven system meets on its first deploy: orders is up, mailer is being restarted, two orders come in.

=== 5. What happens when the subscriber is down ===
--- events: orders runs, mailer is NOT running; create 2 orders, then start the mailer
  → orders.create returned #1
  → orders.create returned #2
--- mailer log:
(nothing — the two events were delivered to nobody and are gone)

emit resolved happily both times. The registry had no mailer endpoint online, so there was no one to send to, so it sent to no one. The mailer came back three seconds later and got nothing, because nothing was stored anywhere — not in NATS, not in Redis, and (see above) not in a way Kafka would replay either. This is at-most-once delivery, and for a lot of events it is exactly right: a cache-invalidation, a "user is typing" indicator, a metric sample, a UI push. Losing one costs nothing, and the source of truth can be re-read.

An order confirmation email is not in that category. Neither is a payment webhook, an inventory reservation, or anything that ends up in an audit log. For these you need a message that is stored until acknowledged, retried when the handler throws, and parked somewhere visible when it keeps failing. In Moleculer that is a separate module — @moleculer/channels — with adapters for Redis Streams, Kafka, NATS JetStream and AMQP. It is one middleware in the config:

// moleculer.config.js (with CHANNELS=1)
const { Middleware: ChannelsMiddleware } = require("@moleculer/channels");

module.exports = {
  transporter: "nats://localhost:4222",   // still the transporter for calls and events
  middlewares: [
    ChannelsMiddleware({
      adapter: {
        type: "Redis",
        options: {
          maxRetries: 3,                                              // then dead-letter (or drop, if dead-lettering is off)
          deadLettering: { enabled: true, queueName: "FAILED_MESSAGES" },
          redis: {
            url: "redis://localhost:6379",
            consumerOptions: { minIdleTime: 1000, claimInterval: 200, processingAttemptsInterval: 500 },
            // minIdleTime: an unacked message is retried after 1 s (default: 1 hour)
          },
        },
      },
    }),
  ],
};

The producer change is one line — sendToChannel instead of emit:

// services/orders.service.js — create handler
if (process.env.CHANNELS === "1") {
  await this.broker.sendToChannel("order.created", order); // stored in the stream until a consumer acknowledges it
} else {
  await ctx.emit("order.created", order);
}

And the consumer gets a channels block next to events. The handler receives the raw payload — no Moleculer context, because the message may have been written to the stream by anything, not just a broker (set context: true if you want ctx.meta and tracing carried through headers):

// services/mailer.service.js
channels: {
  "order.created": {
    async handler(payload) {
      log(this, `[channel] order.created #${payload.id} (${payload.sku})`);
      if (payload.id === FAIL_ORDER) throw new Error(`SMTP is down, could not mail order #${payload.id}`);
    },
  },
},

Plus a support service that consumes the dead-letter queue, with context: true so the error the adapter attached is readable from ctx.headers:

// services/support.service.js
channels: {
  FAILED_MESSAGES: {
    context: true,
    async handler(ctx) {
      log(this, `[dead-letter] order #${ctx.params.id} — ${ctx.headers["x-error-name"]}: ${ctx.headers["x-error-message"]}`);
    },
  },
},

Same scenario, with a twist at the end: the mailer is started once (so its consumer group exists), then killed; two orders are created while it is down; it comes back; then order #3 is created, which this mailer instance was told to fail (FAIL_ORDER=3).

--- channels (CHANNELS=1, Redis Streams): same scenario
mailer-1 stopped
  → orders.create returned #1
  → orders.create returned #2
messages waiting in the stream: 2
--- order #3 makes the mailer throw: retries, then dead-letter
  → orders.create returned #3
[app-1] orders.create #1 (SKU-1) — requestID bf278ea3, level 1
[app-1] orders.create #2 (SKU-2) — requestID a68bf01b, level 1
[mailer-1] [channel] order.created #1 (SKU-1)
[mailer-1] [channel] order.created #2 (SKU-2)
[app-1] orders.create #3 (SKU-1) — requestID a7031da1, level 1
[mailer-1] [channel] order.created #3 (SKU-1)
[mailer-1] [channel] order.created #3 (SKU-1)
[mailer-1] [channel] order.created #3 (SKU-1)
[support-1] [dead-letter] order #3 — Error: SMTP is down, could not mail order #3
WARN  CHANNELS: Moved message to 'FAILED_MESSAGES' 1788550869933-0

Orders #1 and #2 sat in the Redis stream (XLEN said 2) and were delivered the moment the mailer's consumer reconnected. Order #3 was attempted three times, a second apart — the handler threw, the message was never acknowledged, and after minIdleTime the adapter claimed it back and redelivered it — and after maxRetries it was moved to FAILED_MESSAGES, where the support service picked it up with the original payload and the error message in headers. Multiple mailer instances would share the consumer group, so this also load-balances, like emit did, but with the stream keeping score.

One trap worth knowing before your first outage: Redis Streams consumer groups start reading at $ (new messages only) by default, so a subscriber that has never run does not get messages sent before its first start. Deploy consumers before producers, or set redis.startID: "0" on the channel. Kafka and JetStream adapters have the equivalent knobs (fromBeginning, stream config).

Which one, when

The framework gives you two verbs with different guarantees, and the mistake is treating them as one:

• emit / broadcast (events) — routed by the registry to whoever is online now, one copy per service (or per instance), at-most-once, transporter-agnostic, sub-10 ms. Use them to decouple services when losing an occasional message is acceptable or recoverable: cache invalidation, live updates, metrics, "something changed, go re-read it". This covers more of a typical system than people expect.
• sendToChannel (channels) — stored in Redis Streams / Kafka / JetStream / AMQP until acknowledged, retried, dead-lettered, consumer-group balanced. Use it for anything with a cost of loss: emails, payments, stock movements, outbox patterns, integration with non-Moleculer producers and consumers.
• The transporter is neither of these. It is the wire between nodes for calls and events, and its choice does not change what emit guarantees. NATS if you can choose; Redis if it is already there; Kafka as a transporter only when you have a reason that is not "durability" — for durability use it as a channel adapter instead.

If you take one thing from the logs: in section 1 the mailer got every event, in section 5 it got none, and the code was identical. The difference was whether it was running. Design your event handlers for that — and reach for a channel when you cannot.

---

Code for this post: examples/07-event-driven — run.sh reproduces all five demos against local NATS, Redis and Kafka containers. Moleculer 0.15.2, @moleculer/channels 0.3.1, Node 22.

Previous in the series: Circuit breaker, retry and timeout in Node.js microservices. Next: Moleculer vs gRPC and tRPC — RPC transport vs. a full service layer.