diff --git a/packages/effect-drizzle-sqlite/src/effect-sqlite/migrator.ts b/packages/effect-drizzle-sqlite/src/effect-sqlite/migrator.ts index 91aacccf37..6d0d155143 100644 --- a/packages/effect-drizzle-sqlite/src/effect-sqlite/migrator.ts +++ b/packages/effect-drizzle-sqlite/src/effect-sqlite/migrator.ts @@ -1,8 +1,7 @@ /* oxlint-disable */ -import type { MigrationConfig, MigrationFromJournalConfig, MigrationsJournal } from "drizzle-orm/migrator" +import type { MigrationConfig } 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" @@ -13,21 +12,3 @@ 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 fb5ccd6757..d6606b7d9f 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, migrateFromJournal } from "./effect-sqlite/migrator" +export { migrate } from "./effect-sqlite/migrator" export * as EffectDrizzleSqlite from "." diff --git a/packages/opencode/src/storage/db.ts b/packages/opencode/src/storage/db.ts index 55600bd1a4..6df731b530 100644 --- a/packages/opencode/src/storage/db.ts +++ b/packages/opencode/src/storage/db.ts @@ -48,21 +48,12 @@ export type Transaction = SQLiteTransaction<"sync", void> type Client = ReturnType -export type Journal = MigrationsJournal +type Journal = MigrationsJournal function applyMigrations(db: SQLiteBunDatabase, entries: Journal) { 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) { const match = /^(\d{4})(\d{2})(\d{2})(\d{2})(\d{2})(\d{2})/.exec(tag) if (!match) return 0 @@ -116,12 +107,20 @@ export const Client = Object.assign( db.run("PRAGMA wal_checkpoint(PASSIVE)") // Apply schema migrations - const entries = migrationJournal(flags) + const entries = + typeof OPENCODE_MIGRATIONS !== "undefined" + ? OPENCODE_MIGRATIONS + : migrations(path.join(import.meta.dirname, "../../migration")) 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) } diff --git a/packages/opencode/src/v2/storage/database.ts b/packages/opencode/src/v2/storage/database.ts new file mode 100644 index 0000000000..47c5a375a6 --- /dev/null +++ b/packages/opencode/src/v2/storage/database.ts @@ -0,0 +1,39 @@ +import { Database as LegacyDatabase } from "@/storage/db" +import { SqliteClient } from "@effect/sql-sqlite-bun" +import { EffectDrizzleSqlite } from "@opencode-ai/effect-drizzle-sqlite" +import { Context, Effect, Layer } from "effect" +import path from "path" + +const makeDatabase = EffectDrizzleSqlite.makeWithDefaults() +type DatabaseShape = Effect.Success + +export class Service extends Context.Service()("@opencode/v2/storage/Database") {} + +export const layer = Layer.unwrap( + Effect.sync(() => { + const filename = LegacyDatabase.getPath() + return Layer.effect( + Service, + Effect.gen(function* () { + LegacyDatabase.Client() + 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)") + if (filename === ":memory:") { + yield* EffectDrizzleSqlite.migrate(db, { + migrationsFolder: path.join(import.meta.dirname, "../../../migration"), + }) + } + return db + }), + ).pipe(Layer.provide(SqliteClient.layer({ filename, disableWAL: filename === ":memory:" }))) + }), +) + +export const defaultLayer = layer + +export * as StorageDatabase from "./database" diff --git a/packages/opencode/src/v2/storage/session-sql.ts b/packages/opencode/src/v2/storage/session-sql.ts index 80819ace6a..cc0ed781eb 100644 --- a/packages/opencode/src/v2/storage/session-sql.ts +++ b/packages/opencode/src/v2/storage/session-sql.ts @@ -1,47 +1,25 @@ import { SessionMessageTable, SessionTable } from "@/session/session.sql" -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 { and, asc, desc, eq, gt, gte, isNull, like, lt, or, type SQL } from "@/storage/db" import { SessionMessage } from "@opencode-ai/core/session-message" -import { EffectDrizzleSqlite } from "@opencode-ai/effect-drizzle-sqlite" -import { Context, Effect, Layer, Schema } from "effect" +import { Effect, Layer, Schema } from "effect" +import { StorageDatabase } from "./database" 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:" }))) - }), +const mapStorageError = Effect.mapError( + (cause) => new SessionStorage.StorageError({ message: "Session storage SQL operation failed", cause }), ) export const layer = Layer.effect( SessionStorage.Service, Effect.gen(function* () { - const db = yield* Database + const db = yield* StorageDatabase.Service 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()) + const row = yield* db.select().from(SessionTable).where(eq(SessionTable.id, sessionID)).get() return row ? fromSessionRow(row) : undefined - }) + }, mapStorageError) const list: SessionStorage.Interface["list"] = Effect.fn("SessionStorageSql.list")(function* (input) { const direction = input.cursor?.direction ?? "next" @@ -65,9 +43,9 @@ export const layer = Layer.effect( 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)) + const rows = yield* input.limit === undefined ? query : query.limit(input.limit) return (direction === "previous" ? rows.toReversed() : rows).map(fromSessionRow) - }) + }, mapStorageError) const messages: SessionStorage.Interface["messages"] = Effect.fn("SessionStorageSql.messages")(function* (input) { const direction = input.cursor?.direction ?? "next" @@ -85,62 +63,50 @@ export const layer = Layer.effect( 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)) + const rows = yield* 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 }), ) - }) + }, mapStorageError) const context: SessionStorage.Interface["context"] = Effect.fn("SessionStorageSql.context")((sessionID) => 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(), - ) + const compaction = yield* 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() - const rows = yield* attempt( - db - .select() - .from(SessionMessageTable) - .where( - and( - eq(SessionMessageTable.session_id, sessionID), - compaction - ? or( - gt(SessionMessageTable.time_created, compaction.time_created), - and( - eq(SessionMessageTable.time_created, compaction.time_created), - gte(SessionMessageTable.id, compaction.id), - ), - ) - : undefined, - ), - ) - .orderBy(asc(SessionMessageTable.time_created), asc(SessionMessageTable.id)), - ) + const rows = yield* db + .select() + .from(SessionMessageTable) + .where( + and( + eq(SessionMessageTable.session_id, sessionID), + compaction + ? or( + gt(SessionMessageTable.time_created, compaction.time_created), + and( + eq(SessionMessageTable.time_created, compaction.time_created), + gte(SessionMessageTable.id, compaction.id), + ), + ) + : undefined, + ), + ) + .orderBy(asc(SessionMessageTable.time_created), asc(SessionMessageTable.id)) return rows.map((row) => decodeMessage({ ...row.data, id: row.id, type: row.type })) - }), + }).pipe(mapStorageError), ) return SessionStorage.Service.of({ get, list, messages, context }) }), ) -export const defaultLayer = layer.pipe(Layer.provide(databaseLayer.pipe(Layer.orDie))) - -function attempt(effect: Effect.Effect) { - return effect.pipe( - Effect.mapError( - (cause) => new SessionStorage.StorageError({ message: "Session storage SQL operation failed", cause }), - ), - ) -} +export const defaultLayer = layer.pipe(Layer.provide(StorageDatabase.defaultLayer.pipe(Layer.orDie))) function sessionCursorBoundary(cursor: SessionStorage.SessionCursor, order: SessionStorage.SortOrder) { if (order === "asc") diff --git a/packages/opencode/test/v2/session-storage.test.ts b/packages/opencode/test/v2/session-storage.test.ts index 7533bf0ce4..7083950a1b 100644 --- a/packages/opencode/test/v2/session-storage.test.ts +++ b/packages/opencode/test/v2/session-storage.test.ts @@ -4,6 +4,7 @@ import { ProjectTable } from "@/project/project.sql" import { SessionID } from "@/session/schema" import { SessionMessageTable, SessionTable } from "@/session/session.sql" import { SessionStorage } from "@/v2/storage/session" +import { StorageDatabase } from "@/v2/storage/database" import { SessionStorageMemory } from "@/v2/storage/session-memory" import { SessionStorageSql } from "@/v2/storage/session-sql" import { EventV2 } from "@opencode-ai/core/event" @@ -156,9 +157,9 @@ function sessionStorageContract(name: string, layer: Layer.Layer = { +const sqlSeeds: Seeds = { reset: resetSqlSeeds(), project: seedProject(), session: seedSession, @@ -192,7 +193,7 @@ sessionStorageContract("SessionStorageMemory", memoryLayer, memorySeeds) function seedProject() { return Effect.gen(function* () { - const db = yield* SessionStorageSql.Database + const db = yield* StorageDatabase.Service yield* db .insert(ProjectTable) .values({ @@ -210,7 +211,7 @@ function seedProject() { function resetSqlSeeds() { return Effect.gen(function* () { - const db = yield* SessionStorageSql.Database + const db = yield* StorageDatabase.Service yield* db.delete(SessionMessageTable).where(eq(SessionMessageTable.session_id, sessionA)).run().pipe(Effect.orDie) yield* db .delete(SessionTable) @@ -230,7 +231,7 @@ function resetSqlSeeds() { function seedSession(input: { id: SessionID; title: string; directory?: string; path: string; updated: number }) { return Effect.gen(function* () { - const db = yield* SessionStorageSql.Database + const db = yield* StorageDatabase.Service yield* db .insert(SessionTable) .values({ @@ -320,7 +321,7 @@ function seedMessage( data: typeof SessionMessageTable.$inferInsert.data, ) { return Effect.gen(function* () { - const db = yield* SessionStorageSql.Database + const db = yield* StorageDatabase.Service yield* db .insert(SessionMessageTable) .values({