ANG-011:Incremental updates via workspace events

src/data/Database/GraphCacheRepository.test.ts

@@ -0,0 +1,155 @@
+import { GraphCacheRepository } from './GraphCacheRepository';
+import { IVectorDatabase } from './VectorDatabase';
+import { GraphData } from '../../services/graph/types';
+import { Note } from '../Types';
+
+class FakeConnection implements IVectorDatabase {
+	public opened = false;
+	private graphRow: { notes_json: string; graph_json: string } | null = null;
+	private syncStateRow: { events_cursor: string | null; embeddings_cursor: string | null } | null =
+		null;
+
+	public async open(): Promise<void> {
+		this.opened = true;
+	}
+
+	public async run(sql: string, params: unknown[]): Promise<void> {
+		if (sql.includes('INTO graph_cache')) {
+			const [notesJson, graphJson] = params as [string, string, number];
+			this.graphRow = { notes_json: notesJson, graph_json: graphJson };
+		} else if (sql.includes('embeddings_cursor')) {
+			const [cursor] = params as [string];
+			this.syncStateRow = {
+				events_cursor: this.syncStateRow?.events_cursor ?? null,
+				embeddings_cursor: cursor,
+			};
+		} else if (sql.includes('INTO sync_state')) {
+			const [cursor] = params as [string];
+			this.syncStateRow = {
+				events_cursor: cursor,
+				embeddings_cursor: this.syncStateRow?.embeddings_cursor ?? null,
+			};
+		}
+	}
+
+	public async all<T>(sql: string): Promise<T[]> {
+		if (sql.includes('FROM graph_cache')) {
+			return (this.graphRow ? [this.graphRow] : []) as unknown as T[];
+		}
+		if (sql.includes('FROM sync_state')) {
+			return (this.syncStateRow ? [this.syncStateRow] : []) as unknown as T[];
+		}
+		return [];
+	}
+}
+
+const note: Note = {
+	id: 'n1',
+	parent_id: 'p1',
+	title: 'Note 1',
+	body: 'body',
+	created_time: 1,
+	updated_time: 2,
+};
+
+const graphData: GraphData = {
+	nodes: [{ data: { id: 'n1', label: 'Note 1', noteId: 'n1', degree: 0, community: 0, size: 1 } }],
+	edges: [],
+};
+
+describe('GraphCacheRepository', () => {
+	let db: FakeConnection;
+	let repo: GraphCacheRepository;
+
+	beforeEach(() => {
+		db = new FakeConnection();
+		repo = new GraphCacheRepository(db);
+	});
+
+	describe('graph cache', () => {
+		it('returns null when nothing has been cached yet', async () => {
+			const result = await repo.loadGraph();
+			expect(result).toBeNull();
+		});
+
+		it('round-trips notes and graph data through save/load', async () => {
+			await repo.saveGraph([note], graphData);
+
+			const result = await repo.loadGraph();
+
+			expect(result).toEqual({ notes: [note], graphData });
+		});
+
+		it('overwrites the previous cache on a second save', async () => {
+			await repo.saveGraph([note], graphData);
+			const secondNote = { ...note, title: 'Updated' };
+			await repo.saveGraph([secondNote], graphData);
+
+			const result = await repo.loadGraph();
+
+			expect(result?.notes[0].title).toBe('Updated');
+		});
+	});
+
+	describe('events cursor', () => {
+		it('returns null when sync has never run', async () => {
+			const cursor = await repo.loadEventsCursor();
+			expect(cursor).toBeNull();
+		});
+
+		it('round-trips the cursor through save/load', async () => {
+			await repo.saveEventsCursor('cursor-1');
+			expect(await repo.loadEventsCursor()).toBe('cursor-1');
+		});
+
+		it('overwrites the previous cursor on a second save', async () => {
+			await repo.saveEventsCursor('cursor-1');
+			await repo.saveEventsCursor('cursor-2');
+			expect(await repo.loadEventsCursor()).toBe('cursor-2');
+		});
+	});
+
+	describe('embeddings cursor', () => {
+		it('returns null when the AI-on sweep has never run', async () => {
+			const cursor = await repo.loadEmbeddingsCursor();
+			expect(cursor).toBeNull();
+		});
+
+		it('round-trips the cursor through save/load', async () => {
+			await repo.saveEmbeddingsCursor('embeddings-cursor-1');
+			expect(await repo.loadEmbeddingsCursor()).toBe('embeddings-cursor-1');
+		});
+
+		it('overwrites the previous cursor on a second save', async () => {
+			await repo.saveEmbeddingsCursor('embeddings-cursor-1');
+			await repo.saveEmbeddingsCursor('embeddings-cursor-2');
+			expect(await repo.loadEmbeddingsCursor()).toBe('embeddings-cursor-2');
+		});
+
+		it('saving the events cursor and the embeddings cursor never clobbers the other', async () => {
+			await repo.saveEventsCursor('events-cursor-1');
+			await repo.saveEmbeddingsCursor('embeddings-cursor-1');
+			await repo.saveEventsCursor('events-cursor-2');
+
+			expect(await repo.loadEventsCursor()).toBe('events-cursor-2');
+			expect(await repo.loadEmbeddingsCursor()).toBe('embeddings-cursor-1');
+		});
+	});
+
+	describe('write serialization', () => {
+		it('serializes interleaved graph and cursor writes instead of racing them', async () => {
+			const order: string[] = [];
+			const originalRun = db.run.bind(db);
+			db.run = async (sql: string, params: unknown[]) => {
+				order.push(sql.includes('graph_cache') ? 'graph' : 'cursor');
+				await originalRun(sql, params);
+			};
+
+			await Promise.all([repo.saveGraph([note], graphData), repo.saveEventsCursor('cursor-1')]);
+
+			expect(order).toHaveLength(2);
+			expect(await repo.loadGraph()).not.toBeNull();
+			expect(await repo.loadEventsCursor()).toBe('cursor-1');
+		});
+	});
+});

src/data/Database/GraphCacheRepository.ts

@@ -0,0 +1,124 @@
+import { IVectorDatabase, VectorDatabase } from './VectorDatabase';
+import { Note } from '../Types';
+import { GraphData } from '../../services/graph/types';
+
+const DB_FILE_NAME = 'note-graph-cache.sqlite';
+
+const GRAPH_CACHE_SCHEMA = `
+	CREATE TABLE IF NOT EXISTS graph_cache (
+		id INTEGER PRIMARY KEY CHECK (id = 1),
+		notes_json TEXT NOT NULL,
+		graph_json TEXT NOT NULL,
+		updated_time INTEGER NOT NULL
+	)
+`;
+
+const SYNC_STATE_SCHEMA = `
+	CREATE TABLE IF NOT EXISTS sync_state (
+		id INTEGER PRIMARY KEY CHECK (id = 1),
+		events_cursor TEXT,
+		embeddings_cursor TEXT
+	)
+`;
+
+interface GraphCacheRow {
+	notes_json: string;
+	graph_json: string;
+}
+
+interface SyncStateRow {
+	events_cursor: string | null;
+	embeddings_cursor: string | null;
+}
+
+export class GraphCacheRepository {
+	private writeLock: Promise<void> = Promise.resolve();
+
+	public constructor(
+		private readonly db: IVectorDatabase = new VectorDatabase(DB_FILE_NAME, [
+			GRAPH_CACHE_SCHEMA,
+			SYNC_STATE_SCHEMA,
+		])
+	) {}
+
+	public async loadGraph(): Promise<{ notes: Note[]; graphData: GraphData } | null> {
+		await this.db.open();
+		const rows = await this.db.all<GraphCacheRow>(
+			'SELECT notes_json, graph_json FROM graph_cache WHERE id = 1',
+			[]
+		);
+		const row = rows[0];
+		if (!row) return null;
+
+		return {
+			notes: JSON.parse(row.notes_json) as Note[],
+			graphData: JSON.parse(row.graph_json) as GraphData,
+		};
+	}
+
+	public saveGraph(notes: Note[], graphData: GraphData): Promise<void> {
+		return this.enqueueWrite(async () => {
+			await this.db.open();
+			await this.db.run(
+				`INSERT INTO graph_cache (id, notes_json, graph_json, updated_time)
+				 VALUES (1, ?, ?, ?)
+				 ON CONFLICT(id) DO UPDATE SET
+					notes_json = excluded.notes_json,
+					graph_json = excluded.graph_json,
+					updated_time = excluded.updated_time`,
+				[JSON.stringify(notes), JSON.stringify(graphData), Date.now()]
+			);
+		});
+	}
+
+	public async loadEventsCursor(): Promise<string | null> {
+		await this.db.open();
+		const rows = await this.db.all<SyncStateRow>(
+			'SELECT events_cursor FROM sync_state WHERE id = 1',
+			[]
+		);
+		return rows[0]?.events_cursor ?? null;
+	}
+
+	public saveEventsCursor(cursor: string): Promise<void> {
+		return this.enqueueWrite(async () => {
+			await this.db.open();
+			await this.db.run(
+				`INSERT INTO sync_state (id, events_cursor)
+				 VALUES (1, ?)
+				 ON CONFLICT(id) DO UPDATE SET events_cursor = excluded.events_cursor`,
+				[cursor]
+			);
+		});
+	}
+
+	public async loadEmbeddingsCursor(): Promise<string | null> {
+		await this.db.open();
+		const rows = await this.db.all<SyncStateRow>(
+			'SELECT embeddings_cursor FROM sync_state WHERE id = 1',
+			[]
+		);
+		return rows[0]?.embeddings_cursor ?? null;
+	}
+
+	public saveEmbeddingsCursor(cursor: string): Promise<void> {
+		return this.enqueueWrite(async () => {
+			await this.db.open();
+			await this.db.run(
+				`INSERT INTO sync_state (id, embeddings_cursor)
+				 VALUES (1, ?)
+				 ON CONFLICT(id) DO UPDATE SET embeddings_cursor = excluded.embeddings_cursor`,
+				[cursor]
+			);
+		});
+	}
+
+	private enqueueWrite(write: () => Promise<void>): Promise<void> {
+		const task = this.writeLock.then(write);
+		this.writeLock = task.then(
+			() => undefined,
+			() => undefined
+		);
+		return task;
+	}
+}

src/data/Database/VectorDatabase.ts

@@ -19,30 +19,25 @@ interface Sqlite3Database {
 /**
  * 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.
+ * plugin). Owns only the connection and schema; query logic lives in the
+ * repository classes that use it.
  */
 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;
 
+	public constructor(
+		private readonly dbFileName: string,
+		private readonly schemaStatements: string[]
+	) {}
+
 	/**
-	 * 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.
+	 * Opens (creating if needed) the 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;
@@ -76,7 +71,7 @@ export class VectorDatabase implements IVectorDatabase {
 	private async openInternal(): Promise<void> {
 		const sqlite3 = joplin.require('sqlite3');
 		const dataDir = await joplin.plugins.dataDir();
-		const dbPath = `${dataDir}/${VectorDatabase.DB_FILE_NAME}`;
+		const dbPath = `${dataDir}/${this.dbFileName}`;
 
 		this.db = await new Promise<Sqlite3Database>((resolve, reject) => {
 			const db = new sqlite3.Database(dbPath, (err: Error | null) => {
@@ -85,12 +80,14 @@ export class VectorDatabase implements IVectorDatabase {
 			});
 		});
 
-		await this.run(VectorDatabase.SCHEMA, []);
+		for (const statement of this.schemaStatements) {
+			await this.run(statement, []);
+		}
 	}
 
 	private requireDb(): Sqlite3Database {
 		if (!this.db) {
-			throw new Error('VectorDatabase used before open() completed.');
+			throw new Error(`VectorDatabase (${this.dbFileName}) used before open() completed.`);
 		}
 		return this.db;
 	}

src/data/Database/VectorRepository.ts

@@ -1,5 +1,15 @@
 import { IVectorDatabase, VectorDatabase } from './VectorDatabase';
 
+const DB_FILE_NAME = 'note-graph-vectors.sqlite';
+const 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
+	)
+`;
+
 export interface CachedVector {
 	vector: number[];
 	modelId: string;
@@ -41,7 +51,9 @@ export class VectorRepository implements VectorCache {
 	 */
 	private writeLock: Promise<void> = Promise.resolve();
 
-	public constructor(private readonly db: IVectorDatabase = new VectorDatabase()) {}
+	public constructor(
+		private readonly db: IVectorDatabase = new VectorDatabase(DB_FILE_NAME, [SCHEMA])
+	) {}
 
 	/** 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>> {