"use strict";
const mqtt = require("mqtt");
const WebSocket = require("ws");
const Duplexify = require("duplexify");
const { Transform } = require("stream");
const Agent = require("https-proxy-agent");
const utils = require("../helpers");
const log = require("npmlog");
const { EventEmitter } = require("events");

const TOPICS = [
    "/legacy_web", "/webrtc", "/rtc_multi", "/onevc", "/br_sr", "/sr_res",
    "/t_ms", "/thread_typing", "/orca_typing_notifications",
    "/notify_disconnect", "/orca_presence"
];

module.exports = (funcs, api, ctx) => {
    let onEvent = () => { }, form = {};

    const fetchSeq = async () => {
        ctx.t_mqttCalled = false;
        try {
            const raw = await funcs.post("https://www.facebook.com/api/graphqlbatch/", ctx.jar, form);
            const res = await utils.parseAndCheckLogin(ctx, funcs)(raw);
            const syncId = res[0]?.o0?.data?.viewer?.message_threads?.sync_sequence_id;
            res.die || res.logout || !Array.isArray(res) || !syncId || res.at(-1)?.error_results > 0 ? onEvent(res) : (ctx.lastSeqId = syncId, connect());
        } catch (err) {
            onEvent(err);
        }
    };

    const connect = () => {
        const sid = Math.floor(Math.random() * 9007199254740991) + 1;
        const cookie = ctx.jar.getCookies("https://www.facebook.com").join("; ");
        const host = ctx.mqttEndpoint
            ? `${ctx.mqttEndpoint}&sid=${sid}`
            : `wss://edge-chat.facebook.com/chat?${ctx.region ? `region=${ctx.region.toLowerCase()}&` : ""}sid=${sid}`;

        const user = {
            u: ctx.userID, s: sid, chat_on: ctx.globalOptions.online, fg: false,
            d: utils.getGUID(), ct: "websocket", aid: "219994525426954",
            mqtt_sid: "", cp: 3, ecp: 10, st: [], pm: [], dc: "",
            no_auto_fg: true, gas: null, pack: []
        };

        const mqttOptions = {
            clientId: "mqttwsclient",
            protocolId: "MQIsdp",
            protocolVersion: 3,
            username: JSON.stringify(user),
            clean: true,
            keepalive: 10,
            reconnectPeriod: 5000,
            reschedulePings: true,
            connectTimeout: 10000,
            resubscribe: true,
            queueQoSZero: true
        };

        const client = new mqtt.Client(() => {
            let ws, destroyed = false, pingInterval;
            const stream = Duplexify(undefined, undefined, { objectMode: false });
            const proxy = new Transform({
                objectMode: false,
                transform(chunk, _, cb) {
                    if (ws?.readyState === WebSocket.OPEN) {
                        ws.send(chunk, cb);
                    } else {
                        cb(new Error("WebSocket not open"));
                    }
                },
                flush(cb) {
                    if (ws?.readyState === WebSocket.OPEN) {
                        ws.close(1000);
                    }
                    cb();
                },
            });

            function clearResources() {
                clearInterval(pingInterval);
                pingInterval = null;
                if (ws) {
                    try {
                        ws.removeAllListeners();
                        ws.terminate?.();
                    } catch { }
                    ws = null;
                }
                stream.setReadable(null);
                stream.setWritable(null);
            }

            function reconnect() {
                clearResources();
                if (!destroyed) setTimeout(connect, 2000);
            }

            function destroyStream(err) {
                if (destroyed) return;
                destroyed = true;
                clearResources();
                stream.destroy(err || new Error("WebSocket closed"));
                reconnect();
            }

            function connectWS() {
                destroyed = false;
                const wsOptions = {
                    headers: {
                        Cookie: cookie,
                        Origin: "https://www.facebook.com",
                        "User-Agent": ctx.globalOptions.userAgent,
                        Referer: "https://www.facebook.com/",
                        Host: new URL(host).hostname
                    },
                    origin: "https://www.facebook.com",
                    protocolVersion: 13,
                    ...(ctx.globalOptions.proxy ? { agent: new Agent(ctx.globalOptions.proxy) } : {})
                };

                ws = new WebSocket(host, wsOptions);
                stream.socket = ws;

                ws.on("open", () => {
                    if (destroyed) return;
                    stream.setReadable(proxy);
                    stream.setWritable(proxy);
                    stream.emit("connect");
                    pingInterval = setInterval(() => {
                        try {
                            if (ws?.readyState === WebSocket.OPEN) {
                                ws.ping?.();
                            }
                        } catch (err) {
                            destroyStream(err);
                        }
                    }, 20000);
                });

                ws.on("message", data => {
                    if (destroyed) return;
                    stream.push(Buffer.isBuffer(data) ? data : Buffer.from(data));
                });

                ws.on("error", destroyStream);
                ws.on("close", destroyStream);
            }

            proxy.on("close", destroyStream);
            connectWS();
            return stream;
        }, mqttOptions);

        ctx.mqttClient = global.mqttClient = client;

        client.on("connect", () => {
            log.info("MQTT", "Successfully connected!");
            TOPICS.forEach(t => client.subscribe(t));
            const q = {
                sync_api_version: 10,
                max_deltas_able_to_process: 1000,
                delta_batch_size: 500,
                encoding: "JSON",
                entity_fbid: ctx.userID
            };
            const topic = ctx.syncToken
                ? (q.last_seq_id = ctx.lastSeqId, q.sync_token = ctx.syncToken, "/messenger_sync_get_diffs")
                : (q.initial_titan_sequence_id = ctx.lastSeqId, q.device_params = null, "/messenger_sync_create_queue");

            client.publish(topic, JSON.stringify(q), { qos: 1 });
            const timeout = setTimeout(() => client.end() || fetchSeq(), 5000);
            ctx.tmsWait = () => {
                clearTimeout(timeout);
                ctx.globalOptions.emitReady && onEvent({ type: "ready", error: null });
                delete ctx.tmsWait;
            };
        });

        client.on("message", (topic, raw) => {
            if (topic !== "/t_ms") return;
            let data;
            try { data = JSON.parse(raw); } catch (e) {
                return log.error("MQTT", e);
            }
            typeof ctx.tmsWait === "function" && ctx.tmsWait();
            data.firstDeltaSeqId && data.syncToken && (
                ctx.lastSeqId = data.firstDeltaSeqId,
                ctx.syncToken = data.syncToken
            );
            data.lastIssuedSeqId && (ctx.lastSeqId = parseInt(data.lastIssuedSeqId, 10));
            (data.deltas || []).forEach(delta => require("../handle/parse")(funcs, api, ctx, delta, onEvent));
        });

        ctx.mqttClient.on('error', (err) => {
            console.error(`MQTT Error: ${err.message || err}`);
            const maxAttempts = 5;
            ctx.retryCount = (ctx.retryCount || 0) + 1;

            if (ctx.retryCount > maxAttempts) {
                console.error("Đã đạt giới hạn số lần thử kết nối lại MQTT.");
                return;
            }
            const delay = 1000;
            setTimeout(async () => {
                try {
                    if (ctx.mqttClient) {
                        try {
                            await new Promise((resolve, reject) => {
                                ctx.mqttClient.end(false, (err) => {
                                    if (err) return reject(err);
                                    resolve();
                                });
                                setTimeout(() => resolve(), 3000);
                            });
                        } catch (e) {
                            console.warn('Đóng MQTT lỗi:', e);
                        }
                        ctx.mqttClient.removeAllListeners();
                        ctx.mqttClient = null;
                    }
                    ctx.lastSeqId = null;
                    ctx.syncToken = undefined;
                    ctx.t_mqttCalled = false;
                    console.log('Đang kết nối lại MQTT...');
                    fetchSeq();
                } catch (e) {
                    console.error('Lỗi khi reconnect MQTT:', e);
                }
            }, delay);
        });

    };

    return cb => {
        class Stream extends EventEmitter {
            stopListening(done = () => { }) {
                onEvent = () => { };
                if (!ctx.mqttClient) return done();
                ["/webrtc", "/rtc_multi", "/onevc"].forEach(t => ctx.mqttClient.unsubscribe(t));
                ctx.mqttClient.publish("/browser_close", "{}");
                ctx.mqttClient.end(false, () => (ctx.mqttClient = undefined, done()));
            }
        }

        const stream = new Stream();
        stream.setMaxListeners(30);
        api.stopListening = stream.stopListening.bind(stream);
        onEvent = cb || ((err, msg) => err ? stream.emit("error", err) : stream.emit("message", msg));

        form = {
            av: ctx.globalOptions.pageID,
            queries: JSON.stringify({
                o0: {
                    doc_id: "3336396659757871",
                    query_params: {
                        limit: 1, before: null,
                        tags: ["INBOX"], includeDeliveryReceipts: false, includeSeqID: true
                    }
                }
            })
        };

        (!ctx.firstListen || !ctx.lastSeqId) ? fetchSeq() : connect();
        ctx.firstListen = false;
        return stream;
    };
};