From 23b763274313243e70589972b3d7eca080bad629 Mon Sep 17 00:00:00 2001 From: Kit Langton Date: Thu, 13 Aug 2026 21:52:48 -0400 Subject: [PATCH] refactor(core): remove bus replay all (#42460) --- packages/core/src/bus.ts | 27 -------- packages/core/test/bus.test.ts | 82 ----------------------- packages/core/test/session-create.test.ts | 4 +- packages/core/test/session-prompt.test.ts | 4 +- packages/core/test/session-runner.test.ts | 8 ++- 5 files changed, 11 insertions(+), 114 deletions(-) diff --git a/packages/core/src/bus.ts b/packages/core/src/bus.ts index 91846b2a427..84830889c60 100644 --- a/packages/core/src/bus.ts +++ b/packages/core/src/bus.ts @@ -158,10 +158,6 @@ export interface Interface { event: SerializedEvent, options?: { readonly publish?: boolean; readonly ownerID?: string; readonly strictOwner?: boolean }, ) => Effect.Effect - readonly replayAll: ( - events: SerializedEvent[], - options?: { readonly publish?: boolean; readonly ownerID?: string; readonly strictOwner?: boolean }, - ) => Effect.Effect readonly remove: (aggregateID: string) => Effect.Effect readonly claim: (aggregateID: string, ownerID: string) => Effect.Effect } @@ -648,28 +644,6 @@ export function configured(options?: Options) { }) } - function replayAll( - events: SerializedEvent[], - options?: { readonly publish?: boolean; readonly ownerID?: string; readonly strictOwner?: boolean }, - ) { - return Effect.gen(function* () { - const source = events[0]?.aggregateID - if (!source) return undefined - if (events.some((event) => event.aggregateID !== source)) { - yield* Effect.die( - new InvalidDurableEventError({ - type: events[0]?.type ?? "unknown", - message: "Replay events must belong to the same aggregate", - }), - ) - } - for (const event of events) { - yield* replay(event, options) - } - return source - }) - } - function remove(aggregateID: string) { return db .transaction(() => @@ -866,7 +840,6 @@ export function configured(options?: Options) { listen, project, replay, - replayAll, remove, claim, }) diff --git a/packages/core/test/bus.test.ts b/packages/core/test/bus.test.ts index 7c679d184ce..88021b714c6 100644 --- a/packages/core/test/bus.test.ts +++ b/packages/core/test/bus.test.ts @@ -879,88 +879,6 @@ describe("Bus", () => { }), ) - it.effect("replayAll validates contiguous aggregate events", () => - Effect.gen(function* () { - const bus = yield* Bus.Service - const aggregateID = Session.ID.create() - const source = yield* bus.replayAll([ - { - id: Event.ID.create(), - created: DateTime.makeUnsafe(0), - type: Bus.versionedType(DurableMessage.type, 1), - seq: 0, - aggregateID, - data: durableData(aggregateID, "one"), - }, - { - id: Event.ID.create(), - created: DateTime.makeUnsafe(0), - type: Bus.versionedType(DurableMessage.type, 1), - seq: 1, - aggregateID, - data: durableData(aggregateID, "two"), - }, - ]) - - expect(source).toBe(aggregateID) - }), - ) - - it.effect("replayAll accepts later chunks after the first batch", () => - Effect.gen(function* () { - const bus = yield* Bus.Service - const { db } = yield* Database.Service - const aggregateID = Session.ID.create() - - const one = yield* bus.replayAll([ - { - id: Event.ID.create(), - created: DateTime.makeUnsafe(0), - type: Bus.versionedType(DurableMessage.type, 1), - seq: 0, - aggregateID, - data: durableData(aggregateID, "one"), - }, - { - id: Event.ID.create(), - created: DateTime.makeUnsafe(0), - type: Bus.versionedType(DurableMessage.type, 1), - seq: 1, - aggregateID, - data: durableData(aggregateID, "two"), - }, - ]) - const two = yield* bus.replayAll([ - { - id: Event.ID.create(), - created: DateTime.makeUnsafe(0), - type: Bus.versionedType(DurableMessage.type, 1), - seq: 2, - aggregateID, - data: durableData(aggregateID, "three"), - }, - { - id: Event.ID.create(), - created: DateTime.makeUnsafe(0), - type: Bus.versionedType(DurableMessage.type, 1), - seq: 3, - aggregateID, - data: durableData(aggregateID, "four"), - }, - ]) - const rows = yield* db - .select() - .from(EventTable) - .where(eq(EventTable.aggregate_id, aggregateID)) - .all() - .pipe(Effect.orDie) - - expect(one).toBe(aggregateID) - expect(two).toBe(aggregateID) - expect(rows.map((row) => row.seq)).toEqual([0, 1, 2, 3]) - }), - ) - it.effect("claim fences replay owners", () => Effect.gen(function* () { const bus = yield* Bus.Service diff --git a/packages/core/test/session-create.test.ts b/packages/core/test/session-create.test.ts index 8d6bb0d81b7..66a11508309 100644 --- a/packages/core/test/session-create.test.ts +++ b/packages/core/test/session-create.test.ts @@ -629,7 +629,7 @@ describe("Session.create", () => { .pipe(Effect.orDie) expect(yield* store.get(created.id)).toBeUndefined() - expect(yield* bus.replayAll(serialized.slice(0, 2))).toBe(created.id) + yield* Effect.forEach(serialized.slice(0, 2), (event) => bus.replay(event), { discard: true }) expect(yield* SessionInbox.find(db, admitted.id)).toMatchObject({ id: admitted.id, sessionID: created.id, @@ -639,7 +639,7 @@ describe("Session.create", () => { }) expect(yield* store.context(created.id)).toEqual([]) - expect(yield* bus.replayAll(serialized.slice(2))).toBe(created.id) + yield* Effect.forEach(serialized.slice(2), (event) => bus.replay(event), { discard: true }) expect(yield* SessionInbox.find(db, admitted.id)).toBeUndefined() expect(yield* store.context(created.id)).toMatchObject([ { id: admitted.id, type: "user", text: "Replay lifecycle" }, diff --git a/packages/core/test/session-prompt.test.ts b/packages/core/test/session-prompt.test.ts index d5a71a20c4a..aa60be29a0a 100644 --- a/packages/core/test/session-prompt.test.ts +++ b/packages/core/test/session-prompt.test.ts @@ -767,7 +767,7 @@ describe("Session.prompt", () => { .where(eq(SessionMessageTable.session_id, sessionID)) .run() .pipe(Effect.orDie) - yield* bus.replayAll( + yield* Effect.forEach( recorded.map((event) => ({ id: event.id, created: DateTime.makeUnsafe(event.created), @@ -776,6 +776,8 @@ describe("Session.prompt", () => { type: event.type, data: event.data, })), + (event) => bus.replay(event), + { discard: true }, ) expect(yield* admitted(messageID)).toMatchObject({ diff --git a/packages/core/test/session-runner.test.ts b/packages/core/test/session-runner.test.ts index f7531621e56..e1cf377af0e 100644 --- a/packages/core/test/session-runner.test.ts +++ b/packages/core/test/session-runner.test.ts @@ -635,7 +635,7 @@ const replaySessionProjection = (id: Session.ID) => yield* db.delete(InstructionStateTable).where(eq(InstructionStateTable.session_id, id)).run().pipe(Effect.orDie) yield* db.delete(SessionInboxTable).where(eq(SessionInboxTable.session_id, id)).run().pipe(Effect.orDie) yield* db.delete(SessionMessageTable).where(eq(SessionMessageTable.session_id, id)).run().pipe(Effect.orDie) - yield* bus.replayAll( + yield* Effect.forEach( recorded.map((event) => ({ id: event.id, created: DateTime.makeUnsafe(event.created), @@ -644,6 +644,8 @@ const replaySessionProjection = (id: Session.ID) => type: event.type, data: event.data, })), + (event) => bus.replay(event), + { discard: true }, ) }) @@ -1343,7 +1345,7 @@ describe("SessionRunnerLLM", () => { .all() yield* bus.remove(forked.id) yield* db.delete(SessionTable).where(eq(SessionTable.id, forked.id)).run() - yield* bus.replayAll( + yield* Effect.forEach( recorded.map((event) => ({ id: event.id, created: DateTime.makeUnsafe(event.created), @@ -1352,6 +1354,8 @@ describe("SessionRunnerLLM", () => { type: event.type, data: event.data, })), + (event) => bus.replay(event), + { discard: true }, ) expect( yield* db.select().from(InstructionStateTable).where(eq(InstructionStateTable.session_id, forked.id)).get(),