feat(core): initial implementation of syncing (#17814)

packages/opencode/src/server/projectors.ts

@@ -0,0 +1,28 @@
+import z from "zod"
+import sessionProjectors from "../session/projectors"
+import { SyncEvent } from "@/sync"
+import { Session } from "@/session"
+import { SessionTable } from "@/session/session.sql"
+import { Database, eq } from "@/storage/db"
+
+export function initProjectors() {
+  SyncEvent.init({
+    projectors: sessionProjectors,
+    convertEvent: (type, data) => {
+      if (type === "session.updated") {
+        const id = (data as z.infer<typeof Session.Event.Updated.schema>).sessionID
+        const row = Database.use((db) => db.select().from(SessionTable).where(eq(SessionTable.id, id)).get())
+
+        if (!row) return data
+
+        return {
+          sessionID: id,
+          info: Session.fromRow(row),
+        }
+      }
+      return data
+    },
+  })
+}
+
+initProjectors()

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

@@ -6,7 +6,6 @@ import { BusEvent } from "@/bus/bus-event"
 import { Bus } from "@/bus"
 import { lazy } from "../../util/lazy"
 import { AsyncQueue } from "../../util/queue"
-import { Instance } from "@/project/instance"
 
 const log = Log.create({ service: "server" })
 
@@ -53,13 +52,6 @@ export const EventRoutes = lazy(() =>
           )
         }, 10_000)
 
-        const unsub = Bus.subscribeAll((event) => {
-          q.push(JSON.stringify(event))
-          if (event.type === Bus.InstanceDisposed.type) {
-            stop()
-          }
-        })
-
         const stop = () => {
           if (done) return
           done = true
@@ -69,6 +61,13 @@ export const EventRoutes = lazy(() =>
           log.info("event disconnected")
         }
 
+        const unsub = Bus.subscribeAll((event) => {
+          q.push(JSON.stringify(event))
+          if (event.type === Bus.InstanceDisposed.type) {
+            stop()
+          }
+        })
+
         stream.onAbort(stop)
 
         try {

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

@@ -1,9 +1,9 @@
-import { Hono } from "hono"
-import { describeRoute, validator, resolver } from "hono-openapi"
+import { Hono, type Context } from "hono"
+import { describeRoute, resolver, validator } from "hono-openapi"
 import { streamSSE } from "hono/streaming"
 import z from "zod"
-import { Bus } from "../../bus"
 import { BusEvent } from "@/bus/bus-event"
+import { SyncEvent } from "@/sync"
 import { GlobalBus } from "@/bus/global"
 import { AsyncQueue } from "@/util/queue"
 import { Instance } from "../../project/instance"
@@ -17,6 +17,56 @@ const log = Log.create({ service: "server" })
 
 export const GlobalDisposedEvent = BusEvent.define("global.disposed", z.object({}))
 
+async function streamEvents(c: Context, subscribe: (q: AsyncQueue<string | null>) => () => void) {
+  return streamSSE(c, async (stream) => {
+    const q = new AsyncQueue<string | null>()
+    let done = false
+
+    q.push(
+      JSON.stringify({
+        payload: {
+          type: "server.connected",
+          properties: {},
+        },
+      }),
+    )
+
+    // Send heartbeat every 10s to prevent stalled proxy streams.
+    const heartbeat = setInterval(() => {
+      q.push(
+        JSON.stringify({
+          payload: {
+            type: "server.heartbeat",
+            properties: {},
+          },
+        }),
+      )
+    }, 10_000)
+
+    const stop = () => {
+      if (done) return
+      done = true
+      clearInterval(heartbeat)
+      unsub()
+      q.push(null)
+      log.info("global event disconnected")
+    }
+
+    const unsub = subscribe(q)
+
+    stream.onAbort(stop)
+
+    try {
+      for await (const data of q) {
+        if (data === null) return
+        await stream.writeSSE({ data })
+      }
+    } finally {
+      stop()
+    }
+  })
+}
+
 export const GlobalRoutes = lazy(() =>
   new Hono()
     .get(
@@ -70,55 +120,58 @@ export const GlobalRoutes = lazy(() =>
         log.info("global event connected")
         c.header("X-Accel-Buffering", "no")
         c.header("X-Content-Type-Options", "nosniff")
-        return streamSSE(c, async (stream) => {
-          const q = new AsyncQueue<string | null>()
-          let done = false
-
-          q.push(
-            JSON.stringify({
-              payload: {
-                type: "server.connected",
-                properties: {},
-              },
-            }),
-          )
-
-          // Send heartbeat every 10s to prevent stalled proxy streams.
-          const heartbeat = setInterval(() => {
-            q.push(
-              JSON.stringify({
-                payload: {
-                  type: "server.heartbeat",
-                  properties: {},
-                },
-              }),
-            )
-          }, 10_000)
 
+        return streamEvents(c, (q) => {
           async function handler(event: any) {
             q.push(JSON.stringify(event))
           }
           GlobalBus.on("event", handler)
-
-          const stop = () => {
-            if (done) return
-            done = true
-            clearInterval(heartbeat)
-            GlobalBus.off("event", handler)
-            q.push(null)
-            log.info("event disconnected")
-          }
-
-          stream.onAbort(stop)
-
-          try {
-            for await (const data of q) {
-              if (data === null) return
-              await stream.writeSSE({ data })
-            }
-          } finally {
-            stop()
-          }
+          return () => GlobalBus.off("event", handler)
+        })
+      },
+    )
+    .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("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),
+                },
+              }),
+            )
+          })
         })
       },
     )

packages/opencode/src/server/routes/session.ts

@@ -281,14 +281,14 @@ export const SessionRoutes = lazy(() =>
         const sessionID = c.req.valid("param").sessionID
         const updates = c.req.valid("json")
 
-        let session = await Session.get(sessionID)
         if (updates.title !== undefined) {
-          session = await Session.setTitle({ sessionID, title: updates.title })
+          await Session.setTitle({ sessionID, title: updates.title })
         }
         if (updates.time?.archived !== undefined) {
-          session = await Session.setArchived({ sessionID, time: updates.time.archived })
+          await Session.setArchived({ sessionID, time: updates.time.archived })
         }
 
+        const session = await Session.get(sessionID)
         return c.json(session)
       },
     )

packages/opencode/src/server/server.ts

@@ -44,6 +44,7 @@ import { PermissionRoutes } from "./routes/permission"
 import { GlobalRoutes } from "./routes/global"
 import { MDNS } from "./mdns"
 import { lazy } from "@/util/lazy"
+import { initProjectors } from "./projectors"
 
 // @ts-ignore This global is needed to prevent ai-sdk from logging warnings to stdout https://github.com/vercel/ai/blob/2dc67e0ef538307f21368db32d5a12345d98831b/packages/ai/src/logger/log-warnings.ts#L85
 globalThis.AI_SDK_LOG_WARNINGS = false
@@ -51,6 +52,8 @@ globalThis.AI_SDK_LOG_WARNINGS = false
 const csp = (hash = "") =>
   `default-src 'self'; script-src 'self' 'wasm-unsafe-eval'${hash ? ` 'sha256-${hash}'` : ""}; style-src 'self' 'unsafe-inline'; img-src 'self' data: https:; font-src 'self' data:; media-src 'self' data:; connect-src 'self' data:`
 
+initProjectors()
+
 export namespace Server {
   const log = Log.create({ service: "server" })