ANG-008:Similarity computation, edge creation & SQLite cache

src/data/Database/VectorDatabase.ts

@@ -0,0 +1,97 @@
+import joplin from 'api';
+
+export interface IVectorDatabase {
+	open(): Promise<void>;
+	run(sql: string, params: unknown[]): Promise<void>;
+	all<T>(sql: string, params: unknown[]): Promise<T[]>;
+}
+
+interface Sqlite3Database {
+	run(sql: string, params: unknown[], callback: (err: Error | null) => void): void;
+	all(
+		sql: string,
+		params: unknown[],
+		callback: (err: Error | null, rows: unknown[]) => void
+	): void;
+	close(callback?: (err: Error | null) => void): void;
+}
+
+/**
+ * Thin promisified wrapper around Joplin's bundled sqlite3 module (accessed via
+ * `joplin.require('sqlite3')`, since native packages can't be bundled with a
+ * plugin). Owns only the connection and schema; query logic lives in
+ * VectorRepository.
+ */
+export class VectorDatabase implements IVectorDatabase {
+	private static readonly DB_FILE_NAME = 'note-graph-vectors.sqlite';
+	private static readonly SCHEMA = `
+		CREATE TABLE IF NOT EXISTS note_vectors (
+			note_id TEXT PRIMARY KEY,
+			model_id TEXT NOT NULL,
+			updated_time INTEGER NOT NULL,
+			vector BLOB NOT NULL
+		)
+	`;
+
+	private db: Sqlite3Database | null = null;
+	private opening: Promise<void> | null = null;
+
+	/**
+	 * Opens (creating if needed) the vector cache database. Safe to call
+	 * repeatedly. A failed open is not cached: both `opening` and `db` are
+	 * reset on rejection so a later call can retry from scratch, instead of
+	 * either re-awaiting the same stale rejection or (if the connection
+	 * itself succeeded but schema creation failed) treating a half-open
+	 * database as ready forever.
+	 */
+	public async open(): Promise<void> {
+		if (this.db) return;
+		if (!this.opening) {
+			this.opening = this.openInternal().catch((e) => {
+				this.opening = null;
+				if (this.db) {
+					this.db.close();
+					this.db = null;
+				}
+				throw e;
+			});
+		}
+		await this.opening;
+	}
+
+	public async run(sql: string, params: unknown[]): Promise<void> {
+		const db = this.requireDb();
+		await new Promise<void>((resolve, reject) => {
+			db.run(sql, params, (err) => (err ? reject(err) : resolve()));
+		});
+	}
+
+	public async all<T>(sql: string, params: unknown[]): Promise<T[]> {
+		const db = this.requireDb();
+		return new Promise<T[]>((resolve, reject) => {
+			db.all(sql, params, (err, rows) => (err ? reject(err) : resolve(rows as T[])));
+		});
+	}
+
+	private async openInternal(): Promise<void> {
+		const sqlite3 = joplin.require('sqlite3');
+		const dataDir = await joplin.plugins.dataDir();
+		const dbPath = `${dataDir}/${VectorDatabase.DB_FILE_NAME}`;
+
+		this.db = await new Promise<Sqlite3Database>((resolve, reject) => {
+			const db = new sqlite3.Database(dbPath, (err: Error | null) => {
+				if (err) reject(err);
+				else resolve(db);
+			});
+		});
+
+		await this.run(VectorDatabase.SCHEMA, []);
+	}
+
+	private requireDb(): Sqlite3Database {
+		if (!this.db) {
+			throw new Error('VectorDatabase used before open() completed.');
+		}
+		return this.db;
+	}
+}

src/data/Database/VectorRepository.test.ts

@@ -0,0 +1,139 @@
+import { VectorRepository } from './VectorRepository';
+import { IVectorDatabase } from './VectorDatabase';
+
+/**
+ * In-memory stand-in for VectorDatabase. sqlite3 is only reachable at runtime
+ * via joplin.require(), so VectorRepository is tested against this fake
+ * rather than a real database; it emulates the single upsert statement and
+ * the `note_id IN (...)` select that VectorRepository issues.
+ */
+class FakeVectorDatabase implements IVectorDatabase {
+	public opened = false;
+	public allCallBatchSizes: number[] = [];
+	private rows = new Map<
+		string,
+		{ note_id: string; model_id: string; updated_time: number; vector: Buffer }
+	>();
+
+	public async open(): Promise<void> {
+		this.opened = true;
+	}
+
+	public async run(_sql: string, params: unknown[]): Promise<void> {
+		if (params.length === 0) {
+			return; // BEGIN TRANSACTION / COMMIT / ROLLBACK
+		}
+		const [noteId, modelId, updatedTime, vector] = params as [string, string, number, Buffer];
+		this.rows.set(noteId, {
+			note_id: noteId,
+			model_id: modelId,
+			updated_time: updatedTime,
+			vector,
+		});
+	}
+
+	public async all<T>(_sql: string, params: unknown[]): Promise<T[]> {
+		const ids = params as string[];
+		this.allCallBatchSizes.push(ids.length);
+		const found = ids
+			.map((id) => this.rows.get(id))
+			.filter((r): r is NonNullable<typeof r> => !!r);
+		return found as unknown as T[];
+	}
+}
+
+describe('VectorRepository', () => {
+	let db: FakeVectorDatabase;
+	let repo: VectorRepository;
+
+	beforeEach(() => {
+		db = new FakeVectorDatabase();
+		repo = new VectorRepository(db);
+	});
+
+	describe('getMany', () => {
+		it('returns an empty map without opening the database for no IDs', async () => {
+			const result = await repo.getMany([]);
+			expect(result.size).toBe(0);
+			expect(db.opened).toBe(false);
+		});
+
+		it('returns nothing for IDs that were never saved', async () => {
+			const result = await repo.getMany(['missing']);
+			expect(result.size).toBe(0);
+			expect(db.opened).toBe(true);
+		});
+	});
+
+	describe('saveMany + getMany round trip', () => {
+		it('round-trips vector values through the Float32 BLOB encoding', async () => {
+			const vector = [0.1, -0.25, 0.987654, 1, -1, 0];
+			await repo.saveMany([{ noteId: 'n1', vector, modelId: 'm1', updatedTime: 100 }]);
+
+			const result = await repo.getMany(['n1']);
+			const entry = result.get('n1');
+
+			expect(entry).toBeDefined();
+			expect(entry!.modelId).toBe('m1');
+			expect(entry!.updatedTime).toBe(100);
+			expect(entry!.vector).toHaveLength(vector.length);
+			for (let i = 0; i < vector.length; i++) {
+				// Float32 storage loses some precision relative to the JS float64 input.
+				expect(entry!.vector[i]).toBeCloseTo(vector[i], 5);
+			}
+		});
+
+		it('overwrites the previous entry for the same note ID', async () => {
+			await repo.saveMany([
+				{ noteId: 'n1', vector: [1, 0], modelId: 'm1', updatedTime: 100 },
+			]);
+			await repo.saveMany([
+				{ noteId: 'n1', vector: [0, 1], modelId: 'm2', updatedTime: 200 },
+			]);
+
+			const result = await repo.getMany(['n1']);
+			const entry = result.get('n1');
+
+			expect(entry!.modelId).toBe('m2');
+			expect(entry!.updatedTime).toBe(200);
+			expect(entry!.vector[0]).toBeCloseTo(0, 5);
+			expect(entry!.vector[1]).toBeCloseTo(1, 5);
+		});
+
+		it('only returns entries for the requested IDs that exist', async () => {
+			await repo.saveMany([
+				{ noteId: 'n1', vector: [1, 0], modelId: 'm1', updatedTime: 100 },
+				{ noteId: 'n2', vector: [0, 1], modelId: 'm1', updatedTime: 100 },
+			]);
+
+			const result = await repo.getMany(['n1', 'n3']);
+
+			expect(result.has('n1')).toBe(true);
+			expect(result.has('n2')).toBe(false);
+			expect(result.has('n3')).toBe(false);
+		});
+
+		it('does nothing for an empty entries array', async () => {
+			await repo.saveMany([]);
+			expect(db.opened).toBe(false);
+		});
+	});
+
+	describe('large vaults', () => {
+		it("chunks getMany so a single query never exceeds SQLite's bound-parameter limit", async () => {
+			const noteIds = Array.from({ length: 1200 }, (_, i) => `n${i}`);
+			await repo.saveMany(
+				noteIds.map((id) => ({ noteId: id, vector: [1, 0], modelId: 'm1', updatedTime: 1 }))
+			);
+
+			const result = await repo.getMany(noteIds);
+
+			expect(result.size).toBe(1200);
+			expect(db.allCallBatchSizes.length).toBeGreaterThan(1);
+			for (const size of db.allCallBatchSizes) {
+				expect(size).toBeLessThanOrEqual(500);
+			}
+			expect(db.allCallBatchSizes.reduce((a, b) => a + b, 0)).toBe(1200);
+		});
+	});
+});

src/data/Database/VectorRepository.ts

@@ -0,0 +1,149 @@
+import { IVectorDatabase, VectorDatabase } from './VectorDatabase';
+
+export interface CachedVector {
+	vector: number[];
+	modelId: string;
+	updatedTime: number;
+}
+
+export interface VectorCacheEntry {
+	noteId: string;
+	vector: number[];
+	modelId: string;
+	updatedTime: number;
+}
+
+export interface VectorCache {
+	getMany(noteIds: string[]): Promise<Map<string, CachedVector>>;
+	saveMany(entries: VectorCacheEntry[]): Promise<void>;
+}
+
+interface VectorRow {
+	note_id: string;
+	model_id: string;
+	updated_time: number;
+	vector: Buffer;
+}
+
+/**
+ * Persists note embedding vectors in SQLite so unchanged notes aren't
+ * re-fetched from joplin.ai.getEmbeddings(). A cached vector is only reused
+ * when both its note_id and model_id match, since staleness is decided by
+ * the caller comparing `updatedTime` against the note's current updated_time.
+ */
+export class VectorRepository implements VectorCache {
+	/** SQLite caps bound parameters per statement (as low as 999 on some builds); stay well under it. */
+	private static readonly QUERY_BATCH_SIZE = 500;
+
+	/**
+	 * Serializes writes: sqlite transactions live on the shared connection, so
+	 * two interleaved saveMany calls would nest BEGIN TRANSACTION and error.
+	 */
+	private writeLock: Promise<void> = Promise.resolve();
+
+	public constructor(private readonly db: IVectorDatabase = new VectorDatabase()) {}
+
+	/** Returns cached vectors for the given note IDs, keyed by note ID. Missing notes are omitted. */
+	public async getMany(noteIds: string[]): Promise<Map<string, CachedVector>> {
+		if (noteIds.length === 0) {
+			return new Map();
+		}
+
+		await this.db.open();
+
+		const result = new Map<string, CachedVector>();
+		for (const batch of this.chunk(noteIds, VectorRepository.QUERY_BATCH_SIZE)) {
+			const rows = await this.queryBatch(batch);
+			for (const row of rows) {
+				result.set(row.note_id, {
+					vector: this.decodeVector(row.vector),
+					modelId: row.model_id,
+					updatedTime: row.updated_time,
+				});
+			}
+		}
+		return result;
+	}
+
+	/** Inserts or updates vectors for the given notes, in a single transaction. Calls are serialized. */
+	public saveMany(entries: VectorCacheEntry[]): Promise<void> {
+		if (entries.length === 0) {
+			return Promise.resolve();
+		}
+
+		const task = this.writeLock.then(() => this.saveManyInternal(entries));
+		// Keep the lock chain alive whether this write succeeds or fails.
+		this.writeLock = task.then(
+			() => undefined,
+			() => undefined
+		);
+		return task;
+	}
+
+	private async saveManyInternal(entries: VectorCacheEntry[]): Promise<void> {
+		await this.db.open();
+
+		await this.db.run('BEGIN TRANSACTION', []);
+		try {
+			for (const entry of entries) {
+				await this.db.run(
+					`INSERT INTO note_vectors (note_id, model_id, updated_time, vector)
+					 VALUES (?, ?, ?, ?)
+					 ON CONFLICT(note_id) DO UPDATE SET
+						model_id = excluded.model_id,
+						updated_time = excluded.updated_time,
+						vector = excluded.vector`,
+					[
+						entry.noteId,
+						entry.modelId,
+						entry.updatedTime,
+						this.encodeVector(entry.vector),
+					]
+				);
+			}
+			await this.db.run('COMMIT', []);
+		} catch (e) {
+			// A failed ROLLBACK (e.g. "database is locked") must not mask the
+			// original write error.
+			try {
+				await this.db.run('ROLLBACK', []);
+			} catch (rollbackError) {
+				console.error('Vector cache rollback failed after a write error:', rollbackError);
+			}
+			throw e;
+		}
+	}
+
+	private async queryBatch(noteIds: string[]): Promise<VectorRow[]> {
+		const placeholders = noteIds.map(() => '?').join(',');
+		return this.db.all<VectorRow>(
+			`SELECT note_id, model_id, updated_time, vector FROM note_vectors WHERE note_id IN (${placeholders})`,
+			noteIds
+		);
+	}
+
+	private chunk<T>(items: T[], size: number): T[][] {
+		const batches: T[][] = [];
+		for (let i = 0; i < items.length; i += size) {
+			batches.push(items.slice(i, i + size));
+		}
+		return batches;
+	}
+
+	/** Encodes a vector as a Float32 BLOB for compact SQLite storage. */
+	private encodeVector(vector: number[]): Buffer {
+		const floats = Float32Array.from(vector);
+		return Buffer.from(floats.buffer, floats.byteOffset, floats.byteLength);
+	}
+
+	/**
+	 * Decodes a Float32 BLOB back into a plain number array. Copies the bytes
+	 * first: Node pools small Buffers at arbitrary byte offsets, and viewing
+	 * an unaligned offset with `new Float32Array(buffer, byteOffset, …)`
+	 * throws a RangeError.
+	 */
+	private decodeVector(blob: Buffer): number[] {
+		const copy = blob.buffer.slice(blob.byteOffset, blob.byteOffset + blob.byteLength);
+		return Array.from(new Float32Array(copy));
+	}
+}