'use strict';

import {Worker} from 'node:worker_threads';
import path from 'node:path';
import os from 'node:os';
import {fileURLToPath} from 'node:url';

const __dirname = path.dirname(fileURLToPath(import.meta.url));

/** Round-robin pool of worker threads for off-main-thread cluster decompression. */
class DecompressPool {
	#workers = [];
	#next = 0;
	#pending = new Map(); // id -> {resolve, reject}
	#nextId = 0;
	#active = 0;

	constructor(size = Math.max(1, os.cpus().length - 1)) {
		for (let i = 0; i < size; i++) this.#spawn();
	}

	#spawn() {
		const worker = new Worker(path.join(__dirname, 'decompress-worker.js'));
		worker.on('message', ({id, data, error}) => {
			const task = this.#pending.get(id);
			if (!task) return;
			this.#pending.delete(id);
			// No work left in flight -> safe to let the process exit again.
			if (--this.#active === 0) for (const w of this.#workers) w.unref();
			error ? task.reject(new Error(error)) : task.resolve(Buffer.from(data));
		});
		worker.unref(); // idle workers must never keep a short script/server alive
		this.#workers.push(worker);
	}

	/** Decompresses `{compType, body}` on the next worker in rotation. `body` must be a Uint8Array/Buffer view. */
	run({compType, body}) {
		const worker = this.#workers[this.#next];
		this.#next = (this.#next + 1) % this.#workers.length;
		const id = this.#nextId++;
		const owned = new Uint8Array(body);
		if (this.#active++ === 0) for (const w of this.#workers) w.ref();
		return new Promise((resolve, reject) => {
			this.#pending.set(id, {resolve, reject});
			worker.postMessage({id, compType, body: owned}, [owned.buffer]);
		});
	}
}

export const decompressPool = new DecompressPool();