Open History to see the last commit
1/**
2 * Connects Effect SQL to SQLite when running on Bun, using `bun:sqlite`.
3 *
4 * This module opens a SQLite database and exposes it as both `SqliteClient` and
5 * the generic Effect SQL client. It serializes access to the database, enables
6 * WAL mode unless disabled, and waits up to five seconds for busy databases by
7 * default. Explicit transactions on writable connections use `BEGIN IMMEDIATE`
8 * to avoid read-to-write lock upgrades, which serializes them behind other
9 * writers even when they only read. Clients opened with `readonly: true` are
10 * unaffected. Busy waits block the event loop because `bun:sqlite` is
11 * synchronous. Database export and extension loading are supported; streaming
12 * queries and `updateValues` are not.
13 *
14 * @stability unstable
15 * @since 4.0.0
16 */
17import { constants, Database } from "bun:sqlite"
18import * as Clock from "effect/Clock"
19import * as Config from "effect/Config"
20import * as Context from "effect/Context"
21import * as Duration from "effect/Duration"
22import * as Effect from "effect/Effect"
23import { identity } from "effect/Function"
24import * as Layer from "effect/Layer"
25import * as Reactivity from "effect/reactivity/Reactivity"
26import * as Schedule from "effect/Schedule"
27import type * as Scope from "effect/Scope"
28import * as Semaphore from "effect/Semaphore"
29import * as Client from "effect/sql/SqlClient"
30import type { Connection } from "effect/sql/SqlConnection"
31import { classifySqliteError, SqlError } from "effect/sql/SqlError"
32import * as Statement from "effect/sql/Statement"
33import * as Stream from "effect/Stream"
34
35const ATTR_DB_SYSTEM_NAME = "db.system.name"
36const MAX_BUSY_TIMEOUT = 2_147_483_647
37const NANOS_PER_MILLI = BigInt(1_000_000)
38
39const classifyError = (cause: unknown, message: string, operation: string) =>
40 classifySqliteError(cause, { message, operation })
41
42/**
43 * Runtime type identifier used to mark Bun `SqliteClient` values.
44 *
45 * @stability unstable
46 * @category type IDs
47 * @since 4.0.0
48 */
49export const TypeId: TypeId = "~@effect/sql-sqlite-bun/SqliteClient"
50
51/**
52 * Type-level identifier used to mark Bun `SqliteClient` values.
53 *
54 * @stability unstable
55 * @category type IDs
56 * @since 4.0.0
57 */
58export type TypeId = "~@effect/sql-sqlite-bun/SqliteClient"
59
60/**
61 * Bun SQLite client service, extending `SqlClient` with database export and extension loading helpers. `updateValues` is not supported.
62 *
63 * @stability unstable
64 * @category services
65 * @since 4.0.0
66 */
67export interface SqliteClient extends Client.SqlClient {
68 readonly [TypeId]: TypeId
69 readonly config: SqliteClientConfig
70 readonly export: Effect.Effect<Uint8Array, SqlError>
71 readonly loadExtension: (path: string) => Effect.Effect<void, SqlError>
72
73 /** Not supported in sqlite */
74 readonly updateValues: never
75}
76
77/**
78 * Service tag for the Bun SQLite client service.
79 *
80 * **When to use**
81 *
82 * Use to access or provide a Bun SQLite client through the Effect context.
83 *
84 * @stability unstable
85 * @category services
86 * @since 4.0.0
87 */
88export const SqliteClient = Context.Service<SqliteClient>("@effect/sql-sqlite-bun/Client")
89
90/**
91 * Configuration for a Bun SQLite client, including filename, open mode flags, WAL and busy timeout behavior, span attributes, and query/result name transforms.
92 *
93 * @stability unstable
94 * @category models
95 * @since 4.0.0
96 */
97export interface SqliteClientConfig {
98 readonly filename: string
99 readonly readonly?: boolean | undefined
100 readonly create?: boolean | undefined
101 readonly readwrite?: boolean | undefined
102 readonly disableWAL?: boolean | undefined
103 /**
104 * How long SQLite waits when the database is busy. Defaults to 5 seconds.
105 * `Duration.infinity` is clamped to SQLite's maximum timeout.
106 * Waiting blocks the event loop because `bun:sqlite` is synchronous.
107 */
108 readonly busyTimeout?: Duration.Input | undefined
109
110 readonly spanAttributes?: Record<string, unknown> | undefined
111
112 readonly transformResultNames?: ((str: string) => string) | undefined
113 readonly transformQueryNames?: ((str: string) => string) | undefined
114}
115
116interface SqliteConnection extends Connection {
117 readonly isTransaction: () => boolean
118 readonly export: Effect.Effect<Uint8Array, SqlError>
119 readonly loadExtension: (path: string) => Effect.Effect<void, SqlError>
120}
121
122/**
123 * Creates a scoped Bun SQLite client for a database file, enabling WAL and a 5-second busy timeout by default. Explicit transactions on writable connections take the write lock for their duration, even when they only read; clients opened with `readonly: true` are unaffected. Streaming queries are not implemented.
124 *
125 * @stability unstable
126 * @category constructors
127 * @since 4.0.0
128 */
129export const make = (
130 options: SqliteClientConfig
131): Effect.Effect<SqliteClient, SqlError, Scope.Scope | Reactivity.Reactivity> =>
132 Effect.gen(function*() {
133 const compiler = Statement.makeCompilerSqlite(options.transformQueryNames)
134 const transformRows = options.transformResultNames ?
135 Statement.defaultTransforms(
136 options.transformResultNames
137 ).array :
138 undefined
139
140 const makeConnection = Effect.gen(function*() {
141 const readonly = options.readonly === true
142 const openOptions = {
143 readonly,
144 readwrite: readonly ? false : options.readwrite ?? true,
145 create: readonly ? false : options.create ?? true
146 }
147 const db = yield* Effect.acquireRelease(
148 Effect.try({
149 try: () =>
150 new Database(
151 options.filename,
152 options.filename.startsWith("file:")
153 ? constants.SQLITE_OPEN_URI |
154 (openOptions.readonly ? constants.SQLITE_OPEN_READONLY : 0) |
155 (openOptions.readwrite || openOptions.create ? constants.SQLITE_OPEN_READWRITE : 0) |
156 (openOptions.create ? constants.SQLITE_OPEN_CREATE : 0)
157 : openOptions
158 ),
159 catch: (cause) => new SqlError({ reason: classifyError(cause, "Failed to open database", "connect") })
160 }),
161 (db) => Effect.sync(() => db.close())
162 )
163 const busyTimeout = Math.min(
164 MAX_BUSY_TIMEOUT,
165 Math.max(0, Math.round(Duration.toMillis(options.busyTimeout ?? Duration.seconds(5))))
166 )
167 const exec = (sql: string) =>
168 Effect.try({
169 try: () => db.run(sql),
170 catch: (cause) => new SqlError({ reason: classifyError(cause, "Failed to configure database", "connect") })
171 })
172 yield* exec(`PRAGMA busy_timeout = ${busyTimeout};`)
173
174 if (options.disableWAL !== true && !readonly) {
175 // Switching to WAL can fail with SQLITE_BUSY without invoking the busy
176 // handler, so retry it until busyTimeout has elapsed. Each attempt's
177 // native busy wait is limited to the remaining time.
178 const deadline = (yield* Clock.monotonicTimeNanos) + BigInt(busyTimeout) * NANOS_PER_MILLI
179 const remainingMillis = Effect.map(
180 Clock.monotonicTimeNanos,
181 (now) => Math.max(0, Number((deadline - now) / NANOS_PER_MILLI))
182 )
183 yield* remainingMillis.pipe(
184 Effect.flatMap((remaining) => exec(`PRAGMA busy_timeout = ${remaining}`)),
185 Effect.andThen(exec("PRAGMA journal_mode = WAL;")),
186 Effect.retry({
187 while: (error) =>
188 error.reason._tag === "LockTimeoutError" && Effect.map(remainingMillis, (remaining) => remaining > 0),
189 schedule: Schedule.spaced("10 millis")
190 })
191 )
192 yield* exec(`PRAGMA busy_timeout = ${busyTimeout}`)
193 }
194
195 const prepare = (sql: string, useSafeIntegers: boolean) => {
196 const statement = db.query(sql)
197 // @ts-ignore bun-types missing safeIntegers method, fixed in https://github.com/oven-sh/bun/pull/26627
198 statement.safeIntegers(useSafeIntegers)
199 return statement
200 }
201
202 const run = (
203 sql: string,
204 params: ReadonlyArray<unknown> = []
205 ) =>
206 Effect.withFiber<Array<any>, SqlError>((fiber) => {
207 const useSafeIntegers = Context.get(fiber.context, Client.SafeIntegers)
208 try {
209 return Effect.succeed((prepare(sql, useSafeIntegers).all(...(params as any)) ?? []) as Array<any>)
210 } catch (cause) {
211 return Effect.fail(new SqlError({ reason: classifyError(cause, "Failed to execute statement", "execute") }))
212 }
213 })
214
215 const runValues = (
216 sql: string,
217 params: ReadonlyArray<unknown> = []
218 ) =>
219 Effect.withFiber<Array<any>, SqlError>((fiber) => {
220 const useSafeIntegers = Context.get(fiber.context, Client.SafeIntegers)
221 try {
222 return Effect.succeed((prepare(sql, useSafeIntegers).values(...(params as any)) ?? []) as Array<any>)
223 } catch (cause) {
224 return Effect.fail(
225 new SqlError({ reason: classifyError(cause, "Failed to execute statement", "executeValues") })
226 )
227 }
228 })
229
230 return identity<SqliteConnection>({
231 isTransaction: () => db.inTransaction,
232 execute(sql, params, transformRows) {
233 return transformRows
234 ? Effect.map(run(sql, params), transformRows)
235 : run(sql, params)
236 },
237 executeRaw(sql, params) {
238 return run(sql, params)
239 },
240 executeValues(sql, params) {
241 return runValues(sql, params)
242 },
243 executeValuesUnprepared(sql, params) {
244 return runValues(sql, params)
245 },
246 executeUnprepared(sql, params, transformRows) {
247 return this.execute(sql, params, transformRows)
248 },
249 executeStream(_sql, _params) {
250 return Stream.die("executeStream not implemented")
251 },
252 export: Effect.try({
253 try: () => db.serialize(),
254 catch: (cause) => new SqlError({ reason: classifyError(cause, "Failed to export database", "export") })
255 }),
256 loadExtension: (path) =>
257 Effect.try({
258 try: () => db.loadExtension(path),
259 catch: (cause) =>
260 new SqlError({ reason: classifyError(cause, "Failed to load extension", "loadExtension") })
261 })
262 })
263 })
264
265 const connection = yield* makeConnection
266 const { acquirer, borrower, isTransaction, onCommitFailure, transactionAcquirer } = Client.makeSqliteAcquirers({
267 connection: Effect.succeed(connection),
268 semaphore: yield* Semaphore.make(1),
269 isTransaction: (conn) => conn.isTransaction()
270 })
271
272 const client: SqliteClient = Object.assign(
273 (yield* Client.make({
274 driver: "sqlite-bun",
275 acquirer,
276 borrower,
277 compiler,
278 transactionAcquirer,
279 onCommitFailure,
280 isTransaction,
281 releaseSavepoint: (name) => `RELEASE SAVEPOINT ${name}`,
282 beginTransaction: options.readonly === true ? "BEGIN" : "BEGIN IMMEDIATE",
283 spanAttributes: [
284 ...(options.spanAttributes ? Object.entries(options.spanAttributes) : []),
285 [ATTR_DB_SYSTEM_NAME, "sqlite"]
286 ],
287 transformRows
288 })) as SqliteClient,
289 {
290 [TypeId]: TypeId as TypeId,
291 config: options,
292 export: Effect.withFiber((fiber) =>
293 Context.getOption(fiber.context, client.transactionService)._tag === "Some"
294 ? connection.export
295 : Effect.scoped(Effect.flatMap(acquirer, (_) => _.export))
296 ),
297 loadExtension: (path: string) => Effect.scoped(Effect.flatMap(acquirer, (_) => _.loadExtension(path)))
298 }
299 )
300 return client
301 })
302
303/**
304 * Creates a layer from a `Config`-wrapped Bun SQLite client configuration, providing both `SqliteClient` and `SqlClient`.
305 *
306 * @stability unstable
307 * @category layers
308 * @since 4.0.0
309 */
310export const layerConfig = (
311 config: Config.Wrap<SqliteClientConfig>
312): Layer.Layer<SqliteClient | Client.SqlClient, Config.ConfigError | SqlError> =>
313 Layer.effectContext(
314 Config.unwrap(config).pipe(
315 Effect.flatMap(make),
316 Effect.map((client) =>
317 Context.make(SqliteClient, client).pipe(
318 Context.add(Client.SqlClient, client)
319 )
320 )
321 )
322 ).pipe(Layer.provide(Reactivity.layer))
323
324/**
325 * Creates a layer from a concrete Bun SQLite client configuration, providing both `SqliteClient` and `SqlClient`.
326 *
327 * @stability unstable
328 * @category layers
329 * @since 4.0.0
330 */
331export const layer = (
332 config: SqliteClientConfig
333): Layer.Layer<SqliteClient | Client.SqlClient, SqlError> =>
334 Layer.effectContext(
335 Effect.map(make(config), (client) =>
336 Context.make(SqliteClient, client).pipe(
337 Context.add(Client.SqlClient, client)
338 ))
339 ).pipe(Layer.provide(Reactivity.layer))