openAgent

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