Schema-Registry

Für größere Codebases mit vielen Event-Typen wird zu wissen, was existiert, ein Problem. Jeder PersistentActor deklariert seine Events; einige haben Adapter; Versionen, Codecs und Upcaster liegen verstreut in den Adapter-Configs. Kein zentraler Katalog.

InMemorySchemaRegistry ist der optionale Katalog. Du registrierst jedes (manifest, version) einmal — mit dem Codec, der diese Version validiert, und dem Upcaster, der die vorherige Version nach vorne bringt — und die Registry baut die Adapter für dich:

import { InMemorySchemaRegistry, zodCodec } from 'actor-ts/persistence';
import { z } from 'zod';

const DepositedV1 = z.object({ kind: z.literal('deposited'), amount: z.number() });
const DepositedV2 = z.object({
  kind:     z.literal('deposited'),
  amount:   z.number(),
  currency: z.enum(['USD', 'EUR']),
});
type DepositedV1 = z.infer<typeof DepositedV1>;
type DepositedV2 = z.infer<typeof DepositedV2>;

export const registry = new InMemorySchemaRegistry();

registry.register('Deposited', 1, { codec: zodCodec(DepositedV1) });
registry.register('Deposited', 2, {
  codec:          zodCodec(DepositedV2),
  upcastFromPrev: (v1: DepositedV1): DepositedV2 => ({ ...v1, currency: 'USD' }),
});

register mutiert die Registry an Ort und Stelle und gibt void zurück; erneutes Registrieren desselben (manifest, version) überschreibt es. Verwende die Registry, um:

• Den aktuellen Stand jedes Schemas zu dokumentieren.
• Payloads zu validieren — der codec jeder Version (typischerweise ein zodCodec) wird sowohl auf dem Schreib- als auch dem Lesepfad erzwungen.
• Adapter zu bauen — registry.eventAdapter(manifest) schreibt bei der neuesten Version und liest jede ältere Version desselben Manifests, indem es die registrierten Upcaster nach vorne verkettet. Ein gespeicherter Frame mit einem anderen Manifest wird mit einem MigrationError abgewiesen, selbst wenn dieses Manifest ebenfalls registriert ist — siehe Ein Adapter, ein Manifest.
• Tooling anzutreiben — Admin-Dashboards, Dev-Tools, die Journal-Inhalte zeigen.

Die meisten Projekte brauchen sie nicht. Greif danach, wenn du 10+ Event-Typen hast und dich dabei ertappst, manuell zu verfolgen, welche bei welcher Version ist.

Ein minimales Beispiel

import { InMemorySchemaRegistry, zodCodec } from 'actor-ts/persistence';

// Geteiltes Modul — `schemas.ts`:
export const registry = new InMemorySchemaRegistry();

registry.register('Deposited', 1, { codec: zodCodec(DepositedV1) });
registry.register('Deposited', 2, {
  codec:          zodCodec(DepositedV2),
  upcastFromPrev: (v1: DepositedV1): DepositedV2 => ({ ...v1, currency: 'USD' }),
});

registry.register('Withdrawn', 1, { codec: zodCodec(WithdrawnV1) });
registry.register('Frozen',    1, { codec: zodCodec(FrozenV1) });

Dann in Actors — keine handgeschriebene Chain; die Registry baut den Adapter:

import { registry } from './schemas.js';

class Account extends PersistentActor<...> {
  override eventAdapter() {
    return registry.eventAdapter<DepositedV2>('Deposited');
  }
}

Der Adapter schreibt neue Events bei der neuesten registrierten Version von 'Deposited' und dekodiert beim Lesen jedes gespeicherte Event mit dem Codec seiner eigenen Version, bevor er die Upcaster nach vorne zur neuesten Form verkettet.

Die API

interface SchemaRegistry {
  register<Wire = unknown, Upcasted = unknown, Previous = unknown>(
    manifest: string,
    version: number,
    registration: SchemaRegistration<Wire, Upcasted, Previous>,
  ): void;
  get(manifest: string, version: number): SchemaDescriptor | undefined;
  latestVersion(manifest: string): number | undefined;
  list(): ReadonlyArray<SchemaDescriptor>;
  eventAdapter<E, JournalShape = E>(manifest: string): EventAdapter<E, JournalShape>;
  snapshotAdapter<S, StoredShape = S>(manifest: string): SnapshotAdapter<S, StoredShape>;
}

type SchemaRegistration<Wire = unknown, Upcasted = unknown, Previous = unknown> = {
  readonly codec: Codec<Wire>;
  readonly upcastFromPrev?: (prev: Previous) => Upcasted;
  readonly compatibility?: 'none' | 'backward' | 'sample';
  readonly sample?: unknown;
};

• register(manifest, version, { codec, upcastFromPrev?, compatibility?, sample? }) — eine Version hinzufügen oder ersetzen. Mutiert an Ort und Stelle, gibt void zurück.
• get(manifest, version) — der SchemaDescriptor für eine Version, oder undefined.
• latestVersion(manifest) — höchste registrierte Version, oder undefined, wenn das Manifest unbekannt ist.
• list() — jede Registrierung als ReadonlyArray<SchemaDescriptor> ({ manifest, version, codec, … }).
• eventAdapter(manifest) / snapshotAdapter(manifest) — den migrierenden Adapter für Events / State bauen.

SchemaRegistry ist ein Interface; die ausgelieferte Implementierung ist InMemorySchemaRegistry (new InMemorySchemaRegistry()), die den gesamten State in einem Prozess hält. Implementiere das Interface selbst, wenn du einen anderen Backing-Store brauchst.

Ein Adapter, ein Manifest

Ein von eventAdapter(m) oder snapshotAdapter(m) gebauter Adapter ist auf beiden Pfaden an m gebunden. Er schreibt m, und er liest ausschließlich m: ein gespeicherter Frame mit einem anderen Manifest löst einen MigrationError aus, selbst wenn dieses andere Manifest in derselben Registry registriert ist.

const registry = new InMemorySchemaRegistry();
registry.register('Account.Deposited', 1, { codec: zodCodec(DepositedV1) });
registry.register('Account.Closed', 1, { codec: zodCodec(ClosedV1) });

const adapter = registry.eventAdapter<DepositedV1>('Account.Deposited');

// Reads its own manifest, at any registered version.
adapter.fromJournal({ manifest: 'Account.Deposited', version: 1, payload });

// Refused — MigrationError: manifest mismatch: schema-registry adapter
// is for 'Account.Deposited', got 'Account.Closed'
adapter.fromJournal({ manifest: 'Account.Closed', version: 1, payload });

Ohne diese Prüfung würde der zweite Read sauber durch den Codec von Account.Closed dekodieren und einem statisch als DepositedV1 typisierten Aufrufer ein ClosedV1 zurückgeben — eine Typverwechslung, die der Aufrufer nicht erkennen kann, weil die Payload gültig ist und der statische Typ behauptet, er habe das Erwartete bekommen. Auf einem Recovery-Pfad heißt das: ein fremdes Event landet in onEvent und erzeugt einen State, den es nie gab. MigrationChain.upcast und defaultsAdapter haben das immer schon abgewiesen; der Registry-Adapter stimmt jetzt mit ihnen überein.

Der realistische Auslöser ist eine Fehlverdrahtung, kein Angreifer: vertauschte Manifest-Strings von eventAdapter(...) und snapshotAdapter(...), oder ein zwischen zwei Deploys geändertes Argument, während alte Rows noch den alten Wert tragen. Was die Registry selbst geschrieben hat, kann die Prüfung nie auslösen, denn toJournal taggt immer mit dem eigenen Manifest des Adapters.

Die Konsequenz, die du einplanen solltest: ein Actor, dessen Event-Union mehrere Manifeste umfasst, kann keinen einzelnen von der Registry gebauten Adapter verwenden. Wirklich unterstützt war das nie — toJournal taggt jedes Event mit dem einen Manifest und validiert es gegen den Codec dieses Manifests, ein solcher Actor konnte seine anderen Event-Typen also ohnehin nicht persistieren — aber der Lesepfad ist früher stillschweigend fehlgeschlagen und schlägt jetzt laut fehl. Schreib ein fromJournal, das auf stored.manifest verzweigt und an einen Adapter pro Typ delegiert, so wie es MigrationChain für ein Journal mit mehreren Typen erwartet.

Kompatibilitäts-Checks

Die Registry kann zur Register-Zeit prüfen, dass eine neue Version die vorherige noch lesen kann — und so einen fehlenden oder kaputten Upcaster erwischen, bevor du deployst:

registry.register('Deposited', 2, {
  codec:          zodCodec(DepositedV2),
  upcastFromPrev: (v1: DepositedV1): DepositedV2 => ({ ...v1, currency: 'USD' }),
  compatibility:  'backward',
});

• 'none' (Default) — kein Check.
• 'backward' — verlangt, dass die vorherige Version registriert ist und dass diese Registrierung ein upcastFromPrev liefert. Das strukturelle Minimum, das den Lesepfad für alte Daten funktionsfähig hält.
• 'sample' — alles, was 'backward' prüft, plus ein Round-Trip eines gelieferten sample: mit dem vorherigen Codec dekodieren, upcasten, mit dem neuen Codec neu enkodieren. Erwischt latente Upcaster-Bugs zur Register-Zeit statt zur Deploy-Zeit.

registry.register('Deposited', 2, {
  codec:          zodCodec(DepositedV2),
  upcastFromPrev: (v1: DepositedV1): DepositedV2 => ({ ...v1, currency: 'USD' }),
  compatibility:  'sample',
  sample:         { kind: 'deposited', amount: 100 },
});
// → wirft zur Register-Zeit, wenn der v1 → v2 Round-Trip fehlschlägt

Eine inkompatible Registrierung wirft sofort, daher ist es dein Sicherheitsnetz, die Registry beim Startup zu verdrahten: eine Version hochziehen, aber den Upcaster vergessen, und die Registrierung schlägt laut fehl.

Ein binäres Wire-Format pro Version

Ein codec muss kein Validator sein. serializerCodec verpackt jeden Serializer — auch die mitgelieferten AvroSerializer und ProtobufSerializer —, sodass jede Version ein anderes Wire-Format nutzen kann:

import { serializerCodec } from 'actor-ts/persistence';

registry.register('BankAccount.Deposited', 1, {
  codec: serializerCodec(avroSerializer),
});

registry.register('BankAccount.Deposited', 2, {
  codec:          serializerCodec(protobufSerializer),
  upcastFromPrev: (v1: DepositedV1): DepositedV2 => ({ ...v1, currency: 'USD' }),
  compatibility:  'backward',
});

Neue Events werden als v2 in Protobuf geschrieben; die Avro-Zeilen, die schon auf der Platte liegen, dekodieren weiter über den v1-Codec und werden beim Lesen nach vorne gezogen. Beide Formate koexistieren in einem Stream.

Das ist die einzige Stelle, an der das geht. Das withSerializer(...) eines Stores gilt storeweit — ein Format für jede Payload, die er schreibt —, ein Wechsel pro Version muss also auf der Codec-Ebene passieren.

Was auf der Platte landet. serializerCodec kodiert zu { serializerId, manifest, bytes } und hört da auf: der Payload-Codec des Journals trägt ein Uint8Array bereits als getaggtes JSON, die Bytes werden also genau einmal base64-kodiert. Die serializerId reist mit der Zeile mit, und genau das macht eine Verwechslung laut —

serializer:avro: payload was written by serializer id 101
(manifest '.bank.DepositedV2'), but this codec holds 'avro' (id 100)

— statt still zu Unsinn zu dekodieren, was bei Avro (keine Feld-Tags) und Protobuf (kein Nachrichtenname) der Normalfall ist, wenn man ihnen fremde Bytes gibt.

In Kombination mit compatibility: 'sample' läuft der Register-Aufruf den ganzen Weg Avro → Upcast → Protobuf beim Startup durch: ein v2-Schema, das dein Upcaster gar nicht erfüllen kann, scheitert beim Deploy statt beim ersten Replay.

Tooling

Ein paar häufige Tooling-Formen, die von der Registry profitieren:

// Admin-Panel: jedes registrierte Schema und seine Version auflisten
const all = registry.list();   // → [{ manifest: 'Deposited', version: 2, codec, ... }, ...]

// Dev-Skript: warnen, wenn ein Event im Journal eine Version hat,
// die HÖHER ist als das, was die Registry kennt (ein Deploy-Issue)
for await (const event of query.eventsByPersistenceId(...)) {
  const envelopeVersion = event.event._v;
  const latest = registry.latestVersion(event.event._t);
  if (latest !== undefined && envelopeVersion > latest) {
    console.warn(`event ${event.event._t} has version ${envelopeVersion}, registry knows up to ${latest}`);
  }
}

Wann du sie nicht brauchst

Kleine Projekte

// 3 Event-Typen insgesamt — Registry fügt Komplexität ohne Wert hinzu

Für kleine Projekte lebt die Versions-Konstante neben dem Adapter, und das ist in Ordnung. Adoptiere die Registry, wenn du dich dabei ertappst, dass du über Dateien hinweg suchst, um dich zu erinnern “auf welcher Version ist Deposited?”

Keine Remote-Schema-Registry

// Jede Version trägt einen Codec — die Validierung lebt IN der Registry

Das ist keine Confluent- / Apache-Avro-HTTP-Schema-Registry: Es gibt keine Remote-Subject-Versionierung und keine Kompatibilitäts-Aushandlung über die Leitung — das ist außerhalb des Scopes für v1. Aber das strukturelle Schema ist hier — jedes (manifest, version) trägt einen codec (bring dein eigenes zod / valibot mit), erzwungen auf Lesen und Schreiben, und die Registry verkettet Upcaster, um alte Daten zu migrieren. Validierung ist eingebaut, nicht separat auf Adapter-Ebene angeflanscht.

Mehrere Umgebungen

// Prod-Registry unterscheidet sich von Dev — welche ist kanonisch?

Behandle die Registry als per Deployment — eine Konstante in deinem Code. Wenn verschiedene Umgebungen verschiedene Versionen haben (während eines Staged Rollouts), spiegelt sich das im deployten Code wider, nicht in verschiedenen Laufzeit-Registries.

Wie geht’s weiter

• Migration im Überblick — das größere Bild.
• defaultsAdapter — die typisierten Adapter, die die Registry konsumieren.
• migratingAdapter — die verkettete Alternative.
• Envelope-Format — die On-Disk-Form mit _v / _t / _e.
• Eigene Serializer — die Avro- und Protobuf-Serializer, die serializerCodec verpackt.

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