Persistence Query

Das PersistenceQuery-Interface ist die Read-Side-API für das Journal. So konsumieren Projektionen Events; es ist auch das Interface für jeden One-Shot-Read historischer Events.

Zwei Varianten:

Variante	Gibt zurück	Verwendung
*`current`**	`Promise<Event[]>`	Batch-Read von Events zur Aufrufzeit. Wird einmal aufgelöst.
*`events`**	`AsyncIterable<Event>`	Live-Stream. Pollt, bis der Konsument stoppt.

Implementierungen

Query	Paart sich mit
`InMemoryQuery`	`InMemoryJournal` — und jedem Journal ohne eigene Query
`SqliteQuery`	`SqliteJournal`
`PostgresQuery`	`PostgresJournal`
`MariaDbQuery`	`MariaDbJournal` (auch MySQL)
`CassandraQuery`	`CassandraJournal`
`MongoQuery`	`MongoJournal`
`RelationalQuery`	jedes `RelationalJournal` — die dialektneutrale Basis, von der die beiden SQL-Zeilen darüber benannte Subklassen sind

Das Interface ist über alle hinweg gleich — konstruiere die Query für dein Journal-Backend und übergib ihr die Journal-Instanz, nicht einen Connection-String:

import { PostgresJournal, PostgresJournalOptions, PostgresQuery } from 'actor-ts/persistence';

// PostgresQuery liest dasselbe Journal, in das deine PersistentActors schreiben.
const journalOptions = PostgresJournalOptions.create().withUrl(process.env.DATABASE_URL!);
const journal = new PostgresJournal(journalOptions);
const query = new PostgresQuery(journal);

Wenn es für dein Backend keine Query gibt

InMemoryQuery funktioniert gegen jedes Journal — sie ist die Referenzimplementierung, nicht die Nur-für-In-Memory-Variante. Ein Backend, das in der Tabelle oben fehlt, ist also nicht von Projektionen ausgeschlossen. Was fehlt, ist der Index, und zwar nur bei den By-Tag-Methoden:

currentEventsByPersistenceId und die Id-Methoden delegieren direkt ans Journal und kosten daher in beiden Fällen gleich viel.
currentEventsByTag hat keinen Index, in den der Filter geschoben werden könnte. Die Methode listet jede Persistence-Id auf, spielt jede davon ab Sequenznummer 1 erneut ab und filtert im Prozess — das gesamte Journal, pro Aufruf, und eine By-Tag-Projektion wiederholt das alle pollIntervalMs.

Das ist bei jedem Volumen korrekt und nur bei kleinem schnell. Eine eigene Query-Klasse gibt es genau dort, wo das Backend einen Tag-Index führt, der sich zu lesen lohnt: die SQL-Backends pflegen eine events_tags-Join-Tabelle, Cassandra eine events_by_tag-Seitentabelle, MongoDB einen Index auf dem tags-Array.

Die Methoden

`currentEventsByPersistenceId`

const events = await query.currentEventsByPersistenceId<AccountEvent>(
  'account-42',
  /* fromSeq */ 1,
  /* toSeq? */ 100,
);

One-Shot-Read jedes Events für eine Entity, zwischen den Sequenznummern fromSeq und toSeq (inklusiv).

Verwende für:

Audit-Replays — hole jedes Event für einen User.
Einmalige Backfills — verarbeite Events von einer Entity in ein separates System.

`eventsByPersistenceId`

for await (const event of query.eventsByPersistenceId<AccountEvent>('account-42', 1)) {
  console.log(event.event.kind, event.sequenceNr);
  if (someCondition) break;   // stoppt das Polling
}

Live-Stream. Yieldet zuerst vergangene Events (chronologisch nach sequenceNr), dann neue Events, sobald sie angehängt werden. Der Stream beendet sich nie von selbst — breake aus der Schleife, um zu stoppen.

Verwende für:

Live-Per-Entity-Views — ein Echtzeit-Feed der Aktivität eines Users.
Single-Entity-Projektionen, die ad hoc gebaut werden statt über ProjectionActor.byPersistenceId.

`currentEventsByTag`

const events = await query.currentEventsByTag<Event>(
  'account',
  offsetStart,
);

One-Shot-Read jedes Events, das zu einem Tag passt (oder Tag-Filter), ab einem Offset.

Gibt [Event, Offset][]-Paare zurück — der Offset ist das, was du einem folgenden Aufruf weitergeben würdest, um den nächsten Batch zu lesen.

Verwende für:

Berichte — summiere alle account-Events des aktuellen Monats.
Batch-ETL — periodischer Export aller getaggten Events.

`eventsByTag`

for await (const [event, offset] of query.eventsByTag<Event>('account', offsetStart)) {
  await handle(event);
  await offsetStore.save('my-projection', offset);
}

Live-Stream tag-gematchter Events. Das ist das, was ProjectionActor.byTag intern verwendet — der Offset lässt die Projektion ihren Fortschritt persistieren und von dort fortsetzen.

Persistence-Ids

Drei Methoden beantworten “welche Entities liegen in diesem Journal?”, und sie unterscheiden sich darin, was sie kosten und wann sie aufhören.

Methode	Gibt zurück	Hört auf
`currentPersistenceIds()`	`Promise<string[]>`	Sofort — die ganze Liste, ein Array.
`currentPersistenceIdsPaginated()`	`AsyncIterable<string>`	Wenn das Journal erschöpft ist.
`allPersistenceIds()`	`AsyncIterable<string>`	Nie — es kommen weiter neue Ids.

`currentPersistenceIds`

const ids = await query.currentPersistenceIds();

Die ganze Liste in einem Array. Bequem, und bei kleiner Id-Anzahl die richtige Wahl — ein paar hundert Entities zu materialisieren kostet nichts. Es ist allerdings die einzige Methode hier ohne Schranke für das, was sie allokiert: eine Million Entities sind eine Million Strings.

`currentPersistenceIdsPaginated`

for await (const persistenceId of query.currentPersistenceIdsPaginated({ pageSize: 500 })) {
  await backfill(persistenceId);
}

Dieselben Daten, seitenweise geholt — der Spitzenverbrauch ist damit eine Seite statt der gesamten Menge. Endet, sobald das Journal erschöpft ist.

Der Cursor ist eine Persistence-Id, kein undurchsichtiges Token — setze einen abgebrochenen Durchlauf fort, indem du die zuletzt verarbeitete Id übergibst:

for await (const persistenceId of query.currentPersistenceIdsPaginated({
  afterPersistenceId: lastHandled,
})) {
  // ... beginnt strikt nach `lastHandled`
}

Option	Default	Was
`pageSize`	256	Ids pro Round-Trip.
`afterPersistenceId`	—	Fortsetzen nach dieser Id, exklusiv.

Die Ids kommen in der aufsteigenden Ordnung des Backends zurück — bytewise bei SQLite und Cassandra, collation-abhängig bei Postgres. Welche es ist, spielt für den Konsumenten nie eine Rolle, weil der Cursor in genau derselben Ordnung verglichen wird, nach der die Seite sortiert ist.

`allPersistenceIds`

for await (const persistenceId of query.allPersistenceIds()) {
  startProjectionFor(persistenceId);   // läuft endlos
}

Live-Stream: jede Id, die das Journal je gesehen hat, danach jede neue, sobald sie zum ersten Mal auftaucht. Das ist das Fan-out-Primitiv — starte einen Konsumenten pro Entity, sobald diese auftaucht, statt currentPersistenceIds zu pollen und das Ergebnis selbst zu diffen.

Zwei Eigenschaften, die man kennen sollte, bevor man darauf baut:

Einmal pro Stream, nicht einmal pro Journal. Eine neue Subscription emittiert wieder jede Id. Ein Konsument, der über einen Neustart hinweg nicht zweimal handeln darf, braucht seinen eigenen Checkpoint — dieselbe At-least-once-Haltung wie bei den Event-Queries.
Der Stream merkt sich, was er emittiert hat. Das ist ein In-Process-Set von Ids; ein Stream über ein Journal mit einer Million Entities hält also eine Million Strings, solange er läuft. Eine billigere High-Water-Mark wäre falsch: Ids entstehen nicht in sortierter Reihenfolge, eine Entity, deren Id unterhalb der Marke einsortiert, würde also nie emittiert. Für eine einmalige Aufzählung, die kein solches Set braucht, nimm currentPersistenceIdsPaginated.

Offsets

type Offset = {
  readonly timestamp: number;
  readonly persistenceId: string;
  readonly sequenceNr: number;
};

Compound: (timestamp, persistenceId, sequenceNr). Das ist die Reihenfolge, in der Events herausströmen.

timestamp — die Schreibzeit des Journals für das Event.
persistenceId — Tiebreaker für Events mit demselben Timestamp.
sequenceNr — finaler Tiebreaker für Events in derselben pid.

Die Compound-Reihenfolge bedeutet, dass Events aus verschiedenen Entities nach Timestamp verschachtelt werden, während Events aus derselben Entity in ihrer ursprünglichen Sequenz bleiben.

offsetStart ist die Sentinel-Konstante “vom Anfang.” Persistiere den Offset, der zusammen mit jedem Event zurückgegeben wird; setze beim Neustart vom persistierten Offset fort.

Tag-Filter

import type { TagFilter } from 'actor-ts/persistence';

// Alle Events, die mindestens einen dieser Tags haben:
query.eventsByTag<E>({ any: ['account', 'audit'] }, offset);

// Events, die ALLE diese Tags haben:
query.eventsByTag<E>({ all: ['account', 'production'] }, offset);

// AND über zwei Gruppen:
query.eventsByTag<E>({
  all: ['account'],
  any: ['debit', 'credit'],
}, offset);

Für einfache Fälle funktioniert ein einzelner Tag-String:

query.eventsByTag<E>('account', offset);   // äquivalent zu { all: ['account'] }

Polling

const stream = query.eventsByTag<E>('account', offset, {
  pollIntervalMs: 500,   // Default 1000
});

Live-Streams pollen das Journal. Konfigurierbar über LiveQueryOptions:

Option	Default	Was
`pollIntervalMs`	1000	Wie oft auf neue Events geprüft wird.

Für Sub-Poll-Intervall-Delivery siehe Push-basierte Query.

Wann direkt verwenden vs. über ProjectionActor

// Manuelle Iteration — volle Kontrolle:
for await (const [event, offset] of query.eventsByTag(...)) {
  // ... benutzerdefinierte Fehlerbehandlung, benutzerdefiniertes Dispatching ...
}

// ProjectionActor — supervised, neustartbar, offset-managed:
ProjectionActor.byTag(system, ByTagProjectionOptions.create() /* .withName(...).withTag(...).withQuery(...).withHandle(...) */);

ProjectionActor wickelt die Query-API ein mit:

Offset-Persistenz (du verwaltest sie nicht).
Supervised-Restart bei Handler-Fehlern.
Bewusstsein für koordiniertes Shutdown.
Konfigurierbarem Retry / Backoff.

Für Produktions-Projektionen verwende ProjectionActor. Für Einmal-Skripte und Ad-hoc-Admin-Queries verwende die Query-API direkt.

Performance-Hinweise

current*-Reads scannen das Journal einmal, geben ein materialisiertes Array zurück. Für große pids / viele Tag-Matches kann das langsam und speicherintensiv sein. Verwende die streamenden events*-Varianten für unbegrenzte Reads und currentPersistenceIdsPaginated statt currentPersistenceIds.
Die Id-Paginierung wird ins Backend geschoben, wo der Store einen sortierten Schlüssel über den Ids hat: ein ORDER BY … LIMIT bei SQLite und den SQL-Backends, ein Clustering-Column-Range über der Cassandra-Tabelle all_persistence_ids. MongoDB und DynamoDB haben keinen solchen Index — distinct hat keinen Cursor, und ein DynamoDB-Scan hat über Partition-Keys hinweg keine Ordnung — dort wird die Seite in-process geschnitten, und die Paginierung bringt Ergonomie, kein gespartes I/O.
Live-Polling ist im Idle leicht — leere Queries kehren schnell zurück. Unter hoher Schreiblast kann sich das Polling stauen, wenn die Read-Seite langsamer ist als die Write-Seite. Parallelisiere den Handler, wenn du Lag siehst.
Tag-Indexes sind wichtig — der events_tags-Index ist das, was By-Tag-Queries schnell macht. Das SQLite-Journal erstellt diesen automatisch.
Ein kompaktiertes Event verlässt auch den Tag-Index. delete ist gegen die Read-Seite genauso spezifiziert wie gegen read: sobald ein Präfix kompaktiert ist, gibt currentEventsByTag es nicht mehr zurück. Backends, deren Index auf dem Event-Datensatz selbst sitzt (MongoDBs Multikey-tags), bekommen das geschenkt; die mit einer separaten Struktur — die Tags-Tabelle von SQLite und den SQL-Backends, Cassandras events_by_tag — löschen im selben Aufruf daraus, und zwar vor den Events, damit ein Crash mitten im Delete keine Zeilen zurücklässt, deren Schlüssel sich nicht mehr rekonstruieren lassen. Wichtig zu wissen, wenn du ein eigenes Journal schreibst: read und highestSeq sehen völlig korrekt aus, während ein vergessener Index gelöschte Events weiter ausliefert.

Wie geht’s weiter

Persistenz im Überblick — das größere Bild.
Projektionen — der Higher-Level-Wrapper um die Query-API.
Push-basierte Query — für Sub-Poll-Intervall-Live-Delivery.
Journals — die Write-Seite, die die Query füttert.

Die PersistenceQuery-API-Referenz deckt die vollständige Oberfläche ab.