vdm/daemon/tests/uds_roundtrip_test.cpp

// Integration: a real UdsServer on a temp socket, a real client socket, NDJSON round trips.
// Proves the transport, the generated dispatch() wiring, and the session-layer handling.

#include <sys/socket.h>
#include <sys/un.h>
#include <unistd.h>

#include <cstdlib>
#include <cstring>
#include <string>
#include <thread>

#include <nlohmann/json.hpp>

#include "check.hpp"
#include "rpc/dispatcher.hpp"
#include "rpc/event_loop.hpp"
#include "rpc/ndjson.hpp"
#include "rpc/uds_server.hpp"
#include "store/migrations.hpp"
#include "store/sqlite.hpp"
#include "velox_proto.hpp"

using nlohmann::json;
namespace rpc = velox::daemon::rpc;

namespace {

std::string make_temp_socket_path() {
    char tmpl[] = "/tmp/veloxd-test-XXXXXX";
    const char* dir = ::mkdtemp(tmpl);
    return std::string(dir ? dir : "/tmp") + "/velox.sock";
}

int connect_client(const std::string& path) {
    const int fd = ::socket(AF_UNIX, SOCK_STREAM | SOCK_CLOEXEC, 0);
    sockaddr_un addr{};
    addr.sun_family = AF_UNIX;
    std::memcpy(addr.sun_path, path.c_str(), path.size());
    if (::connect(fd, reinterpret_cast<sockaddr*>(&addr), sizeof(addr)) != 0) {
        ::close(fd);
        return -1;
    }
    return fd;
}

// Send one framed request, read one framed reply (blocking client; the server is async).
json call(int fd, const json& request) {
    const std::string out = rpc::frame(request.dump());
    if (::write(fd, out.data(), out.size()) != static_cast<ssize_t>(out.size())) return {};

    std::string buf;
    char chunk[4096];
    for (;;) {
        const ssize_t n = ::read(fd, chunk, sizeof(chunk));
        if (n <= 0) return {};
        buf.append(chunk, static_cast<std::size_t>(n));
        if (const auto nl = buf.find('\n'); nl != std::string::npos)
            return json::parse(buf.substr(0, nl), nullptr, false);
    }
}

}  // namespace

void run() {
    const std::string sock = make_temp_socket_path();

    auto db = velox::daemon::store::Db::open(":memory:");
    CHECK(db.has_value());
    if (!db) return;
    CHECK(velox::daemon::store::migrate_to_head(*db).has_value());

    rpc::EventLoop loop;
    rpc::VeloxDispatcher dispatcher(*db);
    rpc::UdsServer server(loop, dispatcher, sock);
    const auto ec = server.start();
    CHECK(!ec);
    if (ec) return;

    std::thread loop_thread([&loop] { loop.run(); });

    // --- session.hello, matching major -> a real SessionHelloResult -------------------
    {
        const int c = connect_client(sock);
        CHECK(c >= 0);
        const json reply = call(c, {{"jsonrpc", "2.0"},
                                    {"id", 1},
                                    {"method", "session.hello"},
                                    {"params",
                                     {{"clientType", "test"},
                                      {"clientName", "roundtrip"},
                                      {"protocolVersion", std::string(velox::proto::kProtocolVersion)}}}});
        CHECK(reply.contains("result"));
        CHECK_EQ(reply["id"].get<int>(), 1);
        CHECK_EQ(reply["result"]["protocolVersion"].get<std::string>(),
                 std::string(velox::proto::kProtocolVersion));
        CHECK_EQ(reply["result"]["transport"].get<std::string>(), std::string("uds"));
        CHECK(!reply["result"]["sessionId"].get<std::string>().empty());
        ::close(c);
    }

    // --- session.hello, wrong major -> -32001, connection closed after the reply ------
    {
        const int c = connect_client(sock);
        const json reply = call(c, {{"jsonrpc", "2.0"},
                                    {"id", 2},
                                    {"method", "session.hello"},
                                    {"params",
                                     {{"clientType", "gui"},
                                      {"clientName", "from the future"},
                                      {"protocolVersion", "2.0.0"}}}});
        CHECK(reply.contains("error"));
        CHECK_EQ(reply["error"]["code"].get<int>(), -32001);
        CHECK_EQ(reply["error"]["data"]["actual"].get<std::string>(), std::string("2.0.0"));
        ::close(c);
    }

    // --- download.list -> an empty table (dispatcher answers this one for real) -------
    {
        const int c = connect_client(sock);
        const json reply =
            call(c, {{"jsonrpc", "2.0"}, {"id", 3}, {"method", "download.list"}, {"params", json::object()}});
        CHECK(reply.contains("result"));
        CHECK_EQ(reply["result"]["total"].get<int>(), 0);
        CHECK(reply["result"]["items"].is_array());
        CHECK_EQ(reply["result"]["items"].size(), 0u);
        ::close(c);
    }

    // --- an unknown method -> -32601 ------------------------------------------------
    {
        const int c = connect_client(sock);
        const json reply =
            call(c, {{"jsonrpc", "2.0"}, {"id", 4}, {"method", "no.such.method"}, {"params", json::object()}});
        CHECK(reply.contains("error"));
        CHECK_EQ(reply["error"]["code"].get<int>(), -32601);
        ::close(c);
    }

    // --- malformed JSON -> -32700, id null ----------------------------------------
    {
        const int c = connect_client(sock);
        const std::string bad = "{ this is not json )\n";
        CHECK(::write(c, bad.data(), bad.size()) == static_cast<ssize_t>(bad.size()));
        std::string buf;
        char chunk[1024];
        const ssize_t n = ::read(c, chunk, sizeof(chunk));
        CHECK(n > 0);
        if (n > 0) {
            buf.assign(chunk, static_cast<std::size_t>(n));
            const json reply = json::parse(buf.substr(0, buf.find('\n')), nullptr, false);
            CHECK_EQ(reply["error"]["code"].get<int>(), -32700);
            CHECK(reply["id"].is_null());
        }
        ::close(c);
    }

    // --- download.get on an unknown id -> -32010, with data.taskId ------------------
    // (contracts/ error fixture download.get.not-found; reachable now that 1.4.0 gave
    // handlers the HandlerError channel — ADR 0014.)
    {
        const int c = connect_client(sock);
        const std::string missing = "00000000-0000-4000-8000-000000000000";
        const json reply = call(c, {{"jsonrpc", "2.0"},
                                    {"id", 6},
                                    {"method", "download.get"},
                                    {"params", {{"taskId", missing}}}});
        CHECK(reply.contains("error"));
        CHECK_EQ(reply["error"]["code"].get<int>(), -32010);
        CHECK_EQ(reply["error"]["data"]["taskId"].get<std::string>(), missing);
        ::close(c);
    }

    // --- two requests in one write, pipelined on one connection --------------------
    {
        const int c = connect_client(sock);
        std::string out = rpc::frame(json({{"jsonrpc", "2.0"}, {"id", 7}, {"method", "download.list"}, {"params", json::object()}}).dump());
        out += rpc::frame(json({{"jsonrpc", "2.0"}, {"id", 8}, {"method", "download.list"}, {"params", json::object()}}).dump());
        CHECK(::write(c, out.data(), out.size()) == static_cast<ssize_t>(out.size()));
        std::string buf;
        char chunk[4096];
        int seen = 0;
        while (seen < 2) {
            const ssize_t n = ::read(c, chunk, sizeof(chunk));
            if (n <= 0) break;
            buf.append(chunk, static_cast<std::size_t>(n));
            std::size_t nl;
            while ((nl = buf.find('\n')) != std::string::npos) {
                const json reply = json::parse(buf.substr(0, nl), nullptr, false);
                CHECK(reply.contains("result"));
                ++seen;
                buf.erase(0, nl + 1);
            }
        }
        CHECK_EQ(seen, 2);
        ::close(c);
    }

    loop.stop();
    loop_thread.join();
    ::unlink(sock.c_str());
}

TEST_MAIN()