CultLib/src/GameCult.Networking/CultNetRudpSchemaServer.cs

using System;
using System.Runtime.CompilerServices;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Net;
using System.Net.Sockets;
using System.Text;
using System.Threading.Tasks;
using GameCult.Logging;

namespace GameCult.Networking
{
    /// <summary>
    /// Result of a bounded non-blocking RUDP schema-server drain.
    /// </summary>
    public readonly struct RudpCultNetSchemaPollResult
    {
        public RudpCultNetSchemaPollResult(int transportItemsConsumed, int messagesDispatched)
        {
            TransportItemsConsumed = transportItemsConsumed;
            MessagesDispatched = messagesDispatched;
        }

        /// <summary>Gets transport work consumed, including ACK and connection control packets.</summary>
        public int TransportItemsConsumed { get; }
        /// <summary>Gets application schema messages dispatched.</summary>
        public int MessagesDispatched { get; }
    }

    /// <summary>
    /// Options for hosting schema-v0 messages over the native CultNet RUDP listener.
    /// </summary>
    public sealed class RudpCultNetSchemaServerOptions
    {
        /// <summary>
        /// Gets or sets the runtime id advertised by the RUDP listener.
        /// </summary>
        public string RuntimeId { get; set; } = "csharp-rudp-schema-server";
        /// <summary>
        /// Gets or sets the UDP socket used by the listener. When omitted, a loopback socket is created.
        /// </summary>
        public Socket? Socket { get; set; }
        /// <summary>
        /// Gets or sets the accepted RUDP connection id.
        /// </summary>
        public uint ConnectionId { get; set; } = 0x43554c54;
        /// <summary>
        /// Gets or sets the first local packet sequence for each accepted peer. Unset, each peer's
        /// session draws its own at random.
        /// </summary>
        public uint? InitialSequence { get; set; }
        /// <summary>
        /// Gets or sets the resend delay in milliseconds.
        /// </summary>
        public long ResendDelayMs { get; set; } = 25;
        /// <summary>
        /// Gets or sets the advertised transport id.
        /// </summary>
        public string TransportId { get; set; } = "schema-rudp";
        /// <summary>
        /// Gets or sets the maximum fragment size for schema-channel payloads.
        /// </summary>
        public int? MaxFragmentBytes { get; set; } = 1024;
        /// <summary>
        /// Gets or sets the maximum pending reliable packet count.
        /// </summary>
        public int? MaxPendingReliablePackets { get; set; }
        /// <summary>
        /// Gets or sets the payload sent with accept packets.
        /// </summary>
        public byte[]? AcceptPayload { get; set; }
    }

    /// <summary>
    /// Peer context for a schema-v0 RUDP server.
    /// </summary>
    public sealed class RudpCultNetSchemaServerPeer : ICultNetSchemaServerPeer, ICultNetSchemaServerPeerLocation
    {
        private readonly RudpCultNetSchemaServer _server;

        internal RudpCultNetSchemaServerPeer(
            RudpCultNetSchemaServer server,
            CultNetRudpSocketServerPeer transportPeer)
        {
            _server = server;
            TransportPeer = transportPeer;
        }

        /// <summary>
        /// Gets the underlying RUDP transport peer.
        /// </summary>
        public CultNetRudpSocketServerPeer TransportPeer { get; }
        /// <summary>
        /// Gets the UDP endpoint for this peer.
        /// </summary>
        public EndPoint RemoteEndPoint => TransportPeer.RemoteEndPoint;

        /// <summary>
        /// Sends a schema-v0 response to this peer.
        /// </summary>
        public void SendCultNet<TMessage>(TMessage message)
            where TMessage : ICultNetSchemaMessage
        {
            _server.SendCultNet(TransportPeer, message);
        }
    }

    /// <summary>
    /// Multi-peer schema-v0 host over the native CultNet RUDP transport.
    /// </summary>
    public sealed class RudpCultNetSchemaServer : ICultNetSchemaServer, ICultNetSchemaServerPeerLifecycle, IDisposable
    {
        private readonly Socket _socket;
        private readonly CultNetRudpSocketTransportServer _transport;
        private readonly ConcurrentDictionary<Type, Delegate> _handlers = new();
        private readonly ConcurrentDictionary<Delegate, Delegate> _schemaPeerAdapters = new();
        private readonly ConditionalWeakTable<CultNetRudpSocketServerPeer, StrongBox<int>> _undecodableFramesInARow = new();
        private bool _disposed;
        private ILogger _logger = new NullLogger();

        /// <summary>
        /// How many consecutive undecodable frames a peer's session may send before it is ended. Fewer are
        /// answered with a typed <c>message_invalid</c> and the session is kept; the frame that reaches this
        /// number is answered too, then the session ends. Any frame that decodes resets the run. The same
        /// number is the cap in every runtime.
        /// </summary>
        public const int MaxConsecutiveRefusedFrames = 8;

        /// <summary>
        /// How long a session whose capped frame was answered waits for the peer to acknowledge the answer
        /// before it ends anyway.
        /// </summary>
        internal const long CappedAnswerGraceMs = 2000;

        /// <summary>
        /// Gets or sets the logger used by the server. R-AM: the dispatch backstop in DispatchAsync
        /// logs every untyped fault through this instead of swallowing it silently.
        /// </summary>
        public ILogger Logger
        {
            get => _logger;
            set => _logger = value ?? new NullLogger();
        }

        /// <summary>
        /// Creates a schema-v0 RUDP server.
        /// </summary>
        public RudpCultNetSchemaServer(RudpCultNetSchemaServerOptions? options = null)
        {
            options ??= new RudpCultNetSchemaServerOptions();
            var socket = options.Socket;
            if (socket == null)
            {
                socket = new Socket(AddressFamily.InterNetwork, SocketType.Dgram, ProtocolType.Udp);
                socket.Bind(new IPEndPoint(IPAddress.Loopback, 0));
            }

            _socket = socket;
            _transport = new CultNetRudpSocketTransportServer(new CultNetRudpSocketTransportServerOptions
            {
                RuntimeId = options.RuntimeId,
                Socket = socket,
                ConnectionId = options.ConnectionId,
                InitialSequence = options.InitialSequence,
                ResendDelayMs = options.ResendDelayMs,
                TransportId = options.TransportId,
                MaxFragmentBytes = options.MaxFragmentBytes,
                MaxPendingReliablePackets = options.MaxPendingReliablePackets,
                AcceptPayload = options.AcceptPayload
            });
            _transport.PeerDisconnected += OnTransportPeerDisconnected;
        }

        /// <summary>
        /// Gets the underlying RUDP listener profile.
        /// </summary>
        public CultNetTransportProfile Profile => _transport.Profile;

        /// <summary>
        /// Gets the local UDP endpoint.
        /// </summary>
        public IPEndPoint LocalEndPoint => (IPEndPoint)_socket.LocalEndPoint!;

        /// <summary>
        /// Gets a snapshot of the current transfer counters.
        /// </summary>
        public CultNetTransportStats Stats => _transport.Stats;

        // The listener this server drives, for tests that inject send failures through its seams.
        internal CultNetRudpSocketTransportServer Transport => _transport;

        /// <summary>
        /// Gets the currently tracked RUDP peers.
        /// </summary>
        public IReadOnlyCollection<CultNetRudpSocketServerPeer> Peers => _transport.Peers;

        /// <inheritdoc />
        public event Action<ICultNetSchemaServerPeer>? PeerDisconnected;

        /// <summary>
        /// Registers a schema-v0 handler.
        /// </summary>
        public void OnCultNet<TMessage>(Func<TMessage, RudpCultNetSchemaServerPeer, Task> callback)
            where TMessage : ICultNetSchemaMessage
        {
            if (callback == null) throw new ArgumentNullException(nameof(callback));
            _handlers.AddOrUpdate(
                typeof(TMessage),
                _ => callback,
                (type, current) =>
                {
                    var combined = Delegate.Combine(current, callback) as Func<TMessage, RudpCultNetSchemaServerPeer, Task>;
                    return combined ?? throw new InvalidOperationException($"Failed to combine delegates for {type.Name}.");
                });
        }

        /// <summary>
        /// Registers a schema-v0 handler.
        /// </summary>
        public void OnCultNet<TMessage>(Action<TMessage, RudpCultNetSchemaServerPeer> callback)
            where TMessage : ICultNetSchemaMessage
        {
            if (callback == null) throw new ArgumentNullException(nameof(callback));
            OnCultNet<TMessage>((message, peer) =>
            {
                callback(message, peer);
                return Task.CompletedTask;
            });
        }

        /// <inheritdoc />
        public void OnCultNet<TMessage>(Func<TMessage, ICultNetSchemaServerPeer, Task> callback)
            where TMessage : ICultNetSchemaMessage
        {
            if (callback == null) throw new ArgumentNullException(nameof(callback));
            Func<TMessage, RudpCultNetSchemaServerPeer, Task> adapter = (message, peer) => callback(message, peer);
            _schemaPeerAdapters[callback] = adapter;
            OnCultNet(adapter);
        }

        /// <inheritdoc />
        public void RemoveCultNetMessageListener<TMessage>(Delegate callback)
            where TMessage : ICultNetSchemaMessage
        {
            if (_schemaPeerAdapters.TryRemove(callback, out var adapter))
            {
                callback = adapter;
            }

            if (!_handlers.TryGetValue(typeof(TMessage), out var current))
            {
                return;
            }

            var next = Delegate.Remove(current, callback);
            if (next == null)
            {
                _handlers.TryRemove(typeof(TMessage), out _);
            }
            else
            {
                _handlers[typeof(TMessage)] = next;
            }
        }

        /// <summary>
        /// Sends a schema-v0 message to a transport peer.
        /// </summary>
        public void SendCultNet<TMessage>(CultNetRudpSocketServerPeer peer, TMessage message)
            where TMessage : ICultNetSchemaMessage
        {
            _transport.SendSchemaMessage(peer, message);
        }

        /// <summary>
        /// Polls one packet/frame and dispatches at most one schema-v0 message.
        /// </summary>
        public async Task<bool> PollOnceAsync()
        {
            if (!_transport.TryReceiveOnce(out var delivered) || delivered == null)
            {
                return false;
            }

            return await DispatchAsync(delivered).ConfigureAwait(false);
        }

        /// <summary>
        /// Drains up to <paramref name="maxTransportItems"/> immediately available transport work items.
        /// Control packets count as progress so callers only idle when the transport is actually empty.
        /// </summary>
        public async Task<RudpCultNetSchemaPollResult> PollAvailableAsync(int maxTransportItems)
        {
            if (maxTransportItems <= 0) throw new ArgumentOutOfRangeException(nameof(maxTransportItems));
            var consumed = 0;
            var dispatched = 0;
            while (consumed < maxTransportItems && _transport.TryReceiveOnce(out var delivered))
            {
                consumed++;
                if (delivered != null && await DispatchAsync(delivered).ConfigureAwait(false))
                    dispatched++;
            }
            return new RudpCultNetSchemaPollResult(consumed, dispatched);
        }

        /// <summary>
        /// The goodbye's reason when a handler fails on a peer's message: the server could not handle
        /// it, which says nothing about whether the peer's packet was well formed.
        /// </summary>
        internal static readonly byte[] HandlerFailedReason = Encoding.UTF8.GetBytes("server could not handle a message");

        private static string SessionOutcome(bool ended) =>
            ended ? "that peer's session ended." : "that peer's session had already ended.";

        private async Task<bool> DispatchAsync(CultNetRudpSocketServerFrame delivered)
        {
            if (!string.Equals(delivered.Frame.ChannelId, "schema", StringComparison.Ordinal))
                return false;
            // The capped frame was answered: nothing after it is served while the answer is delivered.
            if (delivered.Peer.EndingReason != null)
                return false;

            // A peer's message that cannot be decoded (malformed bytes, an unknown schemaVersion) or
            // handled ends that peer's session, never the poll: the drain goes on serving every other
            // peer. The transport already acknowledged the frame, so the session cannot be kept (a
            // retransmit would never be handled). The peer is told, and the drop is counted and logged.
            // The C# TCP schema server (CultNetTcpSchemaTransport.ServePeerAsync) likewise ends the one
            // peer whose message fails to decode or handle. A handler is still expected to answer its
            // own typed refusals on the wire (HandleSnapshotRequestV1Async does, R-AM); this is the
            // backstop for whatever it does not. The log names the fault's type only: its message can
            // quote what the peer sent.
            ICultNetSchemaMessage message;
            try
            {
                message = CultNetSchemaMessageSerialization.Deserialize(delivered.Frame.Payload);
            }
            catch (Exception ex)
            {
                // The peer sent something no runtime can read: the refusal is the peer's. It is answered
                // message_invalid and the session kept, until MaxConsecutiveRefusedFrames in a row end it.
                var run = _undecodableFramesInARow.GetOrCreateValue(delivered.Peer);
                run.Value++;
                _transport.SendSchemaMessage(delivered.Peer, CultNetErrorMessage.ForMessageInvalid("message", "CultNet RUDP schema server could not read the message."));
                if (run.Value < MaxConsecutiveRefusedFrames)
                {
                    Logger.LogError($"CultNet RUDP schema server could not decode a peer's message ({ex.GetType().Name}); it was refused and that peer's session kept.");
                    return false;
                }
                // The answer was sent reliably: the session ends once the peer has acknowledged it (a lost
                // datagram is resent first), or after the grace.
                _transport.EndPeerSessionWhenAnswered(delivered.Peer, CultNetRudpSession.RefusedPacketReason, CappedAnswerGraceMs);
                Logger.LogError($"CultNet RUDP schema server could not decode a peer's message ({ex.GetType().Name}); that peer's session ends once its answer is acknowledged.");
                return false;
            }
            _undecodableFramesInARow.GetOrCreateValue(delivered.Peer).Value = 0;

            if (!_handlers.TryGetValue(message.GetType(), out var handler) || handler == null)
            {
                return false;
            }

            try
            {
                var result = handler.DynamicInvoke(message, new RudpCultNetSchemaServerPeer(this, delivered.Peer));
                if (result is Task task)
                {
                    await task.ConfigureAwait(false);
                }
            }
            catch (Exception ex)
            {
                // DynamicInvoke wraps a synchronously-thrown handler fault in TargetInvocationException;
                // name the real fault, not that wrapper. The fault is the server's (a response too large
                // to send, a handler bug) or a refusal the handler did not type, so the goodbye does not
                // say the peer's packet was refused. A response that failed after its first fragment
                // already ended the session over the send, and the log says so.
                var reported = ex is System.Reflection.TargetInvocationException { InnerException: { } inner } ? inner : ex;
                var ended = _transport.EndPeerSession(delivered.Peer, HandlerFailedReason);
                Logger.LogError($"CultNet RUDP schema server's {message.GetType().Name} handler failed ({reported.GetType().Name}); {SessionOutcome(ended)}");
                return false;
            }
            return true;
        }

        /// <summary>
        /// Sends any reliable packets whose resend timers are due.
        /// </summary>
        public void PollResends()
        {
            _transport.PollResends();
        }

        /// <inheritdoc />
        public void Dispose()
        {
            if (_disposed)
            {
                return;
            }

            _disposed = true;
            _transport.PeerDisconnected -= OnTransportPeerDisconnected;
            _transport.Dispose();
        }

        private void OnTransportPeerDisconnected(CultNetRudpSocketServerPeer peer)
        {
            PeerDisconnected?.Invoke(new RudpCultNetSchemaServerPeer(this, peer));
        }
    }
}