import type { DatabaseSync, SQLInputValue, SQLOutputValue, StatementSync, } from "node:sqlite"; import { CompiledQuery, createQueryId, type DatabaseConnection, type Dialect, type Driver, IdentifierNode, type Kysely, type QueryCompiler, type QueryResult, RawNode, type RootOperationNode, SelectQueryNode, SqliteAdapter, SqliteIntrospector, SqliteQueryCompiler, } from "kysely"; /** * Query kinds whose compiled SQL is stable enough to keep a prepared statement * around for. Everything else (DDL, raw SQL, `begin`/`commit`) is prepared fresh. */ const CACHEABLE_QUERY_KINDS = new Set([ "SelectQueryNode", "InsertQueryNode", "UpdateQueryNode", "DeleteQueryNode", "MergeQueryNode", ]); /** * Leading keywords of raw statements that can not change the schema, so the * column lists the statement cache is holding stay valid across them. Raw DDL * (`create`, `alter`, `drop`, ...) is not here and clears the cache. */ const SCHEMA_PRESERVING_RAW_COMMANDS = new Set([ "begin", "commit", "rollback", "savepoint", "release", "select", "with", "insert", "update", "delete", "replace", "pragma", "analyze", "explain", ]); const STATEMENT_CACHE_SIZE = 5000; const NO_JSON_OUTPUT_NAMES: ReadonlySet = new Set(); export interface NodeSqliteDialectConfig { database: DatabaseSync; /** * Keeps prepared statements around between queries, keyed by their SQL. Saves * a re-compile per query at the cost of holding onto the compiled programs. * Off by default because it assumes the schema does not change under the * connection, which is not true while migrations run. */ cacheStatements?: boolean; /** * "Table.column" names whose text content is a JSON document. When given, * result values of these columns are parsed into objects. Other text columns * are always returned verbatim, so JSON-shaped user input stays a string. * Origin metadata from `statement.columns()` sees through aliases, views, * subqueries and CTEs, so the names here are the underlying table names * (e.g. `AllTeam`, not the `Team` view). */ jsonColumns?: ReadonlySet; /** * Output names of a query's computed result columns whose value is a JSON * document, which is what `jsonArrayFrom`/`jsonObjectFrom` subqueries compile * to. SQLite reports no origin for a computed expression, so only the query's * own AST tells those apart from an ordinary `coalesce(...)` over user text. * Called once per prepared statement. Requires {@link jsonColumns}. */ computedJsonColumns?: (query: RootOperationNode) => ReadonlySet; } /** * Kysely dialect backed by Node's built-in `node:sqlite` module, replacing the * `better-sqlite3` native addon that Kysely's own `SqliteDialect` expects. * * Rows come back from `node:sqlite` as arrays rather than objects: the objects it * builds itself are both slower to produce and have a `null` prototype, which is * not what the rest of the codebase (or Kysely's own dialects) hand out. */ export class NodeSqliteDialect implements Dialect { readonly #config: NodeSqliteDialectConfig; constructor(config: NodeSqliteDialectConfig) { this.#config = config; } createDriver(): Driver { return new NodeSqliteDriver(this.#config); } createQueryCompiler(): QueryCompiler { return new SqliteQueryCompiler(); } createAdapter(): SqliteAdapter { return new SqliteAdapter(); } createIntrospector(db: Kysely): SqliteIntrospector { return new SqliteIntrospector(db); } } class NodeSqliteDriver implements Driver { readonly #config: NodeSqliteDialectConfig; #connection?: NodeSqliteConnection; constructor(config: NodeSqliteDialectConfig) { this.#config = config; } async init(): Promise { this.#connection = new NodeSqliteConnection(this.#config); } async acquireConnection(): Promise { return this.#connection!; } async beginTransaction(connection: DatabaseConnection): Promise { await connection.executeQuery(CompiledQuery.raw("begin")); } async commitTransaction(connection: DatabaseConnection): Promise { await connection.executeQuery(CompiledQuery.raw("commit")); } async rollbackTransaction(connection: DatabaseConnection): Promise { await connection.executeQuery(CompiledQuery.raw("rollback")); } async savepoint( connection: DatabaseConnection, savepointName: string, compileQuery: QueryCompiler["compileQuery"], ): Promise { await connection.executeQuery( compileQuery( savepointCommand("savepoint", savepointName), createQueryId(), ), ); } async rollbackToSavepoint( connection: DatabaseConnection, savepointName: string, compileQuery: QueryCompiler["compileQuery"], ): Promise { await connection.executeQuery( compileQuery( savepointCommand("rollback to", savepointName), createQueryId(), ), ); } async releaseSavepoint( connection: DatabaseConnection, savepointName: string, compileQuery: QueryCompiler["compileQuery"], ): Promise { await connection.executeQuery( compileQuery(savepointCommand("release", savepointName), createQueryId()), ); } async releaseConnection(): Promise { // the single connection is never handed back to a pool } async destroy(): Promise { this.#connection?.dispose(); this.#config.database.close(); } } /** Which result columns of one query hold a JSON document: by column origin, and by output name for the columns that have no origin. */ interface JsonColumns { byOrigin: ReadonlySet; byOutputName: ReadonlySet; } interface PreparedStatement { statement: StatementSync; /** Empty for statements that return no rows, which is how writes are detected. */ columnNames: string[]; /** Per result column: parse text values as JSON when building rows. */ jsonColumnFlags: boolean[]; /** Kept for re-deriving the flags when the column list turns out to be stale. */ jsonColumns?: JsonColumns; } class NodeSqliteConnection implements DatabaseConnection { readonly #database: DatabaseSync; readonly #cacheStatements: boolean; readonly #jsonColumns?: ReadonlySet; readonly #computedJsonColumns?: ( query: RootOperationNode, ) => ReadonlySet; readonly #cache = new Map(); constructor(config: NodeSqliteDialectConfig) { this.#database = config.database; this.#cacheStatements = config.cacheStatements ?? false; this.#jsonColumns = config.jsonColumns; this.#computedJsonColumns = config.computedJsonColumns; } async executeQuery(compiledQuery: CompiledQuery): Promise> { const prepared = this.#preparedStatementFor(compiledQuery); const parameters = compiledQuery.parameters as SQLInputValue[]; if (prepared.columnNames.length > 0) { return { rows: readRows(prepared, parameters) }; } const { changes, lastInsertRowid } = prepared.statement.run(...parameters); return { insertId: BigInt(lastInsertRowid), numAffectedRows: BigInt(changes), rows: [], }; } async *streamQuery( compiledQuery: CompiledQuery, ): AsyncIterableIterator> { if (!SelectQueryNode.is(compiledQuery.query)) { throw new Error( "Sqlite driver only supports streaming of select queries", ); } // deliberately uncached: the cursor stays open across yields, so sharing the // statement with another query would reset it mid-iteration const prepared = prepare( this.#database, compiledQuery.sql, this.#jsonColumnsFor(compiledQuery.query), ); const parameters = compiledQuery.parameters as SQLInputValue[]; for (const row of prepared.statement.iterate(...parameters)) { yield { rows: [toRow(prepared, row as unknown as SQLOutputValue[])], }; } } dispose() { this.#cache.clear(); } #preparedStatementFor(compiledQuery: CompiledQuery): PreparedStatement { const { sql, query } = compiledQuery; if (!this.#cacheStatements || !CACHEABLE_QUERY_KINDS.has(query.kind)) { // a schema change invalidates every column list the cache is holding if (query.kind !== "RawNode" || canChangeSchema(sql)) { this.#cache.clear(); } return prepare(this.#database, sql, this.#jsonColumnsFor(query)); } const cached = this.#cache.get(sql); if (cached) { // re-insert so the least recently used entry stays at the front this.#cache.delete(sql); this.#cache.set(sql, cached); return cached; } const prepared = prepare(this.#database, sql, this.#jsonColumnsFor(query)); if (this.#cache.size >= STATEMENT_CACHE_SIZE) { this.#cache.delete(this.#cache.keys().next().value!); } this.#cache.set(sql, prepared); return prepared; } #jsonColumnsFor(query: RootOperationNode): JsonColumns | undefined { if (!this.#jsonColumns) return undefined; return { byOrigin: this.#jsonColumns, byOutputName: this.#computedJsonColumns?.(query) ?? NO_JSON_OUTPUT_NAMES, }; } } function canChangeSchema(sql: string) { const firstKeyword = sql .trimStart() .split(/[\s;(]/, 1)[0] .toLowerCase(); return !SCHEMA_PRESERVING_RAW_COMMANDS.has(firstKeyword); } function prepare( database: DatabaseSync, sql: string, jsonColumns: JsonColumns | undefined, ): PreparedStatement { const statement = database.prepare(sql); statement.setReturnArrays(true); return { statement, jsonColumns, ...columnMetadata(statement, jsonColumns) }; } function columnMetadata( statement: StatementSync, jsonColumns: JsonColumns | undefined, ) { const columns = statement.columns(); return { columnNames: columns.map((it) => it.name), jsonColumnFlags: columns.map((it) => { if (!jsonColumns) return false; // a null origin is a computed expression (a jsonArrayFrom subquery, but also // e.g. a coalesce over user text), which only the query itself can classify if (it.column === null) return jsonColumns.byOutputName.has(it.name); return jsonColumns.byOrigin.has(`${it.table}.${it.column}`); }), }; } function readRows( prepared: PreparedStatement, parameters: SQLInputValue[], ): R[] { const rawRows = prepared.statement.all( ...parameters, ) as unknown as SQLOutputValue[][]; if (rawRows.length === 0) return []; // `select *` widens when a migration adds a column, leaving a cached statement // with a stale column list until the next read notices the mismatch if (rawRows[0].length !== prepared.columnNames.length) { Object.assign( prepared, columnMetadata(prepared.statement, prepared.jsonColumns), ); } const rows = new Array(rawRows.length); for (let i = 0; i < rawRows.length; i++) { rows[i] = toRow(prepared, rawRows[i]); } return rows; } function toRow(prepared: PreparedStatement, rawRow: SQLOutputValue[]): R { const { columnNames, jsonColumnFlags } = prepared; const row: Record = {}; for (let i = 0; i < columnNames.length; i++) { const value = rawRow[i]; row[columnNames[i]] = jsonColumnFlags[i] && typeof value === "string" && maybeJson(value) ? parseJsonValue(value) : value; } return row as R; } function maybeJson(value: string) { return ( (value.startsWith("{") && value.endsWith("}")) || (value.startsWith("[") && value.endsWith("]")) ); } function parseJsonValue(value: string): unknown { let parsed: unknown; try { parsed = JSON.parse(value); } catch { return value; } sanitizeParsedJson(parsed); return parsed; } /** Strips `__proto__` and `constructor.prototype` so the parsed document can not prototype-pollute downstream merges. */ function sanitizeParsedJson(value: unknown) { if (Array.isArray(value)) { for (const item of value) { sanitizeParsedJson(item); } return; } if (!isPlainObject(value)) return; for (const key of Object.keys(value)) { if (key === "__proto__") { delete value[key]; continue; } const child = value[key]; if ( key === "constructor" && isPlainObject(child) && Object.hasOwn(child, "prototype") ) { delete child.prototype; } sanitizeParsedJson(child); } } function isPlainObject(value: unknown): value is Record { if (typeof value !== "object" || value === null) return false; const proto = Object.getPrototypeOf(value); return proto === null || proto === Object.prototype; } function savepointCommand(command: string, savepointName: string) { return RawNode.createWithChildren([ RawNode.createWithSql(`${command} `), IdentifierNode.create(savepointName), ]); }