Projektionen

Ein PersistentActor schreibt Events. Eine Projektion konsumiert sie und baut eine Read-Side-View, die für Queries zugeschnitten ist:

Das Journal ist Append-only und autoritativ. Projektionen sind abgeleitet — sie können von Grund auf neu aufgebaut werden, indem das Journal abgespielt wird. Das entkoppelt Write-Durchsatz (durable, append-only) von Read-Durchsatz (denormalisiert, query-optimiert).

Ein minimales Beispiel

import { match } from 'ts-pattern';
import { ProjectionActor, ByTagProjectionOptions } from 'actor-ts';
import { SqliteJournal, SqliteJournalOptions, SqliteQuery, InMemoryOffsetStore } from 'actor-ts';

type AccountEvent =
  | { kind: 'deposited'; amount: number }
  | { kind: 'withdrawn'; amount: number };

// Die Read-Seite fragt dasselbe Journal ab, in das deine PersistentActors
// schreiben — SqliteQuery umschließt eine SqliteJournal-Instanz (keinen Pfad).
const eventsJournal = new SqliteJournal(SqliteJournalOptions.create().withPath('/var/lib/events.db'));

const byTagProjectionOptions = ByTagProjectionOptions.create<AccountEvent>()
  .withName('account-balance-view')
  .withTag('account')
  .withQuery(new SqliteQuery(eventsJournal))
  .withOffsetStore(new InMemoryOffsetStore())
  .withHandle(async (event) => {
    await match(event.event)
      .with({ kind: 'deposited' }, (e) => viewDb.execute(
        'UPDATE balances SET balance = balance + ? WHERE pid = ?',
        [e.amount, event.persistenceId],
      ))
      .otherwise(() => Promise.resolve());
  });
const projection = ProjectionActor.byTag<AccountEvent>(system, byTagProjectionOptions);

Der Actor:

1. Lädt seinen Offset-Cursor beim preStart aus dem Offset-Store.
2. Pollt die Query-Schicht für Events, die zum tag passen, ab dem Cursor.
3. Ruft handle für jedes Event auf.
4. Persistiert den neuen Cursor.
5. Wiederholt.

Zwei Query-Formen

Factory	Cursor-Typ	Verwendung
ProjectionActor.byPersistenceId(...)	sequenceNr (pro pid)	Lies die vollständige Historie einer Entity.
ProjectionActor.byTag(...)	Offset (Timestamp + Tiebreaker)	Lies alles, was mit <tag> getaggt ist, im Journal.

Tag-basiert ist der häufige Fall — der PersistentActor ruft tagsFor(event) auf, um Events zu labeln; die Projektion abonniert das Tag. Per-pid ist nützlich für schmale Views (die Aktivität eines Users).

At-least-once Delivery

// Crash-Recovery-Sequenz:
//   1. Offset-Cursor auf Wert N speichern.
//   2. Handler verarbeitet Event N+1.
//   3. Crash bevor der Cursor gespeichert wird.
//   4. Beim Neustart: Cursor ist immer noch N → Handler erhält Event N+1 erneut.

Wenn der Handler läuft, aber der Cursor nicht persistiert wird, verarbeitet die Projektion dasselbe Event erneut beim Neustart. Das ist At-least-once Delivery — das Framework garantiert, dass kein Event verpasst wird, aber Duplikate sind möglich.

Handler müssen idempotent sein:

• UPSERT in das Read-Model (kein blindes INSERT).
• Verarbeitete Event-IDs verfolgen im Read-Model selbst für Dedup.
• Die sequenceNr des Events als Per-pid-Dedup-Key verwenden — sie nimmt nie ab, ist monoton pro persistenceId.

Wenn du den Handler nicht idempotent machen kannst, muss die Projektion an einem 2-Phase-Commit mit dem Offset-Save teilnehmen — viel komplexer, nicht out-of-the-box bereitgestellt.

OffsetStore

import { InMemoryOffsetStore, DurableStateOffsetStore } from 'actor-ts';

// Default (verloren beim Neustart):
new InMemoryOffsetStore();

// Dauerhaft — speichert den Cursor in einem beliebigen DurableStateStore
// (Postgres, MariaDB, libSQL, SQL Server, MongoDB, DynamoDB, Object-Storage), sodass ein Neustart fortsetzt:
new DurableStateOffsetStore(durableStateStore);

Der Cursor ist nur eine Zahl (oder ein Compound-Offset) — pro Projection-Name + Scope gespeichert. Implementierungen:

• InMemoryOffsetStore — okay für Tests, nutzlos in Produktion (jeder Neustart verarbeitet vom Anfang neu).
• DurableStateOffsetStore — umschließt einen beliebigen DurableStateStore (Postgres, MariaDB, libSQL, SQL Server, MongoDB, DynamoDB, Object-Storage), sodass Offsets Neustarts überstehen.
• Benutzerdefiniert — implementiere OffsetStore gegen deinen eigenen Store (Redis, was auch immer dein Read-Model verwendet).

Für echte Produktions-Setups kollokiere den Offset-Store mit dem Read-Model, sodass sie zusammen abstürzen — das minimiert das Re-Processing-Fenster.

Polling

const byTagProjectionOptions = ByTagProjectionOptions.create()
  .withName('...')
  .withTag('...')
  .withLiveOptions({
    pollIntervalMs: 500,    // Default 1000 ms
  });
ProjectionActor.byTag(system, byTagProjectionOptions
  /* .withQuery(...).withOffsetStore(...).withHandle(...) */);

Die Projektion pollt. Im Idle sind die Poll-Kosten eine Journal-Query pro pollIntervalMs. Tuning:

• Niedriger (250-500 ms) → schnellere End-to-End-Propagation, mehr Datenbank-Last.
• Höher (5-10 s) → langsame Sichtbarkeit, aber billig.

Für sehr niedriglatente Read-Model-Updates siehe Push-basierte Query, die Sub-Poll-Intervall-Delivery über den In-Process-Event-Bus erhält.

Per-pid-Projektionen

const byPidProjectionOptions = ByPersistenceIdProjectionOptions.create<AccountEvent>()
  .withName('account-42-view')
  .withPersistenceId('account-42')
  .withQuery(query)
  .withOffsetStore(offsetStore)
  .withHandle(async (event) => {
    // ... nur die Events dieses Accounts behandeln
  });
const projection = ProjectionActor.byPersistenceId<AccountEvent>(system, byPidProjectionOptions);

Ein Actor pro persistenceId. Nützlich für Per-Entity-Views:

• Eine “User-Activity-Timeline”-Projektion pro User.
• Eine “Per-Order-Audit-Trail”-Projektion pro Order.

Für große Zahlen von pids ist das nicht der Weg zu skalieren — eine Projektion pro pid zu spawnen skaliert nicht auf Millionen von Usern. Verwende dafür eine einzelne Tag-basierte Projektion, die nach pid hasht.

Fan-out: eine starten, sobald eine Entity auftaucht

Eine Per-pid-Projektion muss ihre pid vorab kennen — passend für eine feste Menge, nutzlos für eine wachsende. allPersistenceIds schließt diese Lücke: ein Live-Stream jeder Id, die das Journal gesehen hat, plus jeder neuen, sobald sie zum ersten Mal auftaucht:

for await (const persistenceId of query.allPersistenceIds()) {
  const options = ByPersistenceIdProjectionOptions.create<AccountEvent>()
    .withName(`view-${persistenceId}`)
    .withPersistenceId(persistenceId)
    .withQuery(query)
    .withOffsetStore(offsetStore)
    .withHandle(async (event) => { /* ... */ });
  ProjectionActor.byPersistenceId<AccountEvent>(system, options);
}

Zwei Vorbehalte, dieselben wie bei der Per-pid-Form — hier nur sichtbar gemacht:

• Die Schleife endet nie, führe sie also in einem eigenen Task aus und breake beim Shutdown heraus.
• Ein Actor pro Entity bleibt ein Actor pro Entity. Fan-out ist für hunderte oder tausende langlebige Entities gedacht, nicht für Millionen kurzlebiger — dort ist eine Tag-Projektion die Antwort.

Für einen einmaligen Durchlauf statt eines Live-Streams — ein Backfill über die Entities, die es jetzt gibt — nimm currentPersistenceIdsPaginated: es endet, sobald das Journal erschöpft ist, und hält eine Seite auf einmal statt der ganzen Id-Liste. Siehe Persistence Query.

Mehrere Projektionen, dasselbe Journal

// Drei Projektionen, drei Offset-Cursors, drei unabhängige Views:
const byTagProjectionOptions = ByTagProjectionOptions.create<E>().withName('balance');
ProjectionActor.byTag<E>(system, byTagProjectionOptions /* .withTag(...).withQuery(...).withHandle(...) */);
const byTagProjection2Options = ByTagProjectionOptions.create<E>().withName('audit-log');
ProjectionActor.byTag<E>(system, byTagProjection2Options /* .withTag(...).withQuery(...).withHandle(...) */);
const byTagProjection3Options = ByTagProjectionOptions.create<E>().withName('monthly-stats');
ProjectionActor.byTag<E>(system, byTagProjection3Options /* .withTag(...).withQuery(...).withHandle(...) */);

Jede hat ihren eigenen Offset-Cursor; sie lesen unabhängig vom Journal. Das ist die Stärke von Event Sourcing — ein Event-Stream, viele abgeleitete Views, keine an die andere gekoppelt.

Idempotenz ist nicht optional

await db.execute('INSERT INTO events (...) VALUES (...)');   // ✗ Duplikate beim Retry

Ohne Idempotenz wird At-least-once zu “gelegentlich doppelt.” Der Retry passiert bei jedem Neustart, der einen ausstehenden Handler hat. Verwende UPSERT oder schließe die sequenceNr des Events in die Zeile ein und dedupe zur Schreib-Zeit.

Von Grund auf neu aufbauen

# Um eine Projektion von vorne neu aufzubauen:
DELETE FROM offsets WHERE projection_name = 'balance';
# Oder gib beim nächsten Start einen frischen OffsetStore mit.

Das Zurücksetzen des Offsets spielt jedes Event für diesen Tag ab. Für große Journals kann das eine Weile dauern. Manchmal beabsichtigt (Schema-Migration des Read-Models); meist nicht das, was du zufällig willst.

Langsame Handler bremsen die Projektion

async handle(event) {
  await callSlowExternalService();   // 500ms pro Event
}

Die Handler-Latenz begrenzt direkt den Durchsatz. Bei 500 ms/Event verarbeitet die Projektion 2 Events/sec. Für Journals mit hohem Durchsatz parallelisiere innerhalb des Handlers (Promise.all) oder splitte die Arbeit in mehrere Projektionen.

Wie geht’s weiter

• Persistenz im Überblick — das größere Bild.
• PersistentActor — was die Events produziert.
• Persistence Query — die Read-Side-API, die die Projektion verwendet.
• Push-basierte Query — Sub-Poll-Intervall-Delivery über den Event-Bus.

Die ProjectionActor-API-Referenz deckt alle Settings ab.