#include "rpc/event_hub.hpp"

#include <algorithm>

#include <nlohmann/json.hpp>

namespace velox::daemon::rpc {

namespace proto = velox::proto;

EventHub::SubId EventHub::subscribe(Sink sink) {
    std::lock_guard<std::mutex> lk(mu_);
    const SubId id = next_++;
    subs_.emplace(id, Sub{std::move(sink), {}, std::nullopt});
    return id;
}

void EventHub::set_filter(SubId id, std::vector<proto::Event> events,
                          std::optional<std::vector<std::string>> task_ids) {
    std::lock_guard<std::mutex> lk(mu_);
    if (auto it = subs_.find(id); it != subs_.end()) {
        it->second.events = std::move(events);
        it->second.task_ids = std::move(task_ids);
    }
}

void EventHub::unsubscribe(SubId id) {
    std::lock_guard<std::mutex> lk(mu_);
    subs_.erase(id);
}

void EventHub::publish(proto::Event kind, const nlohmann::json& notification,
                       std::string_view task_id) {
    // Copy the sinks to call out to while holding the lock only long enough to build the
    // list — a sink runs arbitrary connection code (framing + a write syscall) and must
    // not run with mu_ held.
    std::vector<Sink> targets;
    {
        std::lock_guard<std::mutex> lk(mu_);
        targets.reserve(subs_.size());
        for (const auto& [id, sub] : subs_) {
            (void)id;
            if (std::find(sub.events.begin(), sub.events.end(), kind) == sub.events.end())
                continue;
            if (!task_id.empty() && sub.task_ids &&
                std::find(sub.task_ids->begin(), sub.task_ids->end(), task_id) ==
                    sub.task_ids->end())
                continue;
            targets.push_back(sub.sink);
        }
    }
    for (const auto& sink : targets) sink(notification);
}

}  // namespace velox::daemon::rpc