dmf-studio-rnd/studio-manager/main.cpp

#include <filesystem>
#include <mutex>
#include <string>
#include <thread>
#include <unordered_set>
#include <crow.h>
#include "StudioManager.hpp"

namespace fs = std::filesystem;

int main(int argc, char* argv[]) {
    dmf::install_signal_handlers();

    const fs::path bin_dir = fs::path(argv[0]).parent_path();

#ifdef __APPLE__
    const std::string domain = "/tmp/dmf-studio";
#else
    const std::string domain = "/dev/shm/dmf-studio";
#endif

    std::error_code ec;
    fs::create_directories(domain, ec);
    if (ec) {
        fprintf(stderr, "[studio-manager] cannot create domain dir: %s\n", ec.message().c_str());
        return 1;
    }

    dmf::StudioManager manager(bin_dir, domain);

    {
        bool clean = false;
        const char* explicit_path = nullptr;
        for (int i = 1; i < argc; ++i) {
            if (std::string(argv[i]) == "--clean") clean = true;
            else explicit_path = argv[i];
        }

        fs::path load_path = explicit_path ? fs::path(explicit_path) : bin_dir / "last_graph.json";
        if (!clean && fs::exists(load_path)) {
            fprintf(stderr, "[studio-manager] restoring graph from %s\n", load_path.c_str());
            auto r = manager.load_graph_file(load_path.string());
            if (r.value("type", "") == "error") {
                fprintf(stderr, "[studio-manager] %s\n",
                    r.value("message", "load failed").c_str());
                if (explicit_path) return 1;  // explicit path failure is fatal; auto-restore is not
            }
        }
    }

    // ── WebSocket server ──────────────────────────────────────────────────────

    crow::SimpleApp app;
    app.signal_clear();  // let our signal handler manage SIGTERM/SIGINT

    std::mutex ws_mutex;
    std::unordered_set<crow::websocket::connection*> clients;

    // Push status to every connected client except the one whose command triggered
    // the change — that client gets the direct response from the handler instead.
    // tl_requester is thread-local: notify() is called synchronously from within
    // the manager command, so it runs on the same thread as onmessage.
    thread_local crow::websocket::connection* tl_requester = nullptr;

    manager.on_status_change([&](nlohmann::json status) {
        std::lock_guard lk(ws_mutex);
        const std::string msg = status.dump();
        for (auto* c : clients)
            if (c != tl_requester) c->send_text(msg);
    });

    CROW_WEBSOCKET_ROUTE(app, "/ws")
        .onopen([&](crow::websocket::connection& conn) {
            {
                std::lock_guard lk(ws_mutex);
                clients.insert(&conn);
            }
            // get_status() acquires manager mutex — must NOT hold ws_mutex here
            conn.send_text(manager.get_status().dump());
            fprintf(stderr, "[studio-manager] client connected\n");
        })
        .onclose([&](crow::websocket::connection& conn, const std::string&) {
            std::lock_guard lk(ws_mutex);
            clients.erase(&conn);
            fprintf(stderr, "[studio-manager] client disconnected\n");
        })
        .onmessage([&](crow::websocket::connection& conn, const std::string& data, bool) {
            auto j = nlohmann::json::parse(data, nullptr, false);
            if (j.is_discarded()) {
                conn.send_text(R"({"type":"error","message":"invalid JSON"})");
                return;
            }

            nlohmann::json response;
            const std::string type = j.value("type", "");
            tl_requester = &conn;
            try {
                if      (type == "load_graph") response = manager.load_graph(j.at("graph"));
                else if (type == "stop_node")  response = manager.stop_node(j.at("id"));
                else if (type == "start_node") response = manager.start_node(j.at("id"));
                else if (type == "get_status") response = manager.get_status();
                else response = {{"type","error"},{"message","unknown command: " + type}};
            } catch (const std::exception& e) {
                response = {{"type","error"},{"message", e.what()}};
            }
            tl_requester = nullptr;
            conn.send_text(response.dump());
        });

    // ── Monitor thread — detects crashes, stops Crow on shutdown ─────────────

    std::thread monitor([&] {
        manager.run_monitor();
        app.stop();
    });

    fprintf(stderr, "[studio-manager] WebSocket API at ws://0.0.0.0:7070/ws\n");
    app.port(7070).multithreaded().run();

    manager.shutdown();
    monitor.join();
    fprintf(stderr, "[studio-manager] done\n");
    return 0;
}