This module provides the projection system for transforming event streams into queryable read models using PostgreSQL.

Overview

Projections subscribe to events from ANY backend (Memory, Filesystem, or PostgreSQL) and execute handlers within PostgreSQL transactions. State is persisted in the projections table, enabling:

• Automatic resumption after restarts
• Progress tracking via cursors
• LISTEN/NOTIFY for efficient waiting

Backend-Agnostic Design

The key insight: projection execution and state are ALWAYS in PostgreSQL, but events can come from any storage backend:

-- Events from MemoryStore, projections in PostgreSQL
runProjection projId pool Nothing memoryStore handlers

-- Events from PostgreSQL, projections in PostgreSQL
runProjection projId pool Nothing sqlStore handlers

This enables testing with fast in-memory events while validating real SQL projection logic.

Asynchronous Projections (This Module)

Projections run in a separate thread and process events asynchronously:

• Eventually consistent (small delay between insert and projection update)
• Work with ANY event store backend
• Failures don't block event insertion

For synchronous projections (PostgreSQL-only), see Hindsight.Store.PostgreSQL.

Example: User Directory Projection

import Hindsight.Projection
import Hindsight.Projection.Matching (ProjectionHandlers(..))
import Hindsight.Store (match)

-- Define handler
userProjection :: ProjectionHandlers '["user_registered"] backend
userProjection =
  match "user_registered" handleUser :-> ProjectionEnd
  where
    handleUser envelope = do
      let user = envelope.payload
      Transaction.statement (user.userId, user.email)
        [resultlessStatement|
          INSERT INTO users (id, email) VALUES ($1 :: uuid, $2 :: text)
        |]

-- Run projection
main = do
  pool <- createPool postgresSettings
  store <- newMemoryStore  -- Or any backend
  runProjection (ProjectionId "users") pool Nothing store userProjection

Waiting for Projection Progress

Use waitForEvent to synchronize with projection progress:

result <- insertEvents store Nothing batch
case result of
  SuccessfulInsertion cursor -> do
    -- Wait for projection to catch up
    bracket
      (Connection.acquire settings)
      Connection.release
      (waitForEvent projId cursor.finalCursor)

This uses PostgreSQL LISTEN/NOTIFY for efficient waiting without polling.

Projection State Management

Projection state is tracked in the projections table with:

• id - Unique projection identifier
• last_updated - Timestamp of last event processed
• head_position - Cursor position of last processed event (JSON)
• is_active - Whether projection is currently running

The loadProjectionState function reads this state, and handlers automatically update it after each successful event.