// src/sink/postgres.ts
/* eslint-disable @typescript-eslint/no-explicit-any */
import { Sink, SinkConfig } from "./types.js";
import { createPgPool, getPgPool, closePgPool } from "../db/pg.js";
import { ensureCorePartitions } from "../db/partitions.js";
import type { PoolClient } from "pg";

export type PostgresMode = "block-atomic" | "batch-insert";

export interface PostgresSinkConfig extends SinkConfig {
    pg: {
        connectionString?: string;
        host?: string;
        port?: number;
        user?: string;
        password?: string;
        database?: string;
        ssl?: boolean;
    };
    mode?: PostgresMode;
    batchSizes?: {
        blocks?: number;
        txs?: number;
        msgs?: number;
        events?: number;
        attrs?: number;
    };
}

type BlockLine = any;

/* ---------------- helpers (вне класса) ---------------- */

function normArray<T>(x: any): T[] {
    return Array.isArray(x) ? x : [];
}

function pickMessages(tx: any): any[] {
    if (Array.isArray(tx?.msgs)) return tx.msgs;
    if (Array.isArray(tx?.messages)) return tx.messages;
    if (Array.isArray(tx?.body?.messages)) return tx.body.messages;
    if (Array.isArray(tx?.decoded?.body?.messages)) return tx.decoded?.body?.messages;
    return [];
}

type NormalizedLog = {
    msg_index: number;
    events: Array<{ type: string; attributes: any }>;
};

function pickLogs(tx: any): NormalizedLog[] {
    // 1) уже нормализовано
    if (Array.isArray(tx?.logsNormalized)) {
        return tx.logsNormalized as NormalizedLog[];
    }
    // 2) классические ABCI logs [{msg_index, events:[{type, attributes:[{key,value}]}]}]
    if (Array.isArray(tx?.tx_response?.logs)) {
        return tx.tx_response.logs.map((l: any) => ({
            msg_index: Number(l?.msg_index ?? -1),
            events: normArray(l?.events).map((ev: any) => ({
                type: String(ev?.type ?? "unknown"),
                attributes: ev?.attributes ?? [],
            })),
        }));
    }
    // 3) плоские events без msg_index → обернём в один лог
    const flat =
        (Array.isArray(tx?.eventsNormalized) && tx.eventsNormalized) ||
        (Array.isArray(tx?.events) && tx.events) ||
        null;

    if (Array.isArray(flat)) {
        return [
            {
                msg_index: -1,
                events: flat.map((ev: any) => ({
                    type: String(ev?.type ?? "unknown"),
                    attributes: ev?.attributes ?? [],
                })),
            },
        ];
    }
    return [];
}

function attrsToPairs(attrs: any): Array<{ key: string; value: string | null }> {
    // массив пар [{key,value}]
    if (Array.isArray(attrs)) {
        return attrs.map((a) => ({
            key: String(a?.key ?? ""),
            value: a?.value != null ? String(a.value) : null,
        }));
    }
    // объект-карта {k: v}
    if (attrs && typeof attrs === "object") {
        return Object.entries(attrs).map(([k, v]) => ({
            key: String(k),
            value: v != null ? String(v as any) : null,
        }));
    }
    return [];
}

/* ---------------- основной класс ---------------- */

export class PostgresSink implements Sink {
    private cfg: PostgresSinkConfig;
    private mode: PostgresMode;

    private bufBlocks: any[] = [];
    private bufTxs: any[] = [];
    private bufMsgs: any[] = [];
    private bufEvents: any[] = [];
    private bufAttrs: any[] = [];

    private batchSizes = {
        blocks: 1000,
        txs: 2000,
        msgs: 5000,
        events: 5000,
        attrs: 10000,
    };

    constructor(cfg: PostgresSinkConfig) {
        this.cfg = cfg;
        this.mode = cfg.mode ?? "batch-insert";
        if (cfg.batchSizes) Object.assign(this.batchSizes, cfg.batchSizes);
    }

    async init(): Promise<void> {
        createPgPool({ ...this.cfg.pg, applicationName: "cosmos-indexer" });
    }

    async write(line: string): Promise<void> {
        let obj: BlockLine;
        try {
            obj = JSON.parse(line);
        } catch {
            return;
        }
        if (obj.error) return;

        if (this.mode === "block-atomic") {
            await this.persistBlockAtomic(obj);
        } else {
            await this.persistBlockBuffered(obj);
        }
    }

    async flush(): Promise<void> {
        if (this.mode === "batch-insert") {
            await this.flushAll();
        }
    }

    async close(): Promise<void> {
        await this.flush?.();
        await closePgPool();
    }

    /* -------------- mapping -------------- */
    private extractRows(blockLine: BlockLine) {
        const height = Number(blockLine?.meta?.height);
        const time = new Date(blockLine?.meta?.time);

        const b = blockLine.block;
        const blockRow = {
            height,
            block_hash: b?.block_id?.hash ?? null,
            time,
            // если у тебя отдельный расчёт proposer — можешь подменить здесь
            proposer_address:
                b?.block?.last_commit?.signatures?.[0]?.validator_address ?? null,
            tx_count: Array.isArray(blockLine?.txs) ? blockLine.txs.length : 0,
            size_bytes: b?.block?.size ?? null,
            last_commit_hash: b?.block?.last_commit?.block_id?.hash ?? null,
            data_hash: b?.block?.data?.hash ?? null,
            evidence_count: Array.isArray(b?.block?.evidence?.evidence)
                ? b.block.evidence.evidence.length
                : 0,
            app_hash: b?.block?.header?.app_hash ?? null,
        };

        const txRows: any[] = [];
        const msgRows: any[] = [];
        const evRows: any[] = [];
        const attrRows: any[] = [];

        const txs = Array.isArray(blockLine?.txs) ? blockLine.txs : [];
        for (const tx of txs) {
            const tx_hash = tx.hash ?? tx.txhash ?? tx.tx_hash ?? null;
            const tx_index = Number(tx.index ?? tx.tx_index ?? 0);
            const code = Number(tx.code ?? 0);
            const gas_wanted = tx.gas_wanted != null ? Number(tx.gas_wanted) : null;
            const gas_used = tx.gas_used != null ? Number(tx.gas_used) : null;
            const fee = tx.fee ?? null;
            const memo = tx.memo ?? null;
            const signers: string[] | null = Array.isArray(tx.signers) ? tx.signers : null;
            const raw_tx = tx.raw_tx ?? null;
            const log_summary = tx.log_summary ?? null;

            txRows.push({
                tx_hash,
                height,
                tx_index,
                code,
                gas_wanted,
                gas_used,
                fee,
                memo,
                signers,
                raw_tx,
                log_summary,
                time,
            });

            // сообщения
            const msgs = pickMessages(tx);
            for (let i = 0; i < msgs.length; i++) {
                const m = msgs[i];
                msgRows.push({
                    tx_hash,
                    msg_index: i,
                    height,
                    type_url: m?.["@type"] ?? m?.type_url ?? "",
                    value: m,
                    signer: m?.signer ?? m?.from_address ?? m?.delegator_address ?? null,
                });
            }

            // события
            const logs = pickLogs(tx);
            for (const log of logs) {
                const msg_index = Number(log?.msg_index ?? -1);
                const events = normArray<any>(log?.events);
                for (let ei = 0; ei < events.length; ei++) {
                    const ev = events[ei];
                    const event_type = String(ev?.type ?? "unknown");
                    const attrsPairs = attrsToPairs(ev?.attributes);
                    // строка события (атрибуты храним как массив пар)
                    evRows.push({
                        tx_hash,
                        msg_index,
                        event_index: ei,
                        event_type,
                        attributes: attrsPairs,
                        height,
                    });

                    // разворачиваем атрибуты
                    for (const { key, value } of attrsPairs) {
                        attrRows.push({
                            tx_hash,
                            msg_index,
                            event_index: ei,
                            key,
                            value,
                            height,
                        });
                    }
                }
            }
        }

        return { blockRow, txRows, msgRows, evRows, attrRows, height };
    }

    /* -------------- block-atomic -------------- */
    private async persistBlockAtomic(blockLine: BlockLine): Promise<void> {
        const pool = getPgPool();
        const { blockRow, txRows, msgRows, evRows, attrRows, height } =
            this.extractRows(blockLine);

        const client = await pool.connect();
        try {
            await ensureCorePartitions(client, height);
            await client.query("BEGIN");
            await this.insertBlocks(client, [blockRow]);
            if (txRows.length) await this.insertTxs(client, txRows);
            if (msgRows.length) await this.insertMsgs(client, msgRows);
            if (evRows.length) await this.insertEvents(client, evRows);
            if (attrRows.length) await this.insertAttrs(client, attrRows);
            await client.query("COMMIT");
        } catch (e) {
            await client.query("ROLLBACK");
            throw e;
        } finally {
            client.release();
        }
    }

    /* -------------- batch-insert -------------- */
    private async persistBlockBuffered(blockLine: BlockLine): Promise<void> {
        const { blockRow, txRows, msgRows, evRows, attrRows } = this.extractRows(blockLine);

        this.bufBlocks.push(blockRow);
        this.bufTxs.push(...txRows);
        this.bufMsgs.push(...msgRows);
        this.bufEvents.push(...evRows);
        this.bufAttrs.push(...attrRows);

        const needFlush =
            this.bufBlocks.length >= this.batchSizes.blocks ||
            this.bufTxs.length >= this.batchSizes.txs ||
            this.bufMsgs.length >= this.batchSizes.msgs ||
            this.bufEvents.length >= this.batchSizes.events ||
            this.bufAttrs.length >= this.batchSizes.attrs;

        if (needFlush) {
            await this.flushAll();
        }
    }

    private async flushAll(): Promise<void> {
        if (
            this.bufBlocks.length === 0 &&
            this.bufTxs.length === 0 &&
            this.bufMsgs.length === 0 &&
            this.bufEvents.length === 0 &&
            this.bufAttrs.length === 0
        )
            return;

        const pool = getPgPool();
        const client = await pool.connect();
        try {
            const heights: number[] = [
                ...this.bufBlocks.map(r => r.height),
                ...this.bufTxs.map(r => r.height),
                ...this.bufMsgs.map(r => r.height),
                ...this.bufEvents.map(r => r.height),
                ...this.bufAttrs.map(r => r.height),
            ].filter((h): h is number => Number.isFinite(h));  // ← фильтр

            if (heights.length === 0) { client.release(); return; }

            const minH = Math.min(...heights);
            const maxH = Math.max(...heights);

            await ensureCorePartitions(client, minH, maxH);

            await client.query("BEGIN");
            await this.flushBlocks(client);
            await this.flushTxs(client);
            await this.flushMsgs(client);
            await this.flushEvents(client);
            await this.flushAttrs(client);
            await client.query("COMMIT");
        } catch (e) {
            await client.query("ROLLBACK");
            throw e;
        } finally {
            client.release();
        }
    }

    /* -------------- SQL helpers -------------- */
    private async flushBlocks(client: PoolClient): Promise<void> {
        if (!this.bufBlocks.length) return;
        const rows = this.bufBlocks;
        this.bufBlocks = [];
        const cols = [
            "height",
            "block_hash",
            "time",
            "proposer_address",
            "tx_count",
            "size_bytes",
            "last_commit_hash",
            "data_hash",
            "evidence_count",
            "app_hash",
        ];
        const { text, values } = makeMultiInsert(
            "core.blocks",
            cols,
            rows,
            "ON CONFLICT (height) DO NOTHING"
        );
        await client.query(text, values);
    }

    private async flushTxs(client: PoolClient): Promise<void> {
        if (!this.bufTxs.length) return;
        const rows = this.bufTxs;
        this.bufTxs = [];
        const cols = [
            "tx_hash",
            "height",
            "tx_index",
            "code",
            "gas_wanted",
            "gas_used",
            "fee",
            "memo",
            "signers",
            "raw_tx",
            "log_summary",
            "time",
        ];
        const { text, values } = makeMultiInsert(
            "core.transactions",
            cols,
            rows,
            "ON CONFLICT (height, tx_hash) DO UPDATE SET gas_used = EXCLUDED.gas_used, log_summary = EXCLUDED.log_summary",
            { fee: "jsonb", raw_tx: "jsonb" }
        );
        await client.query(text, values);
    }

    private async flushMsgs(client: PoolClient): Promise<void> {
        if (!this.bufMsgs.length) return;
        const rows = this.bufMsgs;
        this.bufMsgs = [];
        const cols = ["tx_hash", "msg_index", "height", "type_url", "value", "signer"];
        const { text, values } = makeMultiInsert(
            "core.messages",
            cols,
            rows,
            "ON CONFLICT (height, tx_hash, msg_index) DO NOTHING",
            { value: "jsonb" }
        );
        await client.query(text, values);
    }

    private async flushEvents(client: PoolClient): Promise<void> {
        if (!this.bufEvents.length) return;
        const rows = this.bufEvents;
        this.bufEvents = [];
        const cols = ["tx_hash", "msg_index", "event_index", "event_type", "attributes"];
        const { text, values } = makeMultiInsert(
            "core.events",
            cols,
            rows,
            "ON CONFLICT (tx_hash, msg_index, event_index) DO NOTHING",
            { attributes: "jsonb" }
        );
        await client.query(text, values);
    }

    private async flushAttrs(client: PoolClient): Promise<void> {
        if (!this.bufAttrs.length) return;
        const rows = this.bufAttrs;
        this.bufAttrs = [];
        const cols = ["tx_hash", "msg_index", "event_index", "key", "value"];
        const { text, values } = makeMultiInsert(
            "core.event_attrs",
            cols,
            rows,
            "ON CONFLICT (tx_hash, msg_index, event_index, key) DO NOTHING",
            { attributes: "jsonb" }
        );
        await client.query(text, values);
    }

    /* -------------- per-block INSERTs -------------- */
    private async insertBlocks(client: PoolClient, rows: any[]): Promise<void> {
        const cols = [
            "height",
            "block_hash",
            "time",
            "proposer_address",
            "tx_count",
            "size_bytes",
            "last_commit_hash",
            "data_hash",
            "evidence_count",
            "app_hash",
        ];
        const { text, values } = makeMultiInsert(
            "core.blocks",
            cols,
            rows,
            "ON CONFLICT (height) DO NOTHING"
        );
        await client.query(text, values);
    }

    private async insertTxs(client: PoolClient, rows: any[]): Promise<void> {
        const cols = [
            "tx_hash",
            "height",
            "tx_index",
            "code",
            "gas_wanted",
            "gas_used",
            "fee",
            "memo",
            "signers",
            "raw_tx",
            "log_summary",
            "time",
        ];
        const { text, values } = makeMultiInsert(
            "core.transactions",
            cols,
            rows,
            "ON CONFLICT (height, tx_hash) DO UPDATE SET gas_used = EXCLUDED.gas_used, log_summary = EXCLUDED.log_summary"
        );
        await client.query(text, values);
    }

    private async insertMsgs(client: PoolClient, rows: any[]): Promise<void> {
        const cols = ["tx_hash", "msg_index", "height", "type_url", "value", "signer"];
        const { text, values } = makeMultiInsert(
            "core.messages",
            cols,
            rows,
            "ON CONFLICT (height, tx_hash, msg_index) DO NOTHING"
        );
        await client.query(text, values);
    }

    private async insertEvents(client: PoolClient, rows: any[]): Promise<void> {
        const cols = ["tx_hash", "msg_index", "event_index", "event_type", "attributes"];
        const { text, values } = makeMultiInsert(
            "core.events",
            cols,
            rows,
            "ON CONFLICT (tx_hash, msg_index, event_index) DO NOTHING"
        );
        await client.query(text, values);
    }

    private async insertAttrs(client: PoolClient, rows: any[]): Promise<void> {
        const cols = ["tx_hash", "msg_index", "event_index", "key", "value"];
        const { text, values } = makeMultiInsert(
            "core.event_attrs",
            cols,
            rows,
            "ON CONFLICT (tx_hash, msg_index, event_index, key) DO NOTHING"
        );
        await client.query(text, values);
    }
}

type Casts = Record<string, "jsonb" | "json">;

function makeMultiInsert(
    table: string,
    columns: string[],
    rows: any[],
    conflictClause: string,
    casts: Casts = {}
) {
    const values: any[] = [];
    const chunks: string[] = [];
    let p = 1;

    for (const r of rows) {
        const tuple: string[] = [];
        for (const c of columns) {
            let v = r[c] ?? null;

            // если колонка json/jsonb — приведём к строке JSON (если это объект/массив)
            if (casts[c] && v !== null) {
                if (typeof v !== "string") v = JSON.stringify(v);
            }

            values.push(v);

            const cast = casts[c] ? `::${casts[c]}` : "";
            tuple.push(`$${p++}${cast}`);
        }
        chunks.push(`(${tuple.join(",")})`);
    }

    const text =
        `INSERT INTO ${table} (${columns.join(",")}) ` +
        `VALUES ${chunks.join(",")} ${conflictClause}`;

    return { text, values };
}