#include "engine.hpp"
#include <fstream>
#include <sstream>
#include <cmath>
#include <sys/stat.h>
#include <sys/statvfs.h>
#include <sys/inotify.h>
#include <arpa/inet.h>
#include <pwd.h>
#include <grp.h>
#include "http.hpp"

ThreadPool::ThreadPool(size_t n) {
    workers_.reserve(n);
    for (size_t i = 0; i < n; ++i) {
        workers_.emplace_back([this] {
            for (;;) {
                std::function<void()> task;
                {
                    std::unique_lock<std::mutex> lock(mutex_);
                    cv_.wait(lock, [this] { return shutdown_ || !tasks_.empty(); });
                    if (shutdown_ && tasks_.empty()) return;
                    task = std::move(tasks_.front());
                    tasks_.pop();
                }
                try { task(); } catch (const std::exception& e) { LOG(ERROR, "pool: " << e.what()); }
            }
        });
    }
    LOG(INFO, "pool: " << n << " workers");
}

ThreadPool::~ThreadPool() {
    { std::lock_guard<std::mutex> lock(mutex_); shutdown_ = true; }
    cv_.notify_all();
    for (auto& w : workers_) if (w.joinable()) w.join();
    LOG(INFO, "pool: shutdown");
}

HbMgr::HbMgr(OriginBridge* b, EngineConfig* c) : bridge_(b), cfg_(c) {}
HbMgr::~HbMgr() { stop(); }

void HbMgr::start() {
    if (running_.exchange(true)) return;
    thread_ = std::thread(&HbMgr::run_loop, this);
    LOG(INFO, "hb: started");
}

void HbMgr::stop() {
    if (!running_.exchange(false)) return;
    if (thread_.joinable()) thread_.join();
    LOG(INFO, "hb: stopped");
}

void HbMgr::run_loop() {
    while (running_.load()) {
        if (bridge_) {
            if (!bridge_->send_heartbeat(cpu_pct_.load(), mem_pct_.load(), disk_pct_.load(),
                    wg_peers_.load(), ServiceState::HEALTHY, cfg_->node_role)) {
                LOG(WARN, "hb: send failed");
            }
        }
        for (uint32_t i = 0; i < cfg_->heartbeat_interval_s && running_.load(); ++i)
            std::this_thread::sleep_for(std::chrono::seconds(1));
    }
}

WdMgr::WdMgr(EngineConfig* c) : cfg_(c) {}
WdMgr::~WdMgr() { stop(); }

void WdMgr::start() {
    if (running_.exchange(true)) return;
    thread_ = std::thread(&WdMgr::run_loop, this);
    LOG(INFO, "wd: started");
}

void WdMgr::stop() {
    if (!running_.exchange(false)) return;
    if (thread_.joinable()) thread_.join();
    LOG(INFO, "wd: stopped");
}

void WdMgr::run_loop() {
    uint32_t chattr_cnt = 0;
    while (running_.load()) {
        if (++chattr_cnt >= cfg_->chattr_check_interval_s) {
            chattr_cnt = 0;
            std::string cmd = std::string("lsattr -d ") + cfg_->scripts_dir + " 2>/dev/null";
            FILE* fp = popen(cmd.c_str(), "r");
            if (fp) { char b[256]; if (fgets(b, sizeof(b), fp) && !strstr(b, "i")) LOG(WARN, "wd: scripts dir mutable"); pclose(fp); }
            cmd = std::string("lsattr -d ") + cfg_->scripts_enc + " 2>/dev/null";
            fp = popen(cmd.c_str(), "r");
            if (fp) { char b[256]; if (fgets(b, sizeof(b), fp) && !strstr(b, "i")) LOG(WARN, "wd: scripts_enc mutable"); pclose(fp); }
        }
        if (access(E_PID, F_OK) != 0) LOG(WARN, "wd: PID file missing");
        for (uint32_t i = 0; i < cfg_->watchdog_interval_s && running_.load(); ++i)
            std::this_thread::sleep_for(std::chrono::seconds(1));
    }
}

DdMon::DdMon(EngineConfig* c) : cfg_(c) {}
DdMon::~DdMon() { stop(); }

void DdMon::start() {
    if (running_.exchange(true)) return;
    thread_ = std::thread(&DdMon::run_loop, this);
    LOG(INFO, "dd: started");
}

void DdMon::stop() {
    if (!running_.exchange(false)) return;
    if (thread_.joinable()) thread_.join();
    LOG(INFO, "dd: stopped");
}

void DdMon::run_loop() {
    while (running_.load()) {
        uint32_t syn = syn_pps_.exchange(0);
        uint32_t udp = udp_pps_.exchange(0);
        uint32_t http5xx = http_5xx_per_sec_.exchange(0);
        uint32_t bw = bw_mbps_.exchange(0);
        AttackType detected = AttackType::NONE;
        if (syn > cfg_->ddos_syn_threshold) {
            detected = AttackType::DDOS_SYN;
            LOG(WARN, "dd: SYN flood " << syn << " pps");
        } else if (udp > cfg_->ddos_udp_threshold) {
            detected = AttackType::DDOS_UDP;
            LOG(WARN, "dd: UDP flood " << udp << " pps");
        } else if (http5xx > cfg_->ddos_http_err_threshold) {
            detected = AttackType::L7_HTTP;
            LOG(WARN, "dd: HTTP flood " << http5xx << " 5xx/s");
        } else if (bw > cfg_->ddos_bw_threshold_mb) {
            detected = AttackType::DDOS_ICMP;
            LOG(WARN, "dd: BW flood " << bw << " MB/s");
        }
        if (detected != AttackType::NONE) {
            current_attack_.store(static_cast<uint8_t>(detected));
        } else if (current_attack_.load() != static_cast<uint8_t>(AttackType::NONE)) {
            LOG(INFO, "dd: attack cleared");
            current_attack_.store(static_cast<uint8_t>(AttackType::NONE));
        }
        std::this_thread::sleep_for(std::chrono::seconds(cfg_->ddos_check_interval_s));
    }
}

FoMgr::FoMgr(OriginBridge* b, EngineConfig* c) : bridge_(b), cfg_(c) {}
FoMgr::~FoMgr() { stop(); }

void FoMgr::start() {
    if (running_.exchange(true)) return;
    thread_ = std::thread(&FoMgr::run_loop, this);
    LOG(INFO, "fo: started");
}

void FoMgr::stop() {
    if (!running_.exchange(false)) return;
    if (thread_.joinable()) thread_.join();
    LOG(INFO, "fo: stopped");
}

void FoMgr::run_loop() {
    while (running_.load()) {
        if (failover_active_.load()) {
            std::string node;
            { std::lock_guard<std::mutex> lock(failover_mutex_); node = failed_node_; }
            if (!node.empty()) {
                std::string cmd = "ping -c 1 -W 3 " + node + " 2>/dev/null | grep -c '1 received'";
                FILE* fp = popen(cmd.c_str(), "r");
                bool recovered = false;
                if (fp) { char b[16]; if (fgets(b, sizeof(b), fp)) recovered = (atoi(b) == 1); pclose(fp); }
                if (recovered) { LOG(INFO, "fo: recovered " << node); recover_node(node); }
            }
        }
        std::this_thread::sleep_for(std::chrono::milliseconds(cfg_->failover_timeout_ms));
    }
}

bool FoMgr::trigger_failover(const std::string& node) {
    std::lock_guard<std::mutex> lock(failover_mutex_);
    LOG(INFO, "fo: failover " << node);
    failed_node_ = node;
    failover_active_.store(true);
    if (bridge_) {
        Json::Value r; r["event"] = "failover"; r["node"] = node; r["timestamp"] = static_cast<Json::UInt>(now_epoch());
        Json::FastWriter w; std::string b = w.write(r); bridge_->sign_request("POST", "/api/events/failover", b);
    }
    return true;
}

bool FoMgr::recover_node(const std::string& node) {
    std::lock_guard<std::mutex> lock(failover_mutex_);
    LOG(INFO, "fo: recover " << node);
    failed_node_.clear();
    failover_active_.store(false);
    return true;
}

ScMgr::ScMgr(EngineConfig* c) : cfg_(c) {}
ScMgr::~ScMgr() { stop(); if (watch_fd_ >= 0) close(watch_fd_); }

void ScMgr::start() {
    if (running_.exchange(true)) return;
    load_whitelist();
    watch_fd_ = inotify_init1(IN_NONBLOCK | IN_CLOEXEC);
    if (watch_fd_ >= 0) watch_wd_ = inotify_add_watch(watch_fd_, cfg_->scripts_dir.c_str(), IN_MODIFY | IN_CREATE | IN_DELETE);
    thread_ = std::thread(&ScMgr::run_loop, this);
    LOG(INFO, "sc: started");
}

void ScMgr::stop() {
    if (!running_.exchange(false)) return;
    if (thread_.joinable()) thread_.join();
    if (watch_wd_ >= 0) { inotify_rm_watch(watch_fd_, watch_wd_); watch_wd_ = -1; }
    if (watch_fd_ >= 0) { close(watch_fd_); watch_fd_ = -1; }
    LOG(INFO, "sc: stopped");
}

void ScMgr::run_loop() {
    while (running_.load()) {
        if (watch_fd_ >= 0) {
            char b[4096]; ssize_t n;
            while ((n = read(watch_fd_, b, sizeof(b))) > 0) {
                for (char* p = b; p < b + n; ) {
                    struct inotify_event* ev = reinterpret_cast<struct inotify_event*>(p);
                    if (ev->len > 0) {
                        std::string name(ev->name);
                        if (ev->mask & IN_CREATE) LOG(INFO, "sc: created " << name);
                        if (ev->mask & IN_MODIFY) LOG(INFO, "sc: modified " << name);
                        if (ev->mask & IN_DELETE) LOG(INFO, "sc: deleted " << name);
                    }
                    p += sizeof(struct inotify_event) + ev->len;
                }
            }
        }
        std::this_thread::sleep_for(std::chrono::seconds(5));
    }
}

void ScMgr::load_whitelist() {
    std::lock_guard<std::mutex> lock(whitelist_mutex_);
    script_whitelist_ = {"nodes.sh","logs.sh","sign.sh","watchdog.sh","encrypt.sh","healthcheck.sh","backup.sh","sync.sh"};
    std::string p = std::string(E_BASE) + "/config/script_whitelist.txt";
    std::ifstream fs(p);
    if (fs.is_open()) {
        std::string line;
        while (std::getline(fs, line)) {
            line.erase(0, line.find_first_not_of(" \t\r\n"));
            line.erase(line.find_last_not_of(" \t\r\n") + 1);
            if (!line.empty() && line[0] != '#') script_whitelist_.insert(line);
        }
        fs.close();
    }
}

std::pair<int, std::string> ScMgr::execute_script(const std::string& name, const std::vector<std::string>& args) {
    {
        std::lock_guard<std::mutex> lock(whitelist_mutex_);
        if (script_whitelist_.find(name) == script_whitelist_.end()) {
            LOG(WARN, "sc: not whitelisted: " << name);
            return {-1, "not whitelisted"};
        }
    }
    std::string path = cfg_->scripts_dir + "/" + name;
    struct stat st;
    if (stat(path.c_str(), &st) != 0 || !S_ISREG(st.st_mode)) {
        LOG(ERROR, "sc: not found: " << path);
        return {-1, "not found"};
    }
    std::string cmd = path;
    for (const auto& a : args) {
        if (a.find_first_of(";|&$`<>\\\"'") != std::string::npos) {
            LOG(WARN, "sc: bad chars in arg");
            return {-1, "bad arg"};
        }
        cmd += " " + a;
    }
    cmd += " 2>&1";
    LOG(INFO, "sc: exec " << name);
    FILE* fp = popen(cmd.c_str(), "r");
    if (!fp) return {-1, "popen failed"};
    std::ostringstream out;
    char b[1024]; size_t total = 0;
    while (fgets(b, sizeof(b), fp)) {
        total += strlen(b);
        if (total > MAX_SCRIPT_OUTPUT) { out << "\n[TRUNCATED]"; break; }
        out << b;
    }
    int rc = pclose(fp);
    LOG(INFO, "sc: " << name << " exit=" << WEXITSTATUS(rc));
    return {WEXITSTATUS(rc), out.str()};
}

bool ScMgr::reload_scripts() {
    LOG(INFO, "sc: reload whitelist");
    load_whitelist();
    return true;
}

FwMgr::FwMgr(EngineConfig* c) : cfg_(c) {}
FwMgr::~FwMgr() { stop(); }

void FwMgr::start() {
    if (running_.exchange(true)) return;
    thread_ = std::thread(&FwMgr::run_loop, this);
    LOG(INFO, "fw: started");
}

void FwMgr::stop() {
    if (!running_.exchange(false)) return;
    if (thread_.joinable()) thread_.join();
    LOG(INFO, "fw: stopped");
}

void FwMgr::run_loop() {
    while (running_.load()) {
        {
            std::lock_guard<std::mutex> lock(blocks_mutex_);
            uint32_t now = now_epoch();
            auto it = active_blocks_.begin();
            while (it != active_blocks_.end()) {
                if (it->expire_at > 0 && it->expire_at < now) {
                    std::string c = "iptables -D INPUT -s " + it->ip + " -j DROP 2>/dev/null";
                    system(c.c_str());
                    it = active_blocks_.erase(it);
                } else ++it;
            }
        }
        std::this_thread::sleep_for(std::chrono::seconds(5));
    }
}

bool FwMgr::apply_rule(const FirewallRule& r) {
    std::string cmd = "iptables -A " + r.chain;
    if (!r.protocol.empty() && r.protocol != "all") cmd += " -p " + r.protocol;
    if (!r.src.empty()) cmd += " -s " + r.src;
    if (!r.dst.empty()) cmd += " -d " + r.dst;
    if (r.dport > 0) cmd += " --dport " + std::to_string(r.dport);
    cmd += " -j " + r.action;
    if (!r.comment.empty()) cmd += " -m comment --comment \"" + r.comment + "\"";
    cmd += " 2>&1";
    FILE* fp = popen(cmd.c_str(), "r");
    if (!fp) return false;
    char e[256]; std::string err;
    while (fgets(e, sizeof(e), fp)) err += e;
    int rc = pclose(fp);
    if (rc != 0) { LOG(WARN, "fw: iptables fail: " << err); return false; }
    LOG(INFO, "fw: rule " << r.chain << " " << r.action << " " << r.protocol);
    return true;
}

bool FwMgr::apply_rules(const std::vector<FirewallRule>& rules) {
    bool ok = true;
    for (const auto& r : rules) if (!apply_rule(r)) ok = false;
    return ok;
}

bool FwMgr::clear_dynamic_rules() {
    system("iptables -F SE-CHAIN 2>/dev/null || true");
    { std::lock_guard<std::mutex> lock(blocks_mutex_); active_blocks_.clear(); }
    LOG(INFO, "fw: cleared dynamic");
    return true;
}

bool FwMgr::block_ip(const std::string& ip, const std::string& reason, uint32_t expire_at) {
    struct in_addr addr;
    if (inet_pton(AF_INET, ip.c_str(), &addr) != 1) { LOG(WARN, "fw: bad IP " << ip); return false; }
    {
        std::lock_guard<std::mutex> lock(blocks_mutex_);
        for (const auto& b : active_blocks_) if (b.ip == ip) return true;
    }
    std::string cmd = "iptables -I INPUT 1 -s " + ip + " -j DROP -m comment --comment \"NN:" + reason + "\" 2>&1";
    FILE* fp = popen(cmd.c_str(), "r");
    if (!fp) return false;
    char e[256]; std::string err; while (fgets(e, sizeof(e), fp)) err += e;
    int rc = pclose(fp);
    if (rc != 0) { LOG(WARN, "fw: block fail: " << err); return false; }
    { std::lock_guard<std::mutex> lock(blocks_mutex_); active_blocks_.push_back({ip, expire_at, reason}); }
    LOG(INFO, "fw: blocked " << ip << " " << reason);
    return true;
}

bool FwMgr::unblock_ip(const std::string& ip) {
    std::string cmd = "iptables -D INPUT -s " + ip + " -j DROP 2>/dev/null";
    system(cmd.c_str());
    { std::lock_guard<std::mutex> lock(blocks_mutex_); active_blocks_.erase(
        std::remove_if(active_blocks_.begin(), active_blocks_.end(), [&ip](const BlockEntry& b) { return b.ip == ip; }),
        active_blocks_.end()); }
    LOG(INFO, "fw: unblocked " << ip);
    return true;
}

LdMgr::LdMgr(FwMgr* f, OriginBridge* b, EngineConfig* c) : fw_(f), bridge_(b), cfg_(c) {}
LdMgr::~LdMgr() { if (locked_down_.load()) exit_lockdown(); }

bool LdMgr::enter_lockdown(const std::string& reason) {
    if (locked_down_.exchange(true)) { LOG(WARN, "ld: already locked"); return false; }
    { std::lock_guard<std::mutex> lock(mutex_); lockdown_reason_ = reason; }
    LOG(FATAL, "ld: ENTER " << reason);
    if (fw_) {
        FirewallRule b_all; b_all.chain = "INPUT"; b_all.action = "DROP"; b_all.protocol = "all"; b_all.comment = "SE-LOCK:" + reason;
        FirewallRule a_lo; a_lo.chain = "INPUT"; a_lo.action = "ACCEPT"; a_lo.src = "127.0.0.1"; a_lo.comment = "SE-LOCK:lo";
        FirewallRule a_est; a_est.chain = "INPUT"; a_est.action = "ACCEPT"; a_est.comment = "SE-LOCK:est";
        FirewallRule a_wg; a_wg.chain = "INPUT"; a_wg.action = "ACCEPT"; a_wg.protocol = "udp"; a_wg.comment = "SE-LOCK:wg";
        fw_->apply_rule(a_lo);
        fw_->apply_rule(a_est);
        fw_->apply_rule(a_wg);
        fw_->apply_rule(b_all);
    }
    if (bridge_) {
        Json::Value r; r["event"] = "lockdown"; r["reason"] = reason; r["node_id"] = cfg_->node_id; r["timestamp"] = static_cast<Json::UInt>(now_epoch());
        Json::FastWriter w; std::string b = w.write(r); bridge_->sign_request("POST", "/api/events/lockdown", b);
    }
    return true;
}

bool LdMgr::exit_lockdown() {
    if (!locked_down_.exchange(false)) { LOG(WARN, "ld: not locked"); return false; }
    LOG(INFO, "ld: EXIT");
    if (fw_) fw_->clear_dynamic_rules();
    { std::lock_guard<std::mutex> lock(mutex_); lockdown_reason_.clear(); }
    if (bridge_) {
        Json::Value r; r["event"] = "recover"; r["node_id"] = cfg_->node_id; r["timestamp"] = static_cast<Json::UInt>(now_epoch());
        Json::FastWriter w; std::string b = w.write(r); bridge_->sign_request("POST", "/api/events/recover", b);
    }
    return true;
}

Engine::Engine(EngineConfig& cfg) : cfg_(cfg) {
    LOG(INFO, "Engine v" << E_VERSION << " init");
    if (sodium_init() < 0) throw std::runtime_error("sodium init failed");
    LOG(INFO, "sodium OK");
    init_crypto();
    init_bridge();
    pool_ = std::make_unique<ThreadPool>(cfg_.thread_pool_size);
    if (cfg_.enable_heartbeat) hb_ = std::make_unique<HbMgr>(bridge_.get(), &cfg_);
    if (cfg_.enable_watchdog) wd_ = std::make_unique<WdMgr>(&cfg_);
    if (cfg_.enable_ddos) dd_ = std::make_unique<DdMon>(&cfg_);
    if (cfg_.enable_failover) fo_ = std::make_unique<FoMgr>(bridge_.get(), &cfg_);
    if (cfg_.enable_scripts) sc_ = std::make_unique<ScMgr>(&cfg_);
    if (cfg_.enable_firewall) fw_ = std::make_unique<FwMgr>(&cfg_);
    if (fw_ && cfg_.enable_lockdown) ld_ = std::make_unique<LdMgr>(fw_.get(), bridge_.get(), &cfg_);
    state_.store(static_cast<uint8_t>(ServiceState::HEALTHY));
    LOG(INFO, "init complete");
}

Engine::~Engine() { shutdown(); }

void Engine::init_crypto() {
    crypto_ = std::make_unique<CryptoEngine>();
    try { crypto_->load_key(cfg_.key_file); }
    catch (const std::exception& e) {
        std::vector<std::string> alt = {"/var/www/private/.env", "/var/lib/sysc-engine/master.key", std::string(E_KEYS) + "/master.key"};
        bool loaded = false;
        for (const auto& p : alt) {
            if (access(p.c_str(), R_OK) == 0) { try { crypto_->load_key(p); loaded = true; break; } catch (...) { continue; } }
        }
        if (!loaded) throw std::runtime_error(std::string("no key: ") + e.what());
    }
}

void Engine::init_bridge() {
    bridge_ = std::make_unique<OriginBridge>(cfg_.origin_url, cfg_.node_id, *crypto_);
}

void Engine::start_subsystems() {
    if (hb_) hb_->start();
    if (wd_) wd_->start();
    if (dd_) dd_->start();
    if (fo_) fo_->start();
    if (sc_) sc_->start();
    if (fw_) fw_->start();
    LOG(INFO, "subsystems started");
}

void Engine::stop_subsystems() {
    if (hb_) hb_->stop();
    if (wd_) wd_->stop();
    if (dd_) dd_->stop();
    if (fo_) fo_->stop();
    if (sc_) sc_->stop();
    if (fw_) fw_->stop();
    LOG(INFO, "subsystems stopped");
}

void Engine::run() {
    if (running_.exchange(true)) { LOG(WARN, "already running"); return; }
    shutdown_requested_.store(false);
    state_.store(static_cast<uint8_t>(ServiceState::HEALTHY));
    LOG(INFO, "event loop start");
    start_subsystems();
    HttpServer http_srv(cfg_.listen_port, this);
    http_srv.start();
    event_loop();
    http_srv.stop();
    stop_subsystems();
    running_.store(false);
    LOG(INFO, "event loop exit");
}

void Engine::shutdown() {
    if (!running_.load()) return;
    LOG(INFO, "shutdown");
    shutdown_requested_.store(true);
    state_.store(static_cast<uint8_t>(ServiceState::STOPPED));
}

void Engine::event_loop() {
    uint32_t mc = 0, cc = 0, hc = 0;
    while (!shutdown_requested_.load()) {
        if (++mc >= 5) { mc = 0; collect_metrics(); }
        if (++cc >= 10) { cc = 0; process_command_queue(); }
        if (++hc >= 30) { hc = 0; check_health(); }
        OriginCommand cmd;
        while (command_queue_.pop(cmd, 0)) handle_command(cmd);
        if (hb_) hb_->set_metrics(cpu_pct_.load(), mem_pct_.load(), disk_pct_.load(), wg_peer_count_.load());
        std::this_thread::sleep_for(std::chrono::seconds(1));
    }
}

void Engine::process_command_queue() {
    if (!bridge_) return;
    OriginCommand cmd = bridge_->receive_command();
    if (cmd.valid()) {
        command_queue_.push(cmd);
        LOG(INFO, "cmd type=" << static_cast<int>(cmd.type) << " seq=" << cmd.seq);
    }
}

void Engine::collect_metrics() {
    cpu_pct_.store(get_cpu_usage());
    mem_pct_.store(get_mem_usage());
    disk_pct_.store(get_disk_usage());
    wg_peer_count_.store(get_wg_peers());
}

void Engine::check_health() {
    uint32_t to = consecutive_timeouts_.load();
    if (to > 3) {
        LOG(WARN, "hb timeouts=" << to);
        state_.store(static_cast<uint8_t>(ServiceState::DEGRADED));
    } else {
        state_.store(static_cast<uint8_t>(ServiceState::HEALTHY));
    }
    if (dd_ && dd_->under_attack()) LOG(WARN, "DDoS: " << attack_name(dd_->current_attack()));
    if (fo_ && fo_->is_failover_active()) LOG(INFO, "failover active");
}

void Engine::handle_command(const OriginCommand& cmd) {
    LOG(INFO, "handle cmd=" << static_cast<int>(cmd.type) << " seq=" << cmd.seq);
    bool ok = false;
    std::string out, err;
    try {
        switch (cmd.type) {
            case OriginCommandType::UPDATE_RULES: {
                if (!cmd.payload.empty()) {
                    Json::Value r; Json::Reader rd;
                    if (rd.parse(cmd.payload, r) && r.isArray()) {
                        std::vector<std::string> rules;
                        for (const auto& v : r) if (v.isString()) rules.push_back(v.asString());
                        if (fw_) for (const auto& s : rules) { FirewallRule fr; fr.chain = "INPUT"; fr.action = "DROP"; fr.comment = "SE-CMD:" + std::to_string(cmd.seq); fw_->apply_rule(fr); }
                        ok = true;
                    }
                }
                break;
            }
            case OriginCommandType::EXEC_SCRIPT: {
                if (sc_ && !cmd.payload.empty()) {
                    std::istringstream iss(cmd.payload);
                    std::string name; iss >> name;
                    std::vector<std::string> args;
                    std::string a; while (iss >> a) args.push_back(a);
                    auto r = sc_->execute_script(name, args);
                    ok = (r.first == 0); out = r.second;
                }
                break;
            }
            case OriginCommandType::REBOOT: {
                LOG(INFO, "reboot cmd"); shutdown(); ok = true; out = "reboot";
                break;
            }
            case OriginCommandType::STATUS_REQ: {
                Json::Value s; s["node_id"] = cfg_.node_id; s["role"] = role_name(cfg_.node_role);
                s["state"] = state_name(static_cast<ServiceState>(state_.load())); s["version"] = E_VERSION;
                s["cpu"] = cpu_pct_.load(); s["mem"] = mem_pct_.load(); s["disk"] = disk_pct_.load();
                s["ddos_active"] = (dd_ && dd_->under_attack());
                s["failover_active"] = (fo_ && fo_->is_failover_active());
                s["lockdown"] = (ld_ && ld_->is_locked_down());
                Json::FastWriter w; out = w.write(s); ok = true;
                break;
            }
            case OriginCommandType::LOCKDOWN: {
                if (ld_) ok = ld_->enter_lockdown(cmd.payload.empty() ? "origin_cmd" : cmd.payload);
                break;
            }
            case OriginCommandType::RECOVER: {
                if (ld_) ok = ld_->exit_lockdown();
                break;
            }
            case OriginCommandType::CONFIG_PUSH: {
                if (!cmd.payload.empty()) { LOG(INFO, "config push"); ok = true; out = "applied"; }
                break;
            }
            case OriginCommandType::PING: {
                ok = true; out = "pong";
                break;
            }
            default: {
                err = "unknown cmd: " + std::to_string(static_cast<int>(cmd.type));
                LOG(WARN, err);
                break;
            }
        }
    } catch (const std::exception& e) { err = std::string("exception: ") + e.what(); LOG(ERROR, err); }
    if (bridge_ && cmd.type != OriginCommandType::UNKNOWN) bridge_->report_result(cmd.seq, ok, out, err);
}

bool Engine::drop_privileges() {
    if (!cfg_.drop_privileges) return true;
    LOG(INFO, "drop privs to " << cfg_.run_user << ":" << cfg_.run_group);
    struct group* gr = getgrnam(cfg_.run_group.c_str());
    if (!gr) { LOG(ERROR, "group not found: " << cfg_.run_group); return false; }
    struct passwd* pw = getpwnam(cfg_.run_user.c_str());
    if (!pw) { LOG(ERROR, "user not found: " << cfg_.run_user); return false; }
    if (initgroups(pw->pw_name, gr->gr_gid) != 0) { LOG(ERROR, "initgroups: " << strerror(errno)); return false; }
    if (setgid(gr->gr_gid) != 0) { LOG(ERROR, "setgid: " << strerror(errno)); return false; }
    if (setuid(pw->pw_uid) != 0) { LOG(ERROR, "setuid: " << strerror(errno)); return false; }
    if (setuid(0) == 0 || getuid() == 0) { LOG(FATAL, "priv drop FAILED"); return false; }
    LOG(INFO, "privs dropped UID=" << getuid() << " GID=" << getgid());
    return true;
}

uint8_t Engine::get_cpu_usage() {
    static uint8_t last = 0;
    std::ifstream fs("/proc/stat");
    if (!fs.is_open()) return last;
    std::string line;
    if (!std::getline(fs, line)) return last;
    fs.close();
    std::istringstream iss(line);
    std::string label; iss >> label;
    if (label != "cpu") return last;
    unsigned long u, n, s, i, io, iq, sq, st;
    iss >> u >> n >> s >> i >> io >> iq >> sq >> st;
    unsigned long total = u + n + s + i + io + iq + sq + st;
    unsigned long active = u + n + s + iq + sq + st;
    if (total == 0) return last;
    last = static_cast<uint8_t>((active * 100) / total);
    return last;
}

uint8_t Engine::get_mem_usage() {
    std::ifstream fs("/proc/meminfo");
    if (!fs.is_open()) return 0;
    unsigned long total = 0, free = 0, buffers = 0, cached = 0;
    std::string line;
    while (std::getline(fs, line)) {
        if (line.find("MemTotal:") == 0) sscanf(line.c_str(), "MemTotal: %lu", &total);
        else if (line.find("MemFree:") == 0) sscanf(line.c_str(), "MemFree: %lu", &free);
        else if (line.find("Buffers:") == 0) sscanf(line.c_str(), "Buffers: %lu", &buffers);
        else if (line.find("Cached:") == 0 && line.find("SwapCached") != 0) sscanf(line.c_str(), "Cached: %lu", &cached);
    }
    fs.close();
    if (total == 0) return 0;
    return static_cast<uint8_t>(((total - free - buffers - cached) * 100) / total);
}

uint8_t Engine::get_disk_usage() {
    struct statvfs st;
    if (statvfs("/", &st) != 0) return 0;
    unsigned long total = st.f_blocks * st.f_frsize;
    unsigned long free = st.f_bavail * st.f_frsize;
    if (total == 0) return 0;
    return static_cast<uint8_t>(((total - free) * 100) / total);
}

uint32_t Engine::get_wg_peers() {
    FILE* fp = popen("wg show all peers 2>/dev/null | wc -l", "r");
    if (!fp) return 0;
    char b[32]; uint32_t peers = 0;
    if (fgets(b, sizeof(b), fp)) peers = static_cast<uint32_t>(atoi(b));
    pclose(fp);
    return peers;
}
