Refactor Redis & Posgres notification listeners into listener subclass in new CentralDB class

This allows us to interchangeably use different listeners (pgsql, redis, pubsub) depending on configuration values passed into the constructor.

controller/CV1.hpp

@@ -64,13 +64,6 @@ class CV1 : public DB {
 		return _ready == 2;
 	}
 
-  protected:
-	struct _PairHasher {
-		inline std::size_t operator()(const std::pair<uint64_t, uint64_t>& p) const
-		{
-			return (std::size_t)(p.first ^ p.second);
-		}
-	};
 	virtual void _memberChanged(nlohmann::json& old, nlohmann::json& memberConfig, bool notifyListeners)
 	{
 		DB::_memberChanged(old, memberConfig, notifyListeners);
@@ -81,6 +74,14 @@ class CV1 : public DB {
 		DB::_networkChanged(old, networkConfig, notifyListeners);
 	}
 
+  protected:
+	struct _PairHasher {
+		inline std::size_t operator()(const std::pair<uint64_t, uint64_t>& p) const
+		{
+			return (std::size_t)(p.first ^ p.second);
+		}
+	};
+
   private:
 	void initializeNetworks();
 	void initializeMembers();

controller/CV2.hpp

@@ -52,6 +52,10 @@ class CV2 : public DB {
 		return _ready == 2;
 	}
 
+	virtual void _memberChanged(nlohmann::json& old, nlohmann::json& memberConfig, bool notifyListeners);
+
+	virtual void _networkChanged(nlohmann::json& old, nlohmann::json& networkConfig, bool notifyListeners);
+
   protected:
 	struct _PairHasher {
 		inline std::size_t operator()(const std::pair<uint64_t, uint64_t>& p) const
@@ -59,9 +63,6 @@ class CV2 : public DB {
 			return (std::size_t)(p.first ^ p.second);
 		}
 	};
-	virtual void _memberChanged(nlohmann::json& old, nlohmann::json& memberConfig, bool notifyListeners);
-
-	virtual void _networkChanged(nlohmann::json& old, nlohmann::json& networkConfig, bool notifyListeners);
 
   private:
 	void initializeNetworks();

controller/CentralDB.hpp

@@ -0,0 +1,130 @@
+#ifdef ZT_CONTROLLER_USE_LIBPQ
+
+#ifndef ZT_CONTROLLER_CENTRAL_DB_HPP
+#define ZT_CONTROLLER_CENTRAL_DB_HPP
+
+#define ZT_CENTRAL_CONTROLLER_COMMIT_THREADS 4
+
+#include "../node/Metrics.hpp"
+#include "ConnectionPool.hpp"
+#include "DB.hpp"
+#include "NotificationListener.hpp"
+#include "PostgreSQL.hpp"
+
+#include <memory>
+#include <pqxx/pqxx>
+#include <redis++/redis++.h>
+
+namespace rustybits {
+struct SmeeClient;
+}
+
+namespace ZeroTier {
+
+struct RedisConfig;
+struct PubSubConfig;
+struct PostgresNotifyConfig;
+
+struct ControllerConfig {
+	RedisConfig* redisConfig;
+	PubSubConfig* pubSubConfig;
+	PostgresNotifyConfig* postgresNotifyConfig;
+};
+
+class CentralDB : public DB {
+  public:
+	enum ListenerMode {
+		LISTENER_MODE_PGSQL = 0,
+		LISTENER_MODE_REDIS = 1,
+		LISTENER_MODE_PUBSUB = 2,
+	};
+
+	CentralDB(const Identity& myId, const char* path, int listenPort, CentralDB::ListenerMode mode, ControllerConfig* cc);
+	virtual ~CentralDB();
+
+	virtual bool waitForReady();
+	virtual bool isReady();
+	virtual bool save(nlohmann::json& record, bool notifyListeners);
+	virtual void eraseNetwork(const uint64_t networkId);
+	virtual void eraseMember(const uint64_t networkId, const uint64_t memberId);
+	virtual void nodeIsOnline(const uint64_t networkId, const uint64_t memberId, const InetAddress& physicalAddress);
+	virtual void nodeIsOnline(const uint64_t networkId, const uint64_t memberId, const InetAddress& physicalAddress, const char* osArch);
+	virtual AuthInfo getSSOAuthInfo(const nlohmann::json& member, const std::string& redirectURL);
+
+	virtual bool ready()
+	{
+		return _ready == 2;
+	}
+
+	virtual void _memberChanged(nlohmann::json& old, nlohmann::json& memberConfig, bool notifyListeners)
+	{
+		DB::_memberChanged(old, memberConfig, notifyListeners);
+	}
+
+	virtual void _networkChanged(nlohmann::json& old, nlohmann::json& networkConfig, bool notifyListeners)
+	{
+		DB::_networkChanged(old, networkConfig, notifyListeners);
+	}
+
+  protected:
+	struct _PairHasher {
+		inline std::size_t operator()(const std::pair<uint64_t, uint64_t>& p) const
+		{
+			return (std::size_t)(p.first ^ p.second);
+		}
+	};
+
+  private:
+	void initializeNetworks();
+	void initializeMembers();
+	void heartbeat();
+
+	void commitThread();
+	void onlineNotificationThread();
+	void onlineNotification_Postgres();
+	void onlineNotification_Redis();
+	uint64_t _doRedisUpdate(sw::redis::Transaction& tx, std::string& controllerId, std::unordered_map<std::pair<uint64_t, uint64_t>, NodeOnlineRecord, _PairHasher>& lastOnline);
+
+	void configureSmee();
+	void notifyNewMember(const std::string& networkID, const std::string& memberID);
+
+	enum OverrideMode { ALLOW_PGBOUNCER_OVERRIDE = 0, NO_OVERRIDE = 1 };
+
+	ListenerMode _listenerMode;
+	std::shared_ptr<ConnectionPool<PostgresConnection> > _pool;
+
+	const Identity _myId;
+	const Address _myAddress;
+	std::string _myAddressStr;
+	std::string _connString;
+
+	BlockingQueue<std::pair<nlohmann::json, bool> > _commitQueue;
+
+	std::thread _heartbeatThread;
+	std::shared_ptr<NotificationListener> _membersDbWatcher;
+	std::shared_ptr<NotificationListener> _networksDbWatcher;
+	std::thread _commitThread[ZT_CENTRAL_CONTROLLER_COMMIT_THREADS];
+	std::thread _onlineNotificationThread;
+
+	std::unordered_map<std::pair<uint64_t, uint64_t>, NodeOnlineRecord, _PairHasher> _lastOnline;
+
+	mutable std::mutex _lastOnline_l;
+	mutable std::mutex _readyLock;
+	std::atomic<int> _ready, _connected, _run;
+	mutable volatile bool _waitNoticePrinted;
+
+	int _listenPort;
+	uint8_t _ssoPsk[48];
+
+	RedisConfig* _rc;
+	std::shared_ptr<sw::redis::Redis> _redis;
+	std::shared_ptr<sw::redis::RedisCluster> _cluster;
+	bool _redisMemberStatus;
+
+	rustybits::SmeeClient* _smee;
+};
+
+}	// namespace ZeroTier
+
+#endif	 // ZT_CONTROLLER_CENTRAL_DB_HPP
+#endif	 // ZT_CONTROLLER_USE_LIBPQ

controller/DB.hpp

@@ -145,6 +145,9 @@ class DB {
 		_changeListeners.push_back(listener);
 	}
 
+	virtual void _memberChanged(nlohmann::json& old, nlohmann::json& memberConfig, bool notifyListeners);
+	virtual void _networkChanged(nlohmann::json& old, nlohmann::json& networkConfig, bool notifyListeners);
+
   protected:
 	static inline bool _compareRecords(const nlohmann::json& a, const nlohmann::json& b)
 	{
@@ -181,8 +184,6 @@ class DB {
 		std::shared_mutex lock;
 	};
 
-	virtual void _memberChanged(nlohmann::json& old, nlohmann::json& memberConfig, bool notifyListeners);
-	virtual void _networkChanged(nlohmann::json& old, nlohmann::json& networkConfig, bool notifyListeners);
 	void _fillSummaryInfo(const std::shared_ptr<_Network>& nw, NetworkSummaryInfo& info);
 
 	std::vector<DB::ChangeListener*> _changeListeners;

controller/NotificationListener.hpp

@@ -0,0 +1,32 @@
+#ifndef NOTIFICATION_LISTENER_HPP
+#define NOTIFICATION_LISTENER_HPP
+
+#include <string>
+
+namespace ZeroTier {
+
+/**
+ * Base class for notification listeners
+ *
+ * This class is used to receive notifications from various sources such as Redis, PostgreSQL, etc.
+ */
+class NotificationListener {
+  public:
+	NotificationListener() = default;
+	virtual ~NotificationListener()
+	{
+	}
+
+	/**
+	 * Called when a notification is received.
+	 *
+	 * Payload should be parsed and passed to the database handler's save method.
+	 *
+	 * @param payload The payload of the notification.
+	 */
+	virtual void onNotification(const std::string& payload) = 0;
+};
+
+}	// namespace ZeroTier
+
+#endif	 // NOTIFICATION_LISTENER_HPP

controller/PostgreSQL.cpp

@@ -2,10 +2,168 @@
 
 #include "PostgreSQL.hpp"
 
+#include "opentelemetry/trace/provider.h"
+
 #include <nlohmann/json.hpp>
 
-using namespace nlohmann;
+namespace ZeroTier {
 
-using namespace ZeroTier;
+PostgresMemberListener::PostgresMemberListener(DB* db, std::shared_ptr<ConnectionPool<PostgresConnection> > pool, const std::string& channel, uint64_t timeout)
+	: NotificationListener()
+	, _db(db)
+	, _pool(pool)
+	, _notification_timeout(timeout)
+	, _listenerThread()
+{
+	_conn = _pool->borrow();
+	_receiver = new _notificationReceiver<PostgresMemberListener>(this, *_conn->c, channel);
+	_run = true;
+	_listenerThread = std::thread(&PostgresMemberListener::listen, this);
+}
+
+PostgresMemberListener::~PostgresMemberListener()
+{
+	_run = false;
+	if (_listenerThread.joinable()) {
+		_listenerThread.join();
+	}
+	delete _receiver;
+	if (_conn) {
+		_pool->unborrow(_conn);
+		_conn.reset();
+	}
+}
+
+void PostgresMemberListener::listen()
+{
+	while (_run) {
+		_conn->c->await_notification(_notification_timeout, 0);
+	}
+}
+
+void PostgresMemberListener::onNotification(const std::string& payload)
+{
+	auto provider = opentelemetry::trace::Provider::GetTracerProvider();
+	auto tracer = provider->GetTracer("PostgresMemberNotificationListener");
+	auto span = tracer->StartSpan("PostgresMemberNotificationListener::onNotification");
+	auto scope = tracer->WithActiveSpan(span);
+	span->SetAttribute("payload", payload);
+
+	fprintf(stderr, "Member Notification received: %s\n", payload.c_str());
+	Metrics::pgsql_mem_notification++;
+	nlohmann::json tmp(nlohmann::json::parse(payload));
+	nlohmann::json& ov = tmp["old_val"];
+	nlohmann::json& nv = tmp["new_val"];
+	nlohmann::json oldConfig, newConfig;
+	if (ov.is_object())
+		oldConfig = ov;
+	if (nv.is_object())
+		newConfig = nv;
+
+	if (oldConfig.is_object() && newConfig.is_object()) {
+		_db->save(newConfig, true);
+		fprintf(stderr, "payload sent\n");
+	}
+	else if (newConfig.is_object() && ! oldConfig.is_object()) {
+		// new member
+		Metrics::member_count++;
+		_db->save(newConfig, true);
+		fprintf(stderr, "new member payload sent\n");
+	}
+	else if (! newConfig.is_object() && oldConfig.is_object()) {
+		// member delete
+		uint64_t networkId = OSUtils::jsonIntHex(oldConfig["nwid"], 0ULL);
+		uint64_t memberId = OSUtils::jsonIntHex(oldConfig["id"], 0ULL);
+		if (memberId && networkId) {
+			_db->eraseMember(networkId, memberId);
+			fprintf(stderr, "member delete payload sent\n");
+		}
+	}
+}
+
+PostgresNetworkListener::PostgresNetworkListener(DB* db, std::shared_ptr<ConnectionPool<PostgresConnection> > pool, const std::string& channel, uint64_t timeout)
+	: NotificationListener()
+	, _db(db)
+	, _pool(pool)
+	, _notification_timeout(timeout)
+	, _listenerThread()
+{
+	_conn = _pool->borrow();
+	_receiver = new _notificationReceiver<PostgresNetworkListener>(this, *_conn->c, channel);
+	_run = true;
+	_listenerThread = std::thread(&PostgresNetworkListener::listen, this);
+}
+
+PostgresNetworkListener::~PostgresNetworkListener()
+{
+	_run = false;
+	if (_listenerThread.joinable()) {
+		_listenerThread.join();
+	}
+	delete _receiver;
+	if (_conn) {
+		_pool->unborrow(_conn);
+		_conn.reset();
+	}
+}
+
+void PostgresNetworkListener::listen()
+{
+	while (_run) {
+		_conn->c->await_notification(_notification_timeout, 0);
+	}
+}
+
+void PostgresNetworkListener::onNotification(const std::string& payload)
+{
+	auto provider = opentelemetry::trace::Provider::GetTracerProvider();
+	auto tracer = provider->GetTracer("db_network_notification");
+	auto span = tracer->StartSpan("db_network_notification::operator()");
+	auto scope = tracer->WithActiveSpan(span);
+	span->SetAttribute("payload", payload);
+
+	fprintf(stderr, "Network Notification received: %s\n", payload.c_str());
+	Metrics::pgsql_net_notification++;
+	nlohmann::json tmp(nlohmann::json::parse(payload));
+
+	nlohmann::json& ov = tmp["old_val"];
+	nlohmann::json& nv = tmp["new_val"];
+	nlohmann::json oldConfig, newConfig;
+
+	if (ov.is_object())
+		oldConfig = ov;
+	if (nv.is_object())
+		newConfig = nv;
+
+	if (oldConfig.is_object() && newConfig.is_object()) {
+		std::string nwid = oldConfig["id"];
+		span->SetAttribute("action", "network_change");
+		span->SetAttribute("network_id", nwid);
+		_db->save(newConfig, true);
+		fprintf(stderr, "payload sent\n");
+	}
+	else if (newConfig.is_object() && ! oldConfig.is_object()) {
+		std::string nwid = newConfig["id"];
+		span->SetAttribute("network_id", nwid);
+		span->SetAttribute("action", "new_network");
+		// new network
+		_db->save(newConfig, true);
+		fprintf(stderr, "new network payload sent\n");
+	}
+	else if (! newConfig.is_object() && oldConfig.is_object()) {
+		// network delete
+		span->SetAttribute("action", "delete_network");
+		std::string nwid = oldConfig["id"];
+		span->SetAttribute("network_id", nwid);
+		uint64_t networkId = Utils::hexStrToU64(nwid.c_str());
+		span->SetAttribute("network_id_int", networkId);
+		if (networkId) {
+			_db->eraseNetwork(networkId);
+			fprintf(stderr, "network delete payload sent\n");
+		}
+	}
+}
+
+}	// namespace ZeroTier
 
 #endif

controller/PostgreSQL.hpp

@@ -18,6 +18,7 @@
 
 #include "ConnectionPool.hpp"
 #include "DB.hpp"
+#include "NotificationListener.hpp"
 #include "opentelemetry/trace/provider.h"
 
 #include <memory>
@@ -188,6 +189,67 @@ struct NodeOnlineRecord {
 	std::string osArch;
 };
 
+/**
+ * internal class for listening to PostgreSQL notification channels.
+ */
+template <typename T> class _notificationReceiver : public pqxx::notification_receiver {
+  public:
+	_notificationReceiver(T* p, pqxx::connection& c, const std::string& channel) : pqxx::notification_receiver(c, channel), _listener(p)
+	{
+		fprintf(stderr, "initialize PostgresMemberNotificationListener::_notificationReceiver\n");
+	}
+
+	virtual void operator()(const std::string& payload, int backendPid)
+	{
+		auto provider = opentelemetry::trace::Provider::GetTracerProvider();
+		auto tracer = provider->GetTracer("notification_receiver");
+		auto span = tracer->StartSpan("notification_receiver::operator()");
+		auto scope = tracer->WithActiveSpan(span);
+		_listener->onNotification(payload);
+	}
+
+  private:
+	T* _listener;
+};
+
+class PostgresMemberListener : public NotificationListener {
+  public:
+	PostgresMemberListener(DB* db, std::shared_ptr<ConnectionPool<PostgresConnection> > pool, const std::string& channel, uint64_t timeout);
+	virtual ~PostgresMemberListener();
+
+	virtual void listen();
+
+	virtual void onNotification(const std::string& payload) override;
+
+  private:
+	bool _run = false;
+	DB* _db;
+	std::shared_ptr<ConnectionPool<PostgresConnection> > _pool;
+	std::shared_ptr<PostgresConnection> _conn;
+	uint64_t _notification_timeout;
+	std::thread _listenerThread;
+	_notificationReceiver<PostgresMemberListener>* _receiver;
+};
+
+class PostgresNetworkListener : public NotificationListener {
+  public:
+	PostgresNetworkListener(DB* db, std::shared_ptr<ConnectionPool<PostgresConnection> > pool, const std::string& channel, uint64_t timeout);
+	virtual ~PostgresNetworkListener();
+
+	virtual void listen();
+
+	virtual void onNotification(const std::string& payload) override;
+
+  private:
+	bool _run = false;
+	DB* _db;
+	std::shared_ptr<ConnectionPool<PostgresConnection> > _pool;
+	std::shared_ptr<PostgresConnection> _conn;
+	uint64_t _notification_timeout;
+	std::thread _listenerThread;
+	_notificationReceiver<PostgresNetworkListener>* _receiver;
+};
+
 }	// namespace ZeroTier
 
 #endif	 // ZT_CONTROLLER_POSTGRESQL_HPP

controller/PubSubListener.cpp

@@ -1,8 +1,12 @@
 #ifdef ZT_CONTROLLER_USE_LIBPQ
 #include "PubSubListener.hpp"
 
+#include "DB.hpp"
+#include "opentelemetry/trace/provider.h"
 #include "rustybits.h"
 
+#include <nlohmann/json.hpp>
+
 namespace ZeroTier {
 
 void listener_callback(void* user_ptr, const uint8_t* payload, uintptr_t length)
@@ -17,15 +21,20 @@ void listener_callback(void* user_ptr, const uint8_t* payload, uintptr_t length)
 	listener->onNotification(payload_str);
 }
 
-PubSubNetworkListener::PubSubNetworkListener(const char* controller_id, uint64_t listen_timeout, rustybits::NetworkListenerCallback callback) : _listener(nullptr)
+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, listen_timeout, callback, this);
+	_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;
@@ -34,15 +43,74 @@ PubSubNetworkListener::~PubSubNetworkListener()
 
 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());
-	// TODO: Implement the logic to handle network notifications
+
+	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<std::string>();
+			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 (rustybits::network_listener_listen(_listener)) {
-			// just keep looping
+		while (_run) {
+			rustybits::network_listener_listen(_listener);
 		}
 	}
 }
@@ -54,17 +122,21 @@ void PubSubNetworkListener::changeHandlerThread()
 	}
 }
 
-PubSubMemberListener::PubSubMemberListener(const char* controller_id, uint64_t listen_timeout, rustybits::NetworkListenerCallback callback) : _listener(nullptr)
+PubSubMemberListener::PubSubMemberListener(std::string controller_id, uint64_t listen_timeout, DB* db) : _run(true), _controller_id(controller_id), _db(db), _listener(nullptr)
 {
-	// Initialize the member listener with the provided controller ID and timeout
-	// The callback will be called when a member notification is received
-	{
-		_listener = rustybits::member_listener_new("controller_id", 60, listener_callback, this);
-	}
+	_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;
@@ -73,16 +145,80 @@ PubSubMemberListener::~PubSubMemberListener()
 
 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());
 
-	// TODO: Implement the logic to handle network notifications
+	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>();
+			std::string networkID = oldConfig["nwid"].get<std::string>();
+			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>();
+			std::string networkID = newConfig["nwid"].get<std::string>();
+			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>();
+			std::string networkID = oldConfig["nwid"].get<std::string>();
+			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 (rustybits::member_listener_listen(_listener)) {
-			// just keep looping
+		while (_run) {
+			rustybits::member_listener_listen(_listener);
 		}
 	}
 }

controller/PubSubListener.hpp

@@ -3,6 +3,7 @@
 #ifndef ZT_CONTROLLER_PUBSUBLISTENER_HPP
 #define ZT_CONTROLLER_PUBSUBLISTENER_HPP
 
+#include "NotificationListener.hpp"
 #include "rustybits.h"
 
 #include <memory>
@@ -10,7 +11,17 @@
 #include <thread>
 
 namespace ZeroTier {
-class PubSubListener {
+class DB;
+
+struct PubSubConfig {
+	const char* controller_id;
+	uint64_t listen_timeout;
+};
+
+/**
+ * Base class for GCP PubSub listeners
+ */
+class PubSubListener : public NotificationListener {
   public:
 	virtual ~PubSubListener()
 	{
@@ -19,9 +30,12 @@ class PubSubListener {
 	virtual void onNotification(const std::string& payload) = 0;
 };
 
+/**
+ * Listener for network notifications via GCP PubSub
+ */
 class PubSubNetworkListener : public PubSubListener {
   public:
-	PubSubNetworkListener(const char* controller_id, uint64_t listen_timeout, rustybits::NetworkListenerCallback callback);
+	PubSubNetworkListener(std::string controller_id, uint64_t listen_timeout, DB* db);
 	virtual ~PubSubNetworkListener();
 
 	virtual void onNotification(const std::string& payload) override;
@@ -30,14 +44,20 @@ class PubSubNetworkListener : public PubSubListener {
 	void listenThread();
 	void changeHandlerThread();
 
+	bool _run = false;
+	std::string _controller_id;
+	DB* _db;
 	const rustybits::NetworkListener* _listener;
 	std::thread _listenThread;
 	std::thread _changeHandlerThread;
 };
 
+/**
+ * Listener for member notifications via GCP PubSub
+ */
 class PubSubMemberListener : public PubSubListener {
   public:
-	PubSubMemberListener(const char* controller_id, uint64_t listen_timeout, rustybits::MemberListenerCallback callback);
+	PubSubMemberListener(std::string controller_id, uint64_t listen_timeout, DB* db);
 	virtual ~PubSubMemberListener();
 
 	virtual void onNotification(const std::string& payload) override;
@@ -46,6 +66,9 @@ class PubSubMemberListener : public PubSubListener {
 	void listenThread();
 	void changeHandlerThread();
 
+	bool _run = false;
+	std::string _controller_id;
+	DB* _db;
 	const rustybits::MemberListener* _listener;
 	std::thread _listenThread;
 	std::thread _changeHandlerThread;

controller/RedisListener.cpp

@@ -0,0 +1,214 @@
+#ifdef ZT_CONTROLLER_USE_LIBPQ
+
+#include "RedisListener.hpp"
+
+#include "../node/Metrics.hpp"
+#include "nlohmann/json.hpp"
+#include "opentelemetry/trace/provider.h"
+
+#include <memory>
+#include <string>
+#include <vector>
+
+namespace ZeroTier {
+
+using Attrs = std::vector<std::pair<std::string, std::string> >;
+using Item = std::pair<std::string, Attrs>;
+using ItemStream = std::vector<Item>;
+
+RedisListener::RedisListener(std::string controller_id, std::shared_ptr<sw::redis::Redis> redis) : _controller_id(controller_id), _redis(redis), _is_cluster(false), _run(false)
+{
+}
+
+RedisListener::RedisListener(std::string controller_id, std::shared_ptr<sw::redis::RedisCluster> cluster) : _controller_id(controller_id), _cluster(cluster), _is_cluster(true), _run(false)
+{
+}
+
+RedisListener::~RedisListener()
+{
+	_run = false;
+	if (_listenThread.joinable()) {
+		_listenThread.join();
+	}
+}
+
+RedisNetworkListener::RedisNetworkListener(std::string controller_id, std::shared_ptr<sw::redis::Redis> redis, DB* db) : RedisListener(controller_id, redis), _db(db)
+{
+	// Additional initialization for network listener if needed
+}
+
+RedisNetworkListener::RedisNetworkListener(std::string controller_id, std::shared_ptr<sw::redis::RedisCluster> cluster, DB* db) : RedisListener(controller_id, cluster), _db(db)
+{
+	// Additional initialization for network listener if needed
+}
+
+RedisNetworkListener::~RedisNetworkListener()
+{
+	// Destructor logic if needed
+}
+
+void RedisNetworkListener::listen()
+{
+	std::string key = "network-stream:{" + _controller_id + "}";
+	std::string lastID = "0";
+	while (_run) {
+		auto provider = opentelemetry::trace::Provider::GetTracerProvider();
+		auto tracer = provider->GetTracer("RedisNetworkListener");
+		auto span = tracer->StartSpan("RedisNetworkListener::listen");
+		auto scope = tracer->WithActiveSpan(span);
+
+		try {
+			nlohmann::json tmp;
+			std::unordered_map<std::string, ItemStream> result;
+			if (_is_cluster) {
+				_cluster->xread(key, lastID, std::chrono::seconds(1), 0, std::inserter(result, result.end()));
+			}
+			else {
+				_redis->xread(key, lastID, std::chrono::seconds(1), 0, std::inserter(result, result.end()));
+			}
+
+			if (! result.empty()) {
+				for (auto element : result) {
+					for (auto rec : element.second) {
+						std::string id = rec.first;
+						auto attrs = rec.second;
+
+						for (auto a : attrs) {
+							try {
+								tmp = nlohmann::json::parse(a.second);
+								tmp = nlohmann::json::parse(a.second);
+								nlohmann::json& ov = tmp["old_val"];
+								nlohmann::json& nv = tmp["new_val"];
+								nlohmann::json oldConfig, newConfig;
+								if (ov.is_object())
+									oldConfig = ov;
+								if (nv.is_object())
+									newConfig = nv;
+								if (oldConfig.is_object() || newConfig.is_object()) {
+									_db->_networkChanged(oldConfig, newConfig, true);
+								}
+							}
+							catch (const nlohmann::json::parse_error& e) {
+								fprintf(stderr, "JSON parse error: %s\n", e.what());
+							}
+							catch (const std::exception& e) {
+								fprintf(stderr, "Exception in Redis network listener: %s\n", e.what());
+							}
+						}
+						if (_is_cluster) {
+							_cluster->xdel(key, id);
+						}
+						else {
+							_redis->xdel(key, id);
+						}
+						lastID = id;
+					}
+					Metrics::redis_net_notification++;
+				}
+			}
+		}
+		catch (sw::redis::Error& e) {
+			fprintf(stderr, "Error in Redis network listener: %s\n", e.what());
+		}
+		catch (const std::exception& e) {
+			fprintf(stderr, "Exception in Redis network listener: %s\n", e.what());
+		}
+	}
+}
+
+void RedisNetworkListener::onNotification(const std::string& payload)
+{
+	// Handle notifications if needed
+}
+
+RedisMemberListener::RedisMemberListener(std::string controller_id, std::shared_ptr<sw::redis::Redis> redis, DB* db) : RedisListener(controller_id, redis), _db(db)
+{
+	// Additional initialization for member listener if needed
+}
+
+RedisMemberListener::RedisMemberListener(std::string controller_id, std::shared_ptr<sw::redis::RedisCluster> cluster, DB* db) : RedisListener(controller_id, cluster), _db(db)
+{
+	// Additional initialization for member listener if needed
+}
+
+RedisMemberListener::~RedisMemberListener()
+{
+	// Destructor logic if needed
+}
+
+void RedisMemberListener::listen()
+{
+	std::string key = "member-stream:{" + _controller_id + "}";
+	std::string lastID = "0";
+	fprintf(stderr, "Listening to Redis member stream: %s\n", key.c_str());
+	while (_run) {
+		auto provider = opentelemetry::trace::Provider::GetTracerProvider();
+		auto tracer = provider->GetTracer("RedisMemberListener");
+		auto span = tracer->StartSpan("RedisMemberListener::listen");
+		auto scope = tracer->WithActiveSpan(span);
+
+		try {
+			nlohmann::json tmp;
+			std::unordered_map<std::string, ItemStream> result;
+			if (_is_cluster) {
+				_cluster->xread(key, lastID, std::chrono::seconds(1), 0, std::inserter(result, result.end()));
+			}
+			else {
+				_redis->xread(key, lastID, std::chrono::seconds(1), 0, std::inserter(result, result.end()));
+			}
+
+			if (! result.empty()) {
+				for (auto element : result) {
+					for (auto rec : element.second) {
+						std::string id = rec.first;
+						auto attrs = rec.second;
+
+						for (auto a : attrs) {
+							try {
+								tmp = nlohmann::json::parse(a.second);
+								nlohmann::json& ov = tmp["old_val"];
+								nlohmann::json& nv = tmp["new_val"];
+								nlohmann::json oldConfig, newConfig;
+								if (ov.is_object())
+									oldConfig = ov;
+								if (nv.is_object())
+									newConfig = nv;
+								if (oldConfig.is_object() || newConfig.is_object()) {
+									_db->_memberChanged(oldConfig, newConfig, true);
+								}
+							}
+							catch (const nlohmann::json::parse_error& e) {
+								fprintf(stderr, "JSON parse error: %s\n", e.what());
+							}
+							catch (const std::exception& e) {
+								fprintf(stderr, "Exception in Redis member listener: %s\n", e.what());
+							}
+						}
+						if (_is_cluster) {
+							_cluster->xdel(key, id);
+						}
+						else {
+							_redis->xdel(key, id);
+						}
+						lastID = id;
+					}
+					Metrics::redis_mem_notification++;
+				}
+			}
+		}
+		catch (sw::redis::Error& e) {
+			fprintf(stderr, "Error in Redis member listener: %s\n", e.what());
+		}
+		catch (const std::exception& e) {
+			fprintf(stderr, "Exception in Redis member listener: %s\n", e.what());
+		}
+	}
+}
+
+void RedisMemberListener::onNotification(const std::string& payload)
+{
+}
+
+}	// namespace ZeroTier
+
+#endif	 // ZT_CONTROLLER_USE_LIBPQ

controller/RedisListener.hpp

@@ -0,0 +1,76 @@
+#ifdef ZT_CONTROLLER_USE_LIBPQ
+
+#ifndef ZT_CONTROLLER_REDIS_LISTENER_HPP
+#define ZT_CONTROLLER_REDIS_LISTENER_HPP
+
+#include "DB.hpp"
+#include "NotificationListener.hpp"
+#include "Redis.hpp"
+
+#include <memory>
+#include <redis++/redis++.h>
+#include <string>
+#include <thread>
+
+namespace ZeroTier {
+
+class RedisListener : public NotificationListener {
+  public:
+	RedisListener(std::string controller_id, std::shared_ptr<sw::redis::Redis> redis);
+	RedisListener(std::string controller_id, std::shared_ptr<sw::redis::RedisCluster> cluster);
+
+	virtual ~RedisListener();
+
+	virtual void listen() = 0;
+	virtual void onNotification(const std::string& payload) override
+	{
+	}
+
+	void start()
+	{
+		_run = true;
+		_listenThread = std::thread(&RedisListener::listen, this);
+	}
+
+  protected:
+	std::string _controller_id;
+	std::shared_ptr<sw::redis::Redis> _redis;
+	std::shared_ptr<sw::redis::RedisCluster> _cluster;
+	bool _is_cluster = false;
+	bool _run = false;
+
+  private:
+	std::thread _listenThread;
+};
+
+class RedisNetworkListener : public RedisListener {
+  public:
+	RedisNetworkListener(std::string controller_id, std::shared_ptr<sw::redis::Redis> redis, DB* db);
+	RedisNetworkListener(std::string controller_id, std::shared_ptr<sw::redis::RedisCluster> cluster, DB* db);
+	virtual ~RedisNetworkListener();
+
+	virtual void listen() override;
+	virtual void onNotification(const std::string& payload) override;
+
+  private:
+	DB* _db;
+};
+
+class RedisMemberListener : public RedisListener {
+  public:
+	RedisMemberListener(std::string controller_id, std::shared_ptr<sw::redis::Redis> redis, DB* db);
+	RedisMemberListener(std::string controller_id, std::shared_ptr<sw::redis::RedisCluster> cluster, DB* db);
+	virtual ~RedisMemberListener();
+
+	virtual void listen() override;
+	virtual void onNotification(const std::string& payload) override;
+
+  private:
+	DB* _db;
+};
+
+}	// namespace ZeroTier
+
+#endif	 // ZT_CONTROLLER_REDIS_LISTENER_HPP
+
+#endif	 // ZT_CONTROLLER_USE_LIBPQ

objects.mk

@@ -42,7 +42,10 @@ ONE_OBJS=\
 	controller/CtlUtil.o \
 	controller/CV1.o \
 	controller/CV2.o \
+	controller/CentralDB.o \
 	controller/PubSubListener.o \
+	controller/RedisListener.o \
+	controller/PostgreSQL.o \
 	osdep/EthernetTap.o \
 	osdep/ManagedRoute.o \
 	osdep/Http.o \