Build read models from the global event stream and use the transactional outbox for at-least-once delivery to external systems.

Projections & Outbox

Projections build read models from the global event stream. The transactional outbox guarantees at-least-once delivery of events to external systems. Both patterns are built into the event store.

Projections

Overview

A projection reads events sequentially from the global stream (ordered by global_position) and transforms them into a read-optimized data structure (read model). Each projection tracks its own checkpoint and processes events independently.

Use projections for query models, cross-stream views, filtering, search, reporting, dashboards, and external integration state. If the state is only a performance shortcut for rebuilding one stream, use a snapshot instead.

Need	Use	Example
Faster replay for one long stream	Snapshot	`BankAccountBalance` stored as `BankAccountBalanceSnapshot` in `Transaction-{accountId}`.
Queryable read table	Projection	`account_summaries` table for listing accounts with owner, balance, and status.
Derived state from multiple streams	Projection	Customer risk score from transactions, loans, and profile events.
Async batch processing with checkpoints	Projection	Process 500 events per run and resume from the projection checkpoint.

Defining a Projection

Implement the Projection interface. The handle function receives a StoredEvent and a PoolClient — use the provided client for all database writes to ensure atomicity with the projection checkpoint:

import type { Projection, StoredEvent } from "@lox-solutions/alvyn";
import type { PoolClient } from "pg";

const orderSummaryProjection: Projection = {
  projectionName: "order-summary",
  async handle(event: StoredEvent, client: PoolClient) {
    switch (event.type) {
      case "OrderPlaced":
        await client.query(
          `INSERT INTO order_summaries (stream_id, total, status)
           VALUES ($1, $2, 'placed')`,
          [event.streamId, (event.data as { total: number }).total],
        );
        break;
      case "OrderShipped":
        await client.query(
          `UPDATE order_summaries SET status = 'shipped' WHERE stream_id = $1`,
          [event.streamId],
        );
        break;
    }
  },
};

The client parameter is the same PoolClient that holds the transaction for the checkpoint update. If you use this client for your SQL writes, the read model update and the checkpoint advance are atomic — if anything fails, both roll back. This gives exactly-once projection semantics within a single database.

`defineProjection` Builder (Recommended)

For typed projections tied to an event-sourced aggregate, use defineProjection. It mirrors the defineAggregate pattern — same curried function for type inference, automatic stream prefix filtering, entity ID extraction, and typed event handlers:

import { defineProjection } from "@lox-solutions/alvyn";

type OrderEvents = {
  OrderPlaced: { customerId: string; total: number };
  OrderShipped: { trackingNumber: string };
};

const orderProjection = defineProjection<OrderEvents>()({
  projectionName: "order-summary",
  streamPrefix: "Order",

  handlers: {
    OrderPlaced: async (data, ctx) => {
      await ctx.client.query(
        `INSERT INTO order_summaries (id, total, status) VALUES ($1, $2, 'placed')`,
        [ctx.entityId, data.total],
      );
    },
    OrderShipped: async (_data, ctx) => {
      await ctx.client.query(
        `UPDATE order_summaries SET status = 'shipped' WHERE id = $1`,
        [ctx.entityId],
      );
    },
  },
});

await eventStore.runProjection(orderProjection, 500);

The builder automatically:

Filters by stream prefix — events from other aggregates are skipped without calling any handler
Extracts the entity ID from the stream ID (e.g. "Order-123" -> "123")
Provides typed data — each handler receives the correctly-typed event payload
Passes the transactional PoolClient via ctx.client for atomic writes

ProjectionHandlerContext:

interface ProjectionHandlerContext {
  entityId: string; // Entity ID (prefix stripped)
  streamId: string; // Full stream ID
  globalPosition: bigint; // Event's global position
  streamVersion: number; // Event's stream version
  createdAt: Date; // Event's creation timestamp
  client: PoolClient; // Transactional PoolClient
}

Running Projections

Call runProjection() to process the next batch of events:

const processed = await eventStore.runProjection(orderSummaryProjection, 500);

Catch-up loop — process until fully caught up:

let count: number;
do {
  count = await eventStore.runProjection(orderSummaryProjection, 500);
} while (count > 0);

Continuous processing — run on a schedule:

setInterval(async () => {
  try {
    await eventStore.runProjection(orderSummaryProjection, 500);
  } catch (error) {
    console.error("Projection failed:", error);
  }
}, 5000);

How Projections Work Internally

The projection's checkpoint is stored in the projections table (last_position column).
On each runProjection() call:
- Upserts the checkpoint row (idempotent)
- Reads last_position with FOR UPDATE (row-level lock)
- Computes a commit-safe watermark and fetches the next batch bounded by it: SELECT ... WHERE global_position > last_position AND global_position <= safeWatermark ORDER BY global_position ASC LIMIT batchSize. The watermark prevents skipping a lower global_position whose transaction commits after a higher one (a real hazard with concurrent writers across replicas).
- Calls handle(event) for each event sequentially
- Updates last_position to the last processed event's global_position
Everything runs within a single transaction — if the handler fails, the checkpoint is not advanced (crash recovery).

Important Notes

Projections read plain event data only (no decryption). They are designed for building read models from non-PII data.
Each projection name must be unique. Multiple instances with the same name will contend on the same checkpoint row.
The handle function receives StoredEvent (not ReplayedEvent) — there are no tombstones in projection processing.

Subscriptions (Fan-out)

Overview

subscribe() observes the event store directly as a fan-out stream: it returns an AsyncIterable<StoredEvent> that first replays matching history (catch-up) and then tails live events on the same iterator, with low-latency delivery via PostgreSQL LISTEN/NOTIFY (and a polling fallback).

Unlike a single-checkpoint projection or the competing-consumer outbox, every subscriber — e.g. every replica in a replicaset — observes all matching events independently, each maintaining its own in-memory cursor. This makes it the right primitive for GraphQL subscriptions and SSE endpoints.

const ac = new AbortController();

for await (const event of eventStore.subscribe({
  subject: "Order-", // CloudEvents subject == streamId
  recursive: true, // include child subjects (prefix match)
  eventTypes: ["OrderPlaced", "OrderShipped"], // optional type filter
  signal: ac.signal, // stop the stream + release its LISTEN connection
})) {
  await handle(event);
  // Persist event.globalPosition as your cursor for resume-on-restart.
}

Options

interface SubscribeOptions {
  subject?: string; // subject (streamId) to observe, e.g. "/orders" or "Order-"
  recursive?: boolean; // include child subjects (prefix match) vs exact match
  eventTypes?: string[]; // restrict to these CloudEvents types
  lowerBound?: { id: string; type?: "exclusive" | "inclusive" }; // resume cursor
  signal?: AbortSignal; // stop the stream and release resources
  batchSize?: number; // catch-up batch size (default: 500)
  pollIntervalMs?: number; // polling / fallback cadence (default: 1000)
}

Resuming after a restart

The subscription stream is gap-free and in order, so remembering the last processed globalPosition is enough to resume — no broker offset store required. Delivery is at-least-once, so consumers must remain idempotent.

for await (const event of eventStore.subscribe({
  subject: "Order-",
  recursive: true,
  lowerBound: { id: lastProcessedPosition.toString(), type: "exclusive" },
  signal: ac.signal,
})) {
  await handle(event);
  lastProcessedPosition = event.globalPosition;
}

Use-case 1 — GraphQL subscription

Because subscribe() returns an AsyncIterable, it maps directly onto a GraphQL subscription resolver's asyncIterator. Each replica streams to its own connected clients.

const resolvers = {
  Subscription: {
    orderEvents: {
      subscribe: (_parent, args) => {
        const ac = new AbortController();
        return mapAsyncIterator(
          eventStore.subscribe({
            subject: "Order-",
            recursive: true,
            eventTypes: args.types,
            signal: ac.signal,
          }),
          (event) => ({ orderEvents: event }),
        );
      },
    },
  },
};

Use-case 3 — SSE endpoint

The same primitive backs a thin Server-Sent Events handler so other services can subscribe over HTTP, replica-ready.

import type { IncomingMessage, ServerResponse } from "node:http";

async function observeEvents(req: IncomingMessage, res: ServerResponse) {
  res.writeHead(200, {
    "content-type": "text/event-stream",
    "cache-control": "no-cache",
    connection: "keep-alive",
  });

  const ac = new AbortController();
  req.on("close", () => ac.abort());

  const lastEventId = req.headers["last-event-id"] as string | undefined;

  for await (const event of eventStore.subscribe({
    subject: "/orders",
    recursive: true,
    lowerBound: lastEventId
      ? { id: lastEventId, type: "exclusive" }
      : undefined,
    signal: ac.signal,
  })) {
    // The SSE `id:` doubles as the resume cursor via the Last-Event-ID header.
    res.write(`id: ${event.globalPosition}\n`);
    res.write(`data: ${JSON.stringify(event)}\n\n`);
  }
}

Subscribe vs. outbox. Use subscribe() for in-process fan-out (read models, GraphQL, SSE) where every replica needs every event. Use the transactional outbox when you must publish each event once across the fleet to an external broker (e.g. NATS) — see below.

Transactional Outbox

Overview

The transactional outbox is the opt-in competing-consumer adapter for bridging events to an external broker (e.g. NATS, Kafka, RabbitMQ). It solves the dual-write problem: how to atomically update the event store and notify external systems. Outbox rows are inserted in the same transaction as the events, guaranteeing consistency.

Unlike subscribe() fan-out, the outbox relay distributes work: exactly one replica processes each entry (via FOR UPDATE SKIP LOCKED), so the event is published once to the broker rather than once per replica. Use it for use-case 2 (bridge every event to NATS); use subscribe() for fan-out.

How It Works

1. Begin Transaction
   |
   +-> INSERT INTO events ...
   +-> INSERT INTO outbox ...     (same transaction)
   |
2. Commit Transaction
   |
3. Relay Worker (separate process)
   |
   +-> SELECT ... FOR UPDATE SKIP LOCKED  (poll pending entries)
   +-> Dispatch to message broker
   +-> UPDATE outbox SET processed_at = now()

Publishing to the Outbox

Specify outboxTopics when appending events:

await eventStore.append({
  streamId: "Order-123",
  expectedVersion: 0,
  events: [{ type: "OrderPlaced", data: { total: 99.99 } }],
  outboxTopics: ["orders", "notifications"],
});

Or through the aggregate builder:

await Order.append(eventStore, {
  entityId: "order-123",
  expectedVersion: order.version,
  events: [{ type: "OrderPlaced", data: { total: 99.99 } }],
  outboxTopics: ["orders"],
});

Outbox Payload Format (CloudEvents v1.0.2)

Each outbox entry contains a fully CloudEvents v1.0.2 compliant JSON payload:

{
  "specversion": "1.0",
  "type": "OrderPlaced",
  "source": "urn:my-app:event-store",
  "id": "Order-123/1",
  "time": "2026-03-12T10:00:00.000Z",
  "datacontenttype": "application/json",
  "subject": "Order-123",
  "correlationid": "cmd-abc",
  "actorid": "user-789",
  "schemaversion": 1,
  "data": { "customerId": "cust-1", "total": 99.99 }
}

Polling and Dispatching

Build a relay worker that polls and dispatches:

async function relayOutbox() {
  await eventStore.processOutbox(async (entries) => {
    for (const entry of entries) {
      await messageBroker.publish(entry.topic, entry.payload);
    }
  }, 100);
}

setInterval(relayOutbox, 2000);

Replica Safety

The outbox uses SELECT ... FOR UPDATE SKIP LOCKED for polling. This means:

Multiple relay workers can run concurrently across replicas
Each worker gets a disjoint set of entries — no double-processing
If a worker crashes mid-batch, the entries are automatically unlocked when the transaction rolls back
Delivery guarantee: at-least-once (entries may be redelivered if the worker crashes after dispatching but before marking as processed)

Projections & Outbox

On this page