#pragma once

// The Unix-domain-socket RPC listener: $XDG_RUNTIME_DIR/velox/velox.sock, mode 0600,
// SO_PEERCRED same-UID check (docs/01 §2, AGENT-DAEMON.md build step 1). NDJSON framing.
// Non-blocking throughout; every fd runs through the shared EventLoop so one slow client
// never stalls another.
//
// session.hello and session.subscribe are handled here because they are connection- and
// transport-stateful (protocol-major check, sessionId, per-connection subscription set).
// Every other method is routed through the generated velox::proto::dispatch(), which does
// the envelope, the -32003 privileged-transport refusal and the typed param parse.

#include <cstdint>
#include <memory>
#include <string>
#include <optional>
#include <system_error>
#include <unordered_map>
#include <vector>

#include <nlohmann/json_fwd.hpp>

#include "rpc/event_hub.hpp"
#include "rpc/ndjson.hpp"
#include "velox_proto.hpp"

namespace velox::daemon::rpc {

class EventLoop;

class UdsServer {
public:
    UdsServer(EventLoop& loop, velox::proto::Dispatcher& dispatcher, EventHub& hub,
             std::string socket_path);
    ~UdsServer();

    UdsServer(const UdsServer&) = delete;
    UdsServer& operator=(const UdsServer&) = delete;

    // Create the socket, bind, chmod 0600, listen, and register with the loop. A stale
    // socket file left by a crashed daemon is removed first. Returns a non-ok error_code
    // (and changes nothing) on any failure.
    std::error_code start();

    const std::string& socket_path() const noexcept { return path_; }
    std::size_t connection_count() const noexcept { return conns_.size(); }

private:
    struct Conn {
        int fd;
        FrameReader reader;
        std::string outbuf;
        std::size_t out_off = 0;   // bytes of outbuf already written
        bool close_after_flush = false;
        bool hello_ok = false;
        std::string session_id;
        std::optional<EventHub::SubId> sub_id;
    };

    void on_listener_readable();
    void on_conn_event(int fd, unsigned events);
    void handle_line(Conn& c, const std::string& line);

    // Returns true and fills `reply` if `method` is one this layer answers directly
    // (session.hello / session.subscribe). Returns false to let dispatch() handle it.
    bool handle_session_method(Conn& c, const std::string& method, const nlohmann::json& request,
                               nlohmann::json& reply);

    void queue_reply(Conn& c, const nlohmann::json& reply);
    void flush(Conn& c);
    void close_conn(int fd);

    EventLoop& loop_;
    velox::proto::Dispatcher& dispatcher_;
    EventHub& hub_;
    std::string path_;
    int listen_fd_ = -1;
    bool bound_ = false;   // path_ is ours to unlink on destruction
    std::unordered_map<int, std::unique_ptr<Conn>> conns_;
};

}  // namespace velox::daemon::rpc