1/**
2 * Port of `@effect/sql-sqlite-node` that uses the native `node:sqlite`
3 * bindings instead of `better-sqlite3`.
4 *
5 * @module SqliteClient
6 */
7import * as NodeSqlite from "node:sqlite";
8import * as NodeUtil from "node:util";
9import * as NodeWorkerThreads from "node:worker_threads";
10
11import * as Cache from "effect/Cache";
12import * as Clock from "effect/Clock";
13import * as Duration from "effect/Duration";
14import * as Effect from "effect/Effect";
15import * as Exit from "effect/Exit";
16import * as Fiber from "effect/Fiber";
17import { identity } from "effect/Function";
18import * as Layer from "effect/Layer";
19import * as Option from "effect/Option";
20import * as Predicate from "effect/Predicate";
21import * as References from "effect/References";
22import * as Schema from "effect/Schema";
23import * as Scope from "effect/Scope";
24import * as Semaphore from "effect/Semaphore";
25import * as Context from "effect/Context";
26import * as Stream from "effect/Stream";
27import * as Tracer from "effect/Tracer";
28import * as Reactivity from "effect/reactivity/Reactivity";
29import * as Client from "effect/sql/SqlClient";
30import type { Connection } from "effect/sql/SqlConnection";
31import { SqlError, classifySqliteError } from "effect/sql/SqlError";
32import * as Statement from "effect/sql/Statement";
33
34const ATTR_DB_SYSTEM_NAME = "db.system.name";
35
36export interface SqliteClientConfig {
37 readonly filename: string;
38 readonly readonly?: boolean | undefined;
39 readonly allowExtension?: boolean | undefined;
40 readonly prepareCacheSize?: number | undefined;
41 readonly prepareCacheTTL?: Duration.Input | undefined;
42 readonly spanAttributes?: Record<string, unknown> | undefined;
43 readonly transformResultNames?: ((str: string) => string) | undefined;
44 readonly transformQueryNames?: ((str: string) => string) | undefined;
45 /**
46 * Run transactions marked with `readOnly` on a second connection in a worker
47 * thread, so long reads block neither the event loop nor writes. Ignored for
48 * in-memory databases, which a second connection cannot share.
49 */
50 readonly readerWorker?: boolean | undefined;
51 /** How long a reader worker may take to open the database before reads use the writer. Default 10 seconds. */
52 readonly readerStartupTimeout?: Duration.Input | undefined;
53 /**
54 * Run WAL checkpoints on a read-write connection in a worker thread, so
55 * copying pages into the database and syncing it never block the event loop.
56 * Ignored for in-memory databases.
57 */
58 readonly checkpointWorker?: boolean | undefined;
59}
60
61/**
62 * Set by `readOnly`. A top-level transaction started while it is true runs on
63 * the reader worker when the client has one, and on the writer otherwise.
64 */
65const ReadOnly = Context.Reference<boolean>("@t3tools/shared/nodeSqliteClient/ReadOnly", {
66 defaultValue: () => false,
67});
68
69/**
70 * Marks the SQL in `effect` as read-only. Wrap the whole transaction, as in
71 * `sql.withTransaction(reads).pipe(readOnly)`. Inside an open write
72 * transaction the reads keep using the writer, and a write attempted on the
73 * reader fails with "attempt to write a readonly database".
74 */
75export const readOnly = <A, E, R>(effect: Effect.Effect<A, E, R>): Effect.Effect<A, E, R> =>
76 Effect.provideService(effect, ReadOnly, true);
77
78/**
79 * Which connection ran the SQL, and how long it waited for that connection's
80 * lock. The wait is time queued behind other SQL, not time the event loop was
81 * blocked.
82 */
83interface Acquired {
84 readonly connection: "writer" | "reader";
85 readonly lockWaitMs: number;
86}
87
88const annotateAcquired = (fiber: Fiber.Fiber<unknown, unknown>, acquired: Acquired | undefined) => {
89 const span = Context.getOrUndefined(fiber.context, Tracer.ParentSpan);
90 if (span?._tag !== "Span" || acquired === undefined) return;
91 span.attribute("db.connection", acquired.connection);
92 span.attribute("db.lock_wait_ms", acquired.lockWaitMs);
93};
94
95/**
96 * Provided by the client's `withTransaction`. Effect acquires a transaction's
97 * connection before it makes the `sql.transaction` span current, so the acquire
98 * leaves its result here and the transaction body puts it on the span.
99 */
100const TransactionAcquire = Context.Reference<{ acquired?: Acquired } | undefined>(
101 "@t3tools/shared/nodeSqliteClient/TransactionAcquire",
102 { defaultValue: () => undefined },
103);
104
105export class UnsupportedNodeSqliteVersionError extends Schema.TaggedError<UnsupportedNodeSqliteVersionError>()(
106 "UnsupportedNodeSqliteVersionError",
107 {
108 nodeVersion: Schema.String,
109 requirement: Schema.String,
110 },
111) {
112 override get message(): string {
113 return `Node.js ${this.nodeVersion} is missing required node:sqlite APIs. Upgrade to ${this.requirement}.`;
114 }
115}
116
117export class UnsupportedNodeSqliteOperationError extends Schema.TaggedError<UnsupportedNodeSqliteOperationError>()(
118 "UnsupportedNodeSqliteOperationError",
119 {},
120) {
121 override get message(): string {
122 return "Node SQLite does not support executeStream.";
123 }
124}
125
126/**
127 * Verify that the current Node.js version includes the `node:sqlite` APIs
128 * used by `NodeSqliteClient` — specifically `StatementSync.columns()` (added
129 * in Node 22.16.0 / 23.11.0).
130 *
131 * @see https://github.com/nodejs/node/pull/57490
132 */
133const checkNodeSqliteCompat = () => {
134 const parts = process.versions.node.split(".").map(Number);
135 const major = parts[0] ?? 0;
136 const minor = parts[1] ?? 0;
137 const supported = (major === 22 && minor >= 16) || (major === 23 && minor >= 11) || major >= 24;
138
139 if (!supported) {
140 return Effect.die(
141 new UnsupportedNodeSqliteVersionError({
142 nodeVersion: process.versions.node,
143 requirement: "Node.js >=22.16, >=23.11, or >=24",
144 }),
145 );
146 }
147 return Effect.void;
148};
149
150/**
151 * `node:sqlite` reports the SQLite result code as `errcode`, while
152 * `classifySqliteError` reads `errno`. Copy it across so busy, locked and
153 * constraint failures get their own reasons instead of `UnknownError`.
154 */
155const classifyError = (cause: unknown, message: string, operation: string) => {
156 if (
157 Predicate.hasProperty(cause, "errcode") &&
158 typeof cause.errcode === "number" &&
159 !Predicate.hasProperty(cause, "errno")
160 ) {
161 Object.assign(cause, { errno: cause.errcode });
162 }
163 return classifySqliteError(cause, { message, operation });
164};
165
166/**
167 * SQLite keeps its copy of a statement's text and blob parameters until the
168 * statement is bound again or finalized, and `node:sqlite` cannot clear them.
169 * A cached statement would hold its last payload for as long as it stays
170 * cached, so executions binding more than this many string code units and
171 * blob bytes, or a named parameter behind a getter, get a throwaway statement
172 * instead.
173 */
174const LARGE_BINDING_SIZE = 64 * 1024;
175
176// `node:sqlite` binds both buffer views and raw buffers as blobs.
177const isBlob = (value: unknown): value is ArrayBufferView | ArrayBufferLike =>
178 ArrayBuffer.isView(value) || NodeUtil.types.isAnyArrayBuffer(value);
179
180const bindingSize = (value: unknown): number => {
181 if (typeof value === "string") {
182 return value.length;
183 }
184 return isBlob(value) ? value.byteLength : 0;
185};
186
187const hasLargeBinding = (params: ReadonlyArray<unknown>): boolean => {
188 try {
189 let size = 0;
190 for (const param of params) {
191 if (
192 (typeof param === "object" || typeof param === "function") &&
193 param !== null &&
194 !isBlob(param)
195 ) {
196 // `node:sqlite` takes named parameters as one object, callable or not.
197 // Read data properties only: calling a getter here would run it twice.
198 for (const property of Object.values(Object.getOwnPropertyDescriptors(param))) {
199 if (!property.enumerable) continue;
200 if (!("value" in property)) return true;
201 size += bindingSize(property.value);
202 }
203 } else {
204 size += bindingSize(param);
205 }
206 }
207 return size > LARGE_BINDING_SIZE;
208 } catch {
209 // A parameter that cannot be measured is left for the statement to report.
210 return true;
211 }
212};
213
214// Evaluated source instead of a worker file, so the reader also starts from
215// bundles and single-executable builds that have no module file on disk.
216const readerWorkerSource = `
217const { parentPort, workerData } = require("node:worker_threads");
218const { DatabaseSync } = require("node:sqlite");
219let db;
220try {
221 // A read-only open cannot be undone by SQL, unlike PRAGMA query_only.
222 const next = new DatabaseSync(workerData.filename, {
223 readOnly: true,
224 allowExtension: workerData.allowExtension,
225 });
226 next.exec("PRAGMA busy_timeout = 5000;");
227 // Without WAL a reader's shared lock blocks the writer's commits, so refuse.
228 const mode = String(next.prepare("PRAGMA journal_mode").get().journal_mode).toLowerCase();
229 if (mode !== "wal") throw new Error("database uses journal_mode " + mode + ", not wal");
230 db = next;
231 parentPort.postMessage({ type: "ready" });
232} catch (error) {
233 parentPort.postMessage({ type: "unavailable", message: String(error && error.message ? error.message : error) });
234}
235const statements = new Map();
236const prepare = (sql, cached) => {
237 if (!cached) return db.prepare(sql);
238 let statement = statements.get(sql);
239 if (statement === undefined) {
240 statement = db.prepare(sql);
241 if (statements.size >= workerData.prepareCacheSize) statements.delete(statements.keys().next().value);
242 } else {
243 statements.delete(sql);
244 }
245 statements.set(sql, statement);
246 return statement;
247};
248parentPort.on("message", ({ id, sql, params, cached, values, raw, safeIntegers }) => {
249 let operation = "open";
250 try {
251 if (db === undefined) throw new Error("database reader is unavailable");
252 operation = "prepare";
253 const statement = prepare(sql, cached);
254 operation = "execute";
255 statement.setReadBigInts(safeIntegers);
256 if (statement.columns().length > 0) {
257 statement.setReturnArrays(values);
258 parentPort.postMessage({ id, rows: statement.all(...params) });
259 } else {
260 const result = statement.run(...params);
261 parentPort.postMessage({ id, rows: raw ? result : [] });
262 }
263 } catch (error) {
264 parentPort.postMessage({
265 id,
266 error: {
267 operation,
268 message: String(error && error.message ? error.message : error),
269 code: error && error.code,
270 errcode: error && error.errcode,
271 errstr: error && error.errstr,
272 },
273 });
274 }
275});
276`;
277
278interface ReaderRequest {
279 readonly sql: string;
280 readonly params: ReadonlyArray<unknown>;
281 readonly cached: boolean;
282 readonly values: boolean;
283 readonly raw: boolean;
284}
285
286type ReaderStartup =
287 | { readonly type: "ready" }
288 | { readonly type: "unavailable"; readonly message: string };
289
290type ReaderReply =
291 | { readonly id: number; readonly rows: ReadonlyArray<any> }
292 | {
293 readonly id: number;
294 readonly error: {
295 readonly operation: "open" | "prepare" | "execute";
296 readonly message: string;
297 readonly code?: unknown;
298 readonly errcode?: unknown;
299 readonly errstr?: unknown;
300 };
301 };
302
303const readerFailureMessages = {
304 open: "Failed to open database",
305 prepare: "Failed to prepare statement",
306 execute: "Failed to execute statement",
307} as const;
308
309const failed = (cause: unknown, message: string, operation: string) =>
310 new SqlError({ reason: classifyError(cause, message, operation) });
311
312// COMMIT, ROLLBACK and ROLLBACK TO SAVEPOINT. On a reader they change nothing,
313// so they succeed once the worker is gone instead of failing the cleanup.
314const endsReadTransaction = (sql: string) => /^\s*ROLLBACK\b|^\s*COMMIT\b/i.test(sql);
315
316/**
317 * Read-only connections served by one worker thread at a time. Each worker
318 * gets its own `Connection`, so a transaction stays on the worker it began on:
319 * once that worker exits, its statements fail rather than silently continuing
320 * outside the transaction on a replacement. The next acquire starts a new one.
321 * A worker that cannot open the database read-only, or finds it outside WAL
322 * mode, turns the reader off for the life of the client: reads use the writer.
323 */
324const makeReader = Effect.fnUntraced(function* (options: SqliteClientConfig) {
325 let current:
326 | {
327 readonly worker: NodeWorkerThreads.Worker;
328 readonly connection: Connection;
329 readonly started: Promise<ReaderStartup>;
330 }
331 | undefined;
332 let unavailable = false;
333 let nextRequestId = 0;
334
335 const start = () => {
336 const worker = new NodeWorkerThreads.Worker(readerWorkerSource, {
337 eval: true,
338 workerData: {
339 filename: options.filename,
340 allowExtension: options.allowExtension ?? false,
341 prepareCacheSize: options.prepareCacheSize ?? 200,
342 },
343 });
344 let announce: (startup: ReaderStartup) => void = () => {};
345 const started = new Promise<ReaderStartup>((resolve) => {
346 announce = resolve;
347 });
348 let stoppedMessage: string | undefined;
349 const pending = new Map<
350 number,
351 { readonly endsTransaction: boolean; readonly resume: (reply: ReaderReply) => void }
352 >();
353 const stop = (message: string) => {
354 stoppedMessage = message;
355 announce({ type: "unavailable", message });
356 if (current?.worker === worker) current = undefined;
357 for (const [id, request] of pending) {
358 request.resume(
359 request.endsTransaction
360 ? { id, rows: [] }
361 : { id, error: { operation: "execute", message } },
362 );
363 }
364 pending.clear();
365 };
366 worker.on("message", (reply: ReaderReply | ReaderStartup) => {
367 if ("type" in reply) {
368 announce(reply);
369 // An idle reader must not keep the process alive.
370 if (pending.size === 0) worker.unref();
371 return;
372 }
373 const request = pending.get(reply.id);
374 if (request === undefined) return;
375 pending.delete(reply.id);
376 if (pending.size === 0) worker.unref();
377 request.resume(reply);
378 });
379 worker.on("error", (cause) => stop(`Database reader failed: ${cause.message}`));
380 worker.on("exit", (code) => stop(`Database reader exited with code ${code}`));
381
382 const request = (input: ReaderRequest) =>
383 Effect.withFiber<ReadonlyArray<any>, SqlError>((fiber) => {
384 const endsTransaction = endsReadTransaction(input.sql);
385 if (stoppedMessage !== undefined) {
386 return endsTransaction
387 ? Effect.succeed([])
388 : Effect.fail(
389 failed(new Error(stoppedMessage), readerFailureMessages.execute, "execute"),
390 );
391 }
392 const safeIntegers = Boolean(Context.get(fiber.context, Client.SafeIntegers));
393 return Effect.callback<ReadonlyArray<any>, SqlError>((resume) => {
394 const id = nextRequestId++;
395 pending.set(id, {
396 endsTransaction,
397 resume: (reply) => {
398 if ("rows" in reply) {
399 resume(Effect.succeed(reply.rows));
400 return;
401 }
402 const { operation, message, ...codes } = reply.error;
403 resume(
404 Effect.fail(
405 failed(
406 Object.assign(new Error(message), codes),
407 readerFailureMessages[operation],
408 operation,
409 ),
410 ),
411 );
412 },
413 });
414 if (pending.size === 1) worker.ref();
415 // The worker's statement cache would pin large parameters the same way.
416 const cached = input.cached && !hasLargeBinding(input.params);
417 // oxlint-disable-next-line unicorn/require-post-message-target-origin -- Node workers do not accept a target origin.
418 worker.postMessage({ id, safeIntegers, ...input, cached });
419 });
420 });
421
422 const connection = identity<Connection>({
423 execute(sql, params, rowTransform) {
424 const effect = request({ sql, params, cached: true, values: false, raw: false });
425 return rowTransform ? Effect.map(effect, rowTransform) : effect;
426 },
427 executeRaw(sql, params) {
428 return request({ sql, params, cached: true, values: false, raw: true });
429 },
430 executeValues(sql, params) {
431 return request({ sql, params, cached: true, values: true, raw: false });
432 },
433 executeValuesUnprepared(sql, params) {
434 return request({ sql, params: params ?? [], cached: false, values: true, raw: false });
435 },
436 executeUnprepared(sql, params, rowTransform) {
437 // A read-only connection rejects BEGIN IMMEDIATE and BEGIN EXCLUSIVE,
438 // and a deferred BEGIN is all a read snapshot needs.
439 const statement = /^\s*BEGIN\b/i.test(sql) ? "BEGIN" : sql;
440 const effect = request({
441 sql: statement,
442 params: params ?? [],
443 cached: false,
444 values: false,
445 raw: false,
446 });
447 return rowTransform ? Effect.map(effect, rowTransform) : effect;
448 },
449 executeStream(_sql, _params) {
450 return Stream.die(new UnsupportedNodeSqliteOperationError());
451 },
452 });
453 current = { worker, connection, started };
454 return current;
455 };
456
457 yield* Effect.addFinalizer(() =>
458 Effect.promise(async () => {
459 await current?.worker.terminate();
460 }),
461 );
462
463 const startupTimeout = Duration.fromInputUnsafe(
464 options.readerStartupTimeout ?? Duration.seconds(10),
465 );
466
467 // Runs once: concurrent first acquires share one startup and may all see it fail.
468 const turnOff = (reader: typeof current, reason: string) =>
469 Effect.suspend(() => {
470 if (reader !== undefined) {
471 if (current === reader) current = undefined;
472 void reader.worker.terminate();
473 }
474 if (unavailable) return Effect.succeed(undefined);
475 unavailable = true;
476 return Effect.logWarning("SQLite reader worker unavailable; reads use the main connection", {
477 reason,
478 }).pipe(Effect.as(undefined));
479 });
480
481 /** The live worker's connection once it is ready, or undefined to use the writer. */
482 return Effect.suspend((): Effect.Effect<Connection | undefined> => {
483 if (unavailable) return Effect.succeed(undefined);
484 let reader: NonNullable<typeof current>;
485 try {
486 reader = current ?? start();
487 } catch (cause) {
488 return turnOff(undefined, String(cause));
489 }
490 // SqlClient acquires transaction connections uninterruptibly. Waiting for the
491 // worker happens before any lock is taken, so let a cancelled request leave.
492 return Effect.flatMap(
493 Effect.interruptible(
494 Effect.timeoutOption(
495 Effect.promise(() => reader.started),
496 startupTimeout,
497 ),
498 ),
499 (startup): Effect.Effect<Connection | undefined> =>
500 // Another acquire may have timed out and retired this worker while this one
501 // waited; a late "ready" must not hand out a worker that is being terminated.
502 unavailable || current !== reader
503 ? Effect.succeed(undefined)
504 : Option.isSome(startup) && startup.value.type === "ready"
505 ? Effect.succeed(reader.connection)
506 : turnOff(
507 reader,
508 Option.isSome(startup) && startup.value.type === "unavailable"
509 ? startup.value.message
510 : `worker did not open the database within ${Duration.toMillis(startupTimeout)} ms`,
511 ),
512 );
513 });
514});
515
516// SQLite's default wal_autocheckpoint, in WAL frames (pages).
517const CHECKPOINT_FRAMES = 1000;
518// While the worker runs, the writer still checkpoints a WAL this long. Writes
519// with no pause leave the worker behind, and the WAL can only restart once
520// every frame is checkpointed, so this keeps the file from growing forever.
521const WRITER_BACKSTOP_FRAMES = 10 * CHECKPOINT_FRAMES;
522// How long writes must pause before the worker checkpoints the rest of a long WAL.
523const CHECKPOINT_IDLE_MS = 100;
524
525// Each message asks the worker to checkpoint if CHECKPOINT_FRAMES frames are
526// waiting, and it replies with the same message when done. Once writes pause, it also checkpoints the
527// frames that arrived during its last checkpoint, so the writer can restart the
528// WAL. NOOP only counts frames. SQLite before 3.51 reads NOOP as PASSIVE and
529// would checkpoint on every request, so the worker refuses to start there and
530// the writer keeps checkpointing.
531const checkpointWorkerSource = `
532const { parentPort, workerData } = require("node:worker_threads");
533const { DatabaseSync } = require("node:sqlite");
534const db = new DatabaseSync(workerData.filename);
535const version = db.prepare("SELECT sqlite_version() AS version").get().version;
536const [major, minor] = version.split(".").map(Number);
537if (major < 3 || (major === 3 && minor < 51)) throw new Error("SQLite " + version + " has no NOOP checkpoint");
538// macOS fsync() does not flush the drive cache; F_FULLFSYNC does. Other platforms ignore this.
539db.exec("PRAGMA checkpoint_fullfsync = ON;");
540const frames = db.prepare("PRAGMA wal_checkpoint(NOOP)");
541const checkpoint = db.prepare("PRAGMA wal_checkpoint(PASSIVE)");
542const finish = () => {
543 const { log, checkpointed } = frames.get();
544 if (log >= workerData.frames && checkpointed < log) checkpoint.get();
545};
546let idle;
547parentPort.on("message", (request) => {
548 const { log, checkpointed } = frames.get();
549 if (log - checkpointed >= workerData.frames) checkpoint.get();
550 clearTimeout(idle);
551 idle = setTimeout(finish, workerData.idleMs);
552 parentPort.postMessage(request);
553});
554`;
555
556/**
557 * Moves the writer's WAL checkpoints to a worker. Call the returned function
558 * after each statement on the writer. While the worker runs, the writer only
559 * checkpoints at WRITER_BACKSTOP_FRAMES. If the worker cannot start, fails or
560 * exits, the writer goes back to SQLite's default and one warning says why.
561 */
562const startCheckpointWorker = Effect.fnUntraced(function* (
563 filename: string,
564 writer: NodeSqlite.DatabaseSync,
565) {
566 const warn = (reason: string) =>
567 Effect.logWarning(
568 "SQLite checkpoint worker unavailable; checkpoints run on the main connection",
569 { reason },
570 );
571 let worker: NodeWorkerThreads.Worker;
572 try {
573 worker = new NodeWorkerThreads.Worker(checkpointWorkerSource, {
574 eval: true,
575 workerData: { filename, frames: CHECKPOINT_FRAMES, idleMs: CHECKPOINT_IDLE_MS },
576 });
577 } catch (cause) {
578 // No worker threads here: the writer keeps checkpointing.
579 yield* warn(String(cause));
580 return undefined;
581 }
582 writer.exec(`PRAGMA wal_autocheckpoint = ${WRITER_BACKSTOP_FRAMES};`);
583 // Worker events arrive outside any fiber; this keeps the client's loggers.
584 const runFork = Effect.runForkWith(yield* Effect.context<never>());
585 let stopped = false;
586 let checking = false;
587 let checkAgain = false;
588 const handBack = (reason: string) => {
589 if (stopped) return;
590 stopped = true;
591 try {
592 writer.exec(`PRAGMA wal_autocheckpoint = ${CHECKPOINT_FRAMES};`);
593 } catch {
594 // The writer is already closed.
595 }
596 runFork(warn(reason));
597 };
598 // A worker that throws, such as on SQLite before 3.51, emits "error" before "exit".
599 worker.on("error", (cause) => handBack(cause.message));
600 worker.on("exit", (code) => handBack(`exited with code ${code}`));
601 yield* Effect.addFinalizer(() =>
602 Effect.promise(() => {
603 // A normal shutdown: the writer closes next, so there is nothing to hand back or report.
604 stopped = true;
605 return worker.terminate();
606 }),
607 );
608
609 // At most one request is in flight. A statement that runs during it asks for
610 // one more check, so the frames of a last commit before idle still get one.
611 const check = () => {
612 if (stopped) return;
613 if (checking) {
614 checkAgain = true;
615 return;
616 }
617 checking = true;
618 // oxlint-disable-next-line unicorn/require-post-message-target-origin -- Node workers do not accept a target origin.
619 worker.postMessage(null);
620 };
621 worker.on("message", () => {
622 checking = false;
623 if (checkAgain) {
624 checkAgain = false;
625 check();
626 }
627 });
628 // After the listeners: adding a "message" listener refs the worker again.
629 // A checkpoint in progress must not keep the process alive.
630 worker.unref();
631 return check;
632});
633
634const make = Effect.fn("makeWithDatabase")(function* (
635 options: SqliteClientConfig,
636): Effect.fn.Return<Client.SqlClient, SqlError, Scope.Scope | Reactivity.Reactivity> {
637 yield* checkNodeSqliteCompat();
638
639 const compiler = Statement.makeCompilerSqlite(options.transformQueryNames);
640 const isFileDatabase = options.filename !== ":memory:" && options.filename !== "";
641 const transformRows = options.transformResultNames
642 ? Statement.defaultTransforms(options.transformResultNames).array
643 : undefined;
644
645 const makeConnection = Effect.gen(function* () {
646 const scope = yield* Effect.scope;
647 const db = yield* Effect.try({
648 try: () =>
649 new NodeSqlite.DatabaseSync(options.filename, {
650 readOnly: options.readonly ?? false,
651 allowExtension: options.allowExtension ?? false,
652 }),
653 catch: (cause) =>
654 new SqlError({
655 reason: classifyError(cause, "Failed to open database", "open"),
656 }),
657 });
658 yield* Scope.addFinalizer(
659 scope,
660 Effect.try({
661 try: () => db.close(),
662 catch: (cause) =>
663 new SqlError({
664 reason: classifyError(cause, "Failed to close database", "close"),
665 }),
666 }).pipe(Effect.orDie),
667 );
668 // Started after the close finalizer, so the worker stops before the writer closes.
669 const checkpoint =
670 options.checkpointWorker && isFileDatabase
671 ? yield* startCheckpointWorker(options.filename, db)
672 : undefined;
673
674 const statementReaderCache = new WeakMap<NodeSqlite.StatementSync, boolean>();
675 const hasRows = (statement: NodeSqlite.StatementSync): boolean => {
676 const cached = statementReaderCache.get(statement);
677 if (cached !== undefined) {
678 return cached;
679 }
680 const value = statement.columns().length > 0;
681 statementReaderCache.set(statement, value);
682 return value;
683 };
684
685 const prepare = (sql: string) =>
686 Effect.try({
687 try: () => db.prepare(sql),
688 catch: (cause) =>
689 new SqlError({
690 reason: classifyError(cause, "Failed to prepare statement", "prepare"),
691 }),
692 });
693
694 const prepareCache = yield* Cache.makeWith(prepare, {
695 capacity: options.prepareCacheSize ?? 200,
696 // A transient prepare failure must not outlive the lock or missing schema.
697 timeToLive: (exit) =>
698 Exit.isSuccess(exit) ? (options.prepareCacheTTL ?? Duration.minutes(10)) : Duration.zero,
699 });
700
701 const statementFor = (sql: string, params: ReadonlyArray<unknown>) =>
702 hasLargeBinding(params) ? prepare(sql) : Cache.get(prepareCache, sql);
703
704 const runStatement = (
705 statement: NodeSqlite.StatementSync,
706 params: ReadonlyArray<unknown>,
707 raw: boolean,
708 ) =>
709 Effect.withFiber<ReadonlyArray<any>, SqlError>((fiber) => {
710 try {
711 statement.setReadBigInts(Boolean(Context.get(fiber.context, Client.SafeIntegers)));
712 if (hasRows(statement)) {
713 const rows = statement.all(...(params as any));
714 checkpoint?.();
715 return Effect.succeed(rows);
716 }
717 const result = statement.run(...(params as any));
718 checkpoint?.();
719 return Effect.succeed(raw ? (result as unknown as ReadonlyArray<any>) : []);
720 } catch (cause) {
721 return Effect.fail(
722 new SqlError({
723 reason: classifyError(cause, "Failed to execute statement", "execute"),
724 }),
725 );
726 }
727 });
728
729 const run = (sql: string, params: ReadonlyArray<unknown>, raw = false) =>
730 Effect.flatMap(statementFor(sql, params), (s) => runStatement(s, params, raw));
731
732 const runStatementValues = (
733 statement: NodeSqlite.StatementSync,
734 params: ReadonlyArray<unknown>,
735 ) =>
736 Effect.acquireUseRelease(
737 Effect.succeed(statement),
738 (statement) =>
739 Effect.try({
740 try: () => {
741 statement.setReadBigInts(
742 Boolean(Context.get(Fiber.getCurrent()!.context, Client.SafeIntegers)),
743 );
744 if (hasRows(statement)) {
745 statement.setReturnArrays(true);
746 // Safe to cast to array after we've setReturnArrays(true)
747 const rows = statement.all(...(params as any)) as unknown as ReadonlyArray<
748 ReadonlyArray<unknown>
749 >;
750 checkpoint?.();
751 return rows;
752 }
753 statement.run(...(params as any));
754 checkpoint?.();
755 return [];
756 },
757 catch: (cause) =>
758 new SqlError({
759 reason: classifyError(cause, "Failed to execute statement", "execute"),
760 }),
761 }),
762 (statement) =>
763 Effect.try({
764 try: () => {
765 if (hasRows(statement)) {
766 statement.setReturnArrays(false);
767 }
768 },
769 catch: (cause) =>
770 new SqlError({
771 reason: classifyError(
772 cause,
773 "Failed to reset statement result mode",
774 "resetResultMode",
775 ),
776 }),
777 }).pipe(Effect.orDie),
778 );
779
780 const runValues = (sql: string, params: ReadonlyArray<unknown>) =>
781 Effect.flatMap(statementFor(sql, params), (statement) =>
782 runStatementValues(statement, params),
783 );
784
785 return identity<Connection>({
786 execute(sql, params, rowTransform) {
787 return rowTransform ? Effect.map(run(sql, params), rowTransform) : run(sql, params);
788 },
789 executeRaw(sql, params) {
790 return run(sql, params, true);
791 },
792 executeValues(sql, params) {
793 return runValues(sql, params);
794 },
795 executeValuesUnprepared(sql, params) {
796 return Effect.flatMap(prepare(sql), (statement) =>
797 runStatementValues(statement, params ?? []),
798 );
799 },
800 executeUnprepared(sql, params, rowTransform) {
801 const effect = prepare(sql).pipe(
802 Effect.flatMap((statement) => runStatement(statement, params ?? [], false)),
803 );
804 return rowTransform ? Effect.map(effect, rowTransform) : effect;
805 },
806 executeStream(_sql, _params) {
807 return Stream.die(new UnsupportedNodeSqliteOperationError());
808 },
809 });
810 });
811
812 const semaphore = yield* Semaphore.make(1);
813 const connection = yield* makeConnection;
814 const acquireReader =
815 options.readerWorker && isFileDatabase ? yield* makeReader(options) : undefined;
816 const readerSemaphore = yield* Semaphore.make(1);
817
818 const writer = { name: "writer" as const, lock: semaphore, connection };
819 // Picks the reader only for `readOnly` work, and the writer whenever the
820 // reader is turned off.
821 const route = Effect.withFiber((fiber) =>
822 acquireReader !== undefined && Context.get(fiber.context, ReadOnly)
823 ? Effect.map(acquireReader, (readerConnection) =>
824 readerConnection === undefined
825 ? writer
826 : { name: "reader" as const, lock: readerSemaphore, connection: readerConnection },
827 )
828 : Effect.succeed(writer),
829 );
830
831 const acquired = (
832 target: { readonly name: Acquired["connection"] },
833 clock: Clock.Clock,
834 start: bigint,
835 ): Acquired => ({
836 connection: target.name,
837 lockWaitMs: Number((clock.monotonicTimeNanosUnsafe() - start) / 1000n) / 1000,
838 });
839
840 const acquirer = Effect.flatMap(route, (target) =>
841 Effect.withFiber((fiber) => {
842 // The statement's span is current here only when it propagates, which
843 // Statement skips while tracing is off. Otherwise the current span is
844 // the caller's, so leave it alone.
845 if (
846 !fiber.getRef(References.TracerEnabled) ||
847 !fiber.getRef(Statement.SpanPropagationEnabled)
848 ) {
849 return target.lock.withPermits(1)(Effect.succeed(target.connection));
850 }
851 const clock = fiber.getRef(Clock.Clock);
852 const start = clock.monotonicTimeNanosUnsafe();
853 return target.lock.withPermits(1)(
854 Effect.sync(() => {
855 annotateAcquired(fiber, acquired(target, clock, start));
856 return target.connection;
857 }),
858 );
859 }),
860 );
861 const transactionAcquirer = Effect.flatMap(route, (target) =>
862 Effect.uninterruptibleMask((restore) => {
863 const fiber = Fiber.getCurrent()!;
864 const scope = Context.getUnsafe(fiber.context, Scope.Scope);
865 const clock = fiber.getRef(Clock.Clock);
866 const start = clock.monotonicTimeNanosUnsafe();
867 return Effect.as(
868 Effect.tap(restore(target.lock.take(1)), () => {
869 const pending = fiber.getRef(TransactionAcquire);
870 if (pending !== undefined) pending.acquired = acquired(target, clock, start);
871 return Scope.addFinalizer(scope, target.lock.release(1));
872 }),
873 target.connection,
874 );
875 }),
876 );
877
878 const client = yield* Client.make({
879 acquirer,
880 compiler,
881 transactionAcquirer,
882 // A deferred BEGIN only takes the write lock at the first write. If another
883 // process commits after this transaction's first read, that write fails at
884 // once with SQLITE_BUSY_SNAPSHOT, which busy_timeout cannot wait out. Taking
885 // the lock up front makes it wait instead, at the cost of serializing
886 // read-only transactions behind other processes' writers. Read-only
887 // connections cannot write, so they keep the deferred BEGIN.
888 beginTransaction: options.readonly === true ? "BEGIN" : "BEGIN IMMEDIATE",
889 spanAttributes: [
890 ...(options.spanAttributes ? Object.entries(options.spanAttributes) : []),
891 [ATTR_DB_SYSTEM_NAME, "sqlite"],
892 ],
893 transformRows,
894 });
895 const withTransaction = client.withTransaction;
896 return Object.assign(client, {
897 // Each run gets its own holder, so a reused transaction Effect never
898 // carries an earlier run's attributes.
899 withTransaction: <A, E, R>(effect: Effect.Effect<A, E, R>) =>
900 Effect.suspend(() => {
901 const pending: { acquired?: Acquired } = {};
902 // The body runs with the `sql.transaction` span current. A nested
903 // transaction acquires nothing, so its span gets no attributes.
904 const body = Effect.withFiber<A, E, R>((fiber) => {
905 annotateAcquired(fiber, pending.acquired);
906 return effect;
907 });
908 return withTransaction(body).pipe(Effect.provideService(TransactionAcquire, pending));
909 }),
910 });
911});
912
913export const layer = (config: SqliteClientConfig): Layer.Layer<Client.SqlClient, SqlError> =>
914 Layer.effect(Client.SqlClient, make(config)).pipe(Layer.provide(Reactivity.layer));