| 123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178 |
- 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<void, SqlError>
- 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<string, unknown>
- readonly transformResultNames?: (str: string) => string
- readonly transformQueryNames?: (str: string) => string
- }
- interface SqliteConnection extends Connection {
- readonly loadExtension: (path: string) => Effect.Effect<void, SqlError>
- }
- 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<unknown> = []) =>
- Effect.withFiber<Array<Record<string, unknown>>, 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<Record<string, unknown>>)
- } catch (cause) {
- return Effect.fail(
- new SqlError({
- reason: classifySqliteError(cause, { message: "Failed to execute statement", operation: "execute" }),
- }),
- )
- }
- })
- const runValues = (query: string, params: ReadonlyArray<unknown> = []) =>
- Effect.withFiber<ReadonlyArray<ReadonlyArray<unknown>>, 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<ReadonlyArray<unknown>>,
- )
- } catch (cause) {
- return Effect.fail(
- new SqlError({
- reason: classifySqliteError(cause, { message: "Failed to execute statement", operation: "execute" }),
- }),
- )
- }
- })
- const connection = identity<SqliteConnection>({
- 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),
- )
- }
|