/**
 * @module async/queue.js
 * @author Krzyszot Putyra
 * @date Apr 03, 2025
 * @description An async processing queue
 */

import { PromiseManager } from "./promise.js";

/**
 * @callback QueueItemProcessor
 * @param {*} item
 * @returns {boolean}    true if the item can be removed from the queue
 */

/**
 * A class representing a processing queue
 * 
 * Items are processed one-by-one asynchronously, so that processing do not block
 * the current thread. This is achieved by using `setTimeout` to recursively call
 * the processing method.
 * 
 * Processing can be paused manually or by the processing method if the method
 * throws an error.
 */
export default class ProcessingQueue {

    /**
     * Creates a processing queue
     * @param {QueueItemProcessor} action   A callback function that processes data from the queue.
     *                                      The callback must return true if the item has to be
     *                                      processed again.
     */
    constructor(action) {
        /** @type {boolean} - items are not processed when this value is true */
        this._paused = false;

        /** @type {boolean} - a flag set when processing is going on */
        this._processing = false;

        /** @type {PromiseManager} */
        this._promises = new PromiseManager();

        /** @type {any[]} - an array of items to be processed */
        this._items = [];

        /** @type {QueueItemProcessor} - the callback that processes items */
        this.action = action;
    }

    /**
     * Returns the pause status of the queue
     * @returns {boolean}
     */
    get paused() { return this._paused; }

    /**
     * Returns true if the queue is processing messages
     * @returns {boolean}
     */
    get active() { return this._processing; }

    /**
     * Pauses processing the queue
     */
    pause() {
        this._paused = true;
    }

    /**
     * Resumes processing the queue
     */
    resume() {
        this._paused = false;
        this._process();
    }

    /**
     * Returns the number of unprocessed items
     * @returns {number}
     */
    get size() { return this._items.length; }

    /**
     * Returns true if the queue has no unprocessed items.
     * @returns {boolean}
     */
    get empty() { return !this._items.length; }

    /**
     * Returns a promise that resolves when the queue has no unprocessed items.
     * @returns {Promise<void>}}
     */
    waitForEmpty() {
        return this._items.length ? this._promises.create('empty') : Promise.resolve();
    }

    /**
     * Returns a promise that resolves when the queue stops processing items.
     * @returns {Promise<void>}}
     */
    async processed() {
        this._processing && await this._promises.create('done');
    }

    /**
     * Puts data on the queue and starts processing unless already processing or paused.
     * @param {*} data 
     */
    put(data) {
        this._items.push(data);
        this._paused || this._process();
    }

    /**
     * Ensures processing messages from the internal queue.
     * 
     * This method creates a task that processes messages asynchronously,
     * allowing other tasks to be process in parallel. Nothing happens
     * if processing is already active.
     * @internal
     */
    async _process() {

        if (this._processing) return;

        this._processing = true;
        let processed = 0,
            unprocessed = 0;
        while (true) {
            if (processed >= this._items.length) {
                this._items.length = unprocessed;
                break;
            } else if (this._paused || !this._processing) {
                this._items.copyWithin(unprocessed, processed);
                this._items.length -= (processed - unprocessed);
                break;
            } else {
                const item = this._items[processed];
                const success = await new Promise(
                    res => res(this.action(item, this))
                ).catch(() => false);
                if (!success) this._items[unprocessed++] = item;
                // Proceed to another item asynchronously
                processed++;
            }
        }
        this._processing = false;
        this._promises.resolve('done', unprocessed);
        if (this._items.length === 0) this._promises.resolve('empty');
    }

    /**
     * Removes all messages from the internal queue and resets the counter
     * of sent messages.
     */
    clear() {
        this._items.length = 0;
        if (!this._processing) {
            this._promises.resolve('done', 0);
            this._promises.resolve('empty');
        }
    }
}
