QDAirPortTestSystemBackend/src/network/ConfigHttpServer.cpp

#include "ConfigHttpServer.h"

#include "core/System.h"
#include "utils/Logger.h"
#include "vehicle/ControllableVehicles.h"

#include <boost/asio/dispatch.hpp>
#include <boost/asio/strand.hpp>
#include <nlohmann/json.hpp>
#include <chrono>
#include <unordered_map>

namespace network {

namespace beast = boost::beast;
namespace http = beast::http;
namespace net = boost::asio;
using tcp = boost::asio::ip::tcp;
using json = nlohmann::json;

namespace {

http::response<http::string_body> json_response(const http::request<http::string_body>& req,
    http::status status,
    const json& body) {
    http::response<http::string_body> res{status, req.version()};
    res.set(http::field::server, BOOST_BEAST_VERSION_STRING);
    res.set(http::field::content_type, "application/json");
    res.keep_alive(req.keep_alive());
    res.body() = body.dump();
    res.prepare_payload();
    return res;
}

http::response<http::string_body> method_not_allowed(const http::request<http::string_body>& req,
    const char* allow) {
    http::response<http::string_body> res{http::status::method_not_allowed, req.version()};
    res.set(http::field::server, BOOST_BEAST_VERSION_STRING);
    res.set(http::field::content_type, "application/json");
    res.set(http::field::allow, allow);
    res.keep_alive(req.keep_alive());
    res.body() = json{{"status", "error"}, {"message", "Method Not Allowed"}}.dump();
    res.prepare_payload();
    return res;
}

} // namespace

class ConfigSession : public std::enable_shared_from_this<ConfigSession> {
    tcp::socket socket_;
    beast::flat_buffer buffer_;
    http::request<http::string_body> req_;
    net::strand<net::io_context::executor_type> strand_;
    System& system_;

public:
    explicit ConfigSession(tcp::socket&& socket, net::io_context& ioc, System& system_ref)
        : socket_(std::move(socket)), strand_(ioc.get_executor()), system_(system_ref) {}

    void run() {
        net::dispatch(strand_, [self = shared_from_this()]() { self->do_read(); });
    }

private:
    void do_read() {
        req_ = {};
        http::async_read(socket_, buffer_, req_,
            net::bind_executor(strand_,
                [self = shared_from_this()](beast::error_code ec, std::size_t bytes) {
                    boost::ignore_unused(bytes);
                    self->on_read(ec);
                }));
    }

    void on_read(beast::error_code ec) {
        if (ec == http::error::end_of_stream) {
            return do_close();
        }
        if (ec) {
            Logger::error("ConfigSession read error: ", ec.message());
            return;
        }
        handle_request();
    }

    void handle_request() {
        if (req_.method() != http::verb::post) {
            return send_response(method_not_allowed(req_, "POST"));
        }

        // Endpoint A: 前端车辆注册表（增量更新）
        // POST /api/VehicleRegistry
        // Body: [{"vehicleID":"A001","vehicleType":"WUREN"}, ...]
        if (req_.target() == "/api/VehicleRegistry") {
            try {
                json body = json::parse(req_.body());
                if (!body.is_array()) {
                    return send_response(json_response(req_, http::status::bad_request,
                        json{{"status", "error"}, {"message", "Body must be an array"}}));
                }

                std::vector<VehicleRegistryEntry> entries;
                entries.reserve(body.size());

                std::unordered_map<std::string, int> type_counts;
                std::vector<std::string> errors;

                auto is_allowed_type = [](const std::string& t) {
                    return t == "WUREN" || t == "TEQIN" || t == "HANGKONG" || t == "PUTONG" || t == "JIUYUAN";
                };

                for (size_t i = 0; i < body.size(); ++i) {
                    const auto& item = body.at(i);
                    if (!item.is_object()) {
                        errors.push_back("index " + std::to_string(i) + ": item must be an object");
                        continue;
                    }
                    if (!item.contains("vehicleID") || !item.contains("vehicleType")) {
                        errors.push_back("index " + std::to_string(i) + ": missing vehicleID/vehicleType");
                        continue;
                    }
                    if (!item.at("vehicleID").is_string() || !item.at("vehicleType").is_string()) {
                        errors.push_back("index " + std::to_string(i) + ": vehicleID/vehicleType must be string");
                        continue;
                    }
                    std::string vehicleID = item.at("vehicleID").get<std::string>();
                    std::string vehicleType = item.at("vehicleType").get<std::string>();
                    if (vehicleID.empty()) {
                        errors.push_back("index " + std::to_string(i) + ": vehicleID must be non-empty");
                        continue;
                    }
                    if (!is_allowed_type(vehicleType)) {
                        errors.push_back("index " + std::to_string(i) + ": invalid vehicleType=" + vehicleType);
                        continue;
                    }

                    entries.push_back(VehicleRegistryEntry{vehicleID, vehicleType});
                    type_counts[vehicleType] += 1;
                }

                if (!errors.empty()) {
                    return send_response(json_response(req_, http::status::bad_request,
                        json{{"status", "error"}, {"message", "Invalid request"}, {"errors", errors}}));
                }

                auto& registry = ControllableVehicles::getInstance();
                registry.updateRegistry(entries);

                auto controllables = registry.getControllableVehicleIdsSnapshot();
                std::vector<std::string> controllable_ids;
                controllable_ids.reserve(controllables.size());
                for (const auto& id : controllables) {
                    controllable_ids.push_back(id);
                }

                auto now_ms = std::chrono::duration_cast<std::chrono::milliseconds>(
                    std::chrono::system_clock::now().time_since_epoch()).count();

                return send_response(json_response(req_, http::status::ok,
                    json{{"status", "success"},
                         {"updatedAt", now_ms},
                         {"updated", static_cast<int>(entries.size())},
                         {"controllableCount", static_cast<int>(controllable_ids.size())},
                         {"typesCount", type_counts},
                         {"controllableVehicleIDs", controllable_ids}}));
            } catch (const json::parse_error& e) {
                return send_response(json_response(req_, http::status::bad_request,
                    json{{"status", "error"}, {"message", "Invalid JSON"}, {"detail", e.what()}}));
            } catch (const std::exception& e) {
                return send_response(json_response(req_, http::status::internal_server_error,
                    json{{"status", "error"}, {"message", "Internal error"}, {"detail", e.what()}}));
            }
        }

        // Endpoint B: 运行中动态修改配置
        // POST /config/runway/warning_zone_radius/aircraft
        // Body: {"value": 300}
        if (req_.target() != "/config/runway/warning_zone_radius/aircraft") {
            return send_response(json_response(req_, http::status::not_found,
                json{{"status", "error"}, {"message", "Not Found"}}));
        }

        try {
            json body = json::parse(req_.body());
            if (!body.contains("value")) {
                return send_response(json_response(req_, http::status::bad_request,
                    json{{"status", "error"}, {"message", "Missing field: value"}}));
            }

            double value = body.at("value").get<double>();
            double oldValue = 0.0;
            if (!system_.setRunwayWarningZoneAircraftRadius(value, &oldValue)) {
                return send_response(json_response(req_, http::status::internal_server_error,
                    json{{"status", "error"}, {"message", "Failed to update config"}}));
            }

            Logger::info("Updated runway warning_zone_radius.aircraft: ", oldValue, " -> ", value);

            return send_response(json_response(req_, http::status::ok,
                json{{"status", "success"},
                     {"area", "runway"},
                     {"field", "warning_zone_radius.aircraft"},
                     {"old", oldValue},
                     {"new", value}}));
        } catch (const json::parse_error& e) {
            return send_response(json_response(req_, http::status::bad_request,
                json{{"status", "error"}, {"message", "Invalid JSON"}, {"detail", e.what()}}));
        } catch (const std::exception& e) {
            return send_response(json_response(req_, http::status::internal_server_error,
                json{{"status", "error"}, {"message", "Internal error"}, {"detail", e.what()}}));
        }
    }

    void send_response(http::response<http::string_body>&& res) {
        auto sp = std::make_shared<http::response<http::string_body>>(std::move(res));
        http::async_write(socket_, *sp,
            net::bind_executor(strand_,
                [self = shared_from_this(), sp](beast::error_code ec, std::size_t bytes) {
                    boost::ignore_unused(bytes);
                    self->on_write(sp->need_eof(), ec);
                }));
    }

    void on_write(bool close, beast::error_code ec) {
        if (ec) {
            Logger::error("ConfigSession write error: ", ec.message());
            return;
        }
        if (close) {
            return do_close();
        }
        do_read();
    }

    void do_close() {
        beast::error_code ec;
        socket_.shutdown(tcp::socket::shutdown_send, ec);
    }
};

class ConfigListener : public std::enable_shared_from_this<ConfigListener> {
    net::io_context& ioc_;
    tcp::acceptor acceptor_;
    net::strand<net::io_context::executor_type> strand_;
    int max_connections_;
    System& system_;

public:
    ConfigListener(net::io_context& ioc, tcp::endpoint endpoint, int max_connections, System& system_ref)
        : ioc_(ioc)
        , acceptor_(ioc)
        , strand_(ioc.get_executor())
        , max_connections_(max_connections)
        , system_(system_ref) {
        beast::error_code ec;
        acceptor_.open(endpoint.protocol(), ec);
        if (ec) throw beast::system_error{ec};

        acceptor_.set_option(net::socket_base::reuse_address(true), ec);
        if (ec) throw beast::system_error{ec};

        acceptor_.bind(endpoint, ec);
        if (ec) throw beast::system_error{ec};

        acceptor_.listen(net::socket_base::max_listen_connections, ec);
        if (ec) throw beast::system_error{ec};
    }

    void run() { do_accept(); }

    void stop() {
        beast::error_code ec;
        acceptor_.close(ec);
    }

private:
    void do_accept() {
        acceptor_.async_accept(
            net::bind_executor(strand_,
                [self = shared_from_this()](beast::error_code ec, tcp::socket socket) {
                    self->on_accept(ec, std::move(socket));
                }));
    }

    void on_accept(beast::error_code ec, tcp::socket socket) {
        if (ec) {
            if (ec != net::error::operation_aborted) {
                Logger::error("ConfigListener accept error: ", ec.message());
            }
            if (acceptor_.is_open() && ec != net::error::operation_aborted) {
                do_accept();
            }
            return;
        }

        std::make_shared<ConfigSession>(std::move(socket), ioc_, system_)->run();
        do_accept();
    }
};

ConfigHttpServer::ConfigHttpServer(uint16_t port, int max_connections, System& system_ref)
    : port_(port), max_connections_(max_connections), ioc_(1), system_(system_ref) {}

ConfigHttpServer::~ConfigHttpServer() {
    if (running_.load()) {
        stop();
    }
}

void ConfigHttpServer::start(int num_threads) {
    if (running_.exchange(true)) {
        return;
    }

    if (num_threads < 1) num_threads = 1;

    auto const address = net::ip::make_address("0.0.0.0");
    try {
        listener_ = std::make_shared<ConfigListener>(
            ioc_, tcp::endpoint{address, port_}, max_connections_, system_);
        listener_->run();

        Logger::info("ConfigHttpServer listening on ", address.to_string(), ":", port_);

        threads_.reserve(num_threads);
        for (int i = 0; i < num_threads; ++i) {
            threads_.emplace_back([this] { run_ioc(); });
        }
    } catch (const std::exception& e) {
        running_ = false;
        Logger::error("Failed to start ConfigHttpServer: ", e.what());
        throw;
    }
}

void ConfigHttpServer::stop() {
    if (!running_.exchange(false)) {
        return;
    }

    Logger::info("Stopping ConfigHttpServer...");
    net::post(ioc_, [this]() {
        if (listener_) {
            listener_->stop();
            listener_.reset();
        }
    });
    ioc_.stop();
    for (auto& t : threads_) {
        if (t.joinable()) t.join();
    }
    threads_.clear();
    Logger::info("ConfigHttpServer stopped.");
}

void ConfigHttpServer::run_ioc() {
    try {
        ioc_.run();
    } catch (const std::exception& e) {
        Logger::error("Exception in ConfigHttpServer I/O thread: ", e.what());
    }
}

} // namespace network