From b03ca0d4e223d5916644742bb4f00360145e7c3e Mon Sep 17 00:00:00 2001 From: Kit Langton Date: Wed, 5 Aug 2026 13:41:38 -0400 Subject: [PATCH] refactor(core): remove unused bus sequences (#40681) --- packages/core/src/bus.ts | 18 +----------------- packages/core/test/bus.test.ts | 20 -------------------- packages/core/test/session-log.test.ts | 4 +--- 3 files changed, 2 insertions(+), 40 deletions(-) diff --git a/packages/core/src/bus.ts b/packages/core/src/bus.ts index 2e498e0a397..92934de9610 100644 --- a/packages/core/src/bus.ts +++ b/packages/core/src/bus.ts @@ -3,7 +3,7 @@ export * as Bus from "./bus" import { Cause, Context, DateTime, Effect, Layer, Option, PubSub, Schema, Stream } from "effect" import { Event } from "@opencode-ai/schema/event" import type { EventLog } from "@opencode-ai/schema/event-log" -import { and, asc, eq, gt, inArray, lte, sql } from "drizzle-orm" +import { and, asc, eq, gt, lte, sql } from "drizzle-orm" import { Database } from "./database/database" import { EventSequenceTable, EventTable } from "./event/sql" import { Location } from "./location" @@ -134,8 +134,6 @@ export interface Interface { readonly after?: number readonly follow?: boolean }) => Stream.Stream - /** Latest committed seq per aggregate. Aggregates without events are absent. */ - readonly sequences: (aggregateIDs: ReadonlyArray) => Effect.Effect> /** @deprecated Use `subscribe()` and consume the returned stream. */ readonly listen: (listener: Subscriber) => Effect.Effect readonly project: (definition: D, projector: Subscriber) => Effect.Effect @@ -657,19 +655,6 @@ export const layerWith = (options?: LayerOptions) => }), ) - const sequences = (aggregateIDs: ReadonlyArray): Effect.Effect> => { - if (aggregateIDs.length === 0) return Effect.succeed(new Map()) - return db - .select({ aggregateID: EventSequenceTable.aggregate_id, seq: EventSequenceTable.seq }) - .from(EventSequenceTable) - .where(inArray(EventSequenceTable.aggregate_id, Array.from(aggregateIDs))) - .all() - .pipe( - Effect.orDie, - Effect.map((rows) => new Map(rows.map((row) => [row.aggregateID, Event.Seq.make(row.seq)]))), - ) - } - const listen = (listener: Subscriber): Effect.Effect => Effect.sync(() => { listeners.push(listener) @@ -691,7 +676,6 @@ export const layerWith = (options?: LayerOptions) => publish, subscribe, log, - sequences, listen, project, replay, diff --git a/packages/core/test/bus.test.ts b/packages/core/test/bus.test.ts index 447afae7e63..2b7d049ff60 100644 --- a/packages/core/test/bus.test.ts +++ b/packages/core/test/bus.test.ts @@ -1298,24 +1298,4 @@ describe("Bus", () => { }), ) - it.effect("sequences returns the latest committed seq per aggregate and omits unknown aggregates", () => - Effect.gen(function* () { - const bus = yield* Bus.Service - const first = Session.ID.create() - const second = Session.ID.create() - yield* bus.publish(DurableMessage, durableData(first, "zero")) - yield* bus.publish(DurableMessage, durableData(first, "one")) - yield* bus.publish(DurableMessage, durableData(second, "zero")) - - const sequences = yield* bus.sequences([first, second, Session.ID.create()]) - - expect(sequences).toEqual( - new Map([ - [first, Event.Seq.make(1)], - [second, Event.Seq.make(0)], - ]), - ) - expect(yield* bus.sequences([])).toEqual(new Map()) - }), - ) }) diff --git a/packages/core/test/session-log.test.ts b/packages/core/test/session-log.test.ts index fa3961f2127..c4428f85482 100644 --- a/packages/core/test/session-log.test.ts +++ b/packages/core/test/session-log.test.ts @@ -41,17 +41,15 @@ describe("Session.log", () => { it.effect("replays public session events and marks synced at the aggregate watermark", () => Effect.gen(function* () { const session = yield* Session.Service - const bus = yield* Bus.Service const created = yield* session.create({ location }) yield* session.rename({ sessionID: created.id, title: "session.renamed" }) const items = Array.from(yield* Stream.runCollect(session.log({ sessionID: created.id }))) - const watermark = (yield* bus.sequences([created.id])).get(created.id) // Session creation commits a non-public durable event, so the marker's // seq covers more of the aggregate than the public events emitted. expect(items.map((item) => item.type)).toEqual(["session.renamed", "log.synced"]) - expect(items.at(-1)).toEqual({ type: "log.synced", aggregateID: created.id, seq: watermark }) + expect(items.at(-1)).toEqual({ type: "log.synced", aggregateID: created.id, seq: Event.Seq.make(1) }) }), )