import { AbortError } from '@libp2p/interface';
import { setMaxListeners } from 'main-event';
import { raceSignal } from 'race-signal';
import { JobRecipient } from "./recipient.js";
/**
 * Returns a random string
 */
function randomId() {
    return `${(parseInt(String(Math.random() * 1e9), 10)).toString()}${Date.now()}`;
}
export class Job {
    id;
    fn;
    options;
    recipients;
    status;
    timeline;
    controller;
    dispatchingProgress;
    constructor(fn, options) {
        this.id = randomId();
        this.status = 'queued';
        this.fn = fn;
        this.options = options;
        this.recipients = [];
        this.timeline = {
            created: Date.now()
        };
        this.controller = new AbortController();
        setMaxListeners(Infinity, this.controller.signal);
        this.dispatchingProgress = false;
        this.onAbort = this.onAbort.bind(this);
    }
    abort(err) {
        this.controller.abort(err);
    }
    onAbort() {
        const allAborted = this.recipients.reduce((acc, curr) => {
            return acc && (curr.signal?.aborted === true);
        }, true);
        // if all recipients have aborted the job, actually abort the job
        if (allAborted) {
            this.controller.abort(new AbortError());
            this.cleanup();
        }
    }
    async join(options) {
        const recipient = new JobRecipient(options);
        this.recipients.push(recipient);
        options?.signal?.addEventListener('abort', this.onAbort);
        return recipient.deferred.promise;
    }
    async run() {
        this.status = 'running';
        this.timeline.started = Date.now();
        try {
            this.controller.signal.throwIfAborted();
            const result = await raceSignal(this.fn({
                ...(this.options ?? {}),
                signal: this.controller.signal,
                onProgress: (evt) => {
                    // Recipients can transitively re-enter this dispatcher; without
                    // this guard a single event recurses until the stack overflows.
                    if (this.dispatchingProgress) {
                        return;
                    }
                    this.dispatchingProgress = true;
                    try {
                        this.recipients.forEach(recipient => {
                            recipient.onProgress?.(evt);
                        });
                    }
                    finally {
                        this.dispatchingProgress = false;
                    }
                }
            }), this.controller.signal);
            this.recipients.forEach(recipient => {
                recipient.deferred.resolve(result);
            });
            this.status = 'complete';
        }
        catch (err) {
            this.recipients.forEach(recipient => {
                recipient.deferred.reject(err);
            });
            this.status = 'errored';
        }
        finally {
            this.timeline.finished = Date.now();
            this.cleanup();
        }
    }
    cleanup() {
        this.recipients.forEach(recipient => {
            recipient.cleanup();
            recipient.signal?.removeEventListener('abort', this.onAbort);
        });
    }
}
//# sourceMappingURL=job.js.map