flycast/core/network/ice.cpp

/*
	Copyright 2025 Flyinghead <[email protected]>

	This file is part of Flycast.

    Flycast is free software: you can redistribute it and/or modify
    it under the terms of the GNU General Public License as published by
    the Free Software Foundation, either version 2 of the License, or
    (at your option) any later version.

    Flycast is distributed in the hope that it will be useful,
    but WITHOUT ANY WARRANTY; without even the implied warranty of
    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
    GNU General Public License for more details.

    You should have received a copy of the GNU General Public License
    along with Flycast.  If not, see <https://www.gnu.org/licenses/>.
 */
#if !defined(LIBRETRO) || !defined(__ANDROID__)
#include "ice.h"
#include "types.h"
#include "oslib/http_client.h"
#include "json.hpp"
#include <future>

namespace ice
{

STUNConfig STUNConfig::get()
{
	STUNConfig cfg;
	cfg.stun_host = "lobby.flyca.st";
	cfg.stun_port = 25001;
	cfg.turn_host = cfg.stun_host;
	cfg.turn_port = cfg.stun_port;
	cfg.turn_username = "flycast";
	cfg.turn_password = "rules";

	http::init();
	std::vector<u8> content;
	std::string ctype;
	int status = http::get("https://dcnet.flyca.st/stun.json", content, ctype);
	if (http::success(status))
	{
		using namespace nlohmann;
		try {
			json v = json::parse((const char *)&content[0], (const char *)&content.back() + 1);
			cfg.stun_host = v.at("stun_host").get<std::string>();
			cfg.stun_port = v.at("stun_port").get<u16>();
			cfg.turn_host = v.at("turn_host").get<std::string>();
			cfg.turn_port = v.at("turn_port").get<u16>();
			cfg.turn_username = v.at("turn_username").get<std::string>();
			cfg.turn_password = v.at("turn_password").get<std::string>();
		} catch (const json::exception& e) {
			WARN_LOG(NETWORK, "Corrupted stun.json file: %s", e.what());
		}
	}
	else {
		WARN_LOG(NETWORK, "Can't download stun.json: error %d", status);
	}
	return cfg;
}
}
#endif // !LIBRETRO || !__ANDROID__

#if defined(USE_ICE) && defined(USE_WEBSOCKET)
#include "hw/sh4/modules/modules.h"
#include "util/tsqueue.h"
#include "oslib/oslib.h"
#include "emulator.h"
#include "log/LogManager.h"
#include "ui/gui.h"
#include "ui/gui_util.h"
#include "stdclass.h"
#include "hw/sh4/sh4_sched.h"
#include "oslib/i18n.h"
#include <juice/juice.h>
#include <cstring>
#include <memory>
#include <atomic>
#include <mutex>
#include <websocketpp/config/asio_no_tls_client.hpp>
#include <websocketpp/client.hpp>

namespace ice
{

static void juiceLogHandler(juice_log_level_t jlevel, const char *message)
{
	LogTypes::LOG_LEVELS level;
	switch (jlevel)
	{
	case JUICE_LOG_LEVEL_NONE:
	case JUICE_LOG_LEVEL_VERBOSE:
	case JUICE_LOG_LEVEL_DEBUG:
	default:
		level = LogTypes::LOG_LEVELS::LDEBUG;
		break;
	case JUICE_LOG_LEVEL_INFO:
		level = LogTypes::LOG_LEVELS::LINFO;
		break;
	case JUICE_LOG_LEVEL_WARN:
		if (strstr(message, "TURN CreatePermission") != nullptr)
			// error pops up a lot with standard.relay.metered.ca
			level = LogTypes::LOG_LEVELS::LINFO;
		else
			level = LogTypes::LOG_LEVELS::LWARNING;
		break;
	case JUICE_LOG_LEVEL_ERROR:
	case JUICE_LOG_LEVEL_FATAL:
		level = LogTypes::LOG_LEVELS::LERROR;
		break;
	}
	GenericLog(level, LogTypes::LOG_TYPE::NETWORK, __FILE__, __LINE__, "%s", message);
}

class IcePipe;

class IceSession : public SerialPort::Pipe
{
	struct CustomWsppConfig : public websocketpp::config::asio_client
	{
		static bool const enable_multithreading = false;
		static const size_t connection_read_buffer_size = 4_KB;
		static const size_t max_message_size = 2_KB;
		static const size_t max_http_body_size = 2_KB;
	};
	using WsClient = websocketpp::client<CustomWsppConfig>;

public:
	~IceSession()
	{
		EventManager::unlisten(Event::Start, onEmuEvent, this);
		EventManager::unlisten(Event::Terminate, onEmuEvent, this);
		EventManager::unlisten(Event::LoadState, onEmuEvent, this);
		destroyJuiceAgent();
		destroyWebSocket();
		SCIFSerialPort::Instance().setPipe(nullptr);
	}

	void init(const std::string& username, bool matchCode)
	{
		{
			Lock _(mutex);
			userlist.clear();
			chat.clear();
		}
		this->matchCode = matchCode;
		std::string room;
		if (matchCode) {
			this->username.clear();
			room = "meet" + http::urlEncode(username);
		}
		else
		{
			if (isF355())
				room = "f355";
			else if (settings.content.gameId == "MAXIMUM SPEED")
				room = "maxspeed";
			else if (isVonot())
				// US version: press LT+RT+A+Start on the start screen to enable vs cable mode
				room = "vonot";
			else if (settings.content.gameId == "T6804M")		// Aero Dancing F	FIXME not working.
				room = "aerof";
			else if (settings.content.gameId == "T6807M")		// Aero Dancing i
				room = "aeroi";
			// TODO sega tetris, hell gate
			else
				throw FlycastException(i18n::Ts("Game not supported"));
			this->username = username;
		}
		EventManager::listen(Event::Start, onEmuEvent, this);
		EventManager::listen(Event::Terminate, onEmuEvent, this);
		EventManager::listen(Event::LoadState, onEmuEvent, this);
		createWebSocket(room);
		futureStunConfig = std::async(std::launch::async, []() {
			return STUNConfig::get();
		});
	}

	State getState() const {
		return state;
	}

	std::string getStatusText() const {
		Lock _(mutex);
		return statusText;
	}

	std::vector<std::string> getUserList() const {
		Lock _(mutex);
		return userlist;
	}

	std::string getChallenger() const {
		Lock _(mutex);
		return opponent;
	}

	void sendChallenge(const std::string& user)
	{
		if (state != Online && state != ChalRefused)
			return;
		{
			Lock _(mutex);
			opponent = user;
		}
		wsSend("challenge " + user);
		state = ChalSent;
	}

	void respondChallenge(bool accept)
	{
		if (state == ChalReceived)
		{
			{
				Lock _(mutex);
				wsSend("chalresp " + opponent + " " + std::to_string((int)accept));
			}
			if (accept) {
				state = ChalAccepted;
				createJuiceAgent();
			}
			else
			{
				state = Online;
				Lock _(mutex);
				opponent.clear();
			}
		}
	}

	void say(const std::string& msg)
	{
		if (state != Offline)
			wsSend("say " + msg);
		addChat(username + ": " + msg);
	}

	std::vector<std::string> getChat() {
		Lock _(mutex);
		return chat;
	}

	struct Stats
	{
		float txQueueSize;
		float rxQueueSize;
		float rxSpeed;
		float txSpeed;
	};

	Stats getStats()
	{
		u64 now = getTimeMs();
		Stats s {};
		if (stat.timestamp != 0)
		{
			if (stat.timestamp == now) {
				s = stat.last;
				return s;
			}
			float cur = (float)stat.rxBytes.exchange(0) * 1000 / (now - stat.timestamp);
			s.rxSpeed = stat.last.rxSpeed * .3f + cur * .7f;
			cur = (float)stat.txBytes.exchange(0) * 1000 / (now - stat.timestamp);
			s.txSpeed = stat.last.txSpeed * .3f + cur * .7f;
			s.rxQueueSize = (float)recvQueue.size() * .7f + stat.last.rxQueueSize * .3f;
			s.txQueueSize = (float)txBufferSize * .7f + stat.last.txQueueSize * .3f;
			stat.last = s;
		}
		stat.timestamp = now;
		return s;
	}

private:
	void createWebSocket(const std::string& room)
	{
		// Initialize websocketpp
		wsclient.clear_access_channels(websocketpp::log::alevel::all);
		wsclient.clear_error_channels(websocketpp::log::elevel::all);
		wsclient.init_asio();
		wsclient.start_perpetual();
		asioThread.reset(new std::thread(&WsClient::run, &wsclient));
		// Connect to lobby server
		std::error_code ec;
		WsClient::connection_ptr con = wsclient.get_connection("ws://lobby.flyca.st/" + room, ec);
		if (ec)
			throw FlycastException(i18n::Ts("Connection to lobby failed:") + ' ' + ec.message());
		hdl = con->get_handle();
		// Set handlers
		con->set_open_handler([this](websocketpp::connection_hdl hdl)
		{
			state = Online;
			if (!matchCode) {
				wsSend("join " + this->username);
				addChat(": " + this->username + " joined");
			}
		});
		con->set_fail_handler([this](websocketpp::connection_hdl hdl)
		{
			WsClient::connection_ptr con = wsclient.get_con_from_hdl(hdl);
			{
				Lock _(mutex);
				statusText = con->get_ec().message();
			}
			WARN_LOG(NETWORK, "Connection to lobby failed: %s", statusText.c_str());
			if (state != Playing)
				state = Offline;
			hdl.reset();
		});
		con->set_close_handler([this](websocketpp::connection_hdl hdl)
		{
			WsClient::connection_ptr con = wsclient.get_con_from_hdl(hdl);
			INFO_LOG(NETWORK, "Connection to lobby closed: %s %s",
					websocketpp::close::status::get_string(con->get_remote_close_code()).c_str(),
					con->get_remote_close_reason().c_str());
			if (state != Playing)
			{
				state = Offline;
				if (con->get_remote_close_code() != websocketpp::close::status::going_away)
				{
					Lock _(mutex);
					statusText = con->get_remote_close_reason().empty() ?
							websocketpp::close::status::get_string(con->get_remote_close_code())
							: con->get_remote_close_reason();
				}
			}
			hdl.reset();
			{
				Lock _(mutex);
				userlist.clear();
				chat.clear();
			}
		});
		con->set_message_handler([this](websocketpp::connection_hdl hdl, WsClient::message_ptr msg) {
			onWsReceive(msg->get_payload());
		});
		wsclient.connect(con);
	}

	void destroyWebSocket()
	{
		if (asioThread != nullptr && asioThread->joinable())
		{
			wsclient.stop_perpetual();
			std::error_code ec;
			wsclient.close(hdl, websocketpp::close::status::going_away, "", ec);
			if (ec && ec != websocketpp::error::bad_connection)
				INFO_LOG(NETWORK, "Error closing connection: %s", ec.message().c_str());
			asioThread->join();	// FIXME risk of being blocked here
			asioThread.reset();
		}
	}

	void wsSend(const std::string& message)
	{
		std::error_code ec;
   		wsclient.send(hdl, message, websocketpp::frame::opcode::text, ec);
   		if (ec)
   			INFO_LOG(NETWORK, "Error sending message: %s", ec.message().c_str());
	}

	void onWsReceive(const std::string& message)
	{
		DEBUG_LOG(NETWORK, "onWsReceive: %s", message.c_str());
		auto spc = message.find(' ');
		std::string op = message.substr(0, spc);
		std::string args;
		if (spc != std::string::npos)
			args = message.substr(spc + 1);
		if (op == "join")
		{
			addChat(": " + args + " joined");
			Lock _(mutex);
			userlist.emplace_back(std::move(args));
		}
		else if (op == "userlist")
		{
			Lock _(mutex);
			userlist.clear();
			while (true)
			{
				spc = args.find(' ');
				if (spc == std::string::npos)
				{
					if (!args.empty())
						userlist.emplace_back(std::move(args));
					break;
				}
				else {
					userlist.push_back(args.substr(0, spc));
					args = args.substr(spc + 1);
				}
			}
		}
		else if (op == "candidate") {
			if (agent != nullptr)
				juice_set_remote_description(agent, args.c_str());
		}
		else if (op == "leave")
		{
			{
				Lock _(mutex);
				for (auto it = userlist.begin(); it != userlist.end(); ++it)
					if (*it == args) {
						userlist.erase(it);
						break;
					}
			}
			if (!matchCode)
				addChat(": " + args + " left");
			Lock _(mutex);
			if ((state == ChalSent || state == ChalReceived) && opponent == args) {
				state = Online;
				opponent.clear();
			}
		}
		else if (op == "challenge")
		{
			if (state != Online && state != ChalRefused) {
				// busy -> refused
				wsSend("chalresp " + args + " 0");
			}
			else {
				state = ChalReceived;
				{
					Lock _(mutex);
					opponent = args;
				}
				if (matchCode) {
					state = ChalAccepted;
					createJuiceAgent();
				}
			}
		}
		else if (op == "chalresp")
		{
			if (state == ChalSent)
			{
				if (args == "0")
				{
					state = ChalRefused;
					Lock _(mutex);
					opponent.clear();
				}
				else {
					state = ChalAccepted;
					createJuiceAgent();
				}
			}
		}
		else if (op == "say") {
			addChat(args);
		}
		else {
			WARN_LOG(NETWORK, "Invalid ws message: %s", message.c_str());
		}
	}

	void addChat(const std::string& text)
	{
		Lock _(mutex);
		chat.push_back(text);
		if (chat.size() > 20)
			chat.erase(chat.begin());
	}

	void createJuiceAgent()
	{
		juice_set_log_level(JUICE_LOG_LEVEL_INFO);
		juice_set_log_handler(juiceLogHandler);
		if (futureStunConfig.valid())
			stunConfig = futureStunConfig.get();
		juice_config_t config {};
		config.stun_server_host = stunConfig.stun_host.c_str();
		config.stun_server_port = stunConfig.stun_port;
		config.cb_state_changed = [](juice_agent_t *agent, juice_state_t state, void *user_ptr) {
			((IceSession *)user_ptr)->onJuiceStateChanged(state);
		};
		config.cb_recv = [](juice_agent_t *agent, const char *data, size_t size, void *user_ptr) {
			((IceSession *)user_ptr)->onJuiceReceive(data, size);
		};
		config.cb_gathering_done = [](juice_agent_t *agent, void *user_ptr) {
			((IceSession *)user_ptr)->onJuiceGatheringDone();
		};
		config.user_ptr = this;
		juice_turn_server_t turnServer {};
		turnServer.host = stunConfig.turn_host.c_str();
		turnServer.port = stunConfig.turn_port;
		turnServer.username = stunConfig.turn_username.c_str();
		turnServer.password = stunConfig.turn_password.c_str();
		config.turn_servers = &turnServer;
		config.turn_servers_count = 1;

		agent = juice_create(&config);
		if (agent == nullptr)
			throw FlycastException(i18n::Ts("Connection failed"));
		juice_gather_candidates(agent);
	}

	void destroyJuiceAgent()
	{
		if (agent != nullptr)
			juice_destroy(agent);
		agent = nullptr;
	}

	void sendCandidates()
	{
		Lock _(mutex);
		if (opponent.empty() || agent == nullptr)
			return;
		char sdp[JUICE_MAX_SDP_STRING_LEN];
		juice_get_local_description(agent, sdp, JUICE_MAX_SDP_STRING_LEN);
		wsSend("candidate " + opponent + " " + std::string(sdp));
	}

	void onJuiceStateChanged(juice_state_t state)
	{
		if (state == JUICE_STATE_COMPLETED)
		{
			char local[128] {};
			juice_get_selected_candidates(agent, local, sizeof(local), NULL, 0);
			INFO_LOG(NETWORK, "Juice connection completed: using %s", local);
			{
				Lock _(mutex);
				if (strstr(local, "typ host") != nullptr)
					statusText = i18n::Ts("Direct peer to peer");
				else if (strstr(local, "typ relay") != nullptr)
					statusText = i18n::Ts("Using relay");
				else
					// typ srflx: server reflexive (NAT)
					// typ prflx: peer reflexive (NAT)
					statusText = i18n::Ts("NAT");
			}
			this->state = Playing;
			setPipe();
			SCIFSerialPort::Instance().setPipe(this);
			// TODO disconnect from websocket while in game?
		}
		else if (state == JUICE_STATE_FAILED && this->state == Playing)
		{
			os_notify(i18n::T("Connection lost"), 5000);
			if (asioThread != nullptr)
			{
				if (matchCode)
				{
					// Disconnect from the lobby when using a match code.
					// Actually we might want to disconnect as soon as the connection is completed.
					destroyWebSocket();
					this->state = Offline;
				}
				else {
					this->state = Online;
				}
			}
			else {
				this->state = Offline;
			}
			Lock _(mutex);
			statusText = i18n::Ts("Connection lost");
		}
	}

	void onJuiceReceive(const char *data, size_t size);

	void onJuiceGatheringDone() {
		sendCandidates();
	}

	void receive(const char *data, size_t size)
	{
		while (size-- != 0)
			recvQueue.push(*data++);
	}

	bool isF355() const {
		return settings.content.gameId == "MK-0100"			// F355 (US)
				|| settings.content.gameId == "T8118D  50"	// F355 (EU)
				|| settings.content.gameId == "HDR-0100";	// F355 (JP)
	}

	bool isVonot() const {
		// Cyber Troopers Virtual On: Oratorio Tangram
		return settings.content.gameId == "T13004N"			// US
				|| settings.content.gameId == "HDR-0040";	// JP
	}

	void emuStartGame()
	{
		// If we are connected, restore the pipe interface when starting a new game
		if (state == Playing) {
			setPipe();
			SCIFSerialPort::Instance().setPipe(this);
		}
	}
	void emuTerminateGame()
	{
		// Leave the lobby but keep the active juice session
		destroyWebSocket();
		if (state != Playing)
			state = Offline;
		if (dump != nullptr) {
			fclose(dump);
			dump = nullptr;
		}
		txBufferSize = 0;
		recvQueue.clear();
		icePipe.reset();
	}

	static void onEmuEvent(Event event, void *arg);

	// Serial port
	void write(u8 data) override;

	void sendBreak() override {
		if (state == Playing)
			WARN_LOG(NETWORK, "ice: Ignoring sent break");
	}
	int available() override {
		return recvQueue.size();
	}
	u8 read() override
	{
		if (recvQueue.empty())
			return 0;
		else
			return recvQueue.pop();
	}

	void flushTxBuffer()
	{
		juice_send(agent, (const char *)&txBuffer[0], txBufferSize);
		stat.txBytes += txBufferSize;
		txBufferSize = 0;
	}

	void setPipe();

	juice_agent_t *agent = nullptr;
	STUNConfig stunConfig;
	std::future<STUNConfig> futureStunConfig;
	WsClient wsclient;
	std::unique_ptr<std::thread> asioThread;
	websocketpp::connection_hdl hdl;
	mutable std::mutex mutex;
	std::string username;
	std::vector<std::string> userlist;
	std::string opponent;
	State state = Offline;
	std::string statusText;
	TsQueue<u8> recvQueue;
	std::vector<std::string> chat;
	bool matchCode = false;
	std::array<uint8_t, 256> txBuffer;
	u32 txBufferSize = 0;
	struct {
		u64 timestamp = 0;
		std::atomic_int rxBytes = 0;
		std::atomic_int txBytes = 0;
		Stats last {};
	} stat;
	std::unique_ptr<IcePipe> icePipe;
	FILE *dump = nullptr;

	using Lock = std::lock_guard<std::mutex>;
	friend class IcePipe;
	friend class F355Pipe;
	friend class StandardPipe;
	friend class MaxSpeedPipe;
	friend class VonotPipe;
};

class IcePipe
{
public:
	IcePipe(IceSession& ice) : ice(ice) {}
	virtual ~IcePipe() {}
	// Called every time a byte is received from the serial port.
	virtual void onWrite() = 0;
	// Called for each packet received from the peer.
	// The default implementation delivers the data data to the serial port.
	virtual void receive(const char *data, size_t size) {
		ice.receive(data, size);
	}
	// Reset pipe when a state is loaded
	virtual void reset() {}

protected:
	IceSession& ice;
};

class StandardPipe : public IcePipe
{
public:
	StandardPipe(IceSession& ice) : IcePipe(ice) {}
	~StandardPipe() {
		sh4_sched_unregister(sh4SchedId);
	}

	void onWrite() override
	{
		if (sh4SchedId == -1)
			sh4SchedId = sh4_sched_register(0, sh4ShedCallback, this);
		sh4_sched_request(sh4SchedId, SH4_CYCLES);
	}

private:
	static int sh4ShedCallback(int tag, int cycles, int jitter, void *arg)
	{
		StandardPipe *self = (StandardPipe *)arg;
		if (self->ice.txBufferSize != 0)
			self->ice.flushTxBuffer();
		return 0;
	}

	int sh4SchedId = -1;
	constexpr static int SH4_CYCLES = SH4_MAIN_CLOCK / 10000 * 25;	// 2.5 ms seems to be fine for vonot at least
};

class F355Pipe : public IcePipe
{
public:
	F355Pipe(IceSession& ice) : IcePipe(ice) {}

	void onWrite() override
	{
		curSize++;
		// Parse F355 packets
		// 2 types:
		// 'L' 'Q' followed by 32 bytes
		// 'X' 'X' followed by 156 bytes (only in race)
		switch (curSize)
		{
		case 1:
			switch (ice.txBuffer[0])
			{
			case 'L':
				LQ = true;
				break;
			case 'X':
				LQ = false;
				break;
			default:
				// resynchronize
				INFO_LOG(NETWORK, "Resync: size %d\n", curSize);
				ice.flushTxBuffer();
				curSize = 0;
				break;
			}
			break;
		case 2:
			if ((ice.txBuffer[0] == 'L' && ice.txBuffer[1] != 'Q')
					|| (ice.txBuffer[0] == 'X' && ice.txBuffer[1] != 'X')) {
				// resynchronize
				INFO_LOG(NETWORK, "Resync: size %d\n", curSize);
				ice.flushTxBuffer();
				curSize = 0;
			}
			break;
		case 34:
			ice.flushTxBuffer();
			if (LQ)
				curSize = 0;
			break;
		case 158:
			ice.flushTxBuffer();
			curSize = 0;
			break;
		default:
			// flush every 34 bytes
			if (ice.txBufferSize == 34)
				ice.flushTxBuffer();
			break;
		}
	}

private:
	u32 curSize = 0;
	bool LQ = false;
};

class MaxSpeedPipe : public IcePipe
{
public:
	MaxSpeedPipe(IceSession& ice) : IcePipe(ice) {}

	void onWrite() override
	{
		// packets:
		// 'M' 'A' 'X' <sz> <payload>
		// sz must be >= 3. total packet size is sz + 4
		u8 last = ice.txBuffer[ice.txBufferSize - 1];
		switch (ice.txBufferSize)
		{
		case 1:
			if (last != 'M')
				ice.flushTxBuffer();
			break;
		case 2:
			if (last != 'A')
				ice.flushTxBuffer();
			break;
		case 3:
			if (last != 'X')
				ice.flushTxBuffer();
			break;
		case 4:
			if (last < 3)
				ice.flushTxBuffer();
			break;
		default:
			if (ice.txBufferSize == ice.txBuffer[3] + 4u)
				ice.flushTxBuffer();
			break;
		}
	}
};

//
// Virtual On Oratorio Tangram pipe
// Protocol spoofing inspired by eaudunord
// See https://github.com/eaudunord/dc-taisen-netplay
//
class VonotPipe : public IcePipe
{
public:
	VonotPipe(IceSession& ice) : IcePipe(ice) {}
	~VonotPipe() {
		sh4_sched_unregister(sh4SchedId);
	}

	void onWrite() override
	{
		switch (state)
		{
		case Init:
			{
				if (ice.txBufferSize == 1)
				{
					const u8 c = ice.txBuffer[0];
					if (c == 0x01)
					{
						serialReceive("\x01");
						ice.txBufferSize = 0;
						break;
					}
					else if (c == 'U')
					{
						serialReceive("\xaa");
						ice.txBufferSize = 0;
						break;
					}
					else if (c == 0xaa)
					{
						serialReceive("U");
						ice.txBufferSize = 0;
						break;
					}
				}
				int match = matcherStart.matches(&ice.txBuffer[0], ice.txBufferSize);
				if (match > 0)
				{
					serialReceive("SCIXB START");
					ice.txBufferSize = 0;
					break;
				}
				bool matchingInProgress = match == 0;
				match = matcherSynced.matches(&ice.txBuffer[0], ice.txBufferSize);
				if (match > 0)
				{
					DEBUG_LOG(NETWORK, "vonot: state -> Sync");
					serialReceive("\x02\x01\0\0\0\0\0\0\0", 9);
					state = Sync;
					ice.txBufferSize = 0;
					savedData = -1;
					break;
				}
				matchingInProgress = matchingInProgress || match == 0;
				if (!matchingInProgress)
				{
					DEBUG_LOG(NETWORK, "vonot: ignored %02x '%c'",
							ice.txBuffer[ice.txBufferSize - 1], ice.txBuffer[ice.txBufferSize - 1]);
					ice.txBufferSize = 0;
				}
				break;
			}

		case Sync:
			{
				int match = matcherStart.matches(&ice.txBuffer[0], ice.txBufferSize);
				if (match > 0)
				{
					serialReceive("SCIXB START");
					ice.txBufferSize = 0;
					state = Init;
				}
				if (match >= 0)
					break;
				if (ice.txBufferSize == 6)
				{
					// random seed
					//printf("Sync: random seed: %02x %02x %02x %02x %02x %02x\n",
					//		ice.txBuffer[0], ice.txBuffer[1], ice.txBuffer[2], ice.txBuffer[3], ice.txBuffer[4], ice.txBuffer[5]);
					prefixTxBufferAndFlush('S');
					std::lock_guard<std::mutex> _(mutex);
					if (!randomSeed.empty())
						serialReceive(randomSeed.c_str(), 6);
					else
						waitingForSeed = true;
				}
				else if (ice.txBufferSize == 1 && (ice.txBuffer[0] == 'U' || ice.txBuffer[0] == 0xaa))
				{
					if (sh4SchedId == -1)
						sh4SchedId = sh4_sched_register(0, sh4ShedCallback, this);
					// start in one sec
					sh4_sched_request(sh4SchedId, SH4_MAIN_CLOCK);
				}
				break;
			}

		case Play:
		default:
			if (matcherStart.matches(&ice.txBuffer[0], ice.txBufferSize) > 0)
			{
				DEBUG_LOG(NETWORK, "vonot: local reset");
				ice.txBuffer[0] = 'R';
				ice.txBufferSize = 1;
				ice.flushTxBuffer();
				reset();
				serialReceive("SCIXB START");
			}
			else if (ice.txBufferSize == 14) {
				prefixTxBufferAndFlush('D');
			}
			break;
		}
	}

	void receive(const char *data, size_t size) override
	{
		switch (data[0])
		{
		case 'S':	// random seed
			randomSeed = std::string(data + 1, size - 1);
			//printf("Random seed received from peer: %02x %02x %02x %02x %02x %02x\n",
			//		(u8)data[1], (u8)data[2], (u8)data[3], (u8)data[4], (u8)data[5], (u8)data[6]);
			if (state == Sync || state == Init)
			{
				std::lock_guard<std::mutex> _(mutex);
				if (waitingForSeed) {
					serialReceive(randomSeed.c_str(), 6);
					waitingForSeed = false;
				}
			}
			break;

		case 'D':	// data
			if (state == Play) {
				ice.receive(data + 1, size - 1);
			}
			else if (state == Sync)
			{
				// save for later
				if (size == 2)
					savedData = data[1];
			}
			break;

		case 'R':	// reset state
			if (state == Play) {
				reset();
				DEBUG_LOG(NETWORK, "vonot: reset received");
			}
			break;

		default:
			WARN_LOG(NETWORK, "vonot: invalid packet type %02x", data[0]);
			break;
		}
	}

	void reset() override
	{
		state = Init;
		randomSeed.clear();
	}

private:
	struct Matcher
	{
		Matcher(const char* pattern, size_t size)
			: pattern((const u8 *)pattern), patternSize(size)
		{}

		int matches(const u8 *data, size_t size) const
		{
			const size_t maxSize = std::min(size, patternSize);
			for (size_t i = 0; i < maxSize; i++)
				if (data[i] != pattern[i])
					return -1;
			if (size < patternSize)
				return 0;
			else
				return (int)patternSize;
		}

		const u8 * const pattern;
		const size_t patternSize;
	};
	const Matcher matcherStart{ "SCIXB START", 11 };
	const Matcher matcherSynced{ "\x02\x01\0\0\0\0\0\0\0", 9 };

	void serialReceive(const char *s, size_t len = 0)
	{
		if (len == 0) {
			for (const char *p = s; *p != '\0'; p++)
				ice.recvQueue.push(*p);
		}
		else {
			for (size_t i = 0; i < len; i++)
				ice.recvQueue.push(s[i]);
		}
	}

	void prefixTxBufferAndFlush(char pktType)
	{
		memmove(&ice.txBuffer[1], &ice.txBuffer[0], ice.txBufferSize);
		ice.txBuffer[0] = pktType;
		ice.txBufferSize++;
		ice.flushTxBuffer();
	}

	static int sh4ShedCallback(int tag, int cycles, int jitter, void *arg)
	{
		VonotPipe *self = (VonotPipe *)arg;
		if (!self->randomSeed.empty())
		{
			DEBUG_LOG(NETWORK, "vonot: state -> Play");
			self->state = Play;
			self->prefixTxBufferAndFlush('D');
			if (self->savedData != -1) {
				self->ice.receive((const char *)&self->savedData, 1);
				self->savedData = -1;
			}
		}
		else {
			DEBUG_LOG(NETWORK, "vonot: timeout after 1 sec");
			self->ice.txBufferSize = 0;
		}
		return 0;
	}

	enum { Init, Sync, Play } state = Init;
	std::string randomSeed;
	int sh4SchedId = -1;
	bool waitingForSeed = false;
	std::mutex mutex;
	int savedData = -1;
};

void IceSession::setPipe()
{
	if (isF355())
		icePipe = std::make_unique<F355Pipe>(*this);
	else if (settings.content.gameId == "MAXIMUM SPEED")
		icePipe = std::make_unique<MaxSpeedPipe>(*this);
	else if (isVonot())
		icePipe = std::make_unique<VonotPipe>(*this);
	else
		icePipe = std::make_unique<StandardPipe>(*this);
}

void IceSession::write(u8 data)
{
	if (state == Playing)
	{
#ifdef DUMP_SERIAL
		if (dump == nullptr)
			dump = fopen("serial-dump.bin", "wb");
		if (dump != nullptr)
			fputc(data, dump);
#endif

		txBuffer[txBufferSize++] = data;
		if (txBufferSize == txBuffer.size()) {
			WARN_LOG(NETWORK, "txBuffer full");
			flushTxBuffer();
		}
		else if (icePipe != nullptr) {
			icePipe->onWrite();
		}
	}
}

void IceSession::onJuiceReceive(const char *data, size_t size)
{
	stat.rxBytes += size;
	if (icePipe != nullptr)
		icePipe->receive(data, size);
}

void IceSession::onEmuEvent(Event event, void *arg)
{
	IceSession *ice = (IceSession *)arg;
	switch (event)
	{
	case Event::Start:
		ice->emuStartGame();
		break;
	case Event::Terminate:
		ice->emuTerminateGame();
		break;
	case Event::LoadState:
		ice->recvQueue.clear();
		ice->txBufferSize = 0;
		if (ice->icePipe != nullptr)
			ice->icePipe->reset();
		break;
	default:
		break;
	}
}

std::unique_ptr<IceSession> session;

void init(const std::string& username, bool matchCode)
{
	session.reset();
	session = std::make_unique<IceSession>();
	session->init(username, matchCode);
}

State getState()
{
	if (session == nullptr)
		return Offline;
	else
		return session->getState();
}

std::string getStatusText()
{
	if (session != nullptr)
		return session->getStatusText();
	else
		return {};
}

std::vector<std::string> getUserList()
{
	if (session == nullptr)
		return {};
	else
		return session->getUserList();
}

void sendChallenge(const std::string& user) {
	if (session != nullptr)
		session->sendChallenge(user);
}

std::string getChallenger()
{
	if (session != nullptr)
		return session->getChallenger();
	else
		return {};
}

void respondChallenge(bool accept) {
	if (session != nullptr)
		session->respondChallenge(accept);
}

void say(const std::string& msg) {
	if (session != nullptr)
		session->say(msg);
}

std::vector<std::string> getChat()
{
	if (session == nullptr)
		return {};
	else
		return session->getChat();
}

void term() {
	session.reset();
}

void displayStats()
{
	if (getState() != Playing)
		return;

	IceSession::Stats s = session->getStats();
	ImguiStyleVar _(ImGuiStyleVar_WindowRounding, 0);
	ImguiStyleVar _1(ImGuiStyleVar_WindowBorderSize, 0);
	ImGui::SetNextWindowPos(ImVec2(10, 10));
	ImGui::SetNextWindowSize(ScaledVec2(95, 0));
	ImGui::SetNextWindowBgAlpha(0.7f);
	if (ImGui::Begin("##icestats", NULL, ImGuiWindowFlags_AlwaysAutoResize | ImGuiWindowFlags_NoDecoration | ImGuiWindowFlags_NoInputs))
	{
		ImguiStyleColor _2(ImGuiCol_PlotHistogram, ImVec4(0.557f, 0.268f, 0.965f, 1.f));

		// Games set serial speed:
		// vonot	260416
		// f355		223214
		// aero i	38109
		// aero f	28935
		// tetris	111607
		// maxspeed	223214
		// hell gate 57870

		// TX/RX Queues
		ImGui::Text("Send Q");
		ImGui::ProgressBar(s.txQueueSize / 100.f, ImVec2(-1, uiScaled(10.f)), "");
		ImGui::Text("Tx Speed");
		ImGui::ProgressBar(s.txSpeed / 20000.f, ImVec2(-1, uiScaled(10.f)), "");

		ImGui::Text("Recv Q");
		ImGui::ProgressBar(s.rxQueueSize / 100.f, ImVec2(-1, uiScaled(10.f)), "");
		ImGui::Text("Rx Speed");
		ImGui::ProgressBar(s.rxSpeed / 20000.f, ImVec2(-1, uiScaled(10.f)), "");

		ImGui::TextWrapped("%s", session->getStatusText().c_str());
	}
	ImGui::End();
}

}	// namespace ice
#endif // USE_ICE && USE_WEBSOCKET