sqlite.node.ts 6.3 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178
  1. import { DatabaseSync, type SQLInputValue } from "node:sqlite"
  2. import { drizzle } from "drizzle-orm/node-sqlite"
  3. import * as Context from "effect/Context"
  4. import * as Effect from "effect/Effect"
  5. import * as Fiber from "effect/Fiber"
  6. import { identity } from "effect/Function"
  7. import * as Layer from "effect/Layer"
  8. import * as Scope from "effect/Scope"
  9. import * as Semaphore from "effect/Semaphore"
  10. import * as Stream from "effect/Stream"
  11. import * as Reactivity from "effect/unstable/reactivity/Reactivity"
  12. import * as Client from "effect/unstable/sql/SqlClient"
  13. import type { Connection } from "effect/unstable/sql/SqlConnection"
  14. import { classifySqliteError, SqlError } from "effect/unstable/sql/SqlError"
  15. import * as Statement from "effect/unstable/sql/Statement"
  16. import { Sqlite } from "./sqlite"
  17. const ATTR_DB_SYSTEM_NAME = "db.system.name"
  18. const TypeId = "~@opencode-ai/core/database/SqliteNode" as const
  19. type TypeId = typeof TypeId
  20. interface SqliteClient extends Client.SqlClient {
  21. readonly [TypeId]: TypeId
  22. readonly config: Config
  23. readonly loadExtension: (path: string) => Effect.Effect<void, SqlError>
  24. readonly updateValues: never
  25. }
  26. interface Config {
  27. readonly filename: string
  28. readonly readonly?: boolean
  29. readonly create?: boolean
  30. readonly readwrite?: boolean
  31. readonly disableWAL?: boolean
  32. readonly timeout?: number
  33. readonly allowExtension?: boolean
  34. readonly spanAttributes?: Record<string, unknown>
  35. readonly transformResultNames?: (str: string) => string
  36. readonly transformQueryNames?: (str: string) => string
  37. }
  38. interface SqliteConnection extends Connection {
  39. readonly loadExtension: (path: string) => Effect.Effect<void, SqlError>
  40. }
  41. const make = (options: Config) =>
  42. Effect.gen(function* () {
  43. const native = (yield* Sqlite.Native) as DatabaseSync
  44. const compiler = Statement.makeCompilerSqlite(options.transformQueryNames)
  45. const transformRows = options.transformResultNames
  46. ? Statement.defaultTransforms(options.transformResultNames).array
  47. : undefined
  48. const run = (query: string, params: ReadonlyArray<unknown> = []) =>
  49. Effect.withFiber<Array<Record<string, unknown>>, SqlError>((fiber) => {
  50. const statement = native.prepare(query)
  51. statement.setReadBigInts(Context.get(fiber.context, Client.SafeIntegers))
  52. try {
  53. return Effect.succeed(statement.all(...(params as SQLInputValue[])) as Array<Record<string, unknown>>)
  54. } catch (cause) {
  55. return Effect.fail(
  56. new SqlError({
  57. reason: classifySqliteError(cause, { message: "Failed to execute statement", operation: "execute" }),
  58. }),
  59. )
  60. }
  61. })
  62. const runValues = (query: string, params: ReadonlyArray<unknown> = []) =>
  63. Effect.withFiber<ReadonlyArray<ReadonlyArray<unknown>>, SqlError>((fiber) => {
  64. const statement = native.prepare(query)
  65. statement.setReadBigInts(Context.get(fiber.context, Client.SafeIntegers))
  66. statement.setReturnArrays(true)
  67. try {
  68. return Effect.succeed(
  69. statement.all(...(params as SQLInputValue[])) as unknown as ReadonlyArray<ReadonlyArray<unknown>>,
  70. )
  71. } catch (cause) {
  72. return Effect.fail(
  73. new SqlError({
  74. reason: classifySqliteError(cause, { message: "Failed to execute statement", operation: "execute" }),
  75. }),
  76. )
  77. }
  78. })
  79. const connection = identity<SqliteConnection>({
  80. execute(query, params, transformRows) {
  81. return transformRows ? Effect.map(run(query, params), transformRows) : run(query, params)
  82. },
  83. executeRaw(query, params) {
  84. return run(query, params)
  85. },
  86. executeValues(query, params) {
  87. return runValues(query, params)
  88. },
  89. executeUnprepared(query, params, transformRows) {
  90. return this.execute(query, params, transformRows)
  91. },
  92. executeStream() {
  93. return Stream.die("executeStream not implemented")
  94. },
  95. loadExtension: (path) =>
  96. Effect.try({
  97. try: () => native.loadExtension(path),
  98. catch: (cause) =>
  99. new SqlError({
  100. reason: classifySqliteError(cause, { message: "Failed to load extension", operation: "loadExtension" }),
  101. }),
  102. }),
  103. })
  104. const semaphore = yield* Semaphore.make(1)
  105. const acquirer = semaphore.withPermits(1)(Effect.succeed(connection))
  106. const transactionAcquirer = Effect.uninterruptibleMask((restore) => {
  107. const fiber = Fiber.getCurrent()!
  108. const scope = Context.getUnsafe(fiber.context, Scope.Scope)
  109. return Effect.as(
  110. Effect.tap(restore(semaphore.take(1)), () => Scope.addFinalizer(scope, semaphore.release(1))),
  111. connection,
  112. )
  113. })
  114. const client = Object.assign(
  115. (yield* Client.make({
  116. acquirer,
  117. compiler,
  118. transactionAcquirer,
  119. spanAttributes: [
  120. ...(options.spanAttributes ? Object.entries(options.spanAttributes) : []),
  121. [ATTR_DB_SYSTEM_NAME, "sqlite"],
  122. ],
  123. transformRows,
  124. })) as SqliteClient,
  125. {
  126. [TypeId]: TypeId,
  127. config: options,
  128. loadExtension: (path: string) => Effect.flatMap(acquirer, (_) => _.loadExtension(path)),
  129. },
  130. )
  131. return client
  132. })
  133. const nativeLayer = (config: Config) =>
  134. Layer.effect(
  135. Sqlite.Native,
  136. Effect.gen(function* () {
  137. const native = new DatabaseSync(config.filename, {
  138. readOnly: config.readonly,
  139. timeout: config.timeout,
  140. allowExtension: config.allowExtension,
  141. enableForeignKeyConstraints: true,
  142. open: true,
  143. })
  144. yield* Effect.addFinalizer(() => Effect.sync(() => native.close()))
  145. if (config.disableWAL !== true && config.readonly !== true) native.exec("PRAGMA journal_mode = WAL;")
  146. return native
  147. }),
  148. )
  149. const sqliteLayer = (config: Config) => Layer.effect(Client.SqlClient, make(config))
  150. const drizzleLayer = Layer.effect(
  151. Sqlite.Drizzle,
  152. Effect.gen(function* () {
  153. return drizzle({ client: (yield* Sqlite.Native) as DatabaseSync }) as unknown as Sqlite.DrizzleClient
  154. }),
  155. )
  156. export const layer = (config: Config) => {
  157. const native = nativeLayer(config)
  158. return Layer.merge(native, Layer.merge(sqliteLayer(config), drizzleLayer).pipe(Layer.provide(native))).pipe(
  159. Layer.provide(Reactivity.layer),
  160. )
  161. }