1import { describe, it, expect, vi } from 'vitest';
2import { ConcurrentProcessor } from '../queue';
3
4const deferred = <T>() => {
5    let resolve!: (v: T) => void;
6    let reject!: (e: unknown) => void;
7    const promise = new Promise<T>((res, rej) => { resolve = res; reject = rej; });
8    return { promise, resolve, reject };
9};
10
11describe('ConcurrentProcessor', () => {
12    it('runs tasks immediately while under the concurrency limit', async () => {
13        const p = new ConcurrentProcessor<number>(2);
14        const started: string[] = [];
15        const a = p.add('a', async () => { started.push('a'); return 1; });
16        const b = p.add('b', async () => { started.push('b'); return 2; });
17        expect(await a).toBe(1);
18        expect(await b).toBe(2);
19        expect(started).toEqual(['a', 'b']);
20    });
21
22    it('queues tasks beyond the concurrency limit and runs them once a slot frees up', async () => {
23        const p = new ConcurrentProcessor<string>(1);
24        const first = deferred<string>();
25        const order: string[] = [];
26
27        const a = p.add('a', async () => { order.push('a-start'); const r = await first.promise; order.push('a-end'); return r; });
28        const b = p.add('b', async () => { order.push('b-start'); return 'b-result'; });
29
30        await new Promise(r => setTimeout(r, 0));
31        // 'b' must not have started yet — only 1 concurrent slot, held by 'a'.
32        expect(order).toEqual(['a-start']);
33
34        first.resolve('a-result');
35        expect(await a).toBe('a-result');
36        expect(await b).toBe('b-result');
37        expect(order).toEqual(['a-start', 'a-end', 'b-start']);
38    });
39
40    it('returns the same promise for duplicate keys instead of re-running the task', async () => {
41        const p = new ConcurrentProcessor<number>(2);
42        let calls = 0;
43        const task = async () => { calls++; return 42; };
44
45        const first = p.add('same-key', task);
46        const second = p.add('same-key', task);
47        expect(first).toBe(second);
48        expect(await first).toBe(42);
49        expect(calls).toBe(1);
50    });
51
52    it('allows the same key to run again once the prior task has settled', async () => {
53        const p = new ConcurrentProcessor<number>(2);
54        let calls = 0;
55        const task = async () => { calls++; return calls; };
56
57        expect(await p.add('k', task)).toBe(1);
58        // The waiters-map cleanup runs in a .finally() one microtask after the
59        // caller's await unblocks — flush the queue so 'k' is free again.
60        await new Promise(r => setTimeout(r, 0));
61        expect(await p.add('k', task)).toBe(2);
62    });
63
64    it('converts a synchronous throw inside a task into a rejection instead of hanging', async () => {
65        const p = new ConcurrentProcessor<void>(1);
66        await expect(p.add('boom', () => { throw new Error('sync fail'); })).rejects.toThrow('sync fail');
67    });
68
69    it('propagates an async rejection to the caller', async () => {
70        const p = new ConcurrentProcessor<void>(1);
71        await expect(p.add('boom', async () => { throw new Error('async fail'); })).rejects.toThrow('async fail');
72    });
73
74    it('frees the concurrency slot after a task rejects, so queued tasks still run', async () => {
75        const p = new ConcurrentProcessor<string>(1);
76        const failing = p.add('fail', async () => { throw new Error('nope'); });
77        const succeeding = p.add('ok', async () => 'done');
78
79        await expect(failing).rejects.toThrow('nope');
80        expect(await succeeding).toBe('done');
81    });
82
83    it('stop() rejects queued-but-not-started tasks and lets active ones settle', async () => {
84        const p = new ConcurrentProcessor<string>(1);
85        const first = deferred<string>();
86        const active = p.add('active', async () => first.promise);
87        const queued = p.add('queued', async () => 'never runs');
88
89        const stopping = p.stop();
90        await expect(queued).rejects.toThrow(/stopped before task started/);
91
92        first.resolve('active-result');
93        expect(await active).toBe('active-result');
94        await stopping;
95    });
96
97    it('start() is a no-op that does not throw', () => {
98        const p = new ConcurrentProcessor<void>(3);
99        expect(() => p.start()).not.toThrow();
100    });
101});
102