Projection Pipeline

The projection pipeline takes enriched events from Kafka and materializes them into per-tenant PostgreSQL tables. It is the read-side of Causet’s event-sourced architecture.

Pipeline Overview

Kafka Consumer Group

projection-worker operates as a Kafka consumer group. Multiple worker instances share the same consumer group ID, allowing parallel processing of partitions.

Partitioning in causet.projection-events.v1 is by entity_id. This ensures all events for a given entity are processed in order by the same worker instance.

Event Deserialization

Each message in causet.projection-events.v1 is a projection event envelope:

{
  "eventType": "ARTIST_FOLLOWED",
  "entityId": "user-123",
  "forkId": "prod",
  "tenantSchema": "platform1_concertapp_prod",
  "irVersion": "v42",
  "sequenceNumber": 1891,
  "ts": 1719331200000,
  "payload": {
    "user_id": "user-123",
    "artist_id": "artist-456"
  }
}

IR Loading

For each event (or on startup), the worker resolves the active IR version:

  1. Calls causet-saas-cloud GetActiveRelease(forkId) to get the current irVersion.
  2. Checks the Redis cache for causet.projections.json at key ir:{irVersion}.
  3. On cache miss, loads the artifact from S3: s3://{bucket}/{irVersion}/causet.projections.json.
  4. Caches the loaded artifact in Redis with a TTL.

The projections IR contains:

  • All declared projections with their source_events, fields, derive expressions, and mutations
  • Compiled SQL templates for each projection operation
  • The target table name and primary key per projection

Matching and Evaluation

For each incoming event:

  1. The worker looks up all projections whose source_events list includes the event type.
  2. For each matched projection, derive expressions are evaluated against the event payload and metadata to produce field values.
  3. The configured mutation operation is determined for this event type.

Example DSL:

projections:
  artist_followers:
    source_events: [ARTIST_FOLLOWED, ARTIST_UNFOLLOWED]
    target:
      table: artist_followers
      primary_key: [user_id, artist_id]
    fields:
      user_id:     TEXT
      artist_id:   TEXT
      followed_at: BIGINT
    derive:
      user_id:     event.user_id
      artist_id:   event.artist_id
      followed_at: event.ts
    mutations:
      ARTIST_FOLLOWED:   { op: upsert }
      ARTIST_UNFOLLOWED: { op: delete }

For an ARTIST_FOLLOWED event, the worker evaluates:

  • user_id = event.user_id → "user-123"
  • artist_id = event.artist_id → "artist-456"
  • followed_at = event.ts → 1719331200000

Then executes an UPSERT into artist_followers with these values.

Tenant Schema Resolution

Before executing any SQL, the worker issues:

SET LOCAL search_path = platform1_concertapp_prod;

The tenant schema name comes from the event envelope’s tenantSchema field. It is computed at event publish time from the fork’s platformId, applicationId, and forkId, sanitized to ≤ 63 characters.

This isolates all reads and writes to the correct tenant’s tables. The schema and its tables are created at deploy time from the projections IR DDL.

Mutation Operations

OperationBehavior
upsertInsert or update the row by primary key
deleteDelete the row matching the primary key
replaceDelete existing rows matching a filter, then insert new rows

Checkpoint: Kafka Offset Commit

After successfully processing all projections for a batch of events, the worker commits the Kafka consumer group offset. If the worker crashes between processing and committing, the events will be re-delivered and re-processed. Projections must be idempotent — upsert and delete by primary key satisfy this requirement.

Warning: If the worker crashes after writing to PostgreSQL but before committing the Kafka offset, duplicate processing will occur on restart. The projection mutations are idempotent so the outcome is correct, but upsert will produce a duplicate write.

Failure and Retry

If a projection handler throws an exception:

  1. The worker retries the event with exponential backoff up to the configured maximum retry count.
  2. Each retry attempt is logged with the error, retry number, and event details.
  3. After max retries are exhausted, the event is written to causet.projection-dlq.v1 and the consumer offset advances past it.

A failed event does not block other events or other projections. Processing continues for all other events in the partition.

Dead Letter Queue

Events that exhaust all retries are published to causet.projection-dlq.v1 with full context:

{
  "originalEvent": { ... },
  "projectionName": "artist_followers",
  "error": "column \"artist_id\" does not exist",
  "retryCount": 5,
  "failedAt": 1719331800000
}

See Dead Letter Queues for how to inspect and recover from DLQ events.

Projection Rebuild

To rebuild a projection from scratch (e.g. after fixing a projection bug or adding a new projection to historical events):

  1. Stop the projection-worker (or remove the specific projection handler).
  2. Optionally truncate the target projection table.
  3. Reset the Kafka consumer group offset for causet.projection-events.v1 to the beginning (or a specific checkpoint).
  4. Restart the projection-worker.

Events are re-processed from the reset offset. All mutations are idempotent by design.

Warning: Side effects from side_effects rules (external calls, downstream submits) are triggered at rule evaluation time on the write path, not during projection processing. Rebuilding a projection does not re-trigger side effects.

See Replay for the full rebuild procedure.

Related Pages

What is Causet?