diff --git a/bun.lock b/bun.lock index 037b72a29b..a34b0cdea1 100644 --- a/bun.lock +++ b/bun.lock @@ -429,12 +429,14 @@ "@clack/prompts": "1.0.0-alpha.1", "@effect/opentelemetry": "catalog:", "@effect/platform-node": "catalog:", + "@effect/sql-sqlite-bun": "catalog:", "@gitlab/opencode-gitlab-auth": "1.3.3", "@lydell/node-pty": "catalog:", "@modelcontextprotocol/sdk": "1.27.1", "@octokit/graphql": "9.0.2", "@octokit/rest": "catalog:", "@openauthjs/openauth": "catalog:", + "@opencode-ai/effect-drizzle-sqlite": "workspace:*", "@opencode-ai/llm": "workspace:*", "@opencode-ai/plugin": "workspace:*", "@opencode-ai/script": "workspace:*", diff --git a/packages/effect-drizzle-sqlite/src/effect-sqlite/migrator.ts b/packages/effect-drizzle-sqlite/src/effect-sqlite/migrator.ts index 6d0d155143..91aacccf37 100644 --- a/packages/effect-drizzle-sqlite/src/effect-sqlite/migrator.ts +++ b/packages/effect-drizzle-sqlite/src/effect-sqlite/migrator.ts @@ -1,7 +1,8 @@ /* oxlint-disable */ -import type { MigrationConfig } from "drizzle-orm/migrator" +import type { MigrationConfig, MigrationFromJournalConfig, MigrationsJournal } from "drizzle-orm/migrator" import { readMigrationFiles } from "drizzle-orm/migrator" import type { AnyRelations } from "drizzle-orm/relations" +import crypto from "node:crypto" import { migrate as coreMigrate } from "../sqlite-core/effect/session" import type { EffectSQLiteDatabase } from "./driver" @@ -12,3 +13,21 @@ export function migrate( const migrations = readMigrationFiles(config) return coreMigrate(migrations, db.session, config) } + +export function migrateFromJournal( + db: EffectSQLiteDatabase, + journal: MigrationsJournal, + config: Omit = {}, +) { + return coreMigrate( + journal.map((migration) => ({ + sql: migration.sql.split("--> statement-breakpoint"), + bps: true, + folderMillis: migration.timestamp, + hash: crypto.createHash("sha256").update(migration.sql).digest("hex"), + name: migration.name, + })), + db.session, + { migrationsFolder: "", migrationsTable: config.migrationsTable }, + ) +} diff --git a/packages/effect-drizzle-sqlite/src/index.ts b/packages/effect-drizzle-sqlite/src/index.ts index d6606b7d9f..fb5ccd6757 100644 --- a/packages/effect-drizzle-sqlite/src/index.ts +++ b/packages/effect-drizzle-sqlite/src/index.ts @@ -1,6 +1,6 @@ export { EffectLogger } from "drizzle-orm/effect-core" export * from "./effect-sqlite/driver" export * from "./effect-sqlite/session" -export { migrate } from "./effect-sqlite/migrator" +export { migrate, migrateFromJournal } from "./effect-sqlite/migrator" export * as EffectDrizzleSqlite from "." diff --git a/packages/opencode/package.json b/packages/opencode/package.json index 3c6e1076b1..37a2b3869b 100644 --- a/packages/opencode/package.json +++ b/packages/opencode/package.json @@ -96,12 +96,14 @@ "@clack/prompts": "1.0.0-alpha.1", "@effect/opentelemetry": "catalog:", "@effect/platform-node": "catalog:", + "@effect/sql-sqlite-bun": "catalog:", "@gitlab/opencode-gitlab-auth": "1.3.3", "@lydell/node-pty": "catalog:", "@modelcontextprotocol/sdk": "1.27.1", "@octokit/graphql": "9.0.2", "@octokit/rest": "catalog:", "@openauthjs/openauth": "catalog:", + "@opencode-ai/effect-drizzle-sqlite": "workspace:*", "@opencode-ai/llm": "workspace:*", "@opencode-ai/plugin": "workspace:*", "@opencode-ai/script": "workspace:*", diff --git a/packages/opencode/src/storage/db.ts b/packages/opencode/src/storage/db.ts index 06f1f84a9a..55600bd1a4 100644 --- a/packages/opencode/src/storage/db.ts +++ b/packages/opencode/src/storage/db.ts @@ -1,5 +1,6 @@ import { type SQLiteBunDatabase } from "drizzle-orm/bun-sqlite" import { migrate } from "drizzle-orm/bun-sqlite/migrator" +import type { MigrationsJournal } from "drizzle-orm/migrator" import { type SQLiteTransaction } from "drizzle-orm/sqlite-core" export * from "drizzle-orm" import { RuntimeFlags } from "@/effect/runtime-flags" @@ -47,13 +48,19 @@ export type Transaction = SQLiteTransaction<"sync", void> type Client = ReturnType -type Journal = { sql: string; timestamp: number; name: string }[] - -// Drizzle's migrate overloads trigger expensive variance checks here; narrow to the journal overload we actually use. -const migrateFromJournal = migrate as unknown as (db: SQLiteBunDatabase, entries: Journal) => void +export type Journal = MigrationsJournal function applyMigrations(db: SQLiteBunDatabase, entries: Journal) { - migrateFromJournal(db, entries) + migrate(db, entries) +} + +export function migrationJournal(flags: Pick = readRuntimeFlags()) { + const entries = + typeof OPENCODE_MIGRATIONS !== "undefined" + ? OPENCODE_MIGRATIONS + : migrations(path.join(import.meta.dirname, "../../migration")) + if (!flags.skipMigrations) return entries + return entries.map((item) => ({ ...item, sql: "select 1;" })) } function time(tag: string) { @@ -74,17 +81,17 @@ function migrations(dir: string): Journal { .filter((entry) => entry.isDirectory()) .map((entry) => entry.name) - const sql = dirs + const sql: Journal = dirs .map((name) => { const file = path.join(dir, name, "migration.sql") - if (!existsSync(file)) return + if (!existsSync(file)) return undefined return { sql: readFileSync(file, "utf-8"), timestamp: time(name), name, } }) - .filter(Boolean) as Journal + .filter((entry) => entry !== undefined) return sql.sort((a, b) => a.timestamp - b.timestamp) } @@ -94,7 +101,7 @@ let loaded = false export const Client = Object.assign( (flags: DatabaseFlags = readRuntimeFlags()): Client => { - if (loaded) return client as Client + if (loaded && client) return client const dbPath = getPath(flags) log.info("opening database", { path: dbPath }) @@ -109,20 +116,12 @@ export const Client = Object.assign( db.run("PRAGMA wal_checkpoint(PASSIVE)") // Apply schema migrations - const entries = - typeof OPENCODE_MIGRATIONS !== "undefined" - ? OPENCODE_MIGRATIONS - : migrations(path.join(import.meta.dirname, "../../migration")) + const entries = migrationJournal(flags) if (entries.length > 0) { log.info("applying migrations", { count: entries.length, mode: typeof OPENCODE_MIGRATIONS !== "undefined" ? "bundled" : "dev", }) - if (flags.skipMigrations) { - for (const item of entries) { - item.sql = "select 1;" - } - } applyMigrations(db, entries) } @@ -159,19 +158,19 @@ export function use(callback: (trx: TxOrDb) => T): T { if (err instanceof LocalContext.NotFound) { const effects: (() => void | Promise)[] = [] const result = ctx.provide({ effects, tx: Client() }, () => callback(Client())) - for (const effect of effects) effect() + for (const effect of effects) void effect() return result } throw err } } -export function effect(fn: () => any | Promise) { +export function effect(fn: () => void | Promise) { const bound = EffectBridge.bind(fn) try { ctx.use().effects.push(bound) } catch { - bound() + void bound() } } @@ -190,7 +189,9 @@ export function transaction( 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() + for (const effect of effects) void effect() + // Drizzle's transaction type does not preserve our NotPromise constraint through the callback wrapper. + // oxlint-disable-next-line typescript-eslint/no-unsafe-type-assertion return result as NotPromise } throw err diff --git a/packages/opencode/src/v2/storage/session-sql.ts b/packages/opencode/src/v2/storage/session-sql.ts index 48a6137115..80819ace6a 100644 --- a/packages/opencode/src/v2/storage/session-sql.ts +++ b/packages/opencode/src/v2/storage/session-sql.ts @@ -1,87 +1,110 @@ import { SessionMessageTable, SessionTable } from "@/session/session.sql" -import { and, asc, Database, desc, eq, gt, gte, isNull, like, lt, or, type SQL } from "@/storage/db" +import { and, asc, Database as LegacyDatabase, desc, eq, gt, gte, isNull, like, lt, or, type SQL } from "@/storage/db" +import { SqliteClient } from "@effect/sql-sqlite-bun" import { SessionMessage } from "@opencode-ai/core/session-message" -import { Effect, Layer, Schema } from "effect" +import { EffectDrizzleSqlite } from "@opencode-ai/effect-drizzle-sqlite" +import { Context, Effect, Layer, Schema } from "effect" import { SessionStorage } from "./session" const decodeMessage = Schema.decodeUnknownSync(SessionMessage.Message) const decodeSessionRow = Schema.decodeUnknownSync(SessionStorage.SessionRow) +const makeDatabase = EffectDrizzleSqlite.makeWithDefaults() +type DatabaseShape = Effect.Success + +export class Database extends Context.Service()("@opencode/v2/session/StorageSql/Database") {} + +export const databaseLayer = Layer.unwrap( + Effect.sync(() => { + const filename = LegacyDatabase.getPath() + return Layer.effect( + Database, + Effect.gen(function* () { + const db = yield* makeDatabase + yield* db.run("PRAGMA journal_mode = WAL") + yield* db.run("PRAGMA synchronous = NORMAL") + yield* db.run("PRAGMA busy_timeout = 5000") + yield* db.run("PRAGMA cache_size = -64000") + yield* db.run("PRAGMA foreign_keys = ON") + yield* db.run("PRAGMA wal_checkpoint(PASSIVE)") + yield* EffectDrizzleSqlite.migrateFromJournal(db, LegacyDatabase.migrationJournal()) + return db + }), + ).pipe(Layer.provide(SqliteClient.layer({ filename, disableWAL: filename === ":memory:" }))) + }), +) export const layer = Layer.effect( SessionStorage.Service, Effect.gen(function* () { - const get: SessionStorage.Interface["get"] = Effect.fn("SessionStorageSql.get")((sessionID) => - attempt(() => - Database.use((db) => db.select().from(SessionTable).where(eq(SessionTable.id, sessionID)).get()), - ).pipe(Effect.map((row) => (row ? fromSessionRow(row) : undefined))), - ) + const db = yield* Database - const list: SessionStorage.Interface["list"] = Effect.fn("SessionStorageSql.list")((input) => - attempt(() => { - const direction = input.cursor?.direction ?? "next" - const order = SessionStorage.pageOrder(input.order ?? "desc", direction) - const sortColumn = SessionTable.time_updated - const conditions: SQL[] = [] - if (input.directory) conditions.push(eq(SessionTable.directory, input.directory)) - if (input.path) - conditions.push(or(eq(SessionTable.path, input.path), like(SessionTable.path, `${input.path}/%`))!) - if (input.workspaceID) conditions.push(eq(SessionTable.workspace_id, input.workspaceID)) - if (input.roots) conditions.push(isNull(SessionTable.parent_id)) - if (input.start) conditions.push(gte(sortColumn, input.start)) - if (input.search) conditions.push(like(SessionTable.title, `%${input.search}%`)) - if (input.cursor) conditions.push(sessionCursorBoundary(input.cursor, order)) + const get: SessionStorage.Interface["get"] = Effect.fn("SessionStorageSql.get")(function* (sessionID) { + const row = yield* attempt(db.select().from(SessionTable).where(eq(SessionTable.id, sessionID)).get()) + return row ? fromSessionRow(row) : undefined + }) - return Database.use((db) => { - const query = db - .select() - .from(SessionTable) - .where(conditions.length > 0 ? and(...conditions) : undefined) - .orderBy( - order === "asc" ? asc(sortColumn) : desc(sortColumn), - order === "asc" ? asc(SessionTable.id) : desc(SessionTable.id), - ) - const rows = input.limit === undefined ? query.all() : query.limit(input.limit).all() - return direction === "previous" ? rows.toReversed() : rows - }) - }).pipe(Effect.map((rows) => rows.map(fromSessionRow))), - ) + const list: SessionStorage.Interface["list"] = Effect.fn("SessionStorageSql.list")(function* (input) { + const direction = input.cursor?.direction ?? "next" + const order = SessionStorage.pageOrder(input.order ?? "desc", direction) + const sortColumn = SessionTable.time_updated + const conditions: SQL[] = [] + if (input.directory) conditions.push(eq(SessionTable.directory, input.directory)) + if (input.path) + conditions.push(or(eq(SessionTable.path, input.path), like(SessionTable.path, `${input.path}/%`))!) + if (input.workspaceID) conditions.push(eq(SessionTable.workspace_id, input.workspaceID)) + if (input.roots) conditions.push(isNull(SessionTable.parent_id)) + if (input.start) conditions.push(gte(sortColumn, input.start)) + if (input.search) conditions.push(like(SessionTable.title, `%${input.search}%`)) + if (input.cursor) conditions.push(sessionCursorBoundary(input.cursor, order)) - const messages: SessionStorage.Interface["messages"] = Effect.fn("SessionStorageSql.messages")((input) => - attempt(() => { - const direction = input.cursor?.direction ?? "next" - const order = SessionStorage.pageOrder(input.order ?? "desc", direction) - const boundary = input.cursor ? messageCursorBoundary(input.cursor, order) : undefined - const where = boundary - ? and(eq(SessionMessageTable.session_id, input.sessionID), boundary) - : eq(SessionMessageTable.session_id, input.sessionID) + const query = db + .select() + .from(SessionTable) + .where(conditions.length > 0 ? and(...conditions) : undefined) + .orderBy( + order === "asc" ? asc(sortColumn) : desc(sortColumn), + order === "asc" ? asc(SessionTable.id) : desc(SessionTable.id), + ) + const rows = yield* attempt(input.limit === undefined ? query : query.limit(input.limit)) + return (direction === "previous" ? rows.toReversed() : rows).map(fromSessionRow) + }) - return Database.use((db) => { - const query = db - .select() - .from(SessionMessageTable) - .where(where) - .orderBy( - order === "asc" ? asc(SessionMessageTable.time_created) : desc(SessionMessageTable.time_created), - order === "asc" ? asc(SessionMessageTable.id) : desc(SessionMessageTable.id), - ) - const rows = input.limit === undefined ? query.all() : query.limit(input.limit).all() - return direction === "previous" ? rows.toReversed() : rows - }) - }).pipe(Effect.map((rows) => rows.map((row) => decodeMessage({ ...row.data, id: row.id, type: row.type })))), - ) + const messages: SessionStorage.Interface["messages"] = Effect.fn("SessionStorageSql.messages")(function* (input) { + const direction = input.cursor?.direction ?? "next" + const order = SessionStorage.pageOrder(input.order ?? "desc", direction) + const boundary = input.cursor ? messageCursorBoundary(input.cursor, order) : undefined + const where = boundary + ? and(eq(SessionMessageTable.session_id, input.sessionID), boundary) + : eq(SessionMessageTable.session_id, input.sessionID) + + const query = db + .select() + .from(SessionMessageTable) + .where(where) + .orderBy( + order === "asc" ? asc(SessionMessageTable.time_created) : desc(SessionMessageTable.time_created), + order === "asc" ? asc(SessionMessageTable.id) : desc(SessionMessageTable.id), + ) + const rows = yield* attempt(input.limit === undefined ? query : query.limit(input.limit)) + return (direction === "previous" ? rows.toReversed() : rows).map((row) => + decodeMessage({ ...row.data, id: row.id, type: row.type }), + ) + }) const context: SessionStorage.Interface["context"] = Effect.fn("SessionStorageSql.context")((sessionID) => - attempt(() => - Database.use((db) => { - const compaction = db + Effect.gen(function* () { + const compaction = yield* attempt( + db .select() .from(SessionMessageTable) .where(and(eq(SessionMessageTable.session_id, sessionID), eq(SessionMessageTable.type, "compaction"))) .orderBy(desc(SessionMessageTable.time_created), desc(SessionMessageTable.id)) .limit(1) - .get() + .get(), + ) - return db + const rows = yield* attempt( + db .select() .from(SessionMessageTable) .where( @@ -98,23 +121,25 @@ export const layer = Layer.effect( : undefined, ), ) - .orderBy(asc(SessionMessageTable.time_created), asc(SessionMessageTable.id)) - .all() - }), - ).pipe(Effect.map((rows) => rows.map((row) => decodeMessage({ ...row.data, id: row.id, type: row.type })))), + .orderBy(asc(SessionMessageTable.time_created), asc(SessionMessageTable.id)), + ) + + return rows.map((row) => decodeMessage({ ...row.data, id: row.id, type: row.type })) + }), ) return SessionStorage.Service.of({ get, list, messages, context }) }), ) -export const defaultLayer = layer +export const defaultLayer = layer.pipe(Layer.provide(databaseLayer.pipe(Layer.orDie))) -function attempt(body: () => A) { - return Effect.try({ - try: body, - catch: (cause) => new SessionStorage.StorageError({ message: "Session storage SQL operation failed", cause }), - }) +function attempt(effect: Effect.Effect) { + return effect.pipe( + Effect.mapError( + (cause) => new SessionStorage.StorageError({ message: "Session storage SQL operation failed", cause }), + ), + ) } function sessionCursorBoundary(cursor: SessionStorage.SessionCursor, order: SessionStorage.SortOrder) { diff --git a/packages/opencode/test/v2/session-storage.test.ts b/packages/opencode/test/v2/session-storage.test.ts index d03748a048..7533bf0ce4 100644 --- a/packages/opencode/test/v2/session-storage.test.ts +++ b/packages/opencode/test/v2/session-storage.test.ts @@ -3,7 +3,6 @@ import { ProjectID } from "@/project/schema" import { ProjectTable } from "@/project/project.sql" import { SessionID } from "@/session/schema" import { SessionMessageTable, SessionTable } from "@/session/session.sql" -import { Database } from "@/storage/db" import { SessionStorage } from "@/v2/storage/session" import { SessionStorageMemory } from "@/v2/storage/session-memory" import { SessionStorageSql } from "@/v2/storage/session-sql" @@ -157,15 +156,17 @@ function sessionStorageContract(name: string, layer: Layer.Layer = { - reset: Effect.sync(resetSqlSeeds), - project: Effect.sync(seedProject), - session: (input) => Effect.sync(() => seedSession(input)), - userMessage: (input) => Effect.sync(() => seedUserMessage(input)), - compaction: (input) => Effect.sync(() => seedCompaction(input)), +const sqlLayer = SessionStorageSql.layer.pipe(Layer.provideMerge(SessionStorageSql.databaseLayer)) + +const sqlSeeds: Seeds = { + reset: resetSqlSeeds(), + project: seedProject(), + session: seedSession, + userMessage: seedUserMessage, + compaction: seedCompaction, } -sessionStorageContract("SessionStorageSql", SessionStorageSql.defaultLayer, sqlSeeds) +sessionStorageContract("SessionStorageSql", sqlLayer, sqlSeeds) const memorySeeds: Seeds = { reset: Effect.sync(() => { @@ -190,8 +191,9 @@ const memorySeeds: Seeds = { sessionStorageContract("SessionStorageMemory", memoryLayer, memorySeeds) function seedProject() { - Database.use((db) => - db + return Effect.gen(function* () { + const db = yield* SessionStorageSql.Database + yield* db .insert(ProjectTable) .values({ id: projectID, @@ -201,14 +203,17 @@ function seedProject() { sandboxes: [], }) .onConflictDoNothing() - .run(), - ) + .run() + .pipe(Effect.orDie) + }) } function resetSqlSeeds() { - Database.use((db) => { - db.delete(SessionMessageTable).where(eq(SessionMessageTable.session_id, sessionA)).run() - db.delete(SessionTable) + return Effect.gen(function* () { + const db = yield* SessionStorageSql.Database + yield* db.delete(SessionMessageTable).where(eq(SessionMessageTable.session_id, sessionA)).run().pipe(Effect.orDie) + yield* db + .delete(SessionTable) .where( or( eq(SessionTable.id, sessionA), @@ -218,13 +223,15 @@ function resetSqlSeeds() { ), ) .run() - db.delete(ProjectTable).where(eq(ProjectTable.id, projectID)).run() + .pipe(Effect.orDie) + yield* db.delete(ProjectTable).where(eq(ProjectTable.id, projectID)).run().pipe(Effect.orDie) }) } function seedSession(input: { id: SessionID; title: string; directory?: string; path: string; updated: number }) { - Database.use((db) => - db + return Effect.gen(function* () { + const db = yield* SessionStorageSql.Database + yield* db .insert(SessionTable) .values({ id: input.id, @@ -243,20 +250,21 @@ function seedSession(input: { id: SessionID; title: string; directory?: string; time_created: input.updated, time_updated: input.updated, }) - .run(), - ) + .run() + .pipe(Effect.orDie) + }) } function seedUserMessage(input: { id: SessionMessage.ID; text: string; time: number }) { const encoded = encodeMessage(makeUserMessage(input)) const { id: _, type: __, ...data } = encoded - seedMessage(input.id, "user", input.time, data) + return seedMessage(input.id, "user", input.time, data) } function seedCompaction(input: { id: SessionMessage.ID; summary: string; time: number }) { const encoded = encodeMessage(makeCompaction(input)) const { id: _, type: __, ...data } = encoded - seedMessage(input.id, "compaction", input.time, data) + return seedMessage(input.id, "compaction", input.time, data) } function makeSessionRow(input: { id: SessionID; title: string; directory?: string; path: string; updated: number }) { @@ -311,8 +319,9 @@ function seedMessage( time: number, data: typeof SessionMessageTable.$inferInsert.data, ) { - Database.use((db) => - db + return Effect.gen(function* () { + const db = yield* SessionStorageSql.Database + yield* db .insert(SessionMessageTable) .values({ id, @@ -322,8 +331,9 @@ function seedMessage( time_updated: time, data, }) - .run(), - ) + .run() + .pipe(Effect.orDie) + }) } function appendMemoryMessage(message: SessionMessage.Message) {