From 2a427429b8ac1c0b77fd378701369005083b98a9 Mon Sep 17 00:00:00 2001 From: Luis Guzman Date: Thu, 23 Jul 2026 18:15:57 -0600 Subject: [PATCH 1/3] ADFA-4838: durable job engine core (JobManager + sqlite) + kiwix runner MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Phase 2 of ADFA-4832. Module-scoped, persisted job engine so a content op is owned by the dashboard process (not a socket closure) and survives a client disconnect and a dashboard restart. - jobs.ts: JobManager with a better-sqlite3 journal (jobs table), per-type Runner registry, structured progress (phase/percent/bytes-per-sec), cancellation that kills tracked children, and reconcileOnBoot() to resume interrupted jobs. - kiwix.exec.ts: kiwix runner ported from the Phase 1 socket handler — aria2 download (structured %/rate parsing) + iiab-make-kiwix-lib index + metadata cleanup + a free- space guard. aria2 flags kept in sync with the app + the socket handler. - ADR-4832 refreshed as the accepted full-Phase-2 design of record. Typechecks (tsc --noEmit). Not wired into server.ts yet (REST routes + reconcile call come next); no runtime behavior change until then. --- .../docs/ADR-4832-live-content-channel.md | 30 ++- static/dashboard/sockets/jobs.ts | 198 ++++++++++++++++++ static/dashboard/sockets/kiwix.exec.ts | 133 ++++++++++++ 3 files changed, 360 insertions(+), 1 deletion(-) create mode 100644 static/dashboard/sockets/jobs.ts create mode 100644 static/dashboard/sockets/kiwix.exec.ts diff --git a/controller/docs/ADR-4832-live-content-channel.md b/controller/docs/ADR-4832-live-content-channel.md index db055a82c..8b214a695 100644 --- a/controller/docs/ADR-4832-live-content-channel.md +++ b/controller/docs/ADR-4832-live-content-channel.md @@ -1,6 +1,6 @@ # ADR — Content management on the live system (app ↔ in-server channel) -Status: Accepted. Phase 1 (socket.io bridge) in progress; Phase 2 (REST + durable jobs) planned. Ticket: ADFA-4832. +Status: Accepted. Phase 1 (socket.io bridge) shipped (ADFA-4832). Phase 2 (REST + durable jobs, full scope) accepted and in progress — see "Phase 2 (accepted) — design of record" below. Tickets: ADFA-4832 (epic thread), plus the Phase 2 tickets listed in that section. ## Context @@ -29,6 +29,34 @@ Status: Accepted. Phase 1 (socket.io bridge) in progress; Phase 2 (REST + durabl - Maps/books on a live system still need the same migration (follow-up). Phase 1 skips the maps proot on the live path to avoid re-introducing a collision. - If the app process is fully killed, the job dies (server kills on disconnect). Phase 2 durable jobs remove this. +## Phase 2 (accepted) — design of record + +Scope decision: **full Phase 2 in one rootfs rebuild** — kiwix + maps + books on the REST engine, IPv4/aria2 parity, and the dashboard packaging (build + supervision). Iteration decision: a **dev "push static/" path** copies the updated dashboard into the installed rootfs on-device and restarts `dash-node`, so we don't pay the ~2h rebuild per test; the full rebuild happens once at the end. + +### Server (in-server dashboard) + +1. **Job manager, module-scoped (not per-socket).** The download+index job is owned by the dashboard process, not by any connection. A client `disconnect` no longer kills anything (Phase 1 killed on disconnect). This is the core durability change. +2. **Durable journal via `better-sqlite3`** (already a dependency). Table `jobs(id, type, target, phase, percent, speed, error, created, updated)`. On dashboard boot, reconcile: incomplete download jobs resume (aria2 `--continue` already resumes partials), indexing re-runs if it hadn't finished. +3. **Progress parsed server-side into structured fields.** aria2/index stdout is parsed on the server into `{phase: downloading|indexing|done|error, percent, speedBytesPerSec, error}` and stored on the job. Clients read structured JSON — no more client-side terminal-scraping. +4. **REST contract (localhost:8085 → :4000), short calls:** + - `POST /api/{kiwix|maps|books}/download {ids:[...]}` → `{jobId}` + - `GET /api/{...}/jobs/:id` → the structured job row + - `POST /api/{...}/jobs/:id/cancel` + - `GET /api/{...}/catalog`, `DELETE /api/{...}/item/:id` + The socket.io handlers stay only as a thin compat shell (or are removed) — the job manager is the single source of truth. +5. **aria2 IPv4 profiler parity.** Port the app's IPv4-preference behavior to the server aria2 spawn (closes a divergence noted below). +6. **Packaging.** Compile TS→JS (drop `ts-node` dev mode), run under `pdsm` supervision with a health check; `dash-node` restarts cleanly on failure. This — not a separate cmdsrv — is the answer to "Node is fragile". + +### Android + +7. **REST content client replaces the socket.io `LiveContentClient`.** Short localhost calls: `POST` to start, **poll `GET` every ~1s** for structured progress, `POST cancel`. Owned by the foreground `InstallService`. There is no long-lived connection to lose; the job survives UI/config churn and even a dashboard/socket restart (durable job + `--continue`). Wire into `InstallService.downloadAndIndexKiwix` and the maps/books paths. + +### Tickets + +- Dashboard: durable REST job engine + endpoints + sqlite + aria2 IPv4 parity (kiwix/maps/books). +- Dashboard packaging: TS→JS build + pdsm supervision/health + dev push-static path. +- Android: REST content client (foreground poll) replacing socket.io Phase 1. + ## References `PRootEngine.java`, `InstallService.downloadAndIndexKiwix`, `ServerController`, `static/dashboard/server.ts` + `sockets/kiwix.socket.ts`, `LiveContentClient.java`. Ticket ADFA-4832. diff --git a/static/dashboard/sockets/jobs.ts b/static/dashboard/sockets/jobs.ts new file mode 100644 index 000000000..37ebf2041 --- /dev/null +++ b/static/dashboard/sockets/jobs.ts @@ -0,0 +1,198 @@ +// sockets/jobs.ts — ADFA-4838 (Phase 2 of ADFA-4832) +// +// Durable, module-scoped job engine for live content operations. Unlike the Phase 1 +// socket handlers (job state lived in the socket closure and was SIGKILLed on +// disconnect), a job here is owned by the dashboard process and persisted to sqlite, +// so it survives a client disconnect AND a dashboard restart. Each content type +// (kiwix/maps/books) registers a Runner; the manager owns the lifecycle, persistence, +// cancellation and boot-time reconciliation. Progress is stored structured (phase + +// percent + bytes/sec), so clients never scrape terminal text. +import Database from 'better-sqlite3'; +import { spawn, ChildProcess, SpawnOptions } from 'child_process'; +import fs from 'fs'; +import path from 'path'; + +export type JobType = 'kiwix' | 'maps' | 'books'; +export type JobPhase = + | 'queued' | 'downloading' | 'indexing' | 'processing' + | 'done' | 'error' | 'canceled'; + +/** A persisted job row. `percent` is -1 when indeterminate; `speed` is bytes/sec. */ +export interface Job { + id: string; + type: JobType; + target: string; // JSON-encoded string[] of item ids + phase: JobPhase; + percent: number; + speed: number; + detail: string | null; // free-form current item / message + error: string | null; + created: number; + updated: number; +} + +export interface JobUpdate { + phase?: JobPhase; + percent?: number; + speed?: number; + detail?: string | null; + error?: string | null; +} + +/** Passed to a runner: report progress, check cancellation, spawn tracked children. */ +export interface RunnerContext { + readonly job: Job; + readonly ids: string[]; + update(patch: JobUpdate): void; + isCanceled(): boolean; + throwIfCanceled(): void; + /** Spawn a child that is killed automatically on cancel and untracked on exit. */ + spawn(cmd: string, args: string[], opts?: SpawnOptions): ChildProcess; + log(line: string): void; +} + +export type Runner = (ctx: RunnerContext) => Promise; + +/** Thrown by throwIfCanceled(); mapped to phase 'canceled' rather than 'error'. */ +export class CanceledError extends Error { + constructor() { super('canceled'); this.name = 'CanceledError'; } +} + +const DB_PATH = process.env.K2GO_JOBS_DB || '/library/dashboard/jobs.db'; +const ACTIVE: JobPhase[] = ['queued', 'downloading', 'indexing', 'processing']; + +class JobManager { + private db: Database.Database; + private runners = new Map(); + private runtime = new Map }>(); + + constructor() { + try { fs.mkdirSync(path.dirname(DB_PATH), { recursive: true }); } catch { /* best effort */ } + this.db = new Database(DB_PATH); + this.db.pragma('journal_mode = WAL'); + this.db.exec(`CREATE TABLE IF NOT EXISTS jobs ( + id TEXT PRIMARY KEY, + type TEXT NOT NULL, + target TEXT NOT NULL, + phase TEXT NOT NULL, + percent INTEGER NOT NULL DEFAULT -1, + speed INTEGER NOT NULL DEFAULT 0, + detail TEXT, + error TEXT, + created INTEGER NOT NULL, + updated INTEGER NOT NULL + );`); + } + + registerRunner(type: JobType, runner: Runner): void { + this.runners.set(type, runner); + } + + /** Create + start a job. Returns the persisted row immediately (phase 'queued'). */ + create(type: JobType, ids: string[]): Job { + const now = Date.now(); + const id = `${type}-${now}-${Math.random().toString(36).slice(2, 8)}`; + const job: Job = { + id, type, target: JSON.stringify(ids), phase: 'queued', + percent: -1, speed: 0, detail: null, error: null, created: now, updated: now, + }; + this.db.prepare( + `INSERT INTO jobs (id,type,target,phase,percent,speed,detail,error,created,updated) + VALUES (@id,@type,@target,@phase,@percent,@speed,@detail,@error,@created,@updated)` + ).run(job); + this.launch(job); + return job; + } + + get(id: string): Job | undefined { + return this.db.prepare(`SELECT * FROM jobs WHERE id = ?`).get(id) as Job | undefined; + } + + list(type?: JobType): Job[] { + return (type + ? this.db.prepare(`SELECT * FROM jobs WHERE type = ? ORDER BY created DESC`).all(type) + : this.db.prepare(`SELECT * FROM jobs ORDER BY created DESC`).all()) as Job[]; + } + + /** Cancel a running job (kills its children) and marks it 'canceled'. */ + cancel(id: string): boolean { + const job = this.get(id); + if (!job) return false; + const rt = this.runtime.get(id); + if (rt) { + rt.canceled = true; + for (const p of rt.procs) { try { p.kill('SIGKILL'); } catch { /* already gone */ } } + } + if (ACTIVE.includes(job.phase)) this.patch(id, { phase: 'canceled' }); + return true; + } + + /** Resume jobs that were mid-flight when the dashboard last stopped. */ + reconcileOnBoot(): void { + const stuck = this.db.prepare( + `SELECT * FROM jobs WHERE phase IN ('queued','downloading','indexing','processing')` + ).all() as Job[]; + for (const job of stuck) this.launch(job); + } + + private patch(id: string, patch: JobUpdate): void { + const cur = this.get(id); + if (!cur) return; + const next: Job = { + ...cur, + phase: patch.phase ?? cur.phase, + percent: patch.percent ?? cur.percent, + speed: patch.speed ?? cur.speed, + detail: patch.detail !== undefined ? patch.detail : cur.detail, + error: patch.error !== undefined ? patch.error : cur.error, + updated: Date.now(), + }; + this.db.prepare( + `UPDATE jobs SET phase=@phase,percent=@percent,speed=@speed,detail=@detail,error=@error,updated=@updated WHERE id=@id` + ).run(next); + } + + private launch(job: Job): void { + const runner = this.runners.get(job.type); + if (!runner) { this.patch(job.id, { phase: 'error', error: `no runner for ${job.type}` }); return; } + + const rt = { canceled: false, procs: new Set() }; + this.runtime.set(job.id, rt); + + let ids: string[] = []; + try { ids = JSON.parse(job.target) as string[]; } catch { ids = []; } + + const ctx: RunnerContext = { + job, ids, + update: (p) => this.patch(job.id, p), + isCanceled: () => rt.canceled, + throwIfCanceled: () => { if (rt.canceled) throw new CanceledError(); }, + spawn: (cmd, args, opts) => { + const child = opts ? spawn(cmd, args, opts) : spawn(cmd, args); + rt.procs.add(child); + child.on('exit', () => rt.procs.delete(child)); + return child; + }, + log: (line) => console.log(`[job ${job.id}] ${line}`), + }; + + Promise.resolve() + .then(() => runner(ctx)) + .then(() => { + const cur = this.get(job.id); + if (cur && ACTIVE.includes(cur.phase)) this.patch(job.id, { phase: 'done', percent: 100 }); + }) + .catch((err: unknown) => { + if (err instanceof CanceledError || rt.canceled) { + this.patch(job.id, { phase: 'canceled' }); + } else { + const msg = err instanceof Error ? err.message : String(err); + this.patch(job.id, { phase: 'error', error: msg }); + } + }) + .finally(() => { this.runtime.delete(job.id); }); + } +} + +/** Singleton shared by every type's runner and the REST routes. */ +export const jobs = new JobManager(); diff --git a/static/dashboard/sockets/kiwix.exec.ts b/static/dashboard/sockets/kiwix.exec.ts new file mode 100644 index 000000000..0804d10a9 --- /dev/null +++ b/static/dashboard/sockets/kiwix.exec.ts @@ -0,0 +1,133 @@ +// sockets/kiwix.exec.ts — ADFA-4838 +// +// Kiwix runner for the durable job engine: download the requested ZIM(s) with aria2, +// then rebuild the Kiwix library index. Progress is reported structured (percent + +// bytes/sec) instead of streamed as terminal text. Ported from the Phase 1 +// kiwix.socket handler, minus the socket/closure lifetime — the job outlives any +// client (see jobs.ts). +import { jobs, RunnerContext, CanceledError } from './jobs'; +import { execSync } from 'child_process'; +import fs from 'fs'; +import path from 'path'; + +const ZIMS_DIR = '/library/zims/content/'; +const BASE_URL = 'https://download.kiwix.org/zim/wikipedia/'; +const INDEXER = '/usr/bin/iiab-make-kiwix-lib'; +const SAFETY_BUFFER_BYTES = 5 * 1024 * 1024 * 1024; // keep >=5 GB free + +// ADFA-4832 sync note: these aria2 flags mirror the app-side downloader +// (controller/app/.../Aria2Manager.java) and the Phase 1 socket handler. Keep the +// three in sync; if you change one, change the others and document any divergence. +const ARIA2_ARGS: string[] = [ + '-d', ZIMS_DIR, + '--continue=true', + '--allow-overwrite=true', + '--auto-file-renaming=false', + '--max-connection-per-server=4', + '--split=16', + '--follow-metalink=mem', + '--check-integrity=true', + '--console-log-level=warn', + '--summary-interval=1', + '--download-result=hide', + '--async-dns=false', + '-Z', + '-j', '5', +]; + +/** Convert an aria2 rate token ("34MiB", "512KiB", "1.2MB") to bytes/sec. */ +function parseRate(token: string): number { + const m = /^([\d.]+)\s*([KMGT]?i?B)?/i.exec(token); + if (!m) return 0; + const val = parseFloat(m[1]); + const unit = (m[2] || 'B').toUpperCase(); + const mult: Record = { + B: 1, KIB: 1024, MIB: 1024 ** 2, GIB: 1024 ** 3, TIB: 1024 ** 4, + KB: 1000, MB: 1e6, GB: 1e9, TB: 1e12, + }; + return Math.round(val * (mult[unit] ?? 1)); +} + +/** Best-effort free-space guard so we never fill the disk. */ +function assertFreeSpace(): void { + try { + const df = execSync('df -k /').toString().trim().split('\n'); + const cols = df[df.length - 1].split(/\s+/); + const availableBytes = parseInt(cols[3], 10) * 1024; + if (availableBytes < SAFETY_BUFFER_BYTES) { + const freeGB = (availableBytes / 1024 ** 3).toFixed(1); + throw new Error(`Not enough free space (${freeGB} GB); need a >5 GB buffer.`); + } + } catch (e) { + if (e instanceof Error && e.message.startsWith('Not enough')) throw e; + // df failed for some OS reason — proceed with caution rather than blocking. + } + // TODO(ADFA-4838): also subtract the requested ZIMs' catalog sizes once the + // catalog is wired into the engine (Phase 1 did this with getKiwixCatalog()). +} + +function cleanupMetadata(): void { + try { + for (const f of fs.readdirSync(ZIMS_DIR)) { + if (/\.(meta4|aria2|torrent)$/.test(f)) fs.unlinkSync(path.join(ZIMS_DIR, f)); + } + } catch { /* non-fatal */ } +} + +const kiwixRunner: (ctx: RunnerContext) => Promise = async (ctx) => { + const zims = ctx.ids.map((z) => path.basename(z)).filter((z) => z.endsWith('.zim')); + if (zims.length === 0) throw new Error('no ZIMs requested'); + + assertFreeSpace(); + ctx.throwIfCanceled(); + + // --- Download phase ----------------------------------------------------- + ctx.update({ phase: 'downloading', percent: 0, speed: 0, detail: zims.join(', ') }); + const urls = zims.map((z) => BASE_URL + z); + + await new Promise((resolve, reject) => { + const dl = ctx.spawn('/usr/bin/aria2c', [...ARIA2_ARGS, ...urls]); + const onData = (buf: Buffer) => { + const text = buf.toString(); + // A single chunk can carry several summary lines; take the LAST %/rate. + const re = /\((\d+)%\).*?DL:([^\s]+)/g; + let m: RegExpExecArray | null; + let lastPct = -1; + let lastRate = ''; + while ((m = re.exec(text)) !== null) { lastPct = parseInt(m[1], 10); lastRate = m[2]; } + if (lastPct >= 0) ctx.update({ phase: 'downloading', percent: lastPct, speed: parseRate(lastRate) }); + }; + dl.stdout?.on('data', onData); + dl.stderr?.on('data', onData); + dl.on('error', reject); + dl.on('exit', (code, signal) => { + if (signal === 'SIGKILL' || ctx.isCanceled()) return reject(new CanceledError()); + if (code === 0) resolve(); + else reject(new Error(`aria2 exited with code ${code}`)); + }); + }); + + cleanupMetadata(); + ctx.throwIfCanceled(); + + // --- Index phase -------------------------------------------------------- + if (fs.existsSync(INDEXER)) { + ctx.update({ phase: 'indexing', percent: -1, speed: 0 }); + await new Promise((resolve, reject) => { + const idx = ctx.spawn(INDEXER, []); + idx.stdout?.on('data', (d: Buffer) => ctx.log(d.toString().trim())); + idx.stderr?.on('data', (d: Buffer) => ctx.log(d.toString().trim())); + idx.on('error', reject); + idx.on('exit', (code, signal) => { + if (signal === 'SIGKILL' || ctx.isCanceled()) return reject(new CanceledError()); + resolve(); // indexing exit code is advisory; the catalog reload is the real signal + }); + }); + } + + ctx.update({ phase: 'done', percent: 100, speed: 0 }); +}; + +jobs.registerRunner('kiwix', kiwixRunner); + +export { kiwixRunner }; From 30ac091699fe4b9a0ad27a1d82f5bf0249d1ccc4 Mon Sep 17 00:00:00 2001 From: Luis Guzman Date: Thu, 23 Jul 2026 18:25:43 -0600 Subject: [PATCH 2/3] ADFA-4838: REST endpoints + wire the job engine into server.ts - routes.ts: POST /api/:type/download, GET /api/:type/jobs(/:id), POST .../cancel. Stateless calls over the durable engine; clients re-attach by polling the id. - server.ts: mount express.json() + /api router, import the kiwix runner (registers it), and call jobs.reconcileOnBoot() on listen so interrupted jobs resume. - nginx: add a localhost-only /api/ proxy to :4000 (mirrors the /dashboard/ + /socket.io/ locations). Kiwix now works end-to-end over REST. maps/books runners and retiring the socket-owned job paths are the next steps. Typechecks (tsc --noEmit). --- static/dashboard/dash-node-nginx.conf | 15 +++++++ static/dashboard/routes.ts | 64 +++++++++++++++++++++++++++ static/dashboard/server.ts | 14 +++++- 3 files changed, 92 insertions(+), 1 deletion(-) create mode 100644 static/dashboard/routes.ts diff --git a/static/dashboard/dash-node-nginx.conf b/static/dashboard/dash-node-nginx.conf index f29fd05ff..1f65ee51a 100644 --- a/static/dashboard/dash-node-nginx.conf +++ b/static/dashboard/dash-node-nginx.conf @@ -15,6 +15,21 @@ location ^~ /dashboard/ { proxy_set_header X-Forwarded-Proto $scheme; } +# Proxy for the REST content API (ADFA-4838, localhost only) +location ^~ /api/ { + allow 127.0.0.1; + allow ::1; + deny all; + + error_page 403 =404 /404.html; + + proxy_pass http://127.0.0.1:4000/api/; + proxy_set_header Host $http_host; + proxy_set_header X-Real-IP $remote_addr; + proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for; + proxy_set_header X-Forwarded-Proto $scheme; +} + # Proxy for WebSockets (socket.io uses it by default) location ^~ /socket.io/ { # Basic security (localhost only) diff --git a/static/dashboard/routes.ts b/static/dashboard/routes.ts new file mode 100644 index 000000000..1f7828c09 --- /dev/null +++ b/static/dashboard/routes.ts @@ -0,0 +1,64 @@ +// routes.ts — ADFA-4838 +// +// REST surface over the durable job engine. Short, stateless calls the app (and the +// web UI) use instead of a long-lived socket: start a job, poll its structured status, +// cancel it. The job itself lives in the dashboard process (see sockets/jobs.ts), so +// none of these calls hold state — a client can drop and re-attach by polling the id. +import express, { Router, Request, Response } from 'express'; +import { jobs, Job, JobType } from './sockets/jobs'; + +const VALID_TYPES: JobType[] = ['kiwix', 'maps', 'books']; +function isType(t: string): t is JobType { + return (VALID_TYPES as string[]).includes(t); +} + +/** Public shape returned to clients: ids as an array, no raw JSON column. */ +function toApi(job: Job) { + let ids: string[] = []; + try { ids = JSON.parse(job.target) as string[]; } catch { ids = []; } + return { + id: job.id, + type: job.type, + ids, + phase: job.phase, + percent: job.percent, + speed: job.speed, + detail: job.detail, + error: job.error, + updated: job.updated, + }; +} + +export const apiRouter: Router = express.Router(); + +// Start a content job → 202 { ...job } +apiRouter.post('/:type/download', (req: Request, res: Response): void => { + const type = String(req.params.type); + if (!isType(type)) { res.status(404).json({ error: 'unknown type' }); return; } + const body = req.body as { ids?: unknown }; + const ids = Array.isArray(body?.ids) ? body.ids.map((x) => String(x)) : []; + if (ids.length === 0) { res.status(400).json({ error: 'ids required' }); return; } + res.status(202).json(toApi(jobs.create(type, ids))); +}); + +// Poll one job's structured status. +apiRouter.get('/:type/jobs/:id', (req: Request, res: Response): void => { + const job = jobs.get(String(req.params.id)); + if (!job || job.type !== String(req.params.type)) { res.status(404).json({ error: 'not found' }); return; } + res.json(toApi(job)); +}); + +// List a type's jobs (most recent first). +apiRouter.get('/:type/jobs', (req: Request, res: Response): void => { + const type = String(req.params.type); + if (!isType(type)) { res.status(404).json({ error: 'unknown type' }); return; } + res.json(jobs.list(type).map(toApi)); +}); + +// Cancel a running job. +apiRouter.post('/:type/jobs/:id/cancel', (req: Request, res: Response): void => { + const job = jobs.get(String(req.params.id)); + if (!job || job.type !== String(req.params.type)) { res.status(404).json({ error: 'not found' }); return; } + jobs.cancel(job.id); + res.json({ ok: true }); +}); diff --git a/static/dashboard/server.ts b/static/dashboard/server.ts index 780cf797b..262620f98 100644 --- a/static/dashboard/server.ts +++ b/static/dashboard/server.ts @@ -10,6 +10,12 @@ import { handleKiwixEvents } from './sockets/kiwix.socket'; import { handleHomeEvents } from './sockets/home.socket'; import { handleBooksEvents } from './sockets/books.socket'; +// ADFA-4838: durable REST job engine. Importing the runner modules registers them with +// the shared JobManager; the router exposes them over /api. +import { jobs } from './sockets/jobs'; +import './sockets/kiwix.exec'; +import { apiRouter } from './routes'; + const app = express(); const server = http.createServer(app); const io = new Server(server); @@ -28,9 +34,13 @@ app.set('view engine', 'ejs'); app.set('views', path.join(__dirname, 'views')); app.use(express.static(path.join(__dirname, 'public'))); +// ADFA-4838: JSON body parsing + the REST content API. +app.use(express.json()); +app.use('/api', apiRouter); + // Main route app.get('/', (req, res) => { - res.render('index'); + res.render('index'); }); // Main Socket Connection Handler @@ -53,6 +63,8 @@ server.listen(PORT, () => { console.log(`===========================================`); console.log(`K2Go Dashboard active on port ${PORT}`); console.log(`===========================================`); + // ADFA-4838: resume any content jobs that were mid-flight before a restart. + try { jobs.reconcileOnBoot(); } catch (e) { console.error('[jobs] reconcile failed', e); } }); // ========================================== From 9c6b7dcdeb8462db830a91b8a481c767349c0293 Mon Sep 17 00:00:00 2001 From: Luis Guzman Date: Thu, 23 Jul 2026 18:44:46 -0600 Subject: [PATCH 3/3] ADFA-4838: maps + books runners (migrate to the REST engine like kiwix) Port the remaining live-content types off the Phase 1 socket handlers onto the durable job engine, so all three run the same way. - Generalize the job payload: items are now arbitrary (strings for kiwix ZIMs; objects for maps/books). RunnerContext exposes items (raw) + ids (string convenience); routes accept { items } or { ids }. - maps.exec.ts: tile-extract.py extract, reusing the socket handler's box validation; structured progress (parses a % when the script prints one). - books.exec.ts: Calibre-Web auth/CSRF + Gutenberg fetch + upload, per-book progress. Deletion (maps/books) stays a separate op for now. Typechecks (tsc --noEmit). --- static/dashboard/routes.ts | 11 ++- static/dashboard/server.ts | 2 + static/dashboard/sockets/books.exec.ts | 120 +++++++++++++++++++++++++ static/dashboard/sockets/jobs.ts | 17 ++-- static/dashboard/sockets/maps.exec.ts | 55 ++++++++++++ 5 files changed, 195 insertions(+), 10 deletions(-) create mode 100644 static/dashboard/sockets/books.exec.ts create mode 100644 static/dashboard/sockets/maps.exec.ts diff --git a/static/dashboard/routes.ts b/static/dashboard/routes.ts index 1f7828c09..b30e59880 100644 --- a/static/dashboard/routes.ts +++ b/static/dashboard/routes.ts @@ -35,10 +35,13 @@ export const apiRouter: Router = express.Router(); apiRouter.post('/:type/download', (req: Request, res: Response): void => { const type = String(req.params.type); if (!isType(type)) { res.status(404).json({ error: 'unknown type' }); return; } - const body = req.body as { ids?: unknown }; - const ids = Array.isArray(body?.ids) ? body.ids.map((x) => String(x)) : []; - if (ids.length === 0) { res.status(400).json({ error: 'ids required' }); return; } - res.status(202).json(toApi(jobs.create(type, ids))); + // kiwix sends { ids: ["file.zim"] }; maps/books send { items: [ {...} ] }. + const body = req.body as { ids?: unknown; items?: unknown }; + const items: unknown[] = Array.isArray(body?.items) + ? body.items + : Array.isArray(body?.ids) ? body.ids : []; + if (items.length === 0) { res.status(400).json({ error: 'items (or ids) required' }); return; } + res.status(202).json(toApi(jobs.create(type, items))); }); // Poll one job's structured status. diff --git a/static/dashboard/server.ts b/static/dashboard/server.ts index 262620f98..6060c3ac8 100644 --- a/static/dashboard/server.ts +++ b/static/dashboard/server.ts @@ -14,6 +14,8 @@ import { handleBooksEvents } from './sockets/books.socket'; // the shared JobManager; the router exposes them over /api. import { jobs } from './sockets/jobs'; import './sockets/kiwix.exec'; +import './sockets/maps.exec'; +import './sockets/books.exec'; import { apiRouter } from './routes'; const app = express(); diff --git a/static/dashboard/sockets/books.exec.ts b/static/dashboard/sockets/books.exec.ts new file mode 100644 index 000000000..f94311ff9 --- /dev/null +++ b/static/dashboard/sockets/books.exec.ts @@ -0,0 +1,120 @@ +// sockets/books.exec.ts — ADFA-4838 +// +// Books runner for the durable job engine: for each requested Gutenberg book, download +// the EPUB and upload it to the local Calibre-Web. Ported from the Phase 1 books.socket +// download_books_batch handler (auth/CSRF + fetch + upload), made durable and reporting +// structured per-book progress. A job item is { id, title, url }. +import { jobs, RunnerContext } from './jobs'; +import fs from 'fs'; +import path from 'path'; + +const CALIBRE_WEB_LOCAL_URL = 'http://127.0.0.1:8083'; +const TMP_DIR = '/tmp/books_downloader/'; +const SYSTEM_USER_AGENT = 'K2Go Dashboard/1.0 (https://github.com/appdevforall/KnowledgeToGo)'; +// Defaults match the Phase 1 handler; a credential override can be added later if needed. +const CALIBRE_WEB_USER = 'Admin'; +const CALIBRE_WEB_PASS = 'changeme'; + +interface BookItem { id?: string; title?: string; url?: string; } + +/** Authenticate against Calibre-Web and return a usable cookie + fresh CSRF token. */ +async function getCalibreSession(): Promise<{ cookie: string; csrfToken: string }> { + const loginPageRes = await fetch(`${CALIBRE_WEB_LOCAL_URL}/login`); + const initialCookies = loginPageRes.headers.getSetCookie().map((c) => c.split(';')[0]).join('; '); + const loginHtml = await loginPageRes.text(); + + const csrfMatch = loginHtml.match(/name="csrf_token" value="(.*?)"/); + if (!csrfMatch) throw new Error('Could not find CSRF token on login page'); + const csrfToken = csrfMatch[1]; + + const loginData = new URLSearchParams(); + loginData.append('csrf_token', csrfToken); + loginData.append('username', CALIBRE_WEB_USER); + loginData.append('password', CALIBRE_WEB_PASS); + + const authRes = await fetch(`${CALIBRE_WEB_LOCAL_URL}/login`, { + method: 'POST', + headers: { + Cookie: initialCookies, + 'Content-Type': 'application/x-www-form-urlencoded', + Referer: `${CALIBRE_WEB_LOCAL_URL}/login`, + }, + body: loginData, + redirect: 'manual', + }); + + if (authRes.status !== 302 && authRes.status !== 303) { + throw new Error('Invalid Calibre-Web credentials'); + } + + const authCookieString = authRes.headers.getSetCookie().map((c) => c.split(';')[0]).join('; '); + + const homePageRes = await fetch(`${CALIBRE_WEB_LOCAL_URL}/`, { headers: { Cookie: authCookieString } }); + const homeHtml = await homePageRes.text(); + const finalCsrfMatch = + homeHtml.match(/name="csrf_token"\s+value="([^"]+)"/i) || + homeHtml.match(/value="([^"]+)"\s+name="csrf_token"/i); + const finalCsrfToken = finalCsrfMatch ? finalCsrfMatch[1] : csrfToken; + + return { cookie: authCookieString, csrfToken: finalCsrfToken }; +} + +const booksRunner: (ctx: RunnerContext) => Promise = async (ctx) => { + if (!fs.existsSync(TMP_DIR)) fs.mkdirSync(TMP_DIR, { recursive: true }); + + const books = ctx.items + .map((x) => x as BookItem) + .filter((b) => b && typeof b.url === 'string' && typeof b.id === 'string'); + if (books.length === 0) throw new Error('no books requested'); + + ctx.update({ phase: 'processing', percent: 0 }); + + let session: { cookie: string; csrfToken: string }; + try { + session = await getCalibreSession(); + } catch (e) { + throw new Error('Calibre-Web authentication failed'); + } + + let done = 0; + for (const book of books) { + ctx.throwIfCanceled(); + const id = String(book.id); + const title = String(book.title ?? id); + const url = String(book.url); + ctx.update({ phase: 'processing', detail: title, percent: Math.round((done / books.length) * 100) }); + + const tmp = path.join(TMP_DIR, `pg_${id}.epub`); + try { + const response = await fetch(url, { + headers: { 'User-Agent': SYSTEM_USER_AGENT, Accept: 'application/epub+zip' }, + }); + if (!response.ok) throw new Error(`HTTP ${response.status} from Gutenberg`); + + const fileBuffer = await response.arrayBuffer(); + fs.writeFileSync(tmp, Buffer.from(fileBuffer)); + + const form = new FormData(); + form.append('csrf_token', session.csrfToken); + form.append('btn-upload', new Blob([fileBuffer], { type: 'application/epub+zip' }), `${title}.epub`); + + const uploadRes = await fetch(`${CALIBRE_WEB_LOCAL_URL}/upload`, { + method: 'POST', + headers: { Cookie: session.cookie, Referer: `${CALIBRE_WEB_LOCAL_URL}/` }, + body: form, + }); + if (!uploadRes.ok) throw new Error(`Calibre-Web rejected upload: ${uploadRes.status}`); + } finally { + if (fs.existsSync(tmp)) fs.unlinkSync(tmp); + } + + done++; + ctx.update({ phase: 'processing', percent: Math.round((done / books.length) * 100) }); + } + + ctx.update({ phase: 'done', percent: 100 }); +}; + +jobs.registerRunner('books', booksRunner); + +export { booksRunner }; diff --git a/static/dashboard/sockets/jobs.ts b/static/dashboard/sockets/jobs.ts index 37ebf2041..31be3f42a 100644 --- a/static/dashboard/sockets/jobs.ts +++ b/static/dashboard/sockets/jobs.ts @@ -42,6 +42,9 @@ export interface JobUpdate { /** Passed to a runner: report progress, check cancellation, spawn tracked children. */ export interface RunnerContext { readonly job: Job; + /** Raw request items (strings for kiwix; objects for maps/books). */ + readonly items: unknown[]; + /** Convenience view: just the string items (kiwix ZIM filenames). */ readonly ids: string[]; update(patch: JobUpdate): void; isCanceled(): boolean; @@ -88,12 +91,13 @@ class JobManager { this.runners.set(type, runner); } - /** Create + start a job. Returns the persisted row immediately (phase 'queued'). */ - create(type: JobType, ids: string[]): Job { + /** Create + start a job. Returns the persisted row immediately (phase 'queued'). + * `items` are strings for kiwix (ZIM filenames) or objects for maps/books. */ + create(type: JobType, items: unknown[]): Job { const now = Date.now(); const id = `${type}-${now}-${Math.random().toString(36).slice(2, 8)}`; const job: Job = { - id, type, target: JSON.stringify(ids), phase: 'queued', + id, type, target: JSON.stringify(items), phase: 'queued', percent: -1, speed: 0, detail: null, error: null, created: now, updated: now, }; this.db.prepare( @@ -159,11 +163,12 @@ class JobManager { const rt = { canceled: false, procs: new Set() }; this.runtime.set(job.id, rt); - let ids: string[] = []; - try { ids = JSON.parse(job.target) as string[]; } catch { ids = []; } + let items: unknown[] = []; + try { const parsed = JSON.parse(job.target); items = Array.isArray(parsed) ? parsed : []; } catch { items = []; } + const ids = items.filter((x): x is string => typeof x === 'string'); const ctx: RunnerContext = { - job, ids, + job, items, ids, update: (p) => this.patch(job.id, p), isCanceled: () => rt.canceled, throwIfCanceled: () => { if (rt.canceled) throw new CanceledError(); }, diff --git a/static/dashboard/sockets/maps.exec.ts b/static/dashboard/sockets/maps.exec.ts new file mode 100644 index 000000000..cb68ae6db --- /dev/null +++ b/static/dashboard/sockets/maps.exec.ts @@ -0,0 +1,55 @@ +// sockets/maps.exec.ts — ADFA-4838 +// +// Maps runner for the durable job engine: extract a tile region via tile-extract.py. +// Ported from the Phase 1 maps.socket handler; reuses its box validation. A job item +// is { name, box, noninteractive? }; deletion stays a separate op (not a long job). +import { jobs, RunnerContext, CanceledError } from './jobs'; +import { parseBox } from './maps.socket'; +import path from 'path'; + +const SCRIPTS_DIR = '/opt/iiab/maps/tile-extract/'; +const EXTRACT_SCRIPT = path.join(SCRIPTS_DIR, 'tile-extract.py'); +// name: letters/digits/hyphen/underscore, 1..34 (same rule as the socket handler). +const NAME_RE = /^[A-Za-z0-9_-]{1,34}$/; + +interface MapItem { name?: string; box?: string; noninteractive?: boolean; } + +const mapsRunner: (ctx: RunnerContext) => Promise = async (ctx) => { + const item = (ctx.items[0] ?? {}) as MapItem; + const name = String(item.name ?? ''); + const rawBox = String(item.box ?? ''); + if (!NAME_RE.test(name)) throw new Error('invalid region name (A-Z a-z 0-9 _ -, length 1-34)'); + const parsed = parseBox(rawBox); + if (!parsed.ok) throw new Error(parsed.error); + + ctx.update({ phase: 'processing', percent: -1, detail: name }); + const args = [EXTRACT_SCRIPT, 'extract', name, parsed.box, 'noninteractive']; + + await new Promise((resolve, reject) => { + const p = ctx.spawn('sudo', args, { env: { ...process.env, PYTHONUNBUFFERED: '1' } }); + const onData = (buf: Buffer) => { + const text = buf.toString(); + ctx.log(text.trim()); + // tile-extract prints progress lines; surface a % if present, else stay indeterminate. + const re = /(\d+)\s*%/g; + let m: RegExpExecArray | null; + let last = -1; + while ((m = re.exec(text)) !== null) last = parseInt(m[1], 10); + if (last >= 0) ctx.update({ phase: 'processing', percent: last }); + }; + p.stdout?.on('data', onData); + p.stderr?.on('data', onData); + p.on('error', reject); + p.on('exit', (code, signal) => { + if (signal === 'SIGKILL' || ctx.isCanceled()) return reject(new CanceledError()); + if (code === 0) resolve(); + else reject(new Error(`tile-extract exited with code ${code}`)); + }); + }); + + ctx.update({ phase: 'done', percent: 100 }); +}; + +jobs.registerRunner('maps', mapsRunner); + +export { mapsRunner };