Add Postgres store, path safety, and the crawl indexer
Store: crawl generations as the sync cursor. Diffs compare generations
on entity identity plus a content digest, never on upstream timestamps —
GET /course-rooms/{id}/board returns request time as updatedAt for most
elements, so a timestamp cursor would report every board as changed on
every crawl. Identity diffing also yields deletions, which no timestamp
scheme can. A per-course crawl carries the other courses' rows forward
so every completed generation is a complete picture and any two diff
directly; without that a partial crawl reads as a mass deletion.
FTS uses the german dictionary with weighted title/body, plus a pg_trgm
arm because stemming will not match "Datenschutz" inside
"Datenschutzgrundverordnung" and German compounds make that the common
case. file_texts is keyed by file record id and deliberately outlives
generations: records are immutable upstream, so text extracted once is
valid forever and a re-crawl of unchanged content costs nothing.
Store.open returns undefined instead of throwing when Postgres is
unreachable — the index is an accelerator, and a Pi that loses its
database should get slower, not broken.
core/paths.ts is the security boundary for the mirror. Course titles,
card titles and filenames are all user-supplied upstream, so this is
where a hostile name stops being text and becomes a path. Two bugs found
by its own tests: "///" produced "---" instead of falling back, and dot
runs survived mid-component. Now no ".." can survive anywhere, which
makes the invariant checkable rather than a claim about ordering.
Indexer coalesces concurrent refreshes onto one run and enforces a
minimum interval, since a full crawl is ~270 requests from an account
that looks like a student.
9 store tests against a real Postgres (mocks would test nothing here)
and 13 path tests; 47 total.
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
@@ -23,9 +23,18 @@ export interface Config {
|
||||
/**
|
||||
* How often to ping the instance to hold the session open. Must stay well
|
||||
* under the instance's `JWT_TIMEOUT_SECONDS` (7200s here) — see
|
||||
* src/keepalive.ts. Zero disables the keepalive.
|
||||
* src/core/keepalive.ts. Zero disables the keepalive.
|
||||
*/
|
||||
keepaliveIntervalMs: number;
|
||||
|
||||
/** Postgres for the search index and file mirror. Unset = live-only mode. */
|
||||
databaseUrl: string | undefined;
|
||||
/** Where mirrored file bytes live on disk. */
|
||||
mirrorDir: string;
|
||||
/** Files larger than this are indexed as metadata but not mirrored. */
|
||||
mirrorMaxBytes: number;
|
||||
/** How often to re-crawl on a timer. Zero = only on demand. */
|
||||
crawlIntervalMs: number;
|
||||
}
|
||||
|
||||
function required(name: string): string {
|
||||
@@ -66,5 +75,9 @@ export function loadConfig(): Config {
|
||||
maxExtractedChars: int('MAX_EXTRACTED_CHARS', 120_000),
|
||||
requestTimeoutMs: int('REQUEST_TIMEOUT_MS', 30_000),
|
||||
keepaliveIntervalMs: intAllowingZero('KEEPALIVE_INTERVAL_MS', 30 * 60_000),
|
||||
databaseUrl: process.env.DATABASE_URL?.trim() || undefined,
|
||||
mirrorDir: process.env.MIRROR_DIR?.trim() || '/var/lib/schulcloud-mcp/mirror',
|
||||
mirrorMaxBytes: int('MIRROR_MAX_BYTES', 64 * 1024 * 1024),
|
||||
crawlIntervalMs: intAllowingZero('CRAWL_INTERVAL_MS', 6 * 60 * 60_000),
|
||||
};
|
||||
}
|
||||
|
||||
107
src/core/paths.ts
Normal file
107
src/core/paths.ts
Normal file
@@ -0,0 +1,107 @@
|
||||
import type { Breadcrumb } from './crawl.ts';
|
||||
|
||||
/**
|
||||
* Builds filesystem paths for mirrored files.
|
||||
*
|
||||
* Every component originates in Schulcloud — course titles, card titles and
|
||||
* filenames are all user-supplied upstream — so this is the boundary where a
|
||||
* hostile name stops being text and becomes a path. A file called
|
||||
* `../../.ssh/authorized_keys` must not be able to escape the mirror root, and
|
||||
* that is this module's whole job. Both the server's mirror and the CLI's sync
|
||||
* go through it.
|
||||
*/
|
||||
|
||||
/** Windows reserved device names, rejected regardless of platform for portability. */
|
||||
const RESERVED = new Set([
|
||||
'con', 'prn', 'aux', 'nul',
|
||||
'com1', 'com2', 'com3', 'com4', 'com5', 'com6', 'com7', 'com8', 'com9',
|
||||
'lpt1', 'lpt2', 'lpt3', 'lpt4', 'lpt5', 'lpt6', 'lpt7', 'lpt8', 'lpt9',
|
||||
]);
|
||||
|
||||
const MAX_COMPONENT = 100;
|
||||
|
||||
/**
|
||||
* Reduces one arbitrary string to a single safe path component.
|
||||
*
|
||||
* Separators, traversal, control characters, NUL and the characters Windows
|
||||
* forbids are all removed rather than escaped — a mirror path is for humans
|
||||
* browsing their coursework, not a reversible encoding.
|
||||
*/
|
||||
export function safeComponent(raw: string, fallback = 'untitled'): string {
|
||||
let value = raw.normalize('NFC');
|
||||
|
||||
// Control characters and NUL first: they could terminate a path early.
|
||||
// eslint-disable-next-line no-control-regex
|
||||
value = value.replace(/[\u0000-\u001f\u007f]/g, '');
|
||||
// Separators become a dash rather than vanishing, so words stay apart.
|
||||
value = value.replace(/[/\\]/g, '-');
|
||||
value = value.replace(/[<>:"|?*]/g, '');
|
||||
// Collapse dot runs so no ".." survives anywhere in the component, not just
|
||||
// at the start — this is what makes "contains no traversal" a simple,
|
||||
// checkable property rather than a claim about ordering.
|
||||
value = value.replace(/\.{2,}/g, '.');
|
||||
value = value.replace(/\s+/g, ' ');
|
||||
// Trim the punctuation a name can start or end with: a leading dot hides
|
||||
// the file, a trailing dot or space is silently dropped by Windows and
|
||||
// would make two distinct names collide.
|
||||
value = value.replace(/^[.\s-]+/, '').replace(/[.\s]+$/, '');
|
||||
value = value.replace(/-{2,}/g, '-').replace(/^-+|-+$/g, '');
|
||||
value = value.trim();
|
||||
|
||||
// Whatever is left must contain something other than punctuation, or the
|
||||
// name carried no information and the fallback is more useful.
|
||||
if (!value || !/[\p{L}\p{N}]/u.test(value)) return fallback;
|
||||
if (RESERVED.has(value.toLowerCase())) return `_${value}`;
|
||||
|
||||
if (value.length > MAX_COMPONENT) {
|
||||
// Preserve the extension when truncating, so file type survives.
|
||||
const dot = value.lastIndexOf('.');
|
||||
if (dot > 0 && value.length - dot <= 12) {
|
||||
const ext = value.slice(dot);
|
||||
value = value.slice(0, MAX_COMPONENT - ext.length) + ext;
|
||||
} else {
|
||||
value = value.slice(0, MAX_COMPONENT);
|
||||
}
|
||||
}
|
||||
return value;
|
||||
}
|
||||
|
||||
/**
|
||||
* Relative mirror path for a file: `Course/Container/Card/name.ext`.
|
||||
*
|
||||
* Always relative, always forward-slashed, never absolute and never escaping.
|
||||
* `fileId` disambiguates the rare case of two files with the same name in the
|
||||
* same card, which the API permits.
|
||||
*/
|
||||
export function mirrorPath(at: Breadcrumb, fileName: string, fileId: string): string {
|
||||
const parts = [at.courseTitle, at.containerTitle, at.cardTitle]
|
||||
.filter((part): part is string => Boolean(part && part.trim()))
|
||||
.map((part) => safeComponent(part));
|
||||
|
||||
const name = safeComponent(fileName, fileId);
|
||||
parts.push(name);
|
||||
return parts.join('/');
|
||||
}
|
||||
|
||||
/**
|
||||
* Resolves a relative mirror path under `root`, refusing anything that escapes.
|
||||
*
|
||||
* The last line of defence: even if a component slipped through
|
||||
* `safeComponent`, this rejects the result rather than writing outside the
|
||||
* root. Callers must use this instead of `path.join` on untrusted input.
|
||||
*/
|
||||
export function resolveWithin(root: string, relative: string): string {
|
||||
if (relative.startsWith('/') || /^[a-zA-Z]:/.test(relative)) {
|
||||
throw new Error(`refusing absolute path in mirror: ${relative}`);
|
||||
}
|
||||
const segments = relative.split('/').filter((segment) => segment.length > 0);
|
||||
if (segments.some((segment) => segment === '.' || segment === '..')) {
|
||||
throw new Error(`refusing path traversal in mirror: ${relative}`);
|
||||
}
|
||||
const normalizedRoot = root.endsWith('/') ? root.slice(0, -1) : root;
|
||||
const resolved = `${normalizedRoot}/${segments.join('/')}`;
|
||||
if (!resolved.startsWith(`${normalizedRoot}/`)) {
|
||||
throw new Error(`refusing path outside mirror root: ${relative}`);
|
||||
}
|
||||
return resolved;
|
||||
}
|
||||
240
src/indexer/indexer.ts
Normal file
240
src/indexer/indexer.ts
Normal file
@@ -0,0 +1,240 @@
|
||||
import { mkdir, stat, writeFile } from 'node:fs/promises';
|
||||
import { dirname } from 'node:path';
|
||||
import type { Config } from '../config.ts';
|
||||
import type { SchulcloudClient } from '../core/client.ts';
|
||||
import { crawl, forEachLimited, type Snapshot } from '../core/crawl.ts';
|
||||
import { extractContent, formatBytes } from '../core/extract.ts';
|
||||
import { mirrorPath, resolveWithin } from '../core/paths.ts';
|
||||
import type { Store } from '../store/store.ts';
|
||||
|
||||
/**
|
||||
* Crawls Schulcloud, persists a generation, mirrors file bytes and indexes
|
||||
* their text.
|
||||
*
|
||||
* Two properties shape this. First, file records are immutable — editing a file
|
||||
* upstream produces a new record — so a file only ever needs downloading and
|
||||
* extracting once, and `file_texts` deliberately outlives the generation that
|
||||
* discovered it. Second, a crawl is ~270 upstream requests from an account that
|
||||
* looks like a student, so concurrent requests coalesce onto one run and a
|
||||
* minimum interval keeps a misbehaving caller from hammering the instance.
|
||||
*/
|
||||
|
||||
export interface IndexResult {
|
||||
crawlId: number;
|
||||
scope: string;
|
||||
courses: number;
|
||||
files: number;
|
||||
mirrored: number;
|
||||
extracted: number;
|
||||
skipped: number;
|
||||
failures: { courseId: string; reason: string }[];
|
||||
durationMs: number;
|
||||
/** Set when the caller joined a run already in progress. */
|
||||
joined?: boolean;
|
||||
}
|
||||
|
||||
export interface IndexerStatus {
|
||||
running: boolean;
|
||||
scope?: string;
|
||||
startedAt?: string;
|
||||
lastResult?: IndexResult;
|
||||
lastError?: string;
|
||||
}
|
||||
|
||||
export class Indexer {
|
||||
private readonly client: SchulcloudClient;
|
||||
private readonly store: Store;
|
||||
private readonly config: Config;
|
||||
private readonly minIntervalMs: number;
|
||||
|
||||
private inFlight = new Map<string, Promise<IndexResult>>();
|
||||
private startedAt: Date | undefined;
|
||||
private runningScope: string | undefined;
|
||||
private lastFinishedAt = new Map<string, number>();
|
||||
private lastResult: IndexResult | undefined;
|
||||
private lastError: string | undefined;
|
||||
|
||||
constructor(client: SchulcloudClient, store: Store, config: Config, minIntervalMs = 60_000) {
|
||||
this.client = client;
|
||||
this.store = store;
|
||||
this.config = config;
|
||||
this.minIntervalMs = minIntervalMs;
|
||||
}
|
||||
|
||||
status(): IndexerStatus {
|
||||
return {
|
||||
running: this.inFlight.size > 0,
|
||||
scope: this.runningScope,
|
||||
startedAt: this.startedAt?.toISOString(),
|
||||
lastResult: this.lastResult,
|
||||
lastError: this.lastError,
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* Re-crawls and re-indexes. `scope` is 'full' or a single course id.
|
||||
*
|
||||
* A per-course refresh is ~3-10 requests against ~270 for a full one, so it
|
||||
* is the right default for "I just uploaded something". Callers arriving
|
||||
* while a run is in progress join it rather than starting a second.
|
||||
*/
|
||||
async refresh(scope: string, options: { force?: boolean } = {}): Promise<IndexResult> {
|
||||
// A full crawl covers every course, so a per-course request can ride along.
|
||||
const existing = this.inFlight.get('full') ?? this.inFlight.get(scope);
|
||||
if (existing) return existing.then((result) => ({ ...result, joined: true }));
|
||||
|
||||
const since = Date.now() - (this.lastFinishedAt.get(scope) ?? 0);
|
||||
if (!options.force && since < this.minIntervalMs) {
|
||||
const wait = Math.ceil((this.minIntervalMs - since) / 1000);
|
||||
throw new Error(
|
||||
`${scope === 'full' ? 'A full re-crawl' : `Course ${scope}`} was refreshed ${Math.round(since / 1000)}s ago. ` +
|
||||
`Wait ${wait}s, or pass force to override — a full crawl is ~270 requests against Schulcloud.`,
|
||||
);
|
||||
}
|
||||
|
||||
const run = this.run(scope).finally(() => {
|
||||
this.inFlight.delete(scope);
|
||||
this.lastFinishedAt.set(scope, Date.now());
|
||||
this.runningScope = undefined;
|
||||
this.startedAt = undefined;
|
||||
});
|
||||
this.inFlight.set(scope, run);
|
||||
this.runningScope = scope;
|
||||
this.startedAt = new Date();
|
||||
return run;
|
||||
}
|
||||
|
||||
private async run(scope: string): Promise<IndexResult> {
|
||||
const began = Date.now();
|
||||
try {
|
||||
const schoolId = (await this.client.me()).school.id;
|
||||
const snapshot: Snapshot = await crawl(this.client, {
|
||||
schoolId,
|
||||
courseIds: scope === 'full' ? undefined : [scope],
|
||||
includeLessonContents: true,
|
||||
includeFiles: true,
|
||||
});
|
||||
|
||||
const crawlId = await this.store.saveSnapshot(snapshot, scope);
|
||||
const { mirrored, extracted, skipped } = await this.ingestFiles(snapshot);
|
||||
|
||||
const result: IndexResult = {
|
||||
crawlId,
|
||||
scope,
|
||||
courses: snapshot.courses.length,
|
||||
files: snapshot.files.length,
|
||||
mirrored,
|
||||
extracted,
|
||||
skipped,
|
||||
failures: snapshot.failures,
|
||||
durationMs: Date.now() - began,
|
||||
};
|
||||
this.lastResult = result;
|
||||
this.lastError = undefined;
|
||||
return result;
|
||||
} catch (error) {
|
||||
this.lastError = error instanceof Error ? error.message : String(error);
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Downloads, mirrors and extracts every file the index has no text for.
|
||||
*
|
||||
* Only files new to the store are touched, so a re-crawl of unchanged
|
||||
* content costs nothing here — which is what makes a 6-hourly crawl cheap
|
||||
* enough to run unattended.
|
||||
*/
|
||||
private async ingestFiles(snapshot: Snapshot): Promise<{ mirrored: number; extracted: number; skipped: number }> {
|
||||
const pending = await this.store.filesNeedingText();
|
||||
if (pending.length === 0) return { mirrored: 0, extracted: 0, skipped: 0 };
|
||||
|
||||
const byId = new Map(snapshot.files.map((file) => [file.record.id, file]));
|
||||
// Same path function the store recorded, so mirror and index agree.
|
||||
const paths = new Map(
|
||||
snapshot.files.map((file) => [file.record.id, mirrorPath(file.at, file.record.name, file.record.id)]),
|
||||
);
|
||||
|
||||
let mirrored = 0;
|
||||
let extracted = 0;
|
||||
let skipped = 0;
|
||||
|
||||
await forEachLimited(pending, 3, async (entry) => {
|
||||
const file = byId.get(entry.fileId);
|
||||
if (!file) return;
|
||||
|
||||
// The instance scans uploads; a file it rejected must not be mirrored.
|
||||
if (file.record.securityCheckStatus === 'blocked') {
|
||||
await this.store.recordFileText({
|
||||
fileId: entry.fileId, name: entry.name, mimeType: entry.mimeType, size: entry.size,
|
||||
content: null, note: 'blocked by the instance virus scanner; not downloaded',
|
||||
mirrorPath: null, mirrorSize: null,
|
||||
});
|
||||
skipped++;
|
||||
return;
|
||||
}
|
||||
|
||||
if (entry.size > this.config.mirrorMaxBytes) {
|
||||
await this.store.recordFileText({
|
||||
fileId: entry.fileId, name: entry.name, mimeType: entry.mimeType, size: entry.size,
|
||||
content: null,
|
||||
note: `too large to mirror (${formatBytes(entry.size)} > ${formatBytes(this.config.mirrorMaxBytes)}); indexed as metadata only`,
|
||||
mirrorPath: null, mirrorSize: null,
|
||||
});
|
||||
skipped++;
|
||||
return;
|
||||
}
|
||||
|
||||
try {
|
||||
const downloaded = await this.client.downloadFile(file.record);
|
||||
const relative = paths.get(entry.fileId);
|
||||
if (!relative) return;
|
||||
const absolute = resolveWithin(this.config.mirrorDir, relative);
|
||||
await mkdir(dirname(absolute), { recursive: true });
|
||||
await writeFile(absolute, downloaded.bytes);
|
||||
mirrored++;
|
||||
|
||||
const extraction = await extractContent(
|
||||
downloaded.bytes,
|
||||
downloaded.mimeType || entry.mimeType,
|
||||
entry.name,
|
||||
this.config.maxExtractedChars,
|
||||
);
|
||||
const content = extraction.kind === 'text' ? (extraction.text ?? '') : null;
|
||||
if (content) extracted++;
|
||||
|
||||
await this.store.recordFileText({
|
||||
fileId: entry.fileId, name: entry.name, mimeType: entry.mimeType, size: entry.size,
|
||||
content, note: extraction.note, mirrorPath: relative, mirrorSize: downloaded.bytes.length,
|
||||
});
|
||||
} catch (error) {
|
||||
// One unreadable file must not abort the crawl; record and move on.
|
||||
await this.store.recordFileText({
|
||||
fileId: entry.fileId, name: entry.name, mimeType: entry.mimeType, size: entry.size,
|
||||
content: null,
|
||||
note: `download or extraction failed: ${error instanceof Error ? error.message : String(error)}`,
|
||||
mirrorPath: null, mirrorSize: null,
|
||||
});
|
||||
skipped++;
|
||||
}
|
||||
});
|
||||
|
||||
return { mirrored, extracted, skipped };
|
||||
}
|
||||
|
||||
/** Verifies a mirrored file is present and the expected size. */
|
||||
async verifyMirror(fileId: string): Promise<{ ok: boolean; path?: string; reason?: string }> {
|
||||
const entry = await this.store.mirrorEntry(fileId);
|
||||
if (!entry) return { ok: false, reason: 'not mirrored' };
|
||||
try {
|
||||
const absolute = resolveWithin(this.config.mirrorDir, entry.path);
|
||||
const info = await stat(absolute);
|
||||
// Size only: the API exposes no ETag or checksum, and hashing would
|
||||
// mean re-downloading every file to learn what it already told us.
|
||||
if (info.size !== entry.size) return { ok: false, path: entry.path, reason: `size ${info.size} != ${entry.size}` };
|
||||
return { ok: true, path: entry.path };
|
||||
} catch (error) {
|
||||
return { ok: false, path: entry.path, reason: error instanceof Error ? error.message : String(error) };
|
||||
}
|
||||
}
|
||||
}
|
||||
80
src/store/db.ts
Normal file
80
src/store/db.ts
Normal file
@@ -0,0 +1,80 @@
|
||||
import { readdir, readFile } from 'node:fs/promises';
|
||||
import { dirname, join } from 'node:path';
|
||||
import { fileURLToPath } from 'node:url';
|
||||
import pg from 'pg';
|
||||
|
||||
/**
|
||||
* Postgres connection and migrations.
|
||||
*
|
||||
* The database is an accelerator, never a source of truth: every fact in it was
|
||||
* read from Schulcloud and can be read again. So nothing here is allowed to
|
||||
* take the server down — `Store.open` returns undefined when the database is
|
||||
* unreachable and the server falls back to live crawls. A Pi that loses its
|
||||
* database should get slower, not broken.
|
||||
*/
|
||||
|
||||
const HERE = dirname(fileURLToPath(import.meta.url));
|
||||
|
||||
export type Db = pg.Pool;
|
||||
|
||||
export interface StoreOptions {
|
||||
connectionString: string;
|
||||
/** Statement timeout, so a runaway query cannot wedge a tool call. */
|
||||
statementTimeoutMs?: number;
|
||||
}
|
||||
|
||||
export async function connect(options: StoreOptions): Promise<Db> {
|
||||
const pool = new pg.Pool({
|
||||
connectionString: options.connectionString,
|
||||
max: 8,
|
||||
idleTimeoutMillis: 30_000,
|
||||
connectionTimeoutMillis: 5_000,
|
||||
statement_timeout: options.statementTimeoutMs ?? 30_000,
|
||||
application_name: 'schulcloud-mcp',
|
||||
});
|
||||
// An idle-client error would otherwise be an unhandled 'error' event and
|
||||
// take the process down; the pool replaces the client on its own.
|
||||
pool.on('error', (error) => console.error('[schulcloud-mcp] postgres idle client error:', error.message));
|
||||
|
||||
const client = await pool.connect();
|
||||
client.release();
|
||||
return pool;
|
||||
}
|
||||
|
||||
/**
|
||||
* Applies any migration files not yet recorded, in filename order.
|
||||
*
|
||||
* Each runs inside a transaction together with the row that records it, so a
|
||||
* failed migration leaves no partial schema and no phantom bookkeeping.
|
||||
*/
|
||||
export async function migrate(db: Db): Promise<string[]> {
|
||||
await db.query(`CREATE TABLE IF NOT EXISTS migrations (
|
||||
name TEXT PRIMARY KEY,
|
||||
applied_at TIMESTAMPTZ NOT NULL DEFAULT now()
|
||||
)`);
|
||||
|
||||
const dir = join(HERE, 'migrations');
|
||||
const files = (await readdir(dir)).filter((name) => name.endsWith('.sql')).sort();
|
||||
const { rows } = await db.query<{ name: string }>('SELECT name FROM migrations');
|
||||
const applied = new Set(rows.map((row) => row.name));
|
||||
|
||||
const ran: string[] = [];
|
||||
for (const file of files) {
|
||||
if (applied.has(file)) continue;
|
||||
const sql = await readFile(join(dir, file), 'utf8');
|
||||
const client = await db.connect();
|
||||
try {
|
||||
await client.query('BEGIN');
|
||||
await client.query(sql);
|
||||
await client.query('INSERT INTO migrations (name) VALUES ($1)', [file]);
|
||||
await client.query('COMMIT');
|
||||
ran.push(file);
|
||||
} catch (error) {
|
||||
await client.query('ROLLBACK').catch(() => {});
|
||||
throw new Error(`migration ${file} failed: ${error instanceof Error ? error.message : String(error)}`);
|
||||
} finally {
|
||||
client.release();
|
||||
}
|
||||
}
|
||||
return ran;
|
||||
}
|
||||
85
src/store/migrations/001_init.sql
Normal file
85
src/store/migrations/001_init.sql
Normal file
@@ -0,0 +1,85 @@
|
||||
-- Trigram matching complements the german dictionary: stemming alone will not
|
||||
-- match "Datenschutz" inside "Datenschutzgrundverordnung", and German compounds
|
||||
-- make that the common case rather than the exception.
|
||||
CREATE EXTENSION IF NOT EXISTS pg_trgm;
|
||||
|
||||
-- One row per crawl. The id is the sync cursor: diffs are computed between
|
||||
-- generations by identity, never from upstream timestamps — the course-board
|
||||
-- projection returns request time as `updatedAt`, so timestamps there are noise.
|
||||
CREATE TABLE IF NOT EXISTS crawls (
|
||||
id BIGSERIAL PRIMARY KEY,
|
||||
started_at TIMESTAMPTZ NOT NULL DEFAULT now(),
|
||||
finished_at TIMESTAMPTZ,
|
||||
status TEXT NOT NULL DEFAULT 'running', -- running | ok | failed
|
||||
scope TEXT NOT NULL DEFAULT 'full', -- 'full' or a course id
|
||||
course_count INTEGER NOT NULL DEFAULT 0,
|
||||
file_count INTEGER NOT NULL DEFAULT 0,
|
||||
error TEXT
|
||||
);
|
||||
|
||||
CREATE INDEX IF NOT EXISTS crawls_finished_idx ON crawls (finished_at DESC) WHERE status = 'ok';
|
||||
|
||||
-- Every entity observed in a given crawl. A partial (per-course) crawl carries
|
||||
-- forward the untouched courses' rows, so each completed generation is a
|
||||
-- complete picture and any two can be diffed directly.
|
||||
CREATE TABLE IF NOT EXISTS nodes (
|
||||
crawl_id BIGINT NOT NULL REFERENCES crawls(id) ON DELETE CASCADE,
|
||||
kind TEXT NOT NULL, -- course | board | lesson | task | file
|
||||
node_id TEXT NOT NULL, -- Schulcloud id
|
||||
course_id TEXT,
|
||||
title TEXT NOT NULL DEFAULT '',
|
||||
body TEXT NOT NULL DEFAULT '',
|
||||
path TEXT NOT NULL DEFAULT '', -- breadcrumb, also the mirror path for files
|
||||
meta JSONB NOT NULL DEFAULT '{}'::jsonb,
|
||||
-- Content fingerprint. Diffing on this rather than on timestamps is what
|
||||
-- makes "changed" meaningful for entities the API reports as always-changed.
|
||||
digest TEXT NOT NULL DEFAULT '',
|
||||
PRIMARY KEY (crawl_id, kind, node_id)
|
||||
);
|
||||
|
||||
CREATE INDEX IF NOT EXISTS nodes_crawl_idx ON nodes (crawl_id);
|
||||
CREATE INDEX IF NOT EXISTS nodes_course_idx ON nodes (crawl_id, course_id);
|
||||
CREATE INDEX IF NOT EXISTS nodes_title_trgm ON nodes USING gin (title gin_trgm_ops);
|
||||
|
||||
-- Extracted file text, keyed by file record id and deliberately NOT scoped to a
|
||||
-- crawl: a Schulcloud file record is immutable — editing a file produces a new
|
||||
-- record — so text extracted once is valid forever and survives every re-crawl.
|
||||
CREATE TABLE IF NOT EXISTS file_texts (
|
||||
file_id TEXT PRIMARY KEY,
|
||||
name TEXT NOT NULL,
|
||||
mime_type TEXT,
|
||||
size BIGINT,
|
||||
content TEXT,
|
||||
extract_note TEXT,
|
||||
extracted_at TIMESTAMPTZ,
|
||||
mirror_path TEXT, -- relative to the mirror root, null if not mirrored
|
||||
mirror_size BIGINT,
|
||||
mirrored_at TIMESTAMPTZ
|
||||
);
|
||||
|
||||
CREATE INDEX IF NOT EXISTS file_texts_name_trgm ON file_texts USING gin (name gin_trgm_ops);
|
||||
|
||||
-- Full-text over the current generation's nodes and over file text. Kept in one
|
||||
-- view-shaped table so a single query covers "titles, board text and the inside
|
||||
-- of PDFs" without the caller unioning by hand.
|
||||
CREATE TABLE IF NOT EXISTS search_docs (
|
||||
crawl_id BIGINT NOT NULL REFERENCES crawls(id) ON DELETE CASCADE,
|
||||
kind TEXT NOT NULL,
|
||||
node_id TEXT NOT NULL,
|
||||
course_id TEXT,
|
||||
course_title TEXT NOT NULL DEFAULT '',
|
||||
title TEXT NOT NULL DEFAULT '',
|
||||
body TEXT NOT NULL DEFAULT '',
|
||||
path TEXT NOT NULL DEFAULT '',
|
||||
-- 'german' gives correct stemming for the content language. Weighted so a
|
||||
-- title hit outranks a body hit for the same query.
|
||||
fts tsvector GENERATED ALWAYS AS (
|
||||
setweight(to_tsvector('german', coalesce(title, '')), 'A') ||
|
||||
setweight(to_tsvector('german', coalesce(body, '')), 'B')
|
||||
) STORED,
|
||||
PRIMARY KEY (crawl_id, kind, node_id)
|
||||
);
|
||||
|
||||
CREATE INDEX IF NOT EXISTS search_docs_fts_idx ON search_docs USING gin (fts);
|
||||
CREATE INDEX IF NOT EXISTS search_docs_trgm_idx ON search_docs USING gin (title gin_trgm_ops);
|
||||
CREATE INDEX IF NOT EXISTS search_docs_crawl_idx ON search_docs (crawl_id);
|
||||
609
src/store/store.ts
Normal file
609
src/store/store.ts
Normal file
@@ -0,0 +1,609 @@
|
||||
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<string, unknown>;
|
||||
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<Store | undefined> {
|
||||
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<void> {
|
||||
await this.db.end().catch(() => {});
|
||||
}
|
||||
|
||||
// --- generations -----------------------------------------------------
|
||||
|
||||
async latestCrawlId(): Promise<number | undefined> {
|
||||
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;
|
||||
}
|
||||
|
||||
/**
|
||||
* 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<number | undefined> {
|
||||
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<number> {
|
||||
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<number | undefined> {
|
||||
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<DiffResult> {
|
||||
const added = await this.db.query<NodeRow>(
|
||||
`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<NodeRow>(
|
||||
`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<NodeRow>(
|
||||
`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<SearchResult[]> {
|
||||
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<SearchRow>(
|
||||
`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<SearchRow>(
|
||||
`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<string>();
|
||||
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<NodeRow & { course_title: string }>(
|
||||
`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<string>();
|
||||
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<NodeRow>(
|
||||
`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<string, unknown> }>(
|
||||
`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<void> {
|
||||
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<string, unknown>;
|
||||
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 immutable, so identity alone decides change; the size
|
||||
// is included only to catch an upstream record being rewritten in place.
|
||||
digest: digestOf([file.record.id, file.record.size]),
|
||||
};
|
||||
}
|
||||
|
||||
async function insertNodes(client: { query: Db['query'] }, crawlId: number, nodes: StoredNode[]): Promise<void> {
|
||||
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<void> {
|
||||
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],
|
||||
);
|
||||
}
|
||||
Reference in New Issue
Block a user