From 055ecce080ab38a4ecf4e6f61e9773b3f0469b1b Mon Sep 17 00:00:00 2001 From: Armin Ronacher Date: Thu, 6 Aug 2026 18:10:42 +0200 Subject: [PATCH] feat(agent): add SQLite SQL template queries --- .../session-backends/sqlite-node/CHANGELOG.md | 4 + .../session-backends/sqlite-node/src/index.ts | 7 +- .../sqlite-node/src/sqlite/branch-cache.ts | 28 ++-- .../sqlite-node/src/sqlite/index.ts | 1 + .../sqlite-node/src/sqlite/migrations.ts | 12 +- .../sqlite-node/src/sqlite/repo.ts | 7 +- .../sqlite-node/src/sqlite/search-backend.ts | 62 ++++---- .../sqlite-node/src/sqlite/sql.ts | 62 ++++++++ .../src/sqlite/storage/branch-entries.ts | 134 ++++++++---------- .../src/sqlite/storage/branch-tips.ts | 25 ++-- .../sqlite-node/src/sqlite/storage/entries.ts | 48 +++---- .../sqlite-node/src/sqlite/storage/facts.ts | 60 +++----- .../sqlite-node/src/sqlite/storage/lanes.ts | 116 ++++++--------- .../sqlite-node/src/sqlite/storage/records.ts | 88 ++++-------- .../src/sqlite/storage/session-sequences.ts | 13 +- .../src/sqlite/storage/session-stats.ts | 41 +++--- .../src/sqlite/storage/sessions.ts | 82 +++++------ .../src/sqlite/storage/writer-leases.ts | 43 +++--- .../sqlite-node/test/sql.test.ts | 39 +++++ 19 files changed, 417 insertions(+), 455 deletions(-) create mode 100644 packages/session-backends/sqlite-node/src/sqlite/sql.ts create mode 100644 packages/session-backends/sqlite-node/test/sql.test.ts diff --git a/packages/session-backends/sqlite-node/CHANGELOG.md b/packages/session-backends/sqlite-node/CHANGELOG.md index e18c7330a..3260d5a1b 100644 --- a/packages/session-backends/sqlite-node/CHANGELOG.md +++ b/packages/session-backends/sqlite-node/CHANGELOG.md @@ -2,6 +2,10 @@ ## [Unreleased] +### Added + +- Added the composable, parameterized `sql` template tag for SQLite queries. + ## [0.84.0] - 2026-08-06 ### Breaking Changes diff --git a/packages/session-backends/sqlite-node/src/index.ts b/packages/session-backends/sqlite-node/src/index.ts index 98c7fdfa4..83552d1cc 100644 --- a/packages/session-backends/sqlite-node/src/index.ts +++ b/packages/session-backends/sqlite-node/src/index.ts @@ -1,5 +1,6 @@ import type { SQLInputValue } from "node:sqlite"; import { DatabaseSync } from "node:sqlite"; +import { sql } from "./sqlite/sql.ts"; import type { SqliteDatabase, SqliteDatabaseFactory, SqliteRunResult, SqliteStatement } from "./sqlite/types.ts"; function isNamedParameters(value: unknown): value is Record { @@ -65,17 +66,17 @@ class NodeSqliteDatabase implements SqliteDatabase { } transaction(fn: () => T): T { - this.db.exec("BEGIN IMMEDIATE"); + sql`BEGIN IMMEDIATE`.exec(this); try { const result = fn(); if (isAsyncResult(result)) { throw new TypeError("SQLite transaction callbacks must be synchronous"); } - this.db.exec("COMMIT"); + sql`COMMIT`.exec(this); return result; } catch (error) { try { - this.db.exec("ROLLBACK"); + sql`ROLLBACK`.exec(this); } catch { // Ignore rollback errors to rethrow original error. } diff --git a/packages/session-backends/sqlite-node/src/sqlite/branch-cache.ts b/packages/session-backends/sqlite-node/src/sqlite/branch-cache.ts index 0e5024cb8..eaef85f58 100644 --- a/packages/session-backends/sqlite-node/src/sqlite/branch-cache.ts +++ b/packages/session-backends/sqlite-node/src/sqlite/branch-cache.ts @@ -1,5 +1,6 @@ import { SessionError } from "@earendil-works/pi-agent-core"; import { uuidv7 } from "@earendil-works/pi-ai"; +import { sql } from "./sql.ts"; import { copyBranchEntriesThroughSeq, deleteBranchEntries, @@ -7,7 +8,6 @@ import { insertBranchEntry, readBranchContainingEntry, } from "./storage/branch-entries.ts"; - import { deleteBranchTips, insertBranchTip, readBranchTipBranchId, updateBranchTip } from "./storage/branch-tips.ts"; import type { SqliteDatabase } from "./types.ts"; @@ -17,32 +17,28 @@ export function deleteBranchCache(db: SqliteDatabase, sessionId: string) { } export function rebuildBranchCache(db: SqliteDatabase, sessionId: string) { - const tips = db - .prepare( - `SELECT leaf.id - FROM entries AS leaf - WHERE leaf.session_id = ? - AND NOT EXISTS ( - SELECT 1 FROM entries AS child WHERE child.session_id = leaf.session_id AND child.parent_id = leaf.id - ) - ORDER BY leaf.seq`, - ) - .all<{ id: string }>(sessionId); + const tips = sql`SELECT leaf.id + FROM entries AS leaf + WHERE leaf.session_id = ${sessionId} + AND NOT EXISTS ( + SELECT 1 FROM entries AS child WHERE child.session_id = leaf.session_id AND child.parent_id = leaf.id + ) + ORDER BY leaf.seq`.all<{ id: string }>(db); deleteBranchCache(db, sessionId); for (const tip of tips) buildCachedBranch(db, sessionId, tip.id); } export function buildCachedBranch(db: SqliteDatabase, sessionId: string, leafId: string) { - db.exec("SAVEPOINT build_branch_cache"); + sql`SAVEPOINT build_branch_cache`.exec(db); try { const branchId = uuidv7(); insertBranchEntriesForPath(db, sessionId, branchId, leafId); insertBranchTip(db, sessionId, leafId, branchId); - db.exec("RELEASE SAVEPOINT build_branch_cache"); + sql`RELEASE SAVEPOINT build_branch_cache`.exec(db); } catch (error) { try { - db.exec("ROLLBACK TO SAVEPOINT build_branch_cache"); - db.exec("RELEASE SAVEPOINT build_branch_cache"); + sql`ROLLBACK TO SAVEPOINT build_branch_cache`.exec(db); + sql`RELEASE SAVEPOINT build_branch_cache`.exec(db); } catch { // Preserve the original build failure. } diff --git a/packages/session-backends/sqlite-node/src/sqlite/index.ts b/packages/session-backends/sqlite-node/src/sqlite/index.ts index 738f086a1..73f6a62dd 100644 --- a/packages/session-backends/sqlite-node/src/sqlite/index.ts +++ b/packages/session-backends/sqlite-node/src/sqlite/index.ts @@ -5,6 +5,7 @@ export { type SqliteWriterLeaseOptions, } from "./repo.ts"; export * from "./search-backend.ts"; +export * from "./sql.ts"; export type { SqliteDatabase, SqliteDatabaseFactory, diff --git a/packages/session-backends/sqlite-node/src/sqlite/migrations.ts b/packages/session-backends/sqlite-node/src/sqlite/migrations.ts index a0debe3f0..e171f4a01 100644 --- a/packages/session-backends/sqlite-node/src/sqlite/migrations.ts +++ b/packages/session-backends/sqlite-node/src/sqlite/migrations.ts @@ -1,5 +1,6 @@ import { readFile } from "node:fs/promises"; import { fileURLToPath } from "node:url"; +import { sql } from "./sql.ts"; import type { SqliteDatabase } from "./types.ts"; export interface SqliteMigration { @@ -23,28 +24,25 @@ export async function loadMigrations(): Promise { } function ensureMigrationsTable(db: SqliteDatabase): void { - db.exec(` + sql` CREATE TABLE IF NOT EXISTS migrations ( id TEXT PRIMARY KEY, applied_at TEXT NOT NULL ); -`); +`.exec(db); } export async function applyMigrations(db: SqliteDatabase): Promise { ensureMigrationsTable(db); const migrations = await loadMigrations(); - const appliedRows = db.prepare("SELECT id FROM migrations ORDER BY applied_at, id").all<{ id: string }>(); + const appliedRows = sql`SELECT id FROM migrations ORDER BY applied_at, id`.all<{ id: string }>(db); const applied = new Set(appliedRows.map((row) => row.id)); for (const migration of migrations) { if (applied.has(migration.id)) continue; db.transaction(() => { db.exec(migration.sql); - db.prepare("INSERT INTO migrations (id, applied_at) VALUES (?, ?)").run( - migration.id, - new Date().toISOString(), - ); + sql`INSERT INTO migrations (id, applied_at) VALUES (${migration.id}, ${new Date().toISOString()})`.run(db); }); applied.add(migration.id); } diff --git a/packages/session-backends/sqlite-node/src/sqlite/repo.ts b/packages/session-backends/sqlite-node/src/sqlite/repo.ts index d75c7a5ab..8eb8ba96c 100644 --- a/packages/session-backends/sqlite-node/src/sqlite/repo.ts +++ b/packages/session-backends/sqlite-node/src/sqlite/repo.ts @@ -20,6 +20,7 @@ import { import { uuidv7 } from "@earendil-works/pi-ai"; import { appendEntryToBranchCache, buildCachedBranch, deleteBranchCache, rebuildBranchCache } from "./branch-cache.ts"; import { applyMigrations } from "./migrations.ts"; +import { sql } from "./sql.ts"; import { type CachedBranchEntryRow, queryCachedBranchRows, readCachedBranch } from "./storage/branch-entries.ts"; import { readBranchTipIds } from "./storage/branch-tips.ts"; import { @@ -169,9 +170,9 @@ function getParentPath(path: string): string { } function configureSqliteDatabase(db: SqliteDatabase): void { - db.exec("PRAGMA journal_mode=WAL"); - db.exec("PRAGMA synchronous=FULL"); - db.exec("PRAGMA busy_timeout=5000"); + sql`PRAGMA journal_mode=WAL`.exec(db); + sql`PRAGMA synchronous=FULL`.exec(db); + sql`PRAGMA busy_timeout=5000`.exec(db); } function timestampToText(timestamp: number): string { diff --git a/packages/session-backends/sqlite-node/src/sqlite/search-backend.ts b/packages/session-backends/sqlite-node/src/sqlite/search-backend.ts index 11f309f19..e22de3340 100644 --- a/packages/session-backends/sqlite-node/src/sqlite/search-backend.ts +++ b/packages/session-backends/sqlite-node/src/sqlite/search-backend.ts @@ -1,6 +1,7 @@ import type { SessionSearch, SessionSearchHit, SessionSearchOptions } from "@earendil-works/pi-agent-core"; import { getFileSystemResultOrThrow } from "@earendil-works/pi-agent-core"; import { applyMigrations } from "./migrations.ts"; +import { sql } from "./sql.ts"; import { decodeSessionMetadata, type SessionRow } from "./storage/sessions.ts"; import type { SqliteDatabase, @@ -18,9 +19,9 @@ function getParentPath(path: string): string { } function configureSqliteDatabase(db: SqliteDatabase): void { - db.exec("PRAGMA journal_mode=WAL"); - db.exec("PRAGMA synchronous=FULL"); - db.exec("PRAGMA busy_timeout=5000"); + sql`PRAGMA journal_mode=WAL`.exec(db); + sql`PRAGMA synchronous=FULL`.exec(db); + sql`PRAGMA busy_timeout=5000`.exec(db); } export interface SqliteSessionSearchOptions { @@ -30,14 +31,14 @@ export interface SqliteSessionSearchOptions { } function tableExists(db: SqliteDatabase, name: string): boolean { - return !!db - .prepare("SELECT 1 AS found FROM sqlite_master WHERE type = 'table' AND name = ? LIMIT 1") - .get<{ found: number }>(name); + return !!sql`SELECT 1 AS found FROM sqlite_master WHERE type = 'table' AND name = ${name} LIMIT 1`.get<{ + found: number; + }>(db); } function ensureSearchSchema(db: SqliteDatabase): void { const ftsExists = tableExists(db, "session_search_fts"); - db.exec(` + sql` CREATE VIRTUAL TABLE IF NOT EXISTS session_search_fts USING fts5( payload, content = 'entries', @@ -54,8 +55,8 @@ CREATE TRIGGER IF NOT EXISTS session_search_fts_au AFTER UPDATE OF payload ON en INSERT INTO session_search_fts(session_search_fts, rowid, payload) VALUES('delete', old.rowid, old.payload); INSERT INTO session_search_fts(rowid, payload) VALUES (new.rowid, new.payload); END; -`); - if (!ftsExists) db.exec("INSERT INTO session_search_fts(session_search_fts) VALUES('rebuild')"); +`.exec(db); + if (!ftsExists) sql`INSERT INTO session_search_fts(session_search_fts) VALUES('rebuild')`.exec(db); } /** SQLite FTS search over a co-located canonical session database. */ @@ -102,32 +103,25 @@ class SqliteSessionSearch implements SessionSearch { const db = await this.openDatabase(); try { const query = `"${text.replaceAll('"', '""')}"`; - const rows = db - .prepare( - `SELECT s.id, s.created_at, s.metadata, s.cwd, s.parent_session_id, - name_fact.seq IS NOT NULL AS has_session_name, - name_fact.value AS session_name, - se.id AS entry_id, se.timestamp, bm25(session_search_fts) AS score - FROM session_search_fts - JOIN entries AS se ON se.rowid = session_search_fts.rowid - JOIN sessions AS s ON s.id = se.session_id - LEFT JOIN facts AS name_fact - ON name_fact.session_id = s.id - AND name_fact.kind = 'name' - AND name_fact.key IS NULL - AND name_fact.seq = ( - SELECT MAX(f.seq) - FROM facts AS f - WHERE f.session_id = s.id AND f.kind = 'name' AND f.key IS NULL - ) - WHERE session_search_fts MATCH ? AND (? IS NULL OR s.cwd = ?) - ORDER BY score`, + const cwd = options.cwd ?? null; + const rows = sql`SELECT s.id, s.created_at, s.metadata, s.cwd, s.parent_session_id, + name_fact.seq IS NOT NULL AS has_session_name, + name_fact.value AS session_name, + se.id AS entry_id, se.timestamp, bm25(session_search_fts) AS score + FROM session_search_fts + JOIN entries AS se ON se.rowid = session_search_fts.rowid + JOIN sessions AS s ON s.id = se.session_id + LEFT JOIN facts AS name_fact + ON name_fact.session_id = s.id + AND name_fact.kind = 'name' + AND name_fact.key IS NULL + AND name_fact.seq = ( + SELECT MAX(f.seq) + FROM facts AS f + WHERE f.session_id = s.id AND f.kind = 'name' AND f.key IS NULL ) - .all( - query, - options.cwd ?? null, - options.cwd ?? null, - ); + WHERE session_search_fts MATCH ${query} AND (${cwd} IS NULL OR s.cwd = ${cwd}) + ORDER BY score`.all(db); const path = await this.getDatabasePath(); return rows.map((row) => ({ metadata: decodeSessionMetadata(row, path), diff --git a/packages/session-backends/sqlite-node/src/sqlite/sql.ts b/packages/session-backends/sqlite-node/src/sqlite/sql.ts new file mode 100644 index 000000000..4eaec7f44 --- /dev/null +++ b/packages/session-backends/sqlite-node/src/sqlite/sql.ts @@ -0,0 +1,62 @@ +import type { SqliteDatabase, SqliteRunResult } from "./types.ts"; + +type SqlTemplateValue = unknown | SqlQuery; + +/** A parameterized SQLite query produced by {@link sql}. */ +export class SqlQuery { + readonly queryText: string; + readonly params: readonly unknown[]; + + constructor(queryText: string, params: readonly unknown[] = []) { + this.queryText = queryText; + this.params = params; + } + + exec(db: SqliteDatabase): void { + if (this.params.length > 0) throw new TypeError("SQLite exec queries cannot have parameters"); + db.exec(this.queryText); + } + + run(db: SqliteDatabase): SqliteRunResult { + return db.prepare(this.queryText).run(...this.params); + } + + get(db: SqliteDatabase): TRow | undefined { + return db.prepare(this.queryText).get(...this.params); + } + + all(db: SqliteDatabase): TRow[] { + return db.prepare(this.queryText).all(...this.params); + } +} + +/** Builds a parameterized query. Nested queries are inlined; other interpolations become `?` parameters. */ +export function sql(strings: TemplateStringsArray, ...values: SqlTemplateValue[]): SqlQuery { + let queryText = strings[0] ?? ""; + const params: unknown[] = []; + for (let index = 0; index < values.length; index++) { + const value = values[index]; + if (value instanceof SqlQuery) { + queryText += value.queryText; + params.push(...value.params); + } else { + queryText += "?"; + params.push(value); + } + queryText += strings[index + 1] ?? ""; + } + return new SqlQuery(queryText, params); +} + +/** Joins trusted query fragments while preserving their parameter order. */ +export function joinSqlFragments(fragments: readonly SqlQuery[], separator: string): SqlQuery { + let queryText = ""; + const params: unknown[] = []; + for (let index = 0; index < fragments.length; index++) { + if (index > 0) queryText += separator; + const fragment = fragments[index]!; + queryText += fragment.queryText; + params.push(...fragment.params); + } + return new SqlQuery(queryText, params); +} diff --git a/packages/session-backends/sqlite-node/src/sqlite/storage/branch-entries.ts b/packages/session-backends/sqlite-node/src/sqlite/storage/branch-entries.ts index 9b604976e..caaf47fd6 100644 --- a/packages/session-backends/sqlite-node/src/sqlite/storage/branch-entries.ts +++ b/packages/session-backends/sqlite-node/src/sqlite/storage/branch-entries.ts @@ -1,4 +1,5 @@ import type { Entry } from "@earendil-works/pi-agent-core"; +import { joinSqlFragments, sql } from "../sql.ts"; import type { SqliteDatabase } from "../types.ts"; /** Derived root-to-tip branch cache membership. Canonical parent links remain in entries. */ @@ -24,11 +25,11 @@ export interface CachedBranchQuery { } export function readCachedBranch(db: SqliteDatabase, sessionId: string, leafId: string) { - const membership = db - .prepare( - "SELECT branch_id, entry_seq FROM branch_entries WHERE session_id = ? AND entry_id = ? ORDER BY branch_id LIMIT 1", - ) - .get<{ branch_id: string; entry_seq: number }>(sessionId, leafId); + const membership = sql`SELECT branch_id, entry_seq + FROM branch_entries + WHERE session_id = ${sessionId} AND entry_id = ${leafId} + ORDER BY branch_id + LIMIT 1`.get<{ branch_id: string; entry_seq: number }>(db); if (!membership) return undefined; return { branchId: membership.branch_id, leafSeq: membership.entry_seq }; } @@ -40,46 +41,45 @@ export function queryCachedBranchRows( query: CachedBranchQuery, ) { const oldestFirst = query.order === "oldestFirst"; - const boundaryParams: unknown[] = [sessionId, branch.branchId, branch.leafSeq]; - const stopPredicates: string[] = []; - if (query.stopAtType !== undefined) { - stopPredicates.push("stop_entry.type = ?"); - boundaryParams.push(query.stopAtType); - } - if (query.stopAtId !== undefined) { - stopPredicates.push("stop.entry_id = ?"); - boundaryParams.push(query.stopAtId); - } + const stopPredicates = []; + if (query.stopAtType !== undefined) stopPredicates.push(sql`stop_entry.type = ${query.stopAtType}`); + if (query.stopAtId !== undefined) stopPredicates.push(sql`stop.entry_id = ${query.stopAtId}`); - const boundary = stopPredicates.length - ? `WITH boundary AS ( - SELECT ${oldestFirst ? "MIN" : "MAX"}(stop.entry_seq) AS entry_seq - FROM branch_entries AS stop - JOIN entries AS stop_entry - ON stop_entry.session_id = stop.session_id AND stop_entry.id = stop.entry_id - WHERE stop.session_id = ? AND stop.branch_id = ? AND stop.entry_seq <= ? - AND (${stopPredicates.join(" OR ")}) - )` - : ""; - const range = stopPredicates.length - ? `AND b.entry_seq ${oldestFirst ? "<=" : ">="} COALESCE( - (SELECT entry_seq FROM boundary), ${oldestFirst ? branch.leafSeq : 0} - )` - : ""; - const sql = `${boundary} + const aggregate = oldestFirst ? sql`MIN` : sql`MAX`; + const comparison = oldestFirst ? sql`<=` : sql`>=`; + const direction = oldestFirst ? sql`ASC` : sql`DESC`; + const boundary = + stopPredicates.length === 0 + ? sql`` + : sql`WITH boundary AS ( + SELECT ${aggregate}(stop.entry_seq) AS entry_seq + FROM branch_entries AS stop + JOIN entries AS stop_entry + ON stop_entry.session_id = stop.session_id AND stop_entry.id = stop.entry_id + WHERE stop.session_id = ${sessionId} + AND stop.branch_id = ${branch.branchId} + AND stop.entry_seq <= ${branch.leafSeq} + AND (${joinSqlFragments(stopPredicates, " OR ")}) + )`; + const range = + stopPredicates.length === 0 + ? sql`` + : sql`AND b.entry_seq ${comparison} COALESCE( + (SELECT entry_seq FROM boundary), ${oldestFirst ? branch.leafSeq : 0} + )`; + return sql`${boundary} SELECT e.session_id, e.id, e.seq AS entry_seq, e.parent_id, e.type, e.timestamp, e.payload FROM branch_entries AS b JOIN entries AS e ON e.session_id = b.session_id AND e.id = b.entry_id - WHERE b.session_id = ? AND b.branch_id = ? AND b.entry_seq <= ? + WHERE b.session_id = ${sessionId} + AND b.branch_id = ${branch.branchId} + AND b.entry_seq <= ${branch.leafSeq} ${range} - ORDER BY b.entry_seq ${oldestFirst ? "ASC" : "DESC"}`; - - const params = [...(stopPredicates.length === 0 ? [] : boundaryParams), sessionId, branch.branchId, branch.leafSeq]; - return db.prepare(sql).all(...params); + ORDER BY b.entry_seq ${direction}`.all(db); } export function deleteBranchEntries(db: SqliteDatabase, sessionId: string) { - db.prepare("DELETE FROM branch_entries WHERE session_id = ?").run(sessionId); + sql`DELETE FROM branch_entries WHERE session_id = ${sessionId}`.run(db); } export function insertBranchEntry( @@ -91,42 +91,34 @@ export function insertBranchEntry( entryType: string, customType: string | null, ) { - db.prepare( - `INSERT INTO branch_entries + sql`INSERT INTO branch_entries (session_id, branch_id, entry_id, entry_seq, entry_type, custom_type) - VALUES (?, ?, ?, ?, ?, ?)`, - ).run(sessionId, branchId, entryId, entrySeq, entryType, customType); + VALUES (${sessionId}, ${branchId}, ${entryId}, ${entrySeq}, ${entryType}, ${customType})`.run(db); } export function insertBranchEntriesForPath(db: SqliteDatabase, sessionId: string, branchId: string, leafId: string) { - db.prepare( - `WITH RECURSIVE path(id, entry_seq, parent_id, type, custom_type) AS ( - SELECT id, seq, parent_id, type, - CASE WHEN type = 'custom' THEN json_extract(payload, '$.customType') ELSE NULL END - FROM entries - WHERE session_id = ? AND id = ? - UNION ALL - SELECT parent.id, parent.seq, parent.parent_id, parent.type, - CASE WHEN parent.type = 'custom' THEN json_extract(parent.payload, '$.customType') ELSE NULL END - FROM entries AS parent - JOIN path AS child ON child.parent_id = parent.id - WHERE parent.session_id = ? - ) - INSERT INTO branch_entries (session_id, branch_id, entry_id, entry_seq, entry_type, custom_type) - SELECT ?, ?, id, entry_seq, type, custom_type FROM path`, - ).run(sessionId, leafId, sessionId, sessionId, branchId); + sql`WITH RECURSIVE path(id, entry_seq, parent_id, type, custom_type) AS ( + SELECT id, seq, parent_id, type, + CASE WHEN type = 'custom' THEN json_extract(payload, '$.customType') ELSE NULL END + FROM entries + WHERE session_id = ${sessionId} AND id = ${leafId} + UNION ALL + SELECT parent.id, parent.seq, parent.parent_id, parent.type, + CASE WHEN parent.type = 'custom' THEN json_extract(parent.payload, '$.customType') ELSE NULL END + FROM entries AS parent + JOIN path AS child ON child.parent_id = parent.id + WHERE parent.session_id = ${sessionId} + ) + INSERT INTO branch_entries (session_id, branch_id, entry_id, entry_seq, entry_type, custom_type) + SELECT ${sessionId}, ${branchId}, id, entry_seq, type, custom_type FROM path`.run(db); } export function readBranchContainingEntry(db: SqliteDatabase, sessionId: string, entryId: string) { - const row = db - .prepare( - `SELECT b.branch_id, b.entry_seq - FROM branch_entries AS b - WHERE b.session_id = ? AND b.entry_id = ? - ORDER BY b.branch_id - LIMIT 1`, - ) - .get<{ branch_id: string; entry_seq: number }>(sessionId, entryId); + const row = sql`SELECT b.branch_id, b.entry_seq + FROM branch_entries AS b + WHERE b.session_id = ${sessionId} AND b.entry_id = ${entryId} + ORDER BY b.branch_id + LIMIT 1`.get<{ branch_id: string; entry_seq: number }>(db); return row === undefined ? undefined : { branchId: row.branch_id, entrySeq: row.entry_seq }; } @@ -137,10 +129,8 @@ export function copyBranchEntriesThroughSeq( sourceBranchId: string, throughSeq: number, ) { - db.prepare( - `INSERT INTO branch_entries (session_id, branch_id, entry_id, entry_seq, entry_type, custom_type) - SELECT session_id, ?, entry_id, entry_seq, entry_type, custom_type - FROM branch_entries - WHERE session_id = ? AND branch_id = ? AND entry_seq <= ?`, - ).run(targetBranchId, sessionId, sourceBranchId, throughSeq); + sql`INSERT INTO branch_entries (session_id, branch_id, entry_id, entry_seq, entry_type, custom_type) + SELECT session_id, ${targetBranchId}, entry_id, entry_seq, entry_type, custom_type + FROM branch_entries + WHERE session_id = ${sessionId} AND branch_id = ${sourceBranchId} AND entry_seq <= ${throughSeq}`.run(db); } diff --git a/packages/session-backends/sqlite-node/src/sqlite/storage/branch-tips.ts b/packages/session-backends/sqlite-node/src/sqlite/storage/branch-tips.ts index e4dffc488..da593e93a 100644 --- a/packages/session-backends/sqlite-node/src/sqlite/storage/branch-tips.ts +++ b/packages/session-backends/sqlite-node/src/sqlite/storage/branch-tips.ts @@ -1,25 +1,21 @@ +import { sql } from "../sql.ts"; import type { SqliteDatabase } from "../types.ts"; export function readBranchTipIds(db: SqliteDatabase, sessionId: string) { - return db - .prepare("SELECT tip_id FROM branch_tips WHERE session_id = ? ORDER BY tip_id") - .all<{ tip_id: string }>(sessionId) + return sql`SELECT tip_id FROM branch_tips WHERE session_id = ${sessionId} ORDER BY tip_id` + .all<{ tip_id: string }>(db) .map((row) => row.tip_id); } export function readBranchTipBranchId(db: SqliteDatabase, sessionId: string, tipId: string) { - const tip = db - .prepare("SELECT branch_id FROM branch_tips WHERE session_id = ? AND tip_id = ?") - .get<{ branch_id: string }>(sessionId, tipId); + const tip = sql`SELECT branch_id FROM branch_tips WHERE session_id = ${sessionId} AND tip_id = ${tipId}`.get<{ + branch_id: string; + }>(db); return tip?.branch_id; } export function insertBranchTip(db: SqliteDatabase, sessionId: string, tipId: string, branchId: string) { - db.prepare("INSERT INTO branch_tips (session_id, tip_id, branch_id) VALUES (?, ?, ?)").run( - sessionId, - tipId, - branchId, - ); + sql`INSERT INTO branch_tips (session_id, tip_id, branch_id) VALUES (${sessionId}, ${tipId}, ${branchId})`.run(db); } export function updateBranchTip( @@ -29,12 +25,11 @@ export function updateBranchTip( oldTipId: string, newTipId: string, ) { - const result = db - .prepare("UPDATE branch_tips SET tip_id = ? WHERE session_id = ? AND branch_id = ? AND tip_id = ?") - .run(newTipId, sessionId, branchId, oldTipId); + const result = sql`UPDATE branch_tips SET tip_id = ${newTipId} + WHERE session_id = ${sessionId} AND branch_id = ${branchId} AND tip_id = ${oldTipId}`.run(db); return result.changes === 1; } export function deleteBranchTips(db: SqliteDatabase, sessionId: string) { - db.prepare("DELETE FROM branch_tips WHERE session_id = ?").run(sessionId); + sql`DELETE FROM branch_tips WHERE session_id = ${sessionId}`.run(db); } diff --git a/packages/session-backends/sqlite-node/src/sqlite/storage/entries.ts b/packages/session-backends/sqlite-node/src/sqlite/storage/entries.ts index 76063c29e..9086f61dd 100644 --- a/packages/session-backends/sqlite-node/src/sqlite/storage/entries.ts +++ b/packages/session-backends/sqlite-node/src/sqlite/storage/entries.ts @@ -1,4 +1,5 @@ import type { Entry, EntryOrder } from "@earendil-works/pi-agent-core"; +import { sql } from "../sql.ts"; import type { SqliteDatabase } from "../types.ts"; export interface EntryRow { @@ -25,22 +26,17 @@ export function entryPayload(entry: Entry): Record { return payload; } -function orderedSql(order: EntryOrder | undefined): string { - return order === "oldestFirst" ? "ASC" : "DESC"; -} - export function insertEntryRow(db: SqliteDatabase, sessionId: string, entry: NewEntryRow) { - db.prepare( - "INSERT INTO entries (session_id, id, seq, parent_id, type, timestamp, payload) VALUES (?, ?, ?, ?, ?, ?, ?)", - ).run(sessionId, entry.id, entry.seq, entry.parentId, entry.type, entry.timestamp, entry.payload); + sql`INSERT INTO entries (session_id, id, seq, parent_id, type, timestamp, payload) + VALUES (${sessionId}, ${entry.id}, ${entry.seq}, ${entry.parentId}, ${entry.type}, ${entry.timestamp}, ${entry.payload})`.run( + db, + ); } export function readEntryRow(db: SqliteDatabase, sessionId: string, entryId: string) { - return db - .prepare( - "SELECT session_id, seq, id, parent_id, type, timestamp, payload FROM entries WHERE session_id = ? AND id = ?", - ) - .get(sessionId, entryId); + return sql`SELECT session_id, seq, id, parent_id, type, timestamp, payload + FROM entries + WHERE session_id = ${sessionId} AND id = ${entryId}`.get(db); } export function readEntryRows( @@ -48,28 +44,20 @@ export function readEntryRows( sessionId: string, options: { afterSeq?: number; order?: EntryOrder } = {}, ) { - const predicates = ["session_id = ?"]; - const params: unknown[] = [sessionId]; - if (options.afterSeq !== undefined) { - predicates.push("seq > ?"); - params.push(options.afterSeq); - } - return db - .prepare( - `SELECT session_id, seq, id, parent_id, type, timestamp, payload - FROM entries - WHERE ${predicates.join(" AND ")} - ORDER BY seq ${orderedSql(options.order)}`, - ) - .all(...params); + const after = options.afterSeq === undefined ? sql`` : sql` AND seq > ${options.afterSeq}`; + const direction = options.order === "oldestFirst" ? sql`ASC` : sql`DESC`; + return sql`SELECT session_id, seq, id, parent_id, type, timestamp, payload + FROM entries + WHERE session_id = ${sessionId}${after} + ORDER BY seq ${direction}`.all(db); } export function idExistsInEntries(db: SqliteDatabase, sessionId: string, id: string) { - return !!db - .prepare("SELECT 1 AS found FROM entries WHERE session_id = ? AND id = ? LIMIT 1") - .get<{ found: number }>(sessionId, id); + return !!sql`SELECT 1 AS found FROM entries WHERE session_id = ${sessionId} AND id = ${id} LIMIT 1`.get<{ + found: number; + }>(db); } export function deleteEntryRows(db: SqliteDatabase, sessionId: string) { - db.prepare("DELETE FROM entries WHERE session_id = ?").run(sessionId); + sql`DELETE FROM entries WHERE session_id = ${sessionId}`.run(db); } diff --git a/packages/session-backends/sqlite-node/src/sqlite/storage/facts.ts b/packages/session-backends/sqlite-node/src/sqlite/storage/facts.ts index 50d056f47..12031cadd 100644 --- a/packages/session-backends/sqlite-node/src/sqlite/storage/facts.ts +++ b/packages/session-backends/sqlite-node/src/sqlite/storage/facts.ts @@ -1,3 +1,4 @@ +import { sql } from "../sql.ts"; import type { SqliteDatabase } from "../types.ts"; export interface FactRow { @@ -16,58 +17,37 @@ export function appendFact( key: string | null, value: string | null, ) { - db.prepare("INSERT INTO facts (session_id, seq, kind, key, value) VALUES (?, ?, ?, ?, ?)").run( - sessionId, - seq, - kind, - key, - value, + sql`INSERT INTO facts (session_id, seq, kind, key, value) VALUES (${sessionId}, ${seq}, ${kind}, ${key}, ${value})`.run( + db, ); } export function readLatestFact(db: SqliteDatabase, sessionId: string, kind: string, key: string | null) { - return db - .prepare( - `SELECT session_id, seq, kind, key, value - FROM facts - WHERE session_id = ? AND kind = ? AND key IS ? - ORDER BY seq DESC - LIMIT 1`, - ) - .get(sessionId, kind, key); + return sql`SELECT session_id, seq, kind, key, value + FROM facts + WHERE session_id = ${sessionId} AND kind = ${kind} AND key IS ${key} + ORDER BY seq DESC + LIMIT 1`.get(db); } export function readLatestLabelFacts(db: SqliteDatabase, sessionId: string) { - return db - .prepare( - `SELECT key, value FROM ( - SELECT key, value, ROW_NUMBER() OVER (PARTITION BY key ORDER BY seq DESC) AS rank - FROM facts - WHERE session_id = ? AND kind = 'label' - ) - WHERE rank = 1 AND value IS NOT NULL - ORDER BY key`, + return sql`SELECT key, value FROM ( + SELECT key, value, ROW_NUMBER() OVER (PARTITION BY key ORDER BY seq DESC) AS rank + FROM facts + WHERE session_id = ${sessionId} AND kind = 'label' ) - .all<{ key: string; value: string }>(sessionId); + WHERE rank = 1 AND value IS NOT NULL + ORDER BY key`.all<{ key: string; value: string }>(db); } export function readFactRows(db: SqliteDatabase, sessionId: string, options: { afterSeq?: number } = {}) { - const predicates = ["session_id = ?"]; - const params: unknown[] = [sessionId]; - if (options.afterSeq !== undefined) { - predicates.push("seq > ?"); - params.push(options.afterSeq); - } - return db - .prepare( - `SELECT session_id, seq, kind, key, value - FROM facts - WHERE ${predicates.join(" AND ")} - ORDER BY seq`, - ) - .all(...params); + const after = options.afterSeq === undefined ? sql`` : sql` AND seq > ${options.afterSeq}`; + return sql`SELECT session_id, seq, kind, key, value + FROM facts + WHERE session_id = ${sessionId}${after} + ORDER BY seq`.all(db); } export function deleteFactRows(db: SqliteDatabase, sessionId: string) { - db.prepare("DELETE FROM facts WHERE session_id = ?").run(sessionId); + sql`DELETE FROM facts WHERE session_id = ${sessionId}`.run(db); } diff --git a/packages/session-backends/sqlite-node/src/sqlite/storage/lanes.ts b/packages/session-backends/sqlite-node/src/sqlite/storage/lanes.ts index d6f54a3a5..fc040b934 100644 --- a/packages/session-backends/sqlite-node/src/sqlite/storage/lanes.ts +++ b/packages/session-backends/sqlite-node/src/sqlite/storage/lanes.ts @@ -1,4 +1,5 @@ import { SessionError } from "@earendil-works/pi-agent-core"; +import { sql } from "../sql.ts"; import type { SqliteDatabase } from "../types.ts"; export interface LaneRow { @@ -16,29 +17,22 @@ export interface LaneMoveRow { } export function createInitialLane(db: SqliteDatabase, sessionId: string, lane = "main", leafId: string | null = null) { - db.prepare("INSERT INTO lanes (session_id, lane, leaf_id, open_operation_id) VALUES (?, ?, ?, NULL)").run( - sessionId, - lane, - leafId, - ); + sql`INSERT INTO lanes (session_id, lane, leaf_id, open_operation_id) + VALUES (${sessionId}, ${lane}, ${leafId}, NULL)`.run(db); } export function readLanes(db: SqliteDatabase, sessionId: string) { - const rows = db - .prepare( - `SELECT - l.session_id, - l.lane, - l.leaf_id, - l.open_operation_id, - (l.leaf_id IS NULL OR EXISTS ( - SELECT 1 FROM entries AS e WHERE e.session_id = l.session_id AND e.id = l.leaf_id - )) AS leaf_exists - FROM lanes AS l - WHERE l.session_id = ? - ORDER BY l.lane`, - ) - .all(sessionId); + const rows = sql`SELECT + l.session_id, + l.lane, + l.leaf_id, + l.open_operation_id, + (l.leaf_id IS NULL OR EXISTS ( + SELECT 1 FROM entries AS e WHERE e.session_id = l.session_id AND e.id = l.leaf_id + )) AS leaf_exists + FROM lanes AS l + WHERE l.session_id = ${sessionId} + ORDER BY l.lane`.all(db); for (const row of rows) { if (row.leaf_exists === 0) { throw new SessionError("storage", `Lane ${row.lane} points at missing entry ${row.leaf_id}`); @@ -53,56 +47,47 @@ export function readLanes(db: SqliteDatabase, sessionId: string) { } export function readLane(db: SqliteDatabase, sessionId: string, lane: string) { - return db - .prepare("SELECT session_id, lane, leaf_id, open_operation_id FROM lanes WHERE session_id = ? AND lane = ?") - .get(sessionId, lane); + return sql`SELECT session_id, lane, leaf_id, open_operation_id + FROM lanes + WHERE session_id = ${sessionId} AND lane = ${lane}`.get(db); } export function readLaneHead(db: SqliteDatabase, sessionId: string, lane: string) { - const row = db - .prepare( - `SELECT - l.leaf_id, - (l.leaf_id IS NULL OR EXISTS ( - SELECT 1 FROM entries AS e WHERE e.session_id = l.session_id AND e.id = l.leaf_id - )) AS leaf_exists - FROM lanes AS l - WHERE l.session_id = ? AND l.lane = ?`, - ) - .get<{ leaf_id: string | null; leaf_exists: number }>(sessionId, lane); + const row = sql`SELECT + l.leaf_id, + (l.leaf_id IS NULL OR EXISTS ( + SELECT 1 FROM entries AS e WHERE e.session_id = l.session_id AND e.id = l.leaf_id + )) AS leaf_exists + FROM lanes AS l + WHERE l.session_id = ${sessionId} AND l.lane = ${lane}`.get<{ + leaf_id: string | null; + leaf_exists: number; + }>(db); if (!row) throw new SessionError("invalid_lane", `Lane not found: ${lane}`); if (row.leaf_exists === 0) throw new SessionError("storage", `Entry ${row.leaf_id} not found`); return { leafId: row.leaf_id }; } export function createLane(db: SqliteDatabase, sessionId: string, seq: number, lane: string, leafId: string | null) { - db.prepare("INSERT INTO lanes (session_id, lane, leaf_id, open_operation_id) VALUES (?, ?, ?, NULL)").run( - sessionId, - lane, - leafId, - ); + sql`INSERT INTO lanes (session_id, lane, leaf_id, open_operation_id) + VALUES (${sessionId}, ${lane}, ${leafId}, NULL)`.run(db); appendLaneMove(db, sessionId, seq, lane, leafId); } export function moveLane(db: SqliteDatabase, sessionId: string, seq: number, lane: string, leafId: string | null) { - const result = db - .prepare("UPDATE lanes SET leaf_id = ? WHERE session_id = ? AND lane = ?") - .run(leafId, sessionId, lane); + const result = sql`UPDATE lanes SET leaf_id = ${leafId} WHERE session_id = ${sessionId} AND lane = ${lane}`.run(db); if (result.changes !== 1) throw new SessionError("invalid_lane", `Lane not found: ${lane}`); appendLaneMove(db, sessionId, seq, lane, leafId); } export function setLaneLeaf(db: SqliteDatabase, sessionId: string, lane: string, leafId: string | null) { - const result = db - .prepare("UPDATE lanes SET leaf_id = ? WHERE session_id = ? AND lane = ?") - .run(leafId, sessionId, lane); + const result = sql`UPDATE lanes SET leaf_id = ${leafId} WHERE session_id = ${sessionId} AND lane = ${lane}`.run(db); if (result.changes !== 1) throw new SessionError("invalid_lane", `Lane not found: ${lane}`); } export function startLaneOperation(db: SqliteDatabase, sessionId: string, lane: string, runId: string) { - const result = db - .prepare("UPDATE lanes SET open_operation_id = ? WHERE session_id = ? AND lane = ? AND open_operation_id IS NULL") - .run(runId, sessionId, lane); + const result = sql`UPDATE lanes SET open_operation_id = ${runId} + WHERE session_id = ${sessionId} AND lane = ${lane} AND open_operation_id IS NULL`.run(db); if (result.changes === 1) return; const current = readLane(db, sessionId, lane); if (!current) throw new SessionError("invalid_lane", `Lane not found: ${lane}`); @@ -110,38 +95,25 @@ export function startLaneOperation(db: SqliteDatabase, sessionId: string, lane: } export function finishLaneOperation(db: SqliteDatabase, sessionId: string, lane: string, runId: string) { - db.prepare( - "UPDATE lanes SET open_operation_id = NULL WHERE session_id = ? AND lane = ? AND open_operation_id = ?", - ).run(sessionId, lane, runId); + sql`UPDATE lanes SET open_operation_id = NULL + WHERE session_id = ${sessionId} AND lane = ${lane} AND open_operation_id = ${runId}`.run(db); } export function readLaneMoveRows(db: SqliteDatabase, sessionId: string, options: { afterSeq?: number } = {}) { - const predicates = ["session_id = ?"]; - const params: unknown[] = [sessionId]; - if (options.afterSeq !== undefined) { - predicates.push("seq > ?"); - params.push(options.afterSeq); - } - return db - .prepare( - `SELECT session_id, seq, lane, leaf_id - FROM lane_moves - WHERE ${predicates.join(" AND ")} - ORDER BY seq`, - ) - .all(...params); + const after = options.afterSeq === undefined ? sql`` : sql` AND seq > ${options.afterSeq}`; + return sql`SELECT session_id, seq, lane, leaf_id + FROM lane_moves + WHERE session_id = ${sessionId}${after} + ORDER BY seq`.all(db); } export function deleteLaneRows(db: SqliteDatabase, sessionId: string) { - db.prepare("DELETE FROM lane_moves WHERE session_id = ?").run(sessionId); - db.prepare("DELETE FROM lanes WHERE session_id = ?").run(sessionId); + sql`DELETE FROM lane_moves WHERE session_id = ${sessionId}`.run(db); + sql`DELETE FROM lanes WHERE session_id = ${sessionId}`.run(db); } function appendLaneMove(db: SqliteDatabase, sessionId: string, seq: number, lane: string, leafId: string | null) { - db.prepare("INSERT INTO lane_moves (session_id, seq, lane, leaf_id) VALUES (?, ?, ?, ?)").run( - sessionId, - seq, - lane, - leafId, + sql`INSERT INTO lane_moves (session_id, seq, lane, leaf_id) VALUES (${sessionId}, ${seq}, ${lane}, ${leafId})`.run( + db, ); } diff --git a/packages/session-backends/sqlite-node/src/sqlite/storage/records.ts b/packages/session-backends/sqlite-node/src/sqlite/storage/records.ts index c68276877..1dc357227 100644 --- a/packages/session-backends/sqlite-node/src/sqlite/storage/records.ts +++ b/packages/session-backends/sqlite-node/src/sqlite/storage/records.ts @@ -1,4 +1,5 @@ import { SessionError } from "@earendil-works/pi-agent-core"; +import { joinSqlFragments, sql } from "../sql.ts"; import type { SqliteDatabase } from "../types.ts"; export interface RecordRow { @@ -25,31 +26,21 @@ export interface NewRecordRow { } export function appendRecordRow(db: SqliteDatabase, sessionId: string, record: NewRecordRow) { - db.prepare( - `INSERT INTO records + sql`INSERT INTO records (session_id, seq, id, lane, run_id, type, op_kind, timestamp, payload) - VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)`, - ).run( - sessionId, - record.seq, - record.id, - record.lane, - record.runId ?? null, - record.type, - record.opKind ?? null, - record.timestamp, - record.payload, + VALUES (${sessionId}, ${record.seq}, ${record.id}, ${record.lane}, ${record.runId ?? null}, ${record.type}, ${record.opKind ?? null}, ${record.timestamp}, ${record.payload})`.run( + db, ); } export function idExistsInRecords(db: SqliteDatabase, sessionId: string, id: string) { - return !!db - .prepare("SELECT 1 AS found FROM records WHERE session_id = ? AND id = ? LIMIT 1") - .get<{ found: number }>(sessionId, id); + return !!sql`SELECT 1 AS found FROM records WHERE session_id = ${sessionId} AND id = ${id} LIMIT 1`.get<{ + found: number; + }>(db); } export function deleteRecordRows(db: SqliteDatabase, sessionId: string) { - db.prepare("DELETE FROM records WHERE session_id = ?").run(sessionId); + sql`DELETE FROM records WHERE session_id = ${sessionId}`.run(db); } export function readRecordRows( @@ -65,39 +56,18 @@ export function readRecordRows( limit?: number; } = {}, ) { - const predicates = ["session_id = ?"]; - const params: unknown[] = [sessionId]; - if (query.lane !== undefined) { - predicates.push("lane = ?"); - params.push(query.lane); - } - if (query.type !== undefined) { - predicates.push("type = ?"); - params.push(query.type); - } - if (query.runId !== undefined) { - predicates.push("run_id = ?"); - params.push(query.runId); - } - if (query.operationKind !== undefined) { - predicates.push("op_kind = ?"); - params.push(query.operationKind); - } - if (query.afterSeq !== undefined) { - predicates.push("seq > ?"); - params.push(query.afterSeq); - } - const limit = query.limit === undefined ? "" : " LIMIT ?"; - if (query.limit !== undefined) params.push(query.limit); - const direction = query.order === "oldestFirst" ? "ASC" : "DESC"; - return db - .prepare( - `SELECT session_id, seq, id, lane, run_id, type, op_kind, timestamp, payload - FROM records - WHERE ${predicates.join(" AND ")} - ORDER BY seq ${direction}${limit}`, - ) - .all(...params); + const predicates = [sql`session_id = ${sessionId}`]; + if (query.lane !== undefined) predicates.push(sql`lane = ${query.lane}`); + if (query.type !== undefined) predicates.push(sql`type = ${query.type}`); + if (query.runId !== undefined) predicates.push(sql`run_id = ${query.runId}`); + if (query.operationKind !== undefined) predicates.push(sql`op_kind = ${query.operationKind}`); + if (query.afterSeq !== undefined) predicates.push(sql`seq > ${query.afterSeq}`); + const direction = query.order === "oldestFirst" ? sql`ASC` : sql`DESC`; + const limit = query.limit === undefined ? sql`` : sql` LIMIT ${query.limit}`; + return sql`SELECT session_id, seq, id, lane, run_id, type, op_kind, timestamp, payload + FROM records + WHERE ${joinSqlFragments(predicates, " AND ")} + ORDER BY seq ${direction}${limit}`.all(db); } export function readOpenOperationRows( @@ -106,19 +76,15 @@ export function readOpenOperationRows( lane: string, _options: { limit?: number } = {}, ): RecordRow[] { - const laneRow = db - .prepare("SELECT open_operation_id FROM lanes WHERE session_id = ? AND lane = ?") - .get<{ open_operation_id: string | null }>(sessionId, lane); + const laneRow = sql`SELECT open_operation_id FROM lanes WHERE session_id = ${sessionId} AND lane = ${lane}`.get<{ + open_operation_id: string | null; + }>(db); if (!laneRow?.open_operation_id) return []; - const record = db - .prepare( - `SELECT session_id, seq, id, lane, run_id, type, op_kind, timestamp, payload - FROM records - WHERE session_id = ? - AND id = ?`, - ) - .get(sessionId, laneRow.open_operation_id); + const record = sql`SELECT session_id, seq, id, lane, run_id, type, op_kind, timestamp, payload + FROM records + WHERE session_id = ${sessionId} + AND id = ${laneRow.open_operation_id}`.get(db); if (!record) { throw new SessionError("storage", `Lane ${lane} points at missing open operation ${laneRow.open_operation_id}`); } diff --git a/packages/session-backends/sqlite-node/src/sqlite/storage/session-sequences.ts b/packages/session-backends/sqlite-node/src/sqlite/storage/session-sequences.ts index 578005c4d..16f7df44c 100644 --- a/packages/session-backends/sqlite-node/src/sqlite/storage/session-sequences.ts +++ b/packages/session-backends/sqlite-node/src/sqlite/storage/session-sequences.ts @@ -1,14 +1,15 @@ import { SessionError } from "@earendil-works/pi-agent-core"; +import { sql } from "../sql.ts"; import type { SqliteDatabase } from "../types.ts"; export function createSequence(db: SqliteDatabase, sessionId: string, nextSeq = 1) { - db.prepare("INSERT INTO session_sequences (session_id, next_seq) VALUES (?, ?)").run(sessionId, nextSeq); + sql`INSERT INTO session_sequences (session_id, next_seq) VALUES (${sessionId}, ${nextSeq})`.run(db); } export function getNextSequence(db: SqliteDatabase, sessionId: string) { - const sequenceRow = db - .prepare("SELECT next_seq FROM session_sequences WHERE session_id = ?") - .get<{ next_seq: number }>(sessionId); + const sequenceRow = sql`SELECT next_seq FROM session_sequences WHERE session_id = ${sessionId}`.get<{ + next_seq: number; + }>(db); if (!sequenceRow) { throw new SessionError("storage", `Missing sequence row for session ${sessionId}`); } @@ -16,7 +17,7 @@ export function getNextSequence(db: SqliteDatabase, sessionId: string) { } export function setNextSequence(db: SqliteDatabase, sessionId: string, nextSeq: number) { - db.prepare("UPDATE session_sequences SET next_seq = ? WHERE session_id = ?").run(nextSeq, sessionId); + sql`UPDATE session_sequences SET next_seq = ${nextSeq} WHERE session_id = ${sessionId}`.run(db); } export function advanceSequence(db: SqliteDatabase, sessionId: string, seq: number) { @@ -24,5 +25,5 @@ export function advanceSequence(db: SqliteDatabase, sessionId: string, seq: numb } export function deleteSequence(db: SqliteDatabase, sessionId: string) { - db.prepare("DELETE FROM session_sequences WHERE session_id = ?").run(sessionId); + sql`DELETE FROM session_sequences WHERE session_id = ${sessionId}`.run(db); } diff --git a/packages/session-backends/sqlite-node/src/sqlite/storage/session-stats.ts b/packages/session-backends/sqlite-node/src/sqlite/storage/session-stats.ts index 370cfb947..85a3a4581 100644 --- a/packages/session-backends/sqlite-node/src/sqlite/storage/session-stats.ts +++ b/packages/session-backends/sqlite-node/src/sqlite/storage/session-stats.ts @@ -1,5 +1,6 @@ import { SessionError, type SessionStats } from "@earendil-works/pi-agent-core"; import type { Usage } from "@earendil-works/pi-ai"; +import { sql } from "../sql.ts"; import type { SqliteDatabase } from "../types.ts"; export interface SessionStatsRow { @@ -12,21 +13,15 @@ export interface SessionStatsRow { } export function createStats(db: SqliteDatabase, sessionId: string, messageCount = 0): void { - db.prepare( - `INSERT INTO session_stats + sql`INSERT INTO session_stats (session_id, message_count, cached_tokens, uncached_tokens, total_tokens, cost_total) - VALUES (?, ?, 0, 0, 0, 0)`, - ).run(sessionId, messageCount); + VALUES (${sessionId}, ${messageCount}, 0, 0, 0, 0)`.run(db); } export function readStats(db: SqliteDatabase, sessionId: string): SessionStats { - const row = db - .prepare( - `SELECT session_id, message_count, cached_tokens, uncached_tokens, total_tokens, cost_total - FROM session_stats - WHERE session_id = ?`, - ) - .get(sessionId); + const row = sql`SELECT session_id, message_count, cached_tokens, uncached_tokens, total_tokens, cost_total + FROM session_stats + WHERE session_id = ${sessionId}`.get(db); if (!row) throw new SessionError("storage", `Missing stats row for session ${sessionId}`); return { messageCount: row.message_count, @@ -38,26 +33,22 @@ export function readStats(db: SqliteDatabase, sessionId: string): SessionStats { } export function incrementMessageCount(db: SqliteDatabase, sessionId: string): void { - const result = db - .prepare("UPDATE session_stats SET message_count = message_count + 1 WHERE session_id = ?") - .run(sessionId); + const result = sql`UPDATE session_stats SET message_count = message_count + 1 WHERE session_id = ${sessionId}`.run( + db, + ); if (result.changes !== 1) throw new SessionError("storage", `Missing stats row for session ${sessionId}`); } export function addUsageToStats(db: SqliteDatabase, sessionId: string, usage: Usage): void { - const result = db - .prepare( - `UPDATE session_stats - SET cached_tokens = cached_tokens + ?, - uncached_tokens = uncached_tokens + ?, - total_tokens = total_tokens + ?, - cost_total = cost_total + ? - WHERE session_id = ?`, - ) - .run(usage.cacheRead, usage.input + usage.cacheWrite, usage.totalTokens, usage.cost.total, sessionId); + const result = sql`UPDATE session_stats + SET cached_tokens = cached_tokens + ${usage.cacheRead}, + uncached_tokens = uncached_tokens + ${usage.input + usage.cacheWrite}, + total_tokens = total_tokens + ${usage.totalTokens}, + cost_total = cost_total + ${usage.cost.total} + WHERE session_id = ${sessionId}`.run(db); if (result.changes !== 1) throw new SessionError("storage", `Missing stats row for session ${sessionId}`); } export function deleteStats(db: SqliteDatabase, sessionId: string): void { - db.prepare("DELETE FROM session_stats WHERE session_id = ?").run(sessionId); + sql`DELETE FROM session_stats WHERE session_id = ${sessionId}`.run(db); } diff --git a/packages/session-backends/sqlite-node/src/sqlite/storage/sessions.ts b/packages/session-backends/sqlite-node/src/sqlite/storage/sessions.ts index d963c5c89..e25e8a809 100644 --- a/packages/session-backends/sqlite-node/src/sqlite/storage/sessions.ts +++ b/packages/session-backends/sqlite-node/src/sqlite/storage/sessions.ts @@ -1,4 +1,5 @@ import { assertJsonSerializable, SessionError } from "@earendil-works/pi-agent-core"; +import { sql } from "../sql.ts"; import type { SqliteDatabase, SqliteSessionMetadata } from "../types.ts"; export interface SessionRow { @@ -38,7 +39,7 @@ function parseMetadata(metadata: string | null, sessionId: string): Record(sessionId); + return !!sql`SELECT 1 AS found FROM sessions WHERE id = ${sessionId}`.get<{ found: number }>(db); } function serializeMetadata(metadata: Record | undefined): string | null { @@ -51,61 +52,50 @@ function serializeMetadata(metadata: Record | undefined): strin } export function insertSessionRow(db: SqliteDatabase, session: NewSessionRow) { - db.prepare("INSERT INTO sessions (id, created_at, metadata, cwd, parent_session_id) VALUES (?, ?, ?, ?, ?)").run( - session.id, - session.createdAt, - serializeMetadata(session.metadata), - session.cwd, - session.parentSessionId ?? null, + sql`INSERT INTO sessions (id, created_at, metadata, cwd, parent_session_id) + VALUES (${session.id}, ${session.createdAt}, ${serializeMetadata(session.metadata)}, ${session.cwd}, ${session.parentSessionId ?? null})`.run( + db, ); } export function readSessionRow(db: SqliteDatabase, sessionId: string) { - return db - .prepare( - `SELECT s.id, s.created_at, s.metadata, s.cwd, s.parent_session_id, - name_fact.seq IS NOT NULL AS has_session_name, - name_fact.value AS session_name - FROM sessions AS s - LEFT JOIN facts AS name_fact - ON name_fact.session_id = s.id - AND name_fact.kind = 'name' - AND name_fact.key IS NULL - AND name_fact.seq = ( - SELECT MAX(f.seq) - FROM facts AS f - WHERE f.session_id = s.id AND f.kind = 'name' AND f.key IS NULL - ) - WHERE s.id = ?`, - ) - .get(sessionId); + return sql`SELECT s.id, s.created_at, s.metadata, s.cwd, s.parent_session_id, + name_fact.seq IS NOT NULL AS has_session_name, + name_fact.value AS session_name + FROM sessions AS s + LEFT JOIN facts AS name_fact + ON name_fact.session_id = s.id + AND name_fact.kind = 'name' + AND name_fact.key IS NULL + AND name_fact.seq = ( + SELECT MAX(f.seq) + FROM facts AS f + WHERE f.session_id = s.id AND f.kind = 'name' AND f.key IS NULL + ) + WHERE s.id = ${sessionId}`.get(db); } export function readSessionRows(db: SqliteDatabase, options: { cwd?: string } = {}) { - const where = options.cwd === undefined ? "" : "WHERE s.cwd = ?"; - return db - .prepare( - `SELECT s.id, s.created_at, s.metadata, s.cwd, s.parent_session_id, - name_fact.seq IS NOT NULL AS has_session_name, - name_fact.value AS session_name - FROM sessions AS s - LEFT JOIN facts AS name_fact - ON name_fact.session_id = s.id - AND name_fact.kind = 'name' - AND name_fact.key IS NULL - AND name_fact.seq = ( - SELECT MAX(f.seq) - FROM facts AS f - WHERE f.session_id = s.id AND f.kind = 'name' AND f.key IS NULL - ) - ${where} - ORDER BY s.created_at DESC`, - ) - .all(...(options.cwd === undefined ? [] : [options.cwd])); + const where = options.cwd === undefined ? sql`` : sql`WHERE s.cwd = ${options.cwd}`; + return sql`SELECT s.id, s.created_at, s.metadata, s.cwd, s.parent_session_id, + name_fact.seq IS NOT NULL AS has_session_name, + name_fact.value AS session_name + FROM sessions AS s + LEFT JOIN facts AS name_fact + ON name_fact.session_id = s.id + AND name_fact.kind = 'name' + AND name_fact.key IS NULL + AND name_fact.seq = ( + SELECT MAX(f.seq) + FROM facts AS f + WHERE f.session_id = s.id AND f.kind = 'name' AND f.key IS NULL + ) + ${where} + ORDER BY s.created_at DESC`.all(db); } export function deleteSessionRow(db: SqliteDatabase, sessionId: string) { - db.prepare("DELETE FROM sessions WHERE id = ?").run(sessionId); + sql`DELETE FROM sessions WHERE id = ${sessionId}`.run(db); } function parseSessionName(value: string | null, sessionId: string): string { diff --git a/packages/session-backends/sqlite-node/src/sqlite/storage/writer-leases.ts b/packages/session-backends/sqlite-node/src/sqlite/storage/writer-leases.ts index 02cf38ec0..ec7fcaa91 100644 --- a/packages/session-backends/sqlite-node/src/sqlite/storage/writer-leases.ts +++ b/packages/session-backends/sqlite-node/src/sqlite/storage/writer-leases.ts @@ -1,3 +1,4 @@ +import { sql } from "../sql.ts"; import type { SqliteDatabase } from "../types.ts"; export interface WriterLease { @@ -19,18 +20,14 @@ export function acquireWriterLease( now: number, expiresAtMs: number, ) { - const row = db - .prepare( - `INSERT INTO writer_leases (session_id, owner_id, fence, expires_at_ms) - VALUES (?, ?, 1, ?) - ON CONFLICT(session_id) DO UPDATE SET - owner_id = excluded.owner_id, - fence = writer_leases.fence + 1, - expires_at_ms = excluded.expires_at_ms - WHERE writer_leases.expires_at_ms <= ? - RETURNING owner_id, fence, expires_at_ms`, - ) - .get(sessionId, ownerId, expiresAtMs, now); + const row = sql`INSERT INTO writer_leases (session_id, owner_id, fence, expires_at_ms) + VALUES (${sessionId}, ${ownerId}, 1, ${expiresAtMs}) + ON CONFLICT(session_id) DO UPDATE SET + owner_id = excluded.owner_id, + fence = writer_leases.fence + 1, + expires_at_ms = excluded.expires_at_ms + WHERE writer_leases.expires_at_ms <= ${now} + RETURNING owner_id, fence, expires_at_ms`.get(db); return row === undefined ? undefined : { ownerId: row.owner_id, fence: row.fence, expiresAtMs: row.expires_at_ms }; } @@ -41,25 +38,21 @@ export function renewWriterLease( now: number, expiresAtMs: number, ) { - const result = db - .prepare( - `UPDATE writer_leases - SET expires_at_ms = ? - WHERE session_id = ? AND owner_id = ? AND fence = ? AND expires_at_ms > ?`, - ) - .run(expiresAtMs, sessionId, lease.ownerId, lease.fence, now); + const result = sql`UPDATE writer_leases + SET expires_at_ms = ${expiresAtMs} + WHERE session_id = ${sessionId} + AND owner_id = ${lease.ownerId} + AND fence = ${lease.fence} + AND expires_at_ms > ${now}`.run(db); if (result.changes === 1) lease.expiresAtMs = expiresAtMs; return result.changes === 1; } export function releaseWriterLease(db: SqliteDatabase, sessionId: string, lease: WriterLease) { - db.prepare("DELETE FROM writer_leases WHERE session_id = ? AND owner_id = ? AND fence = ?").run( - sessionId, - lease.ownerId, - lease.fence, - ); + sql`DELETE FROM writer_leases + WHERE session_id = ${sessionId} AND owner_id = ${lease.ownerId} AND fence = ${lease.fence}`.run(db); } export function deleteWriterLease(db: SqliteDatabase, sessionId: string) { - db.prepare("DELETE FROM writer_leases WHERE session_id = ?").run(sessionId); + sql`DELETE FROM writer_leases WHERE session_id = ${sessionId}`.run(db); } diff --git a/packages/session-backends/sqlite-node/test/sql.test.ts b/packages/session-backends/sqlite-node/test/sql.test.ts new file mode 100644 index 000000000..020d1c463 --- /dev/null +++ b/packages/session-backends/sqlite-node/test/sql.test.ts @@ -0,0 +1,39 @@ +import { describe, expect, it } from "vitest"; +import { createNodeSqliteFactory, joinSqlFragments, sql } from "../src/index.ts"; + +describe("sql", () => { + it("composes SQLite queries without renumbering parameters", async () => { + const db = await createNodeSqliteFactory().open(":memory:"); + try { + sql`CREATE TABLE entries (id TEXT PRIMARY KEY, kind TEXT NOT NULL, active INTEGER NOT NULL)`.exec(db); + sql`INSERT INTO entries (id, kind, active) VALUES (${"one"}, ${"message"}, ${1})`.run(db); + sql`INSERT INTO entries (id, kind, active) VALUES (${"two"}, ${"message"}, ${0})`.run(db); + const filters = joinSqlFragments([sql`kind = ${"message"}`, sql`active = ${1}`], " AND "); + + expect(sql`SELECT id FROM entries WHERE ${filters} LIMIT ${10}`.all<{ id: string }>(db)).toEqual([ + { id: "one" }, + ]); + } finally { + db.close(); + } + }); + + it("executes parameterized queries", async () => { + const db = await createNodeSqliteFactory().open(":memory:"); + try { + sql`CREATE TABLE values_table (id INTEGER PRIMARY KEY, value TEXT NOT NULL)`.exec(db); + sql`INSERT INTO values_table (id, value) VALUES (${1}, ${"one"})`.run(db); + sql`INSERT INTO values_table (id, value) VALUES (${2}, ${"two"})`.run(db); + + expect(sql`SELECT value FROM values_table WHERE id = ${1}`.get<{ value: string }>(db)).toEqual({ + value: "one", + }); + expect(sql`SELECT value FROM values_table ORDER BY id`.all<{ value: string }>(db)).toEqual([ + { value: "one" }, + { value: "two" }, + ]); + } finally { + db.close(); + } + }); +});