import { Hono } from "hono"; import { stream } from "hono/streaming"; import { getAllConfig, getConfig, getDb, setConfig } from "../db/index"; import { log, error as logError, warn } from "../lib/log"; import { getAllItems, getDevItems } from "../services/jellyfin"; import { loadLibrary as loadRadarrLibrary, isUsable as radarrUsable } from "../services/radarr"; import { upsertJellyfinItem } from "../services/rescan"; import { isInScanWindow, msUntilScanWindow, nextScanWindowTime, waitForScanWindow } from "../services/scheduler"; import { loadLibrary as loadSonarrLibrary, isUsable as sonarrUsable } from "../services/sonarr"; const app = new Hono(); /** * Validate a scan `limit` input. Must be a positive integer or absent — * NaN/negatives/non-numerics would disable the progress cap * (`processed >= NaN` never trips) or produce bogus totals via * `Math.min(NaN, …)`. Exported for unit tests. */ export function parseScanLimit(raw: unknown): { ok: true; value: number | null } | { ok: false } { if (raw == null || raw === "") return { ok: true, value: null }; const n = typeof raw === "number" ? raw : Number(raw); if (!Number.isInteger(n) || n <= 0) return { ok: false }; return { ok: true, value: n }; } type ScanStatusFilter = "all" | "pending" | "scanned" | "error"; type ScanTypeFilter = "all" | "movie" | "episode"; type ScanSourceFilter = "all" | "scan" | "webhook"; export interface ScanItemsQuery { offset: number; limit: number; search: string; status: ScanStatusFilter; type: ScanTypeFilter; source: ScanSourceFilter; } function parsePositiveInt(raw: unknown, fallback: number): number { const n = typeof raw === "number" ? raw : Number(raw); if (!Number.isFinite(n)) return fallback; if (!Number.isInteger(n)) return fallback; return n; } function clamp(n: number, min: number, max: number): number { if (n < min) return min; if (n > max) return max; return n; } function parseOneOf(raw: unknown, allowed: T, fallback: T[number]): T[number] { if (typeof raw !== "string") return fallback; const lowered = raw.toLowerCase(); return (allowed as readonly string[]).includes(lowered) ? (lowered as T[number]) : fallback; } export function parseScanItemsQuery(raw: Record): ScanItemsQuery { const limit = clamp(parsePositiveInt(raw.limit, 50), 1, 200); const offset = Math.max(0, parsePositiveInt(raw.offset, 0)); const search = typeof raw.q === "string" ? raw.q.trim() : ""; return { offset, limit, search, status: parseOneOf(raw.status, ["all", "pending", "scanned", "error"] as const, "all"), type: parseOneOf(raw.type, ["all", "movie", "episode"] as const, "all"), source: parseOneOf(raw.source, ["all", "scan", "webhook"] as const, "all"), }; } export function buildScanItemsWhere(query: ScanItemsQuery): { sql: string; args: unknown[] } { const clauses: string[] = []; const args: unknown[] = []; if (query.status !== "all") { clauses.push("scan_status = ?"); args.push(query.status); } if (query.type !== "all") { clauses.push("lower(type) = ?"); args.push(query.type); } if (query.source !== "all") { clauses.push("ingest_source = ?"); args.push(query.source); } if (query.search.length > 0) { clauses.push("(lower(name) LIKE ? OR lower(file_path) LIKE ?)"); const needle = `%${query.search.toLowerCase()}%`; args.push(needle, needle); } return { sql: clauses.length > 0 ? `WHERE ${clauses.join(" AND ")}` : "", args, }; } // ─── State ──────────────────────────────────────────────────────────────────── let scanAbort: AbortController | null = null; const scanListeners = new Set<(data: string) => void>(); function emitSse(type: string, data: unknown): void { const line = `event: ${type}\ndata: ${JSON.stringify(data)}\n\n`; for (const listener of scanListeners) listener(line); } function currentScanLimit(): number | null { const v = getConfig("scan_limit"); return v ? Number(v) : null; } function parseLanguageList(raw: string | null, fallback: string[]): string[] { if (!raw) return fallback; try { const parsed = JSON.parse(raw); return Array.isArray(parsed) ? parsed.filter((v): v is string => typeof v === "string") : fallback; } catch { return fallback; } } // ─── Status ─────────────────────────────────────────────────────────────────── app.get("/", (c) => { const db = getDb(); const running = getConfig("scan_running") === "1"; const total = (db.prepare("SELECT COUNT(*) as n FROM media_items").get() as { n: number }).n; const scanned = ( db.prepare("SELECT COUNT(*) as n FROM media_items WHERE scan_status = 'scanned'").get() as { n: number } ).n; const errors = (db.prepare("SELECT COUNT(*) as n FROM media_items WHERE scan_status = 'error'").get() as { n: number }) .n; const recentItems = db .prepare( "SELECT name, type, scan_status, file_path, last_scanned_at, ingest_source FROM media_items ORDER BY COALESCE(last_scanned_at, created_at) DESC, id DESC LIMIT 5", ) .all() as { name: string; type: string; scan_status: string; file_path: string; last_scanned_at: string | null; ingest_source: string | null; }[]; return c.json({ running, progress: { scanned, total, errors }, recentItems, scanLimit: currentScanLimit() }); }); app.get("/items", (c) => { const db = getDb(); const query = parseScanItemsQuery({ offset: c.req.query("offset"), limit: c.req.query("limit"), q: c.req.query("q"), status: c.req.query("status"), type: c.req.query("type"), source: c.req.query("source"), }); const where = buildScanItemsWhere(query); const rows = db .prepare( ` SELECT id, jellyfin_id, name, type, series_name, season_number, episode_number, scan_status, original_language, orig_lang_source, container, file_size, file_path, last_scanned_at, ingest_source FROM media_items ${where.sql} ORDER BY COALESCE(last_scanned_at, created_at) DESC, id DESC LIMIT ? OFFSET ? `, ) .all(...where.args, query.limit, query.offset) as Array<{ id: number; jellyfin_id: string; name: string; type: string; series_name: string | null; season_number: number | null; episode_number: number | null; scan_status: string; original_language: string | null; orig_lang_source: string | null; container: string | null; file_size: number | null; file_path: string; last_scanned_at: string | null; ingest_source: string | null; }>; const total = (db.prepare(`SELECT COUNT(*) as n FROM media_items ${where.sql}`).get(...where.args) as { n: number }).n; return c.json({ rows, total, hasMore: query.offset + rows.length < total, query }); }); // ─── Start ──────────────────────────────────────────────────────────────────── app.post("/start", async (c) => { const db = getDb(); // Atomic claim: only succeed if scan_running is not already '1'. const claim = db.prepare("UPDATE config SET value = '1' WHERE key = 'scan_running' AND value != '1'").run(); if (claim.changes === 0) { return c.json({ ok: false, error: "Scan already running" }, 409); } const body = await c.req.json<{ limit?: unknown }>().catch(() => ({ limit: undefined })); const formLimit = parseScanLimit(body.limit); const envLimit = parseScanLimit(process.env.SCAN_LIMIT); if (!formLimit.ok || !envLimit.ok) { db.prepare("UPDATE config SET value = '0' WHERE key = 'scan_running'").run(); return c.json({ ok: false, error: "limit must be a positive integer" }, 400); } const limit: number | null = formLimit.value ?? envLimit.value ?? null; setConfig("scan_limit", limit != null ? String(limit) : ""); runScan(limit).catch((err) => { logError("Scan failed:", err); setConfig("scan_running", "0"); emitSse("error", { message: String(err) }); }); return c.json({ ok: true }); }); // ─── Stop ───────────────────────────────────────────────────────────────────── app.post("/stop", (c) => { scanAbort?.abort(); setConfig("scan_running", "0"); return c.json({ ok: true }); }); // ─── SSE ────────────────────────────────────────────────────────────────────── app.get("/events", (c) => { return stream(c, async (s) => { c.header("Content-Type", "text/event-stream"); c.header("Cache-Control", "no-cache"); c.header("Connection", "keep-alive"); const queue: string[] = []; let resolve: (() => void) | null = null; const listener = (data: string) => { queue.push(data); resolve?.(); }; scanListeners.add(listener); s.onAbort(() => { scanListeners.delete(listener); }); try { while (!s.closed) { if (queue.length > 0) { await s.write(queue.shift()!); } else { await new Promise((res) => { resolve = res; setTimeout(res, 25_000); }); resolve = null; if (queue.length === 0) await s.write(": keepalive\n\n"); } } } finally { scanListeners.delete(listener); } }); }); // ─── Core scan logic ────────────────────────────────────────────────────────── async function runScan(limit: number | null = null): Promise { log(`Scan started${limit ? ` (limit: ${limit})` : ""}`); scanAbort = new AbortController(); const { signal } = scanAbort; const isDev = process.env.NODE_ENV === "development"; const db = getDb(); if (isDev) { // Order matters only if foreign keys are enforced without CASCADE; we // have ON DELETE CASCADE on media_streams/review_plans/stream_decisions/ // subtitle_files/jobs, so deleting media_items would be enough. List // them explicitly for clarity and to survive future schema drift. db.prepare("DELETE FROM jobs").run(); db.prepare("DELETE FROM subtitle_files").run(); db.prepare("DELETE FROM stream_decisions").run(); db.prepare("DELETE FROM review_plans").run(); db.prepare("DELETE FROM media_streams").run(); db.prepare("DELETE FROM media_items").run(); } const cfg = getAllConfig(); const jellyfinCfg = { url: cfg.jellyfin_url, apiKey: cfg.jellyfin_api_key, userId: cfg.jellyfin_user_id }; const audioLanguages = parseLanguageList(cfg.audio_languages ?? null, []); const radarrCfg = { url: cfg.radarr_url, apiKey: cfg.radarr_api_key }; const sonarrCfg = { url: cfg.sonarr_url, apiKey: cfg.sonarr_api_key }; // 'enabled' in config means the user toggled it on. Only actually use it // if the URL+key pass URL parsing — otherwise we'd hit ERR_INVALID_URL on // every item. Refuse to call invalid endpoints rather than spamming logs. const radarrEnabled = cfg.radarr_enabled === "1" && radarrUsable(radarrCfg); const sonarrEnabled = cfg.sonarr_enabled === "1" && sonarrUsable(sonarrCfg); if (cfg.radarr_enabled === "1" && !radarrEnabled) { warn(`Radarr is enabled in config but URL/API key is invalid (url='${cfg.radarr_url}') — skipping Radarr lookups`); } if (cfg.sonarr_enabled === "1" && !sonarrEnabled) { warn(`Sonarr is enabled in config but URL/API key is invalid (url='${cfg.sonarr_url}') — skipping Sonarr lookups`); } // Pre-load both libraries once so per-item lookups are O(1) cache hits // instead of HTTP round-trips. The previous code called /api/v3/movie // (the entire library!) once per item that didn't match by tmdbId. const [radarrLibrary, sonarrLibrary] = await Promise.all([ radarrEnabled ? loadRadarrLibrary(radarrCfg) : Promise.resolve(null), sonarrEnabled ? loadSonarrLibrary(sonarrCfg) : Promise.resolve(null), ]); log( `External language sources: radarr=${radarrEnabled ? `enabled (${cfg.radarr_url}, ${radarrLibrary?.byTmdbId.size ?? 0} movies in library)` : "disabled"}, sonarr=${sonarrEnabled ? `enabled (${cfg.sonarr_url}, ${sonarrLibrary?.byTvdbId.size ?? 0} series in library)` : "disabled"}`, ); let processed = 0; let errors = 0; let total = 0; const rescanCfg = { audioLanguages, radarr: radarrEnabled ? radarrCfg : null, sonarr: sonarrEnabled ? sonarrCfg : null, radarrLibrary, sonarrLibrary, }; let radarrMisses = 0; let radarrHits = 0; let sonarrMisses = 0; let sonarrHits = 0; let missingProviderIds = 0; const itemSource = isDev ? getDevItems(jellyfinCfg) : getAllItems(jellyfinCfg, (_fetched, jellyfinTotal) => { total = limit != null ? Math.min(limit, jellyfinTotal) : jellyfinTotal; }); for await (const jellyfinItem of itemSource) { if (signal.aborted) break; if (!isDev && limit != null && processed >= limit) break; if (!jellyfinItem.Name || !jellyfinItem.Path) { warn(`Skipping item without name/path: id=${jellyfinItem.Id}`); continue; } // Honour the scan window between items so overnight-only setups don't hog // Jellyfin during the day. Checked between items rather than mid-item so // we don't leave a partial upsert sitting in flight. if (!isInScanWindow()) { emitSse("paused", { until: nextScanWindowTime(), seconds: Math.round(msUntilScanWindow() / 1000), }); await waitForScanWindow(); if (signal.aborted) break; emitSse("resumed", {}); } processed++; emitSse("progress", { scanned: processed, total, current_item: jellyfinItem.Name, errors, running: true }); try { const result = await upsertJellyfinItem(db, jellyfinItem, rescanCfg); if (result.radarrHit) radarrHits++; if (result.radarrMiss) radarrMisses++; if (result.sonarrHit) sonarrHits++; if (result.sonarrMiss) sonarrMisses++; if (result.missingProviderId) missingProviderIds++; emitSse("log", { name: jellyfinItem.Name, type: jellyfinItem.Type, status: "scanned", file: jellyfinItem.Path }); } catch (err) { errors++; logError(`Error scanning ${jellyfinItem.Name}:`, err); try { db .prepare("UPDATE media_items SET scan_status = 'error', scan_error = ? WHERE jellyfin_id = ?") .run(String(err), jellyfinItem.Id); } catch (dbErr) { // Failed to persist the error status — log it so the incident // doesn't disappear silently. We can't do much more; the outer // loop continues so the scan still finishes. logError(`Failed to record scan error for ${jellyfinItem.Id}:`, dbErr); } emitSse("log", { name: jellyfinItem.Name, type: jellyfinItem.Type, status: "error", file: jellyfinItem.Path }); } } setConfig("scan_running", "0"); log(`Scan complete: ${processed} scanned, ${errors} errors`); log( ` language sources: radarr hits=${radarrHits} misses=${radarrMisses}, sonarr hits=${sonarrHits} misses=${sonarrMisses}, no provider id=${missingProviderIds}`, ); emitSse("complete", { scanned: processed, total, errors }); } export default app;