Subscriptions

EventStore.subscribe() is Alvyn's fan-out primitive for consumers that observe events independently. It returns an AsyncIterable<SubscriptionEvent> that first catches up on matching history and then seamlessly transitions to tailing new events as they commit in PostgreSQL.

Unlike the transactional outbox, subscriptions are not competing consumers. Every subscriber (e.g. every replica or connected client) receives every matching event and maintains its own in-memory cursor.

Basic Usage

const controller = new AbortController();

for await (const event of eventStore.subscribe({
  subject: "Order-",
  recursive: true,
  eventTypes: ["OrderPlaced", "OrderShipped"],
  signal: controller.signal,
})) {
  await handleOrderEvent(event);
  // Persist event.globalPosition after successful processing
}

// Stops iteration and releases the PostgreSQL LISTEN connection
controller.abort();

The iterator is ordered by globalPosition, delivers at least once, and is independent for each subscriber. Consumers should be idempotent and persist their cursor after handling an event.

Catch-up and Live Tailing

The subscription reads historical events first, then transitions to live tailing on the same iterator:

1. Catch-up phase: Queries historical events matching your filter ordered by global_position ASC in batches of batchSize.
2. Live tailing phase: Alvyn automatically registers a PostgreSQL LISTEN client on commit notifications (NOTIFY). When new events append, the listener wakes up immediately.
3. Polling fallback: A background timer (default pollIntervalMs: 1000) guarantees delivery even if a database notification is dropped or during network reconnection.

Commit-Safe Watermark

The read side uses a commit-safe watermark. This prevents a concurrent transaction that commits later with a lower globalPosition from being skipped by a live cursor consumer:

Transaction A (position 101) ----------> commits at t=2
Transaction B (position 100) ---------------> commits at t=3 (delayed)

Without a watermark, a subscriber tailing at t=2 would read 101, advance its cursor past 101, and permanently miss 100 when B commits at t=3. Alvyn calculates the safe watermark dynamically so consumers never miss out-of-order commits.

Filters and Options

interface SubscribeOptions {
  subject?: string; // CloudEvents subject (streamId), e.g. "Order-123" or "Order-"
  recursive?: boolean; // When true, matches prefix (e.g. "Order-" matches "Order-1", "Order-2")
  eventTypes?: string[]; // Restrict delivery to specific event types
  lowerBound?: { id: string; type?: "exclusive" | "inclusive" }; // Resume cursor
  signal?: AbortSignal; // Abort signal to stop stream and release DB connection
  batchSize?: number; // Batch size for catch-up queries (default: 500)
  pollIntervalMs?: number; // Polling fallback cadence in ms (default: 1000)
}

subject and eventTypes can be combined. Without a subject filter, the subscription observes matching events from all streams in the store.

Resuming with a Cursor

To resume processing after a worker restart or server disconnect, pass lowerBound.id set to the string representation of the last processed globalPosition:

let lastProcessedPosition = "42";

for await (const event of eventStore.subscribe({
  subject: "Order-",
  recursive: true,
  lowerBound: {
    id: lastProcessedPosition,
    type: "exclusive", // Default: start with the event immediately AFTER 42
  },
})) {
  await handle(event);
  lastProcessedPosition = event.globalPosition.toString();
}

Use type: "inclusive" if the event at the cursor should be delivered again (e.g. if the previous consumer crashed mid-processing before saving its state).

Aggregate Subscriptions

An aggregate handle provides a strongly typed subscription for a specific entity stream. Stream subject derivation and snapshot event filtering happen automatically:

import { defineAggregate } from "alvyn";

type OrderState = { status: "placed" | "shipped" };
type OrderEvents = {
  OrderPlaced: { total: number };
  OrderShipped: { trackingNumber: string };
};

const Order = defineAggregate<OrderState, OrderEvents>()({
  streamPrefix: "Order",
  evolve: {
    OrderPlaced: () => ({ status: "placed" }),
    OrderShipped: () => ({ status: "shipped" }),
  },
});

// Strongly typed: event.type and event.data match OrderEvents
for await (const event of Order.subscribe({
  eventStore,
  entityId: "order-123",
  options: { eventTypes: ["OrderShipped"] },
})) {
  if (event.type === "OrderShipped") {
    console.log(event.type, event.data.trackingNumber);
  }
}

Real-time Integration Use Cases

GraphQL Subscriptions

Because subscribe() returns a standard AsyncIterable, it integrates directly with GraphQL subscription resolvers (such as GraphQL Yoga, Apollo Server, or Mercurius):

const resolvers = {
  Subscription: {
    orderEvents: {
      subscribe: (_parent, args, context) => {
        const ac = new AbortController();

        // Stream matching events to this specific connected client
        return mapAsyncIterator(
          context.eventStore.subscribe({
            subject: "Order-",
            recursive: true,
            eventTypes: args.types,
            signal: ac.signal,
          }),
          (event) => ({ orderEvents: event }),
        );
      },
    },
  },
};

Server-Sent Events (SSE) Endpoint

You can expose a real-time HTTP event stream where the standard Last-Event-ID header acts as the resumable cursor:

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

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

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

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

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

This minimal endpoint serializes globalPosition as a decimal string because JSON cannot encode bigint. Before deployment, add authorization, cursor validation, error cleanup, and bounded backpressure handling. For implementation sketches and their operational limits, see the Event Streaming & Resilient Consumer Playbook.

Encryption and Schema Evolution

Subscriptions upcast complete, unencrypted event payloads by default, including during catch-up. Set raw: true to receive the stored schema instead. Events with encrypted fields are always delivered as RedactedSubscriptionEvent in their original schema, regardless of whether the key is active or revoked. Neither mode decrypts GDPR-encrypted fields.

• The payload starts with the stored data column, with inaccessible encrypted fields represented as null (whether the key is active or revoked). Ciphertext lives separately in encrypted_data and is not returned by subscriptions. These redacted events carry redacted: true, redactedPaths, and schemaVersion. They are never upcasted: a registered upcaster may require encrypted fields. Check for "redacted" in event before treating an aggregate subscription event as a fully typed payload.
• raw: true skips upcasting complete events and returns the stored schema format; Order.subscribe({ options: { raw: true }, ... }) does not promise typed current-schema payloads. Redacted events remain redacted in either mode.
• Aggregate event-map types do not validate runtime payloads or restore encrypted fields. eventTypes filters also do not narrow the TypeScript union; check event.type before accessing event-specific fields.

When decrypted and upcasted domain state is required, use Order.load() or eventStore.load().

Subscription vs. Outbox

Use subscribe() when you need real-time streaming to all replicas or clients. Use the Transactional Outbox when publishing fleet-wide messages to external brokers.