Build read-optimized query models from the global event stream with atomic checkpoints and exactly-once processing semantics.

Projections

Projections build read models from the global event stream. They transform raw, immutable business events into query-optimized PostgreSQL tables, search indexes, or materialized views.

Each projection tracks its own independent checkpoint position in PostgreSQL and processes events in global commit order.

Projections vs. Snapshots

Feature	Snapshot	Projection
Purpose	Fast stream replay shortcut	Queryable read tables, search, analytics
Scope	Single aggregate stream (`stream_id`)	Global event stream across multiple aggregates
Storage	Generated snapshot event in the stream	Custom database tables (e.g. `order_summaries`)
Execution	Synchronous or on aggregate append	Asynchronous batch processing with checkpoint
Example	Cached `Cart` state for `Cart-123`	Cross-customer sales report table

For typed projections tied to event-sourced aggregates, use defineProjection. It mirrors the defineAggregate pattern with type inference, automatic stream prefix filtering, entity ID extraction, and typed handlers:

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

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

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

  handlers: {
    OrderPlaced: async (data, ctx) => {
      // ctx.client is part of the checkpoint transaction (atomic!)
      await ctx.client.query(
        `INSERT INTO order_summaries (id, customer_id, total, status)
         VALUES ($1, $2, $3, 'placed')
         ON CONFLICT (id) DO UPDATE SET total = $3, status = 'placed'`,
        [ctx.entityId, data.customerId, data.total],
      );
    },
    OrderShipped: async (_data, ctx) => {
      await ctx.client.query(
        `UPDATE order_summaries SET status = 'shipped' WHERE id = $1`,
        [ctx.entityId],
      );
    },
  },
});

The builder automatically:

Filters by stream prefix: Events from unrelated aggregates are skipped without invoking handlers.
Extracts the entity ID: Strips the prefix from the stream ID (e.g. "Order-123" becomes "123").
Provides typed data: Handlers receive strongly typed event payloads.
Provides atomic ctx.client: All SQL operations inside the handler execute in the same transaction as the projection checkpoint update.

`ProjectionHandlerContext`

Each handler receives a context object:

interface ProjectionHandlerContext {
  entityId: string; // Aggregate entity ID (e.g. "123")
  streamId: string; // Full stream identifier (e.g. "Order-123")
  globalPosition: bigint; // Global position in the event store
  streamVersion: number; // Version of the stream at this event
  createdAt: Date; // Timestamp when event was recorded
  client: PoolClient; // PostgreSQL transaction client
}

Raw `Projection` Interface

For cross-stream projections that listen to multiple aggregate prefixes or custom event shapes, implement the Projection interface directly:

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

const multiStreamProjection: Projection = {
  projectionName: "global-activity-feed",
  async handle(event: StoredEvent, client: PoolClient) {
    if (event.type === "OrderPlaced") {
      const data = event.data as { total: number };
      await client.query(
        `INSERT INTO activity_feed (stream_id, action, amount) VALUES ($1, $2, $3)`,
        [event.streamId, "order_placed", data.total],
      );
    }
  },
};

Running Projections with `runProjection()`

Call eventStore.runProjection(projection, batchSize) to process the next batch of pending events:

// Process up to 500 events in a single batch
const processedCount = await eventStore.runProjection(
  orderSummaryProjection,
  500,
);
console.log(`Processed ${processedCount} events.`);

Background Scheduling Loop

Run your projections continuously in a background worker:

async function startProjectionWorker() {
  while (true) {
    try {
      const count = await eventStore.runProjection(orderSummaryProjection, 500);

      // If we processed a full batch, process the next batch immediately
      if (count >= 500) {
        continue;
      }

      // Otherwise wait briefly before polling again
      await new Promise((resolve) => setTimeout(resolve, 1000));
    } catch (err) {
      console.error("Projection execution failed:", err);
      // Wait before retrying on database error
      await new Promise((resolve) => setTimeout(resolve, 5000));
    }
  }
}

startProjectionWorker();

How Projections Work Internally

Checkpoint Tracking: The projection's progress is tracked in the projections table (last_position column).
Row Locking: On each runProjection() invocation, Alvyn acquires a row-level lock (FOR UPDATE) on the projection's checkpoint row.
Commit-Safe Watermark: Alvyn queries pending events with WHERE global_position > last_position AND global_position <= safeWatermark ORDER BY global_position ASC LIMIT batchSize. The watermark prevents skipping out-of-order commits from concurrent transactions.
Sequential Processing: Alvyn invokes your handler for each event sequentially.
Atomic Commit: The read-model updates and the last_position checkpoint update commit together in the single PostgreSQL transaction.

Because the read model queries and the checkpoint update share the same transaction via ctx.client, projection processing is exactly-once within PostgreSQL. If the worker crashes or a SQL query fails, the entire batch rolls back and the checkpoint remains unchanged.

Best Practices

Idempotent Handlers: Use INSERT ... ON CONFLICT DO UPDATE or idempotent SQL writes in handlers to remain resilient against rebuilds.
Non-PII Read Models: Projections read raw stored event rows and do not decrypt crypto-shredded fields. Avoid building long-lived public read models from PII without crypto consideration.
Dedicated Workers: Run heavy background projections on dedicated worker instances to prevent projection queries from competing with latency-critical API requests.

Subscriptions — Fan-out streaming for GraphQL, WebSockets, and SSE.
Transactional Outbox — Relay domain events to Kafka, NATS, or RabbitMQ.
Aggregates — Write-side domain models and event emission.

Projections

Projections

Projections vs. Snapshots

Defining a Projection

Using the `defineProjection` Builder (Recommended)

`ProjectionHandlerContext`

Raw `Projection` Interface

Running Projections with `runProjection()`

Background Scheduling Loop

How Projections Work Internally

Best Practices

On this page