From 06bd34c953a0183f5d5dafabc5875af66352b6e5 Mon Sep 17 00:00:00 2001 From: Dax Raad Date: Mon, 25 May 2026 00:28:30 -0400 Subject: [PATCH] feat(core): add sync event replay controls --- packages/core/src/event.ts | 111 ++++++++++++++++++++++++++++--- packages/core/test/event.test.ts | 102 ++++++++++++++++++++++++++++ 2 files changed, 204 insertions(+), 9 deletions(-) diff --git a/packages/core/src/event.ts b/packages/core/src/event.ts index f8ddad439..2ff7313d4 100644 --- a/packages/core/src/event.ts +++ b/packages/core/src/event.ts @@ -37,6 +37,14 @@ export type Payload = { export type Projector = (event: Payload) => Effect.Effect type AnyProjector = (event: Payload) => Effect.Effect +export type SerializedEvent = { + readonly id: ID + readonly type: string + readonly seq: number + readonly aggregateID: string + readonly data: Record +} + export class InvalidSyncEventError extends Schema.TaggedErrorClass()( "EventV2.InvalidSyncEvent", { @@ -50,6 +58,7 @@ export function versionedType(type: string, version: number) { } export const registry = new Map() +const syncRegistry = new Map }>() export function define(input: { readonly type: Type @@ -78,6 +87,7 @@ export function define= existing.sync.version) { registry.set(input.type, definition) } + if (input.sync) syncRegistry.set(versionedType(input.type, input.sync.version), definition as Definition & { readonly sync: NonNullable }) return definition as Schema.Schema>>> & Definition> } @@ -100,6 +110,13 @@ export interface Interface { readonly subscribe: (definition: D) => Stream.Stream> readonly all: () => Stream.Stream readonly project: (definition: D, projector: Projector) => Effect.Effect + readonly replay: (event: SerializedEvent, options?: { readonly publish?: boolean; readonly ownerID?: string }) => Effect.Effect + readonly replayAll: ( + events: SerializedEvent[], + options?: { readonly publish?: boolean; readonly ownerID?: string }, + ) => Effect.Effect + readonly remove: (aggregateID: string) => Effect.Effect + readonly claim: (aggregateID: string, ownerID: string) => Effect.Effect } export class Service extends Context.Service()("@opencode/Event") {} @@ -128,7 +145,7 @@ export const layer = Layer.effect( }), ) - function runProjectors(event: Payload) { + function commitSyncEvent(event: Payload, input?: { readonly seq: number; readonly aggregateID: string; readonly ownerID?: string }) { return Effect.gen(function* () { const definition = registry.get(event.type) const sync = definition?.sync @@ -155,19 +172,30 @@ export const layer = Layer.effect( .transaction( () => Effect.gen(function* () { - for (const projector of list) { - yield* projector(event as Payload) - } const row = yield* db - .select({ seq: EventSequenceTable.seq }) + .select({ seq: EventSequenceTable.seq, ownerID: EventSequenceTable.owner_id }) .from(EventSequenceTable) .where(eq(EventSequenceTable.aggregate_id, aggregateID)) .get() .pipe(Effect.orDie) - const seq = row?.seq != null ? row.seq + 1 : 0 + const latest = row?.seq ?? -1 + if (input && input.seq <= latest) return + if (input && row?.ownerID && row.ownerID !== input.ownerID) return + const seq = input?.seq ?? latest + 1 + if (input && seq !== latest + 1) { + yield* Effect.die( + new InvalidSyncEventError({ + type: event.type, + message: `Sequence mismatch for aggregate ${aggregateID}: expected ${latest + 1}, got ${seq}`, + }), + ) + } + for (const projector of list) { + yield* projector(event as Payload) + } yield* db .insert(EventSequenceTable) - .values([{ aggregate_id: aggregateID, seq }]) + .values([{ aggregate_id: aggregateID, seq, owner_id: input?.ownerID }]) .onConflictDoUpdate({ target: EventSequenceTable.aggregate_id, set: { seq }, @@ -207,7 +235,7 @@ export const layer = Layer.effect( ...(location ? { location } : {}), data, } as Payload - yield* runProjectors(event) + yield* commitSyncEvent(event as Payload) const pubsub = typed.get(event.type) if (pubsub) yield* PubSub.publish(pubsub, event as Payload) yield* PubSub.publish(all, event as Payload) @@ -215,6 +243,71 @@ export const layer = Layer.effect( }) } + function replay(event: SerializedEvent, options?: { readonly publish?: boolean; readonly ownerID?: string }) { + return Effect.gen(function* () { + const definition = syncRegistry.get(event.type) + if (!definition) { + yield* Effect.die( + new InvalidSyncEventError({ type: event.type, message: `Unknown sync event type ${event.type}` }), + ) + } else { + const payload = { + id: event.id, + type: definition.type, + version: definition.sync.version, + data: event.data, + } as Payload + yield* commitSyncEvent(payload, { seq: event.seq, aggregateID: event.aggregateID, ownerID: options?.ownerID }) + if (options?.publish) { + const pubsub = typed.get(payload.type) + if (pubsub) yield* PubSub.publish(pubsub, payload) + yield* PubSub.publish(all, payload) + } + } + }) + } + + function replayAll(events: SerializedEvent[], options?: { readonly publish?: boolean; readonly ownerID?: string }) { + return Effect.gen(function* () { + const source = events[0]?.aggregateID + if (!source) return undefined + if (events.some((event) => event.aggregateID !== source)) { + yield* Effect.die(new InvalidSyncEventError({ type: events[0]?.type ?? "unknown", message: "Replay events must belong to the same aggregate" })) + } + const start = events[0]?.seq ?? 0 + for (const [index, event] of events.entries()) { + const seq = start + index + if (event.seq !== seq) { + yield* Effect.die(new InvalidSyncEventError({ type: event.type, message: `Replay sequence mismatch at index ${index}: expected ${seq}, got ${event.seq}` })) + } + } + for (const event of events) { + yield* replay(event, options) + } + return source + }) + } + + function remove(aggregateID: string) { + return db + .transaction(() => + Effect.gen(function* () { + yield* db.delete(EventSequenceTable).where(eq(EventSequenceTable.aggregate_id, aggregateID)).run() + yield* db.delete(EventTable).where(eq(EventTable.aggregate_id, aggregateID)).run() + }), + ) + .pipe(Effect.orDie) + } + + function claim(aggregateID: string, ownerID: string) { + return db + .update(EventSequenceTable) + .set({ owner_id: ownerID }) + .where(eq(EventSequenceTable.aggregate_id, aggregateID)) + .run() + .pipe(Effect.orDie) + } + const subscribe = (definition: D): Stream.Stream> => Stream.unwrap(getOrCreate(definition).pipe(Effect.map((pubsub) => Stream.fromPubSub(pubsub)))).pipe( Stream.map((event) => event as Payload), @@ -229,7 +322,7 @@ export const layer = Layer.effect( projectors.set(definition.type, list) }) - return Service.of({ publish, subscribe, all: streamAll, project }) + return Service.of({ publish, subscribe, all: streamAll, project, replay, replayAll, remove, claim }) }), ) diff --git a/packages/core/test/event.test.ts b/packages/core/test/event.test.ts index 24b0df0a2..4d6236740 100644 --- a/packages/core/test/event.test.ts +++ b/packages/core/test/event.test.ts @@ -163,4 +163,106 @@ describe("EventV2", () => { expect(received).toEqual([SyncMessage.type, "stream"]) }), ) + + it.effect("replays sync events through projectors", () => + Effect.gen(function* () { + const events = yield* EventV2.Service + const received = new Array() + yield* events.project(SyncMessage, (event) => + Effect.sync(() => { + received.push(event) + }), + ) + const aggregateID = EventV2.ID.create() + + yield* events.replay({ + id: EventV2.ID.create(), + type: EventV2.versionedType(SyncMessage.type, 1), + seq: 0, + aggregateID, + data: { id: aggregateID, text: "hello" }, + }) + + expect(received[0]?.type).toBe(SyncMessage.type) + expect(received[0]?.data).toEqual({ id: aggregateID, text: "hello" }) + }), + ) + + it.effect("replayAll validates contiguous aggregate events", () => + Effect.gen(function* () { + const events = yield* EventV2.Service + const aggregateID = EventV2.ID.create() + const source = yield* events.replayAll([ + { + id: EventV2.ID.create(), + type: EventV2.versionedType(SyncMessage.type, 1), + seq: 0, + aggregateID, + data: { id: aggregateID, text: "one" }, + }, + { + id: EventV2.ID.create(), + type: EventV2.versionedType(SyncMessage.type, 1), + seq: 1, + aggregateID, + data: { id: aggregateID, text: "two" }, + }, + ]) + + expect(source).toBe(aggregateID) + }), + ) + + it.effect("claim fences replay owners", () => + Effect.gen(function* () { + const events = yield* EventV2.Service + const received = new Array() + const aggregateID = EventV2.ID.create() + yield* events.publish(SyncMessage, { id: aggregateID, text: "seed" }) + yield* events.claim(aggregateID, "owner-a") + yield* events.project(SyncMessage, (event) => + Effect.sync(() => { + received.push(event) + }), + ) + + yield* events.replay( + { + id: EventV2.ID.create(), + type: EventV2.versionedType(SyncMessage.type, 1), + seq: 1, + aggregateID, + data: { id: aggregateID, text: "ignored" }, + }, + { ownerID: "owner-b" }, + ) + + expect(received).toHaveLength(0) + }), + ) + + it.effect("remove clears sync event sequence", () => + Effect.gen(function* () { + const events = yield* EventV2.Service + const received = new Array() + const aggregateID = EventV2.ID.create() + yield* events.publish(SyncMessage, { id: aggregateID, text: "seed" }) + yield* events.remove(aggregateID) + yield* events.project(SyncMessage, (event) => + Effect.sync(() => { + received.push(event) + }), + ) + + yield* events.replay({ + id: EventV2.ID.create(), + type: EventV2.versionedType(SyncMessage.type, 1), + seq: 0, + aggregateID, + data: { id: aggregateID, text: "replayed" }, + }) + + expect(received[0]?.data).toEqual({ id: aggregateID, text: "replayed" }) + }), + ) })