Events and the event bus

An event is the third message kind, and the one that fans out: many handlers may answer one event, where a command and a query each have exactly one.

Declaring

import * as Schema from "effect/Schema";
import { Event } from "@effect-server-utils/cqrs";

const OrderPlaced = Event.make("OrderPlaced", {
  orderId: Schema.String,
  customerId: Schema.String,
});

const event = OrderPlaced.make({ orderId: "ord_1", customerId: "cus_1" });

An event holds its schema rather than being one. A definition that was itself a schema could be annotated, made optional, or piped — and each of those returns a new schema carrying neither the tag nor the brand, so it would quietly stop being an event with nothing to say so. make and Type are forwarded because constructing an event and naming the value it carries are what nearly every call site does; .schema is there for the rare one that genuinely wants a schema.

The value is plain data, not a class instance, so its in-memory shape and its serialized shape are the same. That is what lets one definition describe an event dispatched in-process today and one read back off a durable log later.

Event.Base is the minimum a bus routes on — { readonly _tag: string }. It is deliberately that small because a consumer’s domain layer references it, and anything more would be a dependency it did not ask for.

The design: subscriptions choose, not dispatch

const publish = Effect.gen(function* () {
  const eventBus = yield* EventBus;
  yield* eventBus.dispatch([OrderPlaced.make({ orderId, customerId })]);
});

That is the entire producer-side API. A producer knows what happened; it does not know who is listening, and in a layered application it is often forbidden to. Letting it pick the consistency model would mean deciding, on behalf of consumers it cannot name, whether their failure may undo its own write.

So dispatch says only “these events happened”, and each subscriber declares what it needs.

subscribe — same boundary

Runs in the publisher’s fiber, in registration order, inside whatever boundary the publisher is in. The handler’s writes commit with the publisher’s, and its failure rolls the publisher back.

const subscriptions = Effect.gen(function* () {
  const eventBus = yield* EventBus;
  yield* eventBus.subscribe(OrganizationCreated, (event) =>
    wallets.createFor(event.organizationId),
  );
});

This is the contract for a reaction that is part of the same logical operation — a wallet must not exist without its organization.

subscribeAfterCommit — after the fact, isolated

Runs once the publisher’s boundary has completed, each handler in a boundary of its own, its failure logged and isolated.

const subscriptions = Effect.gen(function* () {
  const eventBus = yield* EventBus;
  yield* eventBus.subscribeAfterCommit(OrderPlaced, (event) => email.sendReceipt(event.orderId));
});

The failure direction is the opposite of subscribe, by design: the producer is already durable, so a reaction must not undo it. Handlers here are expected to be idempotent and independently retryable. Failures go to UnhandledFailures.

stream — never awaited

Subscribes to the given tags and hands back a stream. Only events broadcast while subscribed arrive: nothing is replayed, and nothing accumulates for a stream no one is reading.

const consume = Effect.gen(function* () {
  const eventBus = yield* EventBus;
  const events = yield* eventBus.stream(["OrderPlaced", "PaymentSettled"]);
  // ...
});

Subscribing is an effect rather than a property of the stream because when it happens is observable — a consumer that subscribed lazily, on first pull, would miss everything broadcast between its construction and that pull. The subscription lives as long as the ambient scope.

This is a third surface rather than a flag on subscribeAfterCommit because the delivery contracts genuinely differ: an after-commit handler is awaited and its failure isolated, while a stream consumer may be a saga that runs for days and must never hold up a drain.

At a glance

	subscribe	subscribeAfterCommit	stream
Runs in	publisher’s fiber	a boundary of its own	its own fiber
Timing	with the publisher	once the boundary ends	once the boundary ends
Failure	rolls the publisher back	logged and isolated	reported, saga keeps running
Awaited by the publisher	yes	yes	no

What “after commit” waits for — DeferralSink

This package does not know what a transaction is. dispatch hands its deferred surfaces — the broadcast, and the subscribeAfterCommit handlers — to a DeferralSink if one is in context, and runs them itself at the end of the dispatch if none is.

interface DeferralSinkShape {
  readonly defer: (events: ReadonlyArray<Event.Base>) => Effect.Effect<void>;
}

That makes the boundary a wiring decision rather than a requirement:

• No sink. subscribeAfterCommit handlers run at the end of the dispatch that produced them — still after every immediate handler, still isolated from the publisher. Enough for a host with no datastore to coordinate, and the bus works with no extra package installed.
• With a sink. They run once the boundary the sink owns has succeeded, each in a boundary of its own. Install @effect-server-utils/unit-of-work and that means after commit, in a fresh transaction, with rollbacks discarding what they buffered.

A handler is written the same way for both, which is what lets a host adopt a unit of work later without revisiting one of them. What changes is only when it runs and what it can undo.

A sink also gets to reject a dispatch it cannot place — the unit-of-work sink dies with EventDispatchedOutsideUnitOfWork when nothing is open, because once you have a boundary, a dispatch that forgot one is a bug rather than a different delivery. That check happens before any immediate handler runs, so a missing boundary is reported while the dispatch is still whole.

EventBus.drain(events, boundary?) is the other half of the seam: a sink calls it when its boundary completes, and the bus keeps what is its own — which handlers, in what order, spans, isolation — taking only “run each one in here” from the caller.

Building the bus

import { makeEventBus, Event } from "@effect-server-utils/cqrs";

const OrderSpanAttributes = Event.spanAttributes({
  OrderPlaced: (event: typeof OrderPlaced.Type) => ({ "order.id": event.orderId }),
});

const EventBusLive = makeEventBus({
  spanAttributes: { ...OrderSpanAttributes, ...BillingSpanAttributes },
});

Each dispatched event opens an event.<tag> span carrying event.tag, event.handler.count, and whatever the tag’s registered extractor returned. Modules declare their own contributions with Event.spanAttributes (a type-checked identity function, so a contribution is validated where it is written) and they are merged where the bus is built — the same way dispatch surfaces are.

Subscriptions are registered while layers are built, so the registries are only mutated during composition and are read-only by the time anything dispatches.

The broadcast buffer feeding stream is unbounded, so a broadcast never blocks a flush. A bounded buffer would either block the publisher or drop silently; a consumer that stops consuming instead shows up as its own subscription queue growing, which is a diagnosable bug rather than lost events.

Type reference

Export	What it is
Event.make(tag, fields)	declares an event
Event.Base	{ readonly _tag: string } — what a bus routes on
Event.Type<E>	the value an event carries
Event.spanAttributes(map)	type-checked per-tag attribute extractors
Event.is	runtime predicate, for reflection
EventBus / EventBusShape	the bus service
makeEventBus(options?)	builds it
EventHandler	(event: Event.Base) => Effect<void>
DeferralSink / DeferralSinkShape	where deferred surfaces go, when something owns a boundary
ReactionBoundary	(reaction: Effect<void>) => Effect<unknown, unknown>