WebSocketAPI.cpp

#if defined(WITH_POSTGRESQL) && defined(WITH_SSL)

#include "WebSocketAPI.hpp"
#include "apostol/application.hpp"

#include "apostol/base64.hpp"
#include "apostol/http_utils.hpp"
#include "apostol/pg_utils.hpp"

#include <algorithm>
#include <chrono>
#include <fmt/format.h>
#include <nlohmann/json.hpp>
#include <stdexcept>
#include <string>

namespace apostol
{

// ─── JSON-RPC message types ─────────────────────────────────────────────────

namespace
{

enum class MsgType : int
{
    open        = 0,
    close       = 1,
    call        = 2,
    call_result = 3,
    call_error  = 4,
};

/// Generate a unique ID (hex microseconds) for server-initiated messages.
std::string make_unique_id()
{
    auto us = std::chrono::duration_cast<std::chrono::microseconds>(
        std::chrono::system_clock::now().time_since_epoch()).count();
    return fmt::format("{:x}", static_cast<uint64_t>(us));
}

/// Deepest nesting accepted in JSON from outside. nlohmann parses without
/// recursion but copies and serialises recursively, so a document nested
/// deep enough overflows the worker's stack — a signal, not an exception:
/// no catch sees it. Our messages nest a few levels; 128 is far above them.
constexpr int kMaxJsonDepth = 128;

/// Longest token handed to verify_jwt. jwt::decode copies the claims
/// recursively too; a bound on the length bounds their depth.
constexpr std::size_t kMaxTokenLength = 8192;

nlohmann::json parse_bounded(const std::string& text)
{
    return nlohmann::json::parse(text,
        [](int depth, nlohmann::json::parse_event_t, nlohmann::json&) {
            if (depth > kMaxJsonDepth)
                throw std::out_of_range("JSON nested deeper than allowed");
            return true;
        });
}

/// dump() that does not throw on a string which is not UTF-8.
std::string dump_safe(const nlohmann::json& j)
{
    return j.dump(-1, ' ', false, nlohmann::json::error_handler_t::replace);
}

/// Build JSON-RPC call message: {"t":2, "u":"<uid>", "a":"<action>", "p":<payload>}
std::string build_call_msg(std::string_view action, const std::string& payload)
{
    nlohmann::json msg;
    msg["t"] = static_cast<int>(MsgType::call);
    msg["u"] = make_unique_id();
    msg["a"] = action;

    try {
        msg["p"] = parse_bounded(payload);
    } catch (...) {
        msg["p"] = payload;
    }

    return dump_safe(msg);
}

/// Build JSON-RPC call_result message: {"t":3, "u":"<uid>", "p":<payload>}
std::string build_result_msg(std::string_view unique_id,
                             const std::string& payload)
{
    nlohmann::json msg;
    msg["t"] = static_cast<int>(MsgType::call_result);
    msg["u"] = unique_id;

    try {
        msg["p"] = parse_bounded(payload);
    } catch (...) {
        msg["p"] = payload;
    }

    return dump_safe(msg);
}

/// Build JSON-RPC call_error message: {"t":4, "u":"<uid>", "c":<code>, "m":"<msg>"}
std::string build_error_msg(std::string_view unique_id,
                            int code, std::string_view message)
{
    nlohmann::json msg;
    msg["t"] = static_cast<int>(MsgType::call_error);
    msg["u"] = unique_id;
    msg["c"] = code;
    msg["m"] = message;
    return dump_safe(msg);
}

/// Default receive window for signed_fetch (milliseconds).
constexpr int kReceiveWindowMs = 60000;

/// Heartbeat / session-cleanup interval.
constexpr auto kHeartbeatInterval = std::chrono::seconds(60);

/// How long a peer has to answer the server's Close before the session is dropped.
constexpr auto kCloseGrace = std::chrono::seconds(10);

bool close_sent(const auto& session)
{
    return session.close_sent != std::chrono::steady_clock::time_point{};
}

} // anonymous namespace

// ─── Construction ───────────────────────────────────────────────────────────

WebSocketAPI::WebSocketAPI(Application& app)
    : pool_(app.db_pool())
    , loop_(app.worker_loop())
    , providers_(app.providers())
    , enabled_(true)
    , log_(app.logger())
    , keepalive_(WsKeepAlive::from_config(app.module_config("WebSocketAPI"), "module.WebSocketAPI"))
{
    add_allowed_header("Authorization");
    add_allowed_header("Session");
    add_allowed_header("Secret");

    load_allowed_origins(providers_);
}

// ─── check_location ─────────────────────────────────────────────────────────

bool WebSocketAPI::check_location(const HttpRequest& req) const
{
    // HTTP endpoints: /ws/*
    return req.path.substr(0, 4) == "/ws/" || req.path == "/ws";
}

// ─── init_methods ───────────────────────────────────────────────────────────

void WebSocketAPI::init_methods()
{
    add_method("GET",  [this](auto& req, auto& resp) { do_get(req, resp); });
    add_method("POST", [this](auto& req, auto& resp) { do_post(req, resp); });
}

// ─── heartbeat ──────────────────────────────────────────────────────────────

void WebSocketAPI::heartbeat(std::chrono::system_clock::time_point now)
{
    keep_alive();

    if (now < next_check_)
        return;

    next_check_ = now + kHeartbeatInterval;

    // Initialize LISTEN channels (retry until successful)
    if (!listen_initialized_) {
        init_listen();
    }

    // Cleanup dead sessions (connections dropped without proper disconnect)
    check_sessions();

    // Process any queued observer tasks
    unload_queue();
}

// ─── parse_session_path ─────────────────────────────────────────────────────

std::pair<std::string, std::string>
WebSocketAPI::parse_session_path(std::string_view path)
{
    // Expected: /session/<code>[/<identity>]
    // or: /ws/<code>[/<identity>] (for POST push)

    // Find the prefix: skip "/session/" or "/ws/"
    std::string_view rest;
    if (path.substr(0, 9) == "/session/")
        rest = path.substr(9);
    else if (path.substr(0, 4) == "/ws/")
        rest = path.substr(4);
    else
        return {};

    if (rest.empty())
        return {};

    auto slash = rest.find('/');
    if (slash == std::string_view::npos)
        return {std::string(rest), "main"};

    auto code = rest.substr(0, slash);
    auto identity = rest.substr(slash + 1);

    return {std::string(code),
            identity.empty() ? std::string("main") : std::string(identity)};
}

// ─── normalize_action ───────────────────────────────────────────────────────

std::string WebSocketAPI::normalize_action(std::string_view action)
{
    if (action.empty())
        return "/api/v1";

    if (action.substr(0, 8) == "/api/v1/" || action == "/api/v1")
        return std::string(action);

    return fmt::format("/api/v1{}", action);
}

// ─── Session management ─────────────────────────────────────────────────────

std::shared_ptr<WebSocketAPI::WsSession>
WebSocketAPI::add_session(WsConnection ws,
                          std::string session,
                          std::string identity)
{
    auto s = std::make_shared<WsSession>();
    s->session  = std::move(session);
    s->identity = std::move(identity);
    s->ws       = std::make_shared<WsConnection>(std::move(ws));

    int fd = s->ws->fd();
    sessions_by_fd_[fd] = s;
    sessions_by_code_.emplace(s->session, s);

    return s;
}

void WebSocketAPI::remove_session(int fd)
{
    auto it = sessions_by_fd_.find(fd);
    if (it == sessions_by_fd_.end())
        return;

    auto session = it->second;

    // Detach fd from epoll. EPOLLIN is level-triggered, so a closed socket
    // would otherwise wake the event loop on every iteration and spin at
    // 100% CPU. remove_io is idempotent (EPOLL_CTL_DEL on unknown fd is
    // silently ignored).
    loop_.remove_io(fd);

    // Erase from BOTH indexes immediately so subsequent on_notify,
    // dispatch_observer, check_sessions etc. don't see this dead
    // session. In-flight PG callbacks keep shared_ptr<WsSession> alive;
    // they'll complete (on a dead socket send_call returns EPIPE silently
    // via WsConnection::closed_), and the session (with its WsConnection →
    // fd) is destroyed when the last shared_ptr drops.
    sessions_by_fd_.erase(it);

    // sessions_by_code_ is a multimap: one session-code may map to multiple
    // sessions (client opens several WS with different `identity` suffixes
    // on the same session-code — e.g. "driver-pwa", "main"). Collisions are
    // expected and bounded (typically <10 per client); linear scan of the
    // bucket is fine.
    auto range = sessions_by_code_.equal_range(session->session);
    for (auto it2 = range.first; it2 != range.second; ) {
        if (it2->second.get() == session.get())
            it2 = sessions_by_code_.erase(it2);
        else
            ++it2;
    }
}

void WebSocketAPI::rekey_session(WsSession& session, std::string code)
{
    if (code == session.session)
        return;

    // The entry is keyed by the old code: find it there, not under the new
    // one. Left in place, remove_session() would look under the new code,
    // miss it, and the entry (holding the connection) would outlive the
    // socket; /ws pushes to the old code would reach this session.
    std::shared_ptr<WsSession> self;
    auto range = sessions_by_code_.equal_range(session.session);
    for (auto it = range.first; it != range.second; ++it) {
        if (it->second.get() == &session) {
            self = it->second;
            sessions_by_code_.erase(it);
            break;
        }
    }

    session.session = std::move(code);

    // Not found means the socket is already gone (remove_session ran while
    // the query was in flight) — do not bring it back into the index.
    if (self)
        sessions_by_code_.emplace(session.session, std::move(self));
}

// ─── on_ws_upgrade ──────────────────────────────────────────────────────────

void WebSocketAPI::on_ws_upgrade(EventLoop& loop, WsConnection ws,
                                 const HttpRequest& req)
{
    auto [code, identity] = parse_session_path(req.path);
    if (code.empty()) {
        ws.send_close(1008, "Invalid session path");
        return;
    }

    auto session = add_session(std::move(ws), std::move(code),
                               std::move(identity));
    session->agent = get_user_agent(req);
    session->ip    = get_real_ip(req);

    // Check handshake-level authorization (headers)
    check_session_auth(req, *session);

    int fd = session->ws->fd();
    // Let WsConnection arm EPOLLOUT on its own when an outbound frame
    // EAGAIN's — server-push paths (observer notifications, handlers)
    // call send_text outside this lambda's scope, so they can't arm the
    // loop themselves.
    session->ws->bind_event_loop(loop);

    // Replace the HTTP handler with a WebSocket handler.
    // The fd is already registered in epoll by the HTTP accept loop;
    // remove it first to avoid EEXIST on add_io.
    loop.remove_io(fd);
    loop.add_io(fd, EPOLLIN, [this, session](uint32_t events) {
        // Peer reset/half-close. No useful IO can follow; tearing down the
        // session immediately avoids leaving it waiting for an EPOLLIN
        // that never comes (ET disables the fd after this event).
        if (events & (EPOLLERR | EPOLLHUP)) {
            remove_session(session->ws->fd());
            return;
        }
        if (events & EPOLLIN) {
            bool ok = session->ws->on_readable(
                [this, session](uint8_t opcode, const std::string& payload) {
                    on_ws_message(session, opcode, payload);
                },
                [this, session]() {
                    remove_session(session->ws->fd());
                }
            );
            if (!ok) {
                remove_session(session->ws->fd());
                return;
            }
        }
        if (events & EPOLLOUT) {
            session->ws->on_writable();
            if (session->ws->closed()) {
                remove_session(session->ws->fd());
                return;
            }
        }
        // Under APOSTOL_EPOLL_ET the fd was armed with EPOLLONESHOT and must
        // be rearmed to receive further events. Include EPOLLOUT only while
        // there are buffered bytes waiting to drain. Under LT (flag off)
        // rearm_io is a no-op.
        uint32_t mask = EPOLLIN;
        if (session->ws->has_pending_writes())
            mask |= EPOLLOUT;
        loop_.rearm_io(session->ws->fd(), mask);
    });
}

// ─── check_session_auth ─────────────────────────────────────────────────────

int WebSocketAPI::check_session_auth(const HttpRequest& req, WsSession& session)
{
    // Priority 1: Authorization header (Bearer or Basic)
    auto auth_header = req.header("Authorization");
    if (!auth_header.empty()) {
        session.auth = parse_authorization(auth_header);

        if (session.auth.schema == Authorization::Schema::bearer) {
            // A refused token must not stay on the session: handle_call
            // routes by auth.schema, and a later OPEN by secret would still
            // go out as this bearer and be refused.
            const auto refuse = [&session]() {
                session.auth = Authorization{};
                session.authorized = false;
                session.verified   = false;
                return -1;
            };
            if (session.auth.token.size() > kMaxTokenLength)
                return refuse();
            try {
                auto claims = verify_jwt(session.auth.token, providers_);
                // Verify sub matches session code
                if (!claims.sub.empty() && claims.sub != session.session)
                    return refuse();
                session.authorized = true;
                session.verified   = claims.sub == session.session;   // an empty sub fits any code
                return 1;
            } catch (const JwtExpiredError&) {
                return refuse();
            } catch (const JwtVerificationError&) {
                return refuse();
            } catch (const std::exception&) {
                // Malformed token (jwt::decode) — see handle_open.
                return refuse();
            }
        }

        if (session.auth.schema == Authorization::Schema::basic) {
            session.authorized = true;
            session.verified   = false;
            return 1;
        }
    }

    // Priority 2: Session + Secret headers
    auto sess_hdr   = req.header("Session");
    auto secret_hdr = req.header("Secret");

    if (!sess_hdr.empty() && !secret_hdr.empty()) {
        session.secret = std::move(secret_hdr);
        session.authorized = true;
        session.verified   = false;
        return 1;
    }

    // No auth — must OPEN later
    session.authorized = false;
    session.verified   = false;
    return 0;
}

// ─── on_ws_message ──────────────────────────────────────────────────────────

void WebSocketAPI::on_ws_message(std::shared_ptr<WsSession> session,
                                 uint8_t opcode, const std::string& payload)
{
    if (opcode != WS_OP_TEXT || close_sent(*session))
        return;

    nlohmann::json msg;
    try {
        msg = parse_bounded(payload);
    } catch (const std::exception&) {
        send_call_error(*session->ws, "", 400, "Invalid JSON");
        return;
    }

    // The frame comes from anyone who can reach /session/, before any OPEN.
    // This handler runs in the epoll callback of the socket, outside the
    // HTTP dispatcher's try: an exception that leaves it ends the worker
    // process, and every session on it. json::value() throws on a field of
    // another type or on a message that is not an object, so the envelope is
    // read by type, and whatever still throws below is answered, not
    // propagated. A catch does not cover a stack overflow: the depth is
    // bounded at parse (parse_bounded), and p is moved out, not copied.
    if (!msg.is_object()) {
        send_call_error(*session->ws, "", 400, "Invalid message: an object expected");
        return;
    }

    const auto field_str = [&msg](const char* key) -> std::string {
        auto it = msg.find(key);
        return (it != msg.end() && it->is_string()) ? it->get<std::string>() : std::string();
    };

    std::string unique_id = field_str("u");
    std::string action    = field_str("a");

    auto t = msg.find("t");
    if (t == msg.end() || !t->is_number_integer()) {
        send_call_error(*session->ws, unique_id, 400, "Unknown message type");
        return;
    }
    const auto type = t->get<long long>();

    nlohmann::json p = nlohmann::json::object();
    if (auto it = msg.find("p"); it != msg.end())
        p = std::move(*it);

    try {
        switch (type) {
            case static_cast<long long>(MsgType::open):
                handle_open(session, unique_id, p);
                break;
            case static_cast<long long>(MsgType::close):
                handle_close(session, unique_id);
                break;
            case static_cast<long long>(MsgType::call):
                handle_call(session, unique_id, action, p);
                break;
            default:
                send_call_error(*session->ws, unique_id, 400,
                                "Unknown message type");
                break;
        }
    } catch (const nlohmann::json::exception& e) {
        log_.warn("WebSocketAPI: message from {} refused: {}", session->ip, e.what());
        if (!close_sent(*session))
            send_call_error(*session->ws, unique_id, 400, "Invalid message.");
    } catch (const std::exception& e) {
        log_.error("WebSocketAPI: message from {} failed: {}", session->ip, e.what());
        if (!close_sent(*session))
            send_call_error(*session->ws, unique_id, 500, "Internal error.");
    }
}

// ─── handle_open ────────────────────────────────────────────────────────────

void WebSocketAPI::handle_open(std::shared_ptr<WsSession> session,
                               const std::string& unique_id,
                               const nlohmann::json& payload)
{
    const auto field_str = [&payload](const char* key) -> std::string {
        if (!payload.is_object())
            return {};
        auto it = payload.find(key);
        return (it != payload.end() && it->is_string()) ? it->get<std::string>() : std::string();
    };

    std::string secret = field_str("secret");
    std::string token  = field_str("token");

    if (!secret.empty()) {
        // Authenticate via session+secret
        session->auth       = Authorization{};   // OPEN by secret: not a bearer left from the handshake
        session->secret     = secret;
        session->authorized = false;
        session->verified   = false;
        session->open_token_own = false;

        // Build payload for unauthorized_fetch
        nlohmann::json auth_payload;
        auth_payload["session"] = session->session;
        auth_payload["secret"]  = secret;
        auth_payload["agent"]   = session->agent;
        auth_payload["host"]    = session->ip;

        unauthorized_fetch(session, unique_id,
                           "/api/v1/authenticate",
                           dump_safe(auth_payload), /*open=*/true);
    } else if (!token.empty()) {
        // Authenticate via JWT token
        if (token.size() > kMaxTokenLength) {
            send_call_error(*session->ws, unique_id, 401, "Malformed token.");
            return;
        }
        bool own = false;
        try {
            auto claims = verify_jwt(token, providers_);
            // Check that sub matches session
            if (!claims.sub.empty() && claims.sub != session->session) {
                send_call_error(*session->ws, unique_id, 401,
                                "Token subject does not match session.");
                return;
            }
            own = claims.sub == session->session;
        } catch (const JwtExpiredError&) {
            send_call_error(*session->ws, unique_id, 401, "Token expired.");
            return;
        } catch (const JwtVerificationError& e) {
            send_call_error(*session->ws, unique_id, 401, e.what());
            return;
        } catch (const std::exception&) {
            // Not a token at all: verify_jwt starts with jwt::decode outside
            // its own try, so wrong segments or broken base64 throw jwt-cpp's
            // exception, not ours (as in AppServer::check_auth).
            send_call_error(*session->ws, unique_id, 401, "Malformed token.");
            return;
        }

        session->auth.schema = Authorization::Schema::bearer;
        session->auth.token  = token;
        session->authorized  = false;
        session->verified    = false;
        session->open_token_own = own;

        nlohmann::json auth_payload;
        auth_payload["session"] = session->session;
        auth_payload["agent"]   = session->agent;
        auth_payload["host"]    = session->ip;

        unauthorized_fetch(session, unique_id,
                           "/api/v1/authorize",
                           dump_safe(auth_payload), /*open=*/true);
    } else {
        send_call_error(*session->ws, unique_id, 400,
                        "OPEN requires 'secret' or 'token'.");
    }
}

// ─── handle_close ───────────────────────────────────────────────────────────

void WebSocketAPI::handle_close(std::shared_ptr<WsSession> session,
                                const std::string& unique_id)
{
    if (!session->authorized) {
        send_call_error(*session->ws, unique_id, 401, "Unauthorized.");
        return;
    }

    // Sign out via CALL to /api/v1/sign/out
    handle_call(session, unique_id, "/api/v1/sign/out",
                nlohmann::json::object());
}

// ─── handle_call ────────────────────────────────────────────────────────────

void WebSocketAPI::handle_call(std::shared_ptr<WsSession> session,
                               const std::string& unique_id,
                               const std::string& action,
                               const nlohmann::json& payload)
{
    if (!session->authorized) {
        send_call_error(*session->ws, unique_id, 401, "Unauthorized.");
        return;
    }

    auto normalized = normalize_action(action);

    // Bind the WS connection identity to observer subscribe/unsubscribe so the
    // listener is stored under the same identity that daemon.observer filters
    // by; otherwise FilterListener's strict (publisher, session, identity)
    // match fails and events vanish for any WS URL with a custom identity.
    nlohmann::json effective = payload;
    if (normalized == "/api/v1/observer/subscribe" ||
        normalized == "/api/v1/observer/unsubscribe")
    {
        auto inject = [&](nlohmann::json& obj) {
            if (obj.is_object() && !obj.contains("identity"))
                obj["identity"] = session->identity;
        };
        if (effective.is_array()) {
            for (auto& item : effective) inject(item);
        } else {
            inject(effective);
        }
    }

    auto payload_str = dump_safe(effective);

    if (session->auth.schema == Authorization::Schema::bearer) {
        authorized_fetch(session, unique_id, normalized, payload_str);
    } else if (session->auth.schema == Authorization::Schema::basic) {
        authorized_fetch(session, unique_id, normalized, payload_str);
    } else if (!session->secret.empty()) {
        signed_fetch(session, unique_id, normalized, payload_str);
    } else {
        unauthorized_fetch(session, unique_id, normalized, payload_str);
    }
}

// ─── WebSocket response helpers ─────────────────────────────────────────────

void WebSocketAPI::send_call(WsConnection& ws, std::string_view action,
                             const std::string& payload)
{
    ws.send_text(build_call_msg(action, payload));
}

void WebSocketAPI::send_call_result(WsConnection& ws,
                                    std::string_view unique_id,
                                    const std::string& payload)
{
    ws.send_text(build_result_msg(unique_id, payload));
}

void WebSocketAPI::send_call_error(WsConnection& ws,
                                   std::string_view unique_id,
                                   int code, std::string_view message)
{
    ws.send_text(build_error_msg(unique_id, code, message));
}

// ─── Fetch dispatch ─────────────────────────────────────────────────────────

void WebSocketAPI::unauthorized_fetch(std::shared_ptr<WsSession> session,
                                      const std::string& unique_id,
                                      const std::string& action,
                                      const std::string& payload,
                                      bool open)
{
    auto action_q  = pq_quote_literal(action);
    auto payload_q = (payload.empty() || payload == "{}" || payload == "[]")
                         ? std::string("null")
                         : pq_quote_literal(payload);
    auto agent_q   = pq_quote_literal(session->agent);
    auto host_q    = pq_quote_literal(session->ip);

    auto sql = fmt::format(
        "SELECT * FROM daemon.unauthorized_fetch('POST', {}, {}::jsonb, {}, {})",
        action_q, payload_q, agent_q, host_q);

    // quiet: the credential here is not an argument but the CONTENT of the
    // payload — /authenticate carries the session code with its secret on every
    // (re)connect, /sign/in and /sign/up the user's password. PgPool logs
    // statement text, and a dedicated postgres.log keeps it at debug. Measured on
    // ocpp-css prod 28–30.09 (T713): 364 session code + secret pairs in clear
    // text. Mirrors AppServer's unauthorised branch, quiet for the same reason.
    pool_.execute(std::move(sql),
        [this, session, unique_id, action, open](std::vector<PgResult> results) {
            on_fetch_result(session, unique_id, action, std::move(results), open);
        },
        [this, session, unique_id](std::string_view error) {
            send_call_error(*session->ws, unique_id, 500,
                            std::string(error));
        },
        /*quiet=*/true);
}

void WebSocketAPI::authorized_fetch(std::shared_ptr<WsSession> session,
                                    const std::string& unique_id,
                                    const std::string& action,
                                    const std::string& payload)
{
    auto method_q  = pq_quote_literal("POST");
    auto action_q  = pq_quote_literal(action);
    auto payload_q = (payload.empty() || payload == "{}" || payload == "[]")
                         ? std::string("null")
                         : pq_quote_literal(payload);
    auto agent_q   = pq_quote_literal(session->agent);
    auto host_q    = pq_quote_literal(session->ip);

    std::string sql;

    if (session->auth.schema == Authorization::Schema::bearer) {
        sql = fmt::format(
            "SELECT * FROM daemon.fetch({}, {}, {}, {}::jsonb, {}, {})",
            pq_quote_literal(session->auth.token), method_q,
            action_q, payload_q, agent_q, host_q);
    } else if (session->auth.schema == Authorization::Schema::basic) {
        // Session+Secret headers or Basic auth
        if (!session->auth.username.empty() && !session->auth.password.empty()
            && session->auth.username != session->session) {
            // True Basic auth (username/password)
            sql = fmt::format(
                "SELECT * FROM daemon.authorized_fetch({}, {}, {}, {}, {}::jsonb, {}, {})",
                pq_quote_literal(session->auth.username),
                pq_quote_literal(session->auth.password),
                method_q, action_q, payload_q, agent_q, host_q);
        } else {
            // Session fetch
            sql = fmt::format(
                "SELECT * FROM daemon.session_fetch({}, {}, {}, {}, {}::jsonb, {}, {})",
                pq_quote_literal(session->session),
                pq_quote_literal(session->secret),
                method_q, action_q, payload_q, agent_q, host_q);
        }
    }

    if (sql.empty())
        return;

    // quiet: depending on the branch above the statement carries a bearer token,
    // a username and password, or a session code together with its secret. PgPool
    // logs statement text, and a dedicated postgres.log keeps it at debug — this
    // runs on every WebSocket call, so it would be a continuous credential leak
    // into a file that lives inside the container and is readable by any process
    // there. Mirrors AppServer's fetch, which does the same on the HTTP side.
    pool_.execute(std::move(sql),
        [this, session, unique_id, action](std::vector<PgResult> results) {
            on_fetch_result(session, unique_id, action, std::move(results));
        },
        [this, session, unique_id](std::string_view error) {
            send_call_error(*session->ws, unique_id, 500,
                            std::string(error));
        },
        /*quiet=*/true);
}

void WebSocketAPI::signed_fetch(std::shared_ptr<WsSession> session,
                                const std::string& unique_id,
                                const std::string& action,
                                const std::string& payload)
{
    // Nonce = microsecond epoch as string
    auto us = std::chrono::duration_cast<std::chrono::microseconds>(
        std::chrono::system_clock::now().time_since_epoch()).count();
    auto nonce = std::to_string(us);

    // HMAC input: action + nonce + payload (or "null")
    auto sig_payload = (payload.empty() || payload == "{}" || payload == "[]")
                           ? std::string("null")
                           : payload;
    auto data = action + nonce + sig_payload;
    auto signature = hmac_sha256_hex(session->secret, data);

    auto method_q    = pq_quote_literal("POST");
    auto action_q    = pq_quote_literal(action);
    auto payload_q   = (payload.empty() || payload == "{}" || payload == "[]")
                           ? std::string("null")
                           : pq_quote_literal(payload);
    auto session_q   = pq_quote_literal(session->session);
    auto nonce_q     = pq_quote_literal(nonce);
    auto signature_q = pq_quote_literal(signature);
    auto agent_q     = pq_quote_literal(session->agent);
    auto host_q      = pq_quote_literal(session->ip);

    auto sql = fmt::format(
        "SELECT * FROM daemon.signed_fetch({}, {}, {}::json, {}, {}, {}, {}, {}, "
        "INTERVAL '{} milliseconds')",
        method_q, action_q, payload_q, session_q, nonce_q,
        signature_q, agent_q, host_q, kReceiveWindowMs);

    // quiet: the statement carries the session code, the nonce and the signature
    // together. The signature counts as a credential even though it is not one on
    // its own: daemon.signed_fetch accepts the triple within pTimeWindow, so a
    // reader of the log can REPLAY that exact call while the window is open.
    // Forging a new call is not possible — which is why this is lighter than a
    // password — but "not a secret" is the wrong conclusion, and the next reader
    // would otherwise strip the flag as harmless.
    pool_.execute(std::move(sql),
        [this, session, unique_id, action](std::vector<PgResult> results) {
            on_fetch_result(session, unique_id, action, std::move(results));
        },
        [this, session, unique_id](std::string_view error) {
            send_call_error(*session->ws, unique_id, 500,
                            std::string(error));
        },
        /*quiet=*/true);
}

// ─── PG result handling ─────────────────────────────────────────────────────

void WebSocketAPI::on_fetch_result(std::shared_ptr<WsSession> session,
                                   const std::string& unique_id,
                                   const std::string& action,
                                   std::vector<PgResult> results,
                                   bool open)
{
    if (close_sent(*session))
        return;

    if (results.empty() || !results[0].ok()) {
        std::string err = results.empty()
            ? "no result"
            : (results[0].error_message()
                ? results[0].error_message() : "unknown error");
        send_call_error(*session->ws, unique_id, 500, err);
        return;
    }

    const auto& res = results[0];
    const bool is_list = (action.find("/list") != std::string::npos);

    if (res.rows() == 0 || res.columns() == 0) {
        send_call_result(*session->ws, unique_id, is_list ? "[]" : "{}");
        return;
    }

    // Build JSON from result
    std::string body = pg_result_to_json(res, is_list ? "array" : "");

    // Check for application-level error
    std::string error_message;
    int error_code = check_pg_error(body, error_message);
    if (error_code != 0) {
        int status = static_cast<int>(error_code_to_status(error_code));
        send_call_error(*session->ws, unique_id, status, error_message);

        // De-authorize on 401
        if (status == 401) {
            session->secret.clear();
            session->auth = Authorization{};
            session->authorized = false;
            session->verified   = false;
            session->open_token_own = false;
        }
        return;
    }

    // Update session state based on action
    try {
        auto j = nlohmann::json::parse(body);
        after_query(*session, action, j, open);

        // For /list paths, ensure the response is always a JSON array
        if (is_list && !j.is_array()) {
            body = "[" + body + "]";
        }
    } catch (...) {
        // Non-JSON result — still send it
    }

    send_call_result(*session->ws, unique_id, body);
}

// ─── after_query ────────────────────────────────────────────────────────────

void WebSocketAPI::after_query(WsSession& session, std::string_view action,
                               const nlohmann::json& payload, bool open)
{
    if (action == "/api/v1/sign/in") {
        // Verified only when the answer names the session the socket moves
        // to; otherwise it would stay under the code it was opened on.
        const bool named = payload.is_object() && payload.contains("session");
        rekey_session(session, payload.value("session", session.session));
        session.secret     = payload.value("secret", session.secret);
        session.authorized = true;
        session.verified   = named;
    } else if (action == "/api/v1/sign/out") {
        session.secret.clear();
        session.auth = Authorization{};
        session.authorized = false;
        session.verified   = false;
        session.open_token_own = false;
    } else if (action == "/api/v1/authenticate" ||
               action == "/api/v1/authorize") {
        bool auth_ok = payload.value("authorized", false);
        if (auth_ok) {
            session.authorized = true;
            // Only an OPEN's answer: it asked about this code. /authorize
            // checks the code alone (the token was checked here), so a token
            // without sub would verify any code.
            session.verified   = open && (action == "/api/v1/authenticate" || session.open_token_own);
        } else {
            session.secret.clear();
            session.auth = Authorization{};
            session.authorized = false;
            session.verified   = false;
            session.open_token_own = false;
        }
    }
}

// ─── check_sessions ─────────────────────────────────────────────────────────

void WebSocketAPI::check_sessions()
{
    std::vector<int> dead_fds;
    const auto now = std::chrono::steady_clock::now();

    for (const auto& [fd, session] : sessions_by_fd_) {
        if (!session->ws || session->ws->fd() < 0 || session->ws->closed() ||
            (close_sent(*session) && now - session->close_sent > kCloseGrace))
            dead_fds.push_back(fd);
    }

    for (int fd : dead_fds)
        remove_session(fd);
}

// ─── keep_alive ─────────────────────────────────────────────────────────────

void WebSocketAPI::keep_alive()
{
    if (!keepalive_.enabled())
        return;

    std::vector<int> dead_fds;
    const auto now = std::chrono::steady_clock::now();

    for (const auto& [fd, session] : sessions_by_fd_) {
        // A socket the server already closed waits for the peer's answer
        // under kCloseGrace; a ping after it would be noise.
        if (!session->ws || session->ws->closed() || close_sent(*session))
            continue;
        if (keepalive_.step(*session->ws, now) == WsKeepAlive::Step::closed)
            dead_fds.push_back(fd);
    }

    for (int fd : dead_fds)
        remove_session(fd);
}

// ─── Observer (LISTEN/NOTIFY) ───────────────────────────────────────────────

void WebSocketAPI::init_listen()
{
    if (listen_pending_)
        return;  // a publisher_list() query is already out; see listen_pending_

    auto notify_cb = [this](std::string_view channel, std::string_view data) {
        on_notify(channel, data);
    };

    listen_pending_ = true;

    // Query publisher channels via helper pool (apibot) using daemon.publisher_list()
    // (SECURITY DEFINER wrapper for db.publisher). Register each channel on
    // pool_ dedicated listener connection.
    pool_.execute("SELECT * FROM daemon.publisher_list()",
        [this, notify_cb](std::vector<PgResult> results) {
            listen_pending_ = false;

            // One attempt, not a retry loop. db.publisher is written by install
            // and by patch, so within the life of this process the answer is a
            // constant: a zero that arrives once arrives forever, and retrying
            // it every minute is a query that can never start succeeding.
            // Losing the subscription is handled where it belongs — PgPool
            // restores it from its own registry, without asking again.
            listen_initialized_ = true;

            if (results.empty() || !results[0].ok()) {
                log_.error("WebSocketAPI: daemon.publisher_list() returned no usable result "
                           "— no channels subscribed, this worker will publish nothing");
                return;
            }

            int registered = 0;
            for (int i = 0; i < results[0].rows(); ++i) {
                auto channel = std::string(results[0].value(i, 0));
                pool_.listen(channel, notify_cb);
                ++registered;
            }

            if (registered == 0) {
                // Loud on purpose: db.publisher is populated by install/patch,
                // so an empty set means the database is not installed — a
                // deployment fault, not a quiet state to sit in.
                log_.error("WebSocketAPI: daemon.publisher_list() returned ZERO publishers "
                           "— the dashboard and the driver app will receive nothing. "
                           "db.publisher is filled by install/patch: an empty set means "
                           "the database is not installed, not that publishing is off");
                return;
            }

            log_.notice("WebSocketAPI: subscribed to {} publisher channel(s)", registered);
        },
        [this](std::string_view error) {
            listen_pending_ = false;

            // An exception is NOT an empty publisher set, and the two must never
            // read alike: USAGE on schema daemon is granted to daemon and apibot
            // only, so a wrong role fails HERE with "permission denied for schema
            // daemon" and never returns zero rows.
            //
            // The latch is deliberately NOT set: this branch also covers a
            // database that is simply not up yet when the worker starts, and
            // latching would leave that worker subscribed to nothing forever.
            // The price is that a permanent fault repeats — which is why the
            // line is printed at error and names itself as a query failure, so
            // it cannot pass for waiting.
            log_.error("WebSocketAPI: daemon.publisher_list() failed, will retry — this is a "
                       "query or role failure, NOT an empty publisher set: {}", error);
        },
        true);  // quiet
}

void WebSocketAPI::on_notify(std::string_view channel,
                             std::string_view data)
{
    // data is JSON; publisher comes from payload field or channel name
    std::string publisher;
    std::string notify_data(data);

    try {
        auto j = nlohmann::json::parse(data);
        publisher = j.value("publisher", "");
    } catch (...) {
        return;
    }

    // Fall back to channel name as publisher (e.g., pg_notify('transaction', ...))
    if (publisher.empty())
        publisher = std::string(channel);

    if (publisher.empty())
        return;

    // Queue observer dispatch for each socket that proved a credential for its
    // code. daemon.observer admits by the session code alone (SessionIn with no
    // secret) and EventListener returns the events of that code's subscription,
    // keyed by (publisher, code, identity). A socket that only holds the code —
    // Basic or Session+Secret handshake headers, which set authorized with no
    // check and are verified by the database per CALL (WSA.DEV.02), reusing the
    // code holder's identity — would otherwise ride the holder's subscription
    // and receive its events. Gating on verified, not authorized, closes that
    // road the same way the removed POST /ws was closed (apostol-csms T1066).
    for (auto& [fd, session] : sessions_by_fd_) {
        if (!session->verified)
            continue;

        observer_queue_.push_back(
            ObserverTask{session, publisher, notify_data});
    }

    unload_queue();
}

void WebSocketAPI::unload_queue()
{
    while (!observer_queue_.empty() &&
           observer_progress_ < max_observer_queue_) {
        auto task = std::move(observer_queue_.front());
        observer_queue_.pop_front();
        dispatch_observer(std::move(task));
    }
}

void WebSocketAPI::dispatch_observer(ObserverTask task)
{
    auto session = task.session;
    auto publisher = task.publisher;

    auto sql = fmt::format(
        "SELECT * FROM daemon.observer({}, {}, {}, {}::jsonb, {}, {})",
        pq_quote_literal(publisher),
        pq_quote_literal(session->session),
        pq_quote_literal(session->identity),
        pq_quote_literal(task.data),
        pq_quote_literal(session->agent),
        pq_quote_literal(session->ip));

    observer_progress_++;

    // The answer belongs to the session code it was asked for. sign/in on the
    // same socket replaces session->session (after_query); a late answer for
    // the old code must neither close the new one nor reach it with events.
    pool_.execute(std::move(sql),
        [this, session, publisher, code = session->session](std::vector<PgResult> results) {
            observer_progress_--;

            if (session->session != code || close_sent(*session)) {
                unload_queue();
                return;
            }

            if (!results.empty() && results[0].ok() &&
                results[0].rows() > 0 && results[0].columns() > 0) {

                const auto& res = results[0];
                std::string body = pg_result_to_json(res);

                if (!body.empty()) {
                    // Check for error (401 = de-authorize and close).
                    //
                    // daemon.observer() answers 401 when the session is gone
                    // (signed out, deleted) or SessionIn() refused it (locked,
                    // password expired, IP table). De-authorizing alone left
                    // the client on a socket it believed live, receiving no
                    // more events and told nothing. Closing is the signal every
                    // client already handles: onclose → reconnect → authorize,
                    // which either succeeds or fails where the client sees it.
                    // Several notifications may be queued for one session —
                    // only the first 401 closes; the rest see close_sent.
                    std::string error_message;
                    int error_code = check_pg_error(body, error_message);
                    if (error_code != 0) {
                        int status = static_cast<int>(
                            error_code_to_status(error_code));
                        if (status == 401 && session->authorized) {
                            session->secret.clear();
                            session->auth = Authorization{};
                            session->authorized = false;
                            session->verified   = false;
                            session->open_token_own = false;
                            log_.info("WebSocketAPI: observer /{} refused the session on fd {}: {} — closing",
                                      publisher, session->ws ? session->ws->fd() : -1,
                                      error_message);
                            session->close_sent = std::chrono::steady_clock::now();
                            if (session->ws && !session->ws->closed())
                                session->ws->send_close(1008, "Unauthorized");
                        }
                    } else {
                        send_call(*session->ws,
                                  fmt::format("/{}", publisher), body);
                    }
                }
            }

            unload_queue();
        },
        [this, session](std::string_view) {
            observer_progress_--;
            unload_queue();
        },
        true);  // quiet
}

// ─── HTTP handlers ──────────────────────────────────────────────────────────

void WebSocketAPI::do_get(const HttpRequest& req, HttpResponse& resp)
{
    // No GET route. GET /ws/list used to list every live session — its code,
    // identity, authorized flag, peer address and user agent — to anyone who
    // asked, with no authorization at all (SERP finding, apostol-csms T1061).
    // A session code is a credential in this stack: it is what the __Host-SID
    // cookie carries and what FileServer accepts in a Session header. The
    // listing was a diagnostic inherited from apostol-crm with no consumer in
    // the product (web/driver apps use the /session/<code> socket; the OCPP
    // console reads /ws/log in cs), so it is removed rather than gated.
    reply_error(resp, HttpStatus::not_found, "Not found.");
}

void WebSocketAPI::do_post(const HttpRequest& req, HttpResponse& resp)
{
    // No POST route. POST /ws/<code> used to push a body to every socket on a
    // session code. It had no sender anywhere in the product — nginx on prod
    // logged no 2xx for 47 days, and no caller exists in any layer (db/sql,
    // backend/src outside this module, ocpp/src, go, ai-service, the
    // frontends); the only matches are copies of this module and /ws/log,
    // which is cs's own CSService, a different service (apostol-csms T1061
    // measurement, T1066). Its reach was the server push, and the server push
    // now travels only the observer road (/session/<code>, on_notify), gated
    // on a verified credential. A diagnostic with no consumer and the same
    // reach as a gated road is removed, not re-gated — as GET /ws/list was in
    // T1061 (apostol-csms T1066, WSA.DEV.02).
    reply_error(resp, HttpStatus::not_found, "Not found.");
}

} // namespace apostol

#endif // defined(WITH_POSTGRESQL) && defined(WITH_SSL)