vdm/daemon/src/main.cpp

// veloxd — the Velox download-manager daemon.
//
// Wires up both RPC transports (Unix socket + loopback WebSocket), the SQLite store,
// and a dispatcher skeleton so the CLI and GUI have a real server to speak to
// (AGENT-DAEMON.md build order, steps 1 and 3). The scheduler and the engine link land
// next.

#include <csignal>
#include <cstdlib>
#include <cstring>
#include <iostream>
#include <string>

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

#include <sys/timerfd.h>

#include <nlohmann/json.hpp>

#include "rpc/dispatcher.hpp"
#include "rpc/event_hub.hpp"
#include "rpc/event_loop.hpp"
#include "rpc/pairing.hpp"
#include "rpc/runtime_dir.hpp"
#include "rpc/single_instance.hpp"
#include "rpc/uds_server.hpp"
#include "rpc/ws_server.hpp"
#include "sched/engine_port_core.hpp"
#include "sched/governor.hpp"
#include "sched/scheduler.hpp"
#include "store/migrations.hpp"
#include "store/sqlite.hpp"
#include "util/time.hpp"
#include "vdm/engine.hpp"
#include "version.hpp"

namespace {

velox::daemon::rpc::EventLoop* g_loop = nullptr;

void on_signal(int) {
    if (g_loop != nullptr) g_loop->stop();  // stop() is async-signal-safe (writes an eventfd)
}

}  // namespace

int main() {
    std::cout << "veloxd " << velox::daemon::kDaemonVersion << " (protocol "
              << velox::proto::kProtocolVersion << ")\n";

    velox::daemon::rpc::RuntimeDir rt;
    if (const auto ec = velox::daemon::rpc::resolve_runtime_dir(rt)) {
        std::cerr << "veloxd: cannot prepare runtime directory: " << ec.message() << "\n";
        return 1;
    }

    const int lock_fd = velox::daemon::rpc::acquire_single_instance_lock(rt.path);
    if (lock_fd < 0) {
        std::cerr << "veloxd: another instance is already running for this runtime "
                     "directory (" << rt.path << ")\n";
        return 1;
    }

    velox::daemon::rpc::EventLoop loop;
    g_loop = &loop;

    struct sigaction sa{};
    sa.sa_handler = on_signal;
    ::sigemptyset(&sa.sa_mask);
    ::sigaction(SIGINT, &sa, nullptr);
    ::sigaction(SIGTERM, &sa, nullptr);
    ::signal(SIGPIPE, SIG_IGN);  // a client vanishing mid-write is EPIPE, never a signal

    std::string data_dir;
    if (const auto ec = velox::daemon::rpc::resolve_data_dir(data_dir)) {
        std::cerr << "veloxd: cannot prepare data directory: " << ec.message() << "\n";
        return 1;
    }
    auto db = velox::daemon::store::Db::open(data_dir + "/velox.db");
    if (!db) {
        std::cerr << "veloxd: cannot open " << data_dir << "/velox.db: "
                  << db.error().to_string() << "\n";
        return 1;
    }
    if (const auto m = velox::daemon::store::migrate_to_head(*db); !m) {
        std::cerr << "veloxd: schema migration failed: " << m.error().to_string() << "\n";
        return 1;
    }

    // --- engine + scheduler ---------------------------------------------------------
    velox::daemon::rpc::EventHub hub;
    vdm::Engine engine;
    velox::daemon::sched::EnginePortCore engine_port(engine);
    velox::daemon::sched::Scheduler scheduler(
        *db, engine_port, velox::daemon::sched::Governor{}, &hub,
        {/*local_now*/ {},
         /*post_to_loop*/ [&loop](std::function<void()> fn) { loop.post(std::move(fn)); }});

    if (const auto ec = scheduler.reconcile_after_restart(); !ec)
        std::cerr << "veloxd: restart reconcile: " << ec.error().to_string() << "\n";
    (void)scheduler.reload_config();
    (void)scheduler.tick();  // admit anything already queued in the DB

    velox::daemon::rpc::VeloxDispatcher dispatcher(*db, hub, &scheduler);
    dispatcher.set_on_mutation([&loop, &scheduler] {
        loop.post([&scheduler] { (void)scheduler.tick(); });
    });

    // A 1 s timer re-runs the scheduler so schedule windows opening/closing and any
    // missed nudge are picked up. Registered on the loop, no extra thread.
    const int tick_fd = ::timerfd_create(CLOCK_MONOTONIC, TFD_NONBLOCK | TFD_CLOEXEC);
    if (tick_fd >= 0) {
        itimerspec spec{};
        spec.it_value.tv_sec = 1;
        spec.it_interval.tv_sec = 1;
        ::timerfd_settime(tick_fd, 0, &spec, nullptr);
        loop.add_fd(tick_fd, velox::daemon::rpc::kRead, [&](int fd, unsigned) {
            std::uint64_t ticks = 0;
            [[maybe_unused]] ssize_t n = ::read(fd, &ticks, sizeof(ticks));
            (void)scheduler.tick();
        });
    }

    // event.task.progress: one array message at <=4 Hz (schema x-maxRateHz), never one
    // notification per task (AGENT-DAEMON.md item 5). 250 ms keeps every active task's
    // segment bar under 4 Hz without depending on how many tasks are running.
    const int progress_fd = ::timerfd_create(CLOCK_MONOTONIC, TFD_NONBLOCK | TFD_CLOEXEC);
    if (progress_fd >= 0) {
        itimerspec spec{};
        spec.it_value.tv_nsec = 250'000'000;
        spec.it_interval.tv_nsec = 250'000'000;
        ::timerfd_settime(progress_fd, 0, &spec, nullptr);
        loop.add_fd(progress_fd, velox::daemon::rpc::kRead, [&](int fd, unsigned) {
            std::uint64_t ticks = 0;
            [[maybe_unused]] ssize_t n = ::read(fd, &ticks, sizeof(ticks));
            const auto rows = scheduler.progress_snapshot();
            if (rows.empty()) return;

            nlohmann::json tasks_json = nlohmann::json::array();
            for (const auto& r : rows) {
                nlohmann::json t{{"taskId", r.task_id},
                                 {"downloadedBytes", r.downloaded_bytes},
                                 {"speedBps", r.speed_bps}};
                t["etaSeconds"] = r.eta_seconds ? nlohmann::json(*r.eta_seconds) : nlohmann::json(nullptr);
                if (!r.segments.empty()) {
                    nlohmann::json segs = nlohmann::json::array();
                    for (const auto& s : r.segments)
                        segs.push_back({{"index", s.index},
                                       {"downloadedBytes", s.downloaded_bytes},
                                       {"speedBps", s.speed_bps}});
                    t["segments"] = std::move(segs);
                }
                tasks_json.push_back(std::move(t));
            }
            const nlohmann::json params{{"tasks", std::move(tasks_json)},
                                        {"at", velox::daemon::now_iso()}};
            hub.publish(velox::proto::Event::TaskProgress,
                       velox::proto::make_notification(velox::proto::Event::TaskProgress, params));
        });
    }

    velox::daemon::rpc::UdsServer uds(loop, dispatcher, hub, rt.socket_path(), &scheduler);
    if (const auto ec = uds.start()) {
        std::cerr << "veloxd: cannot listen on " << rt.socket_path() << ": " << ec.message()
                  << "\n";
        return 1;
    }
    std::cout << "veloxd: listening on " << uds.socket_path() << "\n";

    // The WebSocket transport is the extension's fallback (docs/05 §4); the Unix socket is
    // the primary. If every port in 52000-52016 is taken, log it and carry on rather than
    // refusing to start — capture must fail open, and the GUI/CLI still have the socket.
    // TODO(build step 7): replace EnvAutoApprover with a GUI-dialog / desktop-notification
    // approver. Until then pairing needs VELOX_PAIR_AUTO=1.
    velox::daemon::rpc::EnvAutoApprover approver;
    velox::daemon::rpc::WsServer ws(loop, dispatcher, *db, approver, hub, rt, &scheduler);
    if (const auto ec = ws.start()) {
        std::cerr << "veloxd: WebSocket transport unavailable (" << ec.message()
                  << "); the extension fallback will not work this run\n";
    } else {
        std::cout << "veloxd: WebSocket transport on 127.0.0.1:" << ws.port() << "\n";
    }

    loop.run();
    std::cout << "veloxd: shutting down\n";

    if (tick_fd >= 0) {
        loop.del_fd(tick_fd);
        ::close(tick_fd);
    }
    if (progress_fd >= 0) {
        loop.del_fd(progress_fd);
        ::close(progress_fd);
    }
    g_loop = nullptr;
    ::close(lock_fd);
    return 0;
}