Implement USB topology monitoring and WebSocket integration

- Added a WebSocket controller for real-time streaming of USB device topology.
- Introduced a background worker to scan USB devices and update the topology state.
- Enhanced the UI to display USB topology with dynamic updates via WebSocket.
- Updated the device data structure to include additional fields for USB nodes.
- Improved error handling and data fetching mechanisms in the UI for better user experience.

Author: tdewey-rpi

provisioner-service/src/devices.cpp

@@ -1,16 +1,78 @@
 #include <drogon/HttpResponse.h>
 #include <drogon/HttpAppFramework.h>
+#include <drogon/WebSocketController.h>
 #include <devices.h>
 #include <sqlite3.h>
 #include <json/json.h>
 #include <fstream>
 #include <sstream>
 #include <systemd/sd-bus.h>
+#include <filesystem>
+#include <unordered_map>
+#include <set>
+#include <thread>
+#include <mutex>
+#include <atomic>
+#include <chrono>
+#include <algorithm>
 #include "utils.h"
 #include "include/audit.h"
 
 using namespace drogon;
 
+// Forward declare internal snapshot function from this TU's anonymous namespace
+namespace { std::string topologySnapshotString(); }
+
+// WebSocket controller for device topology streaming
+class DevicesWebSocketController : public drogon::WebSocketController<DevicesWebSocketController> {
+public:
+    static std::vector<drogon::WebSocketConnectionPtr> subscribers;
+    static std::mutex subscribersMutex;
+
+    static void broadcast(const std::string &message) {
+        std::lock_guard<std::mutex> lock(subscribersMutex);
+        auto it = subscribers.begin();
+        while (it != subscribers.end()) {
+            if ((*it)->connected()) {
+                (*it)->send(message);
+                ++it;
+            } else {
+                it = subscribers.erase(it);
+            }
+        }
+    }
+
+    void handleNewMessage(const drogon::WebSocketConnectionPtr& wsConnPtr, std::string&& message, const drogon::WebSocketMessageType& type) override {
+        // No-op for now; could support subscription filters later
+        (void)wsConnPtr; (void)message; (void)type;
+    }
+
+    void handleNewConnection(const drogon::HttpRequestPtr& req, const drogon::WebSocketConnectionPtr& wsConnPtr) override {
+        (void)req;
+        {
+            std::lock_guard<std::mutex> lock(subscribersMutex);
+            subscribers.push_back(wsConnPtr);
+        }
+        // Send initial snapshot
+        wsConnPtr->send(topologySnapshotString());
+    }
+
+    void handleConnectionClosed(const drogon::WebSocketConnectionPtr& wsConnPtr) override {
+        std::lock_guard<std::mutex> lock(subscribersMutex);
+        subscribers.erase(std::remove(subscribers.begin(), subscribers.end(), wsConnPtr), subscribers.end());
+    }
+
+    WS_PATH_LIST_BEGIN
+    WS_PATH_ADD("/ws/devices");
+    WS_PATH_LIST_END
+};
+
+std::vector<drogon::WebSocketConnectionPtr> DevicesWebSocketController::subscribers;
+std::mutex DevicesWebSocketController::subscribersMutex;
+
+// Forward declaration for snapshot helper
+namespace provisioner { std::string topologySnapshotString(); }
+
 namespace provisioner {
 
     struct Device {
@@ -25,6 +87,203 @@ namespace provisioner {
         std::vector<Device> devices;
     };
 
+    // Representation of a USB node (hub or device)
+    struct UsbNode {
+        std::string id;          // unique path id (e.g. 1-1.4)
+        std::string parentId;    // parent path id or empty for root
+        std::string vendor;      // idVendor
+        std::string product;     // idProduct
+        std::string serial;      // serial (if available)
+        bool isHub{false};
+        // Optional provisioning info
+        std::string state;
+        std::string image;
+        std::string ip;
+    };
+
+    // Tracker state
+    namespace {
+        std::thread topologyThread;
+        std::atomic<bool> topologyRunning{false};
+        std::mutex topologyMutex;
+        // Map id -> node
+        std::unordered_map<std::string, UsbNode> currentTopology;
+
+        std::string readFileTrimmed(const std::filesystem::path &p) {
+            try {
+                std::ifstream f(p);
+                if (!f.is_open()) return "";
+                std::string s;
+                std::getline(f, s);
+                while (!s.empty() && (s.back()=='\n' || s.back()=='\r' || s.back()=='\t' || s.back()==' ')) s.pop_back();
+                return s;
+            } catch (...) { return ""; }
+        }
+
+        std::string baseDeviceDirName(const std::string &name) {
+            // Strip interface suffix after ':' (e.g. 1-1.2:1.0 -> 1-1.2)
+            auto pos = name.find(':');
+            if (pos != std::string::npos) return name.substr(0, pos);
+            return name;
+        }
+
+        std::string computeParentId(const std::string &id) {
+            // Parent is prefix before last '.' if present, otherwise attach to server
+            auto pos = id.rfind('.');
+            if (pos != std::string::npos) {
+                return id.substr(0, pos);
+            }
+            // If id contains '-' treat as directly connected to server
+            return "server";
+        }
+
+        std::unordered_map<std::string, UsbNode> scanUsbSysfs() {
+            std::unordered_map<std::string, UsbNode> nodes;
+            // Always include the server root node
+            UsbNode server;
+            server.id = "server";
+            server.parentId = "";
+            server.isHub = true;
+            server.vendor = "Provisioner";
+            server.product = "Server";
+            nodes.insert({server.id, server});
+
+            const std::filesystem::path usbPath("/sys/bus/usb/devices");
+            if (!std::filesystem::exists(usbPath)) {
+                return nodes;
+            }
+
+            for (const auto &entry : std::filesystem::directory_iterator(usbPath)) {
+                if (!entry.is_directory()) continue;
+                const std::string name = entry.path().filename().string();
+                if (name.find('-') == std::string::npos) continue; // skip non-device entries like usb1
+                const std::string id = baseDeviceDirName(name);
+                UsbNode node;
+                node.id = id;
+                node.parentId = computeParentId(id);
+                node.vendor = readFileTrimmed(entry.path()/"idVendor");
+                node.product = readFileTrimmed(entry.path()/"idProduct");
+                if (node.vendor.empty() && node.product.empty()) {
+                    // Some entries may not be real devices; skip if missing identifiers
+                    // but keep hubs if detectable
+                }
+                node.serial = readFileTrimmed(entry.path()/"serial");
+                std::string bDeviceClass = readFileTrimmed(entry.path()/"bDeviceClass");
+                // 09 (hex) indicates hub
+                node.isHub = (bDeviceClass == "09" || bDeviceClass == "9" || bDeviceClass == "0x09");
+                nodes.insert_or_assign(node.id, node);
+            }
+
+            return nodes;
+        }
+
+        void enrichWithProvisioningState(std::unordered_map<std::string, UsbNode> &nodes) {
+            sqlite3* db;
+            int rc = sqlite3_open("/srv/rpi-sb-provisioner/state.db", &db);
+            if (rc) {
+                return;
+            }
+            const char* sql = "SELECT serial, endpoint, state, image, ip_address FROM devices ORDER BY ts DESC";
+            sqlite3_stmt* stmt;
+            rc = sqlite3_prepare_v2(db, sql, -1, &stmt, nullptr);
+            if (rc != SQLITE_OK) {
+                sqlite3_close(db);
+                return;
+            }
+            while (sqlite3_step(stmt) == SQLITE_ROW) {
+                const unsigned char* serial = sqlite3_column_text(stmt, 0);
+                const unsigned char* endpoint = sqlite3_column_text(stmt, 1);
+                const unsigned char* state = sqlite3_column_text(stmt, 2);
+                const unsigned char* image = sqlite3_column_text(stmt, 3);
+                const unsigned char* ip = sqlite3_column_text(stmt, 4);
+                if (!endpoint) continue;
+                std::string endpointStr = reinterpret_cast<const char*>(endpoint);
+                auto it = nodes.find(endpointStr);
+                if (it != nodes.end()) {
+                    // Only set if not already populated or to reflect latest row (desc order)
+                    it->second.serial = serial ? reinterpret_cast<const char*>(serial) : it->second.serial;
+                    it->second.state = state ? reinterpret_cast<const char*>(state) : it->second.state;
+                    it->second.image = image ? reinterpret_cast<const char*>(image) : it->second.image;
+                    it->second.ip = ip ? reinterpret_cast<const char*>(ip) : it->second.ip;
+                }
+            }
+            sqlite3_finalize(stmt);
+            sqlite3_close(db);
+        }
+
+        Json::Value topologyToJson(const std::unordered_map<std::string, UsbNode> &nodes) {
+            Json::Value root;
+            root["type"] = "topology";
+            Json::Value arr(Json::arrayValue);
+            for (const auto &p : nodes) {
+                const UsbNode &n = p.second;
+                Json::Value j;
+                j["id"] = n.id;
+                if (!n.parentId.empty()) j["parentId"] = n.parentId; else j["parentId"] = Json::nullValue;
+                j["isHub"] = n.isHub;
+                if (!n.vendor.empty()) j["vendor"] = n.vendor;
+                if (!n.product.empty()) j["product"] = n.product;
+                if (!n.serial.empty()) j["serial"] = n.serial;
+                if (!n.state.empty()) j["state"] = n.state;
+                if (!n.image.empty()) j["image"] = n.image;
+                if (!n.ip.empty()) j["ip"] = n.ip;
+                arr.append(j);
+            }
+            root["nodes"] = arr;
+            root["timestamp"] = static_cast<Json::UInt64>(std::chrono::duration_cast<std::chrono::milliseconds>(
+                std::chrono::system_clock::now().time_since_epoch()).count());
+            return root;
+        }
+
+        std::string topologySnapshotString() {
+            std::lock_guard<std::mutex> lock(topologyMutex);
+            Json::FastWriter w; // compact ok
+            return w.write(topologyToJson(currentTopology));
+        }
+        
+        void topologyWorker() {
+            LOG_INFO << "Topology worker started";
+            Json::Value lastJson;
+            while (topologyRunning) {
+                auto newMap = scanUsbSysfs();
+                enrichWithProvisioningState(newMap);
+
+                bool changed = false;
+                {
+                    std::lock_guard<std::mutex> lock(topologyMutex);
+                    // Simple change detection by size and a few key fields; fallback to full JSON compare
+                    if (newMap.size() != currentTopology.size()) {
+                        changed = true;
+                    } else {
+                        for (const auto &p : newMap) {
+                            auto it = currentTopology.find(p.first);
+                            if (it == currentTopology.end()) { changed = true; break; }
+                            const UsbNode &a = p.second;
+                            const UsbNode &b = it->second;
+                            if (a.parentId != b.parentId || a.vendor != b.vendor || a.product != b.product ||
+                                a.serial != b.serial || a.isHub != b.isHub || a.state != b.state ||
+                                a.image != b.image || a.ip != b.ip) { changed = true; break; }
+                        }
+                    }
+                    if (changed) {
+                        currentTopology = std::move(newMap);
+                    }
+                }
+
+                if (changed) {
+                    std::string msg = topologySnapshotString();
+                    DevicesWebSocketController::broadcast(msg);
+                }
+
+                for (int i=0; i<10 && topologyRunning; ++i) {
+                    std::this_thread::sleep_for(std::chrono::milliseconds(300));
+                }
+            }
+            LOG_INFO << "Topology worker stopped";
+        }
+    }
+    }
+
     Devices::Devices() 
         :
         systemd_bus(nullptr, sd_bus_unref)
@@ -35,9 +294,20 @@ namespace provisioner {
             throw std::runtime_error("Failed to connect to system bus: " + std::string(strerror(-ret)));
         }
         systemd_bus = std::unique_ptr<sd_bus, decltype(&sd_bus_unref)>(bus, sd_bus_unref);
+        // Start topology watcher
+        if (!topologyRunning) {
+            topologyRunning = true;
+            topologyThread = std::thread(topologyWorker);
+        }
     }
 
-    Devices::~Devices() = default;
+    Devices::~Devices() {
+        // Stop topology watcher
+        if (topologyRunning) {
+            topologyRunning = false;
+            if (topologyThread.joinable()) topologyThread.join();
+        }
+    }
 
     void Devices::registerHandlers(drogon::HttpAppFramework &app)
     {