// Three bounded request pools for metadata, content, and thumbnails (issue
// #45).
//
// Ente meters differently by traffic class, so quak keeps three independent
// pools instead of one global limit: metadata is cheap and chatty, original
// content is heavy, thumbnails are small but numerous. Each pool is a
// `BoundedPool` — a plain concurrency limiter — and the three run at the
// design's caps (10 / 5 / 25) unless the caller overrides them.
//
// Two behaviours beyond a bare limiter, both per pool:
//
//   - Priority: work waiting for a slot is ordered on-demand before
//     background/precache, so a slot that frees up serves the request a user is
//     waiting on ahead of speculative prefetch. Within one priority the order
//     is first-come-first-served.
//
//   - In-flight dedup: a task submitted under a `key` that a still-pending task
//     already carries is not run a second time; both callers await the one
//     result. The key is released the moment that task settles — success or
//     failure — so a later request for the same key runs afresh. Callers key by
//     the id whose fetch must not be duplicated (a fileID, say).
//
// The pool holds a task's slot for that task's entire lifetime. A task that
// retries internally is doing so inside its slot: the slot is not freed between
// attempts, which is what keeps a retrying request counted against the cap. The
// pool knows nothing of the retry policy; it only holds the slot until the
// task's promise settles.
//
// This module is self-contained infrastructure. Routing a given `Client` call
// to the right pool belongs to the unit that wires the pools into the cache;
// here there is only the machinery.

export const DEFAULT_METADATA_CONCURRENCY = 10;
export const DEFAULT_CONTENT_CONCURRENCY = 5;
export const DEFAULT_THUMBNAIL_CONCURRENCY = 25;

// On-demand work is served before background/precache work waiting in the same
// pool.
export type Priority = "on-demand" | "background";

export interface RunOptions {
    // Defaults to "background": an unmarked request yields to on-demand work.
    priority?: Priority;
    // When set, a task already pending under this key is shared instead of run
    // again. Omit for work that must always execute.
    key?: string | number;
}

// One queued submission awaiting a slot. `start` runs the task and holds the
// slot until it settles.
interface Waiter {
    priority: Priority;
    // Submission order, used to break ties within a priority (FIFO).
    seq: number;
    start: () => void;
}

export class BoundedPool {
    readonly concurrency: number;

    private active = 0;
    private nextSeq = 0;
    private readonly waiting: Waiter[] = [];
    // Keyed by a caller-supplied dedup key; holds the shared promise for as
    // long as that task is pending, cleared when it settles.
    private readonly pending = new Map<string | number, Promise<unknown>>();

    constructor(concurrency: number) {
        if (!Number.isInteger(concurrency) || concurrency < 1) {
            throw new RangeError(
                `concurrency must be a positive integer, got ${concurrency}`,
            );
        }
        this.concurrency = concurrency;
    }

    // Submit `task` to the pool. It runs once a slot is free, subject to
    // priority; the returned promise settles with the task's result. With a
    // `key`, a still-pending submission under the same key is returned instead
    // of running `task` again.
    run<T>(task: () => Promise<T>, opts: RunOptions = {}): Promise<T> {
        const { key } = opts;
        if (key !== undefined) {
            const shared = this.pending.get(key);
            if (shared !== undefined) return shared as Promise<T>;
        }

        const promise = this.enqueue(task, opts.priority ?? "background");

        if (key !== undefined) {
            this.pending.set(key, promise);
            const release = (): void => {
                // Only clear our own entry: a fresh submission under the same
                // key after this one settled must not be evicted here.
                if (this.pending.get(key) === promise) this.pending.delete(key);
            };
            promise.then(release, release);
        }
        return promise;
    }

    private enqueue<T>(task: () => Promise<T>, priority: Priority): Promise<T> {
        return new Promise<T>((resolve, reject) => {
            const start = (): void => {
                this.active++;
                // Hold the slot until the task fully settles — every internal
                // retry included — then admit the next waiter.
                void (async () => {
                    try {
                        resolve(await task());
                    } catch (err) {
                        reject(err);
                    } finally {
                        this.active--;
                        this.pump();
                    }
                })();
            };
            this.waiting.push({ priority, seq: this.nextSeq++, start });
            this.pump();
        });
    }

    // Admit waiters until the pool is full or the queue is empty.
    private pump(): void {
        while (this.active < this.concurrency) {
            const next = this.takeNext();
            if (next === undefined) return;
            next.start();
        }
    }

    // Remove and return the highest-priority waiter: on-demand before
    // background, earliest submission first within a priority.
    private takeNext(): Waiter | undefined {
        let bestIndex = -1;
        let best: Waiter | undefined;
        for (let i = 0; i < this.waiting.length; i++) {
            const w = this.waiting[i];
            if (w === undefined) continue;
            if (best === undefined || this.precedes(w, best)) {
                best = w;
                bestIndex = i;
            }
        }
        if (best === undefined) return undefined;
        this.waiting.splice(bestIndex, 1);
        return best;
    }

    private precedes(a: Waiter, b: Waiter): boolean {
        if (a.priority !== b.priority) return a.priority === "on-demand";
        return a.seq < b.seq;
    }
}

export interface RequestPoolsOptions {
    metadataConcurrency?: number;
    contentConcurrency?: number;
    thumbnailConcurrency?: number;
}

// The three pools the design calls for, each independent: an idle pool never
// lends its slots to a busy one.
export class RequestPools {
    readonly metadata: BoundedPool;
    readonly content: BoundedPool;
    readonly thumbnails: BoundedPool;

    constructor(opts: RequestPoolsOptions = {}) {
        this.metadata = new BoundedPool(
            opts.metadataConcurrency ?? DEFAULT_METADATA_CONCURRENCY,
        );
        this.content = new BoundedPool(
            opts.contentConcurrency ?? DEFAULT_CONTENT_CONCURRENCY,
        );
        this.thumbnails = new BoundedPool(
            opts.thumbnailConcurrency ?? DEFAULT_THUMBNAIL_CONCURRENCY,
        );
    }
}