1/**
2 * Connects Effect SQL to SQLite compiled to WebAssembly with
3 * `@effect/wa-sqlite`.
4 *
5 * This module can create an in-memory SQLite database in the current runtime or
6 * connect to a worker-backed database, such as the OPFS worker from
7 * `OpfsWorker`. Both clients are exposed as `SqliteClient` and the generic
8 * Effect SQL client, serialize access, support database import and export, and
9 * can install reactivity hooks from SQLite update notifications. In-memory
10 * clients can stream query rows; worker-backed clients cannot. `updateValues`
11 * is not supported by this driver.
12 *
13 * @since 4.0.0
14 */
15// oxlint-disable-next-line effect/no-import-from-barrel-package
16import * as WaSqlite from "@effect/wa-sqlite"
17import SQLiteESMFactory from "@effect/wa-sqlite/dist/wa-sqlite.mjs"
18import { MemoryVFS } from "@effect/wa-sqlite/src/examples/MemoryVFS.js"
19import * as Config from "effect/Config"
20import * as Context from "effect/Context"
21import * as Deferred from "effect/Deferred"
22import * as Effect from "effect/Effect"
23import * as Exit from "effect/Exit"
24import { identity } from "effect/Function"
25import * as Layer from "effect/Layer"
26import * as Reactivity from "effect/reactivity/Reactivity"
27import * as Rec from "effect/Record"
28import * as Scope from "effect/Scope"
29import * as ScopedRef from "effect/ScopedRef"
30import * as Semaphore from "effect/Semaphore"
31import * as Client from "effect/sql/SqlClient"
32import type { Connection } from "effect/sql/SqlConnection"
33import { classifySqliteError, SqlError } from "effect/sql/SqlError"
34import * as Statement from "effect/sql/Statement"
35import * as Stream from "effect/Stream"
36import type { OpfsWorkerMessage } from "./internal/opfsWorker.ts"
37
38const ATTR_DB_SYSTEM_NAME = "db.system.name"
39
40const classifyError = (cause: unknown, message: string, operation: string) =>
41 classifySqliteError(cause, { message, operation })
42
43/**
44 * Runtime identifier attached to SQLite WASM client values.
45 *
46 * @category type IDs
47 * @since 4.0.0
48 */
49export const TypeId: TypeId = "~@effect/sql-sqlite-wasm/SqliteClient"
50
51/**
52 * Type-level identifier for SQLite WASM client values.
53 *
54 * @category type IDs
55 * @since 4.0.0
56 */
57export type TypeId = "~@effect/sql-sqlite-wasm/SqliteClient"
58
59/**
60 * SQLite WASM client service interface, extending `SqlClient` with database `export` and `import` operations and marking `updateValues` as unsupported for SQLite.
61 *
62 * @category services
63 * @since 4.0.0
64 */
65export interface SqliteClient extends Client.SqlClient {
66 readonly [TypeId]: TypeId
67 readonly config: SqliteClientMemoryConfig
68 readonly export: Effect.Effect<Uint8Array, SqlError>
69 readonly import: (data: Uint8Array) => Effect.Effect<void, SqlError>
70
71 /** Not supported in sqlite */
72 readonly updateValues: never
73}
74
75/**
76 * Service tag for the SQLite WASM client.
77 *
78 * @category services
79 * @since 4.0.0
80 */
81export const SqliteClient = Context.Service<SqliteClient>("@effect/sql-sqlite-wasm/SqliteClient")
82
83/**
84 * Configuration for an in-memory SQLite WASM client, including optional reactivity hooks, span attributes, and query/result name transforms.
85 *
86 * @category models
87 * @since 4.0.0
88 */
89export interface SqliteClientMemoryConfig {
90 readonly installReactivityHooks?: boolean
91 readonly spanAttributes?: Record<string, unknown>
92 readonly transformResultNames?: (str: string) => string
93 readonly transformQueryNames?: (str: string) => string
94}
95
96/**
97 * Configuration for a worker-backed SQLite WASM client, including the scoped worker or message port, optional reactivity hooks, span attributes, and query/result name transforms.
98 *
99 * @category models
100 * @since 4.0.0
101 */
102export interface SqliteClientConfig {
103 readonly worker: Effect.Effect<Worker | SharedWorker | MessagePort, never, Scope.Scope>
104 readonly installReactivityHooks?: boolean
105 readonly spanAttributes?: Record<string, unknown>
106 readonly transformResultNames?: (str: string) => string
107 readonly transformQueryNames?: (str: string) => string
108}
109
110interface SqliteConnection extends Connection {
111 readonly export: Effect.Effect<Uint8Array, SqlError>
112 readonly import: (data: Uint8Array) => Effect.Effect<void, SqlError>
113}
114
115interface SqliteMemoryConnection extends SqliteConnection {
116 readonly isTransaction: () => boolean
117}
118
119const initModule = Effect.runSync(
120 Effect.cached(Effect.promise(() => SQLiteESMFactory()))
121)
122
123const initEffect = Effect.runSync(
124 Effect.cached(initModule.pipe(Effect.map((module) => WaSqlite.Factory(module))))
125)
126
127const registered = new Set<string>()
128
129/**
130 * Creates a scoped in-memory SQLite WASM client using the memory VFS, serializing access through a semaphore and exposing database `export` and `import` operations.
131 *
132 * @category constructors
133 * @since 4.0.0
134 */
135export const makeMemory = (
136 options: SqliteClientMemoryConfig
137): Effect.Effect<SqliteClient, SqlError, Scope.Scope | Reactivity.Reactivity> =>
138 Effect.gen(function*() {
139 const reactivity = yield* Reactivity.Reactivity
140 const compiler = Statement.makeCompilerSqlite(options.transformQueryNames)
141 const transformRows = options.transformResultNames ?
142 Statement.defaultTransforms(
143 options.transformResultNames
144 ).array :
145 undefined
146
147 const makeConnection = Effect.gen(function*() {
148 const sqlite3 = yield* initEffect
149
150 if (registered.has("memory-vfs") === false) {
151 registered.add("memory-vfs")
152 const module = yield* initModule
153 // @ts-expect-error
154 const vfs = new MemoryVFS("memory-vfs", module)
155 sqlite3.vfs_register(vfs as any, false)
156 }
157 const db = yield* Effect.acquireRelease(
158 Effect.try({
159 try: () => sqlite3.open_v2(":memory:", undefined, "memory-vfs"),
160 catch: (cause) => new SqlError({ reason: classifyError(cause, "Failed to open database", "openDatabase") })
161 }),
162 (db) => Effect.sync(() => sqlite3.close(db))
163 )
164
165 if (options.installReactivityHooks) {
166 sqlite3.update_hook(db, (_op, _db, table, rowid) => {
167 if (!table) return
168 const id = String(Number(rowid))
169 reactivity.invalidateUnsafe({ [table]: [id] })
170 })
171 }
172
173 const run = (
174 sql: string,
175 params: ReadonlyArray<unknown> = [],
176 rowMode: "object" | "array" = "object"
177 ) =>
178 Effect.try({
179 try: () => {
180 const results: Array<any> = []
181 for (const stmt of sqlite3.statements(db, sql)) {
182 let columns: Array<string> | undefined
183 sqlite3.bind_collection(stmt, params as any)
184 while (sqlite3.step(stmt) === WaSqlite.SQLITE_ROW) {
185 columns = columns ?? sqlite3.column_names(stmt)
186 const row = sqlite3.row(stmt)
187 if (rowMode === "object") {
188 const obj: Record<string, any> = {}
189 for (let i = 0; i < columns.length; i++) {
190 Rec.assignProperty(obj, columns[i], row[i])
191 }
192 results.push(obj)
193 } else {
194 results.push(row)
195 }
196 }
197 }
198 return results
199 },
200 catch: (cause) => new SqlError({ reason: classifyError(cause, "Failed to execute statement", "execute") })
201 })
202
203 return identity<SqliteMemoryConnection>({
204 isTransaction: () => sqlite3.get_autocommit(db) === 0,
205 execute(sql, params, transformRows) {
206 return transformRows
207 ? Effect.map(run(sql, params), transformRows)
208 : run(sql, params)
209 },
210 executeRaw(sql, params) {
211 return run(sql, params)
212 },
213 executeValues(sql, params) {
214 return run(sql, params, "array")
215 },
216 executeValuesUnprepared(sql, params) {
217 return run(sql, params, "array")
218 },
219 executeUnprepared(sql, params, transformRows) {
220 return this.execute(sql, params, transformRows)
221 },
222 executeStream(sql, params, transformRows) {
223 function* stream() {
224 for (const stmt of sqlite3.statements(db, sql)) {
225 let columns: Array<string> | undefined
226 sqlite3.bind_collection(stmt, params as any)
227 while (sqlite3.step(stmt) === WaSqlite.SQLITE_ROW) {
228 columns = columns ?? sqlite3.column_names(stmt)
229 const row = sqlite3.row(stmt)
230 const obj: Record<string, any> = {}
231 for (let i = 0; i < columns.length; i++) {
232 Rec.assignProperty(obj, columns[i], row[i])
233 }
234 yield obj
235 }
236 }
237 }
238 return Stream.suspend(() => Stream.fromIteratorSucceed(stream()[Symbol.iterator]())).pipe(
239 transformRows
240 ? Stream.mapArray((chunk) => transformRows(chunk) as any)
241 : identity,
242 Stream.mapError((cause) =>
243 new SqlError({ reason: classifyError(cause, "Failed to execute statement", "stream") })
244 )
245 )
246 },
247 export: Effect.try({
248 try: () => sqlite3.serialize(db, "main"),
249 catch: (cause) => new SqlError({ reason: classifyError(cause, "Failed to export database", "export") })
250 }),
251 import(data) {
252 return Effect.try({
253 try: () => sqlite3.deserialize(db, "main", data, data.length, data.length, 1 | 2),
254 catch: (cause) => new SqlError({ reason: classifyError(cause, "Failed to import database", "import") })
255 })
256 }
257 })
258 })
259
260 const { acquirer, onCommitFailure, transactionAcquirer } = Client.makeSqliteAcquirers({
261 connection: Effect.succeed(yield* makeConnection),
262 semaphore: yield* Semaphore.make(1),
263 isTransaction: (conn) => conn.isTransaction()
264 })
265
266 return Object.assign(
267 (yield* Client.make({
268 acquirer,
269 compiler,
270 transactionAcquirer,
271 onCommitFailure,
272 releaseSavepoint: (name) => `RELEASE SAVEPOINT ${name}`,
273 spanAttributes: [
274 ...(options.spanAttributes ? Object.entries(options.spanAttributes) : []),
275 [ATTR_DB_SYSTEM_NAME, "sqlite"]
276 ],
277 transformRows
278 })) as SqliteClient,
279 {
280 [TypeId]: TypeId as TypeId,
281 config: options,
282 export: Effect.scoped(Effect.flatMap(transactionAcquirer, (conn) => conn.export)),
283 import(data: Uint8Array) {
284 return Effect.scoped(Effect.flatMap(transactionAcquirer, (conn) => conn.import(data)))
285 }
286 }
287 )
288 })
289
290/**
291 * Creates a scoped worker-backed SQLite WASM client, communicating with the configured worker or message port, restarting the scoped connection on worker errors, and exposing database `export` and `import` operations.
292 *
293 * @category constructors
294 * @since 4.0.0
295 */
296export const make = (
297 options: SqliteClientConfig
298): Effect.Effect<SqliteClient, SqlError, Scope.Scope | Reactivity.Reactivity> =>
299 Effect.gen(function*() {
300 const reactivity = yield* Reactivity.Reactivity
301 const compiler = Statement.makeCompilerSqlite(options.transformQueryNames)
302 const transformRows = options.transformResultNames ?
303 Statement.defaultTransforms(options.transformResultNames).array :
304 undefined
305 const makeConnection = Effect.gen(function*() {
306 let currentId = 0
307 const pending = new Map<number, (effect: Exit.Exit<any, SqlError>) => void>()
308 const scope = yield* Effect.scope
309 const readyDeferred = yield* Deferred.make<void>()
310
311 const worker = yield* options.worker
312 const port = "port" in worker ? worker.port : worker
313 const postMessage = (message: OpfsWorkerMessage, transferables?: ReadonlyArray<any>) =>
314 port.postMessage(message, transferables as any)
315
316 yield* Scope.addFinalizer(scope, Effect.sync(() => postMessage(["close"])))
317
318 const onMessage = (event: any) => {
319 const [id, error, results] = event.data
320 if (id === "ready") {
321 Deferred.doneUnsafe(readyDeferred, Exit.void)
322 return
323 } else if (id === "update_hook") {
324 reactivity.invalidateUnsafe({ [error]: [results] })
325 return
326 } else {
327 const resume = pending.get(id)
328 if (!resume) return
329 pending.delete(id)
330 if (error) {
331 resume(
332 Exit.fail(
333 new SqlError({ reason: classifyError(error, "Failed to execute statement", "execute") })
334 )
335 )
336 } else {
337 resume(Exit.succeed(results))
338 }
339 }
340 }
341 port.addEventListener("message", onMessage)
342 if ("start" in port) {
343 port.start()
344 }
345
346 function onError(cause: Event) {
347 const exit = Exit.fail(
348 new SqlError({ reason: classifyError(cause, "SQLite WASM worker failed", "worker") })
349 )
350 const requests = Array.from(pending.values())
351 pending.clear()
352 for (const resume of requests) {
353 resume(exit)
354 }
355 Effect.runFork(ScopedRef.set(connectionRef, makeConnection))
356 }
357 if ("onerror" in worker) {
358 worker.addEventListener("error", onError)
359 }
360
361 yield* Scope.addFinalizer(
362 scope,
363 Effect.sync(() => {
364 port.removeEventListener("message", onMessage)
365 worker.removeEventListener("error", onError)
366 })
367 )
368
369 yield* Deferred.await(readyDeferred)
370
371 if (options.installReactivityHooks) {
372 postMessage(["update_hook"])
373 }
374
375 const send = (id: number, message: OpfsWorkerMessage, transferables?: ReadonlyArray<any>) =>
376 Effect.callback<any, SqlError>((resume) => {
377 pending.set(id, resume)
378 postMessage(message, transferables)
379 })
380
381 const run = (
382 sql: string,
383 params: ReadonlyArray<unknown> = [],
384 rowMode: "object" | "array" = "object"
385 ): Effect.Effect<Array<any>, SqlError, never> => {
386 const rows = Effect.withFiber<WorkerResult, SqlError>((fiber) => {
387 const id = currentId++
388 return send(id, [id, sql, params], fiber.getRef(Transferables))
389 })
390 return rowMode === "object"
391 ? Effect.map(rows, extractObject)
392 : Effect.map(rows, extractRows)
393 }
394
395 return identity<SqliteConnection>({
396 execute(sql, params, transformRows) {
397 return transformRows
398 ? Effect.map(run(sql, params), transformRows)
399 : run(sql, params)
400 },
401 executeRaw(sql, params) {
402 return run(sql, params)
403 },
404 executeValues(sql, params) {
405 return run(sql, params, "array")
406 },
407 executeValuesUnprepared(sql, params) {
408 return run(sql, params, "array")
409 },
410 executeUnprepared(sql, params, transformRows) {
411 return this.execute(sql, params, transformRows)
412 },
413 executeStream() {
414 return Stream.die("executeStream not implemented")
415 },
416 export: Effect.suspend(() => {
417 const id = currentId++
418 return send(id, ["export", id])
419 }),
420 import(data) {
421 return Effect.suspend(() => {
422 const id = currentId++
423 return send(id, ["import", id, data], [data.buffer])
424 })
425 }
426 })
427 })
428
429 const connectionRef = yield* ScopedRef.fromAcquire(makeConnection)
430
431 const { acquirer, onCommitFailure, transactionAcquirer } = Client.makeSqliteAcquirers({
432 connection: ScopedRef.get(connectionRef),
433 semaphore: yield* Semaphore.make(1)
434 })
435
436 return Object.assign(
437 (yield* Client.make({
438 acquirer,
439 compiler,
440 transactionAcquirer,
441 onCommitFailure,
442 releaseSavepoint: (name) => `RELEASE SAVEPOINT ${name}`,
443 spanAttributes: [
444 ...(options.spanAttributes ? Object.entries(options.spanAttributes) : []),
445 [ATTR_DB_SYSTEM_NAME, "sqlite"]
446 ],
447 transformRows
448 })) as SqliteClient,
449 {
450 [TypeId]: TypeId as TypeId,
451 config: options,
452 export: Effect.flatMap(acquirer, (connection) => connection.export),
453 import(data: Uint8Array) {
454 return Effect.flatMap(acquirer, (connection) => connection.import(data))
455 }
456 }
457 )
458 })
459
460function rowToObject(columns: Array<string>, row: Array<any>) {
461 const obj: Record<string, any> = {}
462 for (let i = 0; i < columns.length; i++) {
463 Rec.assignProperty(obj, columns[i], row[i])
464 }
465 return obj
466}
467type WorkerResult = [columns: Array<Array<string>>, rows: Array<any>]
468
469const extractObject = (rows: WorkerResult) => rows[1].map((row, index) => rowToObject(rows[0][index], row))
470const extractRows = (rows: WorkerResult) => rows[1]
471
472/**
473 * Fiber reference that stores transferables to include with worker-backed SQLite WASM query messages.
474 *
475 * @category services
476 * @since 4.0.0
477 */
478export const Transferables = Context.Reference<ReadonlyArray<Transferable>>(
479 "@effect/sql-sqlite-wasm/currentTransferables",
480 { defaultValue: () => [] }
481)
482
483/**
484 * Runs an effect with the supplied transferables attached to worker-backed SQLite WASM query messages.
485 *
486 * @category transferables
487 * @since 4.0.0
488 */
489export const withTransferables =
490 (transferables: ReadonlyArray<Transferable>) => <A, E, R>(effect: Effect.Effect<A, E, R>): Effect.Effect<A, E, R> =>
491 Effect.provideService(effect, Transferables, transferables)
492
493/**
494 * Builds a layer from an Effect `Config` value, providing both the in-memory SQLite WASM `SqliteClient` service and the generic `SqlClient` service.
495 *
496 * @category layers
497 * @since 4.0.0
498 */
499export const layerMemoryConfig = (
500 config: Config.Wrap<SqliteClientMemoryConfig>
501): Layer.Layer<SqliteClient | Client.SqlClient, Config.ConfigError | SqlError> =>
502 Layer.effectContext(
503 Config.unwrap(config).pipe(
504 Effect.flatMap(makeMemory),
505 Effect.map((client) =>
506 Context.make(SqliteClient, client).pipe(
507 Context.add(Client.SqlClient, client)
508 )
509 )
510 )
511 ).pipe(Layer.provide(Reactivity.layer))
512
513/**
514 * Builds a layer from an in-memory SQLite WASM client configuration, providing both `SqliteClient` and the generic `SqlClient` service.
515 *
516 * @category layers
517 * @since 4.0.0
518 */
519export const layerMemory = (
520 config: SqliteClientMemoryConfig
521): Layer.Layer<SqliteClient | Client.SqlClient, SqlError> =>
522 Layer.effectContext(
523 Effect.map(makeMemory(config), (client) =>
524 Context.make(SqliteClient, client).pipe(
525 Context.add(Client.SqlClient, client)
526 ))
527 ).pipe(Layer.provide(Reactivity.layer))
528
529/**
530 * Builds a layer from a worker-backed SQLite WASM client configuration, providing both `SqliteClient` and the generic `SqlClient` service.
531 *
532 * @category layers
533 * @since 4.0.0
534 */
535export const layer = (
536 config: SqliteClientConfig
537): Layer.Layer<SqliteClient | Client.SqlClient, SqlError> =>
538 Layer.effectContext(
539 Effect.map(make(config), (client) =>
540 Context.make(SqliteClient, client).pipe(
541 Context.add(Client.SqlClient, client)
542 ))
543 ).pipe(Layer.provide(Reactivity.layer))
544
545/**
546 * Builds a layer from an Effect `Config` value, providing both the worker-backed SQLite WASM `SqliteClient` service and the generic `SqlClient` service.
547 *
548 * @category layers
549 * @since 4.0.0
550 */
551export const layerConfig = (
552 config: Config.Wrap<SqliteClientConfig>
553): Layer.Layer<SqliteClient | Client.SqlClient, Config.ConfigError | SqlError> =>
554 Layer.effectContext(
555 Config.unwrap(config).pipe(
556 Effect.flatMap(make),
557 Effect.map((client) =>
558 Context.make(SqliteClient, client).pipe(
559 Context.add(Client.SqlClient, client)
560 )
561 )
562 )
563 ).pipe(Layer.provide(Reactivity.layer))