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 globalPosition order. Sequence allocation order is not transaction commit order; the commit-safe watermark holds back positions until earlier in-flight writes are resolved.

Projections vs. Snapshots

Defining a Projection

Using the defineProjection Builder (Recommended)

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 "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 and handler type: Events from unrelated aggregates or event types without a handler (including snapshot events) are skipped before decryption/upcasting.
• 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.

Redacted events after crypto-shredding

An event encrypted with a revoked key is delivered only through onRedacted, not through its regular typed handler. event.data contains the original stored, non-encrypted fields with encrypted paths set to null, while event.redactedPaths lists those field paths and event.schemaVersion identifies the original payload schema. This partial payload is not upcasted: an upcaster may require PII that no longer exists. Inspect or validate the remaining data before using it.

const issueProjection = defineProjection<IssueEvents>()({
  projectionName: "issues",
  streamPrefix: "Issue",
  handlers: {
    /* handlers for complete events */
  },
  onRedacted: async (event, ctx) => {
    // Example only: handle each historical schema explicitly.
    if (event.type === "IssueCreated" && event.schemaVersion === 1) {
      const data = event.data as { title: string };
      await ctx.client.query(
        `INSERT INTO issues (id, title, assignee) VALUES ($1, $2, NULL)`,
        [ctx.entityId, data.title],
      );
      return;
    }
    throw new Error(
      `Unhandled redacted event: ${event.type} v${event.schemaVersion}`,
    );
  },
});

If the projection can safely discard every handled redacted event, set onRedacted: "skip" explicitly. Typed projections skip event types without a handler before decryption; custom Projection implementations may provide handlesEventType(type) for the same optimization. Without a policy, processing fails and the checkpoint rolls back rather than silently omitting an event. A missing crypto configuration or a missing key row is an error even with "skip"; only revoked keys count as intentional redaction. onRedacted can also be defined on the raw Projection interface with (event, client) arguments.

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 "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

1. Checkpoint Tracking: The projection's progress is tracked in the projections table (last_position column).
2. Row Locking: On each runProjection() invocation, Alvyn acquires a row-level lock (FOR UPDATE) on the projection's checkpoint row.
3. 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.
4. Sequential Processing: Alvyn invokes your handler for each event sequentially.
5. Atomic Commit: The read-model updates and the last_position checkpoint update commit together in the single PostgreSQL transaction.

Best Practices

• Idempotent Handlers: Use INSERT ... ON CONFLICT DO UPDATE or idempotent SQL writes in handlers to remain resilient against rebuilds.
• External Effects: HTTP calls, broker publishes, and writes through another connection are not atomic with the checkpoint and may repeat after failure. Use an outbox and downstream deduplication for these effects.
• Schema and PII: Complete projection events are decrypted and then upcast. Redacted events retain only public fields in their original schema and use the explicit onRedacted policy. Projection handlers can receive PII: restrict access to read models accordingly. Revoking a key does not remove PII already stored in read models; clean those up explicitly.
• Dedicated Workers: Run heavy background projections on dedicated worker instances to prevent projection queries from competing with latency-critical API requests.

Related Topics

• 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.