refactor(v2): use effect sqlite session storage

2026-07-24 20:25:36 -04:00 · 2026-05-20 20:34:02 -04:00
parent dc55ece384
commit dcbd244dd7
7 changed files with 181 additions and 122 deletions
@@ -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:*",
@@ -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<TRelations extends AnyRelations>(
  const migrations = readMigrationFiles(config)
  return coreMigrate(migrations, db.session, config)
 }
+
+export function migrateFromJournal<TRelations extends AnyRelations>(
+  db: EffectSQLiteDatabase<TRelations>,
+  journal: MigrationsJournal,
+  config: Omit<MigrationFromJournalConfig, "migrationsJournal"> = {},
+) {
+  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 },
+  )
+}
@@ -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 "."
@@ -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:*",
@@ -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<typeof init>

-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<DatabaseFlags, "skipMigrations"> = 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<T>(callback: (trx: TxOrDb) => T): T {
    if (err instanceof LocalContext.NotFound) {
      const effects: (() => void | Promise<void>)[] = []
      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<any>) {
+export function effect(fn: () => void | Promise<void>) {
  const bound = EffectBridge.bind(fn)
  try {
    ctx.use().effects.push(bound)
  } catch {
-    bound()
+    void bound()
  }
 }

@@ -190,7 +189,9 @@ export function transaction<T>(
      const effects: (() => void | Promise<void>)[] = []
      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<T> constraint through the callback wrapper.
+      // oxlint-disable-next-line typescript-eslint/no-unsafe-type-assertion
      return result as NotPromise<T>
    }
    throw err
@@ -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<typeof makeDatabase>
+
+export class Database extends Context.Service<Database, DatabaseShape>()("@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<A>(body: () => A) {
-  return Effect.try({
-    try: body,
-    catch: (cause) => new SessionStorage.StorageError({ message: "Session storage SQL operation failed", cause }),
-  })
+function attempt<A, E, R>(effect: Effect.Effect<A, E, R>) {
+  return effect.pipe(
+    Effect.mapError(
+      (cause) => new SessionStorage.StorageError({ message: "Session storage SQL operation failed", cause }),
+    ),
+  )
 }

 function sessionCursorBoundary(cursor: SessionStorage.SessionCursor, order: SessionStorage.SortOrder) {
@@ -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<R, E>(name: string, layer: Layer.Layer<SessionSt
  )
 }

-const sqlSeeds: Seeds<never> = {
-  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<SessionStorageSql.Database> = {
+  reset: resetSqlSeeds(),
+  project: seedProject(),
+  session: seedSession,
+  userMessage: seedUserMessage,
+  compaction: seedCompaction,
 }

-sessionStorageContract("SessionStorageSql", SessionStorageSql.defaultLayer, sqlSeeds)
+sessionStorageContract("SessionStorageSql", sqlLayer, sqlSeeds)

 const memorySeeds: Seeds<never> = {
  reset: Effect.sync(() => {
@@ -190,8 +191,9 @@ const memorySeeds: Seeds<never> = {
 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) {