#include "mungal4.h"
#pragma hdrstop

#include "l4steamtransport.h"

//########################################################################
// Is steam_api.dll actually here?
//
// Defined outside the RP412_STEAM guard on purpose: callers ask without
// caring how the build was configured, and a build without the SDK
// truthfully has no client library. See the header for why every Steam
// path has to come through here first.
//
// The answer is cached because it is asked on menu paint, and because a
// DLL that appeared halfway through a session is not a case worth
// supporting - the delay-load helper would have bound the first miss
// anyway.
//########################################################################
Logical
	SteamNetTransport_ClientLibraryPresent()
{
#ifdef RP412_STEAM
	static int present = -1;

	if (present < 0)
	{
		present = (LoadLibraryA("steam_api.dll") != NULL) ? 1 : 0;

		if (!present)
		{
			DEBUG_STREAM << "Steam: steam_api.dll not found beside the exe - "
				<< "Steam features off, staying on TCP\n" << std::flush;
		}
	}
	return present ? True : False;
#else
	return False;
#endif
}

#ifdef RP412_STEAM

#include "l4nettransport.h"

// The engine builds with /Zp1; Valve's ABI expects default packing.
// Callback structs carry their own pack pragmas, but everything else
// must see the platform default.
#pragma pack(push, 8)
#include "steam/steam_api.h"
#include "steam/isteamnetworkingsockets.h"
#include "steam/isteamnetworkingutils.h"
#include "steam/steamnetworkingfakeip.h"
#pragma pack(pop)

//########################################################################
// SteamNetTransport - see l4steamtransport.h for the design. The
// engine is single threaded and so is this: every entry point pumps
// SteamAPI_RunCallbacks, which is where the status-changed callback
// fires (accepting incoming connections and marking drops).
//########################################################################

namespace
{
	enum
	{
		maxListeners	= 4,
		maxConnections	= 32,
		maxPeers		= 16,
		maxPending		= 8,
		leftoverSize	= 4096
	};

	struct ListenerRecord
	{
		HSteamListenSocket	handle;
		int					fakePortIndex;
		unsigned short		enginePort;			// host order
		HSteamNetConnection	pending[maxPending];
		int					pendingCount;
		// The engine closes listeners with TCP semantics (accepted
		// connections survive). Steam's CloseListenSocket kills the
		// accepted connections ungracefully - so an engine close only
		// marks the listener here; the real close waits for Cleanup.
		Logical				closed;
	};

	struct ConnectionRecord
	{
		HSteamNetConnection	handle;
		unsigned long		remoteIP;			// network order
		unsigned short		remoteEnginePort;	// network order
		Logical				connected;
		Logical				dead;
		char				leftover[leftoverSize];
		int					leftoverCount;
	};

	struct PeerRecord
	{
		unsigned long		fakeIP;				// host order
		unsigned short		fakeConsolePort;	// host order
		unsigned short		fakeGamePort;		// host order
		unsigned __int64	steamID;			// identity -> global FakeIP on accept
	};

	// engine-side port convention (lobby default; game port is +1)
	unsigned short		gEngineConsolePort = 1501;

	Logical				gSteamReady = False;
	unsigned long		gLocalFakeIP = 0;		// host order
	unsigned short		gLocalFakePorts[2] = { 0, 0 };
	char				gLocalFakeAddressString[32] = "";

	// engine ports in the order Listen sees them: [0] console, [1] game
	unsigned short		gEnginePorts[2] = { 0, 0 };
	int					gEnginePortCount = 0;

	ListenerRecord		gListeners[maxListeners];
	int					gListenerCount = 0;
	ConnectionRecord	gConnections[maxConnections];
	int					gConnectionCount = 0;
	PeerRecord			gPeers[maxPeers];
	int					gPeerCount = 0;

	ListenerRecord *
		FindListener(HSteamListenSocket handle)
	{
		for (int i = 0; i < gListenerCount; ++i)
		{
			if (gListeners[i].handle == handle)
			{
				return &gListeners[i];
			}
		}
		return NULL;
	}

	ConnectionRecord *
		FindConnection(HSteamNetConnection handle)
	{
		for (int i = 0; i < gConnectionCount; ++i)
		{
			if (gConnections[i].handle == handle)
			{
				return &gConnections[i];
			}
		}
		return NULL;
	}

	ConnectionRecord *
		AddConnection(
			HSteamNetConnection handle,
			unsigned long remote_ip_net,
			unsigned short remote_engine_port_net
		)
	{
		if (gConnectionCount >= maxConnections)
		{
			return NULL;
		}
		ConnectionRecord *record = &gConnections[gConnectionCount++];
		memset(record, 0, sizeof(*record));
		record->handle = handle;
		record->remoteIP = remote_ip_net;
		record->remoteEnginePort = remote_engine_port_net;
		return record;
	}

	void
		RemoveConnection(HSteamNetConnection handle)
	{
		for (int i = 0; i < gConnectionCount; ++i)
		{
			if (gConnections[i].handle == handle)
			{
				gConnections[i] = gConnections[gConnectionCount - 1];
				--gConnectionCount;
				return;
			}
		}
	}

	// engine port in a SOCKADDR_IN -> the peer's matching fake port
	unsigned short
		LookupPeerFakePort(unsigned long fake_ip_host, unsigned short engine_port)
	{
		for (int i = 0; i < gPeerCount; ++i)
		{
			if (gPeers[i].fakeIP == fake_ip_host)
			{
				return (engine_port == gEngineConsolePort)
					? gPeers[i].fakeConsolePort
					: gPeers[i].fakeGamePort;
			}
		}
		return 0;
	}

	//---------------------------------------------------------------
	// Status-changed handler (fires inside SteamAPI_RunCallbacks on
	// the game thread): accept incoming, queue connected, mark drops.
	// Registered through the CCallback dispatcher below - the
	// config-value function pointer is NOT dispatched by the steam_api
	// flavor of the library (Valve's own example uses STEAM_CALLBACK).
	//---------------------------------------------------------------
	void
		OnConnectionStatusChanged(SteamNetConnectionStatusChangedCallback_t *status)
	{
		switch (status->m_info.m_eState)
		{
		case k_ESteamNetworkingConnectionState_Connecting:
			if (status->m_info.m_hListenSocket != k_HSteamListenSocket_Invalid)
			{
				// no more callers once the engine closed the listener
				ListenerRecord *listener = FindListener(status->m_info.m_hListenSocket);
				if (listener == NULL || listener->closed)
				{
					DEBUG_STREAM << "SteamNetTransport: incoming ["
						<< status->m_info.m_szConnectionDescription
						<< "] on a closed listener - rejecting\n" << std::flush;
					SteamNetworkingSockets()->CloseConnection(
						status->m_hConn, 0, "listener closed", false);
					break;
				}

				// incoming: accept immediately, queue it when Connected
				DEBUG_STREAM << "SteamNetTransport: incoming ["
					<< status->m_info.m_szConnectionDescription
					<< "] - accepting\n" << std::flush;
				EResult accepted = SteamNetworkingSockets()->AcceptConnection(status->m_hConn);
				if (accepted != k_EResultOK)
				{
					DEBUG_STREAM << "SteamNetTransport: AcceptConnection failed (EResult "
						<< (int) accepted << ")\n" << std::flush;
				}
			}
			break;

		case k_ESteamNetworkingConnectionState_Connected:
			DEBUG_STREAM << "SteamNetTransport: connected ["
				<< status->m_info.m_szConnectionDescription << "]\n" << std::flush;
			if (status->m_info.m_hListenSocket != k_HSteamListenSocket_Invalid)
			{
				ListenerRecord *listener = FindListener(status->m_info.m_hListenSocket);
				if (listener != NULL && listener->pendingCount < maxPending)
				{
					listener->pending[listener->pendingCount++] = status->m_hConn;
				}
			}
			else
			{
				ConnectionRecord *record = FindConnection(status->m_hConn);
				if (record != NULL)
				{
					record->connected = True;
				}
			}
			break;

		case k_ESteamNetworkingConnectionState_ProblemDetectedLocally:
		case k_ESteamNetworkingConnectionState_ClosedByPeer:
			{
				DEBUG_STREAM << "SteamNetTransport: dropped ["
					<< status->m_info.m_szConnectionDescription
					<< "] end reason " << status->m_info.m_eEndReason
					<< ": " << status->m_info.m_szEndDebug << "\n" << std::flush;
				ConnectionRecord *record = FindConnection(status->m_hConn);
				if (record != NULL)
				{
					record->dead = True;
				}
			}
			break;
		}
	}

	//---------------------------------------------------------------
	// The callback listener: constructed after SteamAPI_Init so the
	// CCallback registration lands in a live callback manager
	//---------------------------------------------------------------
	class SteamTransportCallbacks
	{
	public:
		STEAM_CALLBACK(SteamTransportCallbacks, OnStatusChanged,
			SteamNetConnectionStatusChangedCallback_t);
	};

	void SteamTransportCallbacks::OnStatusChanged(
		SteamNetConnectionStatusChangedCallback_t *status)
	{
		OnConnectionStatusChanged(status);
	}

	SteamTransportCallbacks *gCallbacks = NULL;

	//---------------------------------------------------------------
	// The transport
	//---------------------------------------------------------------
	class SteamNetTransport:
		public NetTransport
	{
	public:
		SteamNetTransport():
			statsNextTick(0)
		{
		}

		Logical
			Startup()
		{
			// Install() already brought Steam up; the network manager
			// just confirms the transport is live
			return gSteamReady;
		}

		void
			Cleanup()
		{
			// mission teardown (single-binary race loop): drop the
			// wire but keep the Steam API up - the lobby lives on
			FlushSends();
			sendQueues.ReleaseAll();

			//
			// RP412NETSTATS: one detailed status per connection on the
			// way out - the route description (direct vs which relay)
			// lives nowhere else.
			//
			if (NetTransport_StatsEnabled())
			{
				for (int d = 0; d < gConnectionCount; ++d)
				{
					char detail[2048];
					if (SteamNetworkingSockets()->GetDetailedConnectionStatus(
							gConnections[d].handle, detail, sizeof(detail)) >= 0)
					{
						DEBUG_STREAM << "NetStats: steam conn "
							<< (unsigned long) gConnections[d].handle
							<< " detail:\n" << detail << "\n" << std::flush;
					}
				}
			}

			for (int i = 0; i < gConnectionCount; ++i)
			{
				SteamNetworkingSockets()->CloseConnection(
					gConnections[i].handle, 0, "mission teardown", true);
			}
			gConnectionCount = 0;
			for (int j = 0; j < gListenerCount; ++j)
			{
				SteamNetworkingSockets()->CloseListenSocket(gListeners[j].handle);
			}
			gListenerCount = 0;
			gEnginePortCount = 0;
		}

		int
			GetLocalAddresses(
				unsigned long *addresses,
				int max_count
			)
		{
			if (!gSteamReady || max_count < 1)
			{
				return 0;
			}
			addresses[0] = htonl(gLocalFakeIP);
			return 1;
		}

		Logical
			Resolve(
				const char *host_name,
				SOCKADDR_IN *address
			)
		{
			memset(address, 0, sizeof(SOCKADDR_IN));
			address->sin_family = AF_INET;

			SteamNetworkingIPAddr parsed;
			if (!parsed.ParseString(host_name))
			{
				return False;
			}
			address->sin_addr.S_un.S_addr = htonl(parsed.GetIPv4());
			address->sin_port = htons(parsed.m_port);	// 0 when absent
			return True;
		}

		Connection
			Connect(
				const SOCKADDR_IN *remote,
				int local_port
			)
		{
			unsigned long remote_fake_ip = ntohl(remote->sin_addr.S_un.S_addr);
			unsigned short fake_port =
				LookupPeerFakePort(remote_fake_ip, ntohs(remote->sin_port));
			if (fake_port == 0)
			{
				DEBUG_STREAM << "SteamNetTransport: no registered peer for "
					<< inet_ntoa(remote->sin_addr) << " - lobby did not feed it\n" << std::flush;
				return InvalidConnection;
			}

			DEBUG_STREAM << "SteamNetTransport: connecting to "
				<< inet_ntoa(remote->sin_addr) << ":" << ntohs(remote->sin_port)
				<< " (fake port " << fake_port << ")...\n" << std::flush;

			SteamNetworkingIPAddr target;
			target.Clear();
			target.SetIPv4(remote_fake_ip, fake_port);

			//
			// Mirror the TCP retry-while-refused loop, bounded: the
			// egg-ACK ordering means the peer may not be listening yet.
			//
			// Every wait below goes through NetTransport_PumpAndSleep, so
			// the window keeps painting and answering instead of being
			// declared "Not responding" for the whole attempt. There is no
			// render loop yet at this point - this runs before the engine
			// block - so nothing else is keeping it alive.
			//
			//
			// The deadline is PER ATTEMPT, and attempts are capped, rather
			// than one budget across all attempts.
			//
			// The 2026-08-11 playtest showed why, on every machine that met
			// the one badly-NATed peer: the first attempt goes for a direct
			// path, burns around ten seconds, and dies with 'timed out
			// attempting to connect' (5003) or 'negotiate rendezvous'
			// (5008); the SECOND attempt comes up through Valve's relay and
			// succeeds - whenever it is given time. Under one shared budget
			// the relay attempt inherited whatever the direct attempt left,
			// and the logs show it being cut off mid-connect ('attempt 2
			// ended in state 1' - still connecting) at the 20s mark. Races
			// then could not assemble, because the mesh needs every pair.
			//
			// Three attempts at RP412CONNECTWAIT each: the observed failure
			// connects on attempt 2 at about half a minute, the truly
			// unreachable peer costs about a minute instead of twenty
			// seconds, and ESC still works throughout.
			//
			DWORD wait_seconds = (DWORD) NetTransport_ConnectWaitSeconds();
			DWORD started = GetTickCount();
			int attempt = 0;
			const int kMaxAttempts = 3;
			char progress[256];

			sprintf(
				progress,
				"Red Planet - connecting to %s ...",
				inet_ntoa(remote->sin_addr)
			);
			NetTransport_SetWaitProgress(progress);

			for (;;)
			{
				++attempt;
				DWORD deadline = GetTickCount() + wait_seconds * 1000;
				HSteamNetConnection handle =
					SteamNetworkingSockets()->ConnectByIPAddress(target, 0, NULL);
				if (handle == k_HSteamNetConnection_Invalid)
				{
					DEBUG_STREAM << "SteamNetTransport: ConnectByIPAddress refused the call\n" << std::flush;
					NetTransport_SetWaitProgress(NULL);
					return InvalidConnection;
				}
				AddConnection(handle, remote->sin_addr.S_un.S_addr, remote->sin_port);

				//
				// Poll the connection state directly (the callback also
				// runs, for the log and the accept queues)
				//
				ESteamNetworkingConnectionState state =
					k_ESteamNetworkingConnectionState_Connecting;
				while ((LONG)(GetTickCount() - deadline) < 0)
				{
					SteamAPI_RunCallbacks();

					SteamNetConnectionInfo_t info;
					if (!SteamNetworkingSockets()->GetConnectionInfo(handle, &info))
					{
						state = k_ESteamNetworkingConnectionState_Dead;
						break;
					}
					state = info.m_eState;
					if (state == k_ESteamNetworkingConnectionState_Connected ||
						state == k_ESteamNetworkingConnectionState_ProblemDetectedLocally ||
						state == k_ESteamNetworkingConnectionState_ClosedByPeer ||
						state == k_ESteamNetworkingConnectionState_None)
					{
						break;
					}

					//
					// Count down out loud. Two minutes of a silent frozen
					// window gave a host nothing to act on - not which peer
					// was missing, not how long was left, not a way out.
					//
					DWORD left = (DWORD)((LONG)(deadline - GetTickCount()) / 1000);

					sprintf(
						progress,
						"Red Planet - connecting to %s (try %d of %d) ... %us left, ESC to cancel",
						inet_ntoa(remote->sin_addr),
						attempt, kMaxAttempts,
						(unsigned) left
					);
					NetTransport_SetWaitProgress(progress);

					NetWaitResult wait = NetTransport_PumpAndSleep(25);
					if (wait != NetWaitContinue)
					{
						DEBUG_STREAM << "SteamNetTransport: connect "
							<< ((wait == NetWaitCancelled) ? "cancelled" : "abandoned, quitting")
							<< " after " << ((GetTickCount() - started) / 1000)
							<< "s\n" << std::flush;
						SteamNetworkingSockets()->CloseConnection(handle, 0, "cancelled", false);
						RemoveConnection(handle);
						NetTransport_SetWaitProgress(NULL);
						return InvalidConnection;
					}
				}
				if (state == k_ESteamNetworkingConnectionState_Connected)
				{
					ConnectionRecord *record = FindConnection(handle);
					if (record != NULL)
					{
						record->connected = True;
					}
					DEBUG_STREAM << "SteamNetTransport: connect succeeded (attempt "
						<< attempt << ")\n" << std::flush;
					NetTransport_SetWaitProgress(NULL);
					return (Connection) handle;
				}

				// attempt failed - drop it and retry with a fresh window
				DEBUG_STREAM << "SteamNetTransport: attempt " << attempt
					<< " ended in state " << (int) state << "\n" << std::flush;
				SteamNetworkingSockets()->CloseConnection(handle, 0, "retry", false);
				RemoveConnection(handle);
				if (attempt >= kMaxAttempts)
				{
					DEBUG_STREAM << "SteamNetTransport: gave up after "
						<< attempt << " attempts of " << wait_seconds
						<< "s each (RP412CONNECTWAIT), "
						<< ((GetTickCount() - started) / 1000)
						<< "s in all\n" << std::flush;
					NetTransport_SetWaitProgress(NULL);
					return InvalidConnection;
				}

				//
				// A second between redials, still answering the window.
				//
				for (int slept = 0; slept < 1000; slept += 50)
				{
					NetWaitResult wait = NetTransport_PumpAndSleep(50);
					if (wait != NetWaitContinue)
					{
						DEBUG_STREAM << "SteamNetTransport: connect "
							<< ((wait == NetWaitCancelled) ? "cancelled" : "abandoned, quitting")
							<< " between attempts\n" << std::flush;
						NetTransport_SetWaitProgress(NULL);
						return InvalidConnection;
					}
				}
			}
		}

		Connection
			Listen(
				int local_port,
				int backlog
			)
		{
			if (gListenerCount >= maxListeners)
			{
				return InvalidConnection;
			}

			// a logically-closed listener for this engine port reopens
			// (the Steam socket outlives engine closes; see Close)
			for (int existing = 0; existing < gListenerCount; ++existing)
			{
				if (gListeners[existing].enginePort == (unsigned short) local_port)
				{
					gListeners[existing].closed = False;
					gListeners[existing].pendingCount = 0;
					DEBUG_STREAM << "SteamNetTransport: reopened listener on engine port "
						<< local_port << "\n" << std::flush;
					return (Connection) gListeners[existing].handle;
				}
			}

			// engine port -> fake port index, in order of appearance:
			// the console channel always listens first, the mesh second
			int index = -1;
			for (int i = 0; i < gEnginePortCount; ++i)
			{
				if (gEnginePorts[i] == (unsigned short) local_port)
				{
					index = i;
				}
			}
			if (index < 0)
			{
				if (gEnginePortCount >= 2)
				{
					DEBUG_STREAM << "SteamNetTransport: only two fake ports allocated!\n" << std::flush;
					return InvalidConnection;
				}
				index = gEnginePortCount;
				gEnginePorts[gEnginePortCount++] = (unsigned short) local_port;
			}

			DEBUG_STREAM << "SteamNetTransport: listening on engine port "
				<< local_port << " (fake port " << gLocalFakePorts[index]
				<< ")...\n" << std::flush;

			HSteamListenSocket handle =
				SteamNetworkingSockets()->CreateListenSocketP2PFakeIP(index, 0, NULL);
			if (handle == k_HSteamListenSocket_Invalid)
			{
				return InvalidConnection;
			}

			ListenerRecord *listener = &gListeners[gListenerCount++];
			memset(listener, 0, sizeof(*listener));
			listener->handle = handle;
			listener->fakePortIndex = index;
			listener->enginePort = (unsigned short) local_port;
			return (Connection) handle;
		}

		Connection
			Accept(Connection listener_handle)
		{
			SteamAPI_RunCallbacks();

			ListenerRecord *listener = FindListener((HSteamListenSocket) listener_handle);
			if (listener == NULL || listener->closed || listener->pendingCount == 0)
			{
				return InvalidConnection;
			}

			HSteamNetConnection handle = listener->pending[0];
			for (int i = 1; i < listener->pendingCount; ++i)
			{
				listener->pending[i - 1] = listener->pending[i];
			}
			--listener->pendingCount;

			//
			// The remote engine port is the same engine port we accept
			// on (all pods share the -net convention under Steam). The
			// caller's address: Steam reports incoming connections
			// under a locally-allocated ALIAS FakeIP, so resolve the
			// caller's identity through the peer table to the GLOBAL
			// FakeIP the egg's [pilots] list promised - the engine's
			// mesh identity checks compare against that.
			//
			unsigned long remote_ip_net = 0;
			SteamNetConnectionInfo_t info;
			if (SteamNetworkingSockets()->GetConnectionInfo(handle, &info))
			{
				unsigned __int64 caller_id = info.m_identityRemote.GetSteamID64();
				for (int p = 0; p < gPeerCount; ++p)
				{
					if (gPeers[p].steamID != 0 && gPeers[p].steamID == caller_id)
					{
						remote_ip_net = htonl(gPeers[p].fakeIP);
						break;
					}
				}
			}
			if (remote_ip_net == 0)
			{
				// unregistered caller: fall back to the alias address
				SteamNetworkingIPAddr remote_fake;
				if (SteamNetworkingSockets()->GetRemoteFakeIPForConnection(
						handle, &remote_fake) == k_EResultOK)
				{
					remote_ip_net = htonl(remote_fake.GetIPv4());
				}
			}
			ConnectionRecord *record = AddConnection(
				handle, remote_ip_net, htons(listener->enginePort));
			if (record != NULL)
			{
				record->connected = True;
			}
			return (Connection) handle;
		}

		void
			Close(Connection connection)
		{
			ListenerRecord *listener = FindListener((HSteamListenSocket) connection);
			if (listener != NULL)
			{
				//
				// TCP semantics: accepted connections must survive a
				// listener close. Steam's CloseListenSocket kills them
				// ungracefully, so only mark it closed (rejecting any
				// new callers) and destroy it for real in Cleanup.
				//
				listener->closed = True;
				for (int i = 0; i < listener->pendingCount; ++i)
				{
					SteamNetworkingSockets()->CloseConnection(
						listener->pending[i], 0, "listener closed", false);
				}
				listener->pendingCount = 0;
				return;
			}
			//
			// Best-effort last drain so an orderly close does not strand
			// queued bytes, then let the connection linger (true) so
			// Steam delivers what it already accepted.
			//
			NetSendQueue *queue = sendQueues.Find(connection);
			if (queue != NULL)
			{
				if (!queue->dead && queue->PendingBytes() > 0)
				{
					DrainQueue(queue);
				}
				sendQueues.Release(connection);
			}
			SteamNetworkingSockets()->CloseConnection(
				(HSteamNetConnection) connection, 0, "closed", true);
			RemoveConnection((HSteamNetConnection) connection);
		}

		int
			Send(
				Connection connection,
				const void *data,
				int size
			)
		{
			NetSendQueue *queue = sendQueues.Find(connection);

			if (queue != NULL && queue->dead)
			{
				return -1;
			}

			if (!NetTransport_SendQueueEnabled())
			{
				//
				// Legacy direct send (RP412NETSENDQ=0) - but name the
				// losses the queue would have prevented. Backpressure
				// (LimitExceeded) used to vanish into this return code.
				//
				EResult result = SteamNetworkingSockets()->SendMessageToConnection(
					(HSteamNetConnection) connection, data, (uint32) size,
					k_nSteamNetworkingSend_Reliable, NULL);
				if (result != k_EResultOK)
				{
					static int said = 0;
					if (said < 8)
					{
						++said;
						DEBUG_STREAM << "SteamNetTransport::Send: legacy mode lost a "
							<< size << " byte message (result " << (int) result
							<< ")\n" << std::flush;
					}
					return -1;
				}
				return size;
			}

			if (queue == NULL)
			{
				queue = sendQueues.FindOrAdopt(connection);
				if (queue == NULL)
				{
					// table full: behave exactly like the legacy send
					EResult result = SteamNetworkingSockets()->SendMessageToConnection(
						(HSteamNetConnection) connection, data, (uint32) size,
						k_nSteamNetworkingSend_Reliable, NULL);
					return (result == k_EResultOK) ? size : -1;
				}
			}
			queue->sentMessages++;
			queue->sentBytes += size;

			//
			// FIFO discipline: anything already queued goes first.
			//
			if (queue->PendingBytes() > 0)
			{
				if (!sendQueues.Append(queue, data, size))
				{
					MarkSendDead(connection, queue);
					return -1;
				}
				DrainQueue(queue);
				return size;
			}

			EResult result = SteamNetworkingSockets()->SendMessageToConnection(
				(HSteamNetConnection) connection, data, (uint32) size,
				k_nSteamNetworkingSend_Reliable, NULL);
			if (result == k_EResultOK)
			{
				queue->wireWrites++;
				return size;
			}
			if (result == k_EResultLimitExceeded)
			{
				//
				// Steam's send buffer is full (a peer stalled in its
				// mission load, most likely). Keep the message; the
				// flush points retry it in order.
				//
				queue->wouldBlocks++;
				if (!sendQueues.Append(queue, data, size))
				{
					MarkSendDead(connection, queue);
					return -1;
				}
				return size;
			}
			DEBUG_STREAM << "SteamNetTransport::Send: result " << (int) result
				<< " - declaring the connection dead\n" << std::flush;
			MarkSendDead(connection, queue);
			return -1;
		}

		void
			FlushSends()
		{
			for (int i = 0; i < NetSendQueueTable::MaxQueues; ++i)
			{
				NetSendQueue *queue = &sendQueues.queues[i];

				if (queue->connection != InvalidConnection &&
					!queue->dead &&
					queue->PendingBytes() > 0)
				{
					DrainQueue(queue);
				}
			}

			//
			// RP412NETSTATS: the 5 s rollup, plus the first read this
			// codebase has ever taken of Steam's own connection health -
			// ping, quality, and how much is sitting unsent or unacked
			// (the backpressure the old send path silently discarded).
			//
			if (NetTransport_StatsEnabled())
			{
				DWORD now = GetTickCount();
				if ((LONG) (now - statsNextTick) >= 0)
				{
					statsNextTick = now + 5000;

					for (int j = 0; j < NetSendQueueTable::MaxQueues; ++j)
					{
						NetSendQueue *queue = &sendQueues.queues[j];
						if (queue->connection == InvalidConnection ||
							queue->sentMessages == 0)
						{
							continue;
						}
						DEBUG_STREAM << "NetStats: steam conn "
							<< (unsigned long) queue->connection << ": "
							<< queue->sentMessages << " msgs "
							<< queue->sentBytes << " B, "
							<< queue->wireWrites << " writes ("
							<< queue->wouldBlocks << " limit-exceeded), queued high "
							<< queue->queuedHighWater << " B, pending "
							<< queue->PendingBytes()
							<< (queue->dead ? " B, DEAD" : " B")
							<< "\n" << std::flush;
					}

					for (int k = 0; k < gConnectionCount; ++k)
					{
						SteamNetConnectionRealTimeStatus_t status;
						if (SteamNetworkingSockets()->GetConnectionRealTimeStatus(
								gConnections[k].handle, &status, 0, NULL) == k_EResultOK)
						{
							DEBUG_STREAM << "NetStats: steam conn "
								<< (unsigned long) gConnections[k].handle
								<< ": ping " << status.m_nPing
								<< " ms, quality " << status.m_flConnectionQualityLocal
								<< "/" << status.m_flConnectionQualityRemote
								<< ", pending reliable " << status.m_cbPendingReliable
								<< " B, unacked " << status.m_cbSentUnackedReliable
								<< " B, queue " << (long) status.m_usecQueueTime
								<< " us\n" << std::flush;
						}
					}
				}
			}
		}

		int
			Receive(
				Connection connection,
				void *buffer,
				int size
			)
		{
			SteamAPI_RunCallbacks();

			ConnectionRecord *record = FindConnection((HSteamNetConnection) connection);
			if (record == NULL)
			{
				return ReceiveNoData;
			}

			char *out = (char *) buffer;
			int copied = 0;

			// leftover bytes from a message that outsized the last call
			if (record->leftoverCount > 0)
			{
				int take = (record->leftoverCount < size) ? record->leftoverCount : size;
				memcpy(out, record->leftover, take);
				memmove(record->leftover, record->leftover + take,
					record->leftoverCount - take);
				record->leftoverCount -= take;
				out += take;
				copied += take;
			}

			// drain messages while they fit; stash any partial tail
			while (copied < size && record->leftoverCount == 0)
			{
				SteamNetworkingMessage_t *message = NULL;
				int count = SteamNetworkingSockets()->ReceiveMessagesOnConnection(
					record->handle, &message, 1);
				if (count <= 0)
				{
					break;
				}
				int room = size - copied;
				int take = ((int) message->m_cbSize < room) ? (int) message->m_cbSize : room;
				memcpy(out, message->m_pData, take);
				out += take;
				copied += take;

				int rest = (int) message->m_cbSize - take;
				if (rest > 0)
				{
					if (rest > leftoverSize)
					{
						rest = leftoverSize;	// cannot happen: engine packets < 2K
					}
					memcpy(record->leftover, (const char *) message->m_pData + take, rest);
					record->leftoverCount = rest;
				}
				message->Release();
			}

			if (copied > 0)
			{
				return copied;
			}
			if (record->dead)
			{
				return ReceiveDisconnected;
			}
			return ReceiveNoData;
		}

		Logical
			GetRemoteAddress(
				Connection connection,
				SOCKADDR_IN *address
			)
		{
			memset(address, 0, sizeof(SOCKADDR_IN));
			ConnectionRecord *record = FindConnection((HSteamNetConnection) connection);
			if (record == NULL || !record->connected)
			{
				return False;
			}
			address->sin_family = AF_INET;
			address->sin_addr.S_un.S_addr = record->remoteIP;
			address->sin_port = record->remoteEnginePort;
			return True;
		}

	private:
		//
		// Wire messages are capped so the receiver's leftover staging
		// buffer can always hold the tail of one: the pump reads with at
		// least MAX_RECEIVE_DATA_SIZE (1600) of room, so a 2048-byte
		// message leaves at most 448 bytes over, against 4096 of staging.
		//
		enum { maxWireMessageBytes = 2048 };

		//
		// Send-side death has to surface where the engine looks for it:
		// Receive() reports ConnectionRecord::dead as a disconnect once
		// the leftovers drain. Mark both sides so drains stop too.
		//
		void
			MarkSendDead(
				Connection connection,
				NetSendQueue *queue)
		{
			if (queue != NULL)
			{
				queue->dead = True;
			}
			ConnectionRecord *record = FindConnection((HSteamNetConnection) connection);
			if (record != NULL)
			{
				record->dead = True;
			}
		}

		//
		// Push pending bytes at the connection in wire-message chunks.
		// A MUNGA message split across two wire messages is fine - the
		// receiver treats the message sequence as a byte stream and
		// re-frames from the length prefixes.
		//
		void
			DrainQueue(NetSendQueue *queue)
		{
			while (queue->PendingBytes() > 0 && !queue->dead)
			{
				int chunk = queue->PendingBytes();
				if (chunk > maxWireMessageBytes)
				{
					chunk = maxWireMessageBytes;
				}
				EResult result = SteamNetworkingSockets()->SendMessageToConnection(
					(HSteamNetConnection) queue->connection,
					&queue->buffer[queue->head], (uint32) chunk,
					k_nSteamNetworkingSend_Reliable, NULL);
				if (result == k_EResultOK)
				{
					queue->wireWrites++;
					queue->head += chunk;
					if (queue->PendingBytes() == 0)
					{
						queue->head = 0;
						queue->tail = 0;
					}
					continue;
				}
				if (result == k_EResultLimitExceeded)
				{
					queue->wouldBlocks++;
					break;
				}
				DEBUG_STREAM << "SteamNetTransport::DrainQueue: result " << (int) result
					<< " with " << queue->PendingBytes()
					<< " bytes pending - declaring the connection dead\n" << std::flush;
				MarkSendDead(queue->connection, queue);
				break;
			}
		}

		NetSendQueueTable	sendQueues;
		DWORD				statsNextTick;
	};

	SteamNetTransport gSteamTransport;
}

//########################################################################
// Install: bring Steam up, get our FakeIP identity, take over the wire
//########################################################################

Logical
	SteamNetTransport_Install()
{
	if (gSteamReady)
	{
		return True;
	}

	//
	// Before ANY Steam symbol: steam_api.dll is delay-loaded, so calling
	// into it when it is missing raises the helper's fatal exception
	// rather than failing. This is the gate that makes the DLL optional.
	//
	if (!SteamNetTransport_ClientLibraryPresent())
	{
		return False;
	}

	if (!SteamAPI_Init())
	{
		DEBUG_STREAM << "SteamNetTransport: SteamAPI_Init failed "
			<< "(Steam not running, or no steam_appid.txt) - staying on TCP\n" << std::flush;
		return False;
	}

	// connection status arrives through the CCallback dispatcher
	if (gCallbacks == NULL)
	{
		gCallbacks = new SteamTransportCallbacks;
	}

	SteamNetworkingUtils()->InitRelayNetworkAccess();

	//
	// Mission load stalls the game thread for 10-30s with nothing
	// pumping - Steam's default 10s connected-timeout would shear
	// every connection mid-load (seen live: end reason 4001 with rx
	// ages right at load duration). TCP never timed out idle arcade
	// links; 90s keeps that spirit.
	//
	SteamNetworkingUtils()->SetGlobalConfigValueInt32(
		k_ESteamNetworkingConfig_TimeoutConnected, 90 * 1000);

	if (!SteamNetworkingSockets()->BeginAsyncRequestFakeIP(2))
	{
		DEBUG_STREAM << "SteamNetTransport: BeginAsyncRequestFakeIP failed - staying on TCP\n" << std::flush;
		return False;
	}

	// FakeIP allocation is async: pump until it lands (or give up)
	SteamNetworkingFakeIPResult_t fake;
	memset(&fake, 0, sizeof(fake));
	DWORD deadline = GetTickCount() + 20 * 1000;
	for (;;)
	{
		SteamAPI_RunCallbacks();
		SteamNetworkingSockets()->GetFakeIP(0, &fake);
		if (fake.m_eResult == k_EResultOK)
		{
			break;
		}
		if (fake.m_eResult != k_EResultBusy && fake.m_eResult != k_EResultNoMatch)
		{
			DEBUG_STREAM << "SteamNetTransport: FakeIP allocation failed (EResult "
				<< (int) fake.m_eResult << ") - staying on TCP\n" << std::flush;
			return False;
		}
		if ((LONG)(GetTickCount() - deadline) >= 0)
		{
			DEBUG_STREAM << "SteamNetTransport: FakeIP allocation timed out - staying on TCP\n" << std::flush;
			return False;
		}
		Sleep(50);
	}

	gLocalFakeIP = fake.m_unIP;
	gLocalFakePorts[0] = fake.m_unPorts[0];
	gLocalFakePorts[1] = fake.m_unPorts[1];

	SteamNetworkingIPAddr mine;
	mine.Clear();
	mine.SetIPv4(gLocalFakeIP, 0);
	mine.ToString(gLocalFakeAddressString, sizeof(gLocalFakeAddressString), false);

	DEBUG_STREAM << "SteamNetTransport: up as " << gLocalFakeAddressString
		<< " (fake ports " << gLocalFakePorts[0] << " console, "
		<< gLocalFakePorts[1] << " game)\n" << std::flush;

	gSteamReady = True;
	NetTransport_Set(&gSteamTransport);

	//
	// dev: RP412STEAMSELFTEST=1 loops a connection back to ourselves,
	// proving the listen/accept/connect/send/receive machinery (and
	// the status-callback dispatch) without a second machine
	//
	const char *self_test = getenv("RP412STEAMSELFTEST");
	if (self_test != NULL && atoi(self_test) != 0)
	{
		DEBUG_STREAM << "SteamNetTransport: SELF TEST starting\n" << std::flush;
		SteamNetTransport_RegisterPeer(
			gLocalFakeAddressString, gLocalFakePorts[0], gLocalFakePorts[1],
			SteamUser()->GetSteamID().ConvertToUint64());

		NetTransport *transport = &gSteamTransport;
		NetTransport::Connection listener = transport->Listen(1501, 1);

		SOCKADDR_IN self_address;
		transport->Resolve(gLocalFakeAddressString, &self_address);
		self_address.sin_port = htons(1501);
		NetTransport::Connection outgoing = transport->Connect(&self_address, 0);

		NetTransport::Connection incoming = NetTransport::InvalidConnection;
		DWORD accept_deadline = GetTickCount() + 10 * 1000;
		while (incoming == NetTransport::InvalidConnection &&
			(LONG)(GetTickCount() - accept_deadline) < 0)
		{
			incoming = transport->Accept(listener);
			Sleep(50);
		}

		Logical ok = False;
		Logical survives_close = False;
		if (outgoing != NetTransport::InvalidConnection &&
			incoming != NetTransport::InvalidConnection)
		{
			transport->Send(outgoing, "PING", 4);
			char buffer[16];
			DWORD recv_deadline = GetTickCount() + 5 * 1000;
			while ((LONG)(GetTickCount() - recv_deadline) < 0)
			{
				if (transport->Receive(incoming, buffer, sizeof(buffer)) == 4)
				{
					ok = (memcmp(buffer, "PING", 4) == 0);
					break;
				}
				Sleep(25);
			}

			//
			// The arcade engine closes a listener right after
			// accepting - the accepted connection MUST survive
			// (Steam kills children on CloseListenSocket; the
			// transport defers the real close to Cleanup)
			//
			transport->Close(listener);
			transport->Send(incoming, "PONG", 4);
			recv_deadline = GetTickCount() + 5 * 1000;
			while ((LONG)(GetTickCount() - recv_deadline) < 0)
			{
				if (transport->Receive(outgoing, buffer, sizeof(buffer)) == 4)
				{
					survives_close = (memcmp(buffer, "PONG", 4) == 0);
					break;
				}
				Sleep(25);
			}
		}
		DEBUG_STREAM << "SteamNetTransport: SELF TEST "
			<< ((ok && survives_close) ? "PASSED" : "FAILED")
			<< " (out " << (outgoing != NetTransport::InvalidConnection)
			<< ", in " << (incoming != NetTransport::InvalidConnection)
			<< ", ping " << ok
			<< ", survives listener close " << survives_close
			<< ")\n" << std::flush;

		// leave the session pristine for the real race
		gSteamTransport.Cleanup();
		gPeerCount = 0;
	}
	return True;
}

const char *
	SteamNetTransport_GetFakeAddressString()
{
	return gLocalFakeAddressString;
}

int
	SteamNetTransport_GetFakeConsolePort()
{
	return gLocalFakePorts[0];
}

int
	SteamNetTransport_GetFakeGamePort()
{
	return gLocalFakePorts[1];
}

void
	SteamNetTransport_SetEnginePorts(int console_port)
{
	gEngineConsolePort = (unsigned short) console_port;
}

void
	SteamNetTransport_RegisterPeer(
		const char *fake_ip,
		int fake_console_port,
		int fake_game_port,
		unsigned __int64 steam_id
	)
{
	SteamNetworkingIPAddr parsed;
	if (!parsed.ParseString(fake_ip))
	{
		return;
	}
	for (int i = 0; i < gPeerCount; ++i)
	{
		if (gPeers[i].fakeIP == parsed.GetIPv4())
		{
			gPeers[i].fakeConsolePort = (unsigned short) fake_console_port;
			gPeers[i].fakeGamePort = (unsigned short) fake_game_port;
			gPeers[i].steamID = steam_id;
			return;
		}
	}
	if (gPeerCount >= maxPeers)
	{
		return;
	}
	gPeers[gPeerCount].fakeIP = parsed.GetIPv4();
	gPeers[gPeerCount].fakeConsolePort = (unsigned short) fake_console_port;
	gPeers[gPeerCount].fakeGamePort = (unsigned short) fake_game_port;
	gPeers[gPeerCount].steamID = steam_id;
	++gPeerCount;
	DEBUG_STREAM << "SteamNetTransport: peer registered: " << fake_ip
		<< " (console " << fake_console_port << ", game "
		<< fake_game_port << ", id " << steam_id << ")\n" << std::flush;
}

#endif	// RP412_STEAM