Merge pull request #253 from appdevforall/feat/ADFA-4838-rest-job-engine

ADFA-4838: durable REST job engine for live content (kiwix)

Author: luisguzman-adfa

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.

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)

static/dashboard/routes.ts

@@ -0,0 +1,67 @@
+// 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; }
+    // 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.
+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 });
+});

static/dashboard/server.ts

@@ -10,6 +10,14 @@ 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 './sockets/maps.exec';
+import './sockets/books.exec';
+import { apiRouter } from './routes';
+
 const app = express();
 const server = http.createServer(app);
 const io = new Server(server);
@@ -28,9 +36,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 +65,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); }
 });
 
 // ==========================================

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<void> = 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 };