1import * as NodeSqlite from "node:sqlite";
2import * as NodeUtil from "node:util";
3import * as NodeWorkerThreads from "node:worker_threads";
4
5import * as Cache from "effect/Cache";
6import * as Clock from "effect/Clock";
7import * as Duration from "effect/Duration";
8import * as Effect from "effect/Effect";
9import * as Result from "effect/Result";
10import * as Exit from "effect/Exit";
11import * as Fiber from "effect/Fiber";
12import { identity } from "effect/Function";
13import * as Layer from "effect/Layer";
14import * as Option from "effect/Option";
15import * as Predicate from "effect/Predicate";
16import * as References from "effect/References";
17import * as Schema from "effect/Schema";
18import * as Scheduler from "effect/Scheduler";
19import * as Scope from "effect/Scope";
20import * as Semaphore from "effect/Semaphore";
21import * as Context from "effect/Context";
22import * as Stream from "effect/Stream";
23import * as Tracer from "effect/Tracer";
24import * as Reactivity from "effect/reactivity/Reactivity";
25import * as Client from "effect/sql/SqlClient";
26import type { Borrower, Connection } from "effect/sql/SqlConnection";
27import { SqlError, classifySqliteError } from "effect/sql/SqlError";
28import * as Statement from "effect/sql/Statement";
29
30const ATTR_DB_SYSTEM_NAME = "db.system.name";
31
32export interface SqliteClientConfig {
33 readonly filename: string;
34 readonly readonly?: boolean | undefined;
35 readonly spanAttributes?: Record<string, unknown> | undefined;
36 /** Ignored for in-memory databases, which cannot share a second connection. */
37 readonly readerWorker?: boolean | undefined;
38 /** Runs the writer and WAL checkpoints off the event loop. Ignored for in-memory databases. */
39 readonly writerWorker?: boolean | undefined;
40}
41
42// SqlClient masks transaction acquisition, so readOnly carries the caller's interruption policy.
43const ReadOnly = Context.Reference<typeof Effect.interruptible | undefined>(
44 "@t3tools/shared/nodeSqliteClient/ReadOnly",
45 { defaultValue: () => undefined },
46);
47
48/** Wrap the whole transaction; existing write transactions retain the writer, and writes on the reader fail as readonly. */
49export const readOnly = <A, E, R>(effect: Effect.Effect<A, E, R>): Effect.Effect<A, E, R> =>
50 Effect.uninterruptibleMask((restore) =>
51 restore(Effect.provideService(effect, ReadOnly, restore)),
52 );
53
54interface Acquired {
55 readonly connection: "writer" | "reader";
56 readonly lockWaitMs: number;
57}
58
59interface ConnectionOwner {
60 readonly fiberId: number;
61 readonly name: string;
62 readonly traceId?: string;
63 readonly spanId?: string;
64}
65
66const connectionOwner = (fiber: Fiber.Fiber<unknown, unknown>): ConnectionOwner => {
67 const current = Context.getOrUndefined(fiber.context, Tracer.ParentSpan);
68 if (current?._tag !== "Span") return { fiberId: fiber.id, name: "untraced" };
69 let span: Tracer.Span = current;
70 while (span.name === "sqlite" || span.name === "sql.transaction") {
71 const parent: Tracer.AnySpan | undefined = Option.getOrUndefined(span.parent);
72 if (parent?._tag !== "Span") break;
73 span = parent;
74 }
75 return { fiberId: fiber.id, name: span.name, traceId: span.traceId, spanId: span.spanId };
76};
77
78const annotateAcquired = (
79 fiber: Fiber.Fiber<unknown, unknown>,
80 acquired: Acquired | undefined,
81 ownerAtWaitStart?: ConnectionOwner,
82) => {
83 const span = Context.getOrUndefined(fiber.context, Tracer.ParentSpan);
84 if (span?._tag !== "Span" || acquired === undefined) return;
85 span.attribute("db.connection", acquired.connection);
86 span.attribute("db.lock_wait_ms", acquired.lockWaitMs);
87 if (ownerAtWaitStart !== undefined) {
88 span.attribute("db.lock_owner_at_start.fiber_id", ownerAtWaitStart.fiberId);
89 span.attribute("db.lock_owner_at_start.name", ownerAtWaitStart.name);
90 if (ownerAtWaitStart.traceId !== undefined) {
91 span.attribute("db.lock_owner_at_start.trace_id", ownerAtWaitStart.traceId);
92 }
93 if (ownerAtWaitStart.spanId !== undefined) {
94 span.attribute("db.lock_owner_at_start.span_id", ownerAtWaitStart.spanId);
95 }
96 }
97};
98
99const TransactionAcquire = Context.Reference<
100 | {
101 acquired?: Acquired;
102 ownerAtWaitStart: ConnectionOwner | undefined;
103 restore: typeof Effect.interruptible | undefined;
104 }
105 | undefined
106>("@t3tools/shared/nodeSqliteClient/TransactionAcquire", { defaultValue: () => undefined });
107
108export class UnsupportedNodeSqliteVersionError extends Schema.TaggedError<UnsupportedNodeSqliteVersionError>()(
109 "UnsupportedNodeSqliteVersionError",
110 {
111 nodeVersion: Schema.String,
112 requirement: Schema.String,
113 },
114) {
115 override get message(): string {
116 return `Node.js ${this.nodeVersion} is missing required node:sqlite APIs. Upgrade to ${this.requirement}.`;
117 }
118}
119
120export class UnsupportedNodeSqliteOperationError extends Schema.TaggedError<UnsupportedNodeSqliteOperationError>()(
121 "UnsupportedNodeSqliteOperationError",
122 {},
123) {
124 override get message(): string {
125 return "Node SQLite does not support executeStream.";
126 }
127}
128
129const checkNodeSqliteCompat = Effect.fn("checkNodeSqliteCompat")(function () {
130 const parts = process.versions.node.split(".").map(Number);
131 const major = parts[0] ?? 0;
132 const minor = parts[1] ?? 0;
133 const supported = (major === 22 && minor >= 16) || (major === 23 && minor >= 11) || major >= 24;
134
135 if (!supported) {
136 return Effect.die(
137 new UnsupportedNodeSqliteVersionError({
138 nodeVersion: process.versions.node,
139 requirement: "Node.js >=22.16, >=23.11, or >=24",
140 }),
141 );
142 }
143 return Effect.void;
144});
145
146// node:sqlite supplies errcode; classifySqliteError expects errno.
147const classifyError = (cause: unknown, message: string, operation: string) => {
148 if (
149 Predicate.hasProperty(cause, "errcode") &&
150 typeof cause.errcode === "number" &&
151 !Predicate.hasProperty(cause, "errno")
152 ) {
153 Object.assign(cause, { errno: cause.errcode });
154 }
155 return classifySqliteError(cause, { message, operation });
156};
157
158// SQLite retains text/blob bindings until rebind or finalization; node:sqlite cannot clear them.
159const LARGE_BINDING_SIZE = 64 * 1024;
160
161const isBlob = (value: unknown): value is ArrayBufferView | ArrayBufferLike =>
162 ArrayBuffer.isView(value) || NodeUtil.types.isAnyArrayBuffer(value);
163
164const bindingSize = (value: unknown): number => {
165 if (typeof value === "string") {
166 return value.length;
167 }
168 return isBlob(value) ? value.byteLength : 0;
169};
170
171const hasLargeBinding = (params: ReadonlyArray<unknown>): boolean => {
172 try {
173 let size = 0;
174 for (const param of params) {
175 if (
176 (typeof param === "object" || typeof param === "function") &&
177 param !== null &&
178 !isBlob(param)
179 ) {
180 // Named parameters may be callable objects; read data properties only to avoid invoking getters twice.
181 for (const property of Object.values(Object.getOwnPropertyDescriptors(param))) {
182 if (!property.enumerable) continue;
183 if (!("value" in property)) return true;
184 size += bindingSize(property.value);
185 }
186 } else {
187 size += bindingSize(param);
188 }
189 }
190 return size > LARGE_BINDING_SIZE;
191 } catch {
192 return true;
193 }
194};
195
196// Inline worker source also runs in bundles and single-executable builds without module files.
197const workerSource = `
198const { parentPort, workerData } = require("node:worker_threads");
199const { DatabaseSync } = require("node:sqlite");
200const describe = (error) => ({
201 message: String(error && error.message ? error.message : error),
202 code: error && error.code,
203 errcode: error && error.errcode,
204 errstr: error && error.errstr,
205});
206let db;
207try {
208 // A read-only open cannot be undone by SQL, unlike PRAGMA query_only.
209 const next = new DatabaseSync(workerData.filename, {
210 readOnly: workerData.readOnly,
211 allowExtension: workerData.allowExtension,
212 });
213 next.exec("PRAGMA busy_timeout = 5000;");
214 // Reads the header, so a file that is not a database fails here rather than on first use.
215 const mode = String(next.prepare("PRAGMA journal_mode").get().journal_mode).toLowerCase();
216 // Without WAL a reader's shared lock blocks the writer's commits, so refuse.
217 if (workerData.requireWal && mode !== "wal") throw new Error("database uses journal_mode " + mode + ", not wal");
218 db = next;
219 parentPort.postMessage({ type: "ready" });
220} catch (error) {
221 parentPort.postMessage({ type: "unavailable", ...describe(error) });
222}
223const statements = new Map();
224const prepare = (sql, cached) => {
225 if (!cached) return db.prepare(sql);
226 let statement = statements.get(sql);
227 if (statement === undefined) {
228 statement = db.prepare(sql);
229 if (statements.size >= workerData.prepareCacheSize) statements.delete(statements.keys().next().value);
230 } else {
231 statements.delete(sql);
232 }
233 statements.set(sql, statement);
234 return statement;
235};
236parentPort.on("message", (request) => {
237 if (request.type === "close") {
238 // Closing lets SQLite checkpoint and remove the WAL, which terminating would skip.
239 try {
240 if (db !== undefined) db.close();
241 } catch {}
242 parentPort.close();
243 return;
244 }
245 const { id, sql, params, cached, values, raw, safeIntegers } = request;
246 let operation = "open";
247 try {
248 if (db === undefined) throw new Error("database connection is unavailable");
249 operation = "prepare";
250 const statement = prepare(sql, cached);
251 operation = "execute";
252 statement.setReadBigInts(safeIntegers);
253 if (statement.columns().length > 0) {
254 statement.setReturnArrays(values);
255 parentPort.postMessage({ id, rows: statement.all(...params) });
256 } else {
257 const result = statement.run(...params);
258 parentPort.postMessage({ id, rows: raw ? result : [] });
259 }
260 } catch (error) {
261 parentPort.postMessage({ id, error: { operation, ...describe(error) } });
262 }
263});
264`;
265
266interface WorkerRequest {
267 readonly sql: string;
268 readonly params: ReadonlyArray<unknown>;
269 readonly cached: boolean;
270 readonly values: boolean;
271 readonly raw: boolean;
272}
273
274interface WorkerFailure {
275 readonly message: string;
276 readonly code?: unknown;
277 readonly errcode?: unknown;
278 readonly errstr?: unknown;
279}
280
281type WorkerStartup =
282 | { readonly type: "ready" }
283 | ({ readonly type: "unavailable" } & WorkerFailure);
284
285type WorkerReply =
286 | { readonly id: number; readonly rows: ReadonlyArray<any> }
287 | {
288 readonly id: number;
289 readonly error: WorkerFailure & { readonly operation: "open" | "prepare" | "execute" };
290 };
291
292const workerFailureMessages = {
293 open: "Failed to open database",
294 prepare: "Failed to prepare statement",
295 execute: "Failed to execute statement",
296} as const;
297
298const failed = (cause: unknown, message: string, operation: string) =>
299 new SqlError({ reason: classifyError(cause, message, operation) });
300
301const failedWith = (
302 { message, code, errcode, errstr }: WorkerFailure,
303 operation: keyof typeof workerFailureMessages,
304) =>
305 failed(
306 Object.assign(new Error(message), { code, errcode, errstr }),
307 workerFailureMessages[operation],
308 operation,
309 );
310
311// Read-transaction cleanup succeeds after worker exit because it can no longer change anything.
312const endsReadTransaction = (sql: string) => /^\s*ROLLBACK\b|^\s*COMMIT\b/i.test(sql);
313
314// WAL readers run concurrently; one worker serialized every read behind the slowest transaction.
315const READER_POOL_SIZE = 4;
316const READER_STARTUP_TIMEOUT = Duration.seconds(10);
317
318// Bind transactions to one worker; fail after its exit instead of resuming outside the transaction on a replacement.
319const makeWorkerConnection = Effect.fnUntraced(function* (options: {
320 readonly filename: string;
321 readonly role: "reader" | "writer";
322 readonly readOnly: boolean;
323 readonly onReply?: () => void;
324}) {
325 const label = options.role === "reader" ? "Database reader" : "Database writer";
326 const readOnly = options.readOnly;
327 let current:
328 | {
329 readonly worker: NodeWorkerThreads.Worker;
330 readonly connection: Connection;
331 readonly started: Promise<WorkerStartup>;
332 }
333 | undefined;
334 let nextRequestId = 0;
335 // A replacement writer would lose the pragmas callers set on it, such as foreign_keys.
336 let writerStopped: string | undefined;
337
338 const start = () => {
339 const worker = new NodeWorkerThreads.Worker(workerSource, {
340 eval: true,
341 workerData: {
342 filename: options.filename,
343 readOnly,
344 requireWal: options.role === "reader",
345 allowExtension: false,
346 prepareCacheSize: 200,
347 },
348 });
349 let announce: (startup: WorkerStartup) => void = () => {};
350 const started = new Promise<WorkerStartup>((resolve) => {
351 announce = resolve;
352 });
353 let stoppedMessage: string | undefined;
354 const pending = new Map<
355 number,
356 { readonly endsTransaction: boolean; readonly resume: (reply: WorkerReply) => void }
357 >();
358 const stop = (message: string) => {
359 stoppedMessage = message;
360 if (options.role === "writer") writerStopped = message;
361 announce({ type: "unavailable", message });
362 if (current?.worker === worker) current = undefined;
363 for (const [id, request] of pending) {
364 request.resume(
365 request.endsTransaction
366 ? { id, rows: [] }
367 : { id, error: { operation: "execute", message } },
368 );
369 }
370 pending.clear();
371 };
372 worker.on("message", (reply: WorkerReply | WorkerStartup) => {
373 if ("type" in reply) {
374 announce(reply);
375 if (pending.size === 0) worker.unref();
376 return;
377 }
378 const request = pending.get(reply.id);
379 if (request === undefined) return;
380 pending.delete(reply.id);
381 if (pending.size === 0) worker.unref();
382 options.onReply?.();
383 request.resume(reply);
384 });
385 worker.on("error", (cause) => stop(`${label} failed: ${cause.message}`));
386 worker.on("exit", (code) => stop(`${label} exited with code ${code}`));
387
388 const request = (input: WorkerRequest) =>
389 Effect.withFiber<ReadonlyArray<any>, SqlError>((fiber) => {
390 // A COMMIT that never reached a writer must fail; only its ROLLBACK is moot.
391 const endsTransaction = readOnly
392 ? endsReadTransaction(input.sql)
393 : /^\s*ROLLBACK\b/i.test(input.sql);
394 if (stoppedMessage !== undefined) {
395 return endsTransaction
396 ? Effect.succeed([])
397 : Effect.fail(failedWith({ message: stoppedMessage }, "execute"));
398 }
399 const safeIntegers = Boolean(Context.get(fiber.context, Client.SafeIntegers));
400 const reply = Effect.callback<ReadonlyArray<any>, SqlError>((resume) => {
401 const id = nextRequestId++;
402 pending.set(id, {
403 endsTransaction,
404 resume: (reply) =>
405 resume(
406 "rows" in reply
407 ? Effect.succeed(reply.rows)
408 : Effect.fail(failedWith(reply.error, reply.error.operation)),
409 ),
410 });
411 if (pending.size === 1) worker.ref();
412 const cached = input.cached && !hasLargeBinding(input.params);
413 // oxlint-disable-next-line unicorn/require-post-message-target-origin -- Node workers do not accept a target origin.
414 worker.postMessage({ id, safeIntegers, ...input, cached });
415 });
416 // A posted write runs to the end, as it did on the event loop; the caller and its lock
417 // must not move on while the worker is still applying it. Reads stay cancellable.
418 return options.role === "writer" ? Effect.uninterruptible(reply) : reply;
419 });
420
421 const connection = identity<Connection>({
422 execute(sql, params, rowTransform) {
423 const effect = request({ sql, params, cached: true, values: false, raw: false });
424 return rowTransform ? Effect.map(effect, rowTransform) : effect;
425 },
426 executeRaw(sql, params) {
427 return request({ sql, params, cached: true, values: false, raw: true });
428 },
429 executeValues(sql, params) {
430 return request({ sql, params, cached: true, values: true, raw: false });
431 },
432 executeValuesUnprepared(sql, params) {
433 return request({ sql, params: params ?? [], cached: false, values: true, raw: false });
434 },
435 executeUnprepared(sql, params, rowTransform) {
436 // Read snapshots need deferred BEGIN; readonly connections reject IMMEDIATE and EXCLUSIVE.
437 const statement = readOnly && /^\s*BEGIN\b/i.test(sql) ? "BEGIN" : sql;
438 const effect = request({
439 sql: statement,
440 params: params ?? [],
441 cached: false,
442 values: false,
443 raw: false,
444 });
445 return rowTransform ? Effect.map(effect, rowTransform) : effect;
446 },
447 executeStream(_sql, _params) {
448 return Stream.die(new UnsupportedNodeSqliteOperationError());
449 },
450 });
451 current = { worker, connection, started };
452 return current;
453 };
454
455 yield* Effect.addFinalizer(() =>
456 Effect.promise(async () => {
457 const open = current;
458 if (open === undefined) return;
459 if (options.role === "writer") {
460 await new Promise<void>((resolve) => {
461 open.worker.once("exit", () => resolve());
462 open.worker.ref();
463 // oxlint-disable-next-line unicorn/require-post-message-target-origin -- Node workers do not accept a target origin.
464 open.worker.postMessage({ type: "close" });
465 });
466 }
467 await open.worker.terminate();
468 }),
469 );
470
471 const retire = (worker: NodeWorkerThreads.Worker) => {
472 if (current?.worker === worker) current = undefined;
473 void worker.terminate();
474 };
475
476 return Effect.suspend((): Effect.Effect<Connection, SqlError> => {
477 if (current === undefined && writerStopped !== undefined) {
478 return Effect.fail(failedWith({ message: writerStopped }, "execute"));
479 }
480 let entry: NonNullable<typeof current>;
481 try {
482 entry = current ?? start();
483 } catch (cause) {
484 return Effect.fail(failed(cause, workerFailureMessages.open, "open"));
485 }
486 const started = Effect.promise(() => entry.started);
487 return Effect.flatMap(
488 Effect.withFiber((fiber) => {
489 const restore = Context.get(fiber.context, ReadOnly) ?? identity;
490 // Readers fall back to the writer; the writer has no fallback, so a slow disk only delays it.
491 return restore(
492 options.role === "reader"
493 ? Effect.timeoutOption(started, READER_STARTUP_TIMEOUT)
494 : Effect.asSome(started),
495 );
496 }),
497 (startup): Effect.Effect<Connection, SqlError> => {
498 const outcome = Option.getOrUndefined(startup);
499 if (outcome?.type === "ready") {
500 // A late ready event must not hand out a worker retired by another acquire timeout.
501 return current === entry
502 ? Effect.succeed(entry.connection)
503 : Effect.fail(failedWith({ message: `${label} stopped while starting` }, "open"));
504 }
505 retire(entry.worker);
506 return Effect.fail(
507 failedWith(
508 outcome ?? {
509 message: `${label} did not open the database within ${Duration.toMillis(READER_STARTUP_TIMEOUT)} ms`,
510 },
511 "open",
512 ),
513 );
514 },
515 );
516 });
517});
518
519// A reader that cannot start turns off the whole pool, so routing never picks a dead slot.
520const makeReader = Effect.fnUntraced(function* (filename: string, pool: { unavailable: boolean }) {
521 const acquire = yield* makeWorkerConnection({ filename, role: "reader", readOnly: true });
522 return Effect.suspend((): Effect.Effect<Connection | undefined> =>
523 pool.unavailable
524 ? Effect.undefined
525 : acquire.pipe(
526 Effect.map((connection) => (pool.unavailable ? undefined : connection)),
527 Effect.catch((error) => {
528 if (pool.unavailable) return Effect.undefined;
529 pool.unavailable = true;
530 const cause = error.reason.cause;
531 return Effect.logWarning("SQLite reader worker unavailable; reads use the writer", {
532 reason: cause instanceof Error ? cause.message : error.message,
533 }).pipe(Effect.as(undefined));
534 }),
535 ),
536 );
537});
538
539// A checkpoint copies WAL pages into the database and syncs it. On the writer's thread every
540// write would wait for that sync, so a second connection checkpoints beside the writes.
541const CHECKPOINT_FRAMES = 1000;
542const CHECKPOINT_IDLE_MS = 100;
543// If checkpoints fall behind constant writes, the writer checkpoints itself, delaying writes only.
544const WRITER_BACKSTOP_FRAMES = 10_000;
545
546const checkpointWorkerSource = `
547const { parentPort, workerData } = require("node:worker_threads");
548const { DatabaseSync } = require("node:sqlite");
549let db;
550let frames;
551let passive;
552let idle;
553const check = (force) => {
554 if (db === undefined) {
555 db = new DatabaseSync(workerData.filename);
556 db.exec("PRAGMA busy_timeout = 5000;");
557 // macOS requires F_FULLFSYNC to flush the drive cache.
558 db.exec("PRAGMA checkpoint_fullfsync = ON;");
559 frames = db.prepare("PRAGMA wal_checkpoint(NOOP)");
560 passive = db.prepare("PRAGMA wal_checkpoint(PASSIVE)");
561 }
562 const before = frames.get();
563 let state = before;
564 if (state.log - state.checkpointed >= workerData.frames || (force && state.log > state.checkpointed)) {
565 passive.get();
566 state = frames.get();
567 }
568 // The WAL restarts only once every frame is checkpointed, so catch up when writes pause.
569 // A reader pinning the WAL stops progress; the next write retries instead of a timer loop.
570 const progressed = !force || state.checkpointed > before.checkpointed;
571 if (state.log > state.checkpointed && progressed) {
572 idle = setTimeout(() => run(true), workerData.idleMs);
573 }
574 parentPort.postMessage({ type: "checked", log: state.log, checkpointed: state.checkpointed });
575};
576const run = (force) => {
577 clearTimeout(idle);
578 try {
579 check(force);
580 } catch (error) {
581 parentPort.postMessage({ type: "failed", message: String(error && error.message ? error.message : error) });
582 }
583};
584parentPort.on("message", () => run(false));
585`;
586
587const sqliteVersion = () => {
588 const db = new NodeSqlite.DatabaseSync(":memory:");
589 try {
590 return String(db.prepare("SELECT sqlite_version() AS version").get()?.version);
591 } finally {
592 db.close();
593 }
594};
595
596// Returns the writer's per-statement notification, or undefined when the writer checkpoints alone.
597const startCheckpointWorker = Effect.fnUntraced(function* (filename: string) {
598 const warn = (reason: string) =>
599 Effect.logWarning("SQLite checkpoint worker unavailable; the writer checkpoints its own WAL", {
600 reason,
601 });
602 const version = sqliteVersion();
603 const [major = 0, minor = 0] = version.split(".").map(Number);
604 // Older SQLite treats NOOP as PASSIVE, so counting frames would checkpoint on every write.
605 if (major < 3 || (major === 3 && minor < 51)) {
606 yield* warn(`SQLite ${version} has no NOOP checkpoint`);
607 return undefined;
608 }
609 const started = Result.try(
610 () =>
611 new NodeWorkerThreads.Worker(checkpointWorkerSource, {
612 eval: true,
613 workerData: { filename, frames: CHECKPOINT_FRAMES, idleMs: CHECKPOINT_IDLE_MS },
614 }),
615 );
616 if (Result.isFailure(started)) {
617 yield* warn(String(started.failure));
618 return undefined;
619 }
620 const worker = started.success;
621 worker.unref();
622 const runFork = Effect.runForkWith(yield* Effect.context<never>());
623 let stopped = false;
624 const stop = (reason: string) => {
625 if (stopped) return;
626 stopped = true;
627 void worker.terminate();
628 runFork(warn(reason));
629 };
630 worker.on("message", (message: { readonly type?: unknown; readonly message?: unknown }) => {
631 if (message?.type === "failed") stop(String(message.message));
632 });
633 worker.on("error", (cause) => stop(cause.message));
634 worker.on("exit", (code) => stop(`exited with code ${code}`));
635 yield* Effect.addFinalizer(() =>
636 Effect.promise(() => {
637 stopped = true;
638 return worker.terminate();
639 }),
640 );
641 let scheduled = false;
642 return () => {
643 if (stopped || scheduled) return;
644 scheduled = true;
645 // One check per event-loop turn is enough, however many statements finished in it.
646 setImmediate(() => {
647 scheduled = false;
648 // oxlint-disable-next-line unicorn/require-post-message-target-origin -- Node workers do not accept a target origin.
649 if (!stopped) worker.postMessage(0);
650 });
651 };
652});
653
654const make = Effect.fn("makeWithDatabase")(function* (
655 options: SqliteClientConfig,
656): Effect.fn.Return<Client.SqlClient, SqlError, Scope.Scope | Reactivity.Reactivity> {
657 yield* checkNodeSqliteCompat();
658
659 const compiler = Statement.makeCompilerSqlite();
660 const isFileDatabase = options.filename !== ":memory:" && options.filename !== "";
661
662 const makeConnection = Effect.gen(function* () {
663 const scope = yield* Effect.scope;
664 const db = yield* Effect.try({
665 try: () =>
666 new NodeSqlite.DatabaseSync(options.filename, {
667 readOnly: options.readonly ?? false,
668 allowExtension: false,
669 }),
670 catch: (cause) =>
671 new SqlError({
672 reason: classifyError(cause, "Failed to open database", "open"),
673 }),
674 });
675 yield* Scope.addFinalizer(
676 scope,
677 Effect.try({
678 try: () => db.close(),
679 catch: (cause) =>
680 new SqlError({
681 reason: classifyError(cause, "Failed to close database", "close"),
682 }),
683 }).pipe(Effect.orDie),
684 );
685 const statementReaderCache = new WeakMap<NodeSqlite.StatementSync, boolean>();
686 const hasRows = (statement: NodeSqlite.StatementSync): boolean => {
687 const cached = statementReaderCache.get(statement);
688 if (cached !== undefined) {
689 return cached;
690 }
691 const value = statement.columns().length > 0;
692 statementReaderCache.set(statement, value);
693 return value;
694 };
695
696 const prepare = (sql: string) =>
697 Effect.try({
698 try: () => db.prepare(sql),
699 catch: (cause) =>
700 new SqlError({
701 reason: classifyError(cause, "Failed to prepare statement", "prepare"),
702 }),
703 });
704
705 const prepareCache = yield* Cache.makeWith(prepare, {
706 capacity: 200,
707 // A transient prepare failure must not outlive the lock or missing schema.
708 timeToLive: (exit) => (Exit.isSuccess(exit) ? Duration.minutes(10) : Duration.zero),
709 });
710
711 const statementFor = (sql: string, params: ReadonlyArray<unknown>) =>
712 hasLargeBinding(params) ? prepare(sql) : Cache.get(prepareCache, sql);
713
714 const runStatement = (
715 statement: NodeSqlite.StatementSync,
716 params: ReadonlyArray<unknown>,
717 mode: "rows" | "raw" | "values" = "rows",
718 ) =>
719 Effect.withFiber<ReadonlyArray<any>, SqlError>((fiber) => {
720 let arrays = false;
721 let result: Exit.Exit<ReadonlyArray<unknown>, SqlError>;
722 try {
723 statement.setReadBigInts(Boolean(Context.get(fiber.context, Client.SafeIntegers)));
724 if (hasRows(statement)) {
725 arrays = mode === "values";
726 if (arrays) statement.setReturnArrays(true);
727 result = Exit.succeed(statement.all(...(params as any)));
728 } else {
729 const value = statement.run(...(params as any));
730 result = Exit.succeed(
731 mode === "raw" ? (value as unknown as ReadonlyArray<unknown>) : [],
732 );
733 }
734 } catch (cause) {
735 result = Exit.fail(
736 new SqlError({
737 reason: classifyError(cause, "Failed to execute statement", "execute"),
738 }),
739 );
740 }
741 if (arrays) {
742 try {
743 statement.setReturnArrays(false);
744 } catch (cause) {
745 return Exit.die(
746 new SqlError({
747 reason: classifyError(
748 cause,
749 "Failed to reset statement result mode",
750 "resetResultMode",
751 ),
752 }),
753 );
754 }
755 }
756 return result;
757 });
758
759 const run = (sql: string, params: ReadonlyArray<unknown>, raw = false) =>
760 Effect.flatMap(statementFor(sql, params), (statement) =>
761 runStatement(statement, params, raw ? "raw" : "rows"),
762 );
763 const runValues = (sql: string, params: ReadonlyArray<unknown>) =>
764 Effect.flatMap(statementFor(sql, params), (statement) =>
765 runStatement(statement, params, "values"),
766 );
767
768 return identity<Connection>({
769 execute(sql, params, rowTransform) {
770 return rowTransform ? Effect.map(run(sql, params), rowTransform) : run(sql, params);
771 },
772 executeRaw(sql, params) {
773 return run(sql, params, true);
774 },
775 executeValues(sql, params) {
776 return runValues(sql, params);
777 },
778 executeValuesUnprepared(sql, params) {
779 return Effect.flatMap(prepare(sql), (statement) =>
780 runStatement(statement, params ?? [], "values"),
781 );
782 },
783 executeUnprepared(sql, params, rowTransform) {
784 const effect = prepare(sql).pipe(
785 Effect.flatMap((statement) => runStatement(statement, params ?? [])),
786 );
787 return rowTransform ? Effect.map(effect, rowTransform) : effect;
788 },
789 executeStream(_sql, _params) {
790 return Stream.die(new UnsupportedNodeSqliteOperationError());
791 },
792 });
793 });
794
795 const semaphore = yield* Semaphore.make(1);
796 const writerWorker = options.writerWorker === true && isFileDatabase;
797 let checkpointAfterWrite: (() => void) | undefined;
798 const writerConnection: Effect.Effect<Connection, SqlError> = writerWorker
799 ? yield* makeWorkerConnection({
800 filename: options.filename,
801 role: "writer",
802 readOnly: options.readonly ?? false,
803 onReply: () => checkpointAfterWrite?.(),
804 })
805 : Effect.succeed(yield* makeConnection);
806 // Opening now fails the layer for a file that is not a database, not its first query.
807 const opened = yield* writerConnection;
808 // Started after the writer, so it stops first and the writer's close can checkpoint.
809 if (writerWorker && options.readonly !== true) {
810 checkpointAfterWrite = yield* startCheckpointWorker(options.filename);
811 if (checkpointAfterWrite !== undefined) {
812 yield* opened.executeUnprepared(
813 `PRAGMA wal_autocheckpoint = ${WRITER_BACKSTOP_FRAMES}`,
814 [],
815 undefined,
816 );
817 }
818 }
819 const readerPool = { unavailable: false };
820 const readers =
821 options.readerWorker && isFileDatabase
822 ? yield* Effect.forEach(Array.from({ length: READER_POOL_SIZE }), () =>
823 Effect.map(
824 Effect.all([makeReader(options.filename, readerPool), Semaphore.make(1)]),
825 ([acquire, lock]) => ({
826 acquire,
827 lock,
828 claims: 0,
829 }),
830 ),
831 )
832 : [];
833
834 interface Target {
835 readonly name: Acquired["connection"];
836 readonly lock: Semaphore.Semaphore;
837 readonly connection: Connection;
838 }
839 const writer = Effect.map(writerConnection, (connection): Target => ({
840 name: "writer",
841 lock: semaphore,
842 connection,
843 }));
844 let activeWriter: ConnectionOwner | undefined;
845 // Callers must run route uninterruptibly and pass `claimed` a finalizer that runs `release` once.
846 const route = (claimed: (release: Effect.Effect<void>) => Effect.Effect<void>) =>
847 Effect.withFiber((fiber): Effect.Effect<Target, SqlError> => {
848 if (readers.length === 0 || readerPool.unavailable || !Context.get(fiber.context, ReadOnly)) {
849 return writer;
850 }
851 // Least claimed wins and ties go to the lowest slot, so idle slots never start a worker.
852 let slot = readers[0]!;
853 for (const candidate of readers) if (candidate.claims < slot.claims) slot = candidate;
854 slot.claims++;
855 const reader = slot;
856 return Effect.andThen(
857 claimed(
858 Effect.sync(() => {
859 reader.claims--;
860 }),
861 ),
862 Effect.flatMap(reader.acquire, (readerConnection) =>
863 readerConnection === undefined
864 ? writer
865 : Effect.succeed<Target>({
866 name: "reader",
867 lock: reader.lock,
868 connection: readerConnection,
869 }),
870 ),
871 );
872 });
873
874 const acquired = (
875 target: { readonly name: Acquired["connection"] },
876 clock: Clock.Clock,
877 start: bigint,
878 ): Acquired => ({
879 connection: target.name,
880 lockWaitMs: Number((clock.monotonicTimeNanosUnsafe() - start) / 1000n) / 1000,
881 });
882
883 const borrower: Borrower = (use) =>
884 Effect.uninterruptibleMask((restore) => {
885 let release: Effect.Effect<void> = Effect.void;
886 const claimed = (next: Effect.Effect<void>) =>
887 Effect.sync(() => {
888 release = next;
889 });
890 return Effect.flatMap(route(claimed), (target) => restore(borrowTarget(target, use))).pipe(
891 Effect.ensuring(Effect.suspend(() => release)),
892 );
893 });
894 const borrowTarget = <A, E, R>(
895 target: Target,
896 use: (connection: Connection) => Effect.Effect<A, E, R>,
897 ): Effect.Effect<A, E, R> =>
898 Effect.withFiber((fiber) => {
899 const clock = fiber.getRef(Clock.Clock);
900 const start = clock.monotonicTimeNanosUnsafe();
901 const ownerAtWaitStart = target.name === "writer" ? activeWriter : undefined;
902 const owner = target.name === "writer" ? connectionOwner(fiber) : undefined;
903 const useConnection = Effect.suspend(() => {
904 if (target.name === "writer") activeWriter = owner;
905 if (
906 fiber.getRef(References.TracerEnabled) &&
907 fiber.getRef(Statement.SpanPropagationEnabled)
908 ) {
909 annotateAcquired(fiber, acquired(target, clock, start), ownerAtWaitStart);
910 }
911 return use(target.connection);
912 });
913 const useOwnedConnection =
914 target.name === "writer"
915 ? Effect.onExit(useConnection, () =>
916 Effect.sync(() => {
917 if (activeWriter === owner) activeWriter = undefined;
918 }),
919 )
920 : useConnection;
921 return target.lock.withPermits(1)(useOwnedConnection);
922 });
923 const claimedUntilScopeCloses = (release: Effect.Effect<void>) =>
924 Effect.withFiber((fiber) =>
925 Scope.addFinalizer(Context.getUnsafe(fiber.context, Scope.Scope), release),
926 );
927 const transactionAcquirer = Effect.uninterruptible(route(claimedUntilScopeCloses)).pipe(
928 Effect.flatMap((target) =>
929 Effect.uninterruptibleMask((restore) => {
930 const fiber = Fiber.getCurrent()!;
931 const scope = Context.getUnsafe(fiber.context, Scope.Scope);
932 const clock = fiber.getRef(Clock.Clock);
933 const start = clock.monotonicTimeNanosUnsafe();
934 const pending = fiber.getRef(TransactionAcquire);
935 const ownerAtWaitStart = target.name === "writer" ? activeWriter : undefined;
936 const owner = target.name === "writer" ? connectionOwner(fiber) : undefined;
937 const acquire = Effect.tap((pending?.restore ?? restore)(target.lock.take(1)), () => {
938 if (target.name === "writer") activeWriter = owner;
939 if (pending !== undefined) {
940 pending.acquired = acquired(target, clock, start);
941 pending.ownerAtWaitStart = ownerAtWaitStart;
942 }
943 return Scope.addFinalizer(
944 scope,
945 Effect.sync(() => {
946 if (target.name === "writer" && activeWriter === owner) activeWriter = undefined;
947 }).pipe(Effect.andThen(target.lock.release(1))),
948 );
949 });
950 return Effect.as(
951 // A scheduler yield after taking the permit must not precede its release finalizer.
952 Effect.provideService(acquire, Scheduler.PreventSchedulerYield, true),
953 target.connection,
954 );
955 }),
956 ),
957 );
958
959 const client = yield* Client.make({
960 driver: "node:sqlite",
961 acquirer: transactionAcquirer,
962 borrower,
963 compiler,
964 transactionAcquirer,
965 // BEGIN IMMEDIATE avoids read-then-write SQLITE_BUSY_SNAPSHOT, which busy_timeout cannot resolve.
966 beginTransaction: options.readonly === true ? "BEGIN" : "BEGIN IMMEDIATE",
967 spanAttributes: [
968 ...(options.spanAttributes ? Object.entries(options.spanAttributes) : []),
969 [ATTR_DB_SYSTEM_NAME, "sqlite"],
970 ],
971 });
972 const withTransaction = client.withTransaction;
973 return Object.assign(client, {
974 withTransaction: <A, E, R>(effect: Effect.Effect<A, E, R>) =>
975 Effect.uninterruptibleMask((restore) => {
976 // SqlClient masks setup; acquisition must preserve the caller's interruptibility.
977 const pending: {
978 acquired?: Acquired;
979 ownerAtWaitStart: ConnectionOwner | undefined;
980 restore: typeof Effect.interruptible | undefined;
981 } = {
982 ownerAtWaitStart: undefined,
983 restore,
984 };
985 const body = Effect.withFiber<A, E, R>((fiber) => {
986 annotateAcquired(fiber, pending.acquired, pending.ownerAtWaitStart);
987 pending.restore = undefined;
988 return effect;
989 });
990 return restore(
991 withTransaction(body).pipe(Effect.provideService(TransactionAcquire, pending)),
992 );
993 }),
994 });
995});
996
997export const layer = (config: SqliteClientConfig): Layer.Layer<Client.SqlClient, SqlError> =>
998 Layer.effect(Client.SqlClient, make(config)).pipe(Layer.provide(Reactivity.layer));