import { DatabaseSync, type SQLInputValue } from "node:sqlite" import { drizzle } from "drizzle-orm/node-sqlite" import * as Context from "effect/Context" import * as Effect from "effect/Effect" import * as Fiber from "effect/Fiber" import { identity } from "effect/Function" import * as Layer from "effect/Layer" import * as Scope from "effect/Scope" import * as Semaphore from "effect/Semaphore" import * as Stream from "effect/Stream" import * as Reactivity from "effect/unstable/reactivity/Reactivity" import * as Client from "effect/unstable/sql/SqlClient" import type { Connection } from "effect/unstable/sql/SqlConnection" import { classifySqliteError, SqlError } from "effect/unstable/sql/SqlError" import * as Statement from "effect/unstable/sql/Statement" import { Sqlite } from "./sqlite" const ATTR_DB_SYSTEM_NAME = "db.system.name" const TypeId = "~@opencode-ai/core/database/SqliteNode" as const type TypeId = typeof TypeId interface SqliteClient extends Client.SqlClient { readonly [TypeId]: TypeId readonly config: Config readonly loadExtension: (path: string) => Effect.Effect readonly updateValues: never } interface Config { readonly filename: string readonly readonly?: boolean readonly create?: boolean readonly readwrite?: boolean readonly disableWAL?: boolean readonly timeout?: number readonly allowExtension?: boolean readonly spanAttributes?: Record readonly transformResultNames?: (str: string) => string readonly transformQueryNames?: (str: string) => string } interface SqliteConnection extends Connection { readonly loadExtension: (path: string) => Effect.Effect } const make = (options: Config) => Effect.gen(function* () { const native = (yield* Sqlite.Native) as DatabaseSync const compiler = Statement.makeCompilerSqlite(options.transformQueryNames) const transformRows = options.transformResultNames ? Statement.defaultTransforms(options.transformResultNames).array : undefined const run = (query: string, params: ReadonlyArray = []) => Effect.withFiber>, SqlError>((fiber) => { const statement = native.prepare(query) statement.setReadBigInts(Context.get(fiber.context, Client.SafeIntegers)) try { return Effect.succeed(statement.all(...(params as SQLInputValue[])) as Array>) } catch (cause) { return Effect.fail( new SqlError({ reason: classifySqliteError(cause, { message: "Failed to execute statement", operation: "execute" }), }), ) } }) const runValues = (query: string, params: ReadonlyArray = []) => Effect.withFiber>, SqlError>((fiber) => { const statement = native.prepare(query) statement.setReadBigInts(Context.get(fiber.context, Client.SafeIntegers)) statement.setReturnArrays(true) try { return Effect.succeed( statement.all(...(params as SQLInputValue[])) as unknown as ReadonlyArray>, ) } catch (cause) { return Effect.fail( 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) }, executeUnprepared(query, params, transformRows) { return this.execute(query, params, transformRows) }, executeStream() { return Stream.die("executeStream not implemented") }, loadExtension: (path) => Effect.try({ try: () => native.loadExtension(path), catch: (cause) => new SqlError({ reason: classifySqliteError(cause, { message: "Failed to load extension", operation: "loadExtension" }), }), }), }) 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* Client.make({ acquirer, compiler, transactionAcquirer, spanAttributes: [ ...(options.spanAttributes ? Object.entries(options.spanAttributes) : []), [ATTR_DB_SYSTEM_NAME, "sqlite"], ], transformRows, })) as SqliteClient, { [TypeId]: TypeId, config: options, loadExtension: (path: string) => Effect.flatMap(acquirer, (_) => _.loadExtension(path)), }, ) return client }) const nativeLayer = (config: Config) => Layer.effect( Sqlite.Native, Effect.gen(function* () { const native = new DatabaseSync(config.filename, { readOnly: config.readonly, timeout: config.timeout, allowExtension: config.allowExtension, enableForeignKeyConstraints: true, open: true, }) yield* Effect.addFinalizer(() => Effect.sync(() => native.close())) if (config.disableWAL !== true && config.readonly !== true) native.exec("PRAGMA journal_mode = WAL;") return native }), ) const sqliteLayer = (config: Config) => Layer.effect(Client.SqlClient, make(config)) const drizzleLayer = Layer.effect( Sqlite.Drizzle, Effect.gen(function* () { return drizzle({ client: (yield* Sqlite.Native) as DatabaseSync }) as unknown as Sqlite.DrizzleClient }), ) export const layer = (config: Config) => { const native = nativeLayer(config) return Layer.merge(native, Layer.merge(sqliteLayer(config), drizzleLayer).pipe(Layer.provide(native))).pipe( Layer.provide(Reactivity.layer), ) }