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// Bounded-concurrency queue — attaches waiters to in-flight tasks rather than duplicating them.
16
17export class ConcurrentProcessor<T = void> {
18    private readonly max: number;
19    private active   = 0;
20    private queue:   Array<{ key: string; run: () => void; reject: (err: unknown) => void }> = [];
21    // key → promise for every task currently in-flight or waiting to run.
22    private waiters: Map<string, Promise<T>> = new Map();
23
24    constructor(maxConcurrent: number) {
25        this.max = maxConcurrent;
26    }
27
28    add(key: string, task: () => Promise<T>): Promise<T> {
29        // Return the existing promise so duplicate callers share the result.
30        const existing = this.waiters.get(key);
31        if (existing) return existing;
32
33        const p = new Promise<T>((resolve, reject) => {
34            const run = () => {
35                this.active++;
36                // Promise.resolve().then(task) guarantees a synchronous throw inside
37                // task() becomes a rejection instead of leaking the concurrency slot
38                // and the waiters entry forever (task() would otherwise throw before
39                // .then(resolve, reject) ever got attached).
40                Promise.resolve()
41                    .then(task)
42                    .then(resolve, reject)
43                    .finally(() => {
44                        this.waiters.delete(key);
45                        this.active--;
46                        this._drain();
47                    });
48            };
49
50            if (this.active < this.max) {
51                run();
52            } else {
53                this.queue.push({ key, run, reject });
54            }
55        });
56
57        this.waiters.set(key, p);
58        return p;
59    }
60
61    start(): void {
62        // No-op for now — tasks run immediately on add when concurrency allows.
63    }
64
65    // Drops queued-but-not-started tasks and rejects their callers (so they never
66    // hang waiting on a task that will now never run), then resolves once every
67    // already-active task has settled.
68    stop(): Promise<void> {
69        const pending = this.queue;
70        this.queue = [];
71        for (const { key, reject } of pending) {
72            this.waiters.delete(key);
73            reject(new Error('[queue] stopped before task started'));
74        }
75        const remaining = [...this.waiters.values()];
76        return Promise.allSettled(remaining).then(() => undefined);
77    }
78
79    private _drain(): void {
80        if (this.queue.length > 0 && this.active < this.max) {
81            const next = this.queue.shift()!;
82            next.run();
83        }
84    }
85}
86