#include "dispatcherserver.h"

#include "piliterals_time.h"
#include "piscreentiles.h"


DispatcherServer::DispatcherServer(PINetworkAddress addr): eth(PIEthernet::TCP_Server) {
	client_gid      = 0;
	max_connections = 1000;
	eth.setParameter(PIEthernet::ReuseAddress);
	eth.setReadAddress(addr);
	//	eth.setDebug(false);
	CONNECTU(&eth, newConnection, this, newConnection);
	CONNECTU(&timeout_timer, tickEvent, this, cleanClients);
}


DispatcherServer::~DispatcherServer() {
	// stop the drainer and watchdog first: they act on the maps below
	watchdog.stop();
	timeout_timer.stop();
	PIVector<DispatcherClient *> all;
	PIVector<CloudServer *> srvs;
	{
		PIMutexLocker locker(map_mutex);
		all = clients;
		// rmrf_* sets hold clients/servers already removed from the main maps
		for (auto c: rmrf_clients)
			all << c;
		auto mi = c_servers.makeIterator();
		while (mi.next())
			srvs << mi.value();
		for (auto s: rmrf_servers)
			srvs << s;
		clients.clear();
		rm_clients.clear();
		rmrf_clients.clear();
		rmrf_servers.clear();
		c_servers.clear();
		index_c_servers.clear();
		index_c_clients.clear();
	}
	// run here (on this thread), not in read thread context: safe to wait
	for (auto c: all) {
		c->stop();
		c->close();
	}
	for (auto c: all) {
		if (c->isPIObject()) delete c;
	}
	for (auto s: srvs)
		delete s;
	eth.close();
	// piCoutObj << "server stoped";
}


void DispatcherServer::start() {
	eth.listen(true);
	timeout_timer.start(0.5_Hz);
	piCoutObj << "server started" << eth.readAddress();
}


void DispatcherServer::picoutStatus() {
	PIMutexLocker locker(map_mutex);
	piCout << PICoutManipulators::NewLine;
	piCout << "Connections:";
	for (auto c: clients) {
		piCout << "    " << c->address();
	}
	piCout << "Servers:";
	auto it = c_servers.makeIterator();
	while (it.next()) {
		piCout << "    " << it.key();
		it.value()->printStatus();
	}
}


void DispatcherServer::cleanClients() {
	wd_cnt++;
	PIVector<DispatcherClient *> to_close;
	PIVector<CloudServer *> to_unlink;
	{
		PIMutexLocker locker(map_mutex);
		for (auto s: rmrf_servers)
			to_unlink << s;
		rmrf_servers.clear();
		for (auto c: rmrf_clients)
			to_close << c;
		rmrf_clients.clear();
		for (auto c: clients) {
			if (!index_c_servers.contains(c) && !index_c_clients.contains(c)) {
				if (!rm_clients.contains(c)) rm_clients << c;
			} else
				rm_clients.remove(c);
		}
		auto ss = c_servers.values();
		for (auto c: ss) {
			if (c->lastPing() > 15.0) {
				piCout << "remove Server by ping timeout" << c->getConnection()->clientId();
				PIVector<DispatcherClient *> cscv = c->getClients();
				for (auto csc: cscv) {
					if (!csc->isPIObject()) {
						piCout << "ACHTUNG! Non-piobject DispatcherClient!";
						continue;
					}
					clients.removeAll(csc);
					index_c_clients.remove(csc);
					c->removeClient(csc);
					rmrf_clients << csc;
				}
				c_servers.remove(c->serverUUID());
				index_c_servers.remove(c->getConnection());
				rmrf_clients << const_cast<DispatcherClient *>(c->getConnection());
				rmrf_servers << c;
			}
		}
		for (auto c: rm_clients) {
			if (clients.contains(c)) {
				rmrf_clients << c;
			}
		}
		for (auto c: rmrf_clients) {
			clients.removeAll(c);
			if (index_c_servers.contains(c)) {
				c_servers.remove(c_servers.key(index_c_servers[c]));
				index_c_servers.remove(c);
			}
			index_c_clients.remove(c);
			rm_clients.remove(c);
			to_close << c;
		}
		rmrf_clients.clear();
	}
	// map_mutex must NOT be held here: read threads of other clients wait for it
	// (registerServer/registerClient/disconnectClient), so any blocking
	// close()/stopAndWait() under the lock was a guaranteed deadlock
	for (auto c: to_close) {
		c->stop();
		c->close();
	}
	for (auto c: to_close) {
		if (!c->isPIObject()) {
			piCout << "ACHTUNG! Non-piobject client!";
			continue;
		}
		delete c;
	}
	// do NOT call s->close() here: it dereferences the server connection,
	// which is already closed and deleted in the to_close loop above;
	// ~CloudServer does not touch the raw pointers it holds
	for (auto s: to_unlink)
		delete s;
}


void DispatcherServer::updateConnectionsTile(TileList * tl) {
	PIMutexLocker locker(map_mutex);
	tl->content.clear();
	for (auto c: clients) {
		PIString role = "Invalid";
		switch (c->role()) {
		case PICloud::TCP::Client: {
			role             = "Client";
			CloudServer * cs = index_c_clients.value(c, nullptr);
			if (cs) role += " \"" + cs->serverUUID().toHex().left(8) + "...\"";
		} break;
		case PICloud::TCP::Server: {
			role             = "Server";
			CloudServer * cs = index_c_servers.value(c, nullptr);
			if (cs) role += " \"" + cs->serverUUID().toHex().left(8) + "...\"";
		} break;
		default: break;
		}
		tl->content << TileList::Row(c->address() + " " + role, PIScreenTypes::CellFormat());
	}
	for (auto c: rm_clients) {
		tl->content << TileList::Row("[deleting]" + c->address(), PIScreenTypes::CellFormat());
	}
	for (auto c: rmrf_clients) {
		tl->content << TileList::Row("[NULL]" + c->address(), PIScreenTypes::CellFormat());
	}
}


void DispatcherServer::updateServersTile(TileList * tl, PISet<const DispatcherClient *> servers) {
	PIMutexLocker locker(map_mutex);
	tl->content.clear();
	auto mi = c_servers.makeIterator();
	while (mi.next()) {
		tl->content << TileList::Row(mi.value()->serverUUID().toHex().left(8) + "... - " +
		                                 PIString::fromNumber(mi.value()->getClients().size()),
		                             PIScreenTypes::CellFormat());
		if (servers.contains(mi.value()->getConnection())) tl->selected << (tl->content.size_s() - 1);
	}
}


void DispatcherServer::updateClientsTile(TileList * tl, PISet<const DispatcherClient *> servers) {
	PIMutexLocker locker(map_mutex);
	tl->content.clear();
	auto mi = c_servers.makeIterator();
	while (mi.next()) {
		for (auto c: mi.value()->getClients()) {
			if (servers.isEmpty() || servers.contains(mi.value()->getConnection()))
				tl->content << TileList::Row(c->address(), PIScreenTypes::CellFormat());
		}
	}
}


const DispatcherClient * DispatcherServer::getConnection(int index) {
	PIMutexLocker locker(map_mutex);
	const DispatcherClient * ret = nullptr;
	if (index >= 0 && index < clients.size_s()) ret = clients[index];
	return ret;
}


const DispatcherClient * DispatcherServer::getServer(int index) {
	PIMutexLocker locker(map_mutex);
	const DispatcherClient * ret = nullptr;
	if (index >= 0 && index < clients.size_s()) {
		if (index_c_servers.contains(clients[index])) ret = clients[index];
	}
	return ret;
}


PISet<const DispatcherClient *> DispatcherServer::getServers(PISet<int> ids) {
	PISet<const DispatcherClient *> ret;
	if (ids.isEmpty()) return ret;
	PIMutexLocker locker(map_mutex);
	int i   = 0;
	auto mi = c_servers.makeIterator();
	while (mi.next()) {
		if (ids.contains(i)) ret << mi.value()->getConnection();
		i++;
	}
	return ret;
}


void DispatcherServer::setMaxConnections(uint max_count) {
	PIMutexLocker locker(map_mutex);
	max_connections = max_count;
}


void DispatcherServer::startWatchdog() {
	wd_cnt++;
	watchdog.start(
		[this]() {
			if (wd_cnt == 0) {
				piCout << "Deadlock detected, abort";
				std::abort();
				return;
			}
			wd_cnt = 0;
		},
		10_s);
}


void DispatcherServer::disconnectClient(DispatcherClient * client) {
	PIVector<DispatcherClient *> stop_now;
	{
		PIMutexLocker locker(map_mutex);
		if (!clients.contains(client)) {
			// piCoutObj << "INVALID client" << client;
			return;
		}
		piCoutObj << "remove ..." << client->clientId();
		client->connect_link.disconnect();
		clients.removeAll(client);
		rm_clients.remove(client);
		CloudServer * cs = index_c_servers.value(client, nullptr);
		if (cs) {
			piCoutObj << "remove Server" << client->clientId();
			PIVector<DispatcherClient *> cscv = cs->getClients();
			for (auto csc: cscv) {
				clients.removeAll(csc);
				index_c_clients.remove(csc);
				rmrf_clients << csc;
				stop_now << csc;
			}
			c_servers.remove(cs->serverUUID());
			index_c_servers.remove(client);
			rmrf_servers << cs;
		}
		CloudServer * cc = index_c_clients.value(client, nullptr);
		if (cc) {
			piCoutObj << "remove Client" << client->clientId();
			cc->removeClient(client);
			index_c_clients.remove(client);
		}
		rmrf_clients << client;
		piCoutObj << "remove done" << client->clientId();
	}
	// this is usually called from the client's own read thread, so blocking
	// close()/stopAndWait() here would wait for the calling thread itself;
	// only request stop, the drainer (cleanClients) waits and deletes
	for (auto c: stop_now)
		c->stop();
	client->stop();
}


void DispatcherServer::newConnection(PIEthernet * cl) {
	// piCout << "DispatcherServer::newConnection" << (void *)cl;
	DispatcherClient * client = nullptr;
	{
		PIMutexLocker locker(map_mutex);
		if (clients.size() >= max_connections) {
			// piCout << "DispatcherServer::newConnection overflow" << (void *)cl;
			delete cl;
			return;
		}
		client               = new DispatcherClient(cl, client_gid++);
		client->connect_link = CONNECTU(client, disconnectEvent, this, disconnectClient);
		CONNECTL(client, registerServer, [this](const PIByteArray & sname, DispatcherClient * c) {
			PIMutexLocker locker(map_mutex);
			CloudServer * cs = c_servers.value(sname, nullptr);
			if (cs) {
				rm_clients << c;
				piCoutObj << "duplicate Server ->" << sname.toHex();
			} else {
				piCoutObj << "add new Server ->" << sname.toHex();
				CloudServer * cs = new CloudServer(c, sname);
				c_servers.insert(sname, cs);
				index_c_servers.insert(c, cs);
				c->authorise(true);
			}
		});
		CONNECTL(client, registerClient, [this](const PIByteArray & sname, DispatcherClient * c) {
			PIMutexLocker locker(map_mutex);
			CloudServer * cs = c_servers.value(sname, nullptr);
			if (cs) {
				piCoutObj << "add new Client to Server ->" << sname.toHex();
				c->authorise(true);
				cs->addClient(c);
				index_c_clients.insert(c, cs);
			} else {
				rm_clients << c;
				piCoutObj << "Client can't connect to Server ->" << sname.toHex();
			}
		});
		// piCoutObj << "add client" << client;
		clients.push_back(client);
		// start the read thread while the lock is held: a client that is in
		// clients must already have its read thread running, otherwise the
		// drainer could close and delete it before start() executes; the
		// thread can only wait for map_mutex after this function releases it
		client->start();
	}
	// piCout << "DispatcherServer::newConnection started" << (void *)cl;
}