import { Context, Effect, Exit, Fiber, Layer, Scope, Semaphore } from "effect" 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 type { NativeTransactionSqlClient } from "./drizzle/effect-sqlite/session.js" import { Sqlite } from "./sqlite.js" const ATTR_DB_SYSTEM_NAME = "db.system.name" const TypeId = "~@opencode/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/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 = Sqlite.makeConnection(run, runValues, {}) 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 detect this and route through withTransaction. transactionStatements: false, } as const, ) satisfies NativeTransactionSqlClient 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.layerFromClient.pipe(Layer.provide(sqliteLayer({ storage })))", ), ) const clientLayer = (config: Config) => Layer.effect(SqlClient.SqlClient, make(config)) export const sqliteLayer = (config: Config) => { const native = nativeLayer(config) return Layer.merge(native, clientLayer(config).pipe(Layer.provide(native))).pipe(Layer.provide(Reactivity.layer)) }