fix(sync): publish events on injected project bus (#27825)

This commit is contained in:
Dax 2026-05-16 01:56:49 -04:00 committed by GitHub
parent e33912bfee
commit 53849bd866
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
2 changed files with 25 additions and 6 deletions

View file

@ -17,6 +17,7 @@ import { EventV2 } from "@opencode-ai/core/event"
import { serviceUse } from "@/effect/service-use"
import { InstanceState } from "@/effect/instance-state"
import { RuntimeFlags } from "@/effect/runtime-flags"
import { attachWith } from "@/effect/run-service"
// Keep `Event["data"]` mutable because projectors mutate the persisted shape
// when writing to the database. Bus payloads (`Properties`) stay readonly —
@ -75,6 +76,7 @@ export class Service extends Context.Service<Service, Interface>()("@opencode/Sy
export const layer = Layer.effect(Service)(
Effect.gen(function* () {
const flags = yield* RuntimeFlags.Service
const bus = yield* ProjectBus.Service
const replay: Interface["replay"] = Effect.fn("SyncEvent.replay")(function* (event, options) {
const def = registry.get(event.type)
@ -112,6 +114,7 @@ export const layer = Layer.effect(Service)(
}
: undefined
process(def, event, {
bus,
publish,
context,
ownerID: options?.ownerID,
@ -172,7 +175,7 @@ export const layer = Layer.effect(Service)(
const seq = row?.seq != null ? row.seq + 1 : 0
const event = { id, seq, aggregateID: agg, data }
process(def, event, { publish, context, experimentalWorkspaces: flags.experimentalWorkspaces })
process(def, event, { bus, publish, context, experimentalWorkspaces: flags.experimentalWorkspaces })
},
{
behavior: "immediate",
@ -209,7 +212,7 @@ export const layer = Layer.effect(Service)(
}),
)
export const defaultLayer = layer.pipe(Layer.provide(RuntimeFlags.defaultLayer))
export const defaultLayer = layer.pipe(Layer.provide([ProjectBus.defaultLayer, RuntimeFlags.defaultLayer]))
export const use = serviceUse(Service)
@ -303,7 +306,13 @@ function register(def: Definition) {
function process<Def extends Definition>(
def: Def,
event: Event<Def>,
options: { publish: boolean; context?: PublishContext; ownerID?: string; experimentalWorkspaces: boolean },
options: {
bus: ProjectBus.Interface
publish: boolean
context?: PublishContext
ownerID?: string
experimentalWorkspaces: boolean
},
) {
if (projectors == null) {
throw new Error("No projectors available. Call `SyncEvent.init` to install projectors")
@ -348,7 +357,13 @@ function process<Def extends Definition>(
}
const result = convertEvent(def.type, event.data)
const publish = (data: unknown) => ProjectBus.publish(def, data as Properties<Def>, { id: event.id })
const publish = (data: unknown) =>
Effect.runPromise(
attachWith(options.bus.publish(def, data as Properties<Def>, { id: event.id }), {
instance: options.context?.instance,
workspace: options.context?.workspace,
}),
)
if (result instanceof Promise) {
void result.then(publish)
} else {

View file

@ -13,7 +13,10 @@ import { RuntimeFlags } from "@/effect/runtime-flags"
const it = testEffect(
Layer.mergeAll(
SyncEvent.layer.pipe(Layer.provide(RuntimeFlags.layer({ experimentalWorkspaces: true }))),
SyncEvent.layer.pipe(
Layer.provide(RuntimeFlags.layer({ experimentalWorkspaces: true })),
Layer.provideMerge(Bus.layer),
),
CrossSpawnSpawner.defaultLayer,
),
)
@ -114,7 +117,8 @@ describe("SyncEvent", () => {
const received = new Promise<void>((done) => {
resolve = done
})
const dispose = Bus.subscribeAll((event) => {
const bus = yield* Bus.Service
const dispose = yield* bus.subscribeAllCallback((event) => {
events.push(event)
resolve()
})