/* eslint-disable no-console */
// Worker pool with: init barrier, progress reporting, and persistent message handler.

import { Worker } from "node:worker_threads";

type ProgressMsg = { type: "progress"; loaded: number; total: number };
type ReadyMsg = { type: "ready"; ok: boolean; detail?: string };
type OkMsg = { id: number; ok: true; decoded: any };
type ErrMsg = { id: number; ok: false; error: string };

type AnyOut = ProgressMsg | ReadyMsg | OkMsg | ErrMsg;

export type TxDecodePool = {
    submit: (txBase64: string) => Promise<any>;
    close: () => Promise<void>;
};

const INIT_TIMEOUT_MS = 30000;

export function createTxDecodePool(size: number, opts?: { protoDir?: string }): TxDecodePool {
    const workers: Worker[] = [];
    const idle: number[] = [];
    const pending = new Map<number, { resolve: (v: any) => void; reject: (e: any) => void }>();
    const readyFlags: boolean[] = Array(size).fill(false);
    const readyResolvers: Array<() => void> = [];
    const readyPromises: Array<Promise<void>> = [];
    const perWorkerProgress: Record<number, { loaded: number; total: number }> = {};

    console.log(`[txPool] creating ${size} worker(s)`);

    for (let i = 0; i < size; i++) {
        const w = new Worker(new URL("./txWorker.ts", import.meta.url), {
            execArgv: ["--import", "tsx/esm"],
            stdout: false,
            stderr: false,
        });
        // @ts-ignore
        w.stdout?.pipe(process.stdout);
        // @ts-ignore
        w.stderr?.pipe(process.stderr);

        workers.push(w);

        let resolveReady!: () => void;
        let rejectReady!: (e?: any) => void;
        const p = new Promise<void>((resolve, reject) => ((resolveReady = resolve), (rejectReady = reject)));
        readyPromises.push(p);
        readyResolvers.push(resolveReady);

        const timer = setTimeout(() => {
            if (!readyFlags[i]) {
                console.error(`[txPool] worker #${i} init timeout after ${INIT_TIMEOUT_MS}ms`);
                // не даём процессу висеть: резолвим ready и возвращаем воркера в idle (пусть задачи упадут на decode)
                readyFlags[i] = true;
                idle.push(i);
                resolveReady();
            }
        }, INIT_TIMEOUT_MS);

        w.on("online", () => console.log(`[txPool] worker #${i} online`));

        w.on("message", (m: any) => {
            if (m?.type === "progress") {
                const { loaded, total } = m;
                perWorkerProgress[i] = { loaded, total };
                const totals = Object.values(perWorkerProgress);
                if (totals.length > 0) {
                    const sumLoaded = totals.reduce((a, b) => a + b.loaded, 0);
                    const sumTotal = totals.reduce((a, b) => a + b.total, 0);
                    const pct = sumTotal > 0 ? Math.floor((sumLoaded / sumTotal) * 100) : 0;
                    process.stdout.write(`\r[proto] loading: ${sumLoaded}/${sumTotal} (${pct}%)`);
                    if (sumLoaded === sumTotal) process.stdout.write("\n");
                }
                return;
            }

            if (m?.type === "ready") {
                if (!readyFlags[i]) {
                    readyFlags[i] = true;
                    clearTimeout(timer);
                    if (m.ok !== false) {
                        console.log(`[txPool] worker #${i} ready`);
                    } else {
                        console.warn(`[txPool] worker #${i} init not-ok: ${m.detail ?? ""}`);
                    }
                    idle.push(i);
                    resolveReady();
                }
                return;
            }

            if (typeof m?.id === "number") {
                const entry = pending.get(m.id);
                if (!entry) return;
                pending.delete(m.id);
                idle.push(i);
                if (m.ok) entry.resolve(m.decoded);
                else entry.reject(new Error(m.error));
                return;
            }
        });

        w.on("error", (e) => {
            console.error(`[txPool] worker #${i} error: ${e?.message ?? e}`);
            if (!readyFlags[i]) {
                clearTimeout(timer);
                readyFlags[i] = true;
                idle.push(i);
                resolveReady();
            }
            // фейлим все in-flight задачи этого воркера
            for (const [id, p] of pending) {
                p.reject(e);
                pending.delete(id);
            }
        });

        w.on("exit", (code) => {
            console.warn(`[txPool] worker #${i} exited with code ${code}`);
        });

        w.postMessage({ type: "init", protoDir: opts?.protoDir });
    }

    async function waitAllReady() {
        await Promise.all(readyPromises);
    }

    async function submit(txBase64: string): Promise<any> {
        await waitAllReady();
        while (idle.length === 0) await new Promise((r) => setTimeout(r, 0));
        const wid = idle.shift()!;
        const w = workers[wid];

        const id = (Math.random() * 2 ** 31) | 0;
        const prom = new Promise<any>((resolve, reject) => {
            pending.set(id, { resolve, reject });
        });

        w.postMessage({ type: "decode", id, txBase64 });
        return prom;
    }

    async function close() {
        await Promise.all(workers.map((w) => w.terminate()));
    }

    return { submit, close };
}