#include "rpc/event_loop.hpp"

#include <poll.h>
#include <sys/eventfd.h>
#include <unistd.h>

#include <cerrno>
#include <cstdint>
#include <stdexcept>
#include <vector>

namespace velox::daemon::rpc {

EventLoop::EventLoop() {
    wake_fd_ = ::eventfd(0, EFD_NONBLOCK | EFD_CLOEXEC);
    if (wake_fd_ < 0) throw std::runtime_error("eventfd() failed");
    fds_.emplace(wake_fd_, Entry{kRead, [this](int, unsigned) { drain_wakeup(); }});
}

EventLoop::~EventLoop() {
    if (wake_fd_ >= 0) ::close(wake_fd_);
}

void EventLoop::add_fd(int fd, unsigned interest, Callback cb) {
    fds_[fd] = Entry{interest, std::move(cb)};
}

void EventLoop::mod_fd(int fd, unsigned interest) {
    if (auto it = fds_.find(fd); it != fds_.end()) it->second.interest = interest;
}

void EventLoop::del_fd(int fd) {
    if (fd == wake_fd_) return;  // internal, never removed
    fds_.erase(fd);
}

void EventLoop::wake() noexcept {
    const std::uint64_t one = 1;
    // Best-effort: an EAGAIN here means a wakeup is already pending, which is fine.
    [[maybe_unused]] ssize_t n = ::write(wake_fd_, &one, sizeof(one));
}

void EventLoop::stop() noexcept {
    stop_requested_ = true;
    wake();
}

void EventLoop::drain_wakeup() noexcept {
    std::uint64_t sink = 0;
    while (::read(wake_fd_, &sink, sizeof(sink)) > 0) {
    }
}

void EventLoop::run() {
    if (running_) throw std::logic_error("EventLoop::run() is not re-entrant");
    running_ = true;
    stop_requested_ = false;

    std::vector<pollfd> pfds;
    std::vector<int> fired;

    while (!stop_requested_) {
        pfds.clear();
        pfds.reserve(fds_.size());
        for (const auto& [fd, e] : fds_) {
            short ev = 0;
            if (e.interest & kRead) ev |= POLLIN;
            if (e.interest & kWrite) ev |= POLLOUT;
            if (ev == 0 && fd != wake_fd_) continue;
            pollfd p{};
            p.fd = fd;
            p.events = ev;
            pfds.push_back(p);
        }

        const int rc = ::poll(pfds.data(), pfds.size(), -1);
        if (rc < 0) {
            if (errno == EINTR) continue;
            throw std::runtime_error("poll() failed");
        }
        if (rc == 0) continue;

        // Snapshot the fds that fired before invoking any callback: a callback may erase
        // entries from fds_, which would invalidate iteration over pfds' referents.
        fired.clear();
        for (const auto& p : pfds) {
            if (p.revents != 0) fired.push_back(p.fd);
        }

        for (const int fd : fired) {
            const auto it = fds_.find(fd);
            if (it == fds_.end()) continue;  // removed by an earlier callback this pass

            // Recompute revents for this fd from the snapshot.
            unsigned events = 0;
            for (const auto& p : pfds) {
                if (p.fd != fd) continue;
                if (p.revents & (POLLIN | POLLHUP | POLLERR)) events |= kRead;
                if (p.revents & POLLOUT) events |= kWrite;
                break;
            }
            if (events != 0) it->second.cb(fd, events);
        }
    }

    running_ = false;
}

}  // namespace velox::daemon::rpc