/* eslint-disable no-console */
import { getConfig, printConfig } from "./config.js";
import { createRpcClientFromConfig } from "./rpc/client.js";
import { createTxDecodePool } from "./decode/txPool.js";
import { assembleBlockJsonFromParts } from "./assemble/blockJson.js";
import { safeJsonStringify } from "./utils/json.js";
import { pLimit } from "./utils/pLimit.js";
import { formatDuration } from "./utils/time.js";
import { createSink } from "./sink/index.js";
import { EventEmitter } from "node:events";
EventEmitter.defaultMaxListeners = 0;

async function main() {
    const cfg = getConfig();
    printConfig(cfg);

    const rpc = createRpcClientFromConfig(cfg);

    const defaultProtoDir = new URL("../protos", import.meta.url).pathname;
    const protoDir = process.env.PROTO_DIR || defaultProtoDir;
    console.log("[proto] dir =", protoDir);

    // decode pool
    const poolSize = Math.max(1, Math.min(cfg.concurrency ?? 8, 8));
    const pool = createTxDecodePool(poolSize, { protoDir });

    // progress counters
    const totalBlocks = (cfg.to - cfg.from + 1);
    let processed = 0;                         // blocks completed (assembled)
    const t0 = Date.now();
    let lastLogAt = t0;

    function maybeReportProgress(force = false) {
        const now = Date.now();
        const elapsedSec = (now - t0) / 1000;
        const sinceLastSec = (now - lastLogAt) / 1000;
        const rate = processed > 0 && elapsedSec > 0 ? processed / elapsedSec : 0;
        const remaining = Math.max(0, totalBlocks - processed);
        const etaSec = rate > 0 ? remaining / rate : Infinity;

        const needByCount = processed > 0 && processed % cfg.progressEveryBlocks === 0;
        const needByTime = sinceLastSec >= cfg.progressIntervalSec;

        if (force || needByCount || needByTime) {
            console.log(
                `[speed] ${processed}/${totalBlocks} blocks ` +
                `| elapsed ${formatDuration(elapsedSec)} ` +
                `| rate ${rate.toFixed(1)} blk/s ` +
                `| eta ${formatDuration(etaSec)}`
            );
            lastLogAt = now;
        }
    }

    const sink = createSink({
        kind: cfg.sinkKind,
        outPath: cfg.outPath,
        flushEvery: cfg.flushEvery ?? 1,
        pg: cfg.pg,               // ← pass PG connection params
        batchSizes: {
            blocks: cfg.pg?.batchBlocks,
            txs: cfg.pg?.batchTxs,
            msgs: cfg.pg?.batchMsgs,
            events: cfg.pg?.batchEvents,
            attrs: cfg.pg?.batchAttrs,
        },
    });
    await sink.init();

    // concurrency limiter
    const runLimited = pLimit(cfg.concurrency);

    // ordered output buffer
    const ready = new Map<number, string>();
    let nextToPrint = cfg.from;

    async function tryFlush() {
        let flushed = 0;
        while (ready.has(nextToPrint)) {
            const line = ready.get(nextToPrint)!;
            ready.delete(nextToPrint);
            // console.log(line);              // NDJSON line
            await sink.write(line);
            nextToPrint++;
            flushed++;
            processed++;                    // count block after it is flushed/ready
        }
        if (flushed > 0) maybeReportProgress(false);
    }

    async function processHeight(h: number) {
        try {
            const [b, br] = await Promise.all([rpc.fetchBlock(h), rpc.fetchBlockResults(h)]);
            const txsB64: string[] = b?.block?.data?.txs ?? [];
            const decoded = await Promise.all(txsB64.map((x) => pool.submit(x)));
            const assembled = await assembleBlockJsonFromParts(rpc, b, br, decoded, cfg.caseMode);
            const line = safeJsonStringify(assembled);
            ready.set(h, line);
            tryFlush();
        } catch (e: any) {
            const errLine = JSON.stringify({ height: h, error: String(e?.message ?? e) });
            ready.set(h, errLine);
            tryFlush();
        }
    }

// Sliding window runner: keep at most `cfg.concurrency` in-flight
    let nextHeight = cfg.from;
    let inFlight = 0;

    await new Promise<void>((resolve) => {
        const maybeSpawn = () => {
            while (inFlight < cfg.concurrency && nextHeight <= cfg.to) {
                const h = nextHeight++;
                inFlight++;
                processHeight(h).finally(() => {
                    inFlight--;
                    if (nextHeight > cfg.to && inFlight === 0) {
                        resolve();
                    } else {
                        // schedule next tick to avoid deep recursion
                        setImmediate(maybeSpawn);
                    }
                });
            }
        };
        maybeSpawn();
    });

    await pool.close();
    await sink.flush?.();
    await sink.close();

// финальный отчёт (как у тебя уже сделано)
    const elapsedSec = (Date.now() - t0) / 1000;
    const rate = processed > 0 && elapsedSec > 0 ? processed / elapsedSec : 0;
    console.log(
        `[done] ${processed}/${totalBlocks} blocks | elapsed ${formatDuration(elapsedSec)} | avg ${rate.toFixed(1)} blk/s`
    );
    maybeReportProgress(true);
    // force final progress line (useful if мало блоков и не сработали интервалы)
    maybeReportProgress(true);
}

main().catch((e) => {
    console.error(e);
    process.exit(1);
});