🦄 new(pipeline): sefaria source acquisition (1.1)

.claude/CLAUDE.md

@@ -11,7 +11,7 @@ Targumim, Talmud Babli, Yerushalmi and Midrashic Literature. Deployed as a stati
 - **Runtime:** Browser
 - **Components:** 
 - **Icons:** Font Awesome Pro
-- **Fonts:** Lexend (headings), Atkinson Hyperlegible Next (body), ??? (Hebrew)
+- **Fonts:** Lexend (headings), Atkinson Hyperlegible Next (body), dyslexia-hebrew-extended (Hebrew)
 - **Data:** 
 - **Hosting:** Cloudflare
 - **Lint:** Biome

.coderabbit.yaml

@@ -31,7 +31,8 @@ reviews:
       instructions: |
         General project rules:
         - This is a vanilla JS static site PWA. No bundler, no framework, no npm packages at runtime.
-        - All JS runs in the browser unless under `data/admin/` (which uses Bun).
+        - All JS runs in the browser unless under `admin/` (maintainer tooling, which uses Bun).
+        - Repo layout (v2): `app/` public app, `admin/` tooling (incl. `admin/pipeline/`), `data/` data only.
         - Biome enforces lint and formatting.
         - DOMPurify is loaded via CDN with SRI for XSS sanitization.
 
@@ -44,12 +45,12 @@ reviews:
         - Accessibility (WCAG 2.1 AA compliance)
         - No `var` — use `const`/`let`
 
-    - path: "data/admin/**"
+    - path: "admin/**"
       instructions: |
-        Local dev tooling that runs on Bun (not deployed). Includes:
-        - Admin server (`server.ts`) for annotating dictionary entries
-        - AI classification helper
-        - `console` and `process.env` are expected here
+        Maintainer tooling that runs on Bun (not deployed). Includes:
+        - `admin/pipeline/` — data pipeline from the Sefaria source (fetch, transform, validate, emit)
+        - Admin tool v2 (Phase 3) will live here too
+        - `console` and sequential awaits over streams are expected here
 
     - path: ".github/**"
       instructions: |

.gitignore

@@ -10,3 +10,6 @@ node_modules/
 !.dev.vars.example
 .env*
 !.env.example
+
+# sefaria dump cache (pipeline/fetch.ts)
+.cache/

admin/pipeline/README.md

@@ -0,0 +1,51 @@
+# Pipeline v2
+
+Scripted, re-runnable data pipeline from the true Sefaria source
+([design spec](../docs/specs/2026-07-03-v2-overhaul-design.md), Phase 1+).
+
+## Stage 1 — Source acquisition (`fetch.ts`)
+
+```bash
+bun admin/pipeline/fetch.ts           # download dump + decode + emit
+bun admin/pipeline/fetch.ts --cached  # re-decode from .cache/sefaria (no download)
+```
+
+### Channel decision (spec task 1.1)
+
+The canonical channel is **Sefaria's public MongoDB dump**:
+
+`https://storage.googleapis.com/sefaria-mongo-backup/dump_small.tar.gz`
+
+| Channel | Verdict |
+|---------|---------|
+| MongoDB dump (`lexicon_entry` collection) | **Chosen.** The database Sefaria actually serves, refreshed roughly daily, publicly documented in their [local-install docs](https://developers.sefaria.org/docs/local-installation-instructions) |
+| [Sefaria-Export](https://github.com/Sefaria/Sefaria-Export) | Rejected: texts and links only, no lexicon collections |
+| [Sefaria-Data](https://github.com/Sefaria/Sefaria-Data) | Rejected: import sources for texts; no Jastrow lexicon source |
+| [Words API](https://developers.sefaria.org/reference/get-words) | Rejected: per-word lookup only; a full crawl would need ~30k requests and still reflect the same database the dump snapshots |
+
+The dump is ~2.4 GB compressed (~10+ GB unpacked), so `fetch.ts`
+streams it: gunzip + tar parsing happen in memory and only the three
+lexicon collections are written to disk (`.cache/sefaria/`, gitignored).
+The download is cancelled as soon as all targets are captured.
+
+### One Jastrow lexicon (not two)
+
+Sefaria's code maps a second parent lexicon, `Jastrow Unabbreviated`
+(see `LexiconEntrySubClassMapping` in Sefaria-Project
+`sefaria/model/lexicon.py`), but the deployed database does not carry
+it: the 2026-07-04 dump has no `lexicon` record and zero
+`lexicon_entry` docs under that name. Only `Jastrow Dictionary`
+(32,512 entries) exists and is emitted.
+
+### Outputs (`data/source/`, committed)
+
+| File | Contents |
+|------|----------|
+| `jastrow-dictionary.jsonl` | `lexicon_entry` docs with `parent_lexicon: "Jastrow Dictionary"`, verbatim, dump order, relaxed extended JSON |
+| `lexicons.json` | The Jastrow lexicon registry record |
+| `manifest.json` | Provenance: dump URL, ETag, Last-Modified, fetch time, sha256 + entry count per output |
+
+Documents are emitted **unmodified** — no transformation happens in
+this stage, so `data/source/` is a faithful snapshot for the
+divergence audit (task 1.2). `word_form.bson` is cached for later use
+(search word forms) but not yet emitted.

admin/pipeline/fetch.ts

@@ -0,0 +1,198 @@
+#!/usr/bin/env bun
+/**
+ * Pipeline v2, stage 1: source acquisition (spec task 1.1).
+ *
+ * Streams Sefaria's public MongoDB dump, extracts only the lexicon
+ * collections from the tar (the full dump is ~2.4 GB compressed and is
+ * never written to disk), then decodes the BSON and emits the Jastrow
+ * entries as JSONL plus a provenance manifest.
+ *
+ * Usage:
+ *   bun admin/pipeline/fetch.ts             # full fetch (download + decode)
+ *   bun admin/pipeline/fetch.ts --cached    # decode from .cache/sefaria without downloading
+ *
+ * See admin/pipeline/README.md for the channel decision and output contract.
+ */
+import { mkdir } from 'node:fs/promises';
+import { type Document, EJSON } from 'bson';
+import { bsonDocuments, ChunkReader, extractTargets, sha256 } from './lib.ts';
+
+const DUMP_URL =
+	'https://storage.googleapis.com/sefaria-mongo-backup/dump_small.tar.gz';
+const CACHE_DIR = '.cache/sefaria';
+const OUT_DIR = 'data/source';
+
+/** Tar members to capture, and where each is cached. */
+const TARGETS = new Map([
+	['dump/sefaria/lexicon.bson', `${CACHE_DIR}/lexicon.bson`],
+	['dump/sefaria/lexicon_entry.bson', `${CACHE_DIR}/lexicon_entry.bson`],
+	['dump/sefaria/word_form.bson', `${CACHE_DIR}/word_form.bson`],
+]);
+
+/**
+ * Sefaria's code also maps a 'Jastrow Unabbreviated' lexicon
+ * (LexiconEntrySubClassMapping in Sefaria-Project
+ * sefaria/model/lexicon.py), but the 2026-07-04 dump contains no such
+ * lexicon record and zero entries for it, so only the printed
+ * dictionary is emitted.
+ */
+const JASTROW_LEXICONS = new Map([
+	['Jastrow Dictionary', `${OUT_DIR}/jastrow-dictionary.jsonl`],
+]);
+
+interface DumpProvenance {
+	etag: string;
+	lastModified: string;
+}
+
+async function download(
+	progress: (msg: string) => void,
+): Promise<DumpProvenance> {
+	progress(`downloading ${DUMP_URL}`);
+	const res = await fetch(DUMP_URL);
+	if (!res.ok || res.body === null) {
+		throw new Error(`dump download failed: HTTP ${res.status}`);
+	}
+	const provenance: DumpProvenance = {
+		etag: res.headers.get('etag') ?? '',
+		lastModified: res.headers.get('last-modified') ?? '',
+	};
+	const tar = res.body.pipeThrough(new DecompressionStream('gzip'));
+	const reader = new ChunkReader(tar[Symbol.asyncIterator]());
+	const missing = await extractTargets(reader, TARGETS, progress);
+	if (missing.size > 0) {
+		throw new Error(
+			`archive ended before extracting: ${[...missing].join(', ')}`,
+		);
+	}
+	// All targets captured; stop pulling the remainder of the ~2.4 GB body.
+	await tar.cancel().catch(() => progress('download stream already closed'));
+	await Bun.write(
+		`${CACHE_DIR}/provenance.json`,
+		`${JSON.stringify(provenance, undefined, '\t')}\n`,
+	);
+	return provenance;
+}
+
+async function loadProvenance(
+	cached: boolean,
+	progress: (msg: string) => void,
+): Promise<DumpProvenance> {
+	if (!cached) {
+		return await download(progress);
+	}
+	for (const path of TARGETS.values()) {
+		if (!(await Bun.file(path).exists())) {
+			throw new Error(
+				`--cached given but ${path} is missing; run without --cached first`,
+			);
+		}
+	}
+	progress('using cached collections');
+	const stored = Bun.file(`${CACHE_DIR}/provenance.json`);
+	if (await stored.exists()) {
+		return (await stored.json()) as DumpProvenance;
+	}
+	return { etag: '', lastModified: '' };
+}
+
+async function emitRegistry(progress: (msg: string) => void): Promise<string> {
+	const registry: Document[] = [];
+	for await (const doc of bsonDocuments(
+		TARGETS.get('dump/sefaria/lexicon.bson') as string,
+	)) {
+		if (JASTROW_LEXICONS.has(doc['name'] as string)) {
+			registry.push(doc);
+		}
+	}
+	if (registry.length !== JASTROW_LEXICONS.size) {
+		// Fail fast on lexicon-collection schema drift (renamed name
+		// field, renamed lexicon) instead of emitting empty outputs.
+		throw new Error(
+			`expected ${JASTROW_LEXICONS.size} lexicon record(s), found ${registry.length}`,
+		);
+	}
+	const registryPath = `${OUT_DIR}/lexicons.json`;
+	await Bun.write(
+		registryPath,
+		`${EJSON.stringify(registry, undefined, '\t', { relaxed: true })}\n`,
+	);
+	progress(`wrote ${registryPath} (${registry.length} lexicon records)`);
+	return registryPath;
+}
+
+/**
+ * Emit entries verbatim as relaxed extended JSON, one file per lexicon,
+ * preserving dump order.
+ */
+async function emitEntries(
+	progress: (msg: string) => void,
+): Promise<Map<string, number>> {
+	const counts = new Map<string, number>();
+	const writers = new Map(
+		[...JASTROW_LEXICONS].map(([lexicon, path]) => {
+			counts.set(lexicon, 0);
+			return [lexicon, Bun.file(path).writer()];
+		}),
+	);
+	for await (const doc of bsonDocuments(
+		TARGETS.get('dump/sefaria/lexicon_entry.bson') as string,
+	)) {
+		const parentLexicon = doc['parent_lexicon'] as string;
+		const writer = writers.get(parentLexicon);
+		if (writer === undefined) {
+			continue;
+		}
+		writer.write(`${EJSON.stringify(doc, { relaxed: true })}\n`);
+		counts.set(parentLexicon, (counts.get(parentLexicon) ?? 0) + 1);
+	}
+	for (const [lexicon, writer] of writers) {
+		await writer.end();
+		if ((counts.get(lexicon) ?? 0) === 0) {
+			// Same drift guard as the registry: a renamed parent_lexicon
+			// must not produce an empty snapshot that looks like success.
+			throw new Error(`no entries found for lexicon "${lexicon}"`);
+		}
+		progress(
+			`wrote ${JASTROW_LEXICONS.get(lexicon)} (${counts.get(lexicon)} entries)`,
+		);
+	}
+	return counts;
+}
+
+async function main(): Promise<void> {
+	const cached = Bun.argv.includes('--cached');
+	const progress = (msg: string): void => {
+		console.log(`[fetch] ${msg}`);
+	};
+	await mkdir(CACHE_DIR, { recursive: true });
+	await mkdir(OUT_DIR, { recursive: true });
+
+	const dumpProvenance = await loadProvenance(cached, progress);
+	const registryPath = await emitRegistry(progress);
+	const counts = await emitEntries(progress);
+
+	const manifest = {
+		entryCounts: Object.fromEntries(counts),
+		fetchedAt: new Date().toISOString(),
+		outputs: await Promise.all(
+			[registryPath, ...JASTROW_LEXICONS.values()].map(async (path) => ({
+				path,
+				sha256: await sha256(path),
+			})),
+		),
+		source: {
+			url: DUMP_URL,
+			etag: dumpProvenance.etag,
+			lastModified: dumpProvenance.lastModified,
+		},
+	};
+	const manifestPath = `${OUT_DIR}/manifest.json`;
+	await Bun.write(
+		manifestPath,
+		`${JSON.stringify(manifest, undefined, '\t')}\n`,
+	);
+	progress(`wrote ${manifestPath}`);
+}
+
+await main();