1/*
2 * Copyright (c) 2026 DataPond D-Library Pty Ltd. ("The Sanctuary")
3 * Non-Profit Public Library — The World Library
4 * 
5 * D-SAFE Certification issued by POND Enterprise.
6 * Published under the D-Open Code Sovereign Licence v1.0 framework
7 * DataPond D-Library All exclusive Right reserved on modifiying the code - CC Attribution to data pond, Non modifiable, Non Commercial.
8 * https://registry.world.bibliotech.com/licence
9 * Source code donated to DataPond D-Library by it's author: data pond.
10 * 
11 * Technical Guardian ("The Shield"): Pond Enterprise Pty Ltd. (ACN 694 747 987)
12 * All liability claims about D-SAFE certification issues coming from the published D-Safe direction must be addressed with Pond Enterprise Pty Ltd.
13 * Code Modification automatically voids the certified liability protection provided by Pond Enterprise.
14 */
15// EnsurePdf / EnsurePageImage — single-flight pipeline (§5.5).
16import { matchPage, matchPdf, deletePdf } from './cache-store';
17import { pageKey } from './keys';
18import { ConcurrentProcessor } from './queue';
19import {
20    _setFileState,
21    _setRenderPage,
22} from './state';
23import type { Thread } from './client/worker-client';
24import type { WorkerMessage } from './worker-protocol';
25
26const MAX_RENDER_RETRIES = 2;
27const DEFAULT_PDF_CONCURRENCY = 3;
28
29interface OrchestratorOptions {
30    muWorkerTimeout?:  number; // ms; default 60_000
31    pdfConcurrency?:   number; // max PDFs downloading at once across different books; default 3
32}
33
34// Singletons set at init.
35let _fileThread: Thread | null = null;
36let _muThread:   Thread | null = null;
37let _muTimeout   = 60_000;
38// Bounds how many *different* books can be downloading their PDF at once —
39// EnsurePdf's single-flight map only dedups repeat calls for the SAME txId,
40// it never capped fan-out across many different txIds (e.g. a shelf preloading
41// covers, or a page grid that triggers EnsurePdf per page).
42let _pdfProcessor: ConcurrentProcessor<void> | null = null;
43
44// Single-flight maps.
45const pendingPdf:  Map<string, Promise<void>>     = new Map();
46const pendingPage: Map<string, Promise<Response>> = new Map();
47
48export function initOrchestrator(
49    fileThread: Thread,
50    muThread: Thread,
51    opts: OrchestratorOptions = {},
52): void {
53    _fileThread   = fileThread;
54    _muThread     = muThread;
55    _muTimeout    = opts.muWorkerTimeout ?? 60_000;
56    _pdfProcessor = new ConcurrentProcessor<void>(opts.pdfConcurrency ?? DEFAULT_PDF_CONCURRENCY);
57
58    // Route render page-level messages into state. File-worker messages
59    // (progress/finished/error/prune) are owned by the single listener in
60    // client/files.ts — registering them here too would double-write state.
61    muThread.listen(handleMuWorkerMessage);
62}
63
64function handleMuWorkerMessage(msg: WorkerMessage): void {
65    if (msg.type === 'MuPageDone') {
66        _setRenderPage(msg.txId, msg.page, 'done');
67    } else if (msg.type === 'MuPageError') {
68        _setRenderPage(msg.txId, msg.page, 'error');
69    }
70}
71
72// Download-only (no render). Single-flight by txId.
73export function EnsurePdf(txId: string): Promise<void> {
74    const cached = pendingPdf.get(txId);
75    if (cached) return cached;
76
77    const p = _ensurePdfInner(txId).finally(() => pendingPdf.delete(txId));
78    pendingPdf.set(txId, p);
79    return p;
80}
81
82async function _ensurePdfInner(txId: string): Promise<void> {
83    const hit = await matchPdf(txId);
84    if (hit) return;
85
86    _setFileState(txId, { phase: 'queued' });
87
88    // Gated by _pdfProcessor so at most `pdfConcurrency` different books
89    // download simultaneously — unrelated txIds queue instead of all firing
90    // at once (this is on top of, not instead of, EnsurePdf's per-txId
91    // single-flight de-dup above).
92    await _pdfProcessor!.add(txId, async () => {
93        const result = await _fileThread!.rpc({ type: 'GetFile', txId, mime: 'pdf' });
94        if (!result.ok) {
95            throw new Error(result.error ?? '[orchestrator] download failed');
96        }
97    });
98}
99
100// Download + render for a single page. Single-flight by pageKey.
101export function EnsurePageImage(txId: string, page: number): Promise<Response> {
102    const key    = pageKey(txId, page);
103    const cached = pendingPage.get(key);
104    if (cached) return cached;
105
106    const p = _ensurePageInner(txId, page, key).finally(() => pendingPage.delete(key));
107    pendingPage.set(key, p);
108    return p;
109}
110
111async function _ensurePageInner(
112    txId:    string,
113    page:    number,
114    pKey:    string,
115): Promise<Response> {
116    // Fast path: already in cache.
117    const hit = await matchPage(txId, page);
118    if (hit) return hit;
119
120    _setRenderPage(txId, page, 'queued');
121
122    let retries = 0;
123    while (retries <= MAX_RENDER_RETRIES) {
124        // Ensure PDF is present (may already be cached or downloading).
125        await EnsurePdf(txId);
126
127        // Request render for this single page.
128        _setRenderPage(txId, page, 'downloading');
129        const renderResult = await requestRender(txId, page);
130
131        if (renderResult === 'done') {
132            const resp = await matchPage(txId, page);
133            if (resp) return resp;
134            throw new Error(`[orchestrator] page ${page} not in cache after render`);
135        }
136
137        // MuPageError: re-run EnsurePdf in case PDF was evicted, then retry.
138        retries++;
139        if (retries > MAX_RENDER_RETRIES) {
140            _setRenderPage(txId, page, 'error');
141            throw new Error(`[orchestrator] render failed for page ${page} of ${txId}`);
142        }
143        // The cached PDF may be corrupt (render failed against good bytes too,
144        // but a bad cache entry is the common case) — evict it so the next
145        // EnsurePdf actually re-downloads instead of matchPdf-ing the same bytes.
146        await deletePdf(txId);
147        pendingPdf.delete(txId);
148    }
149
150    throw new Error(`[orchestrator] unreachable`);
151}
152
153// Returns 'done' or 'error' after awaiting MuPageDone/MuPageError (with timeout).
154function requestRender(txId: string, page: number): Promise<'done' | 'error'> {
155    return new Promise((resolve) => {
156        const timer = setTimeout(() => {
157            cleanup();
158            resolve('error');
159        }, _muTimeout);
160
161        const cleanup = _addOneShotMuListener(txId, page, (result) => {
162            clearTimeout(timer);
163            resolve(result);
164        });
165
166        // Issue the render command.
167        _muThread!.send({ type: 'ProcessPages', txId, from: page, to: page });
168    });
169}
170
171// Register a one-shot listener on the mu-worker thread for a specific page event.
172// The returned cleanup detaches the handler so it cannot leak across renders.
173function _addOneShotMuListener(
174    txId:     string,
175    page:     number,
176    cb:       (result: 'done' | 'error') => void,
177): () => void {
178    const thread = _muThread!;
179    let done = false;
180    const cleanup = () => { done = true; thread.unlisten(handler); };
181    const handler = (msg: WorkerMessage) => {
182        if (done) return;
183        if (msg.type === 'MuPageDone'  && msg.txId === txId && msg.page === page) {
184            cleanup(); cb('done');
185        } else if (msg.type === 'MuPageError' && msg.txId === txId && msg.page === page) {
186            cleanup(); cb('error');
187        }
188    };
189    thread.listen(handler);
190    return cleanup;
191}
192