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

Jedes Journal hat eine passende Query:

Query	Paart sich mit
InMemoryQuery	InMemoryJournal
SqliteQuery	SqliteJournal
CassandraQuery	CassandraJournal

Das Interface ist über alle hinweg gleich — konstruiere einfach die passende Query für dein Journal-Backend.

import { SqliteJournal, SqliteJournalOptions, SqliteQuery } from 'actor-ts';

// SqliteQuery liest dasselbe Journal, in das deine PersistentActors schreiben.
const journal = new SqliteJournal(SqliteJournalOptions.create().withPath('/var/lib/events.db'));
const query = new SqliteQuery(journal);

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';

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

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.