safesight-edge/plugins/alarm/actions/http_action.cpp

#include "http_action.h"

#include <algorithm>
#include <chrono>
#include <cctype>
#include <cstdlib>
#include <ctime>
#include <sstream>
#include <string>
#include <string_view>

#include "utils/logger.h"

#if __has_include(<curl/curl.h>)
#include <curl/curl.h>
#define HAS_CURL 1
#else
#define HAS_CURL 0
#endif

namespace rk3588 {

namespace {

std::string ToLower(std::string s) {
    for (char& c : s) c = static_cast<char>(std::tolower(static_cast<unsigned char>(c)));
    return s;
}

std::string JsonEscape(std::string_view s) {
    std::string out;
    out.reserve(s.size() + 8);
    for (char c : s) {
        switch (c) {
            case '\\': out += "\\\\"; break;
            case '"': out += "\\\""; break;
            case '\n': out += "\\n"; break;
            case '\r': out += "\\r"; break;
            case '\t': out += "\\t"; break;
            default:
                if (static_cast<unsigned char>(c) < 0x20) out += "?";
                else out.push_back(c);
        }
    }
    return out;
}

#if HAS_CURL
void EnsureCurlGlobalInitOnce() {
    static std::once_flag once;
    std::call_once(once, []() {
        curl_global_init(CURL_GLOBAL_DEFAULT);
        std::atexit([]() { curl_global_cleanup(); });
    });
}
#endif

}  // namespace

HttpAction::~HttpAction() {
    Drain();
    Stop();
}

bool HttpAction::Init(const SimpleJson& config) {
    url_ = config.ValueOr<std::string>("url", "");
    timeout_ms_ = config.ValueOr<int>("timeout_ms", 3000);
    include_media_url_ = config.ValueOr<bool>("include_media_url", true);
    method_ = config.ValueOr<std::string>("method", "POST");

    max_queue_size_ = static_cast<size_t>(std::max(0, config.ValueOr<int>("max_queue_size", static_cast<int>(max_queue_size_))));
    const std::string policy = config.ValueOr<std::string>("queue_policy", "drop_oldest");
    if (policy == "drop_newest") queue_policy_ = QueuePolicy::DropNewest;
    else queue_policy_ = QueuePolicy::DropOldest;

    if (url_.empty()) {
        LogError("[HttpAction] url is required");
        return false;
    }

#if HAS_CURL
    EnsureCurlGlobalInitOnce();
#endif

    running_.store(true);
    worker_ = std::thread(&HttpAction::WorkerLoop, this);
    LogInfo("[HttpAction] initialized, url=" + url_ + ", method=" + method_);
    return true;
}

void HttpAction::Execute(AlarmEvent& event, std::shared_ptr<Frame> /*frame*/) {
    // Build JSON body
    std::ostringstream oss;
    oss << "{";
    oss << "\"node_id\":\"" << JsonEscape(event.node_id) << "\",";
    oss << "\"rule_name\":\"" << JsonEscape(event.rule_name) << "\",";
    oss << "\"timestamp\":" << event.timestamp_ms << ",";
    oss << "\"frame_id\":" << event.frame_id << ",";

    if (include_media_url_) {
        if (!event.snapshot_url.empty()) {
            oss << "\"snapshot_url\":\"" << JsonEscape(event.snapshot_url) << "\",";
        }
        if (!event.clip_url.empty()) {
            oss << "\"clip_url\":\"" << JsonEscape(event.clip_url) << "\",";
        }
    }

    oss << "\"detections\":[";
    for (size_t i = 0; i < event.detections.size(); ++i) {
        const auto& det = event.detections[i];
        if (i > 0) oss << ",";
        oss << "{\"cls_id\":" << det.cls_id
            << ",\"score\":" << det.score
            << ",\"bbox\":{\"x\":" << det.bbox.x
            << ",\"y\":" << det.bbox.y
            << ",\"w\":" << det.bbox.w
            << ",\"h\":" << det.bbox.h << "}"
            << ",\"track_id\":" << det.track_id << "}";
    }
    oss << "]}";

    std::string json_body = oss.str();

    {
        std::lock_guard<std::mutex> lock(queue_mutex_);
        if (max_queue_size_ > 0 && request_queue_.size() >= max_queue_size_) {
            ++dropped_total_;
            if (queue_policy_ == QueuePolicy::DropOldest) {
                request_queue_.pop();
                request_queue_.push(json_body);
            } else {
                // DropNewest: drop this request
            }
        } else {
            request_queue_.push(json_body);
        }
    }
    queue_cv_.notify_one();
}

void HttpAction::Drain() {
    if (!running_.load()) return;
    // Wait for queue to be fully drained (including in-flight request).
    std::unique_lock<std::mutex> lock(queue_mutex_);
    queue_cv_.wait_for(lock, std::chrono::seconds(5),
                       [this] { return request_queue_.empty() && in_flight_ == 0; });
}

void HttpAction::Stop() {
    bool was_running = running_.exchange(false);
    (void)was_running;
    queue_cv_.notify_all();
    if (worker_.joinable()) worker_.join();
}

void HttpAction::WorkerLoop() {
    while (running_.load()) {
        std::string body;
        {
            std::unique_lock<std::mutex> lock(queue_mutex_);
            queue_cv_.wait_for(lock, std::chrono::milliseconds(100),
                               [this] { return !request_queue_.empty() || !running_.load(); });
            if (!running_.load() && request_queue_.empty()) break;
            if (request_queue_.empty()) continue;
            body = std::move(request_queue_.front());
            request_queue_.pop();
            ++in_flight_;
        }

        if (!body.empty()) {
            SendRequest(body);
        }

        {
            std::lock_guard<std::mutex> lock(queue_mutex_);
            if (in_flight_ > 0) --in_flight_;
        }
        queue_cv_.notify_all();
    }
}

bool HttpAction::SendRequest(const std::string& json_body) {
#if HAS_CURL
    CURL* curl = curl_easy_init();
    if (!curl) {
        LogError("[HttpAction] curl_easy_init failed");
        return false;
    }

    struct curl_slist* headers = nullptr;
    headers = curl_slist_append(headers, "Content-Type: application/json");

    curl_easy_setopt(curl, CURLOPT_URL, url_.c_str());
    curl_easy_setopt(curl, CURLOPT_HTTPHEADER, headers);

    const std::string method_lc = ToLower(method_);
    if (method_lc == "post") {
        curl_easy_setopt(curl, CURLOPT_POST, 1L);
        curl_easy_setopt(curl, CURLOPT_POSTFIELDS, json_body.c_str());
    } else if (method_lc == "put" || method_lc == "patch") {
        curl_easy_setopt(curl, CURLOPT_CUSTOMREQUEST, method_.c_str());
        curl_easy_setopt(curl, CURLOPT_POSTFIELDS, json_body.c_str());
    } else {
        LogWarn("[HttpAction] unsupported method='" + method_ + "', falling back to POST");
        curl_easy_setopt(curl, CURLOPT_POST, 1L);
        curl_easy_setopt(curl, CURLOPT_POSTFIELDS, json_body.c_str());
    }

    curl_easy_setopt(curl, CURLOPT_TIMEOUT_MS, timeout_ms_);
    curl_easy_setopt(curl, CURLOPT_CONNECTTIMEOUT_MS, timeout_ms_ / 2);

    CURLcode res = curl_easy_perform(curl);
    if (res != CURLE_OK) {
        LogWarn(std::string("[HttpAction] request failed: ") + curl_easy_strerror(res));
    }

    curl_slist_free_all(headers);
    curl_easy_cleanup(curl);
    return res == CURLE_OK;
#else
    // Stub implementation for Windows without curl
    LogInfo("[HttpAction] (stub) method=" + method_ + " url=" + url_ + " body=" + json_body);
    return true;
#endif
}

}  // namespace rk3588