123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319
// Maximal valid value that can be passed to `setTimeout`const MAX_TIMEOUT_VALUE = ~(1 << 31); // equals (2 ** 31 - 1)/** * A drift is an element in a doubly linked list, and it stores a task. A task * is represented as a Promise. Drifts remove themselves from the queue (they * _decay_) after their task completes or exceeds the timeout. The timeout is * the maximum time that drifts are allowed to be contained in the queue. * * Drifts are appended to the tail of the queue and _drift_ towards the head as * older elements are removed, hence the name. * * The task of every drift is created by a worker function that operates based * on a source element. A drift keeps a reference to that source element. In * case of a timeout, the timeout handlers will be supplied with the source * element because it is interesting to know for which source elements the * worker function produced a promise that timed out. * * A drift also stores the date at which it was added to the queue. * * In the context of `grammy`, each middleware invocation corresponds to a task * in a drift. Drifts are used to manage concurrent middleware execution. */interface Drift<Y> {    /** Previous drift in the queue. `null` iff this drift is the head element. */    prev: Drift<Y> | null;    /** Next drift in the queue. `null` iff this drift is the tail element. */    next: Drift<Y> | null;
    /**     * Task of the drift. Contains logic that removes this drift from the queue as     * soon as the task completes by itself (either resolves or rejects).     */    task: Promise<void>;
    /**     * Timestamp (milliseconds since The Epoch) when this drift was added. This     * may be inspected when starting a new timer that might purge this drift upon     * timeout.     *     * The timestamp will be set to `-1` when the drift is removed from the queue,     * in other words, checking `date > 0` serves as a containment test.     */    date: number;    /** Reference to the source element that was used to start the task */    elem: Y;}
/** * A _decaying deque_ is a special kind of doubly linked list that serves as a * queue for a special kind of nodes, called _drifts_. * * A decaying deque has a worker function that spawns a task for each element * that is added to the queue. This task then gets wrapped into a drift. The * drifts are then the actual elements (aka. links) in the queue. * * In addition, the decaying deque runs a timer that purges old elements from * the queue. This period of time is determined by the `taskTimeout`. * * When a task completes or exceeds its timeout, the corresponding drift is * removed from the queue. As a result, only drifts with pending tasks are * contained in the queue at all times. * * When a tasks completes with failure (`reject`s or exceeds the timeout), the * respective handler (`catchError` or `catchTimeout`) is called. * * The decaying deque has its name from the observation that new elements are * appended to the tail, and the old elements are removed at arbitrary positions * in the queue whenever a task completes, hence, the queue seems to _decay_. */export class DecayingDeque<Y, R = unknown> {    /**     * Number of drifts in the queue. Equivalent to the number of currently     * pending tasks.     */    private len = 0;    /** Head element (oldest), `null` iff the queue is empty */    private head: Drift<Y> | null = null;    /** Tail element (newest), `null` iff the queue is empty */    private tail: Drift<Y> | null = null;
    /**     * Number of currently pending tasks that we strive for (`add` calls will     * resolve only after the number of pending tasks falls below this value.     *     * In the context of `grammy`, it is possible to `await` calls to `add` to     * determine when to fetch more updates.     */    public readonly concurrency: number;    /**     * Timer that waits for the head element to time out, will be rescheduled     * whenever the head element changes. It is `undefined` iff the queue is     * empty.     */    private timer: ReturnType<typeof setTimeout> | undefined;    /**     * List of subscribers that wait for the queue to have capacity again. All     * functions in this array will be called as soon as new capacity is     * available, i.e. the number of pending tasks falls below `concurrency`.     */    private subscribers: Array<(capacity: number) => void> = [];    private emptySubscribers: Array<() => void> = [];
    /**     * Creates a new decaying queue with the given parameters.     *     * @param taskTimeout Max period of time for a task     * @param worker Task generator     * @param concurrency `add` will return only after the number of pending tasks fell below `concurrency`. `false` means `1`, `true` means `Infinity`, numbers below `1` mean `1`     * @param catchError Error handler, receives the error and the source element     * @param catchTimeout Timeout handler, receives the source element and the promise of the task     */    constructor(        private readonly taskTimeout: number,        private readonly worker: (t: Y) => Promise<void>,        concurrency: boolean | number,        private readonly catchError: (err: R, elem: Y) => void | Promise<void>,        private readonly catchTimeout: (t: Y, task: Promise<void>) => void,    ) {        if (concurrency === false) this.concurrency = 1;        else if (concurrency === true) this.concurrency = Infinity;        else this.concurrency = concurrency < 1 ? 1 : concurrency;    }
    /**     * Adds the provided elements to the queue and starts tasks for all of them     * immediately. Returns a `Promise` that resolves with `concurrency - length`     * once this value becomes positive.     * @param elems Elements to be added     * @returns `this.capacity()`     */    add(elems: Y[]): Promise<number> {        const len = elems.length;        this.len += len;
        if (len > 0) {            let i = 0;            const now = Date.now();
            // emptyness check            if (this.head === null) {                this.head = this.tail = this.toDrift(elems[i++]!, now);                // start timer because head element changed                this.startTimer();            }
            let prev = this.tail!;            while (i < len) {                // create drift from source element                const node = this.toDrift(elems[i++]!, now);                // link it to previous element (append operation)                prev.next = node;                node.prev = prev;                prev = node;            }            this.tail = prev;        }
        return this.capacity();    }
    empty(): Promise<void> {        return new Promise((resolve) => {            if (this.len === 0) resolve();            else this.emptySubscribers.push(resolve);        });    }
    /**     * Returns a `Promise` that resolves with `concurrency - length` once this     * value becomes positive. Use `await queue.capacity()` to wait until the     * queue has free space again.     *     * @returns `concurrency - length` once positive     */    capacity(): Promise<number> {        return new Promise((resolve) => {            const capacity = this.concurrency - this.len;            if (capacity > 0) resolve(capacity);            else this.subscribers.push(resolve);        });    }
    /**     * Called when a node completed its lifecycle and should be removed from the     * queue. Effectively wraps the `remove` call and takes care of the timer.     *     * @param node Drift to decay     */    private decay(node: Drift<Y>): void {        // We only need to restart the timer if we decay the head element of the        // queue, however, if the next element has the same date as `node`, we can        // skip this step, too.        if (this.head === node && node.date !== node.next?.date) {            // Clear previous timeout            if (this.timer !== undefined) clearTimeout(this.timer);            // Emptyness check (do not start if queue is now empty)            if (node.next === null) this.timer = undefined;            // Reschedule timer for the next node's timeout            else {                this.startTimer(node.next.date + this.taskTimeout - Date.now());            }        }        this.remove(node);    }
    /**     * Removes an element from the queue. Calls subscribers if there is capacity     * after performing this operation.     *     * @param node Drift to remove     */    private remove(node: Drift<Y>): void {        // Connecting the links of `prev` and `next` removes `node`        if (this.head === node) this.head = node.next;        else node.prev!.next = node.next;        if (this.tail === node) this.tail = node.prev;        else node.next!.prev = node.prev;
        // Mark this drift as no longer contained        node.date = -1;
        // Notify subscribers if there is capacity by now        const capacity = this.concurrency - --this.len;        if (capacity > 0) {            this.subscribers.forEach((resolve) => resolve(capacity));            this.subscribers = [];        }        // Notify subscribers if the queue is empty now        if (this.len === 0) {            this.emptySubscribers.forEach((resolve) => resolve());            this.emptySubscribers = [];        }    }
    /**     * Takes a source element and starts the task for it by calling the worker     * function. Then wraps this task into a drift. Also makes sure that the drift     * removes itself from the queue once it completes, and that the error handler     * is invoked if it fails (rejects).     *     * @param elem Source element     * @param date Date when this drift is created     * @returns The created drift     */    private toDrift(elem: Y, date: number): Drift<Y> {        const node: Drift<Y> = {            prev: null,            task: this.worker(elem)                .catch(async (err) => {                    // Rethrow iff the drift is no longer contained (timed out)                    if (node.date > 0) await this.catchError(err, elem);                    else throw err;                })                .finally(() => {                    // Decay the node once the task completes (unless the drift was                    // removed due to a timeout before)                    if (node.date > 0) this.decay(node);                }),            next: null,            date,            elem,        };        return node;    }
    /**     * Starts a timer that fires off a timeout after the given period of time.     *     * @param ms Number of milliseconds to wait before the timeout kicks in     */    private startTimer(ms = this.taskTimeout): void {        this.timer = ms > MAX_TIMEOUT_VALUE            ? undefined            : setTimeout(() => this.timeout(), ms);    }
    /**     * Performs a timeout event. This removes the head element as well as all     * subsequent drifts with the same date (added in the same millisecond).     *     * The timeout handler is called in sequence for every removed drift.     */    private timeout(): void {        // Rare cases of the event ordering might fire a timeout even though the        // head element has just decayed.        if (this.head === null) return;        while (this.head.date === this.head.next?.date) {            this.catchTimeout(this.head.elem, this.head.task);            // No need to restart timer here, we'll modify head again anyway            this.remove(this.head);        }        this.catchTimeout(this.head.elem, this.head.task);        this.decay(this.head);    }
    /**     * Number of pending tasks in the queue. Equivalent to     * `this.pendingTasks().length` (but much more efficient).     */    get length() {        return this.len;    }
    /**     * Creates a snapshot of the queue by computing a list of those elements that     * are currently being processed.     */    pendingTasks(): Y[] {        const len = this.len;        const snapshot: Y[] = Array(len);        let drift = this.head!;        for (let i = 0; i < len; i++) {            snapshot[i] = drift.elem;            drift = drift.next!;        }        return snapshot;    }}