1import { describe, it, beforeEach, expect, vi } from 'vitest';
2import { installCacheMock } from './cache-mock';
3import { putPage, putPdf, matchPdf } from '../cache-store';
4import { initOrchestrator, EnsurePdf, EnsurePageImage } from '../orchestrator';
5import type { Thread } from '../client/worker-client';
6import type { WorkerMessage } from '../worker-protocol';
7
8class FakeThread implements Thread {
9    handlers: Array<(msg: WorkerMessage) => void> = [];
10    sent: unknown[] = [];
11    rpcCalls: unknown[] = [];
12    rpcResult: { ok: boolean; key: string; error?: string } = { ok: true, key: '' };
13
14    send(cmd: unknown): void { this.sent.push(cmd); }
15    listen(cb: (msg: WorkerMessage) => void): void { this.handlers.push(cb); }
16    unlisten(cb: (msg: WorkerMessage) => void): void { this.handlers = this.handlers.filter(h => h !== cb); }
17    async rpc(cmd: any): Promise<{ ok: boolean; key: string; error?: string }> {
18        this.rpcCalls.push(cmd);
19        return this.rpcResult;
20    }
21    emit(msg: WorkerMessage): void { for (const h of this.handlers) h(msg); }
22}
23
24let fileThread: FakeThread;
25let muThread: FakeThread;
26
27beforeEach(() => {
28    installCacheMock();
29    fileThread = new FakeThread();
30    muThread = new FakeThread();
31    initOrchestrator(fileThread, muThread, { muWorkerTimeout: 200, pdfConcurrency: 2 });
32});
33
34describe('orchestrator — EnsurePdf', () => {
35    it('resolves immediately without an rpc call when the PDF is already cached', async () => {
36        await putPdf('cached-tx', new Response('x'));
37        await EnsurePdf('cached-tx');
38        expect(fileThread.rpcCalls.length).toBe(0);
39    });
40
41    it('issues a GetFile rpc when not cached and resolves on success', async () => {
42        fileThread.rpcResult = { ok: true, key: 'new-tx:pdf' };
43        await EnsurePdf('new-tx');
44        expect(fileThread.rpcCalls).toEqual([{ type: 'GetFile', txId: 'new-tx', mime: 'pdf' }]);
45    });
46
47    it('rejects when the rpc reports failure', async () => {
48        fileThread.rpcResult = { ok: false, key: 'bad-tx:pdf', error: 'gateway down' };
49        await expect(EnsurePdf('bad-tx')).rejects.toThrow('gateway down');
50    });
51
52    it('single-flights concurrent calls for the same txId — only one rpc issued', async () => {
53        let resolveRpc!: (r: any) => void;
54        fileThread.rpc = vi.fn(() => new Promise(res => { resolveRpc = res; }));
55
56        const a = EnsurePdf('dup-tx');
57        const b = EnsurePdf('dup-tx');
58        // matchPdf() is async, so the rpc isn't issued until the microtask
59        // queue drains — give it a tick before asserting single-flight.
60        await new Promise(r => setTimeout(r, 0));
61        expect(fileThread.rpc).toHaveBeenCalledTimes(1);
62
63        resolveRpc({ ok: true, key: 'dup-tx:pdf' });
64        await Promise.all([a, b]);
65    });
66});
67
68describe('orchestrator — EnsurePageImage', () => {
69    it('resolves immediately from cache without touching either thread', async () => {
70        await putPage('cached-page', 0, new Response('page-bytes'));
71        const resp = await EnsurePageImage('cached-page', 0);
72        expect(await resp.text()).toBe('page-bytes');
73        expect(fileThread.rpcCalls.length).toBe(0);
74        expect(muThread.sent.length).toBe(0);
75    });
76
77    it('ensures the PDF, requests a render, and resolves once the page lands in cache', async () => {
78        fileThread.rpcResult = { ok: true, key: 'render-tx:pdf' };
79
80        const pending = EnsurePageImage('render-tx', 3);
81        await new Promise(r => setTimeout(r, 0));
82
83        expect(muThread.sent).toContainEqual({ type: 'ProcessPages', txId: 'render-tx', from: 3, to: 3 });
84
85        await putPage('render-tx', 3, new Response('rendered-page'));
86        muThread.emit({ type: 'MuPageDone', txId: 'render-tx', page: 3 });
87
88        const resp = await pending;
89        expect(await resp.text()).toBe('rendered-page');
90    });
91
92    it('retries after a MuPageError up to the retry limit, then throws', async () => {
93        fileThread.rpcResult = { ok: true, key: 'fail-tx:pdf' };
94
95        const pending = EnsurePageImage('fail-tx', 0);
96        await new Promise(r => setTimeout(r, 0));
97
98        // Fail repeatedly until retries are exhausted (MAX_RENDER_RETRIES = 2 → 3 attempts total).
99        for (let i = 0; i < 3; i++) {
100            await new Promise(r => setTimeout(r, 0));
101            muThread.emit({ type: 'MuPageError', txId: 'fail-tx', page: 0 });
102        }
103
104        await expect(pending).rejects.toThrow(/render failed/);
105    }, 10_000);
106
107    it('evicts a cached PDF on MuPageError so the retry re-downloads instead of reusing corrupt bytes', async () => {
108        // Simulate a corrupt cache entry already sitting there before the first attempt.
109        await putPdf('corrupt-tx', new Response('corrupt-bytes'));
110        fileThread.rpcResult = { ok: true, key: 'corrupt-tx:pdf' };
111
112        const pending = EnsurePageImage('corrupt-tx', 0);
113        await new Promise(r => setTimeout(r, 0));
114
115        // First attempt: PDF was cached, so no rpc issued yet.
116        expect(fileThread.rpcCalls.length).toBe(0);
117
118        muThread.emit({ type: 'MuPageError', txId: 'corrupt-tx', page: 0 });
119        await new Promise(r => setTimeout(r, 0));
120
121        // The corrupt entry must be gone, and the retry's EnsurePdf must have
122        // re-issued a GetFile rpc rather than silently reusing matchPdf's stale hit.
123        expect(await matchPdf('corrupt-tx')).toBeUndefined();
124        expect(fileThread.rpcCalls).toEqual([{ type: 'GetFile', txId: 'corrupt-tx', mime: 'pdf' }]);
125
126        await putPage('corrupt-tx', 0, new Response('rendered-page'));
127        muThread.emit({ type: 'MuPageDone', txId: 'corrupt-tx', page: 0 });
128
129        const resp = await pending;
130        expect(await resp.text()).toBe('rendered-page');
131    });
132
133    it('a render timeout resolves as an error and is retried like MuPageError', async () => {
134        vi.useFakeTimers();
135        fileThread.rpcResult = { ok: true, key: 'timeout-tx:pdf' };
136
137        const pending = EnsurePageImage('timeout-tx', 0).catch(e => e);
138        for (let i = 0; i < 3; i++) {
139            await vi.advanceTimersByTimeAsync(200);
140        }
141        const result = await pending;
142        expect(result).toBeInstanceOf(Error);
143        vi.useRealTimers();
144    });
145
146    it('single-flights concurrent calls for the same txId+page', async () => {
147        fileThread.rpcResult = { ok: true, key: 'concurrent-tx:pdf' };
148        const p1 = EnsurePageImage('concurrent-tx', 1);
149        const p2 = EnsurePageImage('concurrent-tx', 1);
150
151        await new Promise(r => setTimeout(r, 0));
152        await putPage('concurrent-tx', 1, new Response('shared-page'));
153        muThread.emit({ type: 'MuPageDone', txId: 'concurrent-tx', page: 1 });
154
155        // Single-flighted: both callers are handed the exact same Response
156        // instance (its body can only be consumed once).
157        const [r1, r2] = await Promise.all([p1, p2]);
158        expect(r1).toBe(r2);
159        expect(await r1.text()).toBe('shared-page');
160    });
161});
162