/** * SQLite implementation of StorePort. * Uses bun:sqlite for database operations. * * Note: bun:sqlite is synchronous but we use async for interface consistency. * * @module src/store/sqlite/adapter */ // CRITICAL: Import setup FIRST to configure custom SQLite before any Database use import "./setup"; import { Database } from "bun:sqlite"; // node:async_hooks binds nested transactions to one async request; Bun has no separate native equivalent. import { AsyncLocalStorage } from "node:async_hooks"; // node:path basename: no Bun path utilities. import { basename } from "node:path"; import type { EgressLineage } from "../../core/egress-provenance"; import type { FileRefactorJournalAdvance, FileRefactorRecoveryReceipt, FileRefactorRecoveryReceiptDraft, } from "../../core/file-refactor-journal"; import type { ActivationIndexDocument, ActivationIndexIdentity, ActivationIndexSnapshot, ActivationVerificationReceipt, BacklinkRow, ChunkInput, ChunkRow, CleanupStats, CollectionRow, ContextRow, DocEdgeConfidence, DocEdgeInput, DocEdgeRow, DocEdgeSource, DocEdgeType, DocLinkInput, DocLinkRow, DocLinkSource, DocumentChangeKind, DocumentChangeListOptions, DocumentChangePage, DocumentChangePurgeResult, DocumentChangeRetentionPolicy, DocumentChangeRetentionResult, DocumentInput, DocumentRow, EmbeddingCleanupStats, EgressAuditCursor, EgressAuditDeleteResult, EgressAuditPage, EgressAuditPurgeResult, EgressAuditReceiptInput, EgressAuditReceiptRow, EgressAuditRetentionPolicy, EgressAuditRetentionResult, EgressAuditStatusResult, FtsResult, FtsSearchOptions, FileRefactorResolutionReferrerDocument, FileRefactorResolutionSnapshot, GetGraphNeighborsOptions, GetGraphOptions, GraphEdgeConfidence, GraphEdgeAudit, GraphLinkType, GraphNeighborsResult, GraphQueryOptions, GraphQueryTraversalRows, GraphReportNode, GraphResult, IndexStatus, IngestErrorInput, IngestErrorRow, MemoryEligibleDocument, MemoryEligibleDocumentsOptions, MigrationResult, RetrievalTraceAppendResult, RetrievalTraceBundle, RetrievalTraceCursor, RetrievalTraceBoundedBundle, RetrievalTraceDeleteCounts, RetrievalTraceEventInput, RetrievalTraceExportInput, RetrievalTraceExportBundle, RetrievalTraceExportManifestInput, RetrievalTraceExportManifestRow, RetrievalTraceInput, RetrievalTraceJudgmentInput, RetrievalTracePurgeResult, RetrievalTraceRetentionPolicy, RetrievalTraceRetentionResult, RetrievalTraceRow, RetrievalTraceRunInput, RetrievalTraceTerminalStatus, RenameDocumentOptions, SavedCapsuleRegistrationInput, SavedCapsuleRegistrationRecord, SavedCapsuleRegistrationSnapshot, SavedCapsuleReverificationState, SavedCapsuleVerificationExpectation, SavedCapsuleVerificationRecord, StorePort, StoreResult, TagCount, TagRow, TagSource, UpsertDocumentResult, } from "../types"; import type { SqliteDbProvider } from "./types"; import { buildUri, deriveDocid, stripUriIndex } from "../../app/constants"; import { DEFAULT_CHUNKING_PARAMS, resolveChunkingParams, type ChunkingParams, } from "../../config/chunking"; import { type Collection, type Context, DEFAULT_BUSY_TIMEOUT_MS, type EgressPolicy, type EgressPolicySource, type FtsTokenizer, MAX_BUSY_TIMEOUT_MS, MIN_BUSY_TIMEOUT_MS, resolveConfiguredEgressPolicy, } from "../../config/types"; import { getDocumentCapabilities, isTextLikeReferenceDocument, } from "../../core/document-capabilities"; import { analyzeGraphCommunities } from "../../core/graph-analysis"; import { classifyResolvedGraphEdge, mergeGraphEdgeAudit, } from "../../core/graph-edge-confidence"; import { buildWikiBestMatchSubquery } from "../../core/graph-resolver"; import { buildContentPrefilterNeedles } from "../../core/link-relevance"; import { normalizeWikiName, stripWikiMdExt } from "../../core/links"; import { TYPED_METADATA_INGEST_VERSION, typedMetadataSchema, } from "../../core/typed-metadata"; import { parseActivationReceipt, serializeActivationReceipt, } from "../activation-receipts"; import { ChunkingPolicyConflictError, type ChunkingPolicyToken, type PendingChunkingMirror, } from "../chunking"; import { getSchemaVersion, migrations, runMigrations } from "../migrations"; import { err, ok } from "../types"; import { getStoredEmbeddingFingerprint } from "../vector/freshness"; import { modelTableName } from "../vector/sqlite-vec"; import { getVariantStatus } from "../vector/status"; import { deleteSavedCapsuleRegistration as deleteStoredSavedCapsuleRegistration, getSavedCapsuleRegistration as getStoredSavedCapsuleRegistration, getSavedCapsuleRegistrationSnapshot as getStoredSavedCapsuleRegistrationSnapshot, getSavedCapsuleReverificationState as getStoredSavedCapsuleReverificationState, getSavedCapsuleReverificationSequence as getStoredSavedCapsuleReverificationSequence, listSavedCapsuleIdsAffectedByChanges as listStoredSavedCapsuleIdsAffectedByChanges, listSavedCapsuleRegistrations as listStoredSavedCapsuleRegistrations, setSavedCapsuleReverificationSequence as setStoredSavedCapsuleReverificationSequence, upsertSavedCapsuleRegistration as upsertStoredSavedCapsuleRegistration, upsertSavedCapsuleVerification as upsertStoredSavedCapsuleVerification, } from "./capsule-registry-store"; import { appendDocumentChange as appendStoredDocumentChange, enforceDocumentChangeRetention as enforceStoredDocumentChangeRetention, listDocumentChanges as listStoredDocumentChanges, purgeDocumentChanges as purgeStoredDocumentChanges, snapshotDocumentChange, } from "./change-journal-store"; import { assertChunkingTarget, claimChunkingTarget, getChunkingStatus, markChunkingApplied, pendingChunkingMirrors, pruneChunkingMetadata, readChunkingTarget, } from "./chunking-policy"; import { appendEgressAuditReceipt as appendStoredEgressAuditReceipt, appendEgressAuditReceiptWithRetention as appendStoredEgressAuditReceiptWithRetention, deleteEgressAuditReceipt as deleteStoredEgressAuditReceipt, enforceEgressAuditRetention as enforceStoredEgressAuditRetention, getEgressAuditReceipt as getStoredEgressAuditReceipt, getEgressAuditStatus as getStoredEgressAuditStatus, listEgressAuditReceipts as listStoredEgressAuditReceipts, purgeEgressAuditReceipts as purgeStoredEgressAuditReceipts, } from "./egress-audit-store"; import { buildEligibleDocumentQuery } from "./eligibility"; import { advanceFileRefactorReceipt as advanceStoredFileRefactorReceipt, createFileRefactorPreparedReceipt as createStoredFileRefactorPreparedReceipt, getFileRefactorReceiptById as getStoredFileRefactorReceiptById, getLatestFileRefactorReceiptByPlanDigest as getStoredLatestFileRefactorReceiptByPlanDigest, } from "./file-refactor-journal-store"; import { loadFts5Snowball } from "./fts5-snowball"; import { applyGraphEdges, type DesiredGraphEdge, } from "./graph-edge-application"; import { resolveGraphLinkTargets } from "./graph-link-resolver"; import { queryGraphNeighborsForSeeds } from "./graph-neighbors"; import { createGraphReferenceStore } from "./graph-reference-state"; import { snapshotLegacyTitles, reconcileLegacyTitles, } from "./legacy-vector-ownership"; import { appendExportManifest as appendStoredTraceExportManifest, getBoundedTrace as getBoundedStoredTrace, getExportBundle as getStoredTraceExportBundle, getExportManifest as getStoredTraceExportManifest, getOrCreateRedactionSecret as getOrCreateStoredTraceRedactionSecret, } from "./retrieval-trace-management-store"; import { deleteTrace as deleteStoredTrace, enforceRetention as enforceStoredTraceRetention, purgeTraces as purgeStoredTraces, } from "./retrieval-trace-retention"; import { appendEvent as appendStoredTraceEvent, appendExport as appendStoredTraceExport, appendJudgment as appendStoredTraceJudgment, appendRun as appendStoredTraceRun, createTrace as createStoredTrace, finalizeTrace as finalizeStoredTrace, getTrace as getStoredTrace, listTraces as listStoredTraces, mergeTraceEgressLineage as mergeStoredTraceEgressLineage, } from "./retrieval-trace-store"; // ───────────────────────────────────────────────────────────────────────────── // FTS5 Query Escaping // ───────────────────────────────────────────────────────────────────────────── /** Whitespace regex for splitting FTS5 tokens */ const WHITESPACE_REGEX = /\s+/; const SINGLE_LINE_QUERY_PATTERN = /[\r\n]/; const DOUBLE_QUOTE_PATTERN = /"/g; const DOC_EDGE_TYPE_PATTERN = /^[a-z][a-z0-9_]*$/; const SQLITE_SAFE_PARAMETER_BATCH_SIZE = 900; /** * Effective physical source path for watcher fallback queries. * Record-container logical rows resolve to their source container path. */ const WATCHER_SOURCE_PATH_SQL = "COALESCE(NULLIF(record_source_path, ''), rel_path)"; /** * Parent directory of the effective source path (POSIX), with the collection * root represented as the empty string. */ const WATCHER_SOURCE_PARENT_SQL = `CASE WHEN instr(${WATCHER_SOURCE_PATH_SQL}, '/') = 0 THEN '' ELSE substr(${WATCHER_SOURCE_PATH_SQL}, 1, length(rtrim(${WATCHER_SOURCE_PATH_SQL}, replace(${WATCHER_SOURCE_PATH_SQL}, '/', ''))) - 1) END`; /** * Normalize a collection-relative directory argument for watcher source-path * queries. Returns null for absolute, drive-shaped, or escaping paths. */ function normalizeWatcherSourceDirRelPath(dirRelPath: string): string | null { const normalized = dirRelPath.replaceAll("\\", "/"); if (normalized.startsWith("/")) { return null; } const segments: string[] = []; for (const segment of normalized.split("/")) { if (segment === "" || segment === ".") { continue; } if (segment === "..") { return null; } segments.push(segment); } const canonical = segments.join("/"); // `C:` / `C:/foo` after stripping `.` is a Windows absolute escape. // A single segment like `a:notes` (no slash after the colon) stays legal. if ( /^[A-Za-z]:(\/|$)/.test(canonical) && (canonical.length === 2 || canonical[2] === "/") ) { return null; } return canonical; } const FTS5_FIELD_WEIGHTS = { filepath: 1.5, title: 4.0, body: 1.0, } as const; const uniqueNonEmptyValues = (values: readonly string[]): string[] => [ ...new Set(values.filter((value) => value.trim().length > 0)), ]; /** Content-free activation snapshot query; kept exported for contract tests. */ export const ACTIVATION_INDEX_SNAPSHOT_SQL = `SELECT d.id, d.uri, d.source_hash, d.mirror_hash, d.active, CASE WHEN f.rowid IS NULL THEN 0 ELSE 1 END AS fts_present, CASE WHEN d.mirror_hash IS NULL AND f.rowid IS NULL THEN 1 WHEN f.rowid IS NOT NULL AND f.filepath = d.rel_path AND f.title = COALESCE(d.title, '') AND d.fts_mirror_hash = d.mirror_hash THEN 1 ELSE 0 END AS fts_current FROM documents d LEFT JOIN documents_fts f ON f.rowid = d.id WHERE d.collection = ? AND d.active = 1 ORDER BY d.uri ASC, d.id ASC`; function sanitizeFts5Term(term: string): string { return term.replace(/[^\p{L}\p{N}'_]/gu, "").toLowerCase(); } function isCompoundToken(token: string): boolean { return /^[\p{L}\p{N}][\p{L}\p{N}'+-]*[-+][\p{L}\p{N}][\p{L}\p{N}'+-]*$/u.test( token ); } function sanitizeCompoundTerm(term: string): string { return term .split(/[-+]/) .map((part) => sanitizeFts5Term(part)) .filter((part) => part.length > 0) .join(" "); } function normalizeDocEdgeType(edgeType: DocEdgeType): string { const normalized = edgeType.trim().toLowerCase(); if (!DOC_EDGE_TYPE_PATTERN.test(normalized)) { throw new Error( `Invalid edge_type "${edgeType}" (expected lowercase snake_case)` ); } return normalized; } type FtsQueryBuildResult = | { ok: true; query: string } | { ok: false; error: string }; /** * SQL fragment excluding documents superseded by an active document via the * typed `supersedes` edge. `docIdExpr` names the candidate document id column. */ const SUPERSEDED_EXCLUSION_SQL = (docIdExpr: string): string => `AND NOT EXISTS (SELECT 1 FROM doc_edges se JOIN documents sd ON sd.id = se.src_doc_id AND sd.active = 1 WHERE se.dst_doc_id = ${docIdExpr} AND se.edge_type = 'supersedes')`; /** * Narrow lexical grammar for BM25/FTS queries. * * Supported: * - plain terms -> prefix match * - quoted phrases -> phrase match * - negation with at least one positive term * - hyphenated compounds handled intentionally */ function buildFts5Query( query: string, options: { anyTerm?: boolean } = {} ): FtsQueryBuildResult { const trimmed = query.trim(); if (!trimmed) { return { ok: false, error: "Search query cannot be empty" }; } if (SINGLE_LINE_QUERY_PATTERN.test(trimmed)) { return { ok: false, error: "Lexical query must be a single line. Remove newline characters.", }; } const quoteCount = (trimmed.match(DOUBLE_QUOTE_PATTERN) ?? []).length; if (quoteCount % 2 === 1) { return { ok: false, error: 'Lexical query has an unmatched double quote ("). Add the closing quote or remove it.', }; } const positive: string[] = []; const negative: string[] = []; let i = 0; while (i < trimmed.length) { while (i < trimmed.length && /\s/.test(trimmed[i]!)) { i += 1; } if (i >= trimmed.length) { break; } const negated = trimmed[i] === "-"; if (negated) { i += 1; } if (i < trimmed.length && trimmed[i] === '"') { const start = i + 1; i += 1; while (i < trimmed.length && trimmed[i] !== '"') { i += 1; } const phrase = trimmed.slice(start, i).trim(); i += 1; if (!phrase) { continue; } const sanitized = phrase .split(WHITESPACE_REGEX) .map((token) => isCompoundToken(token) ? sanitizeCompoundTerm(token) : sanitizeFts5Term(token) ) .filter((token) => token.length > 0) .join(" "); if (!sanitized) { continue; } const ftsPhrase = `"${sanitized}"`; if (negated) { negative.push(ftsPhrase); } else { positive.push(ftsPhrase); } continue; } const start = i; while (i < trimmed.length && !/[\s"]/.test(trimmed[i]!)) { i += 1; } const token = trimmed.slice(start, i); if (!token) { continue; } if (isCompoundToken(token)) { const sanitized = sanitizeCompoundTerm(token); if (!sanitized) { continue; } const ftsPhrase = `"${sanitized}"`; if (negated) { negative.push(ftsPhrase); } else { positive.push(ftsPhrase); } continue; } const sanitized = sanitizeFts5Term(token); if (!sanitized) { continue; } const ftsTerm = `"${sanitized}"*`; if (negated) { negative.push(ftsTerm); } else { positive.push(ftsTerm); } } if (positive.length === 0 && negative.length === 0) { return { ok: false, error: "Search query has no searchable terms" }; } if (positive.length === 0) { return { ok: false, error: "Negation requires at least one positive search term in lexical queries.", }; } let ftsQuery = options.anyTerm ? `(${positive.join(" OR ")})` : positive.join(" AND "); for (const negation of negative) { ftsQuery = `${ftsQuery} NOT ${negation}`; } return { ok: true, query: ftsQuery }; } // ───────────────────────────────────────────────────────────────────────────── // SQLite Adapter Implementation // ───────────────────────────────────────────────────────────────────────────── /** Regex to strip .sqlite extension from db path */ const SQLITE_EXT_REGEX = /\.sqlite$/; /** Regex to strip index- prefix from db name */ const INDEX_PREFIX_REGEX = /^index-/; function isDatabaseLockedError(cause: unknown): boolean { return ( cause instanceof Error && cause.message.toLowerCase().includes("database is locked") ); } /** Resolve a caller-supplied busy_timeout, defaulting rather than using 0. */ function resolveBusyTimeoutMs(busyTimeoutMs?: number): number { if ( busyTimeoutMs === undefined || !Number.isInteger(busyTimeoutMs) || busyTimeoutMs < MIN_BUSY_TIMEOUT_MS || busyTimeoutMs > MAX_BUSY_TIMEOUT_MS ) { return DEFAULT_BUSY_TIMEOUT_MS; } return busyTimeoutMs; } export class SqliteAdapter implements StorePort, SqliteDbProvider { private db: Database | null = null; private dbPath = ""; private ftsTokenizer: FtsTokenizer = "unicode61"; private configPath = ""; // Set by CLI layer for status output private txCounter = 0; // Savepoint counter for unique names private readonly txContext = new AsyncLocalStorage<{ depth: number; token: { revoked: boolean }; }>(); private txTail: Promise = Promise.resolve(); private activeTransaction?: { token: { revoked: boolean }; release: () => void; }; private shutdownFenced = false; private shutdownDeadline?: number; private contextGeneration = 0; private chunkingGeneration = 0; // ───────────────────────────────────────────────────────────────────────── // Lifecycle // ───────────────────────────────────────────────────────────────────────── async open( dbPath: string, ftsTokenizer: FtsTokenizer, busyTimeoutMs: number = DEFAULT_BUSY_TIMEOUT_MS ): Promise> { try { this.db = new Database(dbPath, { create: true }); this.shutdownFenced = false; this.shutdownDeadline = undefined; this.dbPath = dbPath; this.ftsTokenizer = ftsTokenizer; // Enable pragmas for performance and safety this.db.exec("PRAGMA foreign_keys = ON"); this.db.exec( `PRAGMA busy_timeout = ${resolveBusyTimeoutMs(busyTimeoutMs)}` ); // Keep WAL everywhere so readers can continue while a writer is active. // CI still relaxes fsync/temp-store for speed, but MEMORY journal mode // breaks the cross-process read/write behavior we rely on in CLI tests. try { const journalMode = this.db .query<{ journal_mode: string }, []>("PRAGMA journal_mode") .get()?.journal_mode; if (journalMode?.toLowerCase() !== "wal") { this.db.exec("PRAGMA journal_mode = WAL"); } } catch (cause) { if (!isDatabaseLockedError(cause)) { throw cause; } // Another process may be switching journal mode or holding a write // lock during startup. In that case we keep the connection usable and // rely on the existing DB journal mode instead of failing open(). } if (process.env.CI) { this.db.exec("PRAGMA synchronous = OFF"); this.db.exec("PRAGMA temp_store = MEMORY"); } // Load fts5-snowball extension if using snowball tokenizer if (ftsTokenizer.startsWith("snowball")) { const snowballResult = loadFts5Snowball(this.db); if (!snowballResult.loaded) { this.db.close(); this.db = null; return err( "EXTENSION_LOAD_FAILED", `Failed to load fts5-snowball: ${snowballResult.error}` ); } } // Run migrations const result = runMigrations(this.db, migrations, ftsTokenizer); if (!result.ok) { this.db.close(); this.db = null; return result; } this.chunkingGeneration = readChunkingTarget(this.db).generation; this.contextGeneration += 1; return result; } catch (cause) { const message = cause instanceof Error ? cause.message : "Failed to open database"; return err("CONNECTION_FAILED", message, cause); } } /** Open an existing index with SQLite enforced query-only semantics. */ openReadOnly( dbPath: string, busyTimeoutMs: number = DEFAULT_BUSY_TIMEOUT_MS ): StoreResult { try { this.db = new Database(dbPath, { readonly: true, strict: true }); this.shutdownFenced = false; this.shutdownDeadline = undefined; this.dbPath = dbPath; this.db.exec("PRAGMA query_only = ON"); this.db.exec( `PRAGMA busy_timeout = ${resolveBusyTimeoutMs(busyTimeoutMs)}` ); this.contextGeneration += 1; return ok(undefined); } catch (cause) { this.db?.close(); this.db = null; return err( "CONNECTION_FAILED", cause instanceof Error ? cause.message : "Failed to open database read-only", cause ); } } async close(): Promise { this.fenceForShutdown(); if (this.db) { this.db.close(); this.db = null; } } /** Cap subsequent SQLite lock waits to the resident settlement deadline. */ beginShutdown(deadline: number): void { this.shutdownDeadline = deadline; if (this.db) this.capShutdownBusyWait(this.db); } private capShutdownBusyWait(db: Database): void { if (this.shutdownDeadline === undefined) return; const remaining = Math.max( 0, Math.floor(this.shutdownDeadline - performance.now()) ); const current = db.query<{ timeout: number }, []>("PRAGMA busy_timeout").get()?.timeout ?? 0; db.exec(`PRAGMA busy_timeout = ${Math.min(current, remaining)}`); } /** Revoke suspended transaction callbacks before closing their connection. */ fenceForShutdown(): void { this.shutdownFenced = true; const transaction = this.activeTransaction; if (!transaction) return; // JS cannot interleave this synchronous rollback with a running callback. // A callback suspended at await must never commit or use the store again. transaction.token.revoked = true; this.db?.exec("ROLLBACK"); transaction.release(); this.activeTransaction = undefined; } isOpen(): boolean { return this.db !== null; } /** * Run an async function within a single SQLite transaction. * Uses SAVEPOINT for nesting safety. * * Note: bun:sqlite's Database#transaction is synchronous, so we use * explicit BEGIN/COMMIT to support async callbacks. */ async withTransaction(fn: () => Promise): Promise> { const parent = this.txContext.getStore(); const connection = this.ensureOpen(); const isOuter = parent === undefined; const savepoint = `sp_${++this.txCounter}`; const releaseWriter = isOuter ? await this.acquireTransactionWriter() : null; const token = parent?.token ?? { revoked: false }; let db: Database | undefined; try { const current = this.ensureOpen(); if (current !== connection) throw new Error("Transaction connection retired before admission"); db = current; if (isOuter) { // IMMEDIATE reduces lock churn for bulk writes db.exec("BEGIN IMMEDIATE"); this.activeTransaction = { token, release: releaseWriter! }; } else { db.exec(`SAVEPOINT ${savepoint}`); } const value = await this.txContext.run( { depth: (parent?.depth ?? 0) + 1, token }, fn ); if (token.revoked || this.shutdownFenced) throw new Error("Transaction revoked by resident shutdown"); this.capShutdownBusyWait(db); if (isOuter) { db.exec("COMMIT"); } else { db.exec(`RELEASE ${savepoint}`); } return ok(value); } catch (cause) { try { if (!db || token.revoked) { // Shutdown already rolled back. Never touch a replacement connection. } else if (isOuter) { db.exec("ROLLBACK"); } else { db.exec(`ROLLBACK TO ${savepoint}`); db.exec(`RELEASE ${savepoint}`); } } catch { // Ignore rollback failures; report original error } const message = cause instanceof Error ? cause.message : "Transaction failed"; return err("TRANSACTION_FAILED", message, cause); } finally { if (isOuter && this.activeTransaction?.token === token) this.activeTransaction = undefined; releaseWriter?.(); } } private async acquireTransactionWriter(): Promise<() => void> { const previous = this.txTail; let release!: () => void; this.txTail = new Promise((resolve) => { release = resolve; }); await previous; return release; } /** * Set config path for status output (called by CLI layer). */ setConfigPath(configPath: string): void { this.configPath = configPath; } /** * Get raw SQLite database handle for vector operations. * Part of SqliteDbProvider interface - use with isSqliteDbProvider() type guard. */ getRawDb(): Database { return this.ensureOpen(); } graphReferenceStore() { return createGraphReferenceStore(this.ensureOpen()); } private ensureOpen(): Database { if (this.shutdownFenced || this.txContext.getStore()?.token.revoked) { throw new Error("Database fenced by resident shutdown"); } if (!this.db) { throw new Error("Database not open"); } this.capShutdownBusyWait(this.db); return this.db; } // ───────────────────────────────────────────────────────────────────────── // Config Sync // ───────────────────────────────────────────────────────────────────────── async syncCollections(collections: Collection[]): Promise> { try { const db = this.ensureOpen(); const transaction = db.transaction(() => { // Get existing collection names const existing = new Set( db .query<{ name: string }, []>("SELECT name FROM collections") .all() .map((r) => r.name) ); const incoming = new Set(collections.map((c) => c.name)); // Delete removed collections for (const name of existing) { if (!incoming.has(name)) { db.run("DELETE FROM collections WHERE name = ?", [name]); } } // Upsert collections const stmt = db.prepare(` INSERT INTO collections ( name, path, pattern, include, exclude, update_cmd, language_hint, egress_policy, egress_policy_source, egress_policy_revision, synced_at ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, datetime('now')) ON CONFLICT(name) DO UPDATE SET path = excluded.path, pattern = excluded.pattern, include = excluded.include, exclude = excluded.exclude, update_cmd = excluded.update_cmd, language_hint = excluded.language_hint, egress_policy = CASE WHEN excluded.egress_policy_source = 'config_default' AND collections.egress_policy_source = 'legacy_default' THEN collections.egress_policy ELSE excluded.egress_policy END, egress_policy_source = CASE WHEN excluded.egress_policy_source = 'config_default' AND collections.egress_policy_source = 'legacy_default' THEN collections.egress_policy_source ELSE excluded.egress_policy_source END, egress_policy_revision = CASE WHEN excluded.egress_policy_source = 'config_default' AND collections.egress_policy_source = 'legacy_default' THEN collections.egress_policy_revision ELSE excluded.egress_policy_revision END, synced_at = datetime('now') `); for (const c of collections) { const egress = resolveConfiguredEgressPolicy(c); stmt.run( c.name, c.path, c.pattern, c.include.length > 0 ? JSON.stringify(c.include) : null, c.exclude.length > 0 ? JSON.stringify(c.exclude) : null, c.updateCmd ?? null, c.languageHint ?? null, egress.policy, egress.source, c.egressPolicyRevision ?? 0 ); } }); transaction(); return ok(undefined); } catch (cause) { return err( "QUERY_FAILED", cause instanceof Error ? cause.message : "Failed to sync collections", cause ); } } async upsertCollections( collections: Collection[] ): Promise> { try { const db = this.ensureOpen(); const stmt = db.prepare(` INSERT INTO collections ( name, path, pattern, include, exclude, update_cmd, language_hint, egress_policy, egress_policy_source, egress_policy_revision, synced_at ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, datetime('now')) ON CONFLICT(name) DO UPDATE SET path = excluded.path, pattern = excluded.pattern, include = excluded.include, exclude = excluded.exclude, update_cmd = excluded.update_cmd, language_hint = excluded.language_hint, egress_policy = CASE WHEN excluded.egress_policy_source = 'config_default' AND collections.egress_policy_source = 'legacy_default' THEN collections.egress_policy ELSE excluded.egress_policy END, egress_policy_source = CASE WHEN excluded.egress_policy_source = 'config_default' AND collections.egress_policy_source = 'legacy_default' THEN collections.egress_policy_source ELSE excluded.egress_policy_source END, egress_policy_revision = CASE WHEN excluded.egress_policy_source = 'config_default' AND collections.egress_policy_source = 'legacy_default' THEN collections.egress_policy_revision ELSE excluded.egress_policy_revision END, synced_at = datetime('now') WHERE collections.path IS NOT excluded.path OR collections.pattern IS NOT excluded.pattern OR collections.include IS NOT excluded.include OR collections.exclude IS NOT excluded.exclude OR collections.update_cmd IS NOT excluded.update_cmd OR collections.language_hint IS NOT excluded.language_hint OR ( NOT ( excluded.egress_policy_source = 'config_default' AND collections.egress_policy_source = 'legacy_default' ) AND ( collections.egress_policy IS NOT excluded.egress_policy OR collections.egress_policy_source IS NOT excluded.egress_policy_source OR collections.egress_policy_revision IS NOT excluded.egress_policy_revision ) ) `); const transaction = db.transaction(() => { for (const collection of collections) { const egress = resolveConfiguredEgressPolicy(collection); stmt.run( collection.name, collection.path, collection.pattern, collection.include.length > 0 ? JSON.stringify(collection.include) : null, collection.exclude.length > 0 ? JSON.stringify(collection.exclude) : null, collection.updateCmd ?? null, collection.languageHint ?? null, egress.policy, egress.source, collection.egressPolicyRevision ?? 0 ); } }); transaction(); return ok(undefined); } catch (cause) { return err( "QUERY_FAILED", cause instanceof Error ? cause.message : "Failed to upsert collections", cause ); } } async syncContexts(contexts: Context[]): Promise> { try { const db = this.ensureOpen(); const transaction = db.transaction(() => { // Delete all and re-insert (contexts are small) db.run("DELETE FROM contexts"); const stmt = db.prepare(` INSERT INTO contexts (scope_type, scope_key, text, synced_at) VALUES (?, ?, ?, datetime('now')) ON CONFLICT(scope_type, scope_key, text) DO UPDATE SET synced_at = excluded.synced_at `); for (const c of contexts) { stmt.run(c.scopeType, c.scopeKey, c.text); } }); transaction(); this.contextGeneration += 1; return ok(undefined); } catch (cause) { return err( "QUERY_FAILED", cause instanceof Error ? cause.message : "Failed to sync contexts", cause ); } } async upsertContexts(contexts: Context[]): Promise> { try { const db = this.ensureOpen(); const stmt = db.prepare(` INSERT INTO contexts (scope_type, scope_key, text, synced_at) VALUES (?, ?, ?, datetime('now')) ON CONFLICT(scope_type, scope_key, text) DO NOTHING `); let inserted = 0; const transaction = db.transaction(() => { for (const context of contexts) { inserted += stmt.run( context.scopeType, context.scopeKey, context.text ).changes; } }); transaction(); if (inserted > 0) this.contextGeneration += 1; return ok(undefined); } catch (cause) { return err( "QUERY_FAILED", cause instanceof Error ? cause.message : "Failed to upsert contexts", cause ); } } async getCollections(): Promise> { try { const db = this.ensureOpen(); const rows = db .query("SELECT * FROM collections") .all(); return ok(rows.map(mapCollectionRow)); } catch (cause) { return err( "QUERY_FAILED", cause instanceof Error ? cause.message : "Failed to get collections", cause ); } } async getContexts(): Promise> { try { const db = this.ensureOpen(); const rows = db.query("SELECT * FROM contexts").all(); return ok(rows.map(mapContextRow)); } catch (cause) { return err( "QUERY_FAILED", cause instanceof Error ? cause.message : "Failed to get contexts", cause ); } } async getActivationIndexIdentity( collection: string ): Promise> { const snapshot = await this.getActivationIndexSnapshot(collection); return snapshot.ok ? ok(snapshot.value.identity) : snapshot; } async getActivationIndexSnapshot( collection: string ): Promise> { try { const db = this.ensureOpen(); const indexName = basename(this.dbPath) .replace(SQLITE_EXT_REGEX, "") .replace(INDEX_PREFIX_REGEX, "") || "default"; const rows = db .query< { id: number; uri: string; source_hash: string; mirror_hash: string | null; active: number; fts_present: number; fts_current: number; }, [string] >(ACTIVATION_INDEX_SNAPSHOT_SQL) .all(collection); const ftsHasher = new Bun.CryptoHasher("sha256"); ftsHasher.update( JSON.stringify( rows.map(({ uri, fts_present, fts_current }) => ({ uri, present: fts_present, current: fts_current, })) ) ); const documents: ActivationIndexDocument[] = rows.map((row) => ({ id: row.id, uri: row.uri, sourceHash: row.source_hash, mirrorHash: row.mirror_hash, active: row.active === 1, })); return ok({ identity: { indexName, schemaVersion: getSchemaVersion(db), ftsTokenizer: this.ftsTokenizer, ftsStateHash: ftsHasher.digest("hex"), activeDocumentCount: documents.length, ftsSynchronized: rows.every((row) => row.fts_current === 1), }, documents, }); } catch (cause) { return err( "QUERY_FAILED", cause instanceof Error ? cause.message : "Failed to get activation index identity", cause ); } } async getActivationReceipt( collection: string, expectedFingerprint: string, connectorTarget = "" ): Promise> { try { const db = this.ensureOpen(); const row = db .query< { collection: string; connector_target: string; fingerprint: string; receipt_json: string; }, [string, string] >( `SELECT collection, connector_target, fingerprint, receipt_json FROM activation_receipts WHERE collection = ? AND connector_target = ?` ) .get(collection, connectorTarget); if (!row) { return ok(null); } if (row.fingerprint !== expectedFingerprint) { db.run( "DELETE FROM activation_receipts WHERE collection = ? AND connector_target = ?", [collection, connectorTarget] ); return ok(null); } const receipt = parseActivationReceipt(row.receipt_json); if ( !receipt || receipt.fingerprint !== expectedFingerprint || receipt.collection !== row.collection || (receipt.evidence.connectorTarget ?? "") !== row.connector_target ) { db.run( "DELETE FROM activation_receipts WHERE collection = ? AND connector_target = ?", [collection, connectorTarget] ); return ok(null); } return ok(receipt); } catch (cause) { return err( "QUERY_FAILED", cause instanceof Error ? cause.message : "Failed to get activation receipt", cause ); } } async upsertActivationReceipt( receipt: ActivationVerificationReceipt ): Promise> { try { const db = this.ensureOpen(); const serialized = serializeActivationReceipt(receipt); if (!serialized.ok) { return err("INVALID_INPUT", serialized.error); } db.run( `INSERT INTO activation_receipts ( collection, connector_target, schema_version, fingerprint, receipt_json, updated_at ) VALUES (?, ?, ?, ?, ?, datetime('now')) ON CONFLICT(collection, connector_target) DO UPDATE SET schema_version = excluded.schema_version, fingerprint = excluded.fingerprint, receipt_json = excluded.receipt_json, updated_at = datetime('now')`, [ serialized.projected.collection, serialized.connectorTarget, serialized.projected.schemaVersion, serialized.projected.fingerprint, serialized.json, ] ); return ok(undefined); } catch (cause) { return err( "QUERY_FAILED", cause instanceof Error ? cause.message : "Failed to persist activation receipt", cause ); } } async createRetrievalTrace( input: RetrievalTraceInput ): Promise> { return createStoredTrace(this.ensureOpen(), input); } async mergeRetrievalTraceEgressLineage( traceId: string, lineage: EgressLineage ): Promise> { return mergeStoredTraceEgressLineage(this.ensureOpen(), traceId, lineage); } async getRetrievalTrace( traceId: string ): Promise> { return getStoredTrace(this.ensureOpen(), traceId); } async getBoundedRetrievalTrace( traceId: string, detailLimit: number ): Promise> { return getBoundedStoredTrace(this.ensureOpen(), traceId, detailLimit); } async listRetrievalTraces( limit: number, cursor?: RetrievalTraceCursor ): Promise> { return listStoredTraces(this.ensureOpen(), limit, cursor); } async finalizeRetrievalTrace( traceId: string, status: RetrievalTraceTerminalStatus, updatedAtMs: number ): Promise> { return finalizeStoredTrace(this.ensureOpen(), traceId, status, updatedAtMs); } async appendRetrievalTraceRun( input: RetrievalTraceRunInput ): Promise> { return appendStoredTraceRun(this.ensureOpen(), input); } async appendRetrievalTraceEvent( input: RetrievalTraceEventInput ): Promise> { return appendStoredTraceEvent(this.ensureOpen(), input); } async appendRetrievalTraceJudgment( input: RetrievalTraceJudgmentInput ): Promise> { return appendStoredTraceJudgment(this.ensureOpen(), input); } async appendRetrievalTraceExport( input: RetrievalTraceExportInput ): Promise> { return appendStoredTraceExport(this.ensureOpen(), input); } async appendRetrievalTraceExportManifest( input: RetrievalTraceExportManifestInput ): Promise> { return appendStoredTraceExportManifest(this.ensureOpen(), input); } async getRetrievalTraceExportManifest( exportId: string ): Promise> { return getStoredTraceExportManifest(this.ensureOpen(), exportId); } async getRetrievalTraceExportBundle( exportId: string ): Promise> { return getStoredTraceExportBundle(this.ensureOpen(), exportId); } async getOrCreateRetrievalTraceRedactionSecret(): Promise< StoreResult > { return getOrCreateStoredTraceRedactionSecret(this.ensureOpen()); } async deleteRetrievalTrace( traceId: string ): Promise> { return deleteStoredTrace(this.ensureOpen(), traceId); } async purgeRetrievalTraces(): Promise< StoreResult > { return purgeStoredTraces(this.ensureOpen()); } async enforceRetrievalTraceRetention( policy: RetrievalTraceRetentionPolicy, nowMs: number ): Promise> { return enforceStoredTraceRetention(this.ensureOpen(), policy, nowMs); } async appendEgressAuditReceipt( receipt: EgressAuditReceiptInput ): Promise> { return appendStoredEgressAuditReceipt(this.ensureOpen(), receipt); } async appendEgressAuditReceiptWithRetention( receipt: EgressAuditReceiptInput, policy: EgressAuditRetentionPolicy, nowMs: number ): Promise> { return appendStoredEgressAuditReceiptWithRetention( this.ensureOpen(), receipt, policy, nowMs ); } async listEgressAuditReceipts( limit: number, cursor?: EgressAuditCursor ): Promise> { return listStoredEgressAuditReceipts(this.ensureOpen(), limit, cursor); } async getEgressAuditReceipt( auditId: string ): Promise> { return getStoredEgressAuditReceipt(this.ensureOpen(), auditId); } async deleteEgressAuditReceipt( auditId: string ): Promise> { return deleteStoredEgressAuditReceipt(this.ensureOpen(), auditId); } async getEgressAuditStatus(): Promise> { return getStoredEgressAuditStatus(this.ensureOpen()); } async enforceEgressAuditRetention( policy: EgressAuditRetentionPolicy, nowMs: number ): Promise> { return enforceStoredEgressAuditRetention(this.ensureOpen(), policy, nowMs); } async purgeEgressAuditReceipts(): Promise< StoreResult > { return purgeStoredEgressAuditReceipts(this.ensureOpen()); } getContextGeneration(): number { return this.contextGeneration; } // ───────────────────────────────────────────────────────────────────────── // Documents // ───────────────────────────────────────────────────────────────────────── async getTypedMetadataCoverage( options: import("../types").DocumentEligibilityOptions ): Promise> { try { const db = this.ensureOpen(); const eligible = buildEligibleDocumentQuery( { ...options, filter: undefined }, db ); const row = db .query<{ pending: number; invalid: number }, (string | number)[]>(` SELECT COALESCE(SUM(CASE WHEN d.ingest_version IS NULL OR d.ingest_version < ${TYPED_METADATA_INGEST_VERSION} OR (d.typed_metadata IS NULL AND d.metadata_error IS NULL) THEN 1 ELSE 0 END),0) AS pending, COALESCE(SUM(CASE WHEN d.metadata_error IS NOT NULL THEN 1 ELSE 0 END),0) AS invalid FROM documents d WHERE d.mirror_hash IS NOT NULL AND d.id IN (SELECT id FROM (${eligible.sql})) `) .get(...eligible.params); return ok(row ?? { pending: 0, invalid: 0 }); } catch (cause) { return err("QUERY_FAILED", "Cannot determine typed metadata coverage", { cause, }); } } async upsertDocument( doc: DocumentInput ): Promise> { try { const db = this.ensureOpen(); const docid = deriveDocid(doc.sourceHash); const uri = buildUri(doc.collection, doc.relPath); const transaction = db.transaction((): UpsertDocumentResult => { const previousRow = db .query( "SELECT * FROM documents WHERE collection = ? AND rel_path = ?" ) .get(doc.collection, doc.relPath); const legacyTitles = snapshotLegacyTitles( db, !previousRow || !previousRow.active || previousRow.title !== (doc.title ?? null) || previousRow.mirror_hash !== (doc.mirrorHash ?? null) ? [previousRow?.mirror_hash, doc.mirrorHash] : [] ); db.run( ` INSERT INTO documents ( collection, rel_path, source_hash, source_mime, source_ext, source_size, source_mtime, source_ctime, docid, uri, title, mirror_hash, converter_id, converter_version, language_hint, content_type, categories, content_type_source, author, frontmatter_date, date_fields, typed_metadata, metadata_error, record_key, record_source_path, record_source_locator, record_metadata, record_anchors, record_adapter_fingerprint, content_type_rules_fingerprint, active, indexed_at, last_error_code, last_error_message, last_error_at, ingest_version, updated_at ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 1, datetime('now'), ?, ?, ?, ?, datetime('now')) ON CONFLICT(collection, rel_path) DO UPDATE SET source_hash = excluded.source_hash, source_mime = excluded.source_mime, source_ext = excluded.source_ext, source_size = excluded.source_size, source_mtime = excluded.source_mtime, source_ctime = excluded.source_ctime, docid = excluded.docid, uri = excluded.uri, title = excluded.title, mirror_hash = excluded.mirror_hash, converter_id = excluded.converter_id, converter_version = excluded.converter_version, language_hint = excluded.language_hint, content_type = excluded.content_type, categories = excluded.categories, content_type_source = excluded.content_type_source, author = excluded.author, frontmatter_date = excluded.frontmatter_date, date_fields = excluded.date_fields, typed_metadata = excluded.typed_metadata, metadata_error = excluded.metadata_error, record_key = excluded.record_key, record_source_path = excluded.record_source_path, record_source_locator = excluded.record_source_locator, record_metadata = excluded.record_metadata, record_anchors = excluded.record_anchors, record_adapter_fingerprint = excluded.record_adapter_fingerprint, content_type_rules_fingerprint = excluded.content_type_rules_fingerprint, active = 1, indexed_at = datetime('now'), last_error_code = excluded.last_error_code, last_error_message = excluded.last_error_message, last_error_at = excluded.last_error_at, ingest_version = excluded.ingest_version, updated_at = datetime('now') `, [ doc.collection, doc.relPath, doc.sourceHash, doc.sourceMime, doc.sourceExt, doc.sourceSize, doc.sourceMtime, doc.sourceCtime ?? doc.sourceMtime, docid, uri, doc.title ?? null, doc.mirrorHash ?? null, doc.converterId ?? null, doc.converterVersion ?? null, doc.languageHint ?? null, doc.contentType ?? null, doc.categories ? JSON.stringify(doc.categories) : null, doc.contentTypeSource ?? null, doc.author ?? null, doc.frontmatterDate ?? null, doc.dateFields ? JSON.stringify(doc.dateFields) : null, doc.typedMetadata ? JSON.stringify(typedMetadataSchema.parse(doc.typedMetadata)) : null, doc.metadataError ?? null, doc.recordKey ?? null, doc.recordSourcePath ?? null, doc.recordSourceLocator ?? null, doc.recordMetadata ? JSON.stringify(doc.recordMetadata) : null, doc.recordAnchors ? JSON.stringify(doc.recordAnchors) : null, doc.recordAdapterFingerprint ?? null, doc.contentTypeRulesFingerprint ?? null, doc.lastErrorCode ?? null, doc.lastErrorMessage ?? null, doc.lastErrorCode ? new Date().toISOString() : null, doc.ingestVersion ?? null, ] ); // Get the row id (either inserted or updated) const idRow = db .query<{ id: number }, [string, string]>( "SELECT id FROM documents WHERE collection = ? AND rel_path = ?" ) .get(doc.collection, doc.relPath); if (!idRow) { throw new Error("Failed to get document id after upsert"); } reconcileLegacyTitles(db, legacyTitles); // Owner bindings describe the exact current source input. Keep inactive // bindings for identical restoration, but invalidate changed ownership. if ( previousRow && (previousRow.mirror_hash !== (doc.mirrorHash ?? null) || previousRow.title !== (doc.title ?? null)) ) { db.run("DELETE FROM vector_owners WHERE document_id = ?", [idRow.id]); } // Conversion failures deliberately drop mirror ownership. Remove the // old lexical projection in the same transaction so stale content can // never join against the newly updated document metadata. if (doc.mirrorHash == null) { db.run("DELETE FROM documents_fts WHERE rowid = ?", [idRow.id]); db.run("UPDATE documents SET fts_mirror_hash = NULL WHERE id = ?", [ idRow.id, ]); } if (doc.changeJournal !== false) { const nextRow = db .query( "SELECT * FROM documents WHERE id = ?" ) .get(idRow.id); if (!nextRow) { throw new Error("Failed to read document after upsert"); } const previous = previousRow ? snapshotDocumentChange(mapDocumentRow(previousRow)) : null; const next = snapshotDocumentChange(mapDocumentRow(nextRow)); const kind: DocumentChangeKind | null = previous === null ? "create" : !previous.active ? "reactivate" : previous.sourceHash !== next.sourceHash || previous.mirrorHash !== next.mirrorHash ? "update" : null; if (kind) { appendStoredDocumentChange(db, { documentId: idRow.id, collection: doc.collection, kind, oldSnapshot: previous, newSnapshot: next, structureDelta: doc.changeJournal?.structureDelta, observedAtMs: doc.changeJournal?.observedAtMs ?? Date.now(), }); } } return { id: idRow.id, docid }; }); return ok(transaction()); } catch (cause) { return err( "QUERY_FAILED", cause instanceof Error ? cause.message : "Failed to upsert document", cause ); } } async renameDocument( collection: string, oldRelPath: string, newRelPath: string, options: RenameDocumentOptions = {} ): Promise> { try { if (!oldRelPath || !newRelPath) { return err("INVALID_INPUT", "Rename paths must be non-empty"); } const db = this.ensureOpen(); const transaction = db.transaction((): DocumentRow => { const oldRow = db .query( "SELECT * FROM documents WHERE collection = ? AND rel_path = ?" ) .get(collection, oldRelPath); if (!oldRow) { throw new Error("DOCUMENT_RENAME_NOT_FOUND"); } if (oldRelPath === newRelPath) { return mapDocumentRow(oldRow); } const newUri = buildUri(collection, newRelPath); db.run( `UPDATE documents SET rel_path = ?, uri = ?, updated_at = datetime('now') WHERE id = ?`, [newRelPath, newUri, oldRow.id] ); db.run("UPDATE documents_fts SET filepath = ? WHERE rowid = ?", [ newRelPath, oldRow.id, ]); const nextRow = db .query( "SELECT * FROM documents WHERE id = ?" ) .get(oldRow.id); if (!nextRow) { throw new Error("Failed to read document after rename"); } const oldSnapshot = snapshotDocumentChange(mapDocumentRow(oldRow)); const nextDocument = mapDocumentRow(nextRow); appendStoredDocumentChange(db, { documentId: oldRow.id, collection, kind: "rename", oldSnapshot, newSnapshot: snapshotDocumentChange(nextDocument), structureDelta: options.structureDelta, observedAtMs: options.observedAtMs ?? Date.now(), }); return nextDocument; }); return ok(transaction()); } catch (cause) { if ( cause instanceof Error && cause.message === "DOCUMENT_RENAME_NOT_FOUND" ) { return err( "NOT_FOUND", `Document not found for rename: ${collection}/${oldRelPath}` ); } return err( "QUERY_FAILED", cause instanceof Error ? cause.message : "Failed to rename document", cause ); } } async getDocument( collection: string, relPath: string ): Promise> { try { const db = this.ensureOpen(); const row = db .query( "SELECT * FROM documents WHERE collection = ? AND rel_path = ?" ) .get(collection, relPath); return ok(row ? mapDocumentRow(row) : null); } catch (cause) { return err( "QUERY_FAILED", cause instanceof Error ? cause.message : "Failed to get document", cause ); } } async getDocumentByDocid( docid: string ): Promise> { try { const db = this.ensureOpen(); const row = db .query( "SELECT * FROM documents WHERE docid = ? ORDER BY active DESC, id ASC LIMIT 1" ) .get(docid); return ok(row ? mapDocumentRow(row) : null); } catch (cause) { return err( "QUERY_FAILED", cause instanceof Error ? cause.message : "Failed to get document by docid", cause ); } } async getDocumentByUri( uri: string ): Promise> { try { const db = this.ensureOpen(); const canonicalUri = stripUriIndex(uri); const row = db .query("SELECT * FROM documents WHERE uri = ?") .get(canonicalUri); return ok(row ? mapDocumentRow(row) : null); } catch (cause) { return err( "QUERY_FAILED", cause instanceof Error ? cause.message : "Failed to get document by uri", cause ); } } async listDocuments( collection?: string ): Promise> { try { const db = this.ensureOpen(); const rows = collection ? db .query( "SELECT * FROM documents WHERE collection = ?" ) .all(collection) : db.query("SELECT * FROM documents").all(); return ok(rows.map(mapDocumentRow)); } catch (cause) { return err( "QUERY_FAILED", cause instanceof Error ? cause.message : "Failed to list documents", cause ); } } async listDocumentsForAudit(options: { collections: readonly string[]; pathPrefixes: readonly string[]; tags: readonly string[]; limit: number; }): Promise> { try { const db = this.ensureOpen(); const conditions = ["d.active = 1"]; const params: (string | number)[] = []; if (options.collections.length > 0) { conditions.push("d.collection IN (SELECT value FROM json_each(?))"); params.push(JSON.stringify(options.collections)); } if (options.pathPrefixes.length > 0) { conditions.push(`EXISTS ( SELECT 1 FROM json_each(?) prefix WHERE COALESCE(NULLIF(d.record_source_path, ''), d.rel_path) = prefix.value OR (substr(COALESCE(NULLIF(d.record_source_path, ''), d.rel_path), 1, length(prefix.value)) = prefix.value AND substr(COALESCE(NULLIF(d.record_source_path, ''), d.rel_path), length(prefix.value) + 1, 1) = '/') )`); params.push(JSON.stringify(options.pathPrefixes)); } if (options.tags.length > 0) { conditions.push(`NOT EXISTS ( SELECT 1 FROM json_each(?) requested_tag WHERE NOT EXISTS ( SELECT 1 FROM doc_tags dt WHERE dt.document_id = d.id AND dt.tag = requested_tag.value ) )`); params.push(JSON.stringify(options.tags)); } const where = conditions.join(" AND "); const total = db .query<{ count: number }, (string | number)[]>( `SELECT COUNT(*) AS count FROM documents d WHERE ${where}` ) .get(...params)?.count ?? 0; const rows = db .query( `SELECT d.* FROM documents d WHERE ${where} ORDER BY d.id LIMIT ?` ) .all(...params, options.limit); return ok({ documents: rows.map(mapDocumentRow), total }); } catch (cause) { return err( "QUERY_FAILED", cause instanceof Error ? cause.message : "Failed to select bounded audit documents", cause ); } } async listRecordDocuments( collection: string, sourcePath: string ): Promise> { try { const db = this.ensureOpen(); const rows = db .query( `SELECT * FROM documents WHERE collection = ? AND record_source_path = ? ORDER BY rel_path ASC, id ASC` ) .all(collection, sourcePath); return ok(rows.map(mapDocumentRow)); } catch (cause) { return err( "QUERY_FAILED", cause instanceof Error ? cause.message : "Failed to list record documents", cause ); } } async listActiveDirectChildSourcePaths( collection: string, dirRelPath: string, max: number ): Promise> { if (!Number.isInteger(max) || max <= 0) { return err("INVALID_INPUT", "max must be a positive integer"); } const parentPath = normalizeWatcherSourceDirRelPath(dirRelPath); if (parentPath === null) { return err( "INVALID_INPUT", `Directory path escapes the collection root: ${dirRelPath}` ); } try { const db = this.ensureOpen(); // Effective source path: record containers resolve to physical container. // Parent key: empty string for root-level sources. // LIMIT max+1 detects overflow without returning a truncated success. const rows = db .query<{ source_path: string }, [string, string, number]>( `SELECT DISTINCT ${WATCHER_SOURCE_PATH_SQL} AS source_path FROM documents WHERE collection = ? AND active = 1 AND ${WATCHER_SOURCE_PARENT_SQL} = ? ORDER BY source_path ASC LIMIT ?` ) .all(collection, parentPath, max + 1); if (rows.length > max) { return err( "OVERFLOW", `Active direct-child source paths exceed max=${max}` ); } return ok(rows.map((row) => row.source_path)); } catch (cause) { return err( "QUERY_FAILED", cause instanceof Error ? cause.message : "Failed to list active direct child source paths", cause ); } } async listActiveDescendantSourcePaths( collection: string, dirRelPath: string, max: number ): Promise> { if (!Number.isInteger(max) || max <= 0) { return err("INVALID_INPUT", "max must be a positive integer"); } const directory = normalizeWatcherSourceDirRelPath(dirRelPath); if (directory === null) { return err( "INVALID_INPUT", `Directory path escapes the collection root: ${dirRelPath}` ); } if (directory === "") { return err( "INVALID_INPUT", "Descendant lookup requires a directory below the collection root" ); } try { const db = this.ensureOpen(); // Exact prefix boundary: `dir1/` never matches `dir10/...`. // length(?) is code-point-safe (JS prefix.length is UTF-16 and breaks non-BMP). // LIMIT max+1 detects overflow without returning a truncated success. const prefix = `${directory}/`; const rows = db .query<{ source_path: string }, [string, string, string, number]>( `SELECT DISTINCT ${WATCHER_SOURCE_PATH_SQL} AS source_path FROM documents WHERE collection = ? AND active = 1 AND substr(${WATCHER_SOURCE_PATH_SQL}, 1, length(?)) = ? ORDER BY source_path ASC LIMIT ?` ) .all(collection, prefix, prefix, max + 1); if (rows.length > max) { return err( "OVERFLOW", `Active descendant source paths exceed max=${max}` ); } return ok(rows.map((row) => row.source_path)); } catch (cause) { return err( "QUERY_FAILED", cause instanceof Error ? cause.message : "Failed to list active descendant source paths", cause ); } } async listActiveSourcePaths( collection: string, max: number ): Promise> { if (!Number.isInteger(max) || max <= 0) { return err("INVALID_INPUT", "max must be a positive integer"); } try { const db = this.ensureOpen(); // Root-wide DISTINCT physical sources; overflow after collapse (max+1). const rows = db .query<{ source_path: string }, [string, number]>( `SELECT DISTINCT ${WATCHER_SOURCE_PATH_SQL} AS source_path FROM documents WHERE collection = ? AND active = 1 ORDER BY source_path ASC LIMIT ?` ) .all(collection, max + 1); if (rows.length > max) { return err("OVERFLOW", `Active source paths exceed max=${max}`); } return ok(rows.map((row) => row.source_path)); } catch (cause) { return err( "QUERY_FAILED", cause instanceof Error ? cause.message : "Failed to list active source paths", cause ); } } async listActiveDocumentsForBrowse( collection?: string ): Promise> { try { const db = this.ensureOpen(); const rows = collection ? db .query( `SELECT * FROM documents WHERE collection = ? AND active = 1 ORDER BY rel_path ASC, id ASC` ) .all(collection) : db .query( `SELECT * FROM documents WHERE active = 1 ORDER BY collection ASC, rel_path ASC, id ASC` ) .all(); return ok(rows.map(mapDocumentRow)); } catch (cause) { return err( "QUERY_FAILED", cause instanceof Error ? cause.message : "Failed to list active browse documents", cause ); } } async getDocumentsByMirrorHashes( mirrorHashes: string[], options: { collection?: string; activeOnly?: boolean; } = {} ): Promise> { try { if (mirrorHashes.length === 0) { return ok([]); } const uniqueHashes = [ ...new Set(mirrorHashes.filter((hash) => hash.trim().length > 0)), ]; if (uniqueHashes.length === 0) { return ok([]); } const db = this.ensureOpen(); const rows: DbDocumentRow[] = []; // SQLite SQLITE_LIMIT_VARIABLE_NUMBER defaults to 999. // Reserve headroom for optional non-IN parameters. const SQL_PARAM_LIMIT = options.collection ? 899 : 900; for (let i = 0; i < uniqueHashes.length; i += SQL_PARAM_LIMIT) { const batch = uniqueHashes.slice(i, i + SQL_PARAM_LIMIT); const placeholders = batch.map(() => "?").join(","); const clauses = [`mirror_hash IN (${placeholders})`]; const params: string[] = [...batch]; if (options.activeOnly ?? true) { clauses.push("active = 1"); } if (options.collection) { clauses.push("collection = ?"); params.push(options.collection); } const sql = `SELECT * FROM documents WHERE ${clauses.join(" AND ")} ORDER BY id`; rows.push(...db.query(sql).all(...params)); } return ok(rows.map(mapDocumentRow)); } catch (cause) { return err( "QUERY_FAILED", cause instanceof Error ? cause.message : "Failed to get documents by mirror hashes", cause ); } } async getDocumentsByDocids( docids: string[], options: { collection?: string; activeOnly?: boolean; eligibility?: import("../types").DocumentEligibilityOptions; } = {} ): Promise> { try { if (docids.length === 0) { return ok([]); } const uniqueDocids = [ ...new Set(docids.filter((docid) => docid.trim().length > 0)), ]; if (uniqueDocids.length === 0) { return ok([]); } const db = this.ensureOpen(); const rows: DbDocumentRow[] = []; const SQL_PARAM_LIMIT = options.collection ? 899 : 900; for (let i = 0; i < uniqueDocids.length; i += SQL_PARAM_LIMIT) { const batch = uniqueDocids.slice(i, i + SQL_PARAM_LIMIT); const placeholders = batch.map(() => "?").join(","); const clauses = [`docid IN (${placeholders})`]; const params: (string | number)[] = [...batch]; if (options.activeOnly ?? true) { clauses.push("active = 1"); } if (options.collection) { clauses.push("collection = ?"); params.push(options.collection); } if (options.eligibility) { const eligible = buildEligibleDocumentQuery(options.eligibility, db); clauses.push(`id IN (SELECT id FROM (${eligible.sql}))`); params.push(...eligible.params); } const sql = `SELECT * FROM documents WHERE ${clauses.join(" AND ")} ORDER BY id`; rows.push( ...db.query(sql).all(...params) ); } return ok(rows.map(mapDocumentRow)); } catch (cause) { return err( "QUERY_FAILED", cause instanceof Error ? cause.message : "Failed to get documents by docids", cause ); } } async listDocumentsPaginated(options: { collection?: string; limit: number; offset: number; pathPrefix?: string; directChildrenOnly?: boolean; tagsAll?: string[]; tagsAny?: string[]; since?: string; until?: string; categories?: string[]; author?: string; sortField?: string; sortOrder?: "asc" | "desc"; }): Promise> { try { const db = this.ensureOpen(); const { collection, limit, offset, pathPrefix, directChildrenOnly, tagsAll, tagsAny, } = options; // Build WHERE conditions and params const conditions: string[] = ["d.active = 1"]; const params: (string | number)[] = []; if (collection) { conditions.push("d.collection = ?"); params.push(collection); } const normalizedPathPrefix = pathPrefix ?.replaceAll("\\", "/") .replace(/^\/+|\/+$/g, ""); const browsePathSql = "COALESCE(d.record_source_path, d.rel_path)"; if (normalizedPathPrefix) { conditions.push(`${browsePathSql} LIKE ?`); params.push(`${normalizedPathPrefix}/%`); if (directChildrenOnly) { conditions.push(`substr(${browsePathSql}, ?) NOT LIKE '%/%'`); params.push(normalizedPathPrefix.length + 2); } } else if (directChildrenOnly) { conditions.push(`${browsePathSql} NOT LIKE '%/%'`); } if (options.since) { conditions.push("d.source_mtime >= ?"); params.push(options.since); } if (options.until) { conditions.push("d.source_mtime <= ?"); params.push(options.until); } if (options.categories && options.categories.length > 0) { const placeholders = options.categories.map(() => "?").join(","); conditions.push( `(d.content_type IN (${placeholders}) OR EXISTS (SELECT 1 FROM json_each(COALESCE(d.categories, '[]')) jc WHERE jc.value IN (${placeholders})))` ); params.push(...options.categories, ...options.categories); } if (options.author) { conditions.push("LOWER(COALESCE(d.author, '')) LIKE ?"); params.push(`%${options.author.toLowerCase()}%`); } // tagsAny: document has at least one of these tags (OR) if (tagsAny && tagsAny.length > 0) { const placeholders = tagsAny.map(() => "?").join(","); conditions.push( `EXISTS (SELECT 1 FROM doc_tags dt WHERE dt.document_id = d.id AND dt.tag IN (${placeholders}))` ); params.push(...tagsAny); } // tagsAll: document has all of these tags (AND) if (tagsAll && tagsAll.length > 0) { for (const tag of tagsAll) { conditions.push( "EXISTS (SELECT 1 FROM doc_tags dt WHERE dt.document_id = d.id AND dt.tag = ?)" ); params.push(tag); } } const whereClause = conditions.join(" AND "); // Sort options const sortOrder = options.sortOrder === "asc" ? "ASC" : "DESC"; const sortField = options.sortField ?? "modified"; const isSafeDateField = /^[a-z0-9_]+$/.test(sortField); let orderClause = `d.source_mtime ${sortOrder}`; if (sortField !== "modified" && isSafeDateField) { orderClause = `COALESCE(json_extract(d.date_fields, '$."${sortField}"'), d.source_mtime) ${sortOrder}`; } // Get total count // Use COUNT(DISTINCT d.id) to prevent duplicate counting when tag filters match multiple tags const countSql = `SELECT COUNT(DISTINCT d.id) as count FROM documents d WHERE ${whereClause}`; const countRow = db .query<{ count: number }, (string | number)[]>(countSql) .get(...params); const total = countRow?.count ?? 0; // Get paginated documents // Use DISTINCT to prevent duplicate rows when tag filters match multiple tags const selectSql = `SELECT DISTINCT d.* FROM documents d WHERE ${whereClause} ORDER BY ${orderClause}, d.id ASC LIMIT ? OFFSET ?`; const rows = db .query(selectSql) .all(...params, limit, offset); return ok({ documents: rows.map(mapDocumentRow), total }); } catch (cause) { return err( "QUERY_FAILED", cause instanceof Error ? cause.message : "Failed to list documents", cause ); } } async getCollectionDateFields( collection?: string ): Promise> { try { const db = this.ensureOpen(); const conditions: string[] = [ "d.active = 1", "d.date_fields IS NOT NULL", ]; const params: string[] = []; if (collection) { conditions.push("d.collection = ?"); params.push(collection); } const sql = ` SELECT DISTINCT jf.key as field FROM documents d JOIN json_each(COALESCE(d.date_fields, '{}')) jf WHERE ${conditions.join(" AND ")} ORDER BY jf.key ASC `; const rows = db.query<{ field: string }, string[]>(sql).all(...params); return ok(rows.map((r) => r.field)); } catch (cause) { return err( "QUERY_FAILED", cause instanceof Error ? cause.message : "Failed to list collection date fields", cause ); } } async markInactive( collection: string, relPaths: string[] ): Promise> { try { const db = this.ensureOpen(); if (relPaths.length === 0) { return ok(0); } const uniquePaths = [...new Set(relPaths)]; const placeholders = uniquePaths.map(() => "?").join(","); const transaction = db.transaction((): number => { const activeRows = db .query( `SELECT * FROM documents WHERE collection = ? AND active = 1 AND rel_path IN (${placeholders}) ORDER BY rel_path ASC, id ASC` ) .all(collection, ...uniquePaths); if (activeRows.length === 0) { return 0; } const legacyTitles = snapshotLegacyTitles( db, activeRows.map((row) => row.mirror_hash) ); db.run( `UPDATE documents SET active = 0, updated_at = datetime('now') WHERE collection = ? AND active = 1 AND rel_path IN (${placeholders})`, [collection, ...uniquePaths] ); // Bun run().changes includes trigger writes; report logical document rows. const changed = db.query<{ count: number }, []>("SELECT changes() AS count").get() ?.count ?? 0; reconcileLegacyTitles(db, legacyTitles); const observedAtMs = Date.now(); for (const activeRow of activeRows) { const previous = snapshotDocumentChange(mapDocumentRow(activeRow)); appendStoredDocumentChange(db, { documentId: activeRow.id, collection, kind: "inactivate", oldSnapshot: previous, newSnapshot: { ...previous, active: false }, observedAtMs, }); } return changed; }); return ok(transaction()); } catch (cause) { return err( "QUERY_FAILED", cause instanceof Error ? cause.message : "Failed to mark documents inactive", cause ); } } async listDocumentChanges( options: DocumentChangeListOptions = {} ): Promise> { return listStoredDocumentChanges(this.ensureOpen(), options); } async enforceDocumentChangeRetention( policy: DocumentChangeRetentionPolicy, nowMs: number ): Promise> { return enforceStoredDocumentChangeRetention( this.ensureOpen(), policy, nowMs ); } async purgeDocumentChanges(): Promise< StoreResult > { return purgeStoredDocumentChanges(this.ensureOpen()); } async createFileRefactorPreparedReceipt( draft: FileRefactorRecoveryReceiptDraft ): Promise> { return createStoredFileRefactorPreparedReceipt(this.ensureOpen(), draft); } async advanceFileRefactorReceipt( journalId: string, update: FileRefactorJournalAdvance ): Promise> { return advanceStoredFileRefactorReceipt( this.ensureOpen(), journalId, update ); } async getFileRefactorReceiptById( journalId: string ): Promise> { return getStoredFileRefactorReceiptById(this.ensureOpen(), journalId); } async getLatestFileRefactorReceiptByPlanDigest( planDigest: string ): Promise> { return getStoredLatestFileRefactorReceiptByPlanDigest( this.ensureOpen(), planDigest ); } async upsertSavedCapsuleRegistration( input: SavedCapsuleRegistrationInput ): Promise> { return upsertStoredSavedCapsuleRegistration(this.ensureOpen(), input); } async listSavedCapsuleRegistrations(): Promise< StoreResult > { return listStoredSavedCapsuleRegistrations(this.ensureOpen()); } async getSavedCapsuleRegistration( registrationId: string ): Promise> { return getStoredSavedCapsuleRegistration(this.ensureOpen(), registrationId); } async getSavedCapsuleRegistrationSnapshot( registrationId: string ): Promise> { return getStoredSavedCapsuleRegistrationSnapshot( this.ensureOpen(), registrationId ); } async deleteSavedCapsuleRegistration( registrationId: string ): Promise> { return deleteStoredSavedCapsuleRegistration( this.ensureOpen(), registrationId ); } async listSavedCapsuleIdsAffectedByChanges( afterSequence: number, throughSequence: number, limit: number ): Promise> { return listStoredSavedCapsuleIdsAffectedByChanges( this.ensureOpen(), afterSequence, throughSequence, limit ); } async upsertSavedCapsuleVerification( verification: SavedCapsuleVerificationRecord, expectedRegistration: SavedCapsuleVerificationExpectation ): Promise> { return upsertStoredSavedCapsuleVerification( this.ensureOpen(), verification, expectedRegistration ); } async getSavedCapsuleReverificationSequence(): Promise> { return getStoredSavedCapsuleReverificationSequence(this.ensureOpen()); } async getSavedCapsuleReverificationState(): Promise< StoreResult > { return getStoredSavedCapsuleReverificationState(this.ensureOpen()); } async setSavedCapsuleReverificationSequence( sequence: number, expectedRegistrationEpoch: number ): Promise> { return setStoredSavedCapsuleReverificationSequence( this.ensureOpen(), sequence, expectedRegistrationEpoch ); } // ───────────────────────────────────────────────────────────────────────── // Content // ───────────────────────────────────────────────────────────────────────── async upsertContent( mirrorHash: string, markdown: string ): Promise> { try { const db = this.ensureOpen(); db.run( `INSERT INTO content (mirror_hash, markdown) VALUES (?, ?) ON CONFLICT(mirror_hash) DO NOTHING`, [mirrorHash, markdown] ); return ok(undefined); } catch (cause) { return err( "QUERY_FAILED", cause instanceof Error ? cause.message : "Failed to upsert content", cause ); } } async getContent(mirrorHash: string): Promise> { try { const db = this.ensureOpen(); const row = db .query<{ markdown: string }, [string]>( "SELECT markdown FROM content WHERE mirror_hash = ?" ) .get(mirrorHash); return ok(row?.markdown ?? null); } catch (cause) { return err( "QUERY_FAILED", cause instanceof Error ? cause.message : "Failed to get content", cause ); } } async getContentPrefix( mirrorHash: string, maxChars: number ): Promise> { try { const db = this.ensureOpen(); const boundedChars = Math.max(0, Math.floor(maxChars)); const row = db .query<{ markdown: string }, [number, string]>( "SELECT substr(markdown, 1, ?) AS markdown FROM content WHERE mirror_hash = ?" ) .get(boundedChars, mirrorHash); return ok(row?.markdown ?? null); } catch (cause) { return err( "QUERY_FAILED", cause instanceof Error ? cause.message : "Failed to get content prefix", cause ); } } async getContentBatch( mirrorHashes: string[] ): Promise>> { try { const uniqueHashes = uniqueNonEmptyValues(mirrorHashes); if (uniqueHashes.length === 0) { return ok(new Map()); } const db = this.ensureOpen(); interface DbContentRow { mirror_hash: string; markdown: string; } const result = new Map(); for ( let offset = 0; offset < uniqueHashes.length; offset += SQLITE_SAFE_PARAMETER_BATCH_SIZE ) { const batch = uniqueHashes.slice( offset, offset + SQLITE_SAFE_PARAMETER_BATCH_SIZE ); const placeholders = batch.map(() => "?").join(", "); const rows = db .query( `SELECT mirror_hash, markdown FROM content WHERE mirror_hash IN (${placeholders})` ) .all(...batch); for (const row of rows) result.set(row.mirror_hash, row.markdown); } return ok(result); } catch (cause) { return err( "QUERY_FAILED", cause instanceof Error ? cause.message : "Failed to get content batch", cause ); } } // ───────────────────────────────────────────────────────────────────────── // Chunks // ───────────────────────────────────────────────────────────────────────── async upsertChunks( mirrorHash: string, chunks: ChunkInput[], policy?: ChunkingPolicyToken ): Promise> { try { const db = this.ensureOpen(); const transaction = db.transaction(() => { assertChunkingTarget( db, policy ?? { params: DEFAULT_CHUNKING_PARAMS, generation: this.chunkingGeneration, } ); // Retain stable rows: DELETE cascades erase valid legacy vectors even // when duplicate ingestion produces exactly the same embedding input. const nextBySequence = new Map( chunks.map((chunk) => [chunk.seq, chunk]) ); const existing = db .query<{ seq: number; text: string }, [string]>( "SELECT seq, text FROM content_chunks WHERE mirror_hash = ?" ) .all(mirrorHash); for (const old of existing) { const next = nextBySequence.get(old.seq); if (!next || next.text !== old.text) { db.run( "DELETE FROM vector_owners WHERE mirror_hash = ? AND seq = ?", [mirrorHash, old.seq] ); db.run( "DELETE FROM content_chunks WHERE mirror_hash = ? AND seq = ?", [mirrorHash, old.seq] ); } } const stmt = db.prepare(` INSERT INTO content_chunks (mirror_hash, seq, pos, text, start_line, end_line, language, token_count) VALUES (?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT(mirror_hash, seq) DO UPDATE SET pos = excluded.pos, start_line = excluded.start_line, end_line = excluded.end_line, language = excluded.language, token_count = excluded.token_count WHERE content_chunks.pos IS NOT excluded.pos OR content_chunks.start_line IS NOT excluded.start_line OR content_chunks.end_line IS NOT excluded.end_line OR content_chunks.language IS NOT excluded.language OR content_chunks.token_count IS NOT excluded.token_count `); for (const chunk of chunks) { stmt.run( mirrorHash, chunk.seq, chunk.pos, chunk.text, chunk.startLine, chunk.endLine, chunk.language ?? null, chunk.tokenCount ?? null ); } }); transaction(); return ok(undefined); } catch (cause) { return err( cause instanceof ChunkingPolicyConflictError ? cause.code : "QUERY_FAILED", cause instanceof Error ? cause.message : "Failed to upsert chunks", cause ); } } async claimChunkingPolicy( input: ChunkingParams ): Promise> { const params = resolveChunkingParams(input); const result = await this.withTransaction(async () => claimChunkingTarget(this.ensureOpen(), this.chunkingGeneration, params) ); if (result.ok) this.chunkingGeneration = result.value.generation; else if (result.error.cause instanceof ChunkingPolicyConflictError) { return err( "CHUNKING_POLICY_CONFLICT", result.error.message, result.error.cause ); } return result; } async listPendingChunkingMirrors( policy: ChunkingPolicyToken, afterHash = "" ): Promise> { try { return ok(pendingChunkingMirrors(this.ensureOpen(), policy, afterHash)); } catch (cause) { return err( cause instanceof ChunkingPolicyConflictError ? cause.code : "QUERY_FAILED", cause instanceof Error ? cause.message : "Failed to list pending chunk layouts", cause ); } } async applyChunkLayout( mirrorHash: string, chunks: ChunkInput[], policy: ChunkingPolicyToken, sourcePath: string, languageHint?: string ): Promise> { const result = await this.withTransaction(async () => { const db = this.ensureOpen(); assertChunkingTarget(db, policy); const applied = await this.upsertChunks(mirrorHash, chunks, policy); if (!applied.ok) throw applied.error.cause ?? new Error(applied.error.message); const indexed = await this.rebuildFtsForHash(mirrorHash); if (!indexed.ok) throw indexed.error.cause ?? new Error(indexed.error.message); markChunkingApplied(db, mirrorHash, policy, sourcePath, languageHint); }); if ( !result.ok && result.error.cause instanceof ChunkingPolicyConflictError ) { return err( "CHUNKING_POLICY_CONFLICT", result.error.message, result.error.cause ); } return result; } async getChunks(mirrorHash: string): Promise> { try { const db = this.ensureOpen(); const rows = db .query( "SELECT * FROM content_chunks WHERE mirror_hash = ? ORDER BY seq" ) .all(mirrorHash); return ok(rows.map(mapChunkRow)); } catch (cause) { return err( "QUERY_FAILED", cause instanceof Error ? cause.message : "Failed to get chunks", cause ); } } async getChunksBatch( mirrorHashes: string[] ): Promise>> { try { // Early return for empty input if (mirrorHashes.length === 0) { return ok(new Map()); } // Dedupe and filter empty strings const uniqueHashes = uniqueNonEmptyValues(mirrorHashes); if (uniqueHashes.length === 0) { return ok(new Map()); } const db = this.ensureOpen(); const result = new Map(); // SQLite SQLITE_LIMIT_VARIABLE_NUMBER defaults to 999 // Reserve 99 for potential future filter params (collection, language, etc.) // Batch queries to respect SQLite parameter limit for ( let i = 0; i < uniqueHashes.length; i += SQLITE_SAFE_PARAMETER_BATCH_SIZE ) { const batch = uniqueHashes.slice( i, i + SQLITE_SAFE_PARAMETER_BATCH_SIZE ); const placeholders = batch.map(() => "?").join(","); const sql = `SELECT * FROM content_chunks WHERE mirror_hash IN (${placeholders}) ORDER BY mirror_hash, seq`; const rows = db.query(sql).all(...batch); // Group by mirrorHash, preserving seq order from ORDER BY for (const row of rows) { const mapped = mapChunkRow(row); const existing = result.get(mapped.mirrorHash) ?? []; existing.push(mapped); result.set(mapped.mirrorHash, existing); } } return ok(result); } catch (cause) { return err( "QUERY_FAILED", cause instanceof Error ? cause.message : "Failed to get chunks batch", cause ); } } async getChunksBySequenceBatch( keys: { mirrorHash: string; seq: number }[] ): Promise>> { try { const unique = new Map(); for (const key of keys) { if (key.mirrorHash) unique.set(JSON.stringify([key.mirrorHash, key.seq]), key); } const pairs = [...unique.values()]; const result = new Map(); if (pairs.length === 0) return ok(result); const db = this.ensureOpen(); const batchSize = Math.floor(SQLITE_SAFE_PARAMETER_BATCH_SIZE / 2); for (let offset = 0; offset < pairs.length; offset += batchSize) { const batch = pairs.slice(offset, offset + batchSize); const values = batch.map(() => "(?, ?)").join(","); const rows = db .query(` WITH requested(mirror_hash, seq) AS (VALUES ${values}) SELECT c.* FROM requested r JOIN content_chunks c ON c.mirror_hash = r.mirror_hash AND c.seq = r.seq ORDER BY c.mirror_hash, c.seq `) .all(...batch.flatMap(({ mirrorHash, seq }) => [mirrorHash, seq])); for (const row of rows) { const mapped = mapChunkRow(row); const chunks = result.get(mapped.mirrorHash) ?? []; chunks.push(mapped); result.set(mapped.mirrorHash, chunks); } } for (const chunks of result.values()) chunks.sort((a, b) => a.seq - b.seq); return ok(result); } catch (cause) { return err( "QUERY_FAILED", cause instanceof Error ? cause.message : "Failed to get targeted chunks batch", cause ); } } // ───────────────────────────────────────────────────────────────────────── // FTS Search // ───────────────────────────────────────────────────────────────────────── async searchFts( query: string, options: FtsSearchOptions = {} ): Promise> { try { const db = this.ensureOpen(); const limit = options.limit ?? 20; const builtQuery = buildFts5Query(query, { anyTerm: options.anyTerm }); if (!builtQuery.ok) { return err("INVALID_INPUT", builtQuery.error); } const eligible = buildEligibleDocumentQuery(options, db); // Document-level FTS search using an FTS-first CTE to keep collection and // metadata filters from degrading the query plan into a broad scan. const sql = ` WITH fts_matches AS ( SELECT rowid, ${options.snippet ? "snippet(documents_fts, 2, '', '', '...', 32) as snippet," : ""} bm25( documents_fts, ${FTS5_FIELD_WEIGHTS.filepath}, ${FTS5_FIELD_WEIGHTS.title}, ${FTS5_FIELD_WEIGHTS.body} ) as score FROM documents_fts WHERE documents_fts MATCH ? AND EXISTS ( SELECT 1 FROM (${eligible.sql}) eligible_docs WHERE eligible_docs.id = documents_fts.rowid ) ORDER BY score LIMIT ? ) SELECT d.mirror_hash, ${options.chunkLanguage ? "(SELECT min(lc.seq) FROM content_chunks lc WHERE lc.mirror_hash = d.mirror_hash AND lc.language = ?) as seq," : "0 as seq,"} fm.score as score, ${options.snippet ? "fm.snippet as snippet," : ""} d.docid, d.uri, d.title, d.collection, d.rel_path, d.source_mime, d.source_ext, d.source_mtime, d.frontmatter_date, d.source_size, d.source_hash, d.content_type, d.content_type_source, d.categories, d.converter_id, d.converter_version, d.record_key, d.record_source_path, d.record_source_locator, d.record_metadata, d.record_anchors, d.record_adapter_fingerprint FROM fts_matches fm JOIN documents d ON d.id = fm.rowid AND d.active = 1 WHERE 1 = 1 ORDER BY fm.score LIMIT ? `; interface FtsRow { mirror_hash: string; seq: number; score: number; snippet?: string; docid: string; uri: string; title: string | null; collection: string; rel_path: string; source_mime: string | null; source_ext: string | null; source_mtime: string | null; frontmatter_date: string | null; source_size: number | null; source_hash: string | null; content_type: string | null; content_type_source: string | null; categories: string | null; converter_id: string | null; converter_version: string | null; record_key: string | null; record_source_path: string | null; record_source_locator: string | null; record_metadata: string | null; record_anchors: string | null; record_adapter_fingerprint: string | null; } const queryParams = [ builtQuery.query, ...eligible.params, limit, ...(options.chunkLanguage ? [options.chunkLanguage] : []), limit, ]; const rows = db .query(sql) .all(...queryParams); return ok( rows.map((r) => ({ mirrorHash: r.mirror_hash, seq: r.seq, score: r.score, // Raw bm25() - smaller (more negative) is better snippet: r.snippet, docid: r.docid, uri: r.uri, title: r.title ?? undefined, collection: r.collection, relPath: r.rel_path, sourceMime: r.source_mime ?? undefined, sourceExt: r.source_ext ?? undefined, sourceMtime: r.source_mtime ?? undefined, frontmatterDate: r.frontmatter_date ?? undefined, sourceSize: r.source_size ?? undefined, sourceHash: r.source_hash ?? undefined, contentType: r.content_type ?? undefined, contentTypeSource: r.content_type_source ?? undefined, categories: parseCategoriesJson(r.categories) ?? undefined, converterId: r.converter_id ?? undefined, converterVersion: r.converter_version ?? undefined, recordKey: r.record_key ?? undefined, recordSourcePath: r.record_source_path ?? undefined, recordSourceLocator: r.record_source_locator ?? undefined, recordMetadata: parseRecordMetadataJson(r.record_metadata) ?? undefined, recordAnchors: parseRecordAnchorsJson(r.record_anchors) ?? undefined, recordAdapterFingerprint: r.record_adapter_fingerprint ?? undefined, })) ); } catch (cause) { const message = cause instanceof Error ? cause.message : ""; // Detect FTS5 syntax errors and return INVALID_INPUT for consistent handling const isSyntaxError = message.includes("malformed MATCH") || message.includes("fts5: syntax error") || message.includes("fts5:"); return err( isSyntaxError ? "INVALID_INPUT" : "QUERY_FAILED", message || "Failed to search FTS", cause ); } } /** * Sync a document to documents_fts for full-text search. * Must be called after document and content are both upserted. * The FTS rowid matches documents.id for efficient JOINs. */ async syncDocumentFts( collection: string, relPath: string ): Promise> { try { const db = this.ensureOpen(); const transaction = db.transaction(() => { // Get document with its content interface DocWithContent { id: number; rel_path: string; title: string | null; mirror_hash: string | null; markdown: string | null; } const doc = db .query( `SELECT d.id, d.rel_path, d.title, d.mirror_hash, c.markdown FROM documents d LEFT JOIN content c ON c.mirror_hash = d.mirror_hash WHERE d.collection = ? AND d.rel_path = ? AND d.active = 1` ) .get(collection, relPath); if (!doc) { return; // Document not found or inactive } // Delete existing FTS entry for this doc db.run("DELETE FROM documents_fts WHERE rowid = ?", [doc.id]); db.run("UPDATE documents SET fts_mirror_hash = NULL WHERE id = ?", [ doc.id, ]); // Insert new FTS entry if we have content if (doc.markdown !== null) { db.run( "INSERT INTO documents_fts (rowid, filepath, title, body) VALUES (?, ?, ?, ?)", [doc.id, doc.rel_path, doc.title ?? "", doc.markdown] ); db.run("UPDATE documents SET fts_mirror_hash = ? WHERE id = ?", [ doc.mirror_hash, doc.id, ]); } }); transaction(); return ok(undefined); } catch (cause) { return err( "QUERY_FAILED", cause instanceof Error ? cause.message : "Failed to sync document FTS", cause ); } } /** * Rebuild entire documents_fts index from scratch. * Use after migration or for recovery. */ async rebuildAllDocumentsFts(): Promise> { try { const db = this.ensureOpen(); let count = 0; const transaction = db.transaction(() => { // Clear FTS table db.run("DELETE FROM documents_fts"); db.run("UPDATE documents SET fts_mirror_hash = NULL"); // Get all active documents with content interface DocWithContent { id: number; rel_path: string; title: string | null; mirror_hash: string; markdown: string; } const docs = db .query( `SELECT d.id, d.rel_path, d.title, d.mirror_hash, c.markdown FROM documents d JOIN content c ON c.mirror_hash = d.mirror_hash WHERE d.active = 1 AND d.mirror_hash IS NOT NULL` ) .all(); // Insert FTS entries const stmt = db.prepare( "INSERT INTO documents_fts (rowid, filepath, title, body) VALUES (?, ?, ?, ?)" ); const markSynced = db.prepare( "UPDATE documents SET fts_mirror_hash = ? WHERE id = ?" ); for (const doc of docs) { stmt.run(doc.id, doc.rel_path, doc.title ?? "", doc.markdown); markSynced.run(doc.mirror_hash, doc.id); count++; } }); transaction(); return ok(count); } catch (cause) { return err( "QUERY_FAILED", cause instanceof Error ? cause.message : "Failed to rebuild documents FTS", cause ); } } /** * @deprecated Use syncDocumentFts for document-level FTS. * Kept for backwards compat during migration. */ async rebuildFtsForHash(mirrorHash: string): Promise> { try { const db = this.ensureOpen(); const transaction = db.transaction(() => { // Get documents using this hash and sync their FTS interface DocInfo { id: number; rel_path: string; title: string | null; } const docs = db .query( "SELECT id, rel_path, title FROM documents WHERE mirror_hash = ? AND active = 1" ) .all(mirrorHash); // Get content const content = db .query<{ markdown: string }, [string]>( "SELECT markdown FROM content WHERE mirror_hash = ?" ) .get(mirrorHash); if (!content) { return; } // Update FTS for each document using this hash for (const doc of docs) { db.run("DELETE FROM documents_fts WHERE rowid = ?", [doc.id]); db.run("UPDATE documents SET fts_mirror_hash = NULL WHERE id = ?", [ doc.id, ]); db.run( "INSERT INTO documents_fts (rowid, filepath, title, body) VALUES (?, ?, ?, ?)", [doc.id, doc.rel_path, doc.title ?? "", content.markdown] ); db.run("UPDATE documents SET fts_mirror_hash = ? WHERE id = ?", [ mirrorHash, doc.id, ]); } }); transaction(); return ok(undefined); } catch (cause) { return err( "QUERY_FAILED", cause instanceof Error ? cause.message : "Failed to rebuild FTS", cause ); } } // ───────────────────────────────────────────────────────────────────────── // Tags // ───────────────────────────────────────────────────────────────────────── /** * Set tags for a document. * Replaces tags from the given source (frontmatter or user). * User tags are never overwritten by frontmatter updates. */ async setDocTags( documentId: number, tags: string[], source: TagSource ): Promise> { try { const db = this.ensureOpen(); const transaction = db.transaction(() => { // Delete existing tags from this source db.run("DELETE FROM doc_tags WHERE document_id = ? AND source = ?", [ documentId, source, ]); // Insert new tags (skip duplicates from other source) if (tags.length > 0) { const stmt = db.prepare(` INSERT OR IGNORE INTO doc_tags (document_id, tag, source) VALUES (?, ?, ?) `); for (const tag of tags) { stmt.run(documentId, tag, source); } } }); transaction(); return ok(undefined); } catch (cause) { return err( "QUERY_FAILED", cause instanceof Error ? cause.message : "Failed to set document tags", cause ); } } async setDocMemoryScopes( documentId: number, scopes: string[] ): Promise> { try { const db = this.ensureOpen(); const transaction = db.transaction(() => { db.run("DELETE FROM doc_memory_scopes WHERE document_id = ?", [ documentId, ]); if (scopes.length > 0) { const stmt = db.prepare( "INSERT OR IGNORE INTO doc_memory_scopes (document_id, scope) VALUES (?, ?)" ); for (const scope of scopes) { stmt.run(documentId, scope); } } }); transaction(); return ok(undefined); } catch (cause) { return err( "QUERY_FAILED", cause instanceof Error ? cause.message : "Failed to set memory scopes", cause ); } } async getDocMemoryScopes(documentId: number): Promise> { try { const db = this.ensureOpen(); const rows = db .query<{ scope: string }, [number]>( "SELECT scope FROM doc_memory_scopes WHERE document_id = ? ORDER BY scope" ) .all(documentId); return ok(rows.map((row) => row.scope)); } catch (cause) { return err( "QUERY_FAILED", cause instanceof Error ? cause.message : "Failed to read memory scopes", cause ); } } async listMemoryEligibleDocuments( options: MemoryEligibleDocumentsOptions ): Promise> { try { const db = this.ensureOpen(); if (options.scopes.length === 0) { return ok([]); } const placeholders = options.scopes.map(() => "?").join(","); const rows = db .query< { id: number; docid: string; uri: string; mirror_hash: string }, string[] >( ` SELECT d.id, d.docid, d.uri, d.mirror_hash FROM documents d WHERE d.active = 1 AND d.collection = ? AND d.mirror_hash IS NOT NULL AND EXISTS (SELECT 1 FROM doc_memory_scopes ms WHERE ms.document_id = d.id AND ms.scope IN (${placeholders})) ${options.excludeSuperseded ? SUPERSEDED_EXCLUSION_SQL("d.id") : ""} ORDER BY d.id ` ) .all(options.collection, ...options.scopes); return ok( rows.map((row) => ({ id: row.id, docid: row.docid, uri: row.uri, mirrorHash: row.mirror_hash, })) ); } catch (cause) { return err( "QUERY_FAILED", cause instanceof Error ? cause.message : "Failed to list memory-eligible documents", cause ); } } /** * Get all tags for a document. */ async getTagsForDoc(documentId: number): Promise> { try { const db = this.ensureOpen(); interface DbTagRow { tag: string; source: "frontmatter" | "user"; } const rows = db .query( "SELECT tag, source FROM doc_tags WHERE document_id = ? ORDER BY tag" ) .all(documentId); return ok(rows); } catch (cause) { return err( "QUERY_FAILED", cause instanceof Error ? cause.message : "Failed to get tags for document", cause ); } } /** * Get tags for multiple documents in a single query. * Returns a map of documentId -> TagRow[]. */ async getTagsBatch( documentIds: number[] ): Promise>> { try { const db = this.ensureOpen(); if (documentIds.length === 0) { return ok(new Map()); } interface DbTagRow { document_id: number; tag: string; source: "frontmatter" | "user"; } // Use parameterized IN clause to prevent SQL injection const placeholders = documentIds.map(() => "?").join(", "); const rows = db .query( `SELECT document_id, tag, source FROM doc_tags WHERE document_id IN (${placeholders}) ORDER BY document_id, tag` ) .all(...documentIds); // Group by document_id const result = new Map(); for (const row of rows) { const existing = result.get(row.document_id) ?? []; existing.push({ tag: row.tag, source: row.source }); result.set(row.document_id, existing); } return ok(result); } catch (cause) { return err( "QUERY_FAILED", cause instanceof Error ? cause.message : "Failed to get tags batch", cause ); } } /** * Get tag counts across all active documents. * Optionally filter by collection or tag prefix. */ async getTagCounts(options?: { collection?: string; prefix?: string; }): Promise> { try { const db = this.ensureOpen(); const params: (string | number)[] = []; let sql = ` SELECT dt.tag, COUNT(DISTINCT dt.document_id) as count FROM doc_tags dt JOIN documents d ON d.id = dt.document_id AND d.active = 1 `; const conditions: string[] = []; if (options?.collection) { conditions.push("d.collection = ?"); params.push(options.collection); } if (options?.prefix) { // Normalize prefix: trim trailing slashes to avoid double-slash in LIKE pattern const normalizedPrefix = options.prefix.replace(/\/+$/, ""); // Match tags starting with prefix (for hierarchical browsing) // Escape LIKE metacharacters (%, _, \) in prefix const escapedPrefix = normalizedPrefix.replace(/[%_\\]/g, "\\$&"); conditions.push("(dt.tag = ? OR dt.tag LIKE ? ESCAPE '\\')"); params.push(normalizedPrefix, `${escapedPrefix}/%`); } if (conditions.length > 0) { sql += ` WHERE ${conditions.join(" AND ")}`; } sql += " GROUP BY dt.tag ORDER BY count DESC, dt.tag ASC"; interface DbTagCount { tag: string; count: number; } const rows = db .query(sql) .all(...params); return ok(rows); } catch (cause) { return err( "QUERY_FAILED", cause instanceof Error ? cause.message : "Failed to get tag counts", cause ); } } // ───────────────────────────────────────────────────────────────────────── // Links // ───────────────────────────────────────────────────────────────────────── /** * Set links for a document. * Replaces links from the given source (parsed, user, or suggested). */ async setDocLinks( documentId: number, links: DocLinkInput[], source: DocLinkSource ): Promise> { try { const db = this.ensureOpen(); const transaction = db.transaction(() => { // Link edits without a changed source identity cannot be located from // the prior document inventory, especially alongside unrelated changes. // Persist full-recovery authority in the same transaction as those edits. db.run( `UPDATE graph_projection_state SET dirty = 1, in_progress = 1 WHERE id = 1 AND EXISTS ( SELECT 1 FROM graph_reference_documents r JOIN documents d ON d.id = r.document_id WHERE d.id = ? AND r.source_hash IS d.source_hash AND r.mirror_hash IS d.mirror_hash )`, [documentId] ); // Delete existing links from this source db.run("DELETE FROM doc_links WHERE source_doc_id = ? AND source = ?", [ documentId, source, ]); // Insert new links if (links.length > 0) { const stmt = db.prepare(` INSERT INTO doc_links ( source_doc_id, target_ref, target_ref_norm, target_anchor, target_collection, link_type, link_text, start_line, start_col, end_line, end_col, source ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) `); for (const link of links) { // Normalize empty string targetCollection to NULL for consistent semantics // NULL = "same collection as source doc" const normalizedTargetCollection = link.targetCollection?.trim() ? link.targetCollection.trim() : null; stmt.run( documentId, link.targetRef, link.targetRefNorm, link.targetAnchor ?? null, normalizedTargetCollection, link.linkType, link.linkText ?? null, link.startLine, link.startCol, link.endLine, link.endCol, source ); } } }); transaction(); return ok(undefined); } catch (cause) { return err( "QUERY_FAILED", cause instanceof Error ? cause.message : "Failed to set document links", cause ); } } /** * Get all outgoing links for a document. */ async getLinksForDoc(documentId: number): Promise> { try { const db = this.ensureOpen(); interface DbDocLinkRow { target_ref: string; target_ref_norm: string; target_anchor: string | null; target_collection: string | null; link_type: "wiki" | "markdown"; link_text: string | null; start_line: number; start_col: number; end_line: number; end_col: number; source: "parsed" | "user" | "suggested"; } const rows = db .query( `SELECT target_ref, target_ref_norm, target_anchor, target_collection, link_type, link_text, start_line, start_col, end_line, end_col, source FROM doc_links WHERE source_doc_id = ? ORDER BY start_line, start_col` ) .all(documentId); return ok( rows.map((r) => ({ targetRef: r.target_ref, targetRefNorm: r.target_ref_norm, targetAnchor: r.target_anchor, targetCollection: r.target_collection, linkType: r.link_type, linkText: r.link_text, startLine: r.start_line, startCol: r.start_col, endLine: r.end_line, endCol: r.end_col, source: r.source, })) ); } catch (cause) { return err( "QUERY_FAILED", cause instanceof Error ? cause.message : "Failed to get links for document", cause ); } } /** * Get backlinks pointing to a document. * Uses target_ref_norm for matching (wiki=normalized title with path fallbacks, markdown=rel_path). * Only returns links from active source documents. */ async getBacklinksForDoc( documentId: number, options?: { collection?: string } ): Promise> { try { const db = this.ensureOpen(); // Get target document const target = db .query< { title: string | null; rel_path: string; collection: string }, [number] >( "SELECT title, rel_path, collection FROM documents WHERE id = ? AND active = 1" ) .get(documentId); if (!target) { return ok([]); } // Compute normalized wiki keys for fallback matching const keySet = new Set(); const addKey = (value: string): void => { if (value) { keySet.add(value); } }; const addVariants = (value: string): void => { if (!value) return; const base = stripWikiMdExt(value); const md = `${base}.md`; addKey(value); addKey(base); addKey(md); }; const addVariantsWithBasename = (value: string): void => { if (!value) return; addVariants(value); const basename = value.split("/").pop() ?? value; if (basename !== value) { addVariants(basename); } }; const titleKey = normalizeWikiName(target.title ?? ""); addVariants(titleKey); const relPathKey = normalizeWikiName(target.rel_path); addVariantsWithBasename(relPathKey); interface DbBacklinkRow { source_doc_id: number; docid: string; uri: string; title: string | null; link_text: string | null; start_line: number; start_col: number; } const targetCollection = target.collection; const sourceCollectionFilter = options?.collection; // Query wiki backlinks (link_type='wiki') with path-style fallbacks // NULL target_collection means "same collection as source" - enforce this in SQL const wikiConditions: string[] = []; const wikiParams: string[] = []; const addWikiExact = (value: string): void => { wikiConditions.push("dl.target_ref_norm = ?"); wikiParams.push(value); }; const addWikiSuffix = (value: string): void => { wikiConditions.push( `(substr(dl.target_ref_norm, -length(?)) = ? AND (length(dl.target_ref_norm) = length(?) OR substr(dl.target_ref_norm, -length(?) - 1, 1) = '/'))` ); wikiParams.push(value, value, value, value); }; for (const key of keySet) { addWikiExact(key); addWikiSuffix(key); } const wikiBacklinks = wikiConditions.length > 0 ? db .query( `SELECT dl.source_doc_id, src.docid, src.uri, src.title, dl.link_text, dl.start_line, dl.start_col FROM doc_links dl JOIN documents src ON src.id = dl.source_doc_id AND src.active = 1 WHERE dl.link_type = 'wiki' AND (${wikiConditions.join(" OR ")}) AND ( (dl.target_collection IS NULL AND src.collection = ?) OR dl.target_collection = ? ) ${sourceCollectionFilter ? "AND src.collection = ?" : ""} ORDER BY src.uri, dl.start_line, dl.start_col` ) .all( ...wikiParams, targetCollection, targetCollection, ...(sourceCollectionFilter ? [sourceCollectionFilter] : []) ) : []; // Query markdown backlinks (link_type='markdown') // NULL target_collection means "same collection as source" - enforce this in SQL const mdBacklinks = db .query( `SELECT dl.source_doc_id, src.docid, src.uri, src.title, dl.link_text, dl.start_line, dl.start_col FROM doc_links dl JOIN documents src ON src.id = dl.source_doc_id AND src.active = 1 WHERE dl.link_type = 'markdown' AND dl.target_ref_norm = ? AND ( (dl.target_collection IS NULL AND src.collection = ?) OR dl.target_collection = ? ) ${sourceCollectionFilter ? "AND src.collection = ?" : ""} ORDER BY src.uri, dl.start_line, dl.start_col` ) .all( target.rel_path, targetCollection, targetCollection, ...(sourceCollectionFilter ? [sourceCollectionFilter] : []) ); const allBacklinks = [...wikiBacklinks, ...mdBacklinks].map((r) => ({ sourceDocId: r.source_doc_id, sourceDocid: r.docid, sourceDocUri: r.uri, sourceDocTitle: r.title, linkText: r.link_text, startLine: r.start_line, startCol: r.start_col, })); return ok(allBacklinks); } catch (cause) { return err( "QUERY_FAILED", cause instanceof Error ? cause.message : "Failed to get backlinks for document", cause ); } } async resolveLinks( targets: Array<{ targetRefNorm: string; targetCollection: string; linkType: "wiki" | "markdown"; }> ): Promise< StoreResult< Array<{ docid: string; uri: string; title: string | null } | null> > > { try { const db = this.ensureOpen(); const results: Array<{ docid: string; uri: string; title: string | null; } | null> = Array.from({ length: targets.length }, () => null); const wikiTargets: Array<{ idx: number; collection: string; baseRef: string; baseRefMd: string; }> = []; const mdTargets: Array<{ idx: number; collection: string; relPath: string; }> = []; for (const [idx, target] of targets.entries()) { if (target.linkType === "wiki") { const baseRef = stripWikiMdExt(target.targetRefNorm); wikiTargets.push({ idx, collection: target.targetCollection, baseRef, baseRefMd: `${baseRef}.md`, }); } else { mdTargets.push({ idx, collection: target.targetCollection, relPath: target.targetRefNorm, }); } } const chunkArray = (items: T[], chunkSize: number): T[][] => { const chunks: T[][] = []; for (let i = 0; i < items.length; i += chunkSize) { chunks.push(items.slice(i, i + chunkSize)); } return chunks; }; const MAX_SQL_PARAMS = 900; const wikiBatchSize = Math.max(1, Math.floor(MAX_SQL_PARAMS / 4)); const mdBatchSize = Math.max(1, Math.floor(MAX_SQL_PARAMS / 3)); const titleExpr = "lower(trim(d.title))"; const relExpr = "lower(d.rel_path)"; const suffixMatchExprExpr = ( targetExpr: string, valueExpr: string ): string => `(substr(${targetExpr}, -length(${valueExpr})) = ${valueExpr} AND (length(${targetExpr}) = length(${valueExpr}) OR substr(${targetExpr}, -length(${valueExpr}) - 1, 1) = '/'))`; if (wikiTargets.length > 0) { for (const batch of chunkArray(wikiTargets, wikiBatchSize)) { const valuesClause = batch.map(() => "(?, ?, ?, ?)").join(", "); const wikiParams = batch.flatMap((t) => [ t.idx, t.collection, t.baseRef, t.baseRefMd, ]); const baseRefExpr = "t.base_ref"; const baseRefMdExpr = "t.base_ref_md"; const wikiWhere = ` ${titleExpr} = ${baseRefExpr} OR ${titleExpr} = ${baseRefMdExpr} OR ${suffixMatchExprExpr(baseRefExpr, titleExpr)} OR ${suffixMatchExprExpr(baseRefMdExpr, `${titleExpr} || '.md'`)} OR ${relExpr} = ${baseRefExpr} OR ${relExpr} = ${baseRefMdExpr} OR ${suffixMatchExprExpr(relExpr, baseRefMdExpr)} OR ${suffixMatchExprExpr(relExpr, baseRefExpr)} OR ${suffixMatchExprExpr(baseRefMdExpr, relExpr)} OR ${suffixMatchExprExpr(baseRefExpr, relExpr)} `; const wikiRank = `CASE WHEN ${titleExpr} = ${baseRefExpr} THEN 1 WHEN ${titleExpr} = ${baseRefMdExpr} THEN 2 WHEN ${suffixMatchExprExpr(baseRefExpr, titleExpr)} THEN 3 WHEN ${suffixMatchExprExpr( baseRefMdExpr, `${titleExpr} || '.md'` )} THEN 4 WHEN ${relExpr} = ${baseRefExpr} THEN 5 WHEN ${relExpr} = ${baseRefMdExpr} THEN 6 WHEN ${suffixMatchExprExpr(relExpr, baseRefMdExpr)} THEN 7 WHEN ${suffixMatchExprExpr(relExpr, baseRefExpr)} THEN 8 WHEN ${suffixMatchExprExpr(baseRefMdExpr, relExpr)} THEN 9 WHEN ${suffixMatchExprExpr(baseRefExpr, relExpr)} THEN 10 ELSE 99 END`; const wikiQuery = ` WITH targets(idx, collection, base_ref, base_ref_md) AS ( VALUES ${valuesClause} ), candidates AS ( SELECT t.idx, d.docid, d.uri, d.title, d.id as doc_id, ${wikiRank} as rank FROM targets t JOIN documents d ON d.active = 1 AND d.collection = t.collection WHERE ${wikiWhere} ), ranked AS ( SELECT *, ROW_NUMBER() OVER (PARTITION BY idx ORDER BY rank, doc_id) as rn FROM candidates ) SELECT idx, docid, uri, title FROM ranked WHERE rn = 1 `; const wikiRows = db .query< { idx: number; docid: string; uri: string; title: string | null }, (string | number)[] >(wikiQuery) .all(...wikiParams); for (const row of wikiRows) { results[row.idx] = { docid: row.docid, uri: row.uri, title: row.title, }; } } } if (mdTargets.length > 0) { for (const batch of chunkArray(mdTargets, mdBatchSize)) { const valuesClause = batch.map(() => "(?, ?, ?)").join(", "); const mdParams = batch.flatMap((t) => [ t.idx, t.collection, t.relPath, ]); const mdQuery = ` WITH targets(idx, collection, rel_path) AS ( VALUES ${valuesClause} ), ranked AS ( SELECT t.idx, d.docid, d.uri, d.title, d.id as doc_id, ROW_NUMBER() OVER (PARTITION BY t.idx ORDER BY d.id) as rn FROM targets t JOIN documents d ON d.active = 1 AND d.collection = t.collection AND d.rel_path = t.rel_path ) SELECT idx, docid, uri, title FROM ranked WHERE rn = 1 `; const mdRows = db .query< { idx: number; docid: string; uri: string; title: string | null }, (string | number)[] >(mdQuery) .all(...mdParams); for (const row of mdRows) { results[row.idx] = { docid: row.docid, uri: row.uri, title: row.title, }; } } } return ok(results); } catch (cause) { return err( "QUERY_FAILED", cause instanceof Error ? cause.message : "Failed to resolve links", cause ); } } /** * Bounded read seam for reference-safe rename/move planning. * Loads catalog metadata plus content for indexed backlinks unioned with a * conservative SQL content prefilter (opaque embeds/HTML/code/malformed may * not appear in doc_links). Missing mirror content fails closed. */ async getFileRefactorResolutionSnapshot(input: { sourceUri: string; maxCatalogDocuments?: number; maxReferrerDocuments?: number; maxContentCharsPerDocument?: number; maxTotalContentChars?: number; }): Promise> { try { const db = this.ensureOpen(); const maxCatalog = Math.max( 1, Math.floor(input.maxCatalogDocuments ?? 5_000) ); const maxReferrers = Math.max( 1, Math.floor(input.maxReferrerDocuments ?? 5_000) ); const maxChars = Math.max( 1, Math.floor(input.maxContentCharsPerDocument ?? 1_000_000) ); const maxTotalChars = Math.max( maxChars, Math.floor(input.maxTotalContentChars ?? 20_000_000) ); const truncationReasons: string[] = []; const sourceRow = db .query< { id: number; uri: string; rel_path: string; collection: string; title: string | null; mirror_hash: string | null; source_ext: string; source_mime: string; record_key: string | null; }, [string] >( `SELECT id, uri, rel_path, collection, title, mirror_hash, source_ext, source_mime, record_key FROM documents WHERE uri = ? AND active = 1` ) .get(input.sourceUri); if (!sourceRow) { return err("NOT_FOUND", `Document not found: ${input.sourceUri}`); } const sourceCaps = getDocumentCapabilities({ sourceExt: sourceRow.source_ext, sourceMime: sourceRow.source_mime, contentAvailable: Boolean(sourceRow.mirror_hash), recordKey: sourceRow.record_key, }); let sourceContent: string | null = null; let sourceContentTruncated = false; if (sourceRow.mirror_hash) { const contentRow = db .query<{ markdown: string }, [number, string]>( `SELECT substr(markdown, 1, ?) AS markdown FROM content WHERE mirror_hash = ?` ) .get(maxChars + 1, sourceRow.mirror_hash); if (contentRow) { sourceContentTruncated = contentRow.markdown.length > maxChars; sourceContent = sourceContentTruncated ? contentRow.markdown.slice(0, maxChars) : contentRow.markdown; if (sourceContentTruncated) { truncationReasons.push("source_content_truncated"); } } else { truncationReasons.push("source_content_missing"); } } else { // Null mirror_hash means source bytes are unavailable — fail closed. truncationReasons.push("source_content_missing"); } const catalogRows = db .query< { id: number; uri: string; rel_path: string; collection: string; title: string | null; }, [string, number] >( `SELECT id, uri, rel_path, collection, title FROM documents WHERE active = 1 AND collection = ? ORDER BY id LIMIT ?` ) .all(sourceRow.collection, maxCatalog + 1); if (catalogRows.length > maxCatalog) { truncationReasons.push("catalog_truncated"); } const catalog = catalogRows.slice(0, maxCatalog).map((row) => ({ id: row.id, uri: row.uri, relPath: row.rel_path, collection: row.collection, title: row.title, })); const occupiedRelPaths = catalog.map((doc) => doc.relPath); const backlinksResult = await this.getBacklinksForDoc(sourceRow.id, { collection: sourceRow.collection, }); if (!backlinksResult.ok) { return backlinksResult; } const candidateIds = new Set(); for (const row of backlinksResult.value) { if (row.sourceDocId !== sourceRow.id) { candidateIds.add(row.sourceDocId); } } const needles = buildContentPrefilterNeedles({ relPath: sourceRow.rel_path, title: sourceRow.title, }); const escapeLike = (value: string): string => value .replaceAll("\\", "\\\\") .replaceAll("%", "\\%") .replaceAll("_", "\\_"); if (needles.length > 0) { const likeClauses = needles .map(() => `c.markdown LIKE '%' || ? || '%' ESCAPE '\\'`) .join(" OR "); const prefilterRows = db .query<{ id: number }, (string | number)[]>( `SELECT DISTINCT d.id AS id FROM documents d INNER JOIN content c ON c.mirror_hash = d.mirror_hash WHERE d.active = 1 AND d.collection = ? AND d.id != ? AND (${likeClauses}) ORDER BY d.id LIMIT ?` ) .all( sourceRow.collection, sourceRow.id, ...needles.map(escapeLike), maxReferrers + 1 ); if (prefilterRows.length > maxReferrers) { truncationReasons.push("content_prefilter_truncated"); } for (const row of prefilterRows.slice(0, maxReferrers + 1)) { candidateIds.add(row.id); } } // Completeness: union same-collection docs whose mirror content cannot be // loaded. Text-like docs (including read-only logical .md records) fail // closed; non-text binaries are omitted (they cannot hold markdown/wiki/ // HTML refs). Preserve referrer caps via LIMIT + sorted slice below. const missingMirrorRows = db .query< { id: number; source_ext: string; source_mime: string; }, [string, number, number] >( `SELECT d.id AS id, d.source_ext AS source_ext, d.source_mime AS source_mime FROM documents d WHERE d.active = 1 AND d.collection = ? AND d.id != ? AND ( d.mirror_hash IS NULL OR NOT EXISTS ( SELECT 1 FROM content c WHERE c.mirror_hash = d.mirror_hash ) ) ORDER BY d.id LIMIT ?` ) .all(sourceRow.collection, sourceRow.id, maxReferrers + 1); if (missingMirrorRows.length > maxReferrers) { truncationReasons.push("missing_mirror_scan_truncated"); } for (const row of missingMirrorRows) { if (!isTextLikeReferenceDocument(row.source_ext, row.source_mime)) { continue; } candidateIds.add(row.id); } const sortedCandidateIds = [...candidateIds].sort((a, b) => a - b); if (sortedCandidateIds.length > maxReferrers) { truncationReasons.push("referrers_truncated"); } const referrers: FileRefactorResolutionReferrerDocument[] = []; let totalContentChars = sourceContent?.length ?? 0; for (const referrerId of sortedCandidateIds.slice(0, maxReferrers)) { const referrerRow = db .query< { id: number; uri: string; rel_path: string; collection: string; title: string | null; mirror_hash: string | null; source_ext: string; source_mime: string; record_key: string | null; }, [number] >( `SELECT id, uri, rel_path, collection, title, mirror_hash, source_ext, source_mime, record_key FROM documents WHERE id = ? AND active = 1` ) .get(referrerId); if (!referrerRow) continue; const caps = getDocumentCapabilities({ sourceExt: referrerRow.source_ext, sourceMime: referrerRow.source_mime, contentAvailable: Boolean(referrerRow.mirror_hash), recordKey: referrerRow.record_key, }); if (!referrerRow.mirror_hash) { truncationReasons.push("referrer_content_missing"); referrers.push({ id: referrerRow.id, uri: referrerRow.uri, relPath: referrerRow.rel_path, collection: referrerRow.collection, title: referrerRow.title, content: null, contentTruncated: false, contentMissing: true, editable: caps.editable, editableReason: caps.editable ? undefined : caps.reason ? "read_only_document" : "capability_denied", sourceExt: referrerRow.source_ext, sourceMime: referrerRow.source_mime, recordKey: referrerRow.record_key, }); continue; } const contentRow = db .query<{ markdown: string }, [number, string]>( `SELECT substr(markdown, 1, ?) AS markdown FROM content WHERE mirror_hash = ?` ) .get(maxChars + 1, referrerRow.mirror_hash); if (!contentRow) { truncationReasons.push("referrer_content_missing"); referrers.push({ id: referrerRow.id, uri: referrerRow.uri, relPath: referrerRow.rel_path, collection: referrerRow.collection, title: referrerRow.title, content: null, contentTruncated: false, contentMissing: true, editable: caps.editable, editableReason: caps.editable ? undefined : caps.reason ? "read_only_document" : "capability_denied", sourceExt: referrerRow.source_ext, sourceMime: referrerRow.source_mime, recordKey: referrerRow.record_key, }); continue; } const contentTruncated = contentRow.markdown.length > maxChars; if (contentTruncated) { truncationReasons.push("referrer_content_truncated"); } const content = contentTruncated ? contentRow.markdown.slice(0, maxChars) : contentRow.markdown; totalContentChars += content.length; if (totalContentChars > maxTotalChars) { truncationReasons.push("total_content_truncated"); referrers.push({ id: referrerRow.id, uri: referrerRow.uri, relPath: referrerRow.rel_path, collection: referrerRow.collection, title: referrerRow.title, content: null, contentTruncated: false, contentMissing: true, editable: caps.editable, editableReason: caps.editable ? undefined : caps.reason ? "read_only_document" : "capability_denied", sourceExt: referrerRow.source_ext, sourceMime: referrerRow.source_mime, recordKey: referrerRow.record_key, }); continue; } referrers.push({ id: referrerRow.id, uri: referrerRow.uri, relPath: referrerRow.rel_path, collection: referrerRow.collection, title: referrerRow.title, content, contentTruncated, contentMissing: false, editable: caps.editable, editableReason: caps.editable ? undefined : caps.reason ? "read_only_document" : "capability_denied", sourceExt: referrerRow.source_ext, sourceMime: referrerRow.source_mime, recordKey: referrerRow.record_key, }); } const snapshot: FileRefactorResolutionSnapshot = { source: { id: sourceRow.id, uri: sourceRow.uri, relPath: sourceRow.rel_path, collection: sourceRow.collection, title: sourceRow.title, mirrorHash: sourceRow.mirror_hash, sourceExt: sourceRow.source_ext, sourceMime: sourceRow.source_mime, recordKey: sourceRow.record_key, content: sourceContent, contentTruncated: sourceContentTruncated, editable: sourceCaps.editable, editableReason: sourceCaps.editable ? undefined : sourceCaps.reason ? "read_only_document" : "capability_denied", }, catalog, referrers, occupiedRelPaths, truncated: truncationReasons.length > 0, truncationReasons: [...new Set(truncationReasons)].sort(), }; return ok(snapshot); } catch (cause) { return err( "QUERY_FAILED", cause instanceof Error ? cause.message : "Failed to load file refactor resolution snapshot", cause ); } } /** * Set semantic edges for a document. * Replaces edges from the given source. */ async setDocEdges( documentId: number, edges: DocEdgeInput[], source: DocEdgeSource ): Promise> { try { const db = this.ensureOpen(); applyGraphEdges( db, edges.map((edge) => ({ sourceId: documentId, targetId: edge.targetDocId, edgeType: normalizeDocEdgeType(edge.edgeType), confidence: edge.confidence, source, })), [source], [documentId] ); return ok(undefined); } catch (cause) { return err( "QUERY_FAILED", cause instanceof Error ? cause.message : "Failed to set document edges", cause ); } } async getEdgesForDoc( documentId: number, options?: { edgeType?: DocEdgeType } ): Promise> { try { const db = this.ensureOpen(); const params: (number | string)[] = [documentId]; const edgeTypeClause = options?.edgeType ? "AND e.edge_type = ?" : ""; if (options?.edgeType) { params.push(normalizeDocEdgeType(options.edgeType)); } const rows = db .query( ` WITH ranked AS ( SELECT e.src_doc_id, src.docid AS source_docid, src.uri AS source_uri, src.title AS source_title, e.dst_doc_id, dst.docid AS target_docid, dst.uri AS target_uri, dst.title AS target_title, e.edge_type, e.confidence, e.source, ROW_NUMBER() OVER ( PARTITION BY e.src_doc_id, e.dst_doc_id, e.edge_type ORDER BY CASE e.confidence WHEN 'manual' THEN 1 WHEN 'configured' THEN 2 WHEN 'parsed' THEN 3 WHEN 'inferred' THEN 4 ELSE 5 END, e.source ASC, dst.docid ASC, dst.uri ASC ) AS rn FROM doc_edges e JOIN documents src ON src.id = e.src_doc_id AND src.active = 1 JOIN documents dst ON dst.id = e.dst_doc_id AND dst.active = 1 WHERE e.src_doc_id = ? ${edgeTypeClause} ) SELECT * FROM ranked WHERE rn = 1 ORDER BY edge_type ASC, target_uri ASC, target_docid ASC ` ) .all(...params); return ok(rows.map(mapDocEdgeRow)); } catch (cause) { return err( "QUERY_FAILED", cause instanceof Error ? cause.message : "Failed to get edges for document", cause ); } } async getEdgeBacklinksForDoc( documentId: number, options?: { collection?: string; edgeType?: DocEdgeType } ): Promise> { try { const db = this.ensureOpen(); const params: (number | string)[] = [documentId]; const conditions: string[] = ["e.dst_doc_id = ?"]; if (options?.collection) { conditions.push("src.collection = ?"); params.push(options.collection); } if (options?.edgeType) { conditions.push("e.edge_type = ?"); params.push(normalizeDocEdgeType(options.edgeType)); } const rows = db .query( ` WITH ranked AS ( SELECT e.src_doc_id, src.docid AS source_docid, src.uri AS source_uri, src.title AS source_title, e.dst_doc_id, dst.docid AS target_docid, dst.uri AS target_uri, dst.title AS target_title, e.edge_type, e.confidence, e.source, ROW_NUMBER() OVER ( PARTITION BY e.src_doc_id, e.dst_doc_id, e.edge_type ORDER BY CASE e.confidence WHEN 'manual' THEN 1 WHEN 'configured' THEN 2 WHEN 'parsed' THEN 3 WHEN 'inferred' THEN 4 ELSE 5 END, e.source ASC, src.docid ASC, src.uri ASC ) AS rn FROM doc_edges e JOIN documents src ON src.id = e.src_doc_id AND src.active = 1 JOIN documents dst ON dst.id = e.dst_doc_id AND dst.active = 1 WHERE ${conditions.join(" AND ")} ) SELECT * FROM ranked WHERE rn = 1 ORDER BY edge_type ASC, source_uri ASC, source_docid ASC ` ) .all(...params); return ok(rows.map(mapDocEdgeRow)); } catch (cause) { return err( "QUERY_FAILED", cause instanceof Error ? cause.message : "Failed to get edge backlinks for document", cause ); } } async queryGraphTraversal( rootDocumentId: number, options: GraphQueryOptions = {} ): Promise> { try { const db = this.ensureOpen(); const direction = options.direction ?? "both"; const maxDepth = Math.max(1, Math.min(options.maxDepth ?? 2, 6)); const maxNodes = Math.max(1, Math.min(options.maxNodes ?? 100, 1_000)); const frontierLimit = Math.max( 1, Math.min(options.frontierLimit ?? 100, 1_000) ); const visitedLimit = Math.max( 1, Math.min(options.visitedLimit ?? 500, 5_000) ); const edgeType = options.edgeType ? normalizeDocEdgeType(options.edgeType) : undefined; const nodeLimit = Math.min(maxNodes, visitedLimit); const edgeTypeFilter = edgeType ? "AND e.edge_type = ?" : ""; const edgeTypeFilterFor = (alias: string): string => edgeType ? `AND ${alias}.edge_type = ?` : ""; const candidateEdgeLimit = Math.min(frontierLimit * 4, 4_000); const frontierCtes: string[] = []; const frontierParams: (number | string)[] = []; const frontierNames = ["f0"]; const allFrontiers = (): string => frontierNames .map((name) => `SELECT doc_id FROM ${name}`) .join(" UNION "); const frontierJoinClause = (frontierName: string): string => direction === "out" ? `e.src_doc_id = ${frontierName}.doc_id` : direction === "in" ? `e.dst_doc_id = ${frontierName}.doc_id` : `(e.src_doc_id = ${frontierName}.doc_id OR e.dst_doc_id = ${frontierName}.doc_id)`; const boundedEdgePredicate = (frontierName: string): string => { if (direction === "out") { return ` e.id IN ( SELECT edge_id FROM ( SELECT edge_id, edge_type, next_doc_id FROM ( SELECT e2.id AS edge_id, e2.edge_type, e2.dst_doc_id AS next_doc_id, row_number() OVER ( PARTITION BY e2.dst_doc_id ORDER BY e2.edge_type ASC, e2.id ASC ) AS next_rank FROM doc_edges e2 JOIN documents next_doc ON next_doc.id = e2.dst_doc_id AND next_doc.active = 1 WHERE e2.src_doc_id = ${frontierName}.doc_id ${edgeTypeFilterFor("e2")} AND instr(${frontierName}.path, printf(',%d,', e2.dst_doc_id)) = 0 AND e2.dst_doc_id NOT IN (${allFrontiers()}) ) WHERE next_rank = 1 ORDER BY edge_type ASC, next_doc_id ASC, edge_id ASC LIMIT ? ) )`; } if (direction === "in") { return ` e.id IN ( SELECT edge_id FROM ( SELECT edge_id, edge_type, next_doc_id FROM ( SELECT e2.id AS edge_id, e2.edge_type, e2.src_doc_id AS next_doc_id, row_number() OVER ( PARTITION BY e2.src_doc_id ORDER BY e2.edge_type ASC, e2.id ASC ) AS next_rank FROM doc_edges e2 JOIN documents next_doc ON next_doc.id = e2.src_doc_id AND next_doc.active = 1 WHERE e2.dst_doc_id = ${frontierName}.doc_id ${edgeTypeFilterFor("e2")} AND instr(${frontierName}.path, printf(',%d,', e2.src_doc_id)) = 0 AND e2.src_doc_id NOT IN (${allFrontiers()}) ) WHERE next_rank = 1 ORDER BY edge_type ASC, next_doc_id ASC, edge_id ASC LIMIT ? ) )`; } return ` ( e.id IN ( SELECT edge_id FROM ( SELECT edge_id, edge_type, next_doc_id FROM ( SELECT e2.id AS edge_id, e2.edge_type, e2.dst_doc_id AS next_doc_id, row_number() OVER ( PARTITION BY e2.dst_doc_id ORDER BY e2.edge_type ASC, e2.id ASC ) AS next_rank FROM doc_edges e2 JOIN documents next_doc ON next_doc.id = e2.dst_doc_id AND next_doc.active = 1 WHERE e2.src_doc_id = ${frontierName}.doc_id ${edgeTypeFilterFor("e2")} AND instr(${frontierName}.path, printf(',%d,', e2.dst_doc_id)) = 0 AND e2.dst_doc_id NOT IN (${allFrontiers()}) ) WHERE next_rank = 1 ORDER BY edge_type ASC, next_doc_id ASC, edge_id ASC LIMIT ? ) ) OR e.id IN ( SELECT edge_id FROM ( SELECT edge_id, edge_type, next_doc_id FROM ( SELECT e3.id AS edge_id, e3.edge_type, e3.src_doc_id AS next_doc_id, row_number() OVER ( PARTITION BY e3.src_doc_id ORDER BY e3.edge_type ASC, e3.id ASC ) AS next_rank FROM doc_edges e3 JOIN documents next_doc ON next_doc.id = e3.src_doc_id AND next_doc.active = 1 WHERE e3.dst_doc_id = ${frontierName}.doc_id ${edgeTypeFilterFor("e3")} AND instr(${frontierName}.path, printf(',%d,', e3.src_doc_id)) = 0 AND e3.src_doc_id NOT IN (${allFrontiers()}) ) WHERE next_rank = 1 ORDER BY edge_type ASC, next_doc_id ASC, edge_id ASC LIMIT ? ) ) )`; }; const nextExprFor = (frontierName: string): string => `CASE WHEN e.src_doc_id = ${frontierName}.doc_id THEN e.dst_doc_id ELSE e.src_doc_id END`; const boundaryCandidateDepth = maxDepth + 1; for (let depth = 1; depth <= boundaryCandidateDepth; depth += 1) { const previous = `f${depth - 1}`; const raw = `c${depth}_raw`; const deduped = `c${depth}_deduped`; const nodeDeduped = `c${depth}_node_deduped`; const ranked = `c${depth}_ranked`; const candidates = `c${depth}`; const frontier = `f${depth}`; const nextExpr = nextExprFor(previous); frontierCtes.push(` ${raw} AS ( SELECT ${previous}.doc_id AS from_doc_id, ${nextExpr} AS doc_id, ${depth} AS depth, ${previous}.path || ${nextExpr} || ',' AS path, printf('%06d|%s|%012d', ${depth}, e.edge_type, ${nextExpr}) AS sort_key, e.src_doc_id, src.docid AS source_docid, src.uri AS source_uri, src.title AS source_title, e.dst_doc_id, dst.docid AS target_docid, dst.uri AS target_uri, dst.title AS target_title, e.edge_type, e.confidence, e.source, row_number() OVER ( PARTITION BY ${previous}.doc_id, e.src_doc_id, e.dst_doc_id, e.edge_type ORDER BY CASE e.confidence WHEN 'manual' THEN 1 WHEN 'configured' THEN 2 WHEN 'parsed' THEN 3 WHEN 'inferred' THEN 4 ELSE 5 END, e.source ASC, src.docid ASC, dst.docid ASC ) AS dedup_rank FROM ${previous} JOIN doc_edges e ON ${frontierJoinClause(previous)} AND ${boundedEdgePredicate(previous)} JOIN documents src ON src.id = e.src_doc_id AND src.active = 1 JOIN documents dst ON dst.id = e.dst_doc_id AND dst.active = 1 WHERE 1 = 1 ${edgeTypeFilter} AND instr(${previous}.path, printf(',%d,', ${nextExpr})) = 0 AND ${nextExpr} NOT IN (${allFrontiers()}) ), ${deduped} AS ( SELECT * FROM ${raw} WHERE dedup_rank = 1 ), ${nodeDeduped} AS ( SELECT from_doc_id, doc_id, depth, min(path) AS path, min(sort_key) AS sort_key FROM ${deduped} GROUP BY from_doc_id, doc_id, depth ), ${ranked} AS ( SELECT *, row_number() OVER ( PARTITION BY from_doc_id ORDER BY sort_key ASC, doc_id ASC ) AS expansion_rank FROM ${nodeDeduped} ), ${candidates} AS ( SELECT doc_id, depth, min(path) AS path, min(sort_key) AS sort_key FROM ${ranked} WHERE expansion_rank <= ? GROUP BY doc_id, depth ), ${frontier} AS ( SELECT doc_id, depth, path, sort_key FROM ( SELECT doc_id, depth, path, sort_key, row_number() OVER (ORDER BY sort_key ASC, doc_id ASC) AS depth_rank FROM ${candidates} ) WHERE depth_rank <= ? AND depth_rank <= ? - ( SELECT count(*) FROM (${allFrontiers()}) ) )`); if (direction === "both") { if (edgeType) { frontierParams.push(edgeType); } frontierParams.push(candidateEdgeLimit); if (edgeType) { frontierParams.push(edgeType); } frontierParams.push(candidateEdgeLimit); } else { if (edgeType) { frontierParams.push(edgeType); } frontierParams.push(candidateEdgeLimit); } if (edgeType) { frontierParams.push(edgeType); } frontierParams.push(frontierLimit, frontierLimit, visitedLimit); if (depth <= maxDepth) { frontierNames.push(frontier); } } const returnedEdgeDirectionClause = direction === "out" ? "AND ns.depth < nt.depth" : direction === "in" ? "AND ns.depth > nt.depth" : ""; const frontierUnion = frontierNames .map((name) => `SELECT doc_id, depth, sort_key FROM ${name}`) .join(" UNION ALL "); const visitedOverflowChecks = Array.from( { length: maxDepth }, (_, index) => { const previousFrontiers = Array.from( { length: index + 1 }, (__, frontierIndex) => `SELECT doc_id FROM f${frontierIndex}` ).join(" UNION "); return ` SELECT 1 AS has_more FROM c${index + 1} GROUP BY 1 HAVING count(*) > ? - ( SELECT count(*) FROM (${previousFrontiers}) )`; } ).join(" UNION ALL "); const frontierOverflowChecks = Array.from( { length: maxDepth }, (_, index) => ` SELECT 1 AS has_more FROM c${index + 1} GROUP BY 1 HAVING count(*) > ? UNION ALL SELECT 1 AS has_more FROM c${index + 1}_ranked WHERE expansion_rank > ?` ).join(" UNION ALL "); const walkCte = ` WITH RECURSIVE f0(doc_id, depth, path, sort_key) AS ( SELECT ?, 0, printf(',%d,', ?), '' ), ${frontierCtes.join(",")}, node_depth AS ( SELECT doc_id, min(depth) AS depth, min(sort_key) AS sort_key FROM (${frontierUnion}) GROUP BY doc_id ), ranked_nodes AS ( SELECT d.*, nd.depth, row_number() OVER (ORDER BY nd.depth ASC, nd.sort_key ASC, d.id ASC) AS global_rank, count(*) OVER () AS total_count FROM node_depth nd JOIN documents d ON d.id = nd.doc_id AND d.active = 1 ) `; const baseParams: (number | string)[] = []; baseParams.push(rootDocumentId, rootDocumentId, ...frontierParams); const nodeRows = db .query< DbDocumentRow & { depth: number; global_rank: number; total_count: number; }, (number | string)[] >( `${walkCte} SELECT * FROM ranked_nodes WHERE global_rank <= ? ORDER BY depth ASC, uri ASC, docid ASC LIMIT ?` ) .all(...baseParams, nodeLimit + 1, nodeLimit + 1); const returnedNodeRows = nodeRows.slice(0, nodeLimit); const returnedIds = new Set(returnedNodeRows.map((row) => row.id)); const totalNodes = nodeRows[0]?.total_count ?? 0; const warnings: string[] = []; let truncated = false; if (totalNodes > maxNodes) { truncated = true; warnings.push("maxNodes reached"); } if (totalNodes > visitedLimit) { truncated = true; warnings.push("visitedLimit reached"); } const visitedRows = db .query<{ has_more: number }, (number | string)[]>( `${walkCte} SELECT 1 AS has_more FROM (${visitedOverflowChecks}) LIMIT 1` ) .get(...baseParams, ...Array(maxDepth).fill(visitedLimit)); if (visitedRows) { truncated = true; warnings.push("visitedLimit reached"); } const frontierRows = db .query<{ has_more: number }, (number | string)[]>( `${walkCte} SELECT 1 AS has_more FROM (${frontierOverflowChecks}) LIMIT 1` ) .get(...baseParams, ...Array(maxDepth * 2).fill(frontierLimit)); if (frontierRows) { truncated = true; warnings.push("frontierLimit reached"); } const edgeRows = db .query( `${walkCte} , returned_edges AS ( SELECT e.src_doc_id, src.docid AS source_docid, src.uri AS source_uri, src.title AS source_title, e.dst_doc_id, dst.docid AS target_docid, dst.uri AS target_uri, dst.title AS target_title, e.edge_type, e.confidence, e.source, CASE WHEN ns.depth >= nt.depth THEN ns.depth ELSE nt.depth END AS traversal_depth, row_number() OVER ( PARTITION BY e.src_doc_id, e.dst_doc_id, e.edge_type ORDER BY CASE e.confidence WHEN 'manual' THEN 1 WHEN 'configured' THEN 2 WHEN 'parsed' THEN 3 WHEN 'inferred' THEN 4 ELSE 5 END, e.source ASC, src.docid ASC, dst.docid ASC ) AS dedup_rank FROM doc_edges e JOIN node_depth ns ON ns.doc_id = e.src_doc_id JOIN node_depth nt ON nt.doc_id = e.dst_doc_id JOIN documents src ON src.id = e.src_doc_id AND src.active = 1 JOIN documents dst ON dst.id = e.dst_doc_id AND dst.active = 1 WHERE ns.doc_id IN (${[...returnedIds].map(() => "?").join(",")}) AND nt.doc_id IN (${[...returnedIds].map(() => "?").join(",")}) ${edgeTypeFilter} ${returnedEdgeDirectionClause} ) SELECT * FROM returned_edges WHERE dedup_rank = 1 ORDER BY traversal_depth ASC, edge_type ASC, source_docid ASC, target_docid ASC` ) .all( ...baseParams, ...returnedIds, ...returnedIds, ...(edgeType ? [edgeType] : []) ); const boundaryRows = db .query<{ has_more: number }, (number | string)[]>( `${walkCte} SELECT 1 AS has_more FROM c${boundaryCandidateDepth}_ranked LIMIT 1` ) .get(...baseParams); if (boundaryRows) { truncated = true; warnings.push("maxDepth reached"); } return ok({ nodes: returnedNodeRows.map((row) => ({ doc: mapDocumentRow(row), depth: row.depth, })), edges: edgeRows.map((row) => ({ edge: mapDocEdgeRow(row), depth: row.traversal_depth, })), truncated, warnings: [...new Set(warnings)], }); } catch (cause) { return err( "QUERY_FAILED", cause instanceof Error ? cause.message : "Failed to run graph traversal", cause ); } } async backfillDocEdges( sourceDocumentIds?: number[] ): Promise> { try { const db = this.ensureOpen(); const sourceIds = sourceDocumentIds ? [...new Set(sourceDocumentIds)].filter((id) => id > 0) : undefined; if (sourceIds?.length === 0) { return ok({ inserted: 0 }); } const sourceFilter = sourceIds ? "AND src.id IN (SELECT value FROM json_each(?))" : ""; const params = sourceIds ? [JSON.stringify(sourceIds)] : []; const inserted = db.transaction(() => { const wiki = db .query(` SELECT DISTINCT src.id AS sourceId, tgt.id AS targetId, 'mentions' AS edgeType, 'parsed' AS confidence, 'wikilink' AS source FROM documents src JOIN doc_links dl ON dl.source_doc_id = src.id JOIN documents tgt ON tgt.id = (${buildWikiBestMatchSubquery( "COALESCE(dl.target_collection, src.collection)", "dl.target_ref_norm" )}) WHERE src.active = 1 AND tgt.active = 1 AND dl.link_type = 'wiki' ${sourceFilter} `) .all(...params); const markdown = db .query(` SELECT DISTINCT src.id AS sourceId, tgt.id AS targetId, 'related_to' AS edgeType, 'parsed' AS confidence, 'markdown-link' AS source FROM documents src JOIN doc_links dl ON dl.source_doc_id = src.id JOIN documents tgt ON tgt.active = 1 AND tgt.collection = COALESCE(dl.target_collection, src.collection) AND tgt.rel_path = dl.target_ref_norm WHERE src.active = 1 AND dl.link_type = 'markdown' ${sourceFilter} `) .all(...params); return applyGraphEdges( db, [...wiki, ...markdown], ["wikilink", "markdown-link"], sourceIds ); })(); return ok({ inserted }); } catch (cause) { return err( "QUERY_FAILED", cause instanceof Error ? cause.message : "Failed to backfill document edges", cause ); } } // ───────────────────────────────────────────────────────────────────────── // Graph // ───────────────────────────────────────────────────────────────────────── /** * Seed-scoped one-hop neighbors for query-time graph expansion. * Does not rebuild the full collection graph or similarity edges. */ async getGraphNeighborsForSeeds( options: GetGraphNeighborsOptions ): Promise> { try { const db = this.ensureOpen(); return ok(queryGraphNeighborsForSeeds(db, options)); } catch (cause) { return err( "QUERY_FAILED", cause instanceof Error ? cause.message : "Failed to get seed graph neighbors", cause ); } } async getGraph(options?: GetGraphOptions): Promise> { try { const db = this.ensureOpen(); // Apply defaults const collection = options?.collection ?? null; // Clamp all limits defensively (store is last line of defense) const limitNodes = Math.max( 1, Math.min(5000, options?.limitNodes ?? 2000) ); const limitEdges = Math.max( 1, Math.min(50000, options?.limitEdges ?? 10000) ); const includeSimilar = options?.includeSimilar ?? false; const threshold = Math.max(0, Math.min(1, options?.threshold ?? 0.7)); const linkedOnly = options?.linkedOnly ?? true; const similarTopK = Math.max(1, Math.min(20, options?.similarTopK ?? 5)); const warnings: string[] = []; // Always probe sqlite-vec availability (not just when similarity requested) let similarAvailable = false; try { db.query("SELECT vec_version()").get(); similarAvailable = true; } catch { // sqlite-vec not loaded } interface ResolvedEdgeRow { source_id: number; source_docid: string; target_id: number; target_docid: string; link_type: "wiki" | "markdown"; match_rank: number | null; match_count: number | null; } interface GraphLinkResolutionRow { source_id: number; source_docid: string; source_collection: string; target_ref_norm: string; target_collection: string | null; link_type: "wiki" | "markdown"; } interface NodeMetaRow { id: number; docid: string; uri: string; title: string | null; collection: string; rel_path: string; } const linkParams: string[] = []; let sourceCollectionClause = ""; if (collection) { sourceCollectionClause = "AND src.collection = ?"; linkParams.push(collection); } const graphLinkRows = db .query( ` SELECT src.id as source_id, src.docid as source_docid, src.collection as source_collection, dl.target_ref_norm, dl.target_collection, dl.link_type FROM documents src JOIN doc_links dl ON dl.source_doc_id = src.id WHERE src.active = 1 ${sourceCollectionClause} ORDER BY src.id ASC, dl.id ASC ` ) .all(...linkParams); const resolutions = resolveGraphLinkTargets( db, graphLinkRows.map((row) => ({ targetRefNorm: row.target_ref_norm, targetCollection: row.target_collection ?? row.source_collection, linkType: row.link_type, })) ); const resolvedEdgeRows: ResolvedEdgeRow[] = []; const unresolvedByType: Record<"wiki" | "markdown", number> = { wiki: 0, markdown: 0, }; for (const [index, row] of graphLinkRows.entries()) { const resolution = resolutions[index]; if (!resolution) { unresolvedByType[row.link_type] += 1; continue; } const targetCollection = row.target_collection ?? row.source_collection; if (collection && targetCollection !== collection) continue; resolvedEdgeRows.push({ source_id: row.source_id, source_docid: row.source_docid, target_id: resolution.targetId, target_docid: resolution.targetDocid, link_type: row.link_type, match_rank: resolution.matchRank, match_count: resolution.matchCount, }); } resolvedEdgeRows.sort( (left, right) => left.source_id - right.source_id || left.target_id - right.target_id || left.link_type.localeCompare(right.link_type) ); const totalEdgesUnresolved = unresolvedByType.wiki + unresolvedByType.markdown; const outNeighbors = new Map>(); const inNeighbors = new Map>(); const connectedNodeIds = new Set(); for (const row of resolvedEdgeRows) { connectedNodeIds.add(row.source_id); connectedNodeIds.add(row.target_id); const sourceSet = outNeighbors.get(row.source_id) ?? new Set(); sourceSet.add(row.target_id); outNeighbors.set(row.source_id, sourceSet); const targetSet = inNeighbors.get(row.target_id) ?? new Set(); targetSet.add(row.source_id); inNeighbors.set(row.target_id, targetSet); } const connectedIdList = [...connectedNodeIds].sort((a, b) => a - b); const connectedMetaMap = new Map(); if (connectedIdList.length > 0) { const placeholders = connectedIdList.map(() => "?").join(","); const connectedRows = db .query( `SELECT id, docid, uri, title, collection, rel_path FROM documents WHERE id IN (${placeholders})` ) .all(...connectedIdList); for (const row of connectedRows) { connectedMetaMap.set(row.id, row); } } const connectedNodes = connectedIdList .map((id) => { const meta = connectedMetaMap.get(id); if (!meta) { return null; } return { ...meta, degree: (outNeighbors.get(id)?.size ?? 0) + (inNeighbors.get(id)?.size ?? 0), }; }) .filter((row): row is NodeMetaRow & { degree: number } => row !== null) .sort((a, b) => b.degree - a.degree || a.id - b.id); const toReportNode = ( row: NodeMetaRow & { degree: number } ): GraphReportNode => ({ id: row.docid, uri: row.uri, title: row.title, collection: row.collection, relPath: row.rel_path, degree: row.degree, }); const isolatedBaseParams: (string | number)[] = []; const isolatedBaseConditions = ["active = 1"]; if (collection) { isolatedBaseConditions.push("collection = ?"); isolatedBaseParams.push(collection); } if (connectedIdList.length > 0) { const placeholders = connectedIdList.map(() => "?").join(","); isolatedBaseConditions.push(`id NOT IN (${placeholders})`); isolatedBaseParams.push(...connectedIdList); } const isolatedWhereClause = isolatedBaseConditions.join(" AND "); const isolatedCountRow = db .query<{ cnt: number }, (string | number)[]>( `SELECT COUNT(*) as cnt FROM documents WHERE ${isolatedWhereClause}` ) .get(...isolatedBaseParams); const isolatedTotal = isolatedCountRow?.cnt ?? 0; const isolatedExampleRows = db .query( `SELECT id, docid, uri, title, collection, rel_path FROM documents WHERE ${isolatedWhereClause} ORDER BY id ASC LIMIT 10` ) .all(...isolatedBaseParams) .map((row) => ({ ...row, degree: 0 })); let totalNodes = linkedOnly ? connectedNodes.length : 0; if (!linkedOnly) { const countParams: string[] = []; let countClause = ""; if (collection) { countClause = "WHERE active = 1 AND collection = ?"; countParams.push(collection); } else { countClause = "WHERE active = 1"; } const countRow = db .query<{ cnt: number }, string[]>( `SELECT COUNT(*) as cnt FROM documents ${countClause}` ) .get(...countParams); totalNodes = countRow?.cnt ?? connectedNodes.length; } const selectedNodeRows = [...connectedNodes]; if (!linkedOnly && selectedNodeRows.length < limitNodes) { const remaining = limitNodes - selectedNodeRows.length; const excluded = selectedNodeRows.map((row) => row.id); const isolatedParams: (string | number)[] = []; const conditions = ["active = 1"]; if (collection) { conditions.push("collection = ?"); isolatedParams.push(collection); } if (excluded.length > 0) { const placeholders = excluded.map(() => "?").join(","); conditions.push(`id NOT IN (${placeholders})`); isolatedParams.push(...excluded); } isolatedParams.push(remaining); const isolatedRows = db .query( `SELECT id, docid, uri, title, collection, rel_path FROM documents WHERE ${conditions.join(" AND ")} ORDER BY id ASC LIMIT ?` ) .all(...isolatedParams) .map((row) => ({ ...row, degree: 0 })); selectedNodeRows.push(...isolatedRows); } const nodes = selectedNodeRows.slice(0, limitNodes).map((row) => ({ id: row.docid, uri: row.uri, title: row.title, collection: row.collection, relPath: row.rel_path, degree: row.degree, })); const truncatedNodes = totalNodes > nodes.length; const selectedDocids = new Set(nodes.map((node) => node.id)); const nodeDocids = new Set(nodes.map((node) => node.id)); const edgeMap = new Map< string, { type: GraphLinkType; weight: number; confidence: GraphEdgeConfidence; audit: GraphEdgeAudit; } >(); for (const row of resolvedEdgeRows) { if ( !selectedDocids.has(row.source_docid) || !selectedDocids.has(row.target_docid) ) { continue; } const key = `${row.source_docid}:${row.target_docid}:${row.link_type}`; const { confidence, audit } = classifyResolvedGraphEdge( row.link_type, row.match_rank, row.match_count ); const existing = edgeMap.get(key); if (existing) { existing.weight += 1; mergeGraphEdgeAudit(existing, confidence, audit); } else { edgeMap.set(key, { type: row.link_type, weight: 1, confidence, audit, }); } } // Phase 3: Similarity edges (if requested) let similarTruncatedByComputeBudget = false; if (includeSimilar && nodeDocids.size > 0 && similarAvailable) { // Cap similarity work to avoid blocking the server event loop const SIMILARITY_NODE_CAP = 200; const nodesForSimilarity = [...nodeDocids].slice( 0, SIMILARITY_NODE_CAP ); if (nodeDocids.size > SIMILARITY_NODE_CAP) { similarTruncatedByComputeBudget = true; warnings.push( `Similarity capped at ${SIMILARITY_NODE_CAP} nodes (requested ${nodeDocids.size})` ); } // Track if any similarity queries fail let similarityFailures = 0; const mirrorByDocid = new Map(); if (nodesForSimilarity.length > 0) { const placeholders = nodesForSimilarity.map(() => "?").join(","); const mirrorRows = db .query<{ docid: string; mirror_hash: string }, string[]>( `SELECT docid, mirror_hash FROM documents WHERE active = 1 AND docid IN (${placeholders})` ) .all(...nodesForSimilarity); for (const row of mirrorRows) { if (row.mirror_hash) { mirrorByDocid.set(row.docid, row.mirror_hash); } } } const allowedMirrorHashes = [...mirrorByDocid.values()]; if (allowedMirrorHashes.length === 0) { warnings.push("Similarity unavailable: no embedded nodes in graph"); } const allowedPlaceholders = allowedMirrorHashes .map(() => "?") .join(","); // Get kNN for each node // Query content_vectors for embedded chunks, find similar for (const docid of nodesForSimilarity) { if (allowedMirrorHashes.length === 0) break; const mirrorHash = mirrorByDocid.get(docid); if (!mirrorHash) continue; // Find similar docs using vec_distance, aggregate by doc to get max score interface SimilarRow { target_docid: string; score: number; } // Use GROUP BY to get one best score per doc (avoids duplicate rows from multi-chunk docs) const similarQuery = ` SELECT d.docid as target_docid, MAX(1 - vec_distance_cosine(v1.embedding, v2.embedding)) as score FROM content_vectors v1 JOIN content_vectors v2 ON v2.model = v1.model AND v2.mirror_hash != v1.mirror_hash AND v2.seq = 0 JOIN documents d ON d.mirror_hash = v2.mirror_hash AND d.active = 1 WHERE v1.mirror_hash = ? AND v1.seq = 0 AND d.docid != ? AND v2.mirror_hash IN (${allowedPlaceholders}) GROUP BY d.docid HAVING score >= ? ORDER BY score DESC LIMIT ? `; try { const similarRows = db .query(similarQuery) .all( mirrorHash, docid, ...allowedMirrorHashes, threshold, similarTopK ); for (const sim of similarRows) { if (!nodeDocids.has(sim.target_docid)) continue; // Clamp score to [0, 1] for schema compliance const clampedScore = Math.max(0, Math.min(1, sim.score)); // Canonicalize by lexicographic order (undirected edge) const [a, b] = docid < sim.target_docid ? [docid, sim.target_docid] : [sim.target_docid, docid]; const key = `${a}:${b}:similar`; // Keep max score const existing = edgeMap.get(key); if (!existing || clampedScore > existing.weight) { edgeMap.set(key, { type: "similar", weight: clampedScore, confidence: "similarity", audit: { resolution: "similarity", score: clampedScore }, }); } } } catch { similarityFailures++; } } // Report partial failures if (similarityFailures > 0) { warnings.push( `Similarity query failed for ${similarityFailures} nodes; results may be incomplete` ); } } else if (includeSimilar && !similarAvailable) { warnings.push("Similarity edges unavailable: sqlite-vec not loaded"); } // Convert edge map to array, apply limit const allEdges = [...edgeMap.entries()].map(([key, val]) => { const parts = key.split(":"); return { source: parts[0] ?? "", target: parts[1] ?? "", type: val.type, weight: val.weight, confidence: val.confidence, audit: val.audit, }; }); const edgeTypes: Record = { wiki: 0, markdown: 0, similar: 0, }; for (const edge of allEdges) { edgeTypes[edge.type] += 1; } const edgeConfidence: Record = { explicit: 0, inferred: 0, ambiguous: 0, similarity: 0, }; for (const edge of allEdges) { edgeConfidence[edge.confidence] += 1; } const communityAnalysis = analyzeGraphCommunities(nodes, allEdges); warnings.push(...communityAnalysis.warnings); const nodesWithCommunities = nodes.map((node) => { const communityId = communityAnalysis.assignments[node.id]; return communityId ? { ...node, communityId } : node; }); const communityByNodeId = new Map( Object.entries(communityAnalysis.assignments) ); const truncatedEdges = allEdges.length > limitEdges; const links = allEdges.slice(0, limitEdges); // Add truncation warnings if (truncatedNodes) { warnings.push(`Nodes truncated: ${totalNodes} → ${limitNodes}`); } if (truncatedEdges) { warnings.push(`Edges truncated: ${allEdges.length} → ${limitEdges}`); } return ok({ nodes: nodesWithCommunities, links, report: { hubs: connectedNodes.slice(0, 10).map((node) => ({ ...toReportNode(node), communityId: communityByNodeId.get(node.docid), })), bridgeCandidates: connectedNodes .filter( (row) => (inNeighbors.get(row.id)?.size ?? 0) > 0 && (outNeighbors.get(row.id)?.size ?? 0) > 0 ) .slice(0, 10) .map((node) => ({ ...toReportNode(node), communityId: communityByNodeId.get(node.docid), })), isolated: { total: isolatedTotal, examples: isolatedExampleRows.map((node) => ({ ...toReportNode(node), communityId: communityByNodeId.get(node.docid), })), }, unresolvedLinks: { total: totalEdgesUnresolved, byType: unresolvedByType, }, edgeTypes, edgeConfidence, audit: { inferredEdges: edgeConfidence.inferred, ambiguousEdges: edgeConfidence.ambiguous, similarityEdges: edgeConfidence.similarity, }, communities: { total: communityAnalysis.total, algorithm: communityAnalysis.algorithm, skipped: communityAnalysis.skipped, assignments: communityAnalysis.assignments, top: communityAnalysis.communities, }, }, meta: { collection, nodeLimit: limitNodes, edgeLimit: limitEdges, totalNodes, // totalEdges = collapsed edge count within selected nodes (matches allEdges) totalEdges: allEdges.length, totalEdgesUnresolved, returnedNodes: nodesWithCommunities.length, returnedEdges: links.length, truncated: truncatedNodes || truncatedEdges, linkedOnly, includedSimilar: includeSimilar && similarAvailable, similarAvailable, similarTopK, similarTruncatedByComputeBudget, warnings, }, }); } catch (cause) { return err( "QUERY_FAILED", cause instanceof Error ? cause.message : "Failed to get graph", cause ); } } // ───────────────────────────────────────────────────────────────────────── // Status // ───────────────────────────────────────────────────────────────────────── async getStatus(options?: { embedModel?: string; embedFingerprint?: string; chunking?: Partial; }): Promise> { try { const db = this.ensureOpen(); const embedModel = options?.embedModel ?? null; const embedFingerprint = options?.embedFingerprint ?? (embedModel ? getStoredEmbeddingFingerprint(db, embedModel) : null); const variantStatus = getVariantStatus(db, options); // Get version const versionRow = db .query<{ value: string }, []>( "SELECT value FROM schema_meta WHERE key = 'version'" ) .get(); const version = versionRow?.value ?? "0"; // Derive indexName from dbPath (basename without extension) const indexName = basename(this.dbPath) .replace(SQLITE_EXT_REGEX, "") .replace(INDEX_PREFIX_REGEX, "") || "default"; // Get collection stats with chunk counts interface CollectionStat { name: string; path: string; egress_policy: EgressPolicy; egress_policy_source: EgressPolicySource; total: number; active: number; errored: number; chunked: number; chunk_count: number; embedded_count: number; } const collectionStats = db .query( ` WITH document_stats AS ( SELECT collection, COUNT(*) AS total, SUM(CASE WHEN active = 1 THEN 1 ELSE 0 END) AS active, SUM(CASE WHEN last_error_code IS NOT NULL THEN 1 ELSE 0 END) AS errored, SUM(CASE WHEN mirror_hash IS NOT NULL THEN 1 ELSE 0 END) AS chunked FROM documents GROUP BY collection ), active_collection_mirrors AS ( SELECT DISTINCT collection, mirror_hash FROM documents WHERE active = 1 AND mirror_hash IS NOT NULL ), matching_vectors AS ( SELECT mirror_hash, seq, MAX(embedded_at) AS embedded_at FROM content_vectors WHERE (? IS NULL OR ( model = ? AND embed_fingerprint = ? )) GROUP BY mirror_hash, seq ), collection_chunks AS ( SELECT acm.collection, COUNT(*) AS chunk_count, SUM(CASE WHEN mv.embedded_at >= cc.created_at THEN 1 ELSE 0 END) AS embedded_count FROM active_collection_mirrors acm JOIN content_chunks cc ON cc.mirror_hash = acm.mirror_hash LEFT JOIN matching_vectors mv ON mv.mirror_hash = cc.mirror_hash AND mv.seq = cc.seq GROUP BY acm.collection ) SELECT c.name, c.path, c.egress_policy, c.egress_policy_source, COALESCE(ds.total, 0) AS total, COALESCE(ds.active, 0) AS active, COALESCE(ds.errored, 0) AS errored, COALESCE(ds.chunked, 0) AS chunked, COALESCE(ch.chunk_count, 0) AS chunk_count, COALESCE(ch.embedded_count, 0) AS embedded_count FROM collections c LEFT JOIN document_stats ds ON ds.collection = c.name LEFT JOIN collection_chunks ch ON ch.collection = c.name ORDER BY c.name ` ) .all(embedModel, embedModel, embedFingerprint); // Get totals const totalsRow = db .query<{ total: number; active: number }, []>( ` SELECT COUNT(*) as total, SUM(CASE WHEN active = 1 THEN 1 ELSE 0 END) as active FROM documents ` ) .get(); const chunkCount = db .query<{ count: number }, []>( "SELECT COUNT(*) as count FROM content_chunks" ) .get()?.count ?? 0; // Embedding backlog: chunks from active docs without vectors // Uses EXISTS to avoid duplicates when multiple docs share mirror_hash const backlogRow = db .query< { count: number }, [string | null, string | null, string | null] >( ` WITH active_mirrors AS ( SELECT DISTINCT mirror_hash FROM documents WHERE active = 1 AND mirror_hash IS NOT NULL ), matching_vectors AS ( SELECT mirror_hash, seq, MAX(embedded_at) AS embedded_at FROM content_vectors WHERE (? IS NULL OR ( model = ? AND embed_fingerprint = ? )) GROUP BY mirror_hash, seq ) SELECT COUNT(*) AS count FROM active_mirrors am JOIN content_chunks c ON c.mirror_hash = am.mirror_hash LEFT JOIN matching_vectors mv ON mv.mirror_hash = c.mirror_hash AND mv.seq = c.seq AND mv.embedded_at >= c.created_at WHERE mv.mirror_hash IS NULL ` ) .get(embedModel, embedModel, embedFingerprint); // Recent errors (last 24h) const recentErrorsRow = db .query<{ count: number }, []>( ` SELECT COUNT(*) as count FROM ingest_errors WHERE occurred_at > datetime('now', '-1 day') ` ) .get(); // Last updated (max updated_at from documents) const lastUpdatedRow = db .query<{ last_updated: string | null }, []>( "SELECT strftime('%Y-%m-%dT%H:%M:%fZ', MAX(updated_at)) as last_updated FROM documents" ) .get(); // Health check: no recent errors and DB is accessible const recentErrors = recentErrorsRow?.count ?? 0; const healthy = recentErrors === 0; const metadataCoverage = await this.getTypedMetadataCoverage({}); if (!metadataCoverage.ok) return metadataCoverage; return ok({ typedMetadata: metadataCoverage.value, version, indexName, configPath: this.configPath, dbPath: this.dbPath, ftsTokenizer: this.ftsTokenizer, collections: collectionStats.map((s) => ({ name: s.name, path: s.path, egressPolicy: s.egress_policy, egressPolicySource: s.egress_policy_source, totalDocuments: s.total, activeDocuments: s.active, errorDocuments: s.errored, chunkedDocuments: s.chunked, totalChunks: s.chunk_count, embeddedChunks: variantStatus ? (variantStatus.embeddedByCollection.get(s.name) ?? 0) : s.embedded_count, })), totalDocuments: totalsRow?.total ?? 0, activeDocuments: totalsRow?.active ?? 0, totalChunks: chunkCount, embeddingBacklog: variantStatus?.backlog ?? backlogRow?.count ?? 0, chunking: getChunkingStatus(db, options?.chunking), recentErrors, lastUpdatedAt: lastUpdatedRow?.last_updated ?? null, healthy, }); } catch (cause) { return err( "QUERY_FAILED", cause instanceof Error ? cause.message : "Failed to get status", cause ); } } // ───────────────────────────────────────────────────────────────────────── // Errors // ───────────────────────────────────────────────────────────────────────── async recordError(error: IngestErrorInput): Promise> { try { const db = this.ensureOpen(); db.run( `INSERT INTO ingest_errors (collection, rel_path, code, message, details_json) VALUES (?, ?, ?, ?, ?)`, [ error.collection, error.relPath, error.code, error.message, error.details ? JSON.stringify(error.details) : null, ] ); return ok(undefined); } catch (cause) { return err( "QUERY_FAILED", cause instanceof Error ? cause.message : "Failed to record error", cause ); } } async getRecentErrors(limit = 50): Promise> { try { const db = this.ensureOpen(); const rows = db .query( "SELECT * FROM ingest_errors ORDER BY occurred_at DESC LIMIT ?" ) .all(limit); return ok(rows.map(mapIngestErrorRow)); } catch (cause) { return err( "QUERY_FAILED", cause instanceof Error ? cause.message : "Failed to get recent errors", cause ); } } // ───────────────────────────────────────────────────────────────────────── // Cleanup // ───────────────────────────────────────────────────────────────────────── async cleanupOrphans(): Promise> { try { const db = this.ensureOpen(); let orphanedContent = 0; let orphanedChunks = 0; let orphanedVectors = 0; let expiredCache = 0; const transaction = db.transaction(() => { // Delete content not referenced by any active document const contentResult = db.run(` DELETE FROM content WHERE mirror_hash NOT IN ( SELECT DISTINCT mirror_hash FROM documents WHERE mirror_hash IS NOT NULL AND active = 1 ) `); orphanedContent = contentResult.changes; pruneChunkingMetadata(db); // Delete chunks for deleted content const chunksResult = db.run(` DELETE FROM content_chunks WHERE mirror_hash NOT IN ( SELECT mirror_hash FROM content ) `); orphanedChunks = chunksResult.changes; // Delete vectors for deleted chunks const vectorsResult = db.run(` DELETE FROM content_vectors WHERE (mirror_hash, seq) NOT IN ( SELECT mirror_hash, seq FROM content_chunks ) `); orphanedVectors = vectorsResult.changes; // Delete expired cache entries const cacheResult = db.run(` DELETE FROM llm_cache WHERE expires_at IS NOT NULL AND expires_at < datetime('now') `); expiredCache = cacheResult.changes; // Clean orphaned FTS entries (documents that no longer exist or are inactive) db.run(` DELETE FROM documents_fts WHERE rowid NOT IN ( SELECT id FROM documents WHERE active = 1 ) `); }); transaction(); return ok({ orphanedContent, orphanedChunks, orphanedVectors, expiredCache, }); } catch (cause) { return err( "QUERY_FAILED", cause instanceof Error ? cause.message : "Failed to cleanup orphans", cause ); } } async clearEmbeddingsForCollection( collection: string, options: { mode: "stale" | "all"; activeModel?: string } ): Promise> { try { const db = this.ensureOpen(); const collectionName = collection.toLowerCase(); if (options.mode === "stale" && !options.activeModel) { return err( "INVALID_INPUT", "activeModel is required for stale embedding cleanup" ); } const filterSql = options.mode === "stale" ? "AND cv.model != ?" : ""; const filterParams = options.mode === "stale" ? [options.activeModel ?? ""] : []; const deletableRows = db .query<{ mirror_hash: string; model: string; seq: number }, string[]>( ` SELECT DISTINCT cv.mirror_hash, cv.seq, cv.model FROM content_vectors cv WHERE EXISTS ( SELECT 1 FROM documents d WHERE d.collection = ? AND d.mirror_hash = cv.mirror_hash ) ${filterSql} AND NOT EXISTS ( SELECT 1 FROM documents d2 WHERE d2.mirror_hash = cv.mirror_hash AND d2.collection != ? AND d2.active = 1 ) ` ) .all(collectionName, ...filterParams, collectionName); const protectedRow = db .query<{ count: number }, string[]>( ` SELECT COUNT(*) as count FROM ( SELECT DISTINCT cv.mirror_hash, cv.seq, cv.model FROM content_vectors cv WHERE EXISTS ( SELECT 1 FROM documents d WHERE d.collection = ? AND d.mirror_hash = cv.mirror_hash ) ${filterSql} AND EXISTS ( SELECT 1 FROM documents d2 WHERE d2.mirror_hash = cv.mirror_hash AND d2.collection != ? AND d2.active = 1 ) ) ` ) .get(collectionName, ...filterParams, collectionName); const deletedModels = [...new Set(deletableRows.map((row) => row.model))]; const transaction = db.transaction(() => { const deleteVectorStmt = db.prepare( `DELETE FROM content_vectors WHERE mirror_hash = ? AND seq = ? AND model = ?` ); const vecDeleteStatements = new Map< string, ReturnType | null >(); for (const row of deletableRows) { deleteVectorStmt.run(row.mirror_hash, row.seq, row.model); let deleteVecStmt = vecDeleteStatements.get(row.model); if (deleteVecStmt === undefined) { try { deleteVecStmt = db.prepare( `DELETE FROM ${modelTableName(row.model)} WHERE chunk_id = ?` ); } catch { deleteVecStmt = null; } vecDeleteStatements.set(row.model, deleteVecStmt); } if (deleteVecStmt) { try { deleteVecStmt.run(`${row.mirror_hash}:${row.seq}`); } catch { // Best effort; a later vec sync/rebuild can recover. } } } }); transaction(); return ok({ collection: collectionName, deletedVectors: deletableRows.length, deletedModels, mode: options.mode, protectedSharedVectors: protectedRow?.count ?? 0, }); } catch (cause) { return err( "QUERY_FAILED", cause instanceof Error ? cause.message : "Failed to clear collection embeddings", cause ); } } } // ───────────────────────────────────────────────────────────────────────────── // DB Row Types (snake_case from SQLite) // ───────────────────────────────────────────────────────────────────────────── interface DbCollectionRow { name: string; path: string; pattern: string; include: string | null; exclude: string | null; update_cmd: string | null; language_hint: string | null; egress_policy: EgressPolicy; egress_policy_source: EgressPolicySource; synced_at: string; } interface DbContextRow { scope_type: "global" | "collection" | "prefix"; scope_key: string; text: string; synced_at: string; } interface DbDocumentRow { id: number; collection: string; rel_path: string; source_hash: string; source_mime: string; source_ext: string; source_size: number; source_mtime: string; source_ctime: string | null; docid: string; uri: string; title: string | null; mirror_hash: string | null; converter_id: string | null; converter_version: string | null; language_hint: string | null; content_type: string | null; content_type_source: string | null; categories: string | null; author: string | null; frontmatter_date: string | null; date_fields: string | null; typed_metadata: string | null; metadata_error: string | null; record_key: string | null; record_source_path: string | null; record_source_locator: string | null; record_metadata: string | null; record_anchors: string | null; record_adapter_fingerprint: string | null; content_type_rules_fingerprint: string | null; indexed_at: string | null; active: number; ingest_version: number | null; last_error_code: string | null; last_error_message: string | null; last_error_at: string | null; created_at: string; updated_at: string; } interface DbDocEdgeRow { src_doc_id: number; source_docid: string; source_uri: string; source_title: string | null; dst_doc_id: number; target_docid: string; target_uri: string; target_title: string | null; edge_type: string; confidence: DocEdgeConfidence; source: DocEdgeSource; } interface DbChunkRow { mirror_hash: string; seq: number; pos: number; text: string; start_line: number; end_line: number; language: string | null; token_count: number | null; created_at: string; } interface DbIngestErrorRow { id: number; collection: string; rel_path: string; occurred_at: string; code: string; message: string; details_json: string | null; } // Row Mappers (snake_case -> camelCase) // ───────────────────────────────────────────────────────────────────────────── function mapCollectionRow(row: DbCollectionRow): CollectionRow { return { name: row.name, path: row.path, pattern: row.pattern, include: row.include ? JSON.parse(row.include) : null, exclude: row.exclude ? JSON.parse(row.exclude) : null, updateCmd: row.update_cmd, languageHint: row.language_hint, egressPolicy: row.egress_policy, egressPolicySource: row.egress_policy_source, syncedAt: row.synced_at, }; } function mapContextRow(row: DbContextRow): ContextRow { return { scopeType: row.scope_type, scopeKey: row.scope_key, text: row.text, syncedAt: row.synced_at, }; } function parseCategoriesJson(raw: string | null): string[] | null { if (!raw) { return null; } try { const parsed = JSON.parse(raw); if (Array.isArray(parsed)) { return parsed.filter((v): v is string => typeof v === "string"); } } catch { return null; } return null; } function parseRecordMetadataJson( raw: string | null ): DocumentRow["recordMetadata"] { if (!raw) return null; try { const parsed: unknown = JSON.parse(raw); return parsed && typeof parsed === "object" && !Array.isArray(parsed) ? (parsed as NonNullable) : null; } catch { return null; } } function parseRecordAnchorsJson( raw: string | null ): DocumentRow["recordAnchors"] { if (!raw) return null; try { const parsed: unknown = JSON.parse(raw); return Array.isArray(parsed) ? (parsed as NonNullable) : null; } catch { return null; } } function mapDocumentRow(row: DbDocumentRow): DocumentRow { const categories = parseCategoriesJson(row.categories); let dateFields: Record | null = null; if (row.date_fields) { try { const parsed = JSON.parse(row.date_fields); if (parsed && typeof parsed === "object" && !Array.isArray(parsed)) { const normalized: Record = {}; for (const [key, value] of Object.entries(parsed)) { if (typeof value === "string") { normalized[key] = value; } } if (Object.keys(normalized).length > 0) { dateFields = normalized; } } } catch { dateFields = null; } } return { id: row.id, collection: row.collection, relPath: row.rel_path, sourceHash: row.source_hash, sourceMime: row.source_mime, sourceExt: row.source_ext, sourceSize: row.source_size, sourceMtime: row.source_mtime, sourceCtime: row.source_ctime, docid: row.docid, uri: row.uri, title: row.title, mirrorHash: row.mirror_hash, converterId: row.converter_id, converterVersion: row.converter_version, languageHint: row.language_hint, contentType: row.content_type, contentTypeSource: row.content_type_source, categories, author: row.author, frontmatterDate: row.frontmatter_date, dateFields, typedMetadata: row.typed_metadata ? typedMetadataSchema.parse(JSON.parse(row.typed_metadata)) : null, metadataError: row.metadata_error, recordKey: row.record_key, recordSourcePath: row.record_source_path, recordSourceLocator: row.record_source_locator, recordMetadata: parseRecordMetadataJson(row.record_metadata), recordAnchors: parseRecordAnchorsJson(row.record_anchors), recordAdapterFingerprint: row.record_adapter_fingerprint, indexedAt: row.indexed_at, active: row.active === 1, ingestVersion: row.ingest_version, contentTypeRulesFingerprint: row.content_type_rules_fingerprint, lastErrorCode: row.last_error_code, lastErrorMessage: row.last_error_message, lastErrorAt: row.last_error_at, createdAt: row.created_at, updatedAt: row.updated_at, }; } function mapDocEdgeRow(row: DbDocEdgeRow): DocEdgeRow { return { sourceDocId: row.src_doc_id, sourceDocid: row.source_docid, sourceUri: row.source_uri, sourceTitle: row.source_title, targetDocId: row.dst_doc_id, targetDocid: row.target_docid, targetUri: row.target_uri, targetTitle: row.target_title, edgeType: row.edge_type, relationType: row.edge_type, confidence: row.confidence, edgeSource: row.source, }; } function mapChunkRow(row: DbChunkRow): ChunkRow { return { mirrorHash: row.mirror_hash, seq: row.seq, pos: row.pos, text: row.text, startLine: row.start_line, endLine: row.end_line, language: row.language, tokenCount: row.token_count, createdAt: row.created_at, }; } function mapIngestErrorRow(row: DbIngestErrorRow): IngestErrorRow { return { id: row.id, collection: row.collection, relPath: row.rel_path, occurredAt: row.occurred_at, code: row.code, message: row.message, detailsJson: row.details_json, }; }