refactor(core): unify sqlite database clients

packages/core/src/database/sqlite.node.ts

@@ -1,3 +1,170 @@
-import { NodeSqliteClient } from "@opencode-ai/effect-sqlite-node"
+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"
 
-export const layer = NodeSqliteClient.layer
+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) =>
+  Layer.merge(
+    nativeLayer(config),
+    Layer.merge(sqliteLayer(config), drizzleLayer).pipe(Layer.provide(nativeLayer(config))),
+  ).pipe(Layer.provide(Reactivity.layer))