MoShiro/backend/fleet/src/Fleet.Server.Link/NodeConnection.Wire.cs

using System.Buffers;
using System.Net.WebSockets;
using System.Text.Json;
using Fleet.Rpc;

namespace Fleet.Server;

/// <summary>线上的帧：WebSocket 收帧、解析、状态名。</summary>
internal sealed partial class NodeConnection
{
    private async ValueTask<Frame?> ReceiveFrameAsync(CancellationToken ct)
    {
        var writer = new ArrayBufferWriter<byte>(1024);
        while (true)
        {
            Memory<byte> buffer = writer.GetMemory(4096);
            ValueWebSocketReceiveResult result = await _socket.ReceiveAsync(buffer, ct);
            if (result.MessageType == WebSocketMessageType.Close)
            {
                return null;
            }

            writer.Advance(result.Count);
            if (writer.WrittenCount > _host.Options.MaxFrameBytes)
            {
                _closeReason ??= "frame too large";
                await _socket.CloseOutputAsync(WebSocketCloseStatus.MessageTooBig, "frame too large", CancellationToken.None);
                return null;
            }

            if (result.EndOfMessage)
            {
                if (_cipher is { } cipher)
                {
                    // 加密链路上只接受通过认证的二进制帧；明文、被改过的、重放的、乱序的一律断开
                    if (result.MessageType != WebSocketMessageType.Binary || !cipher.TryOpen(writer.WrittenSpan, out byte kind, out byte[] plain))
                    {
                        _closeReason ??= "frame failed authentication";
                        await _socket.CloseOutputAsync(WebSocketCloseStatus.PolicyViolation, "frame failed authentication", CancellationToken.None);
                        return null;
                    }

                    return new Frame(kind == 0, plain);
                }

                return new Frame(result.MessageType == WebSocketMessageType.Text, writer.WrittenSpan.ToArray());
            }
        }
    }

    private static bool TryParse(byte[] data, [System.Diagnostics.CodeAnalysis.NotNullWhen(true)] out RpcMessage? message)
    {
        try
        {
            message = JsonSerializer.Deserialize(data, FleetRpcContext.Default.RpcMessage);
            return message is not null;
        }
        catch (JsonException)
        {
            message = null;
            return false;
        }
    }

    private static bool TryDecode(string text, out byte[] data)
    {
        try
        {
            data = Convert.FromBase64String(text);
            return true;
        }
        catch (FormatException)
        {
            data = [];
            return false;
        }
    }

    private static string StatusName(NodeStatus status) => status switch
    {
        NodeStatus.Active => "active",
        NodeStatus.Pending => "pending",
        _ => "disabled",
    };

    private sealed record Frame(bool IsText, byte[] Data);
}


[System.Text.Json.Serialization.JsonSerializable(typeof(EmptyObject))]
internal sealed partial class FleetEmptyObjectContext : System.Text.Json.Serialization.JsonSerializerContext;

internal sealed class EmptyObject;

internal static class FleetEmptyObject
{
    public static System.Text.Json.Serialization.Metadata.JsonTypeInfo<EmptyObject> Info => FleetEmptyObjectContext.Default.EmptyObject;
}