diff --git a/packages/core/src/database/database.ts b/packages/core/src/database/database.ts index 140063051..a12f4f07d 100644 --- a/packages/core/src/database/database.ts +++ b/packages/core/src/database/database.ts @@ -3,7 +3,6 @@ export * as Database from "./database" import { EffectDrizzleSqlite } from "@opencode-ai/effect-drizzle-sqlite" import { layer as sqliteLayer } from "#sqlite" import { Context, Effect, Layer } from "effect" -import { Sqlite } from "./sqlite" import { Global } from "../global" import { Flag } from "../flag/flag" import { isAbsolute, join } from "path" @@ -15,7 +14,6 @@ type DatabaseShape = Effect.Success export interface Interface { db: DatabaseShape - drizzle: Sqlite.DrizzleClient } export class Service extends Context.Service()("@opencode/v2/storage/Database") {} @@ -23,7 +21,6 @@ export class Service extends Context.Service()("@opencode/v2 export const layer = Layer.effect( Service, Effect.gen(function* () { - const drizzle = yield* Sqlite.Drizzle const db = yield* makeDatabase yield* db.run("PRAGMA journal_mode = WAL") @@ -35,7 +32,7 @@ export const layer = Layer.effect( yield* Effect.log("Applying database migrations") yield* DatabaseMigration.apply(db) - return { db, drizzle } + return { db } }).pipe(Effect.orDie), ) diff --git a/packages/opencode/src/cli/cmd/db.ts b/packages/opencode/src/cli/cmd/db.ts index b113455f3..9e7e37e18 100644 --- a/packages/opencode/src/cli/cmd/db.ts +++ b/packages/opencode/src/cli/cmd/db.ts @@ -1,17 +1,14 @@ import type { Argv } from "yargs" import { spawn } from "child_process" -import { Database } from "@/storage/db" -import { drizzle } from "drizzle-orm/bun-sqlite" -import { Database as BunDatabase } from "bun:sqlite" -import { UI } from "../ui" -import { cmd } from "./cmd" -import { JsonMigration } from "@/storage/json-migration" -import { EOL } from "os" -import { errorMessage } from "../../util/error" +import { Database } from "@opencode-ai/core/database/database" +import { Effect } from "effect" +import { sql } from "drizzle-orm" +import { effectCmd } from "../effect-cmd" -const QueryCommand = cmd({ +const QueryCommand = effectCmd({ command: "$0 [query]", describe: "open an interactive sqlite3 shell or run a query", + instance: false, builder: (yargs: Argv) => { return yargs .positional("query", { @@ -25,96 +22,41 @@ const QueryCommand = cmd({ describe: "Output format", }) }, - handler: async (args: { query?: string; format: string }) => { + handler: Effect.fn("Cli.db.query")(function* (args: { query?: string; format: string }) { const query = args.query as string | undefined if (query) { - const db = new BunDatabase(Database.getPath(), { readonly: true }) - try { - const result = db.query(query).all() as Record[] - if (args.format === "json") { - console.log(JSON.stringify(result, null, 2)) - } else if (result.length > 0) { - const keys = Object.keys(result[0]) - console.log(keys.join("\t")) - for (const row of result) { - console.log(keys.map((k) => row[k]).join("\t")) - } - } - } catch (err) { - UI.error(errorMessage(err)) - process.exit(1) + const { db } = yield* Database.Service + const result = yield* db.all>(sql.raw(query)).pipe(Effect.orDie) + if (args.format === "json") console.log(JSON.stringify(result, null, 2)) + else if (result.length > 0) { + const keys = Object.keys(result[0]) + console.log(keys.join("\t")) + for (const row of result) console.log(keys.map((key) => row[key]).join("\t")) } - db.close() return } - const child = spawn("sqlite3", [Database.getPath()], { + const child = spawn("sqlite3", [Database.path()], { stdio: "inherit", }) - await new Promise((resolve) => child.on("close", resolve)) - }, + yield* Effect.promise(() => new Promise((resolve) => child.on("close", resolve))) + }), }) -const PathCommand = cmd({ +const PathCommand = effectCmd({ command: "path", describe: "print the database path", - handler: () => { - console.log(Database.getPath()) - }, + instance: false, + handler: Effect.fn("Cli.db.path")(function* () { + console.log(Database.path()) + }), }) -const MigrateCommand = cmd({ - command: "migrate", - describe: "migrate JSON data to SQLite (merges with existing data)", - handler: async () => { - const sqlite = new BunDatabase(Database.getPath()) - const tty = process.stderr.isTTY - const width = 36 - const orange = "\x1b[38;5;214m" - const muted = "\x1b[0;2m" - const reset = "\x1b[0m" - let last = -1 - if (tty) process.stderr.write("\x1b[?25l") - try { - const stats = await JsonMigration.run(drizzle({ client: sqlite }), { - progress: (event) => { - const percent = Math.floor((event.current / event.total) * 100) - if (percent === last) return - last = percent - if (tty) { - const fill = Math.round((percent / 100) * width) - const bar = `${"■".repeat(fill)}${"・".repeat(width - fill)}` - process.stderr.write( - `\r${orange}${bar} ${percent.toString().padStart(3)}%${reset} ${muted}${event.current}/${event.total}${reset} `, - ) - } else { - process.stderr.write(`sqlite-migration:${percent}${EOL}`) - } - }, - }) - if (tty) process.stderr.write("\n") - if (tty) process.stderr.write("\x1b[?25h") - else process.stderr.write(`sqlite-migration:done${EOL}`) - UI.println( - `Migration complete: ${stats.projects} projects, ${stats.sessions} sessions, ${stats.messages} messages`, - ) - if (stats.errors.length > 0) { - UI.println(`${stats.errors.length} errors occurred during migration`) - } - } catch (err) { - if (tty) process.stderr.write("\x1b[?25h") - UI.error(`Migration failed: ${errorMessage(err)}`) - process.exit(1) - } finally { - sqlite.close() - } - }, -}) - -export const DbCommand = cmd({ +export const DbCommand = effectCmd({ command: "db", describe: "database tools", + instance: false, builder: (yargs: Argv) => { - return yargs.command(QueryCommand).command(PathCommand).command(MigrateCommand).demandCommand() + return yargs.command(QueryCommand).command(PathCommand).demandCommand() }, - handler: () => {}, + handler: Effect.fn("Cli.db")(function* () {}), }) diff --git a/packages/opencode/src/cli/cmd/import.ts b/packages/opencode/src/cli/cmd/import.ts index 44bd41119..7cad05bae 100644 --- a/packages/opencode/src/cli/cmd/import.ts +++ b/packages/opencode/src/cli/cmd/import.ts @@ -3,7 +3,7 @@ import { SessionLegacy } from "@opencode-ai/core/session/legacy" import { Session } from "@/session/session" import { MessageV2 } from "../../session/message-v2" import { CliError, effectCmd } from "../effect-cmd" -import { Database } from "@/storage/db" +import { Database } from "@opencode-ai/core/database/database" import { SessionTable, MessageTable, PartTable } from "@opencode-ai/core/session/sql" import { InstanceRef } from "@/effect/instance-ref" import { ShareNext } from "@/share/share-next" @@ -99,6 +99,7 @@ export const ImportCommand = effectCmd({ const runImport = Effect.fn("Cli.import.body")(function* (file: string, ctx: InstanceContext) { const share = yield* ShareNext.Service const fs = yield* AppFileSystem.Service + const { db } = yield* Database.Service let exportData: ExportData | undefined @@ -176,48 +177,45 @@ const runImport = Effect.fn("Cli.import.body")(function* (file: string, ctx: Ins path: path.relative(path.resolve(ctx.worktree), ctx.directory).replaceAll("\\", "/"), }) as Session.Info const row = Session.toRow(info) - Database.use((db) => - db - .insert(SessionTable) - .values(row) - .onConflictDoUpdate({ - target: SessionTable.id, - set: { project_id: row.project_id, directory: row.directory, path: row.path }, - }) - .run(), - ) + yield* db + .insert(SessionTable) + .values(row) + .onConflictDoUpdate({ + target: SessionTable.id, + set: { project_id: row.project_id, directory: row.directory, path: row.path }, + }) + .run() + .pipe(Effect.orDie) for (const msg of exportData.messages) { const msgInfo = decodeMessageInfo(msg.info) as SessionLegacy.Info const { id, sessionID: _, ...msgData } = msgInfo - Database.use((db) => - db - .insert(MessageTable) - .values({ - id, - session_id: row.id, - time_created: msgInfo.time?.created ?? Date.now(), - data: msgData as never, - }) - .onConflictDoNothing() - .run(), - ) + yield* db + .insert(MessageTable) + .values({ + id, + session_id: row.id, + time_created: msgInfo.time?.created ?? Date.now(), + data: msgData as never, + }) + .onConflictDoNothing() + .run() + .pipe(Effect.orDie) for (const part of msg.parts) { const partInfo = decodePart(part) as SessionLegacy.Part const { id: partId, sessionID: _s, messageID, ...partData } = partInfo - Database.use((db) => - db - .insert(PartTable) - .values({ - id: partId, - message_id: messageID, - session_id: row.id, - data: partData, - }) - .onConflictDoNothing() - .run(), - ) + yield* db + .insert(PartTable) + .values({ + id: partId, + message_id: messageID, + session_id: row.id, + data: partData, + }) + .onConflictDoNothing() + .run() + .pipe(Effect.orDie) } } diff --git a/packages/opencode/src/cli/cmd/stats.ts b/packages/opencode/src/cli/cmd/stats.ts index 3a937b4fb..22dee1477 100644 --- a/packages/opencode/src/cli/cmd/stats.ts +++ b/packages/opencode/src/cli/cmd/stats.ts @@ -2,7 +2,7 @@ import { Effect } from "effect" import { effectCmd } from "../effect-cmd" import { Session } from "@/session/session" import { NotFoundError } from "@/storage/storage" -import { Database } from "@/storage/db" +import { Database } from "@opencode-ai/core/database/database" import { SessionTable } from "@opencode-ai/core/session/sql" import { Project } from "@/project/project" import { InstanceRef } from "@/effect/instance-ref" @@ -80,9 +80,10 @@ export const StatsCommand = effectCmd({ }), }) -const getAllSessions = Effect.sync(() => - Database.use((db) => db.select().from(SessionTable).all()).map((row) => Session.fromRow(row)), -) +const getAllSessions = Effect.fnUntraced(function* () { + const { db } = yield* Database.Service + return (yield* db.select().from(SessionTable).all().pipe(Effect.orDie)).map((row) => Session.fromRow(row)) +}) const aggregateSessionStats = Effect.fn("Cli.stats.aggregate")(function* ( days?: number, @@ -90,7 +91,7 @@ const aggregateSessionStats = Effect.fn("Cli.stats.aggregate")(function* ( currentProject?: Project.Info, ) { const svc = yield* Session.Service - const sessions = yield* getAllSessions + const sessions = yield* getAllSessions() const MS_IN_DAY = 24 * 60 * 60 * 1000 const cutoffTime = (() => { diff --git a/packages/opencode/src/cli/cmd/tui/context/editor-zed.ts b/packages/opencode/src/cli/cmd/tui/context/editor-zed.ts index 0b8b58fc7..e17d520a0 100644 --- a/packages/opencode/src/cli/cmd/tui/context/editor-zed.ts +++ b/packages/opencode/src/cli/cmd/tui/context/editor-zed.ts @@ -1,7 +1,10 @@ -import { Database } from "bun:sqlite" import os from "node:os" import path from "node:path" -import { Option, Schema } from "effect" +import { Effect, Option, Schema } from "effect" +import { sql } from "drizzle-orm" +import { Sqlite } from "@opencode-ai/core/database/sqlite" +import { layer as sqliteLayer } from "@opencode-ai/core/database/sqlite.bun" +import { makeRuntime } from "@opencode-ai/core/effect/runtime" import { Filesystem } from "@/util/filesystem" import type { EditorSelection } from "./editor" @@ -88,12 +91,10 @@ export async function resolveZedSelection(dbPath: string, cwd = process.cwd()): } function queryZedActiveEditor(dbPath: string, cwd: string) { - let db: Database | undefined try { - db = new Database(dbPath, { readonly: true }) - const raw = db - .query( - `select + const raw = zedDatabase(dbPath).runSync((db) => + Effect.succeed( + db.all(sql.raw(`select i.kind as item_kind, e.item_id as editor_id, i.workspace_id as workspace_id, @@ -105,9 +106,9 @@ function queryZedActiveEditor(dbPath: string, cwd: string) { join workspaces w on w.workspace_id = i.workspace_id left join editors e on e.item_id = i.item_id and e.workspace_id = i.workspace_id where i.active = 1 and p.active = 1 - order by w.timestamp desc`, - ) - .all() + order by w.timestamp desc`)), + ), + ) const rows = raw.flatMap((row) => { const parsed = decodeZedEditorRow(row) @@ -126,24 +127,22 @@ function queryZedActiveEditor(dbPath: string, cwd: string) { return { type: "row" as const, row } } catch { return { type: "unavailable" as const } - } finally { - db?.close() } } function queryZedEditorSelections(dbPath: string, row: ZedActiveEditorRow) { - let db: Database | undefined try { - db = new Database(dbPath, { readonly: true }) - const raw = db - .query( - `select + const raw = zedDatabase(dbPath).runSync((db) => + Effect.succeed( + db.all>( + sql`select start as selection_start, end as selection_end from editor_selections - where editor_id = $editorID and workspace_id = $workspaceID`, - ) - .all({ $editorID: row.editor_id, $workspaceID: row.workspace_id }) + where editor_id = ${row.editor_id} and workspace_id = ${row.workspace_id}`, + ), + ), + ) const selections = raw.flatMap((selection) => { const parsed = decodeZedSelectionRow(selection) @@ -154,33 +153,33 @@ function queryZedEditorSelections(dbPath: string, row: ZedActiveEditorRow) { return { type: "selections" as const, selections } } catch { return { type: "unavailable" as const } - } finally { - db?.close() } } function queryZedEditorContents(dbPath: string, row: ZedActiveEditorRow) { - let db: Database | undefined try { - db = new Database(dbPath, { readonly: true }) const parsed = decodeZedEditorContents( - db - .query( - `select contents + zedDatabase(dbPath).runSync((db) => + Effect.succeed( + db.get( + sql`select contents from editors - where item_id = $editorID and workspace_id = $workspaceID`, - ) - .get({ $editorID: row.editor_id, $workspaceID: row.workspace_id }), + where item_id = ${row.editor_id} and workspace_id = ${row.workspace_id}`, + ), + ), + ), ) if (Option.isNone(parsed)) return { type: "unavailable" as const } return { type: "contents" as const, contents: parsed.value.contents } } catch { return { type: "unavailable" as const } - } finally { - db?.close() } } +function zedDatabase(dbPath: string) { + return makeRuntime(Sqlite.Drizzle, sqliteLayer({ filename: dbPath, readonly: true, create: false })) +} + function isZedActiveEditorRow(row: ZedEditorRow): row is ZedActiveEditorRow { return row.item_kind === "Editor" && row.editor_id != null } diff --git a/packages/opencode/src/control-plane/workspace.ts b/packages/opencode/src/control-plane/workspace.ts index c10e7b7b9..4139096d0 100644 --- a/packages/opencode/src/control-plane/workspace.ts +++ b/packages/opencode/src/control-plane/workspace.ts @@ -9,7 +9,8 @@ import { Project } from "@/project/project" import { BusEvent } from "@/bus/bus-event" import { GlobalBus } from "@/bus/global" import { Auth } from "@/auth" -import { SyncEvent } from "@/sync" +import { EventV2 } from "@opencode-ai/core/event" +import { EventV2Bridge } from "@/event-v2-bridge" import { EventSequenceTable, EventTable } from "@opencode-ai/core/event/sql" import { AppFileSystem } from "@opencode-ai/core/filesystem" import * as Log from "@opencode-ai/core/util/log" @@ -174,7 +175,7 @@ export const layer = Layer.effect( const session = yield* Session.Service const prompt = yield* SessionPrompt.Service const http = yield* HttpClient.HttpClient - const sync = yield* SyncEvent.Service + const events = yield* EventV2Bridge.Service const vcs = yield* Vcs.Service const flags = yield* RuntimeFlags.Service const fs = yield* AppFileSystem.Service @@ -372,20 +373,20 @@ export const layer = Layer.effect( }) } - const events = (yield* response.json) as HistoryEvent[] + const history = (yield* response.json) as HistoryEvent[] log.info("workspace history synced", { workspaceID: space.id, - events: events.length, + events: history.length, }) yield* Effect.forEach( - events, + history, (event) => - sync + events .replay( { - id: event.id, + id: EventV2.ID.make(event.id), aggregateID: event.aggregate_id, seq: event.seq, type: event.type, @@ -432,11 +433,11 @@ export const layer = Layer.effect( yield* parseSSE(stream, (evt) => Effect.gen(function* () { if (!evt || typeof evt !== "object" || !("payload" in evt)) return - const payload = evt.payload as { type?: string; syncEvent?: SyncEvent.SerializedEvent } + const payload = evt.payload as { type?: string; syncEvent?: EventV2.SerializedEvent } if (payload.type === "server.heartbeat") return if (payload.type === "sync" && payload.syncEvent) { - const failed = yield* sync.replay(payload.syncEvent).pipe( + const failed = yield* events.replay(payload.syncEvent, { publish: true }).pipe( Effect.as(false), Effect.catchCause((error) => Effect.sync(() => { @@ -634,7 +635,7 @@ export const layer = Layer.effect( // "claim" this session so any future events coming from // the old workspace are ignored - yield* sync.claim(input.sessionID, input.workspaceID ?? previous.projectID) + yield* events.claim(input.sessionID, input.workspaceID ?? previous.projectID) } } @@ -1002,12 +1003,12 @@ export const layer = Layer.effect( export const defaultLayer = layer.pipe( Layer.provide(Auth.defaultLayer), Layer.provide(Session.defaultLayer), - Layer.provide(SyncEvent.defaultLayer), Layer.provide(SessionPrompt.defaultLayer), Layer.provide(Project.defaultLayer), Layer.provide(Vcs.defaultLayer), Layer.provide(AppFileSystem.defaultLayer), Layer.provide(Database.defaultLayer), + Layer.provide(EventV2Bridge.defaultLayer), Layer.provide(FetchHttpClient.layer), Layer.provide(RuntimeFlags.defaultLayer), ) diff --git a/packages/opencode/src/data-migration.ts b/packages/opencode/src/data-migration.ts deleted file mode 100644 index ddd0f1337..000000000 --- a/packages/opencode/src/data-migration.ts +++ /dev/null @@ -1,161 +0,0 @@ -import { Context, Effect, Layer } from "effect" -import { Database } from "./storage/db" -import { DataMigrationTable } from "@opencode-ai/core/data-migration.sql" -import * as Log from "@opencode-ai/core/util/log" -import { and, asc, eq, gt, inArray, sql } from "drizzle-orm" -import { MessageTable, SessionTable } from "@opencode-ai/core/session/sql" -import type { SessionID } from "./session/schema" - -export type Migration = { - name: string - run: Effect.Effect -} - -const log = Log.create({ service: "data-migration" }) - -export interface Interface {} - -export class Service extends Context.Service()("@opencode/DataMigration") {} - -export const layer = Layer.effect( - Service, - Effect.gen(function* () { - const migrations: Migration[] = [ - { - name: "session_usage_from_messages", - run: Effect.gen(function* () { - type Usage = { - cost: number - tokens: { input: number; output: number; reasoning: number; cache: { read: number; write: number } } - } - - for (let cursor: SessionID | undefined, page = 1; ; page++) { - const next = yield* Effect.gen(function* () { - const sessions = yield* Effect.sync(() => - Database.use((db) => - db - .select({ id: SessionTable.id }) - .from(SessionTable) - .where(cursor ? gt(SessionTable.id, cursor) : undefined) - .orderBy(asc(SessionTable.id)) - .limit(100) - .all(), - ), - ) - if (sessions.length === 0) return - - yield* Effect.sync(() => - Database.transaction((db) => { - const usageBySession = new Map( - sessions.map((session) => [ - session.id, - { cost: 0, tokens: { input: 0, output: 0, reasoning: 0, cache: { read: 0, write: 0 } } }, - ]), - ) - - for (const row of db - .select({ - session_id: MessageTable.session_id, - cost: sql`coalesce(sum(coalesce(json_extract(${MessageTable.data}, '$.cost'), 0)), 0)`, - tokens_input: sql`coalesce(sum(coalesce(json_extract(${MessageTable.data}, '$.tokens.input'), 0)), 0)`, - tokens_output: sql`coalesce(sum(coalesce(json_extract(${MessageTable.data}, '$.tokens.output'), 0)), 0)`, - tokens_reasoning: sql`coalesce(sum(coalesce(json_extract(${MessageTable.data}, '$.tokens.reasoning'), 0)), 0)`, - tokens_cache_read: sql`coalesce(sum(coalesce(json_extract(${MessageTable.data}, '$.tokens.cache.read'), 0)), 0)`, - tokens_cache_write: sql`coalesce(sum(coalesce(json_extract(${MessageTable.data}, '$.tokens.cache.write'), 0)), 0)`, - }) - .from(MessageTable) - .where( - and( - inArray( - MessageTable.session_id, - sessions.map((session) => session.id), - ), - sql`json_extract(${MessageTable.data}, '$.role') = 'assistant'`, - ), - ) - .groupBy(MessageTable.session_id) - .all()) { - const current = usageBySession.get(row.session_id) - if (!current) continue - current.cost = row.cost - current.tokens.input = row.tokens_input - current.tokens.output = row.tokens_output - current.tokens.reasoning = row.tokens_reasoning - current.tokens.cache.read = row.tokens_cache_read - current.tokens.cache.write = row.tokens_cache_write - } - - for (const [sessionID, value] of usageBySession) { - db.update(SessionTable) - .set({ - cost: value.cost, - tokens_input: value.tokens.input, - tokens_output: value.tokens.output, - tokens_reasoning: value.tokens.reasoning, - tokens_cache_read: value.tokens.cache.read, - tokens_cache_write: value.tokens.cache.write, - time_updated: sql`${SessionTable.time_updated}`, - }) - .where(eq(SessionTable.id, sessionID)) - .run() - } - }), - ) - - return sessions.at(-1)?.id - }).pipe( - Effect.withSpan("DataMigration.sessionUsage.page", { - attributes: { - "data_migration.name": "session_usage_from_messages", - "data_migration.page": page, - "data_migration.cursor": cursor ?? "", - }, - }), - ) - if (!next) return - cursor = next - yield* Effect.sleep("10 millis") - } - }), - }, - ] - - yield* Effect.gen(function* () { - if (migrations.length === 0) return - - // Migrations run in a background fiber, so they must be resumable until - // their completion row is written. - for (const migration of migrations) { - const completed = Database.use((db) => - db - .select({ name: DataMigrationTable.name }) - .from(DataMigrationTable) - .where(eq(DataMigrationTable.name, migration.name)) - .get(), - ) - if (completed) continue - - log.info("running data migration", { name: migration.name }) - yield* migration.run.pipe(Effect.withSpan("DataMigration", { attributes: { name: migration.name } })) - Database.use((db) => - db - .insert(DataMigrationTable) - .values({ name: migration.name, time_completed: Date.now() }) - .onConflictDoNothing() - .run(), - ) - } - }).pipe( - Effect.tapCause((cause) => - Effect.logError("failed to run data migrations").pipe(Effect.annotateLogs("cause", cause)), - ), - Effect.ignore, - Effect.forkScoped, - ) - return Service.of({}) - }), -) - -export const defaultLayer = layer - -export * as DataMigration from "./data-migration" diff --git a/packages/opencode/src/effect/app-runtime.ts b/packages/opencode/src/effect/app-runtime.ts index 2bef35ed0..f02943c15 100644 --- a/packages/opencode/src/effect/app-runtime.ts +++ b/packages/opencode/src/effect/app-runtime.ts @@ -3,6 +3,7 @@ import { attach } from "./run-service" import * as Observability from "@opencode-ai/core/effect/observability" import { AppFileSystem } from "@opencode-ai/core/filesystem" +import { Database } from "@opencode-ai/core/database/database" import { Bus } from "@/bus" import { Auth } from "@/auth" import { Account } from "@/account/account" @@ -51,17 +52,15 @@ import { PtyTicket } from "@/pty/ticket" import { Installation } from "@/installation" import { ShareNext } from "@/share/share-next" import { SessionShare } from "@/share/session" -import { SyncEvent } from "@/sync" import { Npm } from "@opencode-ai/core/npm" import { memoMap } from "@opencode-ai/core/effect/memo-map" -import { DataMigration } from "@/data-migration" import { BackgroundJob } from "@/background/job" -import { EventV2Bridge } from "@/event-v2-bridge" import { RuntimeFlags } from "@/effect/runtime-flags" export const AppLayer = Layer.mergeAll( Npm.defaultLayer, AppFileSystem.defaultLayer, + Database.defaultLayer, Bus.defaultLayer, Auth.defaultLayer, Account.defaultLayer, @@ -111,9 +110,6 @@ export const AppLayer = Layer.mergeAll( Installation.defaultLayer, ShareNext.defaultLayer, SessionShare.defaultLayer, - SyncEvent.defaultLayer, - EventV2Bridge.defaultLayer, - DataMigration.defaultLayer, ).pipe(Layer.provideMerge(InstanceLayer.layer), Layer.provideMerge(Observability.layer)) const rt = ManagedRuntime.make(AppLayer, { memoMap }) diff --git a/packages/opencode/src/event-v2-bridge.ts b/packages/opencode/src/event-v2-bridge.ts index 3a29a6ebf..6e84ce82a 100644 --- a/packages/opencode/src/event-v2-bridge.ts +++ b/packages/opencode/src/event-v2-bridge.ts @@ -1,5 +1,5 @@ // Temporary V2 bridge: core events are the publish path, but the rest of -// opencode and the HTTP event stream still expect legacy bus/sync payloads. +// opencode and the HTTP event stream still expect legacy bus payloads. // This layer goes away once consumers subscribe to core EventV2 directly. import { Bus as ProjectBus } from "@/bus" import { GlobalBus } from "@/bus/global" @@ -31,8 +31,8 @@ export const layer = Layer.effect( }) }) - const provideEventLocation = (event: EventV2.Payload, effect: Effect.Effect) => { - return Effect.gen(function* () { + const provideEventLocation = (event: EventV2.Payload, effect: Effect.Effect) => + Effect.gen(function* () { const ctx = yield* InstanceRef if (ctx) return yield* effect const store = Option.getOrUndefined(yield* Effect.serviceOption(InstanceStore.Service)) @@ -45,7 +45,6 @@ export const layer = Layer.effect( }), ) }) - } yield* events.all().pipe( Stream.runForEach((event) => { @@ -58,6 +57,7 @@ export const layer = Layer.effect( }), Effect.forkScoped, ) + return Service.of(events) }), ) diff --git a/packages/opencode/src/index.ts b/packages/opencode/src/index.ts index 3498d0ef7..d8bcdff7b 100644 --- a/packages/opencode/src/index.ts +++ b/packages/opencode/src/index.ts @@ -30,10 +30,9 @@ import { WebCommand } from "./cli/cmd/web" import { PrCommand } from "./cli/cmd/pr" import { SessionCommand } from "./cli/cmd/session" import { DbCommand } from "./cli/cmd/db" -import path from "path" import { Global } from "@opencode-ai/core/global" import { JsonMigration } from "@/storage/json-migration" -import { Database } from "@/storage/db" +import { Database } from "@opencode-ai/core/database/database" import { errorMessage } from "./util/error" import { PluginCommand } from "./cli/cmd/plug" import { Heap } from "./cli/heap" @@ -116,7 +115,7 @@ const cli = yargs(args) run_id: processMetadata.runID, }) - const marker = Database.getPath() + const marker = Database.path() if (!(await Filesystem.exists(marker))) { const tty = process.stderr.isTTY process.stderr.write("Performing one time database migration, may take a few minutes..." + EOL) @@ -126,8 +125,9 @@ const cli = yargs(args) const reset = "\x1b[0m" let last = -1 if (tty) process.stderr.write("\x1b[?25l") + const sqlite = new (await import("bun:sqlite")).Database(marker) try { - await JsonMigration.run(drizzle({ client: Database.Client().$client }), { + await JsonMigration.run(drizzle({ client: sqlite }), { progress: (event) => { const percent = Math.floor((event.current / event.total) * 100) if (percent === last && event.current !== event.total) return @@ -145,6 +145,7 @@ const cli = yargs(args) }, }) } finally { + sqlite.close() if (tty) process.stderr.write("\x1b[?25h") else { process.stderr.write(`sqlite-migration:done${EOL}`) diff --git a/packages/opencode/src/node.ts b/packages/opencode/src/node.ts index 9c29dcd98..48b4293d1 100644 --- a/packages/opencode/src/node.ts +++ b/packages/opencode/src/node.ts @@ -2,5 +2,5 @@ export { Config } from "@/config/config" export { Server } from "./server/server" export { bootstrap } from "./cli/bootstrap" export * as Log from "@opencode-ai/core/util/log" -export { Database } from "@/storage/db" +export { Database } from "@opencode-ai/core/database/database" export { JsonMigration } from "@/storage/json-migration" diff --git a/packages/opencode/src/permission/index.ts b/packages/opencode/src/permission/index.ts index 7cc211f2f..ec1c24d4f 100644 --- a/packages/opencode/src/permission/index.ts +++ b/packages/opencode/src/permission/index.ts @@ -5,7 +5,7 @@ import { InstanceState } from "@/effect/instance-state" import { ProjectV2 } from "@opencode-ai/core/project" import { MessageID, SessionID } from "@/session/schema" import { PermissionTable } from "@opencode-ai/core/session/sql" -import { Database } from "@/storage/db" +import { Database } from "@opencode-ai/core/database/database" import { eq } from "drizzle-orm" import * as Log from "@opencode-ai/core/util/log" import { Wildcard } from "@opencode-ai/core/util/wildcard" @@ -145,11 +145,10 @@ export const layer = Layer.effect( Service, Effect.gen(function* () { const bus = yield* Bus.Service + const { db } = yield* Database.Service const state = yield* InstanceState.make( Effect.fn("Permission.state")(function* (ctx) { - const row = Database.use((db) => - db.select().from(PermissionTable).where(eq(PermissionTable.project_id, ctx.project.id)).get(), - ) + const row = yield* db.select().from(PermissionTable).where(eq(PermissionTable.project_id, ctx.project.id)).get().pipe(Effect.orDie) const state = { pending: new Map(), approved: [...(row?.data ?? [])], @@ -307,6 +306,6 @@ export function disabled(tools: string[], ruleset: Ruleset): Set { return PermissionV2.disabled(tools, ruleset) } -export const defaultLayer = layer.pipe(Layer.provide(Bus.layer)) +export const defaultLayer = layer.pipe(Layer.provide(Database.defaultLayer), Layer.provide(Bus.layer)) export * as Permission from "." diff --git a/packages/opencode/src/server/routes/instance/httpapi/api.ts b/packages/opencode/src/server/routes/instance/httpapi/api.ts index eff336b3c..59392a03b 100644 --- a/packages/opencode/src/server/routes/instance/httpapi/api.ts +++ b/packages/opencode/src/server/routes/instance/httpapi/api.ts @@ -1,7 +1,6 @@ import { Schema } from "effect" import { HttpApi } from "effect/unstable/httpapi" import { BusEvent } from "@/bus/bus-event" -import { SyncEvent } from "@/sync" import { ConfigApi } from "./groups/config" import { ControlApi } from "./groups/control" import { EventApi } from "./groups/event" @@ -23,9 +22,8 @@ import { V2Api } from "./groups/v2" import { Authorization } from "./middleware/authorization" import { SchemaErrorMiddleware } from "./middleware/schema-error" -// SSE event schemas built from the BusEvent/SyncEvent registries. +// SSE event schemas built from the BusEvent and EventV2 registries. const EventSchema = Schema.Union(BusEvent.effectPayloads()).annotate({ identifier: "Event" }) -const SyncEventSchemas = SyncEvent.effectPayloads() export const RootHttpApi = HttpApi.make("opencode-root") .addHttpApi(ControlApi) @@ -56,7 +54,7 @@ export const OpenCodeHttpApi = HttpApi.make("opencode") .addHttpApi(EventApi) .addHttpApi(InstanceHttpApi) .addHttpApi(PtyConnectApi) - .annotate(HttpApi.AdditionalSchemas, [EventSchema, ...SyncEventSchemas]) + .annotate(HttpApi.AdditionalSchemas, [EventSchema]) export type RootHttpApiType = typeof RootHttpApi export type InstanceHttpApiType = typeof InstanceHttpApi diff --git a/packages/opencode/src/server/routes/instance/httpapi/groups/global.ts b/packages/opencode/src/server/routes/instance/httpapi/groups/global.ts index 75441b4ca..6d2af6695 100644 --- a/packages/opencode/src/server/routes/instance/httpapi/groups/global.ts +++ b/packages/opencode/src/server/routes/instance/httpapi/groups/global.ts @@ -1,6 +1,5 @@ import { Config } from "@/config/config" import { BusEvent } from "@/bus/bus-event" -import { SyncEvent } from "@/sync" import "@/server/event" import { Schema } from "effect" import { HttpApi, HttpApiEndpoint, HttpApiError, HttpApiGroup, OpenApi } from "effect/unstable/httpapi" @@ -15,7 +14,7 @@ const GlobalEventSchema = Schema.Struct({ directory: Schema.String, project: Schema.optional(Schema.String), workspace: Schema.optional(Schema.String), - payload: Schema.Union([...BusEvent.effectPayloads(), ...SyncEvent.effectPayloads()]), + payload: Schema.Union(BusEvent.effectPayloads()), }).annotate({ identifier: "GlobalEvent" }) export const GlobalUpgradeInput = Schema.Struct({ diff --git a/packages/opencode/src/server/routes/instance/httpapi/handlers/sync.ts b/packages/opencode/src/server/routes/instance/httpapi/handlers/sync.ts index 2a988753d..5269f3546 100644 --- a/packages/opencode/src/server/routes/instance/httpapi/handlers/sync.ts +++ b/packages/opencode/src/server/routes/instance/httpapi/handlers/sync.ts @@ -1,8 +1,9 @@ import { Workspace } from "@/control-plane/workspace" import * as InstanceState from "@/effect/instance-state" import { Session } from "@/session/session" -import { Database } from "@/storage/db" -import { SyncEvent } from "@/sync" +import { Database } from "@opencode-ai/core/database/database" +import { EventV2 } from "@opencode-ai/core/event" +import { EventV2Bridge } from "@/event-v2-bridge" import { EventTable } from "@opencode-ai/core/event/sql" import { asc } from "drizzle-orm" import { and } from "drizzle-orm" @@ -23,7 +24,8 @@ export const syncHandlers = HttpApiBuilder.group(InstanceHttpApi, "sync", (handl const workspace = yield* Workspace.Service const session = yield* Session.Service const scope = yield* Scope.Scope - const sync = yield* SyncEvent.Service + const events = yield* EventV2Bridge.Service + const { db } = yield* Database.Service const start = Effect.fn("SyncHttpApi.start")(function* () { yield* workspace @@ -33,27 +35,27 @@ export const syncHandlers = HttpApiBuilder.group(InstanceHttpApi, "sync", (handl }) const replay = Effect.fn("SyncHttpApi.replay")(function* (ctx: { payload: typeof ReplayPayload.Type }) { - const events: SyncEvent.SerializedEvent[] = ctx.payload.events.map((event) => ({ - id: event.id, + const payload: EventV2.SerializedEvent[] = ctx.payload.events.map((event) => ({ + id: EventV2.ID.make(event.id), aggregateID: event.aggregateID, seq: event.seq, type: event.type, data: { ...event.data }, })) - const source = events[0].aggregateID + const source = payload[0].aggregateID log.info("sync replay requested", { sessionID: source, - events: events.length, - first: events[0]?.seq, - last: events.at(-1)?.seq, + events: payload.length, + first: payload[0]?.seq, + last: payload.at(-1)?.seq, directory: ctx.payload.directory, }) - yield* sync.replayAll(events) + yield* events.replayAll(payload) log.info("sync replay complete", { sessionID: source, - events: events.length, - first: events[0]?.seq, - last: events.at(-1)?.seq, + events: payload.length, + first: payload[0]?.seq, + last: payload.at(-1)?.seq, }) return { sessionID: source } }) @@ -74,18 +76,17 @@ export const syncHandlers = HttpApiBuilder.group(InstanceHttpApi, "sync", (handl const history = Effect.fn("SyncHttpApi.history")(function* (ctx: { payload: typeof HistoryPayload.Type }) { const exclude = Object.entries(ctx.payload) - return Database.use((db) => - db - .select() - .from(EventTable) - .where( - exclude.length > 0 - ? not(or(...exclude.map(([id, seq]) => and(eq(EventTable.aggregate_id, id), lte(EventTable.seq, seq))))!) - : undefined, - ) - .orderBy(asc(EventTable.seq)) - .all(), - ) + return yield* db + .select() + .from(EventTable) + .where( + exclude.length > 0 + ? not(or(...exclude.map(([id, seq]) => and(eq(EventTable.aggregate_id, id), lte(EventTable.seq, seq))))!) + : undefined, + ) + .orderBy(asc(EventTable.seq)) + .all() + .pipe(Effect.orDie) }) return handlers.handle("start", start).handle("replay", replay).handle("steal", steal).handle("history", history) diff --git a/packages/opencode/src/server/routes/instance/httpapi/server.ts b/packages/opencode/src/server/routes/instance/httpapi/server.ts index 1b98134ee..94990fdba 100644 --- a/packages/opencode/src/server/routes/instance/httpapi/server.ts +++ b/packages/opencode/src/server/routes/instance/httpapi/server.ts @@ -49,7 +49,6 @@ import { EventV2Bridge } from "@/event-v2-bridge" import { Database } from "@opencode-ai/core/database/database" import { Skill } from "@/skill" import { Snapshot } from "@/snapshot" -import { SyncEvent } from "@/sync" import { ToolRegistry } from "@/tool/registry" import { lazy } from "@/util/lazy" import { Vcs } from "@/project/vcs" @@ -192,12 +191,12 @@ export function createRoutes( corsVaryFix, fenceLayer, cors(corsOptions), + Database.defaultLayer, Account.defaultLayer, Agent.defaultLayer, Auth.defaultLayer, Command.defaultLayer, Config.defaultLayer, - Database.defaultLayer, File.defaultLayer, FileWatcher.defaultLayer, Format.defaultLayer, @@ -225,7 +224,6 @@ export function createRoutes( SessionSummary.defaultLayer, ShareNext.defaultLayer, Snapshot.defaultLayer, - SyncEvent.defaultLayer, EventV2Bridge.defaultLayer, Skill.defaultLayer, Todo.defaultLayer, @@ -233,7 +231,7 @@ export function createRoutes( Vcs.defaultLayer, Workspace.defaultLayer, Worktree.appLayer, - Bus.layer, + Bus.defaultLayer, AppFileSystem.defaultLayer, FetchHttpClient.layer, HttpServer.layerServices, diff --git a/packages/opencode/src/server/shared/fence.ts b/packages/opencode/src/server/shared/fence.ts index 8cf27ed50..a3e86b165 100644 --- a/packages/opencode/src/server/shared/fence.ts +++ b/packages/opencode/src/server/shared/fence.ts @@ -1,25 +1,28 @@ -import { Database } from "@/storage/db" +import { Database } from "@opencode-ai/core/database/database" import { inArray } from "drizzle-orm" import { EventSequenceTable } from "@opencode-ai/core/event/sql" import { Workspace } from "@/control-plane/workspace" import type { WorkspaceV2 } from "@opencode-ai/core/workspace" import * as Log from "@opencode-ai/core/util/log" import { Effect } from "effect" +import { makeRuntime } from "@/effect/run-service" export const HEADER = "x-opencode-sync" export type State = Record const log = Log.create({ service: "fence" }) +const runtime = makeRuntime(Database.Service, Database.defaultLayer) export function load(ids?: string[]) { - const rows = Database.use((db) => { - if (!ids?.length) { - return db.select().from(EventSequenceTable).all() - } + return runtime.runSync(({ db }) => + Effect.gen(function* () { + const rows = yield* (ids?.length + ? db.select().from(EventSequenceTable).where(inArray(EventSequenceTable.aggregate_id, ids)).all() + : db.select().from(EventSequenceTable).all() + ).pipe(Effect.orDie) - return db.select().from(EventSequenceTable).where(inArray(EventSequenceTable.aggregate_id, ids)).all() - }) - - return Object.fromEntries(rows.map((row) => [row.aggregate_id, row.seq])) + return Object.fromEntries(rows.map((row) => [row.aggregate_id, row.seq])) + }), + ) } export function diff(prev: State, next: State) { diff --git a/packages/opencode/src/session/prompt.ts b/packages/opencode/src/session/prompt.ts index 041438c50..757d1b58f 100644 --- a/packages/opencode/src/session/prompt.ts +++ b/packages/opencode/src/session/prompt.ts @@ -1652,7 +1652,6 @@ export const defaultLayer = Layer.suspend(() => Layer.provide(Image.defaultLayer), Layer.provide( Layer.mergeAll( - EventV2Bridge.defaultLayer, Agent.defaultLayer, Database.defaultLayer, SystemPrompt.defaultLayer, @@ -1661,6 +1660,7 @@ export const defaultLayer = Layer.suspend(() => Bus.layer, CrossSpawnSpawner.defaultLayer, RuntimeFlags.defaultLayer, + EventV2Bridge.defaultLayer, ), ), ), diff --git a/packages/opencode/src/session/session.ts b/packages/opencode/src/session/session.ts index 7da998593..fa9436432 100644 --- a/packages/opencode/src/session/session.ts +++ b/packages/opencode/src/session/session.ts @@ -10,7 +10,7 @@ import type { ProviderMetadata, Usage } from "@opencode-ai/llm" import { InstallationVersion } from "@opencode-ai/core/installation/version" import { Database } from "@opencode-ai/core/database/database" import { makeRuntime } from "@opencode-ai/core/effect/runtime" -import { EventV2 } from "@opencode-ai/core/event" +import { EventV2Bridge } from "@/event-v2-bridge" import { SessionV2 } from "@opencode-ai/core/session" import { NotFoundError } from "@/storage/storage" @@ -510,7 +510,7 @@ export type Patch = Omit, "time" | "share" | "summary" | "revert" export const layer: Layer.Layer< Service, never, - BackgroundJob.Service | Bus.Service | Storage.Service | RuntimeFlags.Service | Database.Service | EventV2.Service + BackgroundJob.Service | Bus.Service | Storage.Service | RuntimeFlags.Service | Database.Service | EventV2Bridge.Service > = Layer.effect( Service, Effect.gen(function* () { @@ -518,7 +518,7 @@ export const layer: Layer.Layer< const database = yield* Database.Service const background = yield* BackgroundJob.Service const bus = yield* Bus.Service - const events = yield* EventV2.Service + const events = yield* EventV2Bridge.Service const storage = yield* Storage.Service const flags = yield* RuntimeFlags.Service @@ -935,7 +935,7 @@ export const defaultLayer = layer.pipe( Layer.provide(Bus.layer), Layer.provide(Storage.defaultLayer), Layer.provide(Database.defaultLayer), - Layer.provide(EventV2.defaultLayer), + Layer.provide(EventV2Bridge.defaultLayer), Layer.provide(SessionV2.defaultLayer), Layer.provide(RuntimeFlags.defaultLayer), ) diff --git a/packages/opencode/src/storage/db.bun.ts b/packages/opencode/src/storage/db.bun.ts deleted file mode 100644 index f2541c74e..000000000 --- a/packages/opencode/src/storage/db.bun.ts +++ /dev/null @@ -1,12 +0,0 @@ -import { Database } from "bun:sqlite" -import { Sqlite } from "@opencode-ai/core/database/sqlite" -import { layer } from "@opencode-ai/core/database/sqlite.bun" -import { makeRuntime } from "@opencode-ai/core/effect/runtime" -import { drizzle } from "drizzle-orm/bun-sqlite" -import { Effect } from "effect" - -export function init(path: string) { - const runtime = makeRuntime(Sqlite.Native, layer({ filename: path })) - const native = runtime.runSync((native) => Effect.succeed(native)) as Database - return drizzle({ client: native }) -} diff --git a/packages/opencode/src/storage/db.node.ts b/packages/opencode/src/storage/db.node.ts deleted file mode 100644 index ec6eb1243..000000000 --- a/packages/opencode/src/storage/db.node.ts +++ /dev/null @@ -1,12 +0,0 @@ -import { DatabaseSync } from "node:sqlite" -import { Sqlite } from "@opencode-ai/core/database/sqlite" -import { layer } from "@opencode-ai/core/database/sqlite.node" -import { makeRuntime } from "@opencode-ai/core/effect/runtime" -import { drizzle } from "drizzle-orm/node-sqlite" -import { Effect } from "effect" - -export function init(path: string) { - const runtime = makeRuntime(Sqlite.Native, layer({ filename: path })) - const native = runtime.runSync((native) => Effect.succeed(native)) as DatabaseSync - return drizzle({ client: native }) -} diff --git a/packages/opencode/src/storage/db.ts b/packages/opencode/src/storage/db.ts deleted file mode 100644 index 580a0bb81..000000000 --- a/packages/opencode/src/storage/db.ts +++ /dev/null @@ -1,115 +0,0 @@ -import { type SQLiteTransaction } from "drizzle-orm/sqlite-core" -import type { TablesRelationalConfig } from "drizzle-orm/relations" -export * from "drizzle-orm" -import { LocalContext } from "@/util/local-context" -import * as Log from "@opencode-ai/core/util/log" -import { NamedError } from "@opencode-ai/core/util/error" -import { EffectBridge } from "@/effect/bridge" -import { Effect, Schema } from "effect" -import { Database } from "@opencode-ai/core/database/database" -import { makeRuntime } from "@opencode-ai/core/effect/runtime" - -export const NotFoundError = NamedError.create("NotFoundError", { - message: Schema.String, -}) - -const log = Log.create({ service: "db" }) -const runtime = makeRuntime(Database.Service, Database.defaultLayer) -const database = await runtime.runPromise((db) => Effect.succeed(db)) - -export const getPath = () => Database.path() - -export type Transaction = SQLiteTransaction<"sync", void, Record, TablesRelationalConfig> - -type Client = Database.Interface["drizzle"] - -let client: Client | undefined -let loaded = false - -export const Client = Object.assign( - (): Client => { - if (loaded) return client as Client - - const dbPath = getPath() - log.info("opening database", { path: dbPath }) - - const db = database.drizzle - - db.run("PRAGMA journal_mode = WAL") - db.run("PRAGMA synchronous = NORMAL") - db.run("PRAGMA busy_timeout = 5000") - db.run("PRAGMA cache_size = -64000") - db.run("PRAGMA foreign_keys = ON") - db.run("PRAGMA wal_checkpoint(PASSIVE)") - - client = db - loaded = true - return db - }, - { - reset: () => { - loaded = false - client = undefined - }, - loaded: () => loaded, - }, -) - -export function close() { - if (!Client.loaded()) return - Client.reset() -} - -export type TxOrDb = Transaction | Client - -const ctx = LocalContext.create<{ - tx: TxOrDb - effects: (() => void | Promise)[] -}>("database") - -export function use(callback: (trx: TxOrDb) => T): T { - try { - return callback(ctx.use().tx) - } catch (err) { - if (err instanceof LocalContext.NotFound) { - const effects: (() => void | Promise)[] = [] - const result = ctx.provide({ effects, tx: Client() }, () => callback(Client())) - for (const effect of effects) effect() - return result - } - throw err - } -} - -export function effect(fn: () => any | Promise) { - const bound = EffectBridge.bind(fn) - try { - ctx.use().effects.push(bound) - } catch { - bound() - } -} - -type NotPromise = T extends Promise ? never : T - -export function transaction( - callback: (tx: TxOrDb) => NotPromise, - options?: { - behavior?: "deferred" | "immediate" | "exclusive" - }, -): NotPromise { - try { - return callback(ctx.use().tx) - } catch (err) { - if (err instanceof LocalContext.NotFound) { - const effects: (() => void | Promise)[] = [] - const txCallback = EffectBridge.bind((tx: TxOrDb) => ctx.provide({ tx, effects }, () => callback(tx))) - const result = Client().transaction(txCallback, { behavior: options?.behavior }) - for (const effect of effects) effect() - return result as NotPromise - } - throw err - } -} - -export * as Database from "./db" diff --git a/packages/opencode/src/sync/index.ts b/packages/opencode/src/sync/index.ts deleted file mode 100644 index 25cae5b9a..000000000 --- a/packages/opencode/src/sync/index.ts +++ /dev/null @@ -1,411 +0,0 @@ -// Legacy sync event system. It should stay unaware of core EventV2 execution; -// the only temporary V2 coupling here is exposing versioned core event schemas -// in effectPayloads() so existing HTTP/SDK schema generation remains stable. -// Remove that registry read when event schemas are generated from core directly. -import { Database } from "@/storage/db" -import { eq } from "drizzle-orm" -import { GlobalBus } from "@/bus/global" -import { Bus as ProjectBus } from "@/bus" -import { BusEvent } from "@/bus/bus-event" -import { EventSequenceTable, EventTable } from "@opencode-ai/core/event/sql" -import { EventID } from "./schema" -import { Context, Effect, Layer, Schema as EffectSchema } from "effect" -import type { DeepMutable } from "@opencode-ai/core/schema" -import { EventV2 } from "@opencode-ai/core/event" -import { serviceUse } from "@opencode-ai/core/effect/service-use" -import { InstanceState } from "@/effect/instance-state" -import { RuntimeFlags } from "@/effect/runtime-flags" -import { EffectBridge } from "@/effect/bridge" - -// Keep `Event["data"]` mutable because projectors mutate the persisted shape -// when writing to the database. Bus payloads (`Properties`) stay readonly — -// subscribers only read. - -export type Definition< - Type extends string = string, - Schema extends EffectSchema.Top = EffectSchema.Top, - BusSchema extends EffectSchema.Top = Schema, -> = { - type: Type - version: number - aggregate: string - schema: Schema - // Bus event payload schema. Defaults to `schema` unless `busSchema` was - // passed at definition time (see `session.updated`, whose projector - // expands the persisted data to a `{ sessionID, info }` bus payload). - properties: BusSchema -} - -export type Event = { - id: string - seq: number - aggregateID: string - data: DeepMutable> -} - -export type Properties = EffectSchema.Schema.Type - -export type SerializedEvent = Event & { type: string } - -type ProjectorFunc = (db: Database.TxOrDb, data: unknown, event: Event) => void -type ConvertEvent = (type: string, data: Event["data"]) => unknown | Promise - -export interface Interface { - readonly run: ( - def: Def, - data: Event["data"], - options?: { publish?: boolean }, - ) => Effect.Effect - readonly replay: (event: SerializedEvent, options?: { publish: boolean; ownerID?: string }) => Effect.Effect - readonly replayAll: ( - events: SerializedEvent[], - options?: { publish: boolean; ownerID?: string }, - ) => Effect.Effect - readonly remove: (aggregateID: string) => Effect.Effect - readonly claim: (aggregateID: string, ownerID: string) => Effect.Effect -} - -export class Service extends Context.Service()("@opencode/SyncEvent") {} - -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) - if (!def) { - throw new Error(`Unknown event type: ${event.type}`) - } - - const row = Database.use((db) => - db - .select({ seq: EventSequenceTable.seq, ownerID: EventSequenceTable.owner_id }) - .from(EventSequenceTable) - .where(eq(EventSequenceTable.aggregate_id, event.aggregateID)) - .get(), - ) - - const latest = row?.seq ?? -1 - if (event.seq <= latest) return - - if (row?.ownerID && row.ownerID !== options?.ownerID) { - return - } - - const expected = latest + 1 - if (event.seq !== expected) { - throw new Error( - `Sequence mismatch for aggregate "${event.aggregateID}": expected ${expected}, got ${event.seq}`, - ) - } - - const publish = !!options?.publish - // Bridge captures handler-fiber refs (InstanceRef/WorkspaceRef) and the - // full Effect context, so the forked publish + GlobalBus emit run with - // the right state without a per-call attachWith. - const bridge = yield* EffectBridge.make() - process(def, event, { - bus, - bridge, - publish, - ownerID: options?.ownerID, - experimentalWorkspaces: flags.experimentalWorkspaces, - }) - }) - - 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)[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 || {} - const bridge = yield* EffectBridge.make() - - // 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, { bus, bridge, publish, experimentalWorkspaces: flags.experimentalWorkspaces }) - }, - { - 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() - }) - }) - - const claim: Interface["claim"] = Effect.fn("SyncEvent.claim")((aggregateID, ownerID) => - Effect.sync(() => - Database.use((db) => - db - .update(EventSequenceTable) - .set({ owner_id: ownerID }) - .where(eq(EventSequenceTable.aggregate_id, aggregateID)) - .run(), - ), - ), - ) - - return Service.of({ - run, - replay, - replayAll, - remove, - claim, - }) - }), -) - -export const defaultLayer = layer.pipe(Layer.provide([ProjectBus.defaultLayer, RuntimeFlags.defaultLayer])) - -export const use = serviceUse(Service) - -export const registry = new Map() -let projectors: Map | undefined -const versions = new Map() -let frozen = false -let convertEvent: ConvertEvent - -export function reset() { - frozen = false - projectors = undefined - convertEvent = (_, data) => data -} - -export function init(input: { projectors: Array<[Definition, ProjectorFunc]>; convertEvent?: ConvertEvent }) { - projectors = new Map(input.projectors.map(([def, func]) => [versionedType(def.type, def.version), func])) - for (let entry of EventV2.registry.values()) { - if (!entry.sync) continue - register({ - type: entry.type, - version: entry.sync.version, - aggregate: entry.sync.aggregate, - properties: entry.data, - schema: entry.data, - }) - } - - // Install all the latest event defs to the bus. We only ever emit - // latest versions from code, and keep around old versions for - // replaying. Replaying does not go through the bus, and it - // simplifies the bus to only use unversioned latest events - for (let [type, version] of versions.entries()) { - let def = registry.get(versionedType(type, version))! - BusEvent.define(def.type, def.properties) - } - - // Freeze the system so it clearly errors if events are defined - // after `init` which would cause bugs - frozen = true - convertEvent = input.convertEvent ?? ((_, data) => data) -} - -export function versionedType(type: A): A -export function versionedType(type: A, version: B): `${A}/${B}` -export function versionedType(type: string, version?: number) { - return version ? `${type}.${version}` : type -} - -export function define< - Type extends string, - Agg extends string, - Schema extends EffectSchema.Top, - BusSchema extends EffectSchema.Top = Schema, ->(input: { - type: Type - version: number - aggregate: Agg - schema: Schema - busSchema?: BusSchema -}): Definition { - if (frozen) { - throw new Error("Error defining sync event: sync system has been frozen") - } - - const def = { - type: input.type, - version: input.version, - aggregate: input.aggregate, - schema: input.schema, - properties: (input.busSchema ?? input.schema) as BusSchema, - } - - register(def) - - return def -} - -export function project( - def: Def, - func: (db: Database.TxOrDb, data: Event["data"], event: Event) => void, -): [Definition, ProjectorFunc] { - return [def, func as ProjectorFunc] -} - -function register(def: Definition) { - versions.set(def.type, Math.max(def.version, versions.get(def.type) || 0)) - registry.set(versionedType(def.type, def.version), def) -} - -function process( - def: Def, - event: Event, - options: { - bus: ProjectBus.Interface - bridge: EffectBridge.Shape - publish: boolean - ownerID?: string - experimentalWorkspaces: boolean - }, -) { - if (projectors == null) { - throw new Error("No projectors available. Call `SyncEvent.init` to install projectors") - } - - const projector = projectors.get(versionedType(def.type, def.version)) - if (!projector) { - if (!def.type.includes("next")) throw new Error(`Projector not found for event: ${def.type}`) - return - } - - Database.transaction((tx) => { - projector(tx, event.data, event) - - if (options.experimentalWorkspaces) { - tx.insert(EventSequenceTable) - .values({ - aggregate_id: event.aggregateID, - seq: event.seq, - owner_id: options?.ownerID, - }) - .onConflictDoUpdate({ - target: EventSequenceTable.aggregate_id, - set: { seq: event.seq }, - }) - .run() - tx.insert(EventTable) - .values({ - id: event.id as never, - seq: event.seq, - aggregate_id: event.aggregateID, - type: versionedType(def.type, def.version), - data: event.data as never, - }) - .run() - } - - Database.effect(() => { - if (!options.publish) return - const result = convertEvent(def.type, event.data) - // The bridge was built inside the caller's fiber so it already carries - // InstanceRef/WorkspaceRef and the full Effect context. Both the bus - // publish and the GlobalBus emit run inside the forked Effect so they - // share the same instance/workspace lookup. - const publish = (data: unknown) => - options.bridge.fork( - Effect.gen(function* () { - yield* options.bus.publish(def, data as Properties, { id: event.id }) - const instance = yield* InstanceState.context - const workspace = yield* InstanceState.workspaceID - GlobalBus.emit("event", { - directory: instance.directory, - project: instance.project.id, - workspace, - payload: { - type: "sync", - syncEvent: { - type: versionedType(def.type, def.version), - ...event, - }, - }, - }) - }), - ) - if (result instanceof Promise) { - void result.then(publish) - } else { - publish(result) - } - }) - }) -} - -export function effectPayloads() { - return [ - ...registry - .entries() - .map(([type, def]) => - EffectSchema.Struct({ - type: EffectSchema.Literal("sync"), - name: EffectSchema.Literal(type), - id: EffectSchema.String, - seq: EffectSchema.Finite, - aggregateID: EffectSchema.Literal(def.aggregate), - data: def.schema, - }).annotate({ identifier: `SyncEvent.${type}` }), - ) - .toArray(), - ...EventV2.registry - .values() - .filter( - (definition) => - definition.sync !== undefined && !registry.has(versionedType(definition.type, definition.sync.version)), - ) - .map((definition) => - EffectSchema.Struct({ - type: EffectSchema.Literal("sync"), - name: EffectSchema.Literal(versionedType(definition.type, definition.sync!.version)), - id: EffectSchema.String, - seq: EffectSchema.Finite, - aggregateID: EffectSchema.Literal(definition.sync!.aggregate), - data: definition.data, - }).annotate({ identifier: `SyncEvent.${definition.type}` }), - ) - .toArray(), - ] -} - -export * as SyncEvent from "." diff --git a/packages/opencode/src/worktree/index.ts b/packages/opencode/src/worktree/index.ts index 534bba2fc..f17644702 100644 --- a/packages/opencode/src/worktree/index.ts +++ b/packages/opencode/src/worktree/index.ts @@ -2,7 +2,7 @@ import { Global } from "@opencode-ai/core/global" import { InstanceLayer } from "@/project/instance-layer" import { InstanceStore } from "@/project/instance-store" import { Project } from "@/project/project" -import { Database } from "@/storage/db" +import { Database } from "@opencode-ai/core/database/database" import { eq } from "drizzle-orm" import { ProjectTable } from "@opencode-ai/core/project/sql" import type { ProjectV2 } from "@opencode-ai/core/project" @@ -149,7 +149,7 @@ type GitResult = { code: number; text: string; stderr: string } export const layer: Layer.Layer< Service, never, - AppFileSystem.Service | Path.Path | AppProcess.Service | Git.Service | Project.Service | InstanceStore.Service + AppFileSystem.Service | Path.Path | AppProcess.Service | Git.Service | Project.Service | InstanceStore.Service | Database.Service > = Layer.effect( Service, Effect.gen(function* () { @@ -157,6 +157,7 @@ export const layer: Layer.Layer< const fs = yield* AppFileSystem.Service const pathSvc = yield* Path.Path const appProcess = yield* AppProcess.Service + const { db } = yield* Database.Service const gitSvc = yield* Git.Service const project = yield* Project.Service const store = yield* InstanceStore.Service @@ -478,9 +479,7 @@ export const layer: Layer.Layer< directory: string, input: { projectID: ProjectV2.ID; extra?: string }, ) { - const row = yield* Effect.sync(() => - Database.use((db) => db.select().from(ProjectTable).where(eq(ProjectTable.id, input.projectID)).get()), - ) + const row = yield* db.select().from(ProjectTable).where(eq(ProjectTable.id, input.projectID)).get().pipe(Effect.orDie) const project = row ? Project.fromRow(row) : undefined const startup = project?.commands?.start?.trim() ?? "" const ok = yield* runStartScript(directory, startup, "project") @@ -611,6 +610,7 @@ export const appLayer = layer.pipe( Layer.provide(Git.defaultLayer), Layer.provide(AppProcess.defaultLayer), Layer.provide(Project.defaultLayer), + Layer.provide(Database.defaultLayer), Layer.provide(AppFileSystem.defaultLayer), Layer.provide(NodePath.layer), ) diff --git a/packages/opencode/test/account/repo.test.ts b/packages/opencode/test/account/repo.test.ts index 5679794a4..42851fc19 100644 --- a/packages/opencode/test/account/repo.test.ts +++ b/packages/opencode/test/account/repo.test.ts @@ -1,18 +1,19 @@ import { expect } from "bun:test" import { Effect, Layer, Option } from "effect" +import { sql } from "drizzle-orm" import { AccountRepo } from "../../src/account/repo" import { AccessToken, AccountID, OrgID, RefreshToken } from "../../src/account/schema" -import { Database } from "@/storage/db" +import { Database } from "@opencode-ai/core/database/database" import { testEffect } from "../lib/effect" const truncate = Layer.effectDiscard( - Effect.sync(() => { - const db = Database.Client() - db.run(/*sql*/ `DELETE FROM account_state`) - db.run(/*sql*/ `DELETE FROM account`) + Effect.gen(function* () { + const { db } = yield* Database.Service + yield* db.run(sql`DELETE FROM account_state`) + yield* db.run(sql`DELETE FROM account`) }), -) +).pipe(Layer.provide(Database.defaultLayer)) const it = testEffect(Layer.merge(AccountRepo.defaultLayer, truncate)) diff --git a/packages/opencode/test/account/service.test.ts b/packages/opencode/test/account/service.test.ts index 63b15c5b1..04d425e2c 100644 --- a/packages/opencode/test/account/service.test.ts +++ b/packages/opencode/test/account/service.test.ts @@ -1,5 +1,6 @@ import { expect } from "bun:test" import { Duration, Effect, Layer, Option, Schema } from "effect" +import { sql } from "drizzle-orm" import { HttpClient, HttpClientError, HttpClientResponse } from "effect/unstable/http" import { AccountRepo } from "../../src/account/repo" @@ -15,16 +16,16 @@ import { RefreshToken, UserCode, } from "../../src/account/schema" -import { Database } from "@/storage/db" +import { Database } from "@opencode-ai/core/database/database" import { testEffect } from "../lib/effect" const truncate = Layer.effectDiscard( - Effect.sync(() => { - const db = Database.Client() - db.run(/*sql*/ `DELETE FROM account_state`) - db.run(/*sql*/ `DELETE FROM account`) + Effect.gen(function* () { + const { db } = yield* Database.Service + yield* db.run(sql`DELETE FROM account_state`) + yield* db.run(sql`DELETE FROM account`) }), -) +).pipe(Layer.provide(Database.defaultLayer)) const it = testEffect(Layer.merge(AccountRepo.defaultLayer, truncate)) diff --git a/packages/opencode/test/control-plane/workspace.test.ts b/packages/opencode/test/control-plane/workspace.test.ts index 34b6160d8..5eb97c8da 100644 --- a/packages/opencode/test/control-plane/workspace.test.ts +++ b/packages/opencode/test/control-plane/workspace.test.ts @@ -10,14 +10,12 @@ import { eq } from "drizzle-orm" import { AppFileSystem } from "@opencode-ai/core/filesystem" import * as Log from "@opencode-ai/core/util/log" import { GlobalBus, type GlobalEvent } from "@/bus/global" -import { Database } from "@/storage/db" -import { Database as CoreDatabase } from "@opencode-ai/core/database/database" +import { Database } from "@opencode-ai/core/database/database" import { ProjectV2 } from "@opencode-ai/core/project" import { ProjectTable } from "@opencode-ai/core/project/sql" import { Session as SessionNs } from "@/session/session" import { SessionID } from "@/session/schema" import { SessionTable } from "@opencode-ai/core/session/sql" -import { SyncEvent } from "@/sync" import { EventSequenceTable } from "@opencode-ai/core/event/sql" import { resetDatabase } from "../fixture/db" import { disposeAllInstances, provideTmpdirInstance, requireInstance, TestInstance } from "../fixture/fixture" @@ -34,6 +32,7 @@ import { SessionPrompt } from "@/session/prompt" import { Project } from "@/project/project" import { Vcs } from "@/project/vcs" import { RuntimeFlags } from "@/effect/runtime-flags" +import { EventV2Bridge } from "@/event-v2-bridge" void Log.init({ print: false }) @@ -49,11 +48,11 @@ const workspaceLayer = (experimentalWorkspaces: boolean) => Workspace.layer.pipe( Layer.provide(Auth.defaultLayer), Layer.provide(SessionNs.defaultLayer), - Layer.provide(SyncEvent.defaultLayer), Layer.provide(SessionPrompt.defaultLayer), Layer.provide(Project.defaultLayer), Layer.provide(Vcs.defaultLayer), - Layer.provide(CoreDatabase.defaultLayer), + Layer.provide(Database.defaultLayer), + Layer.provide(EventV2Bridge.defaultLayer), Layer.provide(FetchHttpClient.layer), Layer.provide(AppFileSystem.defaultLayer), Layer.provide(RuntimeFlags.layer({ experimentalWorkspaces })), @@ -64,6 +63,7 @@ const testServerLayer = Layer.mergeAll( NodeHttpServer.layer(Http.createServer, { host: "127.0.0.1", port: 0 }), workspaceLayer(true), SessionNs.defaultLayer, + Database.defaultLayer, ) const it = testEffect(testServerLayer) @@ -107,7 +107,6 @@ function restoreEnv() { } beforeEach(() => { - Database.close() restoreEnv() process.env.OPENCODE_EXPERIMENTAL_WORKSPACES = "true" }) @@ -281,7 +280,7 @@ function workspaceInfo(projectID: ProjectV2.ID, type: string, input?: Partial + return Database.Service.use(({ db }) => db .insert(WorkspaceTable) .values({ @@ -294,12 +293,13 @@ function insertWorkspace(info: Workspace.Info) { project_id: info.projectID, time_used: info.timeUsed, }) - .run(), + .run() + .pipe(Effect.orDie), ) } function insertProject(id: ProjectV2.ID, worktree: string) { - Database.use((db) => + return Database.Service.use(({ db }) => db .insert(ProjectTable) .values({ @@ -311,34 +311,37 @@ function insertProject(id: ProjectV2.ID, worktree: string) { time_updated: Date.now(), sandboxes: [], }) - .run(), + .run() + .pipe(Effect.orDie), ) } function attachSessionToWorkspace(sessionID: SessionID, workspaceID: WorkspaceV2.ID) { - Database.use((db) => - db.update(SessionTable).set({ workspace_id: workspaceID }).where(eq(SessionTable.id, sessionID)).run(), + return Database.Service.use(({ db }) => + db.update(SessionTable).set({ workspace_id: workspaceID }).where(eq(SessionTable.id, sessionID)).run().pipe(Effect.orDie), ) } function sessionSequence(sessionID: SessionID) { - return Database.use((db) => + return Database.Service.use(({ db }) => db .select({ seq: EventSequenceTable.seq }) .from(EventSequenceTable) .where(eq(EventSequenceTable.aggregate_id, sessionID)) - .get(), - )?.seq + .get() + .pipe(Effect.orDie, Effect.map((row) => row?.seq)), + ) } function sessionSequenceOwner(sessionID: SessionID) { - return Database.use((db) => + return Database.Service.use(({ db }) => db .select({ ownerID: EventSequenceTable.owner_id }) .from(EventSequenceTable) .where(eq(EventSequenceTable.aggregate_id, sessionID)) - .get(), - )?.ownerID + .get() + .pipe(Effect.orDie, Effect.map((row) => row?.ownerID)), + ) } describe("workspace schemas and exports", () => { @@ -382,7 +385,7 @@ describe("workspace CRUD", () => { const instance = yield* requireInstance const workspace = yield* Workspace.Service const otherProjectID = ProjectV2.ID.make("project-other") - insertProject(otherProjectID, "/tmp/other") + yield* insertProject(otherProjectID, "/tmp/other") const a = workspaceInfo(instance.project.id, "manual", { id: WorkspaceV2.ID.ascending("wrk_a_list"), branch: "a", @@ -396,9 +399,9 @@ describe("workspace CRUD", () => { extra: ["b"], }) const other = workspaceInfo(otherProjectID, "manual", { id: WorkspaceV2.ID.ascending("wrk_c_list") }) - insertWorkspace(b) - insertWorkspace(other) - insertWorkspace(a) + yield* insertWorkspace(b) + yield* insertWorkspace(other) + yield* insertWorkspace(a) expect(yield* workspace.list(instance.project)).toEqual([a, b]) }), @@ -580,7 +583,7 @@ describe("workspace CRUD", () => { name: "existing", directory: path.join(instance.directory, "existing"), }) - insertWorkspace(existing) + yield* insertWorkspace(existing) const discovered = { type, @@ -765,8 +768,8 @@ describe("workspace CRUD", () => { const info = yield* workspace.create({ type, branch: null, projectID: instance.project.id, extra: null }) const one = yield* sessionSvc.create({}) const two = yield* sessionSvc.create({}) - attachSessionToWorkspace(one.id, info.id) - attachSessionToWorkspace(two.id, info.id) + yield* attachSessionToWorkspace(one.id, info.id) + yield* attachSessionToWorkspace(two.id, info.id) const removed = yield* workspace.remove(info.id) @@ -774,10 +777,14 @@ describe("workspace CRUD", () => { expect(yield* workspace.get(info.id)).toBeUndefined() expect(recorded.calls.remove).toEqual([info]) expect((yield* workspace.status()).find((item) => item.workspaceID === info.id)?.status).toBeUndefined() + const { db } = yield* Database.Service expect( - Database.use((db) => - db.select({ id: SessionTable.id }).from(SessionTable).where(eq(SessionTable.workspace_id, info.id)).all(), - ), + yield* db + .select({ id: SessionTable.id }) + .from(SessionTable) + .where(eq(SessionTable.workspace_id, info.id)) + .all() + .pipe(Effect.orDie), ).toEqual([]) }) }, @@ -804,7 +811,7 @@ describe("workspace CRUD", () => { }, }).adapter, ) - insertWorkspace(info) + yield* insertWorkspace(info) expect(yield* workspace.remove(info.id)).toEqual(info) expect(yield* workspace.get(info.id)).toBeUndefined() @@ -824,25 +831,27 @@ describe("workspace CRUD", () => { const targetType = unique("warp-target-local") const previous = workspaceInfo(instance.project.id, previousType) const target = workspaceInfo(instance.project.id, targetType) - insertWorkspace(previous) - insertWorkspace(target) + yield* insertWorkspace(previous) + yield* insertWorkspace(target) registerAdapter(instance.project.id, previousType, localAdapter(path.join(dir, "warp-prev-local")).adapter) registerAdapter(instance.project.id, targetType, localAdapter(path.join(dir, "warp-target-local")).adapter) const session = yield* sessionSvc.create({}) - attachSessionToWorkspace(session.id, previous.id) + yield* attachSessionToWorkspace(session.id, previous.id) yield* workspace.sessionWarp({ workspaceID: target.id, sessionID: session.id }) + const { db } = yield* Database.Service expect( - Database.use((db) => - db + ( + yield* db .select({ workspaceID: SessionTable.workspace_id }) .from(SessionTable) .where(eq(SessionTable.id, session.id)) - .get(), + .get() + .pipe(Effect.orDie) )?.workspaceID, ).toBe(target.id) - expect(sessionSequenceOwner(session.id)).toBe(target.id) + expect(yield* sessionSequenceOwner(session.id)).toBe(target.id) }) }, { git: true }, @@ -867,12 +876,12 @@ describe("workspace CRUD", () => { const previous = workspaceInfo(instance.project.id, previousType) const target = workspaceInfo(instance.project.id, targetType) - insertWorkspace(previous) - insertWorkspace(target) + yield* insertWorkspace(previous) + yield* insertWorkspace(target) registerAdapter(instance.project.id, previousType, localAdapter(previousDir, { createDir: false }).adapter) registerAdapter(instance.project.id, targetType, localAdapter(targetDir, { createDir: false }).adapter) const session = yield* sessionSvc.create({}) - attachSessionToWorkspace(session.id, previous.id) + yield* attachSessionToWorkspace(session.id, previous.id) yield* workspace.sessionWarp({ workspaceID: target.id, sessionID: session.id, copyChanges: true }) @@ -893,23 +902,25 @@ describe("workspace CRUD", () => { const sessionSvc = yield* SessionNs.Service const previousType = unique("warp-detach-local") const previous = workspaceInfo(instance.project.id, previousType) - insertWorkspace(previous) + yield* insertWorkspace(previous) registerAdapter(instance.project.id, previousType, localAdapter(path.join(dir, "warp-detach-local")).adapter) const session = yield* sessionSvc.create({}) - attachSessionToWorkspace(session.id, previous.id) + yield* attachSessionToWorkspace(session.id, previous.id) yield* workspace.sessionWarp({ workspaceID: null, sessionID: session.id }) + const { db } = yield* Database.Service expect( - Database.use((db) => - db + ( + yield* db .select({ workspaceID: SessionTable.workspace_id }) .from(SessionTable) .where(eq(SessionTable.id, session.id)) - .get(), + .get() + .pipe(Effect.orDie) )?.workspaceID, ).toBeNull() - expect(sessionSequenceOwner(session.id)).toBe(instance.project.id) + expect(yield* sessionSequenceOwner(session.id)).toBe(instance.project.id) }) }, { git: true }, @@ -926,9 +937,9 @@ describe("workspace CRUD", () => { const sessionSvc = yield* SessionNs.Service const previousType = unique("warp-detach-workspace-instance") const previous = workspaceInfo(projectID, previousType) - insertWorkspace(previous) + yield* insertWorkspace(previous) const session = yield* sessionSvc.create({}) - attachSessionToWorkspace(session.id, previous.id) + yield* attachSessionToWorkspace(session.id, previous.id) const workspaceProjectID = yield* provideTmpdirInstance( (workspaceDir) => @@ -942,17 +953,19 @@ describe("workspace CRUD", () => { { git: true }, ) + const { db } = yield* Database.Service expect( - Database.use((db) => - db + ( + yield* db .select({ workspaceID: SessionTable.workspace_id }) .from(SessionTable) .where(eq(SessionTable.id, session.id)) - .get(), + .get() + .pipe(Effect.orDie) )?.workspaceID, ).toBeNull() - expect(sessionSequenceOwner(session.id)).toBe(projectID) - expect(sessionSequenceOwner(session.id)).not.toBe(workspaceProjectID) + expect(yield* sessionSequenceOwner(session.id)).toBe(projectID) + expect(yield* sessionSequenceOwner(session.id)).not.toBe(workspaceProjectID) }), { git: true }, ) @@ -1005,14 +1018,14 @@ describe("workspace CRUD", () => { const targetType = unique("warp-remote-target") const previous = workspaceInfo(instance.project.id, previousType) const target = workspaceInfo(instance.project.id, targetType, { directory: "remote-target-dir" }) - insertWorkspace(previous) - insertWorkspace(target) + yield* insertWorkspace(previous) + yield* insertWorkspace(target) registerAdapter(instance.project.id, previousType, remoteAdapter(`${url}/warp-source`).adapter) registerAdapter(instance.project.id, targetType, remoteAdapter(`${url}/warp-target`).adapter) const session = yield* sessionSvc.create({}) - attachSessionToWorkspace(session.id, previous.id) + yield* attachSessionToWorkspace(session.id, previous.id) historySessionID = session.id - historyNextSeq = (sessionSequence(session.id) ?? -1) + 1 + historyNextSeq = ((yield* sessionSequence(session.id)) ?? -1) + 1 yield* workspace.sessionWarp({ workspaceID: target.id, sessionID: session.id, copyChanges: true }) @@ -1042,7 +1055,7 @@ describe("workspace CRUD", () => { }) expect(calls[4].json).toEqual({ sessionID: session.id }) expect((yield* sessionSvc.get(session.id)).title).toBe("from source history") - expect(sessionSequenceOwner(session.id)).toBe(target.id) + expect(yield* sessionSequenceOwner(session.id)).toBe(target.id) }), { git: true }, ) @@ -1062,8 +1075,8 @@ describe("workspace sync state", () => { const type = unique("flag-disabled") const info = workspaceInfo(instance.project.id, type) const session = yield* sessionSvc.create({}) - attachSessionToWorkspace(session.id, info.id) - insertWorkspace(info) + yield* attachSessionToWorkspace(session.id, info.id) + yield* insertWorkspace(info) registerAdapter(instance.project.id, type, localAdapter(path.join(dir, "flag-disabled")).adapter) yield* Effect.promise(() => startWorkspaceSyncingWithFlag(instance.project.id, false)) @@ -1088,12 +1101,10 @@ describe("workspace sync state", () => { const second = workspaceInfo(projectID, secondType) yield* Effect.promise(() => fs.mkdir(path.join(dir, "first"), { recursive: true })) yield* Effect.promise(() => fs.mkdir(path.join(dir, "second"), { recursive: true })) - yield* Effect.sync(() => { - insertWorkspace(first) - insertWorkspace(second) - registerAdapter(projectID, firstType, localAdapter(path.join(dir, "first")).adapter) - registerAdapter(projectID, secondType, localAdapter(path.join(dir, "second")).adapter) - }) + yield* insertWorkspace(first) + yield* insertWorkspace(second) + registerAdapter(projectID, firstType, localAdapter(path.join(dir, "first")).adapter) + registerAdapter(projectID, secondType, localAdapter(path.join(dir, "second")).adapter) yield* Effect.addFinalizer(() => Effect.all([workspace.remove(first.id), workspace.remove(second.id)], { discard: true }).pipe(Effect.ignore), ) @@ -1121,13 +1132,13 @@ describe("workspace sync state", () => { const sessionSvc = yield* SessionNs.Service const type = unique("missing-local") const info = workspaceInfo(instance.project.id, type) - insertWorkspace(info) + yield* insertWorkspace(info) registerAdapter( instance.project.id, type, localAdapter(path.join(dir, "missing-target"), { createDir: false }).adapter, ) - attachSessionToWorkspace((yield* sessionSvc.create({})).id, info.id) + yield* attachSessionToWorkspace((yield* sessionSvc.create({})).id, info.id) yield* workspace.startWorkspaceSyncing(instance.project.id) @@ -1157,9 +1168,9 @@ describe("workspace sync state", () => { const info = workspaceInfo(instance.project.id, type) const target = path.join(dir, "dedupe-local") yield* Effect.promise(() => fs.mkdir(target, { recursive: true })) - insertWorkspace(info) + yield* insertWorkspace(info) registerAdapter(instance.project.id, type, localAdapter(target).adapter) - attachSessionToWorkspace((yield* sessionSvc.create({})).id, info.id) + yield* attachSessionToWorkspace((yield* sessionSvc.create({})).id, info.id) yield* workspace.startWorkspaceSyncing(instance.project.id) yield* workspace.startWorkspaceSyncing(instance.project.id) @@ -1211,9 +1222,9 @@ describe("workspace sync state", () => { try { const type = unique("remote-start") const info = workspaceInfo(instance.project.id, type) - insertWorkspace(info) + yield* insertWorkspace(info) registerAdapter(instance.project.id, type, remoteAdapter(`${url}/sync`).adapter) - attachSessionToWorkspace((yield* sessionSvc.create({})).id, info.id) + yield* attachSessionToWorkspace((yield* sessionSvc.create({})).id, info.id) yield* workspace.startWorkspaceSyncing(instance.project.id) yield* eventuallyEffect( @@ -1265,9 +1276,9 @@ describe("workspace sync state", () => { const instance = yield* requireInstance const type = unique("remote-connect-fail") const info = workspaceInfo(instance.project.id, type) - insertWorkspace(info) + yield* insertWorkspace(info) registerAdapter(instance.project.id, type, remoteAdapter(`${url}/failed`).adapter) - attachSessionToWorkspace((yield* sessionSvc.create({})).id, info.id) + yield* attachSessionToWorkspace((yield* sessionSvc.create({})).id, info.id) yield* workspace.startWorkspaceSyncing(instance.project.id) @@ -1306,9 +1317,9 @@ describe("workspace sync state", () => { const instance = yield* requireInstance const type = unique("remote-history-fail") const info = workspaceInfo(instance.project.id, type) - insertWorkspace(info) + yield* insertWorkspace(info) registerAdapter(instance.project.id, type, remoteAdapter(`${url}/history-failed`).adapter) - attachSessionToWorkspace((yield* sessionSvc.create({})).id, info.id) + yield* attachSessionToWorkspace((yield* sessionSvc.create({})).id, info.id) yield* workspace.startWorkspaceSyncing(instance.project.id) @@ -1364,12 +1375,12 @@ describe("workspace sync state", () => { try { const type = unique("history-replay") const info = workspaceInfo(instance.project.id, type) - insertWorkspace(info) + yield* insertWorkspace(info) registerAdapter(instance.project.id, type, remoteAdapter(`${url}/history`).adapter) const session = yield* sessionSvc.create({ title: "before history" }) - attachSessionToWorkspace(session.id, info.id) + yield* attachSessionToWorkspace(session.id, info.id) historySessionID = session.id - historyNextSeq = (sessionSequence(session.id) ?? -1) + 1 + historyNextSeq = ((yield* sessionSequence(session.id)) ?? -1) + 1 yield* workspace.startWorkspaceSyncing(instance.project.id) @@ -1432,9 +1443,9 @@ describe("workspace sync state", () => { try { const type = unique("sse-forward") const info = workspaceInfo(instance.project.id, type) - insertWorkspace(info) + yield* insertWorkspace(info) registerAdapter(instance.project.id, type, remoteAdapter(`${url}/sse-forward`).adapter) - attachSessionToWorkspace((yield* sessionSvc.create({})).id, info.id) + yield* attachSessionToWorkspace((yield* sessionSvc.create({})).id, info.id) yield* workspace.startWorkspaceSyncing(instance.project.id) @@ -1514,12 +1525,12 @@ describe("workspace sync state", () => { try { const type = unique("sse-sync") const info = workspaceInfo(instance.project.id, type) - insertWorkspace(info) + yield* insertWorkspace(info) registerAdapter(instance.project.id, type, remoteAdapter(`${url}/sse-sync`).adapter) const session = yield* sessionSvc.create({ title: "before sse" }) - attachSessionToWorkspace(session.id, info.id) + yield* attachSessionToWorkspace(session.id, info.id) sseSessionID = session.id - sseNextSeq = (sessionSequence(session.id) ?? -1) + 1 + sseNextSeq = ((yield* sessionSequence(session.id)) ?? -1) + 1 yield* workspace.startWorkspaceSyncing(instance.project.id) @@ -1564,7 +1575,8 @@ describe("workspace waitForSync", () => { Effect.gen(function* () { const workspace = yield* Workspace.Service const sessionID = SessionID.descending("ses_wait_done") - Database.use((db) => db.insert(EventSequenceTable).values({ aggregate_id: sessionID, seq: 4 }).run()) + const { db } = yield* Database.Service + yield* db.insert(EventSequenceTable).values({ aggregate_id: sessionID, seq: 4 }).run().pipe(Effect.orDie) expect(yield* workspace.waitForSync(WorkspaceV2.ID.ascending("wrk_wait_done"), { [sessionID]: 4 })).toBeUndefined() expect( @@ -1581,20 +1593,20 @@ describe("workspace waitForSync", () => { const workspace = yield* Workspace.Service const workspaceID = WorkspaceV2.ID.ascending("wrk_wait_event") const sessionID = SessionID.descending("ses_wait_event") - Database.use((db) => db.insert(EventSequenceTable).values({ aggregate_id: sessionID, seq: 1 }).run()) + const { db } = yield* Database.Service + yield* db.insert(EventSequenceTable).values({ aggregate_id: sessionID, seq: 1 }).run().pipe(Effect.orDie) yield* Effect.all( [ workspace.waitForSync(workspaceID, { [sessionID]: 2 }), Effect.gen(function* () { yield* Effect.sleep("10 millis") - Database.use((db) => - db - .update(EventSequenceTable) - .set({ seq: 2 }) - .where(eq(EventSequenceTable.aggregate_id, sessionID)) - .run(), - ) + yield* db + .update(EventSequenceTable) + .set({ seq: 2 }) + .where(eq(EventSequenceTable.aggregate_id, sessionID)) + .run() + .pipe(Effect.orDie) GlobalBus.emit("event", { workspace: workspaceID, payload: { type: "anything" } }) }), ], @@ -1611,20 +1623,20 @@ describe("workspace waitForSync", () => { const workspace = yield* Workspace.Service const workspaceID = WorkspaceV2.ID.ascending("wrk_wait_sync_any") const sessionID = SessionID.descending("ses_wait_sync_any") - Database.use((db) => db.insert(EventSequenceTable).values({ aggregate_id: sessionID, seq: 0 }).run()) + const { db } = yield* Database.Service + yield* db.insert(EventSequenceTable).values({ aggregate_id: sessionID, seq: 0 }).run().pipe(Effect.orDie) yield* Effect.all( [ workspace.waitForSync(workspaceID, { [sessionID]: 1 }), Effect.gen(function* () { yield* Effect.sleep("10 millis") - Database.use((db) => - db - .update(EventSequenceTable) - .set({ seq: 1 }) - .where(eq(EventSequenceTable.aggregate_id, sessionID)) - .run(), - ) + yield* db + .update(EventSequenceTable) + .set({ seq: 1 }) + .where(eq(EventSequenceTable.aggregate_id, sessionID)) + .run() + .pipe(Effect.orDie) GlobalBus.emit("event", { workspace: WorkspaceV2.ID.ascending("wrk_other_workspace"), payload: { type: "sync" }, diff --git a/packages/opencode/test/fixture/db.ts b/packages/opencode/test/fixture/db.ts index db4a5df20..88f1097f2 100644 --- a/packages/opencode/test/fixture/db.ts +++ b/packages/opencode/test/fixture/db.ts @@ -1,11 +1,10 @@ import { rm } from "fs/promises" -import { Database } from "@/storage/db" +import { Database } from "@opencode-ai/core/database/database" import { disposeAllInstances } from "./fixture" export async function resetDatabase() { await disposeAllInstances().catch(() => undefined) - Database.close() - const dbPath = Database.getPath() + const dbPath = Database.path() await rm(dbPath, { force: true }).catch(() => undefined) await rm(`${dbPath}-wal`, { force: true }).catch(() => undefined) await rm(`${dbPath}-shm`, { force: true }).catch(() => undefined) diff --git a/packages/opencode/test/fixture/workspace.ts b/packages/opencode/test/fixture/workspace.ts index df2e22d10..b3dceddf8 100644 --- a/packages/opencode/test/fixture/workspace.ts +++ b/packages/opencode/test/fixture/workspace.ts @@ -11,17 +11,17 @@ import { Project } from "../../src/project/project" import { Vcs } from "../../src/project/vcs" import { Session } from "../../src/session/session" import { SessionPrompt } from "../../src/session/prompt" -import { SyncEvent } from "../../src/sync" +import { EventV2Bridge } from "../../src/event-v2-bridge" export const workspaceLayerWithRuntimeFlags = (overrides: Partial) => Workspace.layer.pipe( Layer.provide(Auth.defaultLayer), Layer.provide(Session.defaultLayer), - Layer.provide(SyncEvent.defaultLayer), Layer.provide(SessionPrompt.defaultLayer), Layer.provide(Project.defaultLayer), Layer.provide(Vcs.defaultLayer), Layer.provide(Database.defaultLayer), + Layer.provide(EventV2Bridge.defaultLayer), Layer.provide(FetchHttpClient.layer), Layer.provide(AppFileSystem.defaultLayer), Layer.provide(RuntimeFlags.layer(overrides)), diff --git a/packages/opencode/test/permission/next.test.ts b/packages/opencode/test/permission/next.test.ts index e969e67ff..de791653c 100644 --- a/packages/opencode/test/permission/next.test.ts +++ b/packages/opencode/test/permission/next.test.ts @@ -3,6 +3,7 @@ import os from "os" import { Cause, Deferred, Effect, Exit, Fiber, Layer } from "effect" import { Bus } from "../../src/bus" import { CrossSpawnSpawner } from "@opencode-ai/core/cross-spawn-spawner" +import { Database } from "@opencode-ai/core/database/database" import { Permission } from "../../src/permission" import { PermissionID } from "../../src/permission/schema" import { InstanceBootstrap } from "../../src/project/bootstrap-service" @@ -14,7 +15,7 @@ import { MessageID, SessionID } from "../../src/session/schema" const bus = Bus.layer const noopBootstrap = Layer.succeed(InstanceBootstrap.Service, InstanceBootstrap.Service.of({ run: Effect.void })) const env = Layer.mergeAll( - Permission.layer.pipe(Layer.provide(bus)), + Permission.layer.pipe(Layer.provide(Database.defaultLayer), Layer.provide(bus)), bus, CrossSpawnSpawner.defaultLayer, InstanceStore.defaultLayer.pipe(Layer.provide(noopBootstrap)), diff --git a/packages/opencode/test/plugin/workspace-adapter.test.ts b/packages/opencode/test/plugin/workspace-adapter.test.ts index 616244044..216e2e5a6 100644 --- a/packages/opencode/test/plugin/workspace-adapter.test.ts +++ b/packages/opencode/test/plugin/workspace-adapter.test.ts @@ -21,12 +21,12 @@ import { Vcs } from "../../src/project/vcs" import { InstanceState } from "../../src/effect/instance-state" import { Session } from "../../src/session/session" import { SessionPrompt } from "../../src/session/prompt" -import { SyncEvent } from "../../src/sync" import { disposeAllInstances, provideTmpdirInstance } from "../fixture/fixture" import { testEffect } from "../lib/effect" import { AccountTest } from "../fake/account" import { AuthTest } from "../fake/auth" import { NpmTest } from "../fake/npm" +import { EventV2Bridge } from "../../src/event-v2-bridge" const configLayer = Config.layer.pipe( Layer.provide(EffectFlock.defaultLayer), @@ -46,12 +46,12 @@ const noopBootstrapLayer = Layer.succeed(InstanceBootstrap.Service, InstanceBoot const workspaceLayer = Workspace.layer.pipe( Layer.provide(Auth.defaultLayer), Layer.provide(Session.defaultLayer), - Layer.provide(SyncEvent.defaultLayer), Layer.provide(SessionPrompt.defaultLayer), Layer.provide(Project.defaultLayer), Layer.provide(Vcs.defaultLayer), Layer.provide(FetchHttpClient.layer), Layer.provide(Database.defaultLayer), + Layer.provide(EventV2Bridge.defaultLayer), Layer.provide(AppFileSystem.defaultLayer), Layer.provide(InstanceStore.defaultLayer.pipe(Layer.provide(noopBootstrapLayer))), Layer.provide(RuntimeFlags.layer({ experimentalWorkspaces: true })), diff --git a/packages/opencode/test/preload.ts b/packages/opencode/test/preload.ts index 24b804819..7183d992a 100644 --- a/packages/opencode/test/preload.ts +++ b/packages/opencode/test/preload.ts @@ -10,8 +10,6 @@ import { afterAll } from "bun:test" const dir = path.join(os.tmpdir(), "opencode-test-data-" + process.pid) await fs.mkdir(dir, { recursive: true }) afterAll(async () => { - const { Database } = await import("../src/storage/db") - Database.close() const busy = (error: unknown) => typeof error === "object" && error !== null && "code" in error && error.code === "EBUSY" const rm = async (left: number): Promise => { diff --git a/packages/opencode/test/project/migrate-global.test.ts b/packages/opencode/test/project/migrate-global.test.ts index a9a511f18..006ae2473 100644 --- a/packages/opencode/test/project/migrate-global.test.ts +++ b/packages/opencode/test/project/migrate-global.test.ts @@ -1,6 +1,6 @@ import { describe, expect } from "bun:test" import { Project } from "@/project/project" -import { Database } from "@/storage/db" +import { Database } from "@opencode-ai/core/database/database" import { eq } from "drizzle-orm" import { SessionTable } from "@opencode-ai/core/session/sql" import { ProjectTable } from "@opencode-ai/core/project/sql" @@ -15,7 +15,7 @@ import { testEffect } from "../lib/effect" void Log.init({ print: false }) -const it = testEffect(Layer.mergeAll(Project.defaultLayer, CrossSpawnSpawner.defaultLayer)) +const it = testEffect(Layer.mergeAll(Project.defaultLayer, CrossSpawnSpawner.defaultLayer, Database.defaultLayer)) function legacySessionID() { // Global-session migration covers persisted IDs from before prefixed session IDs. @@ -24,7 +24,7 @@ function legacySessionID() { function seed(opts: { id: SessionID; dir: string; project: ProjectV2.ID }) { const now = Date.now() - Database.use((db) => + return Database.Service.use(({ db }) => db .insert(SessionTable) .values({ @@ -37,12 +37,13 @@ function seed(opts: { id: SessionID; dir: string; project: ProjectV2.ID }) { time_created: now, time_updated: now, }) - .run(), + .run() + .pipe(Effect.orDie), ) } function ensureGlobal() { - Database.use((db) => + return Database.Service.use(({ db }) => db .insert(ProjectTable) .values({ @@ -53,7 +54,8 @@ function ensureGlobal() { sandboxes: [], }) .onConflictDoNothing() - .run(), + .run() + .pipe(Effect.orDie), ) } @@ -72,7 +74,7 @@ describe("migrateFromGlobal", () => { // 2. Seed a session under "global" with matching directory const id = legacySessionID() - yield* Effect.sync(() => seed({ id, dir: tmp, project: ProjectV2.ID.global })) + yield* seed({ id, dir: tmp, project: ProjectV2.ID.global }) // 3. Make a commit so the project gets a real ID yield* Effect.promise(() => $`git commit --allow-empty -m "root"`.cwd(tmp).quiet()) @@ -81,7 +83,9 @@ describe("migrateFromGlobal", () => { expect(real.id).not.toBe(ProjectV2.ID.global) // 4. The session should have been migrated to the real project ID - const row = Database.use((db) => db.select().from(SessionTable).where(eq(SessionTable.id, id)).get()) + const row = yield* Database.Service.use(({ db }) => + db.select().from(SessionTable).where(eq(SessionTable.id, id)).get().pipe(Effect.orDie), + ) expect(row).toBeDefined() expect(row!.project_id).toBe(real.id) }), @@ -96,19 +100,21 @@ describe("migrateFromGlobal", () => { expect(project.id).not.toBe(ProjectV2.ID.global) // 2. Ensure "global" project row exists (as it would from a prior no-git session) - yield* Effect.sync(() => ensureGlobal()) + yield* ensureGlobal() // 3. Seed a session under "global" with matching directory. // This simulates a session created before git init that wasn't // present when the real project row was first created. const id = legacySessionID() - yield* Effect.sync(() => seed({ id, dir: tmp, project: ProjectV2.ID.global })) + yield* seed({ id, dir: tmp, project: ProjectV2.ID.global }) // 4. Call fromDirectory again — project row already exists, // so the current code skips migration entirely. This is the bug. yield* projects.fromDirectory(tmp) - const row = Database.use((db) => db.select().from(SessionTable).where(eq(SessionTable.id, id)).get()) + const row = yield* Database.Service.use(({ db }) => + db.select().from(SessionTable).where(eq(SessionTable.id, id)).get().pipe(Effect.orDie), + ) expect(row).toBeDefined() expect(row!.project_id).toBe(project.id) }), @@ -121,16 +127,18 @@ describe("migrateFromGlobal", () => { const { project } = yield* projects.fromDirectory(tmp) expect(project.id).not.toBe(ProjectV2.ID.global) - yield* Effect.sync(() => ensureGlobal()) + yield* ensureGlobal() // Legacy sessions may lack a directory value. // Without a matching origin directory, they should remain global. const id = legacySessionID() - yield* Effect.sync(() => seed({ id, dir: "", project: ProjectV2.ID.global })) + yield* seed({ id, dir: "", project: ProjectV2.ID.global }) yield* projects.fromDirectory(tmp) - const row = Database.use((db) => db.select().from(SessionTable).where(eq(SessionTable.id, id)).get()) + const row = yield* Database.Service.use(({ db }) => + db.select().from(SessionTable).where(eq(SessionTable.id, id)).get().pipe(Effect.orDie), + ) expect(row).toBeDefined() expect(row!.project_id).toBe(ProjectV2.ID.global) }), @@ -143,14 +151,16 @@ describe("migrateFromGlobal", () => { const { project } = yield* projects.fromDirectory(tmp) expect(project.id).not.toBe(ProjectV2.ID.global) - yield* Effect.sync(() => ensureGlobal()) + yield* ensureGlobal() // Seed a session under "global" but for a DIFFERENT directory const id = legacySessionID() - yield* Effect.sync(() => seed({ id, dir: "/some/other/dir", project: ProjectV2.ID.global })) + yield* seed({ id, dir: "/some/other/dir", project: ProjectV2.ID.global }) yield* projects.fromDirectory(tmp) - const row = Database.use((db) => db.select().from(SessionTable).where(eq(SessionTable.id, id)).get()) + const row = yield* Database.Service.use(({ db }) => + db.select().from(SessionTable).where(eq(SessionTable.id, id)).get().pipe(Effect.orDie), + ) expect(row).toBeDefined() // Should remain under "global" — not stolen expect(row!.project_id).toBe(ProjectV2.ID.global) diff --git a/packages/opencode/test/server/httpapi-event-diagnostics.test.ts b/packages/opencode/test/server/httpapi-event-diagnostics.test.ts deleted file mode 100644 index 553369e2b..000000000 --- a/packages/opencode/test/server/httpapi-event-diagnostics.test.ts +++ /dev/null @@ -1,279 +0,0 @@ -// Diagnostic suite for /event SSE delivery. -// -// Each test isolates ONE variable in the publisher chain while keeping the -// subscriber path constant (in-process HttpApi via Server.Default reading the -// SSE body). The pass/fail pattern across tests tells us where the bug lives: -// -// D1 (baseline): publish via Bus.use.publish — mirror of httpapi-event.test.ts -// test 3. Confirms /event SSE delivery works for SOME publish path. -// -// D2: publish N times in quick succession via Bus.use.publish. If the bus -// subscription is acquired correctly there should be no message loss. -// -// D3: publish via SyncEvent.use.run — exercises the same path the HTTP -// handlers use (Session.updatePart → sync.run → bus.publish) without -// the HTTP roundtrip. Tells us whether the sync path itself can deliver -// in-process. -// -// D4: publish via SyncEvent.use.run; subscriber is an in-process Bus -// callback. Confirms pub/sub identity end-to-end without /event SSE. -// -// D5: in-process Bus callback subscriber AND raw /event SSE subscriber -// receive the same publish. If both receive: no bug. If only the -// callback receives: the /event handler has an acquisition race. -// -// D6: same as D5 but the callback subscriber is attached AFTER /event SSE -// subscription is established. Order-of-setup variable. -import { afterEach, describe, expect } from "bun:test" -import { Deferred, Effect, Layer, Schema } from "effect" -import * as Log from "@opencode-ai/core/util/log" -import { Bus } from "../../src/bus" -import { Event as ServerEvent } from "../../src/server/event" -import { Server } from "../../src/server/server" -import { EventPaths } from "../../src/server/routes/instance/httpapi/groups/event" -import { MessageV2 } from "../../src/session/message-v2" -import { MessageID, PartID, SessionID } from "../../src/session/schema" -import { SyncEvent } from "../../src/sync" -import { resetDatabase } from "../fixture/db" -import { disposeAllInstances, TestInstance } from "../fixture/fixture" -import { testEffectShared } from "../lib/effect" - -void Log.init({ print: false }) - -const SseEvent = Schema.Struct({ - id: Schema.optional(Schema.String), - type: Schema.String, - properties: Schema.Record(Schema.String, Schema.Any), -}) - -type SseEvent = Schema.Schema.Type -type BusEvent = { type: string; properties: unknown } - -afterEach(async () => { - await disposeAllInstances() - await resetDatabase() -}) - -const it = testEffectShared(Layer.mergeAll(Bus.defaultLayer, SyncEvent.defaultLayer)) - -const publishConnected = Bus.use.publish(ServerEvent.Connected, {}) - -const publishPartUpdated = (partID: ReturnType) => { - const sessionID = SessionID.make(`ses_${Date.now().toString(36)}${Math.random().toString(36).slice(2, 8)}`) - return Bus.use.publish(MessageV2.Event.PartUpdated, { - sessionID, - part: { id: partID, sessionID, messageID: MessageID.ascending(), type: "text", text: "diag" }, - time: Date.now(), - }) -} - -const subscribeAllCallback = (handler: (event: BusEvent) => void) => - Effect.acquireRelease(Bus.use.subscribeAllCallback(handler), (dispose) => Effect.sync(() => dispose())) - -const openEventStream = (directory: string) => - Effect.gen(function* () { - const response = yield* Effect.promise(async () => - Server.Default().app.request(EventPaths.event, { headers: { "x-opencode-directory": directory } }), - ) - if (!response.body) return yield* Effect.die("missing SSE response body") - const reader = response.body.getReader() - yield* Effect.addFinalizer(() => Effect.promise(() => reader.cancel().catch(() => undefined))) - return reader - }) - -const decoder = new TextDecoder() - -function decodeFrame(value: Uint8Array): SseEvent[] { - return decoder - .decode(value) - .split(/\n\n+/) - .map((part) => part.trim()) - .filter((part) => part.length > 0) - .map((part) => Schema.decodeUnknownSync(SseEvent)(JSON.parse(part.replace(/^data: /, "")))) -} - -const readNextEvent = (reader: ReadableStreamDefaultReader) => - Effect.promise(() => reader.read()).pipe( - Effect.timeoutOrElse({ - duration: "3 seconds", - orElse: () => Effect.fail(new Error("timed out reading SSE chunk")), - }), - Effect.flatMap((result) => { - if (result.done || !result.value) return Effect.fail(new Error("event stream closed")) - const frames = decodeFrame(result.value) - if (frames.length === 0) return Effect.fail(new Error("empty SSE frame")) - return Effect.succeed(frames[0]!) - }), - ) - -const collectUntilEvent = (reader: ReadableStreamDefaultReader, predicate: (event: SseEvent) => boolean) => - Effect.gen(function* () { - const events: SseEvent[] = [] - while (true) { - const event = yield* readNextEvent(reader) - events.push(event) - if (predicate(event)) return events - } - }).pipe( - Effect.timeoutOrElse({ - duration: "4 seconds", - orElse: () => Effect.fail(new Error("collectUntil deadline exceeded")), - }), - ) - -const isPartUpdated = (event: { type: string }) => event.type === MessageV2.Event.PartUpdated.type - -describe("/event SSE delivery diagnostics", () => { - // Sanity: baseline same as httpapi-event.test.ts test 3 (already known to pass) - // but explicit about timing — publish happens with NO wait after reading - // server.connected. If this fails we have a deeper problem than just sync. - it.instance( - "D1: delivers a single bus event published right after server.connected", - () => - Effect.gen(function* () { - const { directory } = yield* TestInstance - const reader = yield* openEventStream(directory) - - expect((yield* readNextEvent(reader)).type).toBe("server.connected") - yield* publishConnected - expect((yield* readNextEvent(reader)).type).toBe("server.connected") - }), - { git: true, config: { formatter: false, lsp: false } }, - ) - - // If D1 passes but D2 fails, we have a queue-drain or partial-loss issue. - it.instance( - "D2: delivers all N bus events published in rapid succession", - () => - Effect.gen(function* () { - const { directory } = yield* TestInstance - const reader = yield* openEventStream(directory) - expect((yield* readNextEvent(reader)).type).toBe("server.connected") - - const N = 5 - yield* Effect.replicateEffect(publishConnected, N) - - const received = yield* Effect.replicateEffect(readNextEvent(reader), N) - expect(received).toHaveLength(N) - for (const event of received) expect(event.type).toBe("server.connected") - }), - { git: true, config: { formatter: false, lsp: false } }, - ) - - // The critical test. If D1 passes but this fails, the bus-identity fix is - // incomplete OR the sync.run publish path doesn't reach the same bus - // /event subscribes to, even when both share the memoMap. - it.instance( - "D3: delivers a SyncEvent published via SyncEvent.use.run after server.connected", - () => - Effect.gen(function* () { - const { directory } = yield* TestInstance - const reader = yield* openEventStream(directory) - expect((yield* readNextEvent(reader)).type).toBe("server.connected") - - const partID = PartID.ascending() - yield* publishPartUpdated(partID) - - const collected = yield* collectUntilEvent(reader, isPartUpdated) - const updated = collected.find(isPartUpdated) - expect(updated?.properties.part.id).toBe(partID) - }), - { git: true, config: { formatter: false, lsp: false } }, - ) - - // If D3 passes but D5 (the SDK E2E in httpapi-sdk.test.ts) fails, then the - // bug is specifically in the cross-request / cross-fiber HTTP path, not in - // the publish itself. If D3 also fails, the publish chain is broken. - // - // D4: ensure the publish reaches an in-process Bus subscriber too. Confirms - // pub/sub identity end-to-end without involving /event SSE. - it.instance( - "D4: SyncEvent.use.run publish reaches an in-process Bus callback", - () => - Effect.gen(function* () { - const received = yield* Deferred.make() - yield* subscribeAllCallback((event) => { - if (isPartUpdated(event)) Deferred.doneUnsafe(received, Effect.succeed(event)) - }) - - const partID = PartID.ascending() - yield* publishPartUpdated(partID) - - const event = yield* Deferred.await(received).pipe( - Effect.timeoutOrElse({ - duration: "3 seconds", - orElse: () => Effect.fail(new Error("D4 timed out waiting for callback")), - }), - ) - expect(event.type).toBe(MessageV2.Event.PartUpdated.type) - expect(event.properties).toMatchObject({ part: { id: partID } }) - }), - { git: true, config: { formatter: false, lsp: false } }, - ) - - // D5: BOTH subscribers attached simultaneously. Trigger ONE publish via - // SyncEvent.use.run. Both subscribers should receive it. If only one does - // we know exactly which side of the chain is failing. - it.instance( - "D5: same SyncEvent.use.run publish reaches BOTH /event SSE and in-process callback", - () => - Effect.gen(function* () { - const { directory } = yield* TestInstance - const callbackReceived = yield* Deferred.make() - yield* subscribeAllCallback((event) => { - if (isPartUpdated(event)) Deferred.doneUnsafe(callbackReceived, Effect.succeed(event)) - }) - const reader = yield* openEventStream(directory) - expect((yield* readNextEvent(reader)).type).toBe("server.connected") - - const partID = PartID.ascending() - yield* publishPartUpdated(partID) - - const sseSaw = yield* collectUntilEvent(reader, isPartUpdated).pipe( - Effect.map((events) => events.some(isPartUpdated)), - Effect.catch(() => Effect.succeed(false)), - ) - const callbackSaw = yield* Deferred.await(callbackReceived).pipe( - Effect.timeoutOrElse({ duration: "1 second", orElse: () => Effect.succeed(undefined) }), - Effect.map((event) => event !== undefined), - ) - - // Single assert with the boolean pair so the failure message tells us - // exactly which side broke. - expect({ sseSaw, callbackSaw }).toEqual({ sseSaw: true, callbackSaw: true }) - }), - { git: true, config: { formatter: false, lsp: false } }, - ) - - // D6: same as D5 but the callback subscriber is attached AFTER /event SSE - // subscription is established. If D5 fails and D6 passes, the order of - // subscriber setup is the determining factor. - it.instance( - "D6: /event SSE receives sync.run publish when callback is attached AFTER /event opens", - () => - Effect.gen(function* () { - const { directory } = yield* TestInstance - const reader = yield* openEventStream(directory) - expect((yield* readNextEvent(reader)).type).toBe("server.connected") - - const callbackReceived = yield* Deferred.make() - yield* subscribeAllCallback((event) => { - if (isPartUpdated(event)) Deferred.doneUnsafe(callbackReceived, Effect.succeed(event)) - }) - - const partID = PartID.ascending() - yield* publishPartUpdated(partID) - - const sseSaw = yield* collectUntilEvent(reader, isPartUpdated).pipe( - Effect.map((events) => events.some(isPartUpdated)), - Effect.catch(() => Effect.succeed(false)), - ) - const callbackSaw = yield* Deferred.await(callbackReceived).pipe( - Effect.timeoutOrElse({ duration: "1 second", orElse: () => Effect.succeed(undefined) }), - Effect.map((event) => event !== undefined), - ) - expect({ sseSaw, callbackSaw }).toEqual({ sseSaw: true, callbackSaw: true }) - }), - { git: true, config: { formatter: false, lsp: false } }, - ) -}) diff --git a/packages/opencode/test/server/httpapi-experimental.test.ts b/packages/opencode/test/server/httpapi-experimental.test.ts index 06f2ec554..36da905c5 100644 --- a/packages/opencode/test/server/httpapi-experimental.test.ts +++ b/packages/opencode/test/server/httpapi-experimental.test.ts @@ -6,7 +6,9 @@ import { Server } from "../../src/server/server" import { ExperimentalPaths } from "../../src/server/routes/instance/httpapi/groups/experimental" import { Session } from "@/session/session" import { SessionTable } from "@opencode-ai/core/session/sql" -import { Database } from "@/storage/db" +import { Database } from "@opencode-ai/core/database/database" +import { AccountV2 } from "@opencode-ai/core/account" +import { AccountTable } from "@opencode-ai/core/account/sql" import * as Log from "@opencode-ai/core/util/log" import { Worktree } from "../../src/worktree" import { resetDatabase } from "../fixture/db" @@ -15,7 +17,7 @@ import { testEffect } from "../lib/effect" void Log.init({ print: false }) -const it = testEffect(Layer.mergeAll(Session.defaultLayer)) +const it = testEffect(Layer.mergeAll(Session.defaultLayer, Database.defaultLayer)) const testWorktreeMutations = process.platform === "win32" ? it.instance.skip : it.instance function app() { @@ -62,34 +64,34 @@ function waitReady(input: { directory?: string; name?: string }) { function insertAccount() { return Effect.acquireRelease( - Effect.sync(() => { - Database.Client() - .$client.prepare( - "INSERT INTO account (id, email, url, access_token, refresh_token, time_created, time_updated) VALUES (?, ?, ?, ?, ?, ?, ?)", - ) - .run( - "account-test", - "test@example.com", - "https://console.example.com", - "access", - "refresh", - Date.now(), - Date.now(), - ) + Effect.gen(function* () { + const { db } = yield* Database.Service + yield* db + .insert(AccountTable) + .values({ + id: AccountV2.ID.make("account-test"), + email: "test@example.com", + url: "https://console.example.com", + access_token: AccountV2.AccessToken.make("access"), + refresh_token: AccountV2.RefreshToken.make("refresh"), + time_created: Date.now(), + time_updated: Date.now(), + }) + .run() + .pipe(Effect.orDie) return "account-test" }), (id) => - Effect.sync(() => { - Database.Client().$client.prepare("DELETE FROM account WHERE id = ?").run(id) - }), + Database.Service.use(({ db }) => + db.delete(AccountTable).where(eq(AccountTable.id, AccountV2.ID.make(id))).run().pipe(Effect.orDie), + ), ) } function setSessionUpdated(session: Session.Info, updated: number) { - return Effect.sync(() => { - Database.use((db) => - db.update(SessionTable).set({ time_updated: updated }).where(eq(SessionTable.id, session.id)).run(), - ) + return Effect.gen(function* () { + const { db } = yield* Database.Service + yield* db.update(SessionTable).set({ time_updated: updated }).where(eq(SessionTable.id, session.id)).run().pipe(Effect.orDie) }) } diff --git a/packages/opencode/test/server/httpapi-instance.test.ts b/packages/opencode/test/server/httpapi-instance.test.ts index 456f2ca1a..65bdfa7c5 100644 --- a/packages/opencode/test/server/httpapi-instance.test.ts +++ b/packages/opencode/test/server/httpapi-instance.test.ts @@ -17,7 +17,7 @@ import { resetDatabase } from "../fixture/db" import { tmpdirScoped } from "../fixture/fixture" import { testEffect } from "../lib/effect" -// Flip the experimental workspaces flag so SyncEvent.run actually writes to +// Flip the experimental workspaces flag so EventV2.run actually writes to // EventSequenceTable (the source of truth the fence middleware reads). Reset // the database around the test so per-instance state does not leak between // runs. resetDatabase() already calls disposeAllInstances(), so we don't diff --git a/packages/opencode/test/server/httpapi-schema-error-body.test.ts b/packages/opencode/test/server/httpapi-schema-error-body.test.ts index 744ce8a68..e8052451e 100644 --- a/packages/opencode/test/server/httpapi-schema-error-body.test.ts +++ b/packages/opencode/test/server/httpapi-schema-error-body.test.ts @@ -1,7 +1,7 @@ import { afterEach, describe, expect } from "bun:test" -import { Effect } from "effect" +import { Effect, Layer } from "effect" import { eq } from "drizzle-orm" -import * as Database from "@/storage/db" +import { Database } from "@opencode-ai/core/database/database" import { Server } from "../../src/server/server" import { Session } from "@/session/session" @@ -14,7 +14,7 @@ import { disposeAllInstances, TestInstance } from "../fixture/fixture" import { testEffect } from "../lib/effect" import { ProviderV2 } from "@opencode-ai/core/provider" -const it = testEffect(Session.defaultLayer) +const it = testEffect(Layer.mergeAll(Session.defaultLayer, Database.defaultLayer)) afterEach(async () => { await disposeAllInstances() @@ -44,22 +44,20 @@ const seedCorruptStepFinishPart = Effect.gen(function* () { }) // Schema.Finite still rejects NaN at encode: exact mirror of the corrupt row // that broke the user's session in the OMO/Windows bug. - yield* Effect.sync(() => - Database.use((db) => - db - .update(PartTable) - .set({ - data: { - type: "step-finish", - reason: "stop", - cost: 0, - tokens: { input: 0, output: NaN, reasoning: 0, cache: { read: 0, write: 0 } }, - } as never, // drizzle's .set() can't narrow the discriminated union - }) - .where(eq(PartTable.id, partID)) - .run(), - ), - ) + const { db } = yield* Database.Service + yield* db + .update(PartTable) + .set({ + data: { + type: "step-finish", + reason: "stop", + cost: 0, + tokens: { input: 0, output: NaN, reasoning: 0, cache: { read: 0, write: 0 } }, + } as never, // drizzle's .set() can't narrow the discriminated union + }) + .where(eq(PartTable.id, partID)) + .run() + .pipe(Effect.orDie) return info.id }) diff --git a/packages/opencode/test/server/httpapi-sdk.test.ts b/packages/opencode/test/server/httpapi-sdk.test.ts index 644dcfa01..13bc05d87 100644 --- a/packages/opencode/test/server/httpapi-sdk.test.ts +++ b/packages/opencode/test/server/httpapi-sdk.test.ts @@ -642,7 +642,7 @@ describe("HttpApi SDK", () => { ), ) - // Regression: SyncEvent must publish on the same ProjectBus the /event handler + // Regression: EventV2 must publish on the same ProjectBus the /event handler // subscribes to, AND the /event stream must forward handler ALS/context into the // body-pump fiber. Drives the full SDK → /event → Session.updatePart → sync.run → // bus.publish → SDK subscriber path. Goes red if either the publisher uses a diff --git a/packages/opencode/test/server/httpapi-session.test.ts b/packages/opencode/test/server/httpapi-session.test.ts index 5e58d436a..71b79c1bc 100644 --- a/packages/opencode/test/server/httpapi-session.test.ts +++ b/packages/opencode/test/server/httpapi-session.test.ts @@ -19,7 +19,7 @@ import { SessionPaths } from "../../src/server/routes/instance/httpapi/groups/se import { Session } from "@/session/session" import { MessageID, PartID, SessionID, type SessionID as SessionIDType } from "../../src/session/schema" import { MessageV2 } from "../../src/session/message-v2" -import { Database } from "@/storage/db" +import { Database } from "@opencode-ai/core/database/database" import { SessionMessageTable, SessionTable } from "@opencode-ai/core/session/sql" import { SessionMessage } from "@opencode-ai/core/session/message" import { ModelV2 } from "@opencode-ai/core/model" @@ -43,7 +43,9 @@ const instanceStoreLayer = InstanceStore.defaultLayer.pipe( Layer.succeed(InstanceBootstrapService.Service, InstanceBootstrapService.Service.of({ run: Effect.void })), ), ) -const it = testEffect(Layer.mergeAll(instanceStoreLayer, Project.defaultLayer, Session.defaultLayer, workspaceLayer)) +const it = testEffect( + Layer.mergeAll(instanceStoreLayer, Project.defaultLayer, Session.defaultLayer, workspaceLayer, Database.defaultLayer), +) function app() { return Server.Default().app @@ -107,7 +109,7 @@ const createLocalWorkspace = (input: { projectID: Project.Info["id"]; type: stri ) const insertLegacyAssistantMessage = (sessionID: SessionIDType, time = 1) => - Effect.sync(() => { + Effect.gen(function* () { const message = new SessionMessage.Assistant({ id: SessionMessage.ID.create(), type: "assistant", @@ -120,76 +122,72 @@ const insertLegacyAssistantMessage = (sessionID: SessionIDType, time = 1) => time: { created: DateTime.makeUnsafe(time) }, content: [], }) - Database.use((db) => - db - .insert(SessionMessageTable) - .values([ - { - id: message.id, - session_id: sessionID, - type: message.type, - time_created: time, - data: { - time: { created: time }, - agent: message.agent, - model: message.model, - content: message.content, - } as NonNullable<(typeof SessionMessageTable.$inferInsert)["data"]>, - }, - ]) - .run(), - ) + const { db } = yield* Database.Service + yield* db + .insert(SessionMessageTable) + .values([ + { + id: message.id, + session_id: sessionID, + type: message.type, + time_created: time, + data: { + time: { created: time }, + agent: message.agent, + model: message.model, + content: message.content, + } as NonNullable<(typeof SessionMessageTable.$inferInsert)["data"]>, + }, + ]) + .run() + .pipe(Effect.orDie) }) const insertCorruptV2Message = (sessionID: SessionIDType, time = 1) => - Effect.sync(() => - Database.use((db) => - db - .insert(SessionMessageTable) - .values([ - { - id: SessionMessage.ID.create(), - session_id: sessionID, - type: "assistant", - time_created: time, - data: {} as NonNullable<(typeof SessionMessageTable.$inferInsert)["data"]>, - }, - ]) - .run(), - ), - ) + Effect.gen(function* () { + const { db } = yield* Database.Service + yield* db + .insert(SessionMessageTable) + .values([ + { + id: SessionMessage.ID.create(), + session_id: sessionID, + type: "assistant", + time_created: time, + data: {} as NonNullable<(typeof SessionMessageTable.$inferInsert)["data"]>, + }, + ]) + .run() + .pipe(Effect.orDie) + }) const setLegacySummaryDiff = (sessionID: SessionIDType) => - Effect.sync(() => - Database.use((db) => - db - .update(SessionTable) - .set({ - summary_additions: 1, - summary_deletions: 0, - summary_files: 1, - summary_diffs: [{ additions: 1, deletions: 0 }], - }) - .where(eq(SessionTable.id, sessionID)) - .run(), - ), - ) + Effect.gen(function* () { + const { db } = yield* Database.Service + yield* db + .update(SessionTable) + .set({ + summary_additions: 1, + summary_deletions: 0, + summary_files: 1, + summary_diffs: [{ additions: 1, deletions: 0 }], + }) + .where(eq(SessionTable.id, sessionID)) + .run() + .pipe(Effect.orDie) + }) const getWorkspaceID = (sessionID: SessionIDType) => - Effect.sync(() => - Database.use((db) => - db - .select({ workspaceID: SessionTable.workspace_id }) - .from(SessionTable) - .where(eq(SessionTable.id, sessionID)) - .get(), - ), - ) + Effect.gen(function* () { + const { db } = yield* Database.Service + return yield* db.select({ workspaceID: SessionTable.workspace_id }).from(SessionTable).where(eq(SessionTable.id, sessionID)).get().pipe(Effect.orDie) + }) const clearSessionPath = (sessionID: SessionIDType) => - Effect.sync(() => - Database.use((db) => db.update(SessionTable).set({ path: null }).where(eq(SessionTable.id, sessionID)).run()), - ) + Effect.gen(function* () { + const { db } = yield* Database.Service + yield* db.update(SessionTable).set({ path: null }).where(eq(SessionTable.id, sessionID)).run().pipe(Effect.orDie) + }) function request(path: string, init?: RequestInit) { return Effect.promise(async () => app().request(path, init)) diff --git a/packages/opencode/test/server/httpapi-workspace-routing.test.ts b/packages/opencode/test/server/httpapi-workspace-routing.test.ts index 5265b9604..642940eb0 100644 --- a/packages/opencode/test/server/httpapi-workspace-routing.test.ts +++ b/packages/opencode/test/server/httpapi-workspace-routing.test.ts @@ -19,6 +19,7 @@ import { WorkspaceV2 } from "@opencode-ai/core/workspace" import type { WorkspaceAdapter } from "../../src/control-plane/types" import { Workspace } from "../../src/control-plane/workspace" import { WorkspaceTable } from "@opencode-ai/core/control-plane/workspace.sql" +import { Database } from "@opencode-ai/core/database/database" import { Project } from "../../src/project/project" import { WorkspacePaths } from "../../src/server/routes/instance/httpapi/groups/workspace" import { @@ -26,7 +27,6 @@ import { workspaceRouterMiddleware, } from "../../src/server/routes/instance/httpapi/middleware/workspace-routing" import { HEADER as FenceHeader } from "../../src/server/shared/fence" -import { Database } from "../../src/storage/db" import { resetDatabase } from "../fixture/db" import { workspaceLayerWithRuntimeFlags } from "../fixture/workspace" import { tmpdirScoped } from "../fixture/fixture" @@ -50,6 +50,7 @@ const it = testEffect( testStateLayer, NodeHttpServer.layerTest, NodeServices.layer, + Database.defaultLayer, Project.defaultLayer, workspaceLayer, Socket.layerWebSocketConstructorGlobal, @@ -160,10 +161,11 @@ const insertRemoteWorkspaceWithoutSync = (input: { type: string url: string }) => - Effect.sync(() => { + Effect.gen(function* () { const id = WorkspaceV2.ID.ascending() registerAdapter(input.projectID, input.type, remoteAdapter(path.join(input.dir, `.${input.type}`), input.url)) - Database.use((db) => db.insert(WorkspaceTable).values({ id, type: input.type, project_id: input.projectID }).run()) + const { db } = yield* Database.Service + yield* db.insert(WorkspaceTable).values({ id, type: input.type, project_id: input.projectID }).run().pipe(Effect.orDie) return id }) diff --git a/packages/opencode/test/server/negative-tokens-regression.test.ts b/packages/opencode/test/server/negative-tokens-regression.test.ts index 4098f98d8..64a0a9c93 100644 --- a/packages/opencode/test/server/negative-tokens-regression.test.ts +++ b/packages/opencode/test/server/negative-tokens-regression.test.ts @@ -6,21 +6,21 @@ // strict `NonNegativeInt` schema then made every load of the message list // fail to encode, killing Desktop boot for every user with such a row. import { describe, expect } from "bun:test" -import { Effect } from "effect" +import { Effect, Layer } from "effect" import { eq } from "drizzle-orm" import { Server } from "../../src/server/server" import { SessionPaths } from "../../src/server/routes/instance/httpapi/groups/session" import { Session } from "@/session/session" import { MessageID, PartID } from "../../src/session/schema" -import * as Database from "@/storage/db" +import { Database } from "@opencode-ai/core/database/database" import { PartTable } from "@opencode-ai/core/session/sql" import { resetDatabase } from "../fixture/db" import { TestInstance } from "../fixture/fixture" import { testEffect } from "../lib/effect" import { ProviderV2 } from "@opencode-ai/core/provider" -const it = testEffect(Session.defaultLayer) +const it = testEffect(Layer.mergeAll(Session.defaultLayer, Database.defaultLayer)) function seedNegativeTokenSession() { return Effect.gen(function* () { @@ -47,20 +47,20 @@ function seedNegativeTokenSession() { // Bypass the schema with a direct SQL update to install the // negative `output` value we want to test loading. - Database.use((db) => - db - .update(PartTable) - .set({ - data: { - type: "step-finish", - reason: "stop", - cost: 0, - tokens: { input: 0, output: -42, reasoning: 0, cache: { read: 0, write: 0 } }, - } as never, - }) - .where(eq(PartTable.id, partID)) - .run(), - ) + const { db } = yield* Database.Service + yield* db + .update(PartTable) + .set({ + data: { + type: "step-finish", + reason: "stop", + cost: 0, + tokens: { input: 0, output: -42, reasoning: 0, cache: { read: 0, write: 0 } }, + } as never, + }) + .where(eq(PartTable.id, partID)) + .run() + .pipe(Effect.orDie) return info.id }) diff --git a/packages/opencode/test/server/session-list.test.ts b/packages/opencode/test/server/session-list.test.ts index 1498bbd5c..0b7de6bdf 100644 --- a/packages/opencode/test/server/session-list.test.ts +++ b/packages/opencode/test/server/session-list.test.ts @@ -1,7 +1,6 @@ import { afterEach, describe, expect } from "bun:test" import { Effect, Layer } from "effect" import { Database } from "@opencode-ai/core/database/database" -import { EventV2 } from "@opencode-ai/core/event" import { SessionProjector } from "@opencode-ai/core/session/projector" import { Session as SessionNs } from "@/session/session" import * as Log from "@opencode-ai/core/util/log" @@ -13,9 +12,9 @@ import { eq } from "drizzle-orm" import { testEffect } from "../lib/effect" import { Bus } from "@/bus" import { Storage } from "@/storage/storage" -import { SyncEvent } from "@/sync" import { RuntimeFlags } from "@/effect/runtime-flags" import { BackgroundJob } from "@/background/job" +import { EventV2Bridge } from "@/event-v2-bridge" void Log.init({ print: false }) const it = testEffect( @@ -24,9 +23,8 @@ const it = testEffect( SessionNs.layer.pipe( Layer.provide(Bus.layer), Layer.provide(Storage.defaultLayer), - Layer.provide(SyncEvent.defaultLayer), Layer.provide(Database.defaultLayer), - Layer.provide(EventV2.defaultLayer), + Layer.provide(EventV2Bridge.defaultLayer), Layer.provide(SessionProjector.defaultLayer), Layer.provide(RuntimeFlags.layer({ experimentalWorkspaces: false })), Layer.provide(BackgroundJob.defaultLayer), diff --git a/packages/opencode/test/session/compaction.test.ts b/packages/opencode/test/session/compaction.test.ts index 3b381d18c..f13705f72 100644 --- a/packages/opencode/test/session/compaction.test.ts +++ b/packages/opencode/test/session/compaction.test.ts @@ -1,6 +1,7 @@ import { afterEach, describe, expect, mock, test } from "bun:test" import { SessionLegacy } from "@opencode-ai/core/session/legacy" import { Database } from "@opencode-ai/core/database/database" +import { EventV2Bridge } from "@/event-v2-bridge" import { APICallError } from "ai" import { Cause, Deferred, Effect, Exit, Fiber, Layer, Schema } from "effect" import * as Stream from "effect/Stream" @@ -29,9 +30,7 @@ import { ProviderTest } from "../fake/provider" import { testEffect } from "../lib/effect" import { CrossSpawnSpawner } from "@opencode-ai/core/cross-spawn-spawner" import { TestConfig } from "../fixture/config" -import { SyncEvent } from "@/sync" import { RuntimeFlags } from "@/effect/runtime-flags" -import { EventV2Bridge } from "@/event-v2-bridge" import { LLMEvent, Usage } from "@opencode-ai/llm" import { ProviderV2 } from "@opencode-ai/core/provider" @@ -234,22 +233,22 @@ const deps = Layer.mergeAll( Plugin.defaultLayer, Bus.layer, Config.defaultLayer, - SyncEvent.defaultLayer, RuntimeFlags.layer({ experimentalEventSystem: true }), - EventV2Bridge.defaultLayer, Database.defaultLayer, + EventV2Bridge.defaultLayer, ) const env = Layer.mergeAll( SessionNs.defaultLayer, Database.defaultLayer, + EventV2Bridge.defaultLayer, CrossSpawnSpawner.defaultLayer, SessionCompaction.layer.pipe(Layer.provide(SessionNs.defaultLayer), Layer.provideMerge(deps)), ) const it = testEffect(env) -const compactionEnv = Layer.mergeAll(SessionNs.defaultLayer, Database.defaultLayer, CrossSpawnSpawner.defaultLayer) +const compactionEnv = Layer.mergeAll(SessionNs.defaultLayer, Database.defaultLayer, EventV2Bridge.defaultLayer, CrossSpawnSpawner.defaultLayer) const itCompaction = testEffect(compactionEnv) type CompactionProcessOptions = { @@ -286,7 +285,6 @@ function compactionProcessLayer(options?: CompactionProcessOptions) { Layer.provide(status), Layer.provide(bus), Layer.provide(options?.config ?? Config.defaultLayer), - Layer.provide(SyncEvent.defaultLayer), Layer.provide(RuntimeFlags.layer({ experimentalEventSystem: true })), Layer.provide(EventV2Bridge.defaultLayer), ) diff --git a/packages/opencode/test/session/processor-effect.test.ts b/packages/opencode/test/session/processor-effect.test.ts index 4375c6aaa..88aea01e8 100644 --- a/packages/opencode/test/session/processor-effect.test.ts +++ b/packages/opencode/test/session/processor-effect.test.ts @@ -1,6 +1,7 @@ import { NodeFileSystem } from "@effect/platform-node" import { SessionLegacy } from "@opencode-ai/core/session/legacy" import { Database } from "@opencode-ai/core/database/database" +import { EventV2Bridge } from "@/event-v2-bridge" import { expect } from "bun:test" import { tool } from "ai" import { Cause, Effect, Exit, Fiber, Layer } from "effect" @@ -28,9 +29,7 @@ import { CrossSpawnSpawner } from "@opencode-ai/core/cross-spawn-spawner" import { provideTmpdirServer } from "../fixture/fixture" import { testEffect } from "../lib/effect" import { raw, reply, TestLLMServer } from "../lib/llm-server" -import { SyncEvent } from "@/sync" import { RuntimeFlags } from "@/effect/runtime-flags" -import { EventV2Bridge } from "@/event-v2-bridge" import { ProviderV2 } from "@opencode-ai/core/provider" void Log.init({ print: false }) @@ -185,9 +184,8 @@ const deps = Layer.mergeAll( LLM.defaultLayer, Provider.defaultLayer, status, - SyncEvent.defaultLayer, - EventV2Bridge.defaultLayer, Database.defaultLayer, + EventV2Bridge.defaultLayer, ).pipe(Layer.provideMerge(infra)) const env = Layer.mergeAll( TestLLMServer.layer, diff --git a/packages/opencode/test/session/prompt.test.ts b/packages/opencode/test/session/prompt.test.ts index 89573053f..9eba7c4f6 100644 --- a/packages/opencode/test/session/prompt.test.ts +++ b/packages/opencode/test/session/prompt.test.ts @@ -1,6 +1,8 @@ import { NodeFileSystem } from "@effect/platform-node" import { SessionLegacy } from "@opencode-ai/core/session/legacy" -import { Database as CoreDatabase } from "@opencode-ai/core/database/database" +import { Database } from "@opencode-ai/core/database/database" +import { eq } from "drizzle-orm" +import { EventV2Bridge } from "@/event-v2-bridge" import { FetchHttpClient } from "effect/unstable/http" import { expect } from "bun:test" import { Cause, Deferred, Duration, Effect, Exit, Fiber, Layer } from "effect" @@ -46,7 +48,6 @@ import { ToolRegistry } from "@/tool/registry" import { Truncate } from "@/tool/truncate" import * as Log from "@opencode-ai/core/util/log" import { CrossSpawnSpawner } from "@opencode-ai/core/cross-spawn-spawner" -import * as Database from "../../src/storage/db" import { Ripgrep } from "../../src/file/ripgrep" import { Format } from "../../src/format" import { Reference } from "../../src/reference/reference" @@ -54,9 +55,7 @@ import { RepositoryCache } from "../../src/reference/repository-cache" import { TestInstance } from "../fixture/fixture" import { awaitWithTimeout, pollWithTimeout, testEffect } from "../lib/effect" import { reply, TestLLMServer } from "../lib/llm-server" -import { SyncEvent } from "@/sync" import { RuntimeFlags } from "@/effect/runtime-flags" -import { EventV2Bridge } from "@/event-v2-bridge" import { ProviderV2 } from "@opencode-ai/core/provider" void Log.init({ print: false }) @@ -184,9 +183,8 @@ function makePrompt(input?: { processor?: "blocking" }) { AppFileSystem.defaultLayer, BackgroundJob.defaultLayer, status, - SyncEvent.defaultLayer, + Database.defaultLayer, EventV2Bridge.defaultLayer, - CoreDatabase.defaultLayer, ).pipe(Layer.provideMerge(infra)) const question = Question.layer.pipe(Layer.provideMerge(deps)) const todo = Todo.layer.pipe(Layer.provideMerge(deps)) @@ -512,9 +510,8 @@ noLLMServer.instance( const messages = yield* SessionV2.Service.use((session) => session.messages({ sessionID: chat.id })).pipe( Effect.provide(SessionV2.defaultLayer), ) - const row = Database.use((db) => - db.select().from(SessionMessageTable).where(Database.eq(SessionMessageTable.session_id, chat.id)).get(), - ) + const { db } = yield* Database.Service + const row = yield* db.select().from(SessionMessageTable).where(eq(SessionMessageTable.session_id, chat.id)).get().pipe(Effect.orDie) expect(messages.find((message) => message.type === "user")).toMatchObject({ type: "user", text: "hello v2" }) expect(typeof row?.data.time.created).toBe("number") expect(messages).toEqual( diff --git a/packages/opencode/test/session/session.test.ts b/packages/opencode/test/session/session.test.ts index f78eb5fb8..932b70d77 100644 --- a/packages/opencode/test/session/session.test.ts +++ b/packages/opencode/test/session/session.test.ts @@ -1,7 +1,6 @@ import { describe, expect } from "bun:test" import { SessionLegacy } from "@opencode-ai/core/session/legacy" import { Database } from "@opencode-ai/core/database/database" -import { EventV2 } from "@opencode-ai/core/event" import { SessionProjector } from "@opencode-ai/core/session/projector" import { Deferred, Effect, Exit, Layer } from "effect" import { Session as SessionNs } from "@/session/session" @@ -14,9 +13,9 @@ import { provideInstance, tmpdirScoped } from "../fixture/fixture" import { testEffect } from "../lib/effect" import { Bus } from "@/bus" import { Storage } from "@/storage/storage" -import { SyncEvent } from "@/sync" import { RuntimeFlags } from "@/effect/runtime-flags" import { BackgroundJob } from "@/background/job" +import { EventV2Bridge } from "@/event-v2-bridge" void Log.init({ print: false }) @@ -25,9 +24,8 @@ const it = testEffect( SessionNs.layer.pipe( Layer.provide(Bus.layer), Layer.provide(Storage.defaultLayer), - Layer.provide(SyncEvent.defaultLayer), Layer.provide(Database.defaultLayer), - Layer.provide(EventV2.defaultLayer), + Layer.provide(EventV2Bridge.defaultLayer), Layer.provide(SessionProjector.defaultLayer), Layer.provide(RuntimeFlags.layer({ experimentalWorkspaces: false })), Layer.provide(BackgroundJob.defaultLayer), diff --git a/packages/opencode/test/session/snapshot-tool-race.test.ts b/packages/opencode/test/session/snapshot-tool-race.test.ts index 18046fbe5..6c40b60d7 100644 --- a/packages/opencode/test/session/snapshot-tool-race.test.ts +++ b/packages/opencode/test/session/snapshot-tool-race.test.ts @@ -31,6 +31,7 @@ import { TestLLMServer } from "../lib/llm-server" // Same layer setup as prompt-effect.test.ts import { NodeFileSystem } from "@effect/platform-node" import { Database } from "@opencode-ai/core/database/database" +import { EventV2Bridge } from "@/event-v2-bridge" import { Agent as AgentSvc } from "../../src/agent/agent" import { BackgroundJob } from "@/background/job" import { Git } from "../../src/git" @@ -62,9 +63,7 @@ import { Ripgrep } from "../../src/file/ripgrep" import { Format } from "../../src/format" import { Reference } from "../../src/reference/reference" import { RepositoryCache } from "../../src/reference/repository-cache" -import { SyncEvent } from "@/sync" import { RuntimeFlags } from "@/effect/runtime-flags" -import { EventV2Bridge } from "@/event-v2-bridge" void Log.init({ print: false }) @@ -132,9 +131,8 @@ function makeHttp() { AppFileSystem.defaultLayer, BackgroundJob.defaultLayer, status, - SyncEvent.defaultLayer, - EventV2Bridge.defaultLayer, Database.defaultLayer, + EventV2Bridge.defaultLayer, ).pipe(Layer.provideMerge(infra)) const question = Question.layer.pipe(Layer.provideMerge(deps)) const todo = Todo.layer.pipe(Layer.provideMerge(deps)) diff --git a/packages/opencode/test/storage/db.test.ts b/packages/opencode/test/storage/db.test.ts deleted file mode 100644 index c25a29a92..000000000 --- a/packages/opencode/test/storage/db.test.ts +++ /dev/null @@ -1,9 +0,0 @@ -import { describe, expect, it } from "bun:test" -import { Database as CoreDatabase } from "@opencode-ai/core/database/database" -import { Database } from "@/storage/db" - -describe("Database.getPath", () => { - it("delegates to the core database path", () => { - expect(Database.getPath()).toBe(CoreDatabase.path()) - }) -}) diff --git a/packages/opencode/test/sync/index.test.ts b/packages/opencode/test/sync/index.test.ts deleted file mode 100644 index 46de313cb..000000000 --- a/packages/opencode/test/sync/index.test.ts +++ /dev/null @@ -1,391 +0,0 @@ -import { describe, expect, beforeEach, afterAll } from "bun:test" -import { provideTmpdirInstance } from "../fixture/fixture" -import { Deferred, Effect, Layer, Schema } from "effect" -import { CrossSpawnSpawner } from "@opencode-ai/core/cross-spawn-spawner" -import { Bus } from "../../src/bus" -import { GlobalBus, type GlobalEvent } from "../../src/bus/global" -import { SyncEvent } from "../../src/sync" -import { Database, eq } from "@/storage/db" -import { EventSequenceTable, EventTable } from "@opencode-ai/core/event/sql" -import { EventV2 } from "@opencode-ai/core/event" -import { MessageID } from "../../src/session/schema" -import { initProjectors } from "../../src/server/projectors" -import { awaitWithTimeout, testEffect } from "../lib/effect" -import { RuntimeFlags } from "@/effect/runtime-flags" - -const it = testEffect( - Layer.mergeAll( - SyncEvent.layer.pipe( - Layer.provide(RuntimeFlags.layer({ experimentalWorkspaces: true })), - Layer.provideMerge(Bus.layer), - ), - CrossSpawnSpawner.defaultLayer, - ), -) - -beforeEach(() => { - Database.close() -}) - -describe("SyncEvent", () => { - function setup() { - SyncEvent.reset() - - const Created = SyncEvent.define({ - type: "item.created", - version: 1, - aggregate: "id", - schema: Schema.Struct({ id: Schema.String, name: Schema.String }), - }) - const Sent = SyncEvent.define({ - type: "item.sent", - version: 1, - aggregate: "item_id", - schema: Schema.Struct({ item_id: Schema.String, to: Schema.String }), - }) - - SyncEvent.init({ - projectors: [SyncEvent.project(Created, () => {}), SyncEvent.project(Sent, () => {})], - }) - - return { Created, Sent } - } - - function expectDefect(effect: Effect.Effect, pattern: RegExp) { - return Effect.gen(function* () { - const exit = yield* Effect.exit(effect) - if (exit._tag === "Success") throw new Error("Expected effect to fail") - expect(String(exit.cause)).toMatch(pattern) - }) - } - - afterAll(() => { - SyncEvent.reset() - initProjectors() - }) - - describe("run", () => { - it.live( - "inserts event row", - provideTmpdirInstance(() => - Effect.gen(function* () { - const { Created } = setup() - yield* SyncEvent.use.run(Created, { id: "evt_1", name: "first" }) - const rows = Database.use((db) => db.select().from(EventTable).all()) - expect(rows).toHaveLength(1) - expect(rows[0].type).toBe("item.created.1") - expect(rows[0].aggregate_id).toBe("evt_1") - }), - ), - ) - - it.live( - "increments seq per aggregate", - provideTmpdirInstance(() => - Effect.gen(function* () { - const { Created } = setup() - yield* SyncEvent.use.run(Created, { id: "evt_1", name: "first" }) - yield* SyncEvent.use.run(Created, { id: "evt_1", name: "second" }) - const rows = Database.use((db) => db.select().from(EventTable).all()) - expect(rows).toHaveLength(2) - expect(rows[1].seq).toBe(rows[0].seq + 1) - }), - ), - ) - - it.live( - "uses custom aggregate field from agg()", - provideTmpdirInstance(() => - Effect.gen(function* () { - const { Sent } = setup() - yield* SyncEvent.use.run(Sent, { item_id: "evt_1", to: "james" }) - const rows = Database.use((db) => db.select().from(EventTable).all()) - expect(rows).toHaveLength(1) - expect(rows[0].aggregate_id).toBe("evt_1") - }), - ), - ) - - it.live( - "emits events", - provideTmpdirInstance(() => - Effect.gen(function* () { - const { Created } = setup() - const events: Array<{ - type: string - properties: { id: string; name: string } - }> = [] - let resolve = () => {} - const received = new Promise((done) => { - resolve = done - }) - const bus = yield* Bus.Service - const dispose = yield* bus.subscribeAllCallback((event) => { - events.push(event) - resolve() - }) - try { - yield* SyncEvent.use.run(Created, { id: "evt_1", name: "test" }) - yield* Effect.promise(() => received) - expect(events).toHaveLength(1) - expect(events[0]).toMatchObject({ - type: "item.created", - properties: { - id: "evt_1", - name: "test", - }, - }) - } finally { - dispose() - } - }), - ), - ) - - // Regression for the EffectBridge migration. GlobalBus.emit used to fire - // synchronously inside the Database.effect post-commit callback. After the - // migration it fires inside the forked publish Effect, AFTER bus.publish - // completes. Consumers don't care about microsecond-level ordering, but - // we still need to prove the emit actually fires. - it.live( - "emits sync events to GlobalBus after publishing to ProjectBus", - provideTmpdirInstance(() => - Effect.gen(function* () { - const { Created } = setup() - // Filter for OUR specific event in the handler so we ignore any - // stray sync events from other tests' lingering forks. - const received = yield* Deferred.make() - const handler = (evt: GlobalEvent) => { - if (evt.payload?.type === "sync" && evt.payload?.syncEvent?.type === "item.created.1") { - Deferred.doneUnsafe(received, Effect.succeed(evt)) - } - } - GlobalBus.on("event", handler) - try { - yield* SyncEvent.use.run(Created, { id: "evt_global_1", name: "global" }) - const event = yield* awaitWithTimeout( - Deferred.await(received), - "timed out waiting for sync event on GlobalBus", - "2 seconds", - ) - expect(event.payload).toMatchObject({ - type: "sync", - syncEvent: { type: "item.created.1", data: { id: "evt_global_1", name: "global" } }, - }) - } finally { - GlobalBus.off("event", handler) - } - }), - ), - ) - }) - - describe("replay", () => { - it.live( - "inserts event from external payload", - provideTmpdirInstance(() => - Effect.gen(function* () { - const id = MessageID.ascending() - yield* SyncEvent.use.replay({ - id: "evt_1", - type: "item.created.1", - seq: 0, - aggregateID: id, - data: { id, name: "replayed" }, - }) - const rows = Database.use((db) => db.select().from(EventTable).all()) - expect(rows).toHaveLength(1) - expect(rows[0].aggregate_id).toBe(id) - }), - ), - ) - - it.live( - "throws on sequence mismatch", - provideTmpdirInstance(() => - Effect.gen(function* () { - const id = MessageID.ascending() - yield* SyncEvent.use.replay({ - id: "evt_1", - type: "item.created.1", - seq: 0, - aggregateID: id, - data: { id, name: "first" }, - }) - yield* expectDefect( - SyncEvent.use.replay({ - id: "evt_1", - type: "item.created.1", - seq: 5, - aggregateID: id, - data: { id, name: "bad" }, - }), - /Sequence mismatch/, - ) - }), - ), - ) - - it.live( - "throws on unknown event type", - provideTmpdirInstance(() => - Effect.gen(function* () { - yield* expectDefect( - SyncEvent.use.replay({ - id: "evt_1", - type: "unknown.event.1", - seq: 0, - aggregateID: "x", - data: {}, - }), - /Unknown event type/, - ) - }), - ), - ) - - it.live( - "replayAll accepts later chunks after the first batch", - provideTmpdirInstance(() => - Effect.gen(function* () { - const { Created } = setup() - const id = MessageID.ascending() - - const one = yield* SyncEvent.use.replayAll([ - { - id: "evt_1", - type: SyncEvent.versionedType(Created.type, Created.version), - seq: 0, - aggregateID: id, - data: { id, name: "first" }, - }, - { - id: "evt_2", - type: SyncEvent.versionedType(Created.type, Created.version), - seq: 1, - aggregateID: id, - data: { id, name: "second" }, - }, - ]) - - const two = yield* SyncEvent.use.replayAll([ - { - id: "evt_3", - type: SyncEvent.versionedType(Created.type, Created.version), - seq: 2, - aggregateID: id, - data: { id, name: "third" }, - }, - { - id: "evt_4", - type: SyncEvent.versionedType(Created.type, Created.version), - seq: 3, - aggregateID: id, - data: { id, name: "fourth" }, - }, - ]) - - expect(one).toBe(id) - expect(two).toBe(id) - - const rows = Database.use((db) => db.select().from(EventTable).all()) - expect(rows.map((row) => row.seq)).toEqual([0, 1, 2, 3]) - }), - ), - ) - - it.live( - "claims unowned event sequence on replay with ownerID", - provideTmpdirInstance(() => - Effect.gen(function* () { - const { Created } = setup() - const id = MessageID.ascending() - - yield* SyncEvent.use.replay( - { - id: "evt_1", - type: SyncEvent.versionedType(Created.type, Created.version), - seq: 0, - aggregateID: id, - data: { id, name: "owned" }, - }, - { publish: false, ownerID: "owner-1" }, - ) - - const row = Database.use((db) => - db - .select({ seq: EventSequenceTable.seq, ownerID: EventSequenceTable.owner_id }) - .from(EventSequenceTable) - .get(), - ) - expect(row).toEqual({ seq: 0, ownerID: "owner-1" }) - }), - ), - ) - - it.live( - "ignores replay from a different owner after sequence is claimed", - provideTmpdirInstance(() => - Effect.gen(function* () { - const { Created } = setup() - const id = MessageID.ascending() - - yield* SyncEvent.use.replay( - { - id: "evt_1", - type: SyncEvent.versionedType(Created.type, Created.version), - seq: 0, - aggregateID: id, - data: { id, name: "first" }, - }, - { publish: false, ownerID: "owner-1" }, - ) - yield* SyncEvent.use.replay( - { - id: "evt_2", - type: SyncEvent.versionedType(Created.type, Created.version), - seq: 1, - aggregateID: id, - data: { id, name: "ignored" }, - }, - { publish: false, ownerID: "owner-2" }, - ) - - const events = Database.use((db) => db.select().from(EventTable).all()) - const sequence = Database.use((db) => - db - .select({ seq: EventSequenceTable.seq, ownerID: EventSequenceTable.owner_id }) - .from(EventSequenceTable) - .get(), - ) - expect(events).toHaveLength(1) - expect(events[0]?.id).toBe(EventV2.ID.make("evt_1")) - expect(sequence).toEqual({ seq: 0, ownerID: "owner-1" }) - }), - ), - ) - - it.live( - "claim updates the event sequence owner", - provideTmpdirInstance(() => - Effect.gen(function* () { - const { Created } = setup() - const id = MessageID.ascending() - - yield* SyncEvent.use.run(Created, { id, name: "claimed" }, { publish: false }) - yield* SyncEvent.use.claim(id, "owner-1") - yield* SyncEvent.use.claim(id, "owner-2") - - const row = Database.use((db) => - db - .select({ seq: EventSequenceTable.seq, ownerID: EventSequenceTable.owner_id }) - .from(EventSequenceTable) - .where(eq(EventSequenceTable.aggregate_id, id)) - .get(), - ) - expect(row).toEqual({ seq: 0, ownerID: "owner-2" }) - }), - ), - ) - }) -}) diff --git a/specs/storage/remove-opencode-db.md b/specs/storage/remove-opencode-db.md index dc11c65a7..3e8344676 100644 --- a/specs/storage/remove-opencode-db.md +++ b/specs/storage/remove-opencode-db.md @@ -25,7 +25,6 @@ Production imports from `packages/opencode/src/storage/db.ts` are concentrated i - `packages/opencode/src/cli/cmd/import.ts` - `packages/opencode/src/cli/cmd/stats.ts` - `packages/opencode/src/control-plane/workspace.ts` -- `packages/opencode/src/data-migration.ts` - `packages/opencode/src/index.ts` - `packages/opencode/src/node.ts` - `packages/opencode/src/permission/index.ts` @@ -47,6 +46,8 @@ There are 65 direct API/type references in those files. The references fall into ## Group 1: Database Runtime And Startup +Status: Completed. Startup, the public node export, and database CLI tooling no longer import the legacy opencode database wrapper; `packages/opencode/src/storage/db.ts` has been deleted. + Files: - `packages/opencode/src/storage/db.ts` @@ -75,6 +76,8 @@ Target shape: ## Group 2: Sync Event Transaction Boundary +Status: Completed. `SyncEvent` and the opencode projector boundary were removed; session/message event projection now lives in core EventV2/projector infrastructure. + Files: - `packages/opencode/src/sync/index.ts` @@ -106,6 +109,8 @@ Suggested first step: ## Group 3: Domain Repositories Already Behind Services +Status: Completed. These services no longer import the legacy opencode database wrapper. + Files: - `packages/opencode/src/account/repo.ts` @@ -139,6 +144,8 @@ Suggested order: ## Group 4: Session And Message Read Models +Status: Completed. Session/message reads and projector writes have moved off the legacy opencode database wrapper. + Files: - `packages/opencode/src/session/session.ts` @@ -176,6 +183,8 @@ Suggested order: ## Group 5: Legacy CLI And One-Off Admin Reads +Status: Completed. Remaining one-off CLI/admin reads and writes now use core database services or domain services instead of the legacy opencode database wrapper. + Files: - `packages/opencode/src/cli/cmd/import.ts` @@ -206,46 +215,21 @@ Target shape: - For HTTP sync reads, move the event log query behind the sync event module. - For permission and worktree reads, call the permission/project services if available; otherwise add narrow repository methods. -## Group 6: Data Migrations - -Files: - -- `packages/opencode/src/data-migration.ts` - -Current usage: - -- Checks `DataMigrationTable` with `Database.use`. -- Runs resumable data migrations with `Database.use` and `Database.transaction`. -- Writes completion rows with `Database.use`. - -Why this group is separate: - -- Data migrations are database-native by definition and may reasonably stay close to SQL. -- They should not depend on the legacy opencode wrapper, but they do need a stable transaction API and migration completion store. - -Target shape: - -- Run data migrations through the same Effect database service used by startup/migrations. -- Keep SQL-heavy migration logic local to `data-migration.ts`, but remove callback-style legacy access. -- Ensure migrations still run in a scoped/background fiber and remain resumable. - ## Recommended Migration Sequence -1. Replace the legacy runtime seam from Group 1 with an Effect-native database module that exposes path, client/query access, transaction, and after-commit behavior. -2. Port Group 2 sync event transaction semantics to the new module before touching projector bodies. -3. Migrate Group 3 repositories that already hide database access behind service interfaces. -4. Migrate Group 4 session/message reads, then projector write helpers once the new projector transaction type exists. -5. Migrate Group 6 data migrations onto the new database service. -6. Clean up Group 5 one-off reads and CLI/admin commands. -7. Remove drizzle helper re-exports from `@/storage/db` imports by importing operators directly from `drizzle-orm` during each file migration. -8. Delete `packages/opencode/src/storage/db.ts` once `rg "@/storage/db|./storage/db|Database\." packages/opencode/src` no longer finds legacy usages. +All migration groups are complete or superseded. `packages/opencode/src/storage/db.ts` has been deleted. + +## Superseded: Data Migrations + +Status: Superseded. No opencode data-migration group remains. + +The previous opencode `data-migration.ts` service only backfilled session usage from message rows. That work is now covered by core database migration `packages/core/src/database/migration/20260510033149_session_usage.ts`, so there is no separate opencode data-migration group. ## Invariants To Preserve - Nested reads inside a transaction must use the active transaction, not the root client. - `SyncEvent.run` sequence allocation must keep immediate transaction behavior. - Post-commit publish effects must not run before the transaction commits. -- Data migrations must remain resumable and record completion only after successful migration work. - Existing schema ownership remains in `packages/core/src/**/*.sql.ts`; do not move table definitions back into `packages/opencode`. ## Verification Commands