Add SyncEvent service (#25158)

2026-04-30 16:45:26 -04:00
parent 3250b814ce
commit fbcbd24063
11 changed files with 430 additions and 350 deletions
--- a/packages/opencode/src/sync/index.ts
+++ b/packages/opencode/src/sync/index.ts
@@ -9,9 +9,11 @@ import { EventSequenceTable, EventTable } from "./event.sql"
 import { WorkspaceContext } from "@/control-plane/workspace-context"
 import { EventID } from "./schema"
 import { Flag } from "@opencode-ai/core/flag/flag"
-import { Schema as EffectSchema } from "effect"
+import { Context, Effect, Layer, Schema as EffectSchema } from "effect"
 import { zodObject } from "@/util/effect-zod"
 import type { DeepMutable } from "@/util/schema"
+import { makeRuntime } from "@/effect/run-service"
+import { serviceUse } from "@/effect/service-use"

 // Keep `Event["data"]` mutable because projectors mutate the persisted shape
 // when writing to the database. Bus payloads (`Properties`) stay readonly —
@@ -46,6 +48,125 @@ export type SerializedEvent<Def extends Definition = Definition> = Event<Def> &
 type ProjectorFunc = (db: Database.TxOrDb, data: unknown) => void
 type ConvertEvent = (type: string, data: Event["data"]) => unknown | Promise<unknown>

+export interface Interface {
+  readonly run: <Def extends Definition>(
+    def: Def,
+    data: Event<Def>["data"],
+    options?: { publish?: boolean },
+  ) => Effect.Effect<void>
+  readonly replay: (event: SerializedEvent, options?: { publish: boolean }) => Effect.Effect<void>
+  readonly replayAll: (events: SerializedEvent[], options?: { publish: boolean }) => Effect.Effect<string | undefined>
+  readonly remove: (aggregateID: string) => Effect.Effect<void>
+}
+
+export class Service extends Context.Service<Service, Interface>()("@opencode/SyncEvent") {}
+
+export const layer = Layer.effect(Service)(
+  Effect.gen(function* () {
+    const replay: Interface["replay"] = Effect.fn("SyncEvent.replay")(function* (event, options) {
+      const def = registry.get(event.type)
+      if (!def) {
+        throw new Error(`Unknown event type: ${event.type}`)
+      }
+
+      const row = Database.use((db) =>
+        db
+          .select({ seq: EventSequenceTable.seq })
+          .from(EventSequenceTable)
+          .where(eq(EventSequenceTable.aggregate_id, event.aggregateID))
+          .get(),
+      )
+
+      const latest = row?.seq ?? -1
+      if (event.seq <= latest) return
+
+      const expected = latest + 1
+      if (event.seq !== expected) {
+        throw new Error(
+          `Sequence mismatch for aggregate "${event.aggregateID}": expected ${expected}, got ${event.seq}`,
+        )
+      }
+
+      process(def, event, { publish: !!options?.publish })
+    })
+
+    const replayAll: Interface["replayAll"] = Effect.fn("SyncEvent.replayAll")(function* (events, options) {
+      const source = events[0]?.aggregateID
+      if (!source) return undefined
+      if (events.some((item) => item.aggregateID !== source)) {
+        throw new Error("Replay events must belong to the same session")
+      }
+      const start = events[0].seq
+      for (const [i, item] of events.entries()) {
+        const seq = start + i
+        if (item.seq !== seq) {
+          throw new Error(`Replay sequence mismatch at index ${i}: expected ${seq}, got ${item.seq}`)
+        }
+      }
+      for (const item of events) {
+        yield* replay(item, options)
+      }
+      return source
+    })
+
+    const run: Interface["run"] = Effect.fn("SyncEvent.run")(function* (def, data, options) {
+      const agg = (data as Record<string, string>)[def.aggregate]
+      // This should never happen: we've enforced it via typescript in
+      // the definition
+      if (agg == null) {
+        throw new Error(`SyncEvent.run: "${def.aggregate}" required but not found: ${JSON.stringify(data)}`)
+      }
+
+      if (def.version !== versions.get(def.type)) {
+        throw new Error(`SyncEvent.run: running old versions of events is not allowed: ${def.type}`)
+      }
+
+      const { publish = true } = options || {}
+
+      // Note that this is an "immediate" transaction which is critical.
+      // We need to make sure we can safely read and write with nothing
+      // else changing the data from under us
+      Database.transaction(
+        (tx) => {
+          const id = EventID.ascending()
+          const row = tx
+            .select({ seq: EventSequenceTable.seq })
+            .from(EventSequenceTable)
+            .where(eq(EventSequenceTable.aggregate_id, agg))
+            .get()
+          const seq = row?.seq != null ? row.seq + 1 : 0
+
+          const event = { id, seq, aggregateID: agg, data }
+          process(def, event, { publish })
+        },
+        {
+          behavior: "immediate",
+        },
+      )
+    })
+
+    const remove: Interface["remove"] = Effect.fn("SyncEvent.remove")(function* (aggregateID) {
+      Database.transaction((tx) => {
+        tx.delete(EventSequenceTable).where(eq(EventSequenceTable.aggregate_id, aggregateID)).run()
+        tx.delete(EventTable).where(eq(EventTable.aggregate_id, aggregateID)).run()
+      })
+    })
+
+    return Service.of({
+      run,
+      replay,
+      replayAll,
+      remove,
+    })
+  }),
+)
+
+export const defaultLayer = layer
+
+export const use = serviceUse(Service)
+
+const runtime = makeRuntime(Service, defaultLayer)
+
 export const registry = new Map<string, Definition>()
 let projectors: Map<Definition, ProjectorFunc> | undefined
 const versions = new Map<string, number>()
@@ -186,92 +307,19 @@ function process<Def extends Definition>(def: Def, event: Event<Def>, options: {
 }

 export function replay(event: SerializedEvent, options?: { publish: boolean }) {
-  const def = registry.get(event.type)
-  if (!def) {
-    throw new Error(`Unknown event type: ${event.type}`)
-  }
-
-  const row = Database.use((db) =>
-    db
-      .select({ seq: EventSequenceTable.seq })
-      .from(EventSequenceTable)
-      .where(eq(EventSequenceTable.aggregate_id, event.aggregateID))
-      .get(),
-  )
-
-  const latest = row?.seq ?? -1
-  if (event.seq <= latest) {
-    return
-  }
-
-  const expected = latest + 1
-  if (event.seq !== expected) {
-    throw new Error(`Sequence mismatch for aggregate "${event.aggregateID}": expected ${expected}, got ${event.seq}`)
-  }
-
-  process(def, event, { publish: !!options?.publish })
+  return runtime.runSync((sync) => sync.replay(event, options))
 }

 export function replayAll(events: SerializedEvent[], options?: { publish: boolean }) {
-  const source = events[0]?.aggregateID
-  if (!source) return
-  if (events.some((item) => item.aggregateID !== source)) {
-    throw new Error("Replay events must belong to the same session")
-  }
-  const start = events[0].seq
-  for (const [i, item] of events.entries()) {
-    const seq = start + i
-    if (item.seq !== seq) {
-      throw new Error(`Replay sequence mismatch at index ${i}: expected ${seq}, got ${item.seq}`)
-    }
-  }
-  for (const item of events) {
-    replay(item, options)
-  }
-  return source
+  return runtime.runSync((sync) => sync.replayAll(events, options))
 }

 export function run<Def extends Definition>(def: Def, data: Event<Def>["data"], options?: { publish?: boolean }) {
-  const agg = (data as Record<string, string>)[def.aggregate]
-  // This should never happen: we've enforced it via typescript in
-  // the definition
-  if (agg == null) {
-    throw new Error(`SyncEvent.run: "${def.aggregate}" required but not found: ${JSON.stringify(data)}`)
-  }
-
-  if (def.version !== versions.get(def.type)) {
-    throw new Error(`SyncEvent.run: running old versions of events is not allowed: ${def.type}`)
-  }
-
-  const { publish = true } = options || {}
-
-  // Note that this is an "immediate" transaction which is critical.
-  // We need to make sure we can safely read and write with nothing
-  // else changing the data from under us
-  Database.transaction(
-    (tx) => {
-      const id = EventID.ascending()
-      const row = tx
-        .select({ seq: EventSequenceTable.seq })
-        .from(EventSequenceTable)
-        .where(eq(EventSequenceTable.aggregate_id, agg))
-        .get()
-      const seq = row?.seq != null ? row.seq + 1 : 0
-
-      const event = { id, seq, aggregateID: agg, data }
-      process(def, event, { publish })
-    },
-    {
-      behavior: "immediate",
-    },
-  )
+  return runtime.runSync((sync) => sync.run(def, data, options))
 }

 export function remove(aggregateID: string) {
-  Database.transaction((tx) => {
-    tx.delete(EventSequenceTable).where(eq(EventSequenceTable.aggregate_id, aggregateID)).run()
-    tx.delete(EventTable).where(eq(EventTable.aggregate_id, aggregateID)).run()
-  })
+  return runtime.runSync((sync) => sync.remove(aggregateID))
 }

 export function payloads() {