diff options
Diffstat (limited to 'packages/opencode/src/control-plane/workspace.ts')
| -rw-r--r-- | packages/opencode/src/control-plane/workspace.ts | 258 |
1 files changed, 237 insertions, 21 deletions
diff --git a/packages/opencode/src/control-plane/workspace.ts b/packages/opencode/src/control-plane/workspace.ts index f330e07b7..78f3d770e 100644 --- a/packages/opencode/src/control-plane/workspace.ts +++ b/packages/opencode/src/control-plane/workspace.ts @@ -1,11 +1,13 @@ import z from "zod" import { setTimeout as sleep } from "node:timers/promises" import { fn } from "@/util/fn" -import { Database, eq } from "@/storage/db" +import { Database, asc, eq } from "@/storage/db" import { Project } from "@/project/project" import { BusEvent } from "@/bus/bus-event" import { GlobalBus } from "@/bus/global" import { SyncEvent } from "@/sync" +import { EventTable } from "@/sync/event.sql" +import { Flag } from "@/flag/flag" import { Log } from "@/util/log" import { Filesystem } from "@/util/filesystem" import { ProjectID } from "@/project/schema" @@ -15,6 +17,11 @@ import { getAdaptor } from "./adaptors" import { WorkspaceInfo } from "./types" import { WorkspaceID } from "./schema" import { parseSSE } from "./sse" +import { Session } from "@/session" +import { SessionTable } from "@/session/session.sql" +import { SessionID } from "@/session/schema" +import { errorData } from "@/util/error" +import { AppRuntime } from "@/effect/app-runtime" export namespace Workspace { export const Info = WorkspaceInfo.meta({ @@ -29,6 +36,13 @@ export namespace Workspace { }) export type ConnectionStatus = z.infer<typeof ConnectionStatus> + const Restore = z.object({ + workspaceID: WorkspaceID.zod, + sessionID: SessionID.zod, + total: z.number().int().min(0), + step: z.number().int().min(0), + }) + export const Event = { Ready: BusEvent.define( "workspace.ready", @@ -42,6 +56,7 @@ export namespace Workspace { message: z.string(), }), ), + Restore: BusEvent.define("workspace.restore", Restore), Status: BusEvent.define("workspace.status", ConnectionStatus), } @@ -102,11 +117,170 @@ export namespace Workspace { return info }) + const SessionRestoreInput = z.object({ + workspaceID: WorkspaceID.zod, + sessionID: SessionID.zod, + }) + + export const sessionRestore = fn(SessionRestoreInput, async (input) => { + log.info("session restore requested", { + workspaceID: input.workspaceID, + sessionID: input.sessionID, + }) + try { + const space = await get(input.workspaceID) + if (!space) throw new Error(`Workspace not found: ${input.workspaceID}`) + + const adaptor = await getAdaptor(space.projectID, space.type) + const target = await adaptor.target(space) + + // Need to switch the workspace of the session + SyncEvent.run(Session.Event.Updated, { + sessionID: input.sessionID, + info: { + workspaceID: input.workspaceID, + }, + }) + + const rows = Database.use((db) => + db + .select({ + id: EventTable.id, + aggregateID: EventTable.aggregate_id, + seq: EventTable.seq, + type: EventTable.type, + data: EventTable.data, + }) + .from(EventTable) + .where(eq(EventTable.aggregate_id, input.sessionID)) + .orderBy(asc(EventTable.seq)) + .all(), + ) + if (rows.length === 0) throw new Error(`No events found for session: ${input.sessionID}`) + + const all = rows + + const size = 10 + const sets = Array.from({ length: Math.ceil(all.length / size) }, (_, i) => all.slice(i * size, (i + 1) * size)) + const total = sets.length + log.info("session restore prepared", { + workspaceID: input.workspaceID, + sessionID: input.sessionID, + workspaceType: space.type, + directory: space.directory, + target: target.type === "remote" ? String(route(target.url, "/sync/replay")) : target.directory, + events: all.length, + batches: total, + first: all[0]?.seq, + last: all.at(-1)?.seq, + }) + GlobalBus.emit("event", { + directory: "global", + workspace: input.workspaceID, + payload: { + type: Event.Restore.type, + properties: { + workspaceID: input.workspaceID, + sessionID: input.sessionID, + total, + step: 0, + }, + }, + }) + for (const [i, events] of sets.entries()) { + log.info("session restore batch starting", { + workspaceID: input.workspaceID, + sessionID: input.sessionID, + step: i + 1, + total, + events: events.length, + first: events[0]?.seq, + last: events.at(-1)?.seq, + target: target.type === "remote" ? String(route(target.url, "/sync/replay")) : target.directory, + }) + if (target.type === "local") { + SyncEvent.replayAll(events) + log.info("session restore batch replayed locally", { + workspaceID: input.workspaceID, + sessionID: input.sessionID, + step: i + 1, + total, + events: events.length, + }) + } else { + const url = route(target.url, "/sync/replay") + const headers = new Headers(target.headers) + headers.set("content-type", "application/json") + const res = await fetch(url, { + method: "POST", + headers, + body: JSON.stringify({ + directory: space.directory ?? "", + events, + }), + }) + if (!res.ok) { + const body = await res.text() + log.error("session restore batch failed", { + workspaceID: input.workspaceID, + sessionID: input.sessionID, + step: i + 1, + total, + status: res.status, + body, + }) + throw new Error( + `Failed to replay session ${input.sessionID} into workspace ${input.workspaceID}: HTTP ${res.status} ${body}`, + ) + } + log.info("session restore batch posted", { + workspaceID: input.workspaceID, + sessionID: input.sessionID, + step: i + 1, + total, + status: res.status, + }) + } + GlobalBus.emit("event", { + directory: "global", + workspace: input.workspaceID, + payload: { + type: Event.Restore.type, + properties: { + workspaceID: input.workspaceID, + sessionID: input.sessionID, + total, + step: i + 1, + }, + }, + }) + } + + log.info("session restore complete", { + workspaceID: input.workspaceID, + sessionID: input.sessionID, + batches: total, + }) + + return { + total, + } + } catch (err) { + log.error("session restore failed", { + workspaceID: input.workspaceID, + sessionID: input.sessionID, + error: errorData(err), + }) + throw err + } + }) + export function list(project: Project.Info) { const rows = Database.use((db) => db.select().from(WorkspaceTable).where(eq(WorkspaceTable.project_id, project.id)).all(), ) const spaces = rows.map(fromRow).sort((a, b) => a.id.localeCompare(b.id)) + for (const space of spaces) startSync(space) return spaces } @@ -120,13 +294,25 @@ export namespace Workspace { }) export const remove = fn(WorkspaceID.zod, async (id) => { + const sessions = Database.use((db) => + db.select({ id: SessionTable.id }).from(SessionTable).where(eq(SessionTable.workspace_id, id)).all(), + ) + for (const session of sessions) { + await AppRuntime.runPromise(Session.Service.use((svc) => svc.remove(session.id))) + } + const row = Database.use((db) => db.select().from(WorkspaceTable).where(eq(WorkspaceTable.id, id)).get()) + if (row) { stopSync(id) const info = fromRow(row) - const adaptor = await getAdaptor(info.projectID, row.type) - adaptor.remove(info) + try { + const adaptor = await getAdaptor(info.projectID, row.type) + await adaptor.remove(info) + } catch (err) { + log.error("adaptor not available when removing workspace", { type: row.type }) + } Database.use((db) => db.delete(WorkspaceTable).where(eq(WorkspaceTable.id, id)).run()) return info } @@ -156,51 +342,81 @@ export namespace Workspace { const log = Log.create({ service: "workspace-sync" }) - async function workspaceEventLoop(space: Info, signal: AbortSignal) { - log.info("starting sync: " + space.id) + function route(url: string | URL, path: string) { + const next = new URL(url) + next.pathname = `${next.pathname.replace(/\/$/, "")}${path}` + next.search = "" + next.hash = "" + return next + } + async function syncWorkspace(space: Info, signal: AbortSignal) { while (!signal.aborted) { - log.info("connecting to sync: " + space.id) + log.info("connecting to global sync", { workspace: space.name }) - setStatus(space.id, "connecting") const adaptor = await getAdaptor(space.projectID, space.type) const target = await adaptor.target(space) if (target.type === "local") return - const res = await fetch(target.url + "/sync/event", { method: "GET", signal }).catch((err: unknown) => { - setStatus(space.id, "error", String(err)) + const res = await fetch(route(target.url, "/global/event"), { + method: "GET", + headers: target.headers, + signal, + }).catch((err: unknown) => { + setStatus(space.id, "error") + + log.info("failed to connect to global sync", { + workspace: space.name, + error: err, + }) return undefined }) - if (!res || !res.ok || !res.body) { - log.info("failed to connect to sync: " + res?.status) - setStatus(space.id, "error", res ? `HTTP ${res.status}` : "no response") + if (!res || !res.ok || !res.body) { + log.info("failed to connect to global sync", { workspace: space.name }) + setStatus(space.id, "error") await sleep(1000) continue } + + log.info("global sync connected", { workspace: space.name }) setStatus(space.id, "connected") - await parseSSE(res.body, signal, (evt) => { - const event = evt as SyncEvent.SerializedEvent + await parseSSE(res.body, signal, (evt: any) => { try { - if (!event.type.startsWith("server.")) { - SyncEvent.replay(event) + if (!("payload" in evt)) return + + if (evt.payload.type === "sync") { + // This name -> type is temporary + SyncEvent.replay({ ...evt.payload, type: evt.payload.name } as SyncEvent.SerializedEvent) } + + GlobalBus.emit("event", { + directory: evt.directory, + project: evt.project, + workspace: space.id, + payload: evt.payload, + }) } catch (err) { - log.warn("failed to replay sync event", { + log.info("failed to replay global event", { workspaceID: space.id, error: err, }) } }) + + log.info("disconnected from global sync: " + space.id) setStatus(space.id, "disconnected") - log.info("disconnected to sync: " + space.id) - await sleep(250) + + // TODO: Implement exponential backoff + await sleep(1000) } } function startSync(space: Info) { + if (!Flag.OPENCODE_EXPERIMENTAL_WORKSPACES) return + if (space.type === "worktree") { void Filesystem.exists(space.directory!).then((exists) => { setStatus(space.id, exists ? "connected" : "error", exists ? undefined : "directory does not exist") @@ -213,9 +429,9 @@ export namespace Workspace { aborts.set(space.id, abort) setStatus(space.id, "disconnected") - void workspaceEventLoop(space, abort.signal).catch((error) => { + void syncWorkspace(space, abort.signal).catch((error) => { setStatus(space.id, "error", String(error)) - log.warn("workspace sync listener failed", { + log.warn("workspace listener failed", { workspaceID: space.id, error, }) |