Something went wrong. Try again.
opencode v2 mirror
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225import { 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 consttype 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<string> raw(): IterableIterator<Array<SqlStorageValue>> toArray(): Array<Record<string, SqlStorageValue>>}
export interface SqlStorage { exec(query: string, ...bindings: Array<unknown>): SqlStorageCursor}
export interface DurableObjectStorage { readonly sql: SqlStorage transaction<T>(closure: (txn: { rollback(): void }) => Promise<T>): Promise<T> transactionSync<T>(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<string, unknown> 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<SqlClient.TransactionConnection, SqlClient.TransactionConnection.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"] => <A, E, R>(effect: Effect.Effect<A, E, R>): Effect.Effect<A, E | SqlError, R> => 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<void>((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<unknown> = []) { const cursor = native.sql.exec(query, ...params) const columns = cursor.columnNames for (const row of cursor.raw()) { const record: Record<string, unknown> = {} 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<unknown> = []) => 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<unknown> = []) => 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))}