Outbound sinks

The endpoint side of event publishing. Every ctx.publish of a .public() event ends in runtime.sinkDrain(event, payload, envelope) after the local fold (ctx.emit never gets here — it's local-only). An outbound endpoint adapter installs a terminal OutboundStage at boot that drains the chain into a real transport — BullMQ, NATS, a webhook, OTLP, anything.

Outbound delivery is symmetric with inbound. Inbound adapters translate transport events into runtime.execute(...). Outbound adapters drain runtime.sinkDrain into the transport. The two halves let you replace the wire without touching the domain.

The .public() gate

Events are private by default. Calling .public() on an event declaration is what makes it reach the outbound chain:

import { defineEvent } from "@nwire/messages";
import { z } from "zod";

// Public — fans out locally AND reaches outbound sinks.
export const OrderPlaced = defineEvent({
  name: "shop.order-placed",
  schema: z.object({ orderId: z.string(), amount: z.number() }),
}).public();

// Private — fans out locally to subscribers only. Never crosses the
// process membrane. Use for module-internal events like audit trails.
export const OrderAudited = defineEvent({
  name: "shop.order-audited",
  schema: z.object({ orderId: z.string() }),
});

That gate is on the event itself, not the adapter. A domain event without .public() stays inside the bounded context — local subscribers see it, but no outbound stage ever does. That's how terminal-private fields (cleaner codes, raw telemetry, internal IDs) stay out of the integration contract.

The verb matters too: a handler announces a cross-boundary fact with ctx.publish(Event, payload) — local fold first, then the drain. An action's ctx.publish refuses events without the marker, and ctx.emit skips the drain even for a .public() event. Forge handler ctxs carry publish out of the box; a plain app installs publishPlugin() from @nwire/app (there, an unmarked event stays local instead of throwing). See events and listeners.

The stage shape

OutboundStage is the contract an outbound adapter implements:

import type { OutboundStage } from "@nwire/runtime";

const stage: OutboundStage = {
  name: "bullmq.terminal",
  position: "terminal",
  kind: "bullmq",
  async run({ event, payload, envelope }) {
    await queue.add(event.name, { payload, envelope });
  },
};

• position — "early" | "middle" | "terminal". The runtime runs them in position order. Early/middle stages are for logging, metrics, routing; terminal stages do the transport delivery.
• kind — adapter-kind tag. Only one terminal per kind is allowed. Installing a second NATS terminal throws. Swap implementations by swapping kinds.
• run({event, payload, envelope}) — called once per drain. Returning { continue: false } short-circuits the chain.

Installing a stage via an endpoint adapter

Outbound adapters set direction: "outbound" and call ctx.installSinkStage(stage) at boot:

import type { Adapter } from "@nwire/endpoint";
import type { OutboundStage } from "@nwire/runtime";

export function captureOutboundAdapter(captured: unknown[]): Adapter {
  return {
    $kind: "adapter",
    kind: "capture",
    direction: "outbound",
    async boot(ctx) {
      if (!ctx.installSinkStage) return; // no app mounted
      const stage: OutboundStage = {
        name: "capture.terminal",
        position: "terminal",
        kind: "capture",
        run: ({ event, payload, envelope }) => {
          captured.push({ name: event.name, payload, envelope });
        },
      };
      ctx.installSinkStage(stage);
    },
    async shutdown() {
      // ...drain transport
    },
  };
}

Then compose with the endpoint:

await endpoint("orders", { exitOnShutdown: false })
  .use(captureOutboundAdapter(captured))
  .mount(app)
  .run();

installSinkStage is only provided when an App is mounted. Standalone endpoints (no app) get undefined and the adapter should fall back gracefully.

See the outbound-sink example for the full pattern. The real BullMQ adapter lives in @nwire/bullmq (bullmqOutboundAdapter) — same contract.

Envelope causality survives

The envelope (messageId, correlationId, causationId) flows unchanged through the sink, so downstream systems can stitch the causal chain. A consumer that receives the BullMQ job sees the same correlationId the original HTTP request set, and Studio's Trace page renders the cross-process tree.

See also

• Event — .public() gate
• Local delivery chain — the priority model for the local fold
• Topology — how Studio renders sinks per app