From c5704d3e3b23c52aac1fb2dfdcd07fbefe8de0a6 Mon Sep 17 00:00:00 2001 From: Kit Langton Date: Tue, 11 Aug 2026 15:45:23 -0400 Subject: [PATCH] feat(core): sqlite driver for durable object storage (#41659) --- packages/core/package.json | 1 + packages/core/src/database/database.ts | 26 +- packages/core/src/database/migration.ts | 15 +- packages/core/src/database/sqlite.bun.ts | 5 + packages/core/src/database/sqlite.node.ts | 5 + packages/core/src/database/sqlite.workerd.ts | 254 ++++++++++++++++++ packages/core/test/database-migration.test.ts | 13 + packages/core/test/sqlite-workerd.test.ts | 141 ++++++++++ 8 files changed, 449 insertions(+), 11 deletions(-) create mode 100644 packages/core/src/database/sqlite.workerd.ts create mode 100644 packages/core/test/sqlite-workerd.test.ts diff --git a/packages/core/package.json b/packages/core/package.json index 90534e5d14..be897e3dbf 100644 --- a/packages/core/package.json +++ b/packages/core/package.json @@ -25,6 +25,7 @@ }, "imports": { "#sqlite": { + "workerd": "./src/database/sqlite.workerd.ts", "bun": "./src/database/sqlite.bun.ts", "node": "./src/database/sqlite.node.ts", "default": "./src/database/sqlite.bun.ts" diff --git a/packages/core/src/database/database.ts b/packages/core/src/database/database.ts index 40e56a5cb1..1354cb8c96 100644 --- a/packages/core/src/database/database.ts +++ b/packages/core/src/database/database.ts @@ -1,8 +1,9 @@ export * as Database from "./database" import { EffectDrizzleSqlite } from "@opencode-ai/effect-drizzle-sqlite" -import { sqliteLayer } from "#sqlite" +import { sqliteLayer, supportsForeignKeyToggle, supportsTuningPragmas } from "#sqlite" import { Context, Effect, Layer, Schema } from "effect" +import type { SqlClient } from "effect/unstable/sql" import { Global } from "@opencode-ai/util/global" import { isAbsolute, join } from "path" import { DatabaseMigration } from "./migration" @@ -27,12 +28,15 @@ const databaseLayer = Layer.effect( 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)") + if (supportsTuningPragmas) { + 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 wal_checkpoint(PASSIVE)") + } + // Durable Object SQLite always enforces foreign keys and rejects the pragma. + if (supportsForeignKeyToggle) yield* db.run("PRAGMA foreign_keys = ON") yield* DatabaseMigration.apply(db) return { db } @@ -42,7 +46,7 @@ const databaseLayer = Layer.effect( export function layer(options: Options = { path: ":memory:" }) { return Layer.unwrap( Effect.gen(function* () { - const provide = (filename: string) => databaseLayer.pipe(Layer.provide(sqliteLayer({ filename }))) + const provide = (filename: string) => layerFromClient.pipe(Layer.provide(sqliteLayer({ filename }))) const filename = options.path ?? ":memory:" if (filename === ":memory:" || isAbsolute(filename)) return provide(filename) const global = yield* Global.Service @@ -51,6 +55,12 @@ export function layer(options: Options = { path: ":memory:" }) { ) } +// The database service over an injected SqlClient, for runtimes that receive +// database storage instead of opening a filesystem path. Any client provided +// here still goes through the pragma guards and migrations; Global is required +// because migrations may read it (the v1 import). +export const layerFromClient: Layer.Layer = databaseLayer + export function configured(options?: Options) { return makeGlobalNode({ service: Service, layer: layer(options), deps: [Global.node] }) } diff --git a/packages/core/src/database/migration.ts b/packages/core/src/database/migration.ts index 3c412f3fc4..a572f2b003 100644 --- a/packages/core/src/database/migration.ts +++ b/packages/core/src/database/migration.ts @@ -2,6 +2,7 @@ export * as DatabaseMigration from "./migration" import { sql } from "drizzle-orm" import { Effect, Semaphore } from "effect" +import { supportsForeignKeyToggle } from "#sqlite" import type { EffectDrizzleSqlite } from "@opencode-ai/effect-drizzle-sqlite" import { migrations } from "./migration.gen" import schema from "./schema.gen" @@ -20,8 +21,10 @@ export type Migration = { export function apply(db: Database) { return lock.withPermit( Effect.gen(function* () { + // OpenCode owns the unprefixed table namespace. Embedders sharing this + // database may own underscore-prefixed tables, which bootstrap ignores. const tables = yield* db.all<{ name: string }>( - sql`SELECT name FROM sqlite_master WHERE type = 'table' AND name NOT LIKE 'sqlite_%'`, + sql`SELECT name FROM sqlite_master WHERE type = 'table' AND name NOT LIKE 'sqlite_%' AND substr(name, 1, 1) <> '_'`, ) if (tables.some((table) => table.name === "session" || table.name === "session_v2")) return yield* applyOnly(db, migrations) @@ -103,9 +106,15 @@ export function applyOnly(db: Database, input: Migration[]) { }) continue } - yield* db.run(sql`PRAGMA foreign_keys = OFF`) + // Durable Object SQLite rejects the foreign_keys toggle; the closest + // allowlisted relaxation is deferring enforcement to transaction commit. + const relaxForeignKeys = supportsForeignKeyToggle + ? db.run(sql`PRAGMA foreign_keys = OFF`) + : db.run(sql`PRAGMA defer_foreign_keys = ON`) + const restoreForeignKeys = supportsForeignKeyToggle ? db.run(sql`PRAGMA foreign_keys = ON`) : Effect.void + yield* relaxForeignKeys yield* apply.pipe( - Effect.ensuring(db.run(sql`PRAGMA foreign_keys = ON`).pipe(Effect.orDie)), + Effect.ensuring(restoreForeignKeys.pipe(Effect.orDie)), Effect.tapError((error) => Effect.logError("database migration failed", { migration: migration.id, diff --git a/packages/core/src/database/sqlite.bun.ts b/packages/core/src/database/sqlite.bun.ts index d7739be087..4a49a9a3e8 100644 --- a/packages/core/src/database/sqlite.bun.ts +++ b/packages/core/src/database/sqlite.bun.ts @@ -8,6 +8,11 @@ import { Sqlite } from "./sqlite" const TypeId = "~@opencode-ai/core/database/SqliteBun" as const +export const supportsTuningPragmas = true + +// Foreign keys default OFF and can be toggled per connection. +export const supportsForeignKeyToggle = true + interface Config extends Sqlite.ClientConfig { readonly filename: string readonly readonly?: boolean diff --git a/packages/core/src/database/sqlite.node.ts b/packages/core/src/database/sqlite.node.ts index 2ead9df734..54e3b40efd 100644 --- a/packages/core/src/database/sqlite.node.ts +++ b/packages/core/src/database/sqlite.node.ts @@ -8,6 +8,11 @@ import { Sqlite } from "./sqlite" const TypeId = "~@opencode-ai/core/database/SqliteNode" as const +export const supportsTuningPragmas = true + +// Foreign keys default OFF and can be toggled per connection. +export const supportsForeignKeyToggle = true + interface Config extends Sqlite.ClientConfig { readonly filename: string readonly readonly?: boolean diff --git a/packages/core/src/database/sqlite.workerd.ts b/packages/core/src/database/sqlite.workerd.ts new file mode 100644 index 0000000000..94a5a5eb76 --- /dev/null +++ b/packages/core/src/database/sqlite.workerd.ts @@ -0,0 +1,254 @@ +import { drizzle } from "drizzle-orm/durable-sqlite" +import { Context, Effect, Exit, Fiber, Layer, Scope, Semaphore, Stream } from "effect" +import { identity } from "effect/Function" +import { Reactivity } from "effect/unstable/reactivity" +import { SqlClient, Statement } from "effect/unstable/sql" +import type { Connection } from "effect/unstable/sql/SqlConnection" +import { classifySqliteError, SqlError, UnknownError } from "effect/unstable/sql/SqlError" +import { Sqlite } from "./sqlite" + +const ATTR_DB_SYSTEM_NAME = "db.system.name" + +const TypeId = "~@opencode-ai/core/database/SqliteWorkerd" as const +type TypeId = typeof TypeId + +// Durable Object SQLite only allowlists introspection pragmas; journal_mode, +// synchronous, busy_timeout, cache_size, and wal_checkpoint all throw, and +// foreign keys are already enforced by default (SQLITE_DEFAULT_FOREIGN_KEYS=1). +export const supportsTuningPragmas = false + +// Durable Object SQLite rejects `PRAGMA foreign_keys`: enforcement is always +// on (SQLITE_DEFAULT_FOREIGN_KEYS=1) and only `defer_foreign_keys` is +// allowlisted for migrations that must relax checking inside a transaction. +export const supportsForeignKeyToggle = false + +// Minimal structural types for the Durable Object storage API so this adapter +// does not depend on @cloudflare/workers-types (whose ambient globals conflict +// with @types/bun). Shapes match the SqlStorage and DurableObjectStorage docs. +type SqlStorageValue = ArrayBuffer | string | number | null + +interface SqlStorageCursor { + readonly columnNames: Array + raw(): IterableIterator> + toArray(): Array> +} + +export interface SqlStorage { + exec(query: string, ...bindings: Array): SqlStorageCursor +} + +export interface DurableObjectStorage { + readonly sql: SqlStorage + transaction(closure: (txn: { rollback(): void }) => Promise): Promise + transactionSync(closure: () => T): T +} + +interface SqliteClient extends SqlClient.SqlClient { + readonly [TypeId]: TypeId + readonly config: Config + readonly updateValues: never +} + +interface Config { + readonly storage: DurableObjectStorage + readonly spanAttributes?: Record + readonly transformResultNames?: (str: string) => string + readonly transformQueryNames?: (str: string) => string +} + +// sql.exec() rejects BEGIN/COMMIT/SAVEPOINT, so SqlClient.make's default +// transaction SQL can never run. withTransaction is replaced below with a +// DurableObjectStorage.transaction-backed implementation; this service only +// tracks the active transaction connection for statements and nesting checks. +const WorkerdTransaction = Context.Service( + "@opencode-ai/core/database/SqliteWorkerdTransaction", +) + +const transactionError = (message: string) => + new SqlError({ + reason: new UnknownError({ cause: new Error(message), message, operation: "transaction" }), + }) + +const makeWithTransaction = + ( + storage: DurableObjectStorage, + connection: Connection, + semaphore: Semaphore.Semaphore, + ): SqlClient.SqlClient["withTransaction"] => + (effect: Effect.Effect): Effect.Effect => + Effect.withFiber((fiber) => { + const services = fiber.context + if (Context.getOption(services, WorkerdTransaction)._tag === "Some") + return Effect.fail( + transactionError("Nested transactions are not supported by Cloudflare Durable Object SQLite storage"), + ) + const effectWithTxn = Effect.provideContext( + effect, + Context.add(services, WorkerdTransaction, [connection, 0] as const), + ) + return semaphore.withPermits(1)( + Effect.callback((resume) => { + let interrupted = false + const promise = storage + .transaction( + (txn) => + new Promise((resolve) => { + if (interrupted) return resolve() + resume( + Effect.onExit(effectWithTxn, (exit) => { + if (Exit.isFailure(exit)) txn.rollback() + resolve() + // wait for the transaction to complete + return Effect.promise(() => promise) + }), + ) + }), + ) + .catch((cause) => + resume( + Effect.fail( + new SqlError({ + reason: classifySqliteError(cause, { message: "Failed transaction", operation: "transaction" }), + }), + ), + ), + ) + return Effect.suspend(() => { + interrupted = true + return Effect.promise(() => promise) + }) + }), + ) + }) + +const make = (options: Config) => + Effect.gen(function* () { + const native = (yield* Sqlite.Native) as DurableObjectStorage + + const compiler = Statement.makeCompilerSqlite(options.transformQueryNames) + const transformRows = options.transformResultNames + ? Statement.defaultTransforms(options.transformResultNames).array + : undefined + + // SqlClient.SafeIntegers is ignored: Durable Object SQLite has no bigint + // mode and always returns integers as numbers. Blobs come back as + // ArrayBuffer and are normalized to Uint8Array to match the other adapters. + function* runIterator(query: string, params: ReadonlyArray = []) { + const cursor = native.sql.exec(query, ...params) + const columns = cursor.columnNames + for (const row of cursor.raw()) { + const record: Record = {} + for (let i = 0; i < columns.length; i++) { + const value = row[i] + record[columns[i]] = value instanceof ArrayBuffer ? new Uint8Array(value) : value + } + yield record + } + } + + const run = (query: string, params: ReadonlyArray = []) => + Effect.try({ + try: () => Array.from(runIterator(query, params)), + catch: (cause) => + new SqlError({ + reason: classifySqliteError(cause, { message: "Failed to execute statement", operation: "execute" }), + }), + }) + + const runValues = (query: string, params: ReadonlyArray = []) => + Effect.try({ + try: () => + Array.from(native.sql.exec(query, ...params).raw(), (row) => + row.map((value) => (value instanceof ArrayBuffer ? new Uint8Array(value) : value)), + ), + catch: (cause) => + new SqlError({ + reason: classifySqliteError(cause, { message: "Failed to execute statement", operation: "execute" }), + }), + }) + + const connection = identity({ + execute(query, params, transformRows) { + return transformRows ? Effect.map(run(query, params), transformRows) : run(query, params) + }, + executeRaw(query, params) { + return run(query, params) + }, + executeValues(query, params) { + return runValues(query, params) + }, + executeValuesUnprepared(query, params) { + return runValues(query, params) + }, + executeUnprepared(query, params, transformRows) { + return this.execute(query, params, transformRows) + }, + executeStream() { + return Stream.die("executeStream not implemented") + }, + }) + + const semaphore = yield* Semaphore.make(1) + const acquirer = semaphore.withPermits(1)(Effect.succeed(connection)) + const transactionAcquirer = Effect.uninterruptibleMask((restore) => { + const fiber = Fiber.getCurrent()! + const scope = Context.getUnsafe(fiber.context, Scope.Scope) + return Effect.as( + Effect.tap(restore(semaphore.take(1)), () => Scope.addFinalizer(scope, semaphore.release(1))), + connection, + ) + }) + + const client = Object.assign( + (yield* SqlClient.make({ + acquirer, + compiler, + transactionAcquirer, + transactionService: WorkerdTransaction, + spanAttributes: [ + ...(options.spanAttributes ? Object.entries(options.spanAttributes) : []), + [ATTR_DB_SYSTEM_NAME, "sqlite"], + ], + transformRows, + })) as SqliteClient, + { + [TypeId]: TypeId, + config: options, + withTransaction: makeWithTransaction(native, connection, semaphore), + // Durable Object SQLite rejects BEGIN/COMMIT/SAVEPOINT; consumers such + // as the drizzle session must route through withTransaction instead. + transactionStatements: false, + }, + ) + + return client + }) + +// Defends against the shared path-based Database.layer, which passes a +// filename instead of storage when resolved under the workerd condition. +const nativeLayer = (config: Config) => + config.storage + ? Layer.succeed(Sqlite.Native, config.storage) + : Layer.effect( + Sqlite.Native, + Effect.die( + "workerd sqlite cannot open a database from a path; use Database.layerWith(sqliteLayer({ storage }))", + ), + ) + +const clientLayer = (config: Config) => Layer.effect(SqlClient.SqlClient, make(config)) + +const drizzleLayer = Layer.effect( + Sqlite.Drizzle, + Effect.gen(function* () { + const native = (yield* Sqlite.Native) as DurableObjectStorage + return drizzle(native) as unknown as Sqlite.DrizzleClient + }), +) + +export const sqliteLayer = (config: Config) => { + const native = nativeLayer(config) + return Layer.merge(native, Layer.merge(clientLayer(config), drizzleLayer).pipe(Layer.provide(native))).pipe( + Layer.provide(Reactivity.layer), + ) +} diff --git a/packages/core/test/database-migration.test.ts b/packages/core/test/database-migration.test.ts index f23cf0e8bb..06de4f58df 100644 --- a/packages/core/test/database-migration.test.ts +++ b/packages/core/test/database-migration.test.ts @@ -84,6 +84,19 @@ describe("DatabaseMigration", () => { ).rejects.toThrow("Database is not empty and has no session table") }) + test("bootstraps alongside underscore-prefixed embedder tables", async () => { + await run( + Effect.gen(function* () { + const db = yield* makeDb + yield* db.run(sql`CREATE TABLE _embedder_state (id text PRIMARY KEY)`) + yield* DatabaseMigration.apply(db) + expect(yield* db.get(sql`SELECT name FROM sqlite_master WHERE type = 'table' AND name = 'session_v2'`)).toEqual( + { name: "session_v2" }, + ) + }), + ) + }) + test("applies generic migrations once and records their order", async () => { await run( Effect.gen(function* () { diff --git a/packages/core/test/sqlite-workerd.test.ts b/packages/core/test/sqlite-workerd.test.ts new file mode 100644 index 0000000000..64194b0418 --- /dev/null +++ b/packages/core/test/sqlite-workerd.test.ts @@ -0,0 +1,141 @@ +import { describe, expect, test } from "bun:test" +import { Database } from "bun:sqlite" +import { Effect, Layer } from "effect" +import { SqlClient } from "effect/unstable/sql" +import { SqlError } from "effect/unstable/sql/SqlError" +import { sqliteLayer } from "@opencode-ai/core/database/sqlite.workerd" +import type { DurableObjectStorage } from "@opencode-ai/core/database/sqlite.workerd" +import { tempGlobalLayer } from "./fixture/global" + +// Emulates the Durable Object storage API over bun:sqlite so the adapter can +// be verified without workerd or Cloudflare runtime dependencies. +const makeFakeStorage = () => { + const native = new Database(":memory:") + const toSqlStorageValue = (value: unknown) => { + if (!(value instanceof Uint8Array)) return value as ArrayBuffer | string | number | null + const buffer = new ArrayBuffer(value.byteLength) + new Uint8Array(buffer).set(value) + return buffer + } + const storage: DurableObjectStorage = { + sql: { + exec(query: string, ...bindings: Array) { + const statement = native.query(query) + const rows = (statement.values(...(bindings as never[])) ?? []).map((row) => row.map(toSqlStorageValue)) + const columnNames = statement.columnNames + return { + columnNames, + raw: () => rows[Symbol.iterator](), + toArray: () => rows.map((row) => Object.fromEntries(columnNames.map((name, i) => [name, row[i]]))), + } + }, + }, + transaction(closure: (txn: { rollback(): void }) => Promise): Promise { + native.run("BEGIN") + let rolledBack = false + return closure({ rollback: () => (rolledBack = true) }).then( + (result) => { + native.run(rolledBack ? "ROLLBACK" : "COMMIT") + return result + }, + (error) => { + native.run("ROLLBACK") + throw error + }, + ) + }, + transactionSync(closure: () => T): T { + return native.transaction(closure)() + }, + } + return storage +} + +const run = (storage: DurableObjectStorage, effect: Effect.Effect) => + Effect.runPromise(effect.pipe(Effect.provide(sqliteLayer({ storage })), Effect.scoped)) + +describe("sqlite.workerd", () => { + test("executes statements with bindings and maps rows to records", async () => { + const rows = await run( + makeFakeStorage(), + Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient + yield* sql`CREATE TABLE item (id INTEGER PRIMARY KEY, name TEXT NOT NULL)` + yield* sql`INSERT INTO item (id, name) VALUES (${1}, ${"one"}), (${2}, ${"two"})` + return yield* sql<{ id: number; name: string }>`SELECT id, name FROM item ORDER BY id` + }), + ) + expect(rows).toEqual([ + { id: 1, name: "one" }, + { id: 2, name: "two" }, + ]) + }) + + test("normalizes ArrayBuffer blob values to Uint8Array", async () => { + const rows = await run( + makeFakeStorage(), + Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient + yield* sql`CREATE TABLE blob (data BLOB NOT NULL)` + yield* sql`INSERT INTO blob (data) VALUES (${new Uint8Array([1, 2, 3])})` + return yield* sql<{ data: Uint8Array }>`SELECT data FROM blob` + }), + ) + expect(rows[0].data).toBeInstanceOf(Uint8Array) + expect(Array.from(rows[0].data)).toEqual([1, 2, 3]) + }) + + test("withTransaction commits on success and rolls back on failure", async () => { + const storage = makeFakeStorage() + const count = await run( + storage, + Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient + yield* sql`CREATE TABLE t (value TEXT NOT NULL)` + yield* sql.withTransaction(sql`INSERT INTO t (value) VALUES (${"kept"})`) + yield* sql + .withTransaction( + Effect.gen(function* () { + yield* sql`INSERT INTO t (value) VALUES (${"discarded"})` + return yield* Effect.fail("rollback") + }), + ) + .pipe(Effect.ignore) + return yield* sql<{ count: number }>`SELECT count(*) AS count FROM t` + }), + ) + expect(count[0].count).toBe(1) + }) + + test("nested withTransaction fails with SqlError", async () => { + const error = await run( + makeFakeStorage(), + Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient + yield* sql`CREATE TABLE t (value TEXT NOT NULL)` + return yield* sql + .withTransaction(sql.withTransaction(sql`INSERT INTO t (value) VALUES (${"nested"})`)) + .pipe(Effect.flip) + }), + ) + expect(error).toBeInstanceOf(SqlError) + }) + + test("boots the full database layer with migrations over injected storage", async () => { + const storage = makeFakeStorage() + const core = await import("@opencode-ai/core/database/database") + await Effect.runPromise( + Effect.scoped( + Layer.build( + core.Database.layerFromClient.pipe(Layer.provide(sqliteLayer({ storage })), Layer.provide(tempGlobalLayer)), + ), + ), + ) + const names = storage.sql + .exec("SELECT name FROM sqlite_master WHERE type = 'table' ORDER BY name") + .toArray() + .map((row) => row.name) + expect(names).toContain("migration") + expect(names).toContain("session_v2") + }) +})