refactor(core): publish sync events to global event stream (#22347)

Author: jlongster

packages/app/src/context/global-sdk.tsx

@@ -155,7 +155,12 @@ export const { use: useGlobalSDK, provider: GlobalSDKProvider } = createSimpleCo
               resetHeartbeat()
               streamErrorLogged = false
               const directory = event.directory ?? "global"
-              const payload = event.payload
+              if (event.payload.type === "sync") {
+                continue
+              }
+
+              const payload = event.payload as Event
+
               const k = key(directory, payload)
               if (k) {
                 const i = coalesced.get(k)

packages/opencode/src/bus/bus-event.ts

@@ -16,25 +16,18 @@ export namespace BusEvent {
   }
 
   export function payloads() {
-    return z
-      .discriminatedUnion(
-        "type",
-        registry
-          .entries()
-          .map(([type, def]) => {
-            return z
-              .object({
-                type: z.literal(type),
-                properties: def.properties,
-              })
-              .meta({
-                ref: "Event" + "." + def.type,
-              })
+    return registry
+      .entries()
+      .map(([type, def]) => {
+        return z
+          .object({
+            type: z.literal(type),
+            properties: def.properties,
+          })
+          .meta({
+            ref: "Event" + "." + def.type,
           })
-          .toArray() as any,
-      )
-      .meta({
-        ref: "Event",
       })
+      .toArray()
   }
 }

packages/opencode/src/cli/cmd/tui/context/event.ts

@@ -8,6 +8,10 @@ export function useEvent() {
 
   function subscribe(handler: (event: Event) => void) {
     return sdk.event.on("event", (event) => {
+      if (event.payload.type === "sync") {
+        return
+      }
+
       // Special hack for truly global events
       if (event.directory === "global") {
         handler(event.payload)

packages/opencode/src/project/instance.ts

@@ -21,8 +21,6 @@ const disposal = {
   all: undefined as Promise<void> | undefined,
 }
 
-function emitDisposed(directory: string) {}
-
 function boot(input: { directory: string; init?: () => Promise<any>; worktree?: string; project?: Project.Info }) {
   return iife(async () => {
     const ctx =

packages/opencode/src/server/instance/event.ts

@@ -1,8 +1,10 @@
+import z from "zod"
 import { Hono } from "hono"
 import { describeRoute, resolver } from "hono-openapi"
 import { streamSSE } from "hono/streaming"
 import { Log } from "@/util/log"
 import { BusEvent } from "@/bus/bus-event"
+import { SyncEvent } from "@/sync"
 import { Bus } from "@/bus"
 import { AsyncQueue } from "../../util/queue"
 
@@ -20,7 +22,11 @@ export const EventRoutes = () =>
           description: "Event stream",
           content: {
             "text/event-stream": {
-              schema: resolver(BusEvent.payloads()),
+              schema: resolver(
+                z.union(BusEvent.payloads()).meta({
+                  ref: "Event",
+                }),
+              ),
             },
           },
         },

packages/opencode/src/server/instance/global.ts

@@ -109,7 +109,7 @@ export const GlobalRoutes = lazy(() =>
                       directory: z.string(),
                       project: z.string().optional(),
                       workspace: z.string().optional(),
-                      payload: BusEvent.payloads(),
+                      payload: z.union([...BusEvent.payloads(), ...SyncEvent.payloads()]),
                     })
                     .meta({
                       ref: "GlobalEvent",
@@ -135,52 +135,6 @@ export const GlobalRoutes = lazy(() =>
         })
       },
     )
-    .get(
-      "/sync-event",
-      describeRoute({
-        summary: "Subscribe to global sync events",
-        description: "Get global sync events",
-        operationId: "global.sync-event.subscribe",
-        responses: {
-          200: {
-            description: "Event stream",
-            content: {
-              "text/event-stream": {
-                schema: resolver(
-                  z
-                    .object({
-                      payload: SyncEvent.payloads(),
-                    })
-                    .meta({
-                      ref: "SyncEvent",
-                    }),
-                ),
-              },
-            },
-          },
-        },
-      }),
-      async (c) => {
-        log.info("global sync event connected")
-        c.header("Cache-Control", "no-cache, no-transform")
-        c.header("X-Accel-Buffering", "no")
-        c.header("X-Content-Type-Options", "nosniff")
-        return streamEvents(c, (q) => {
-          return SyncEvent.subscribeAll(({ def, event }) => {
-            // TODO: don't pass def, just pass the type (and it should
-            // be versioned)
-            q.push(
-              JSON.stringify({
-                payload: {
-                  ...event,
-                  type: SyncEvent.versionedType(def.type, def.version),
-                },
-              }),
-            )
-          })
-        })
-      },
-    )
     .get(
       "/config",
       describeRoute({

packages/opencode/src/sync/index.ts

@@ -2,9 +2,12 @@ import z from "zod"
 import type { ZodObject } from "zod"
 import { EventEmitter } from "events"
 import { Database, eq } from "@/storage/db"
+import { GlobalBus } from "@/bus/global"
 import { Bus as ProjectBus } from "@/bus"
 import { BusEvent } from "@/bus/bus-event"
+import { Instance } from "@/project/instance"
 import { EventSequenceTable, EventTable } from "./event.sql"
+import { WorkspaceContext } from "@/control-plane/workspace-context"
 import { EventID } from "./schema"
 import { Flag } from "@/flag/flag"
 
@@ -37,8 +40,6 @@ export namespace SyncEvent {
   let frozen = false
   let convertEvent: (type: string, event: Event["data"]) => Promise<Record<string, unknown>> | Record<string, unknown>
 
-  const Bus = new EventEmitter<{ event: [{ def: Definition; event: Event }] }>()
-
   export function reset() {
     frozen = false
     projectors = undefined
@@ -140,11 +141,6 @@ export namespace SyncEvent {
       }
 
       Database.effect(() => {
-        Bus.emit("event", {
-          def,
-          event,
-        })
-
         if (options?.publish) {
           const result = convertEvent(def.type, event.data)
           if (result instanceof Promise) {
@@ -154,6 +150,17 @@ export namespace SyncEvent {
           } else {
             ProjectBus.publish({ type: def.type, properties: def.schema }, result)
           }
+
+          GlobalBus.emit("event", {
+            directory: Instance.directory,
+            project: Instance.project.id,
+            workspace: WorkspaceContext.workspaceID,
+            payload: {
+              type: "sync",
+              name: versionedType(def.type, def.version),
+              ...event,
+            },
+          })
         }
       })
     })
@@ -235,31 +242,23 @@ export namespace SyncEvent {
     })
   }
 
-  export function subscribeAll(handler: (event: { def: Definition; event: Event }) => void) {
-    Bus.on("event", handler)
-    return () => Bus.off("event", handler)
-  }
-
   export function payloads() {
-    return z
-      .union(
-        registry
-          .entries()
-          .map(([type, def]) => {
-            return z
-              .object({
-                type: z.literal(type),
-                aggregate: z.literal(def.aggregate),
-                data: def.schema,
-              })
-              .meta({
-                ref: "SyncEvent" + "." + def.type,
-              })
+    return registry
+      .entries()
+      .map(([type, def]) => {
+        return z
+          .object({
+            type: z.literal("sync"),
+            name: z.literal(type),
+            id: z.string(),
+            seq: z.number(),
+            aggregateID: z.literal(def.aggregate),
+            data: def.schema,
+          })
+          .meta({
+            ref: "SyncEvent" + "." + def.type,
           })
-          .toArray() as any,
-      )
-      .meta({
-        ref: "SyncEvent",
       })
+      .toArray()
   }
 }

packages/sdk/js/src/v2/gen/sdk.gen.ts

@@ -51,7 +51,6 @@ import type {
   GlobalDisposeResponses,
   GlobalEventResponses,
   GlobalHealthResponses,
-  GlobalSyncEventSubscribeResponses,
   GlobalUpgradeErrors,
   GlobalUpgradeResponses,
   InstanceDisposeResponses,
@@ -237,20 +236,6 @@ class HeyApiRegistry<T> {
   }
 }
 
-export class SyncEvent extends HeyApiClient {
-  /**
-   * Subscribe to global sync events
-   *
-   * Get global sync events
-   */
-  public subscribe<ThrowOnError extends boolean = false>(options?: Options<never, ThrowOnError>) {
-    return (options?.client ?? this.client).sse.get<GlobalSyncEventSubscribeResponses, unknown, ThrowOnError>({
-      url: "/global/sync-event",
-      ...options,
-    })
-  }
-}
-
 export class Config extends HeyApiClient {
   /**
    * Get global configuration
@@ -350,11 +335,6 @@ export class Global extends HeyApiClient {
     })
   }
 
-  private _syncEvent?: SyncEvent
-  get syncEvent(): SyncEvent {
-    return (this._syncEvent ??= new SyncEvent({ client: this.client }))
-  }
-
   private _config?: Config
   get config(): Config {
     return (this._config ??= new Config({ client: this.client }))