import { AbortError } from '@libp2p/interface';
import { pushable } from 'it-pushable';
import { TypedEventEmitter } from 'main-event';
import { pEvent } from 'p-event';
import { debounce } from "../debounce.js";
import { QueueFullError } from "../errors.js";
import { Job } from "./job.js";
/**
 * Heavily influence by `p-queue` with the following differences:
 *
 * 1. Items remain at the head of the queue while they are running so `queue.size` includes `queue.pending` items - this is so interested parties can join the results of a queue item while it is running
 * 2. The options for a job are stored separately to the job in order for them to be modified while they are still in the queue
 */
export class Queue extends TypedEventEmitter {
    concurrency;
    maxSize;
    queue;
    pending;
    sort;
    paused;
    constructor(init = {}) {
        super();
        this.concurrency = init.concurrency ?? Number.POSITIVE_INFINITY;
        this.maxSize = init.maxSize ?? Number.POSITIVE_INFINITY;
        this.pending = 0;
        this.paused = false;
        if (init.metricName != null) {
            init.metrics?.registerMetricGroup(init.metricName, {
                calculate: () => {
                    return {
                        size: this.queue.length,
                        running: this.pending,
                        queued: this.queue.length - this.pending
                    };
                }
            });
        }
        this.sort = init.sort;
        this.queue = [];
        this.emitEmpty = debounce(this.emitEmpty.bind(this), 1);
        this.emitIdle = debounce(this.emitIdle.bind(this), 1);
    }
    emitEmpty() {
        if (this.size !== 0) {
            return;
        }
        this.safeDispatchEvent('empty');
    }
    emitIdle() {
        if (this.running !== 0) {
            return;
        }
        this.safeDispatchEvent('idle');
    }
    pause() {
        this.paused = true;
    }
    resume() {
        if (!this.paused) {
            return;
        }
        this.paused = false;
        this.tryToStartAnother();
    }
    tryToStartAnother() {
        if (this.paused) {
            return false;
        }
        if (this.size === 0) {
            this.emitEmpty();
            if (this.running === 0) {
                this.emitIdle();
            }
            return false;
        }
        if (this.pending < this.concurrency) {
            let job;
            for (const j of this.queue) {
                if (j.status === 'queued') {
                    job = j;
                    break;
                }
            }
            if (job == null) {
                return false;
            }
            this.safeDispatchEvent('active');
            this.pending++;
            void job.run()
                .finally(() => {
                // remove the job from the queue
                for (let i = 0; i < this.queue.length; i++) {
                    if (this.queue[i] === job) {
                        this.queue.splice(i, 1);
                        break;
                    }
                }
                this.pending--;
                this.tryToStartAnother();
                this.safeDispatchEvent('next');
            });
            return true;
        }
        return false;
    }
    enqueue(job) {
        this.queue.push(job);
        if (this.sort != null) {
            this.queue.sort(this.sort);
        }
    }
    /**
     * Adds a sync or async task to the queue. Always returns a promise.
     */
    async add(fn, options) {
        options?.signal?.throwIfAborted();
        if (this.size === this.maxSize) {
            throw new QueueFullError();
        }
        const job = new Job(fn, options);
        this.enqueue(job);
        this.safeDispatchEvent('add');
        const result = job.join(options)
            .then(result => {
            this.safeDispatchEvent('completed', { detail: result });
            this.safeDispatchEvent('success', { detail: { job, result } });
            return result;
        })
            .catch(err => {
            if (job.status === 'queued') {
                // job was aborted before it started - remove the job from the queue
                for (let i = 0; i < this.queue.length; i++) {
                    if (this.queue[i] === job) {
                        this.queue.splice(i, 1);
                        break;
                    }
                }
            }
            this.safeDispatchEvent('failure', { detail: { job, error: err } });
            throw err;
        });
        this.tryToStartAnother();
        return result;
    }
    /**
     * Clear the queue
     */
    clear() {
        this.queue.splice(0, this.queue.length);
    }
    /**
     * Abort all jobs in the queue and clear it
     */
    abort() {
        this.queue.forEach(job => {
            job.abort(new AbortError());
        });
        this.clear();
    }
    /**
     * Can be called multiple times. Useful if you for example add additional items at a later time.
     *
     * @returns A promise that settles when the queue becomes empty.
     */
    async onEmpty(options) {
        // Instantly resolve if the queue is empty
        if (this.size === 0) {
            return;
        }
        await pEvent(this, 'empty', options);
    }
    /**
     * @returns A promise that settles when the queue size is less than the given
     * limit: `queue.size < limit`.
     *
     * If you want to avoid having the queue grow beyond a certain size you can
     * `await queue.onSizeLessThan()` before adding a new item.
     *
     * Note that this only limits the number of items waiting to start. There
     * could still be up to `concurrency` jobs already running that this call does
     * not include in its calculation.
     */
    async onSizeLessThan(limit, options) {
        // Instantly resolve if the queue is empty.
        if (this.size < limit) {
            return;
        }
        await pEvent(this, 'next', {
            ...options,
            filter: () => this.size < limit
        });
    }
    /**
     * The difference with `.onEmpty` is that `.onIdle` guarantees that all work
     * from the queue has finished. `.onEmpty` merely signals that the queue is
     * empty, but it could mean that some promises haven't completed yet.
     *
     * @returns A promise that settles when the queue becomes empty, and all
     * promises have completed; `queue.size === 0 && queue.pending === 0`.
     */
    async onIdle(options) {
        // Instantly resolve if none pending and if nothing else is queued
        if (this.pending === 0 && this.size === 0) {
            return;
        }
        await pEvent(this, 'idle', options);
    }
    /**
     * Size of the queue including running items
     */
    get size() {
        return this.queue.length;
    }
    /**
     * The number of queued items waiting to run.
     */
    get queued() {
        return this.queue.length - this.pending;
    }
    /**
     * The number of items currently running.
     */
    get running() {
        return this.pending;
    }
    /**
     * Returns an async generator that makes it easy to iterate over the results
     * of jobs added to the queue.
     *
     * The generator will end when the queue becomes idle, that is there are no
     * jobs running and no jobs that have yet to run.
     *
     * If you need to keep the queue open indefinitely, consider using it-pushable
     * instead.
     */
    async *toGenerator(options) {
        options?.signal?.throwIfAborted();
        const stream = pushable({
            objectMode: true
        });
        const cleanup = (err) => {
            if (err != null) {
                this.abort();
            }
            else {
                this.clear();
            }
            stream.end(err);
        };
        const onQueueJobComplete = (evt) => {
            if (evt.detail != null) {
                stream.push(evt.detail);
            }
        };
        const onQueueFailure = (evt) => {
            cleanup(evt.detail.error);
        };
        const onQueueIdle = () => {
            cleanup();
        };
        // clear the queue and throw if the query is aborted
        const onSignalAbort = () => {
            cleanup(new AbortError('Queue aborted'));
        };
        // add listeners
        this.addEventListener('completed', onQueueJobComplete);
        this.addEventListener('failure', onQueueFailure);
        this.addEventListener('idle', onQueueIdle);
        options?.signal?.addEventListener('abort', onSignalAbort);
        try {
            yield* stream;
        }
        finally {
            // remove listeners
            this.removeEventListener('completed', onQueueJobComplete);
            this.removeEventListener('failure', onQueueFailure);
            this.removeEventListener('idle', onQueueIdle);
            options?.signal?.removeEventListener('abort', onSignalAbort);
            // empty the queue for when the user has broken out of a loop early
            cleanup();
        }
    }
}
//# sourceMappingURL=index.js.map