MicroservicesSupported

@zmdb/app/messaging ships the transport-neutral message layer and typed request and event clients. Core NATS ships from @zmdb/transport/nats, RabbitMQ ships from @zmdb/transport/rabbitmq, Redis Pub/Sub ships from @zmdb/transport/redis, and typed gRPC ships from @zmdb/transport/grpc. Applications own those transports through the same module graph and bounded lifecycle as HTTP.

None of those four adapters or their peers is installed by yarn add @zmdb/core@1.0.0-beta.2. The exact optional edges are:

The public seam#

Import the broker-neutral API from @zmdb/app/messaging:

import { EventPattern, MessagePattern, createMessageClient, transportExtension, type MessageContext, type TransportStrategy } from '@zmdb/app/messaging';

A broker delivery is not an HTTP request. MessageContext<T> is therefore a sibling of Ctx, not a subtype: it has no invented method or path. The reusable part is structural:

type WithHeaders = {
  readonly headers: Readonly<Record<string, string>>;
};

function requiresApiKey(ctx: WithHeaders): boolean {
  return ctx.headers['x-api-key'] === env.API_KEY;
}

Both HTTP and message contexts satisfy WithHeaders without casts. An HTTP guard and a message handler can call the same function, while the HTTP middleware interface remains HTTP-only so a path check cannot silently run against a broker delivery.

Declaring consumers#

Each declaration carries its consume-boundary validator:

type OrderId = { readonly id: number };

function orderId(raw: unknown): OrderId {
  if (typeof raw !== 'object' || raw === null || !('id' in raw) || typeof raw.id !== 'number') {
    throw new Error('id must be a number');
  }
  return { id: raw.id };
}

class OrdersConsumer {
  @MessagePattern('order.get', orderId)
  async get(ctx: MessageContext<OrderId>): Promise<Order> {
    return orders.findById(ctx.payload.id);
  }

  @EventPattern('order.refresh', orderId)
  async refresh(ctx: MessageContext<OrderId>): Promise<void> {
    await orders.refresh(ctx.payload.id);
  }
}

@EventPattern rejects a method that returns a value. Decorators write Stage-3 metadata; createApp resolves the exact-pattern map once at startup. There is no filesystem scan, runtime wildcard language or metadata lookup per delivery.

Failure and settlement#

Handlers never call ack() or nack(). The dispatcher returns a DispatchOutcome: a broker Settlement plus an optional correlated reply. The strategy applies the settlement while it still owns the broker delivery token.

The decorators are separate because success and failure mean different things:

A request is not redelivered after its caller has received an error: doing so could perform the same command twice. An event has no waiting caller, so a transient failure can be retried.

• Unknown patterns are acknowledged and reported to onUnhandled.
• Parse and validation failures settle dead and reach onInvalidPayload.
• Private handler error detail goes to onHandlerError; it never enters a correlated reply.
• retry always carries a positive delay. The default is exponential from one second, capped at 30 seconds.
• A strategy that cannot redeliver or dead-letter requires an onUndeliverable sink when it is attached to an application.

The three capability flags are facts, not hints: redelivery, deadLetter and requestResponse. A request through a strategy without request/reply support rejects immediately with TransportUnsupportedError.

Typed request clients#

The strategy moves unknown; the client supplies the trusted result type by validating the reply:

type OrderCalls = {
  readonly 'order.get': {
    readonly request: OrderId;
    readonly response: Order;
  };
};

const client = createMessageClient<OrderCalls>(transport, {
  timeoutMs: 5_000,
  validate: {
    'order.get': raw => assert<Order>(raw),
  },
});

const order = await client['order.get']({ id: 7 });

The client generates the correlation id, passes it and an AbortSignal to the strategy, rejects mismatched replies, validates successful payloads and clears its deadline timer on every exit. Timeouts have no framework default.

For one-way events, createEventPublisher<EventMap>(transport) exposes one typed method per event pattern and delegates to transport.emit.

Application ownership#

Attach strategies through the public application extension rather than starting them beside the app:

await using app = createApp(AppModule, {
  extensions: [
    transportExtension({
      transports: [transport],
      dispatcher: {
        onUnhandled: message => audit.unhandled(message),
        onInvalidPayload: (message, error) => audit.invalid(message, error),
        onHandlerError: (message, error) => audit.failed(message, error),
        onUndeliverable: (message, settlement) => audit.dropped(message, settlement),
      },
    }),
  ],
  graceMs: 5_000,
});

await app.init();

Initialization runs module hooks, builds the dispatcher, then calls transport.listen in declaration order. A strategy enters the close ledger before listen, so a partial startup closes the refusing strategy and every earlier strategy in reverse order. Disposal closes transports in reverse order before provider/controller shutdown hooks, so no message handler outlives its dependencies.

Packaged broker strategies#

Install only the client used by the selected strategy:

yarn add @zmdb/transport@1.0.0-beta.2 redis@^6.2.1
yarn add @zmdb/transport@1.0.0-beta.2 @nats-io/transport-node@^3.4.0
yarn add @zmdb/transport@1.0.0-beta.2 amqplib@^2.0.1

One package publishes every strategy and declares each broker client an optional peer, so the install differs only in the peer. Import each adapter from its own entry point:

import { createNatsStrategy } from '@zmdb/transport/nats';
import { createRabbitMqStrategy } from '@zmdb/transport/rabbitmq';
import { createRedisStrategy } from '@zmdb/transport/redis';

Redis and core NATS therefore require dispatcher.onUndeliverable. RabbitMQ requires an explicit positive prefetch, owns its dead-letter destination and uses publisher-confirmed, per-message-TTL retry copies. It never immediately requeues a failed delivery.

See Broker Transports for concrete configuration and the complete settlement table.

What remains deferred#

Kafka is deferred because committing an ordered partition offset also commits every predecessor, which does not implement independent per-message settlement. MQTT is deferred because broker QoS cannot honour retry.afterMs. There is no bespoke length-prefixed TCP protocol.

GraphQL remains out of scope; the message layer has no GraphQL dependency.

Before splitting a deployment, keep the trade-off explicit: an independent service buys independent scaling and failure, while giving up local transactions, joins and atomic reads across the seam.

Cross-links#

• Broker Transports · gRPC · Custom Transports
• Hybrid Applications · Transactional Outbox