import { createHash } from 'node:crypto'; import type { CrawledFile, Snapshot } from '../core/crawl.ts'; import { mirrorPath } from '../core/paths.ts'; import { connect, migrate, type Db } from './db.ts'; /** * The crawl store: generations, diffing, and full-text search. * * Cursors are crawl ids, and diffs are computed by comparing generations on * entity identity and a content digest — never on upstream timestamps. That is * a deliberate response to a measured quirk: `GET /course-rooms/{id}/board` * returns the *request time* as `updatedAt` for most elements, so a * timestamp-based cursor would report every board as changed on every crawl. * Identity diffing also gives deletions for free, which no timestamp scheme can. */ export type NodeKind = 'course' | 'board' | 'lesson' | 'task' | 'file'; export interface StoredNode { kind: NodeKind; nodeId: string; courseId: string | null; title: string; body: string; path: string; meta: Record; digest: string; } export interface SearchResult { kind: NodeKind; nodeId: string; courseId: string | null; courseTitle: string; title: string; path: string; snippet: string; rank: number; } export interface DiffResult { added: StoredNode[]; changed: StoredNode[]; removed: { kind: NodeKind; nodeId: string; title: string; path: string }[]; } export interface ManifestEntry { fileId: string; name: string; path: string; size: number; mimeType: string; courseId: string | null; courseTitle: string; status: 'added' | 'unchanged' | 'removed'; } export class Store { private readonly db: Db; private constructor(db: Db) { this.db = db; } /** * Connects and migrates. Returns undefined rather than throwing when the * database is unreachable: the index is an accelerator, and a Pi that loses * its database should get slower, not broken. */ static async open(connectionString: string | undefined): Promise { if (!connectionString) return undefined; try { const db = await connect({ connectionString }); const ran = await migrate(db); if (ran.length > 0) console.log(`[schulcloud-mcp] applied migrations: ${ran.join(', ')}`); return new Store(db); } catch (error) { console.error( `[schulcloud-mcp] postgres unavailable (${error instanceof Error ? error.message : String(error)}); ` + 'running without the index — search falls back to live crawls.', ); return undefined; } } async close(): Promise { await this.db.end().catch(() => {}); } // --- generations ----------------------------------------------------- async latestCrawlId(): Promise { const { rows } = await this.db.query<{ id: string }>( `SELECT id FROM crawls WHERE status = 'ok' ORDER BY id DESC LIMIT 1`, ); return rows[0] ? Number(rows[0].id) : undefined; } async oldestCrawlId(): Promise { const { rows } = await this.db.query<{ id: string }>( `SELECT id FROM crawls WHERE status = 'ok' ORDER BY id ASC LIMIT 1`, ); return rows[0] ? Number(rows[0].id) : undefined; } /** * Turns a `since` value into a crawl id. * * Accepts a crawl id (the real cursor) or an ISO timestamp, resolved to the * newest crawl at or before it. The timestamp form is a convenience for * humans typing `--since yesterday`; correctness never depends on it. */ async resolveCursor(since: string): Promise { if (/^\d+$/.test(since)) return Number(since); const date = new Date(since); if (Number.isNaN(date.getTime())) return undefined; const { rows } = await this.db.query<{ id: string }>( `SELECT id FROM crawls WHERE status = 'ok' AND finished_at <= $1 ORDER BY id DESC LIMIT 1`, [date.toISOString()], ); return rows[0] ? Number(rows[0].id) : undefined; } /** * Persists a snapshot as a new generation. * * A per-course crawl carries the other courses' rows forward from the * previous generation, so every completed crawl is a *complete* picture and * any two can be diffed directly. Without that, a partial crawl would look * like a mass deletion. */ async saveSnapshot(snapshot: Snapshot, scope: string): Promise { const client = await this.db.connect(); try { await client.query('BEGIN'); const { rows } = await client.query<{ id: string }>( `INSERT INTO crawls (scope, course_count, file_count) VALUES ($1, $2, $3) RETURNING id`, [scope, snapshot.courses.length, snapshot.files.length], ); const crawlId = Number(rows[0]!.id); const previous = await this.latestCrawlIdIn(client); if (previous !== undefined && scope !== 'full') { // Carry forward everything the partial crawl did not look at. await client.query( `INSERT INTO nodes (crawl_id, kind, node_id, course_id, title, body, path, meta, digest) SELECT $1, kind, node_id, course_id, title, body, path, meta, digest FROM nodes WHERE crawl_id = $2 AND course_id IS DISTINCT FROM $3`, [crawlId, previous, scope], ); await client.query( `INSERT INTO search_docs (crawl_id, kind, node_id, course_id, course_title, title, body, path) SELECT $1, kind, node_id, course_id, course_title, title, body, path FROM search_docs WHERE crawl_id = $2 AND course_id IS DISTINCT FROM $3`, [crawlId, previous, scope], ); } const nodes = snapshotToNodes(snapshot); await insertNodes(client, crawlId, nodes); await insertSearchDocs(client, crawlId, nodes, snapshot); await client.query(`UPDATE crawls SET status = 'ok', finished_at = now() WHERE id = $1`, [crawlId]); await client.query('COMMIT'); return crawlId; } catch (error) { await client.query('ROLLBACK').catch(() => {}); throw error; } finally { client.release(); } } private async latestCrawlIdIn(client: { query: Db['query'] }): Promise { const { rows } = await client.query<{ id: string }>( `SELECT id FROM crawls WHERE status = 'ok' ORDER BY id DESC LIMIT 1`, ); return rows[0] ? Number(rows[0].id) : undefined; } /** Entities added, changed or removed between two generations. */ async diff(from: number, to: number): Promise { const added = await this.db.query( `SELECT n.* FROM nodes n WHERE n.crawl_id = $2 AND NOT EXISTS (SELECT 1 FROM nodes o WHERE o.crawl_id = $1 AND o.kind = n.kind AND o.node_id = n.node_id)`, [from, to], ); const changed = await this.db.query( `SELECT n.* FROM nodes n JOIN nodes o ON o.crawl_id = $1 AND o.kind = n.kind AND o.node_id = n.node_id WHERE n.crawl_id = $2 AND o.digest IS DISTINCT FROM n.digest`, [from, to], ); const removed = await this.db.query( `SELECT o.* FROM nodes o WHERE o.crawl_id = $1 AND NOT EXISTS (SELECT 1 FROM nodes n WHERE n.crawl_id = $2 AND n.kind = o.kind AND n.node_id = o.node_id)`, [from, to], ); return { added: added.rows.map(toNode), changed: changed.rows.map(toNode), removed: removed.rows.map((row) => ({ kind: row.kind as NodeKind, nodeId: row.node_id, title: row.title, path: row.path, })), }; } // --- search ---------------------------------------------------------- /** * Full-text search over the newest generation. * * Two arms, merged: the `german` dictionary for stemmed matching, and * trigram similarity for what stemming misses — German compounds mean * "Datenschutz" does not stem-match inside "Datenschutzgrundverordnung", * and that is the common case, not an edge case. */ async search(query: string, options: { limit?: number; kinds?: NodeKind[] } = {}): Promise { const crawlId = await this.latestCrawlId(); if (crawlId === undefined) return []; const limit = options.limit ?? 30; const kinds = options.kinds ?? null; const fts = await this.db.query( `SELECT kind, node_id, course_id, course_title, title, path, ts_rank(fts, q) AS rank, ts_headline('german', coalesce(nullif(body, ''), title), q, 'MaxWords=32, MinWords=8, MaxFragments=1, StartSel=**, StopSel=**') AS snippet FROM search_docs, websearch_to_tsquery('german', $2) q WHERE crawl_id = $1 AND fts @@ q AND ($3::text[] IS NULL OR kind = ANY($3)) ORDER BY rank DESC LIMIT $4`, [crawlId, query, kinds, limit], ); const trgm = await this.db.query( `SELECT kind, node_id, course_id, course_title, title, path, similarity(title, $2) AS rank, title AS snippet FROM search_docs WHERE crawl_id = $1 AND title %> $2 AND ($3::text[] IS NULL OR kind = ANY($3)) ORDER BY rank DESC LIMIT $4`, [crawlId, query, kinds, limit], ); const seen = new Set(); const merged: SearchResult[] = []; for (const row of [...fts.rows, ...trgm.rows]) { const key = `${row.kind}:${row.node_id}`; if (seen.has(key)) continue; seen.add(key); merged.push({ kind: row.kind as NodeKind, nodeId: row.node_id, courseId: row.course_id, courseTitle: row.course_title, title: row.title, path: row.path, snippet: (row.snippet ?? '').replace(/\s+/g, ' ').trim(), rank: Number(row.rank), }); } return merged.sort((a, b) => b.rank - a.rank).slice(0, limit); } // --- files ----------------------------------------------------------- /** File records in the newest generation, optionally diffed against a cursor. */ async manifest(since?: number): Promise<{ crawlId: number; entries: ManifestEntry[] }> { const crawlId = await this.latestCrawlId(); if (crawlId === undefined) return { crawlId: 0, entries: [] }; const current = await this.db.query( `SELECT n.*, coalesce(s.course_title, '') AS course_title FROM nodes n LEFT JOIN search_docs s ON s.crawl_id = n.crawl_id AND s.kind = n.kind AND s.node_id = n.node_id WHERE n.crawl_id = $1 AND n.kind = 'file'`, [crawlId], ); let previousIds = new Set(); if (since !== undefined) { const previous = await this.db.query<{ node_id: string }>( `SELECT node_id FROM nodes WHERE crawl_id = $1 AND kind = 'file'`, [since], ); previousIds = new Set(previous.rows.map((row) => row.node_id)); } const entries: ManifestEntry[] = current.rows.map((row) => { const meta = row.meta as { size?: number; mimeType?: string }; return { fileId: row.node_id, name: row.title, path: row.path, size: Number(meta.size ?? 0), mimeType: String(meta.mimeType ?? 'application/octet-stream'), courseId: row.course_id, courseTitle: row.course_title, status: since === undefined ? 'added' : previousIds.has(row.node_id) ? 'unchanged' : 'added', }; }); if (since !== undefined) { const currentIds = new Set(current.rows.map((row) => row.node_id)); const gone = await this.db.query( `SELECT * FROM nodes WHERE crawl_id = $1 AND kind = 'file'`, [since], ); for (const row of gone.rows) { if (currentIds.has(row.node_id)) continue; const meta = row.meta as { size?: number; mimeType?: string }; entries.push({ fileId: row.node_id, name: row.title, path: row.path, size: Number(meta.size ?? 0), mimeType: String(meta.mimeType ?? 'application/octet-stream'), courseId: row.course_id, courseTitle: '', status: 'removed', }); } } return { crawlId, entries }; } /** File ids in the newest generation that have no extracted text yet. */ async filesNeedingText(): Promise<{ fileId: string; name: string; mimeType: string; size: number }[]> { const crawlId = await this.latestCrawlId(); if (crawlId === undefined) return []; const { rows } = await this.db.query<{ node_id: string; title: string; meta: Record }>( `SELECT n.node_id, n.title, n.meta FROM nodes n WHERE n.crawl_id = $1 AND n.kind = 'file' AND NOT EXISTS (SELECT 1 FROM file_texts f WHERE f.file_id = n.node_id AND f.extracted_at IS NOT NULL)`, [crawlId], ); return rows.map((row) => ({ fileId: row.node_id, name: row.title, mimeType: String((row.meta as { mimeType?: string }).mimeType ?? ''), size: Number((row.meta as { size?: number }).size ?? 0), })); } async recordFileText(entry: { fileId: string; name: string; mimeType: string; size: number; content: string | null; note: string; mirrorPath: string | null; mirrorSize: number | null; }): Promise { await this.db.query( `INSERT INTO file_texts (file_id, name, mime_type, size, content, extract_note, extracted_at, mirror_path, mirror_size, mirrored_at) VALUES ($1,$2,$3,$4,$5,$6, now(), $7,$8, CASE WHEN $7::text IS NULL THEN NULL ELSE now() END) ON CONFLICT (file_id) DO UPDATE SET content = EXCLUDED.content, extract_note = EXCLUDED.extract_note, extracted_at = now(), mirror_path = COALESCE(EXCLUDED.mirror_path, file_texts.mirror_path), mirror_size = COALESCE(EXCLUDED.mirror_size, file_texts.mirror_size), mirrored_at = CASE WHEN EXCLUDED.mirror_path IS NULL THEN file_texts.mirrored_at ELSE now() END`, [ entry.fileId, entry.name, entry.mimeType, entry.size, entry.content, entry.note, entry.mirrorPath, entry.mirrorSize, ], ); // Make the newly extracted text searchable in the current generation. await this.db.query( `UPDATE search_docs SET body = $2 WHERE kind = 'file' AND node_id = $1 AND crawl_id = (SELECT id FROM crawls WHERE status = 'ok' ORDER BY id DESC LIMIT 1)`, [entry.fileId, entry.content ?? ''], ); } async mirrorEntry(fileId: string): Promise<{ path: string; size: number; name: string; mimeType: string } | undefined> { const { rows } = await this.db.query<{ mirror_path: string | null; mirror_size: string | null; name: string; mime_type: string | null }>( `SELECT mirror_path, mirror_size, name, mime_type FROM file_texts WHERE file_id = $1`, [fileId], ); const row = rows[0]; if (!row?.mirror_path) return undefined; return { path: row.mirror_path, size: Number(row.mirror_size ?? 0), name: row.name, mimeType: row.mime_type ?? 'application/octet-stream', }; } async stats(): Promise<{ crawlId?: number; crawledAt?: string; nodes: number; files: number; extracted: number; mirrored: number }> { const crawlId = await this.latestCrawlId(); if (crawlId === undefined) return { nodes: 0, files: 0, extracted: 0, mirrored: 0 }; const { rows } = await this.db.query<{ crawled_at: string; nodes: string; files: string; extracted: string; mirrored: string }>( `SELECT (SELECT finished_at FROM crawls WHERE id = $1) AS crawled_at, (SELECT count(*) FROM nodes WHERE crawl_id = $1) AS nodes, (SELECT count(*) FROM nodes WHERE crawl_id = $1 AND kind = 'file') AS files, (SELECT count(*) FROM file_texts WHERE extracted_at IS NOT NULL) AS extracted, (SELECT count(*) FROM file_texts WHERE mirror_path IS NOT NULL) AS mirrored`, [crawlId], ); const row = rows[0]!; return { crawlId, crawledAt: row.crawled_at, nodes: Number(row.nodes), files: Number(row.files), extracted: Number(row.extracted), mirrored: Number(row.mirrored), }; } } // --- mapping ------------------------------------------------------------- interface NodeRow { kind: string; node_id: string; course_id: string | null; title: string; body: string; path: string; meta: Record; digest: string; } interface SearchRow { kind: string; node_id: string; course_id: string | null; course_title: string; title: string; path: string; snippet: string | null; rank: string; } function toNode(row: NodeRow): StoredNode { return { kind: row.kind as NodeKind, nodeId: row.node_id, courseId: row.course_id, title: row.title, body: row.body, path: row.path, meta: row.meta, digest: row.digest, }; } /** Content fingerprint — what "changed" means, independent of any timestamp. */ function digestOf(parts: unknown[]): string { return createHash('sha256').update(JSON.stringify(parts)).digest('hex').slice(0, 32); } export function snapshotToNodes(snapshot: Snapshot): StoredNode[] { const nodes: StoredNode[] = []; for (const course of snapshot.courses) { nodes.push({ kind: 'course', nodeId: course.course.id, courseId: course.course.id, title: course.title, body: '', path: course.title, meta: { shortTitle: course.course.shortTitle, isLocked: course.course.isLocked ?? false }, digest: digestOf([course.title, course.course.isLocked ?? false]), }); for (const board of course.boards) { nodes.push({ kind: 'board', nodeId: board.id, courseId: course.course.id, title: board.title, body: board.text, path: `${course.title}/${board.title}`, meta: { columns: board.board.columns.length, fileCount: board.board.fileCount }, digest: digestOf([board.title, board.text]), }); } for (const lesson of course.lessons) { nodes.push({ kind: 'lesson', nodeId: lesson.id, courseId: course.course.id, title: lesson.name, body: lesson.text, path: `${course.title}/${lesson.name}`, meta: { hidden: lesson.hidden, materials: lesson.materials }, digest: digestOf([lesson.name, lesson.text, lesson.hidden]), }); } for (const task of course.tasks) { nodes.push({ kind: 'task', nodeId: task.id, courseId: course.course.id, title: task.task.name, body: task.text, path: `${course.title}/${task.task.name}`, meta: { dueDate: task.task.dueDate ?? null, status: task.task.status }, digest: digestOf([task.task.name, task.text, task.task.dueDate ?? null]), }); } } for (const file of snapshot.files) { nodes.push(fileNode(file)); } return nodes; } function fileNode(file: CrawledFile): StoredNode { const path = mirrorPath(file.at, file.record.name, file.record.id); return { kind: 'file', nodeId: file.record.id, courseId: file.at.courseId, title: file.record.name, body: '', path, meta: { size: file.record.size, mimeType: file.record.mimeType, parentType: file.parentType, parentId: file.parentId, securityCheckStatus: file.record.securityCheckStatus, at: file.at, }, // File records are less immutable than they look: `PATCH /file/rename/{id}` // changes the name in place, keeping the id and the size, and teachers do // rename files. Leaving the name out made that invisible to what_changed — // the file simply reappeared under its new name with nothing reported. // Size still catches a record rewritten in place under the same name. digest: digestOf([file.record.id, file.record.name, file.record.size]), }; } async function insertNodes(client: { query: Db['query'] }, crawlId: number, nodes: StoredNode[]): Promise { const CHUNK = 200; for (let i = 0; i < nodes.length; i += CHUNK) { const batch = nodes.slice(i, i + CHUNK); const values: unknown[] = []; const tuples = batch.map((node, index) => { const base = index * 9; values.push(crawlId, node.kind, node.nodeId, node.courseId, node.title, node.body, node.path, node.meta, node.digest); return `($${base + 1},$${base + 2},$${base + 3},$${base + 4},$${base + 5},$${base + 6},$${base + 7},$${base + 8},$${base + 9})`; }); await client.query( `INSERT INTO nodes (crawl_id, kind, node_id, course_id, title, body, path, meta, digest) VALUES ${tuples.join(',')} ON CONFLICT (crawl_id, kind, node_id) DO UPDATE SET title = EXCLUDED.title, body = EXCLUDED.body, path = EXCLUDED.path, meta = EXCLUDED.meta, digest = EXCLUDED.digest, course_id = EXCLUDED.course_id`, values, ); } } async function insertSearchDocs( client: { query: Db['query'] }, crawlId: number, nodes: StoredNode[], snapshot: Snapshot, ): Promise { const courseTitles = new Map(snapshot.courses.map((course) => [course.course.id, course.title])); const CHUNK = 200; for (let i = 0; i < nodes.length; i += CHUNK) { const batch = nodes.slice(i, i + CHUNK); const values: unknown[] = []; const tuples = batch.map((node, index) => { const base = index * 8; values.push( crawlId, node.kind, node.nodeId, node.courseId, courseTitles.get(node.courseId ?? '') ?? '', node.title, node.body, node.path, ); return `($${base + 1},$${base + 2},$${base + 3},$${base + 4},$${base + 5},$${base + 6},$${base + 7},$${base + 8})`; }); await client.query( `INSERT INTO search_docs (crawl_id, kind, node_id, course_id, course_title, title, body, path) VALUES ${tuples.join(',')} ON CONFLICT (crawl_id, kind, node_id) DO UPDATE SET course_title = EXCLUDED.course_title, title = EXCLUDED.title, body = EXCLUDED.body, path = EXCLUDED.path`, values, ); } // File bodies live in file_texts and survive re-crawls; pull them in so // already-extracted PDFs are searchable in this generation immediately. await client.query( `UPDATE search_docs s SET body = coalesce(f.content, '') FROM file_texts f WHERE s.crawl_id = $1 AND s.kind = 'file' AND s.node_id = f.file_id AND f.content IS NOT NULL`, [crawlId], ); }