openAgent

1import { Database } from "bun:sqlite"
2import { drizzle } from "drizzle-orm/bun-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/SqliteBun" as const
21type TypeId = typeof TypeId
22 
23interface SqliteClient extends Client.SqlClient {
24  readonly [TypeId]: TypeId
25  readonly config: Config
26  readonly export: Effect.Effect<Uint8Array, SqlError>
27  readonly loadExtension: (path: string) => Effect.Effect<void, SqlError>
28  readonly updateValues: never
29}
30 
31interface Config {
32  readonly filename: string
33  readonly readonly?: boolean
34  readonly create?: boolean
35  readonly readwrite?: boolean
36  readonly disableWAL?: boolean
37  readonly spanAttributes?: Record<string, unknown>
38  readonly transformResultNames?: (str: string) => string
39  readonly transformQueryNames?: (str: string) => string
40}
41 
42interface SqliteConnection extends Connection {
43  readonly export: Effect.Effect<Uint8Array, SqlError>
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 Database
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.query(query)
59        // @ts-ignore bun-types missing safeIntegers method, fixed in https://github.com/oven-sh/bun/pull/26627
60        statement.safeIntegers(Context.get(fiber.context, Client.SafeIntegers))
61        try {
62          return Effect.succeed((statement.all(...(params as any)) ?? []) as Array<Record<string, unknown>>)
63        } catch (cause) {
64          return Effect.fail(
65            new SqlError({
66              reason: classifySqliteError(cause, { message: "Failed to execute statement", operation: "execute" }),
67            }),
68          )
69        }
70      })
71 
72    const runValues = (query: string, params: ReadonlyArray<unknown> = []) =>
73      Effect.withFiber<Array<unknown[]>, SqlError>((fiber) => {
74        const statement = native.query(query)
75        // @ts-ignore bun-types missing safeIntegers method, fixed in https://github.com/oven-sh/bun/pull/26627
76        statement.safeIntegers(Context.get(fiber.context, Client.SafeIntegers))
77        try {
78          return Effect.succeed((statement.values(...(params as any)) ?? []) as Array<unknown[]>)
79        } catch (cause) {
80          return Effect.fail(
81            new SqlError({
82              reason: classifySqliteError(cause, { message: "Failed to execute statement", operation: "execute" }),
83            }),
84          )
85        }
86      })
87 
88    const connection = identity<SqliteConnection>({
89      execute(query, params, transformRows) {
90        return transformRows ? Effect.map(run(query, params), transformRows) : run(query, params)
91      },
92      executeRaw(query, params) {
93        return run(query, params)
94      },
95      executeValues(query, params) {
96        return runValues(query, params)
97      },
98      executeUnprepared(query, params, transformRows) {
99        return this.execute(query, params, transformRows)
100      },
101      executeStream() {
102        return Stream.die("executeStream not implemented")
103      },
104      export: Effect.try({
105        try: () => native.serialize(),
106        catch: (cause) =>
107          new SqlError({
108            reason: classifySqliteError(cause, { message: "Failed to export database", operation: "export" }),
109          }),
110      }),
111      loadExtension: (path) =>
112        Effect.try({
113          try: () => native.loadExtension(path),
114          catch: (cause) =>
115            new SqlError({
116              reason: classifySqliteError(cause, { message: "Failed to load extension", operation: "loadExtension" }),
117            }),
118        }),
119    })
120 
121    const semaphore = yield* Semaphore.make(1)
122    const acquirer = semaphore.withPermits(1)(Effect.succeed(connection))
123    const transactionAcquirer = Effect.uninterruptibleMask((restore) => {
124      const fiber = Fiber.getCurrent()!
125      const scope = Context.getUnsafe(fiber.context, Scope.Scope)
126      return Effect.as(
127        Effect.tap(restore(semaphore.take(1)), () => Scope.addFinalizer(scope, semaphore.release(1))),
128        connection,
129      )
130    })
131 
132    const client = Object.assign(
133      (yield* Client.make({
134        acquirer,
135        compiler,
136        transactionAcquirer,
137        spanAttributes: [
138          ...(options.spanAttributes ? Object.entries(options.spanAttributes) : []),
139          [ATTR_DB_SYSTEM_NAME, "sqlite"],
140        ],
141        transformRows,
142      })) as SqliteClient,
143      {
144        [TypeId]: TypeId,
145        config: options,
146        export: Effect.flatMap(acquirer, (_) => _.export),
147        loadExtension: (path: string) => Effect.flatMap(acquirer, (_) => _.loadExtension(path)),
148      },
149    )
150 
151    return client
152  })
153 
154const nativeLayer = (config: Config) =>
155  Layer.effect(
156    Sqlite.Native,
157    Effect.gen(function* () {
158      const native = new Database(config.filename, {
159        readonly: config.readonly,
160        readwrite: config.readwrite ?? true,
161        create: config.create ?? true,
162      })
163      yield* Effect.addFinalizer(() => Effect.sync(() => native.close()))
164      if (config.disableWAL !== true) native.run("PRAGMA journal_mode = WAL;")
165      return native
166    }),
167  )
168 
169const sqliteLayer = (config: Config) => Layer.effect(Client.SqlClient, make(config))
170 
171const drizzleLayer = Layer.effect(
172  Sqlite.Drizzle,
173  Effect.gen(function* () {
174    return drizzle({ client: (yield* Sqlite.Native) as Database })
175  }),
176)
177 
178export const layer = (config: Config) => {
179  const native = nativeLayer(config)
180  return Layer.merge(native, Layer.merge(sqliteLayer(config), drizzleLayer).pipe(Layer.provide(native))).pipe(
181    Layer.provide(Reactivity.layer),
182  )
183}
184