/*
 * WebSocket 信令客户端 (libdatachannel::rtc::WebSocket + nlohmann::json)
 *
 * 协议匹配 https://jywy.yd-ss.com/js/signal.js + js/client.js (实测):
 *   wss://signal.yd-ss.com/websocket?dev=<roomId>
 * 内置 10s ping / 25s 无 pong 重连 / 断线 2s 自动重连
 */
#include "ws_signaling.h"

#include <rtc/rtc.hpp>
#include <rtc/websocket.hpp>
#include <nlohmann/json.hpp>

#include <atomic>
#include <chrono>
#include <condition_variable>
#include <cstdio>
#include <cstdlib>
#include <fstream>
#include <iostream>
#include <mutex>
#include <stdexcept>
#include <thread>
#include <utility>

namespace webrtcpush {

using json = nlohmann::json;

struct WsSignaling::Impl {
    std::recursive_mutex ws_mutex;
    std::shared_ptr<rtc::WebSocket> ws;
    std::string url;

    std::string member_id;
    std::mutex  id_mutex;

    MessageCallback msg_cb;
    StateCallback   state_cb;
    std::mutex      cb_mutex;

    std::atomic<bool> stopped{false};
    std::atomic<bool> want_reconnect{false};   /* onClosed 触发 */
    std::atomic<long long> last_pong_ms{0};    /* 上次 pong 时间 */
    std::atomic<long long> last_keepalive_tick_ms{0};
    std::atomic<unsigned long> pong_count{0};
    std::atomic<bool> keepalive_exited{false};
    std::atomic<uint64_t> generation{0};       /* 忽略旧 WebSocket 的迟到回调 */

    std::thread keepalive_th;
    std::thread watchdog_th;
    std::mutex keepalive_mutex;
    std::condition_variable keepalive_cv;

    bool wait_or_stop(std::chrono::milliseconds timeout){
        std::unique_lock<std::mutex> lk(keepalive_mutex);
        return keepalive_cv.wait_for(lk, timeout, [this](){ return stopped.load(); });
    }

    static long long now_ms(){
        return std::chrono::duration_cast<std::chrono::milliseconds>(
            std::chrono::steady_clock::now().time_since_epoch()).count();
    }

    /* (重)建 WebSocket 并挂回调; 调用方需已锁 ws_mutex */
    void open_locked(){
        {
            std::lock_guard<std::mutex> lk(id_mutex);
            member_id.clear();
        }
        last_pong_ms = now_ms();
        last_keepalive_tick_ms = now_ms();

        const char *ca_file = std::getenv("WEBRTCPUSH_CA_CERT");
        if(!ca_file || !*ca_file)
            ca_file = std::getenv("SSL_CERT_FILE");
        if(!ca_file || !*ca_file){
            throw std::runtime_error(
                "CA certificate is not configured; expected lib/ca-certificates.crt next to the executable");
        }
        std::ifstream ca_stream(ca_file);
        if(!ca_stream.good()){
            throw std::runtime_error(std::string("CA certificate is not readable: ") + ca_file);
        }

        rtc::WebSocket::Configuration config;
        config.caCertificatePemFile = std::string(ca_file);
        config.connectionTimeout = std::chrono::seconds(10);
        /* The deployed browser sends only the JSON keepalive below. Disable
         * libdatachannel's separate 10s protocol-level PING so both periodic
         * send paths do not contend on WsTransport::mSendMutex. */
        config.pingInterval = std::chrono::milliseconds::zero();

        const uint64_t gen = ++generation;
        auto socket = std::make_shared<rtc::WebSocket>(std::move(config));
        ws = socket;

        socket->onOpen([this, gen](){
            if(gen != generation.load() || stopped) return;
            std::cerr << "[ws] open: " << url << std::endl;
            last_pong_ms = now_ms();
            StateCallback cb;
            { std::lock_guard<std::mutex> lk(cb_mutex); cb = state_cb; }
            if(cb) cb(true);
        });

        socket->onClosed([this, gen](){
            if(gen != generation.load()) return;
            std::cerr << "[ws] closed" << std::endl;
            StateCallback cb;
            { std::lock_guard<std::mutex> lk(cb_mutex); cb = state_cb; }
            if(cb) cb(false);
            if(!stopped) want_reconnect = true;   /* keepalive 线程 2s 后重连 */
        });

        socket->onError([this, gen](std::string err){
            if(gen != generation.load()) return;
            std::cerr << "[ws] error: " << err << std::endl;
        });

        socket->onMessage(
            std::function<void(rtc::binary)>{},
            [this, gen](std::string text){
                if(gen == generation.load()) handle_message(text);
            });

        socket->open(url);
    }

    void handle_message(const std::string &text){
        SignalMessage msg;
        try {
            auto j = json::parse(text);
            if(!j.is_object()) return;

            auto sender_of = [&](const json &o) -> std::string {
                for(const char *k : {"memberId", "senderId"}){
                    if(o.contains(k) && o[k].is_string())
                        return o[k].get<std::string>();
                }
                return "";
            };

            if(j.contains("type") && j["type"].is_string()){
                const std::string t = j["type"].get<std::string>();
                if(t == "id"){
                    msg.type    = SignalMessage::Type::Id;
                    msg.message = j.value("message", "");
                    {
                        std::lock_guard<std::mutex> lk(id_mutex);
                        member_id = msg.message;
                    }
                    std::cerr << "[ws] memberId=" << msg.message << std::endl;
                }else if(t == "num"){
                    msg.type = SignalMessage::Type::Num;
                    /* message 可能是数字也可能是字符串 */
                    if(j.contains("message")){
                        if(j["message"].is_number())
                            msg.num = j["message"].get<int>();
                        else if(j["message"].is_string())
                            msg.num = atoi(j["message"].get<std::string>().c_str());
                    }
                }else if(t == "offer" || t == "answer"){
                    msg.type     = (t == "offer") ? SignalMessage::Type::Offer
                                                  : SignalMessage::Type::Answer;
                    msg.sender_id = sender_of(j);
                    if(j.contains("sdp") && j["sdp"].is_string()){
                        msg.sdp = j["sdp"].get<std::string>();
                    }else if(j.contains("description")){
                        const json &d = j["description"];
                        if(d.is_string()){
                            /* description 是字符串化的 JSON */
                            try { msg.sdp = json::parse(d.get<std::string>()).value("sdp", ""); }
                            catch(...){ msg.sdp = d.get<std::string>(); }
                        }else if(d.is_object() && d.contains("sdp")){
                            msg.sdp = d.value("sdp", "");
                        }
                    }
                }else if(t == "pong"){
                    msg.type = SignalMessage::Type::Pong;
                    last_pong_ms = now_ms();
                    const unsigned long count = ++pong_count;
                    if(count == 1 || count % 6 == 0)
                        std::cerr << "[ws] heartbeat ok (pong=" << count << ")\n";
                }else if(t == "leave"){
                    msg.type = SignalMessage::Type::Leave;
                    msg.sender_id = sender_of(j);
                    msg.message   = j.value("message", "");
                    if(msg.sender_id.empty()) msg.sender_id = msg.message;
                }else if(t == "overload"){
                    msg.type = SignalMessage::Type::Overload;
                    msg.message = j.value("message", "");
                }else{
                    return;
                }
            }else if(j.contains("candidate")){
                /* ICE candidate 转发 (无 type 字段) */
                msg.type           = SignalMessage::Type::Candidate;
                msg.sender_id      = sender_of(j);
                if(j["candidate"].is_object()){
                    msg.candidate_json = j["candidate"].dump();
                }else if(j["candidate"].is_string()){
                    /* 服务端有时会把 candidate 对象展开到消息顶层。 */
                    json candidate;
                    candidate["candidate"] = j["candidate"];
                    if(j.contains("sdpMid") && j["sdpMid"].is_string())
                        candidate["sdpMid"] = j["sdpMid"];
                    if(j.contains("sdpMLineIndex") && j["sdpMLineIndex"].is_number())
                        candidate["sdpMLineIndex"] = j["sdpMLineIndex"];
                    msg.candidate_json = candidate.dump();
                }else{
                    return;
                }
            }else{
                return;
            }
        }catch(const std::exception &e){
            std::cerr << "[ws] JSON parse error: " << e.what()
                      << " data=" << text.substr(0, 80) << std::endl;
            return;
        }

        MessageCallback cb;
        {
            std::lock_guard<std::mutex> lk(cb_mutex);
            cb = msg_cb;
        }
        if(cb) cb(msg);
    }
};

WsSignaling::WsSignaling(std::string url)
    : impl_(std::make_unique<Impl>()) {
    impl_->url = std::move(url);
}

WsSignaling::~WsSignaling(){ stop(); }

bool WsSignaling::start(){
    if(impl_->stopped) return false;
    try {
        impl_->last_keepalive_tick_ms = Impl::now_ms();
        impl_->keepalive_exited = false;
        {
            std::lock_guard<std::recursive_mutex> lk(impl_->ws_mutex);
            impl_->open_locked();
        }

        /* 保活线程: 每 10s ping; 25s 无 pong 或断线 -> 重连 */
        impl_->keepalive_th = std::thread([this](){
            long long last_check = Impl::now_ms();
            while(!impl_->stopped){
                impl_->last_keepalive_tick_ms = Impl::now_ms();
                if(impl_->wait_or_stop(std::chrono::seconds(1))) break;
                long long now = Impl::now_ms();
                impl_->last_keepalive_tick_ms = now;

                bool need = impl_->want_reconnect.exchange(false);
                if(!need && now - impl_->last_pong_ms.load() > 25000){
                    std::cerr << "[ws] 25s no pong, reconnect\n";
                    need = true;
                }
                if(need){
                    std::shared_ptr<rtc::WebSocket> old_ws;
                    {
                        std::lock_guard<std::recursive_mutex> lk(impl_->ws_mutex);
                        if(impl_->stopped) break;
                        old_ws = std::move(impl_->ws);
                        ++impl_->generation;
                    }
                    /* close() only starts a WebSocket closing handshake.  If
                     * the peer has already half-closed TCP without a proper
                     * WebSocket close frame, libdatachannel can remain in
                     * CLOSE_WAIT and never finish reconnecting. */
                    try { if(old_ws) old_ws->forceClose(); }catch(...){}
                    if(impl_->wait_or_stop(std::chrono::seconds(2))) break;
                    std::cerr << "[ws] reconnecting...\n";
                    try {
                        std::lock_guard<std::recursive_mutex> lk(impl_->ws_mutex);
                        if(impl_->stopped) break;
                        impl_->open_locked();
                    }catch(const std::exception &e){
                        std::cerr << "[ws] reconnect failed: " << e.what() << "\n";
                        impl_->want_reconnect = true;
                    }
                    continue;
                }

                if(now - last_check >= 10000){
                    last_check = now;
                    std::shared_ptr<rtc::WebSocket> socket;
                    {
                        std::lock_guard<std::recursive_mutex> lk(impl_->ws_mutex);
                        socket = impl_->ws;
                    }
                    if(socket && socket->isOpen()){
                        try {
                            socket->send(std::string("{\"type\":\"ping\",\"message\":\"keeplive\"}"));
                        }catch(...){ impl_->want_reconnect = true; }
                    }
                }
            }
            impl_->keepalive_exited = true;
            impl_->keepalive_cv.notify_all();
        });

        /* A libdatachannel WebSocket send/forceClose can block indefinitely
         * after a half-closed TCP connection. This independent watchdog never
         * touches the socket. The executable supervisor restarts exit 75. */
        impl_->watchdog_th = std::thread([this](){
            while(true){
                {
                    std::unique_lock<std::mutex> lk(impl_->keepalive_mutex);
                    impl_->keepalive_cv.wait_for(lk, std::chrono::seconds(5));
                }
                if(impl_->stopped && impl_->keepalive_exited) break;

                const long long stall = Impl::now_ms() -
                                        impl_->last_keepalive_tick_ms.load();
                if(stall > 30000){
                    std::cerr << "[ws] keepalive worker stalled for " << stall
                              << "ms; process restart required\n";
                    std::fflush(nullptr);
                    std::_Exit(impl_->stopped ? 130 : 75);
                }
            }
        });
        return true;
    }catch(const std::exception &e){
        std::cerr << "[ws] start error: " << e.what() << std::endl;
        return false;
    }
}

void WsSignaling::stop(){
    if(impl_->stopped.exchange(true)) return;
    ++impl_->generation;
    impl_->want_reconnect = false;
    impl_->keepalive_cv.notify_all();

    if(impl_->keepalive_th.joinable()) impl_->keepalive_th.join();
    impl_->keepalive_cv.notify_all();
    if(impl_->watchdog_th.joinable()) impl_->watchdog_th.join();

    std::shared_ptr<rtc::WebSocket> socket;
    {
        std::lock_guard<std::recursive_mutex> lk(impl_->ws_mutex);
        socket = std::move(impl_->ws);
    }
    try { if(socket) socket->forceClose(); }catch(...){}
}

void WsSignaling::set_message_callback(MessageCallback cb){
    std::lock_guard<std::mutex> lk(impl_->cb_mutex);
    impl_->msg_cb = std::move(cb);
}

void WsSignaling::set_state_callback(StateCallback cb){
    std::lock_guard<std::mutex> lk(impl_->cb_mutex);
    impl_->state_cb = std::move(cb);
}

std::string WsSignaling::member_id() const{
    std::lock_guard<std::mutex> lk(impl_->id_mutex);
    return impl_->member_id;
}

bool WsSignaling::send_description(const std::string &sdp, bool is_offer,
                                   bool log_result){
    std::shared_ptr<rtc::WebSocket> socket;
    {
        std::lock_guard<std::recursive_mutex> lk(impl_->ws_mutex);
        socket = impl_->ws;
    }
    if(!socket || !socket->isOpen()){
        if(!impl_->stopped) impl_->want_reconnect = true;
        return false;
    }
    std::string mid;
    {
        std::lock_guard<std::mutex> lk2(impl_->id_mutex);
        mid = impl_->member_id;
    }
    /* 与网页一致: JSON.stringify({memberId, description:{type,sdp}}) */
    json j;
    j["memberId"]    = mid;
    j["description"] = json::object({
        {"type", is_offer ? "offer" : "answer"},
        {"sdp",  sdp}
    });
    std::string payload = j.dump();
    try {
        bool sent = socket->send(payload);
        if(log_result){
            std::cerr << "[ws] send " << (is_offer ? "offer" : "answer")
                      << " sdp=" << sdp.size()
                      << " json=" << payload.size()
                      << " sent=" << (sent ? "yes" : "no") << "\n";
        }
        return sent;
    }catch(const std::exception &e){
        std::cerr << "[ws] send description failed: " << e.what() << "\n";
        if(!impl_->stopped) impl_->want_reconnect = true;
        return false;
    }
}

bool WsSignaling::send_candidate(const std::string &candidate_json,
                                 bool log_result){
    std::shared_ptr<rtc::WebSocket> socket;
    {
        std::lock_guard<std::recursive_mutex> lk(impl_->ws_mutex);
        socket = impl_->ws;
    }
    if(!socket || !socket->isOpen()){
        if(!impl_->stopped) impl_->want_reconnect = true;
        return false;
    }
    std::string mid;
    {
        std::lock_guard<std::mutex> lk2(impl_->id_mutex);
        mid = impl_->member_id;
    }
    /* 与网页一致: {memberId, candidate:{...}} */
    try {
        json j;
        j["memberId"]  = mid;
        j["candidate"] = json::parse(candidate_json);
        std::string payload = j.dump();
        bool sent = socket->send(payload);
        if(log_result){
            std::cerr << "[ws] send ICE candidate json=" << payload.size()
                      << " sent=" << (sent ? "yes" : "no") << "\n";
        }
        return sent;
    }catch(const std::exception &){
        std::string raw = "{\"memberId\":\"" + mid + "\",\"candidate\":" + candidate_json + "}";
        try { return socket->send(raw); }
        catch(...){
            if(!impl_->stopped) impl_->want_reconnect = true;
            return false;
        }
    }
}

} // namespace webrtcpush
