refactor(core): remove bus replay all (#42460)

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<void>
-  readonly replayAll: (
-    events: SerializedEvent[],
-    options?: { readonly publish?: boolean; readonly ownerID?: string; readonly strictOwner?: boolean },
-  ) => Effect.Effect<string | undefined>
   readonly remove: (aggregateID: string) => Effect.Effect<void>
   readonly claim: (aggregateID: string, ownerID: string) => Effect.Effect<void>
 }
@@ -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,
         })

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

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" },

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({

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(),