#ifdef ZT_CONTROLLER_USE_LIBPQ #include "PubSubListener.hpp" #include "DB.hpp" #include "opentelemetry/trace/provider.h" #include "rustybits.h" #include namespace ZeroTier { void listener_callback(void* user_ptr, const uint8_t* payload, uintptr_t length) { if (! user_ptr || ! payload || length == 0) { fprintf(stderr, "Invalid parameters in listener_callback\n"); return; } auto* listener = static_cast(user_ptr); std::string payload_str(reinterpret_cast(payload), length); listener->onNotification(payload_str); } PubSubNetworkListener::PubSubNetworkListener(std::string controller_id, uint64_t listen_timeout, DB* db) : _run(true), _controller_id(controller_id), _db(db), _listener(nullptr) { _listener = rustybits::network_listener_new(_controller_id.c_str(), listen_timeout, listener_callback, this); _listenThread = std::thread(&PubSubNetworkListener::listenThread, this); _changeHandlerThread = std::thread(&PubSubNetworkListener::changeHandlerThread, this); } PubSubNetworkListener::~PubSubNetworkListener() { _run = false; if (_listenThread.joinable()) { _listenThread.join(); } if (_listener) { rustybits::network_listener_delete(_listener); _listener = nullptr; } } void PubSubNetworkListener::onNotification(const std::string& payload) { auto provider = opentelemetry::trace::Provider::GetTracerProvider(); auto tracer = provider->GetTracer("PubSubNetworkListener"); auto span = tracer->StartSpan("PubSubNetworkListener::onNotification"); auto scope = tracer->WithActiveSpan(span); span->SetAttribute("payload", payload); fprintf(stderr, "Network notification received: %s\n", payload.c_str()); try { nlohmann::json j = nlohmann::json::parse(payload); nlohmann::json& ov_tmp = j["old"]; nlohmann::json& nv_tmp = j["new"]; nlohmann::json oldConfig, newConfig; if (ov_tmp.is_object()) { // TODO: copy old configuration to oldConfig // changing key names along the way } if (nv_tmp.is_object()) { // TODO: copy new configuration to newConfig // changing key names along the way } if (oldConfig.is_object() && newConfig.is_object()) { // network modification std::string nwid = oldConfig["id"].get(); span->SetAttribute("action", "network_change"); span->SetAttribute("network_id", nwid); _db->save(newConfig, _db->isReady()); } else if (newConfig.is_object() && ! oldConfig.is_object()) { // new network std::string nwid = newConfig["id"]; span->SetAttribute("network_id", nwid); span->SetAttribute("action", "new_network"); _db->save(newConfig, _db->isReady()); } else if (! newConfig.is_object() && oldConfig.is_object()) { // network deletion std::string nwid = oldConfig["id"]; span->SetAttribute("action", "delete_network"); span->SetAttribute("network_id", nwid); uint64_t networkId = Utils::hexStrToU64(nwid.c_str()); if (networkId) { _db->eraseNetwork(networkId); } } } catch (const nlohmann::json::parse_error& e) { fprintf(stderr, "JSON parse error: %s\n", e.what()); span->SetAttribute("error", e.what()); span->SetStatus(opentelemetry::trace::StatusCode::kError, e.what()); return; } catch (const std::exception& e) { fprintf(stderr, "Exception in PubSubNetworkListener: %s\n", e.what()); span->SetAttribute("error", e.what()); span->SetStatus(opentelemetry::trace::StatusCode::kError, e.what()); return; } } void PubSubNetworkListener::listenThread() { if (_listener) { while (_run) { rustybits::network_listener_listen(_listener); } } } void PubSubNetworkListener::changeHandlerThread() { if (_listener) { rustybits::network_listener_change_handler(_listener); } } PubSubMemberListener::PubSubMemberListener(std::string controller_id, uint64_t listen_timeout, DB* db) : _run(true), _controller_id(controller_id), _db(db), _listener(nullptr) { _run = true; _listener = rustybits::member_listener_new(_controller_id.c_str(), listen_timeout, listener_callback, this); _listenThread = std::thread(&PubSubMemberListener::listenThread, this); _changeHandlerThread = std::thread(&PubSubMemberListener::changeHandlerThread, this); } PubSubMemberListener::~PubSubMemberListener() { _run = false; if (_listenThread.joinable()) { _listenThread.join(); } if (_listener) { rustybits::member_listener_delete(_listener); _listener = nullptr; } } void PubSubMemberListener::onNotification(const std::string& payload) { auto provider = opentelemetry::trace::Provider::GetTracerProvider(); auto tracer = provider->GetTracer("PubSubMemberListener"); auto span = tracer->StartSpan("PubSubMemberListener::onNotification"); auto scope = tracer->WithActiveSpan(span); span->SetAttribute("payload", payload); fprintf(stderr, "Member notification received: %s\n", payload.c_str()); try { nlohmann::json tmp; nlohmann::json old_tmp = tmp["old"]; nlohmann::json new_tmp = tmp["new"]; nlohmann::json oldConfig, newConfig; if (old_tmp.is_object()) { // TODO: copy old configuration to oldConfig } if (new_tmp.is_object()) { // TODO: copy new configuration to newConfig } if (oldConfig.is_object() && newConfig.is_object()) { // member modification std::string memberID = oldConfig["id"].get(); std::string networkID = oldConfig["nwid"].get(); span->SetAttribute("action", "member_change"); span->SetAttribute("member_id", memberID); span->SetAttribute("network_id", networkID); _db->save(newConfig, _db->isReady()); } else if (newConfig.is_object() && ! oldConfig.is_object()) { // new member std::string memberID = newConfig["id"].get(); std::string networkID = newConfig["nwid"].get(); span->SetAttribute("action", "new_member"); span->SetAttribute("member_id", memberID); span->SetAttribute("network_id", networkID); _db->save(newConfig, _db->isReady()); } else if (! newConfig.is_object() && oldConfig.is_object()) { // member deletion std::string memberID = oldConfig["id"].get(); std::string networkID = oldConfig["nwid"].get(); span->SetAttribute("action", "delete_member"); span->SetAttribute("member_id", memberID); span->SetAttribute("network_id", networkID); uint64_t networkId = Utils::hexStrToU64(networkID.c_str()); uint64_t memberId = Utils::hexStrToU64(memberID.c_str()); if (networkId && memberId) { _db->eraseMember(networkId, memberId); } } } catch (const nlohmann::json::parse_error& e) { fprintf(stderr, "JSON parse error: %s\n", e.what()); span->SetAttribute("error", e.what()); span->SetStatus(opentelemetry::trace::StatusCode::kError, e.what()); return; } catch (const std::exception& e) { fprintf(stderr, "Exception in PubSubMemberListener: %s\n", e.what()); span->SetAttribute("error", e.what()); span->SetStatus(opentelemetry::trace::StatusCode::kError, e.what()); return; } } void PubSubMemberListener::listenThread() { if (_listener) { while (_run) { rustybits::member_listener_listen(_listener); } } } void PubSubMemberListener::changeHandlerThread() { if (_listener) { rustybits::member_listener_change_handler(_listener); } } } // namespace ZeroTier #endif // ZT_CONTROLLER_USE_LIBPQ