IC.Rebuild/Core/NetworkPipeline.cs

using System.Diagnostics;

namespace ICRebuild.Core;

public enum NetworkPhase { Received, BeforeSend, AfterSend }
public enum PacketOrigin { NativeZone, PluginActiveSend }
public enum PacketStage { RawIpcPayload }
public sealed record PacketContext(long ObservationId, PacketOrigin Origin, PacketStage Stage,
    NetworkPhase Phase, byte? NativeResult = null);

// Owned immutable copy. No native pointer or mutable array is exposed to observers.
public sealed class NetworkMessage
{
    private readonly byte[] payload;
    public long Id { get; }
    public DateTimeOffset Time { get; } = DateTimeOffset.UtcNow;
    public long MonotonicTicks { get; } = Stopwatch.GetTimestamp();
    public PacketDirection Direction { get; }
    public string GameVersion { get; }
    public int Opcode { get; }
    public uint? Target { get; }
    public bool Complete { get; }
    public string? Issue { get; }
    public PacketOrigin Origin { get; }
    public ReadOnlySpan<byte> Payload => payload;
    public NetworkMessage(long id, PacketDirection direction, string version, CapturedPacket packet,
        PacketOrigin origin = PacketOrigin.NativeZone)
    {
        Id=id; Direction=direction; GameVersion=version; Opcode=packet.Opcode;
        Target=packet.Target; Complete=packet.Complete; Issue=packet.Issue; Origin=origin;
        payload=packet.Payload.ToArray();
    }
}

public sealed record NetworkObservation(NetworkMessage Message, NetworkPhase Phase, byte? NativeResult = null)
{
    public PacketContext Context => new(Message.Id,Message.Origin,PacketStage.RawIpcPayload,Phase,NativeResult);
}

// Inline on the native calling thread, in registration order per observation. Concurrent
// calls may interleave. Callbacks must be short; no game/UI service access, waits for
// Framework, native sending, or changes to native Hook lifetimes from a callback.
public sealed class NetworkPipeline(string version) : IDisposable
{
    private sealed class Subscription(NetworkPhase phase, Action<NetworkObservation> callback, int? opcode)
    {
        public readonly NetworkPhase Phase=phase;
        public readonly Action<NetworkObservation> Callback=callback;
        public readonly int? Opcode=opcode;
        public int Active=1;
        public bool Matches(NetworkPhase phase, int? opcode=null) => Volatile.Read(ref Active)!=0 &&
            Phase==phase && (!opcode.HasValue || !Opcode.HasValue || Opcode==opcode);
    }
    private readonly object sync=new();
    private Subscription[] subscriptions=[];
    private bool disposed;
    private long sequence,readErrors,observerErrors,received,beforeSend,afterSend;
    private int lastSendResult=-1;
    private string lastIssue="无";
    public long ReadErrors => Interlocked.Read(ref readErrors);
    public long ObserverErrors => Interlocked.Read(ref observerErrors);
    public long Received => Interlocked.Read(ref received);
    public long BeforeSend => Interlocked.Read(ref beforeSend);
    public long AfterSend => Interlocked.Read(ref afterSend);
    public int LastSendResult => Volatile.Read(ref lastSendResult);
    public string LastIssue => Volatile.Read(ref lastIssue);
    public int SubscriberCount => Volatile.Read(ref subscriptions).Length;

    public IDisposable Subscribe(NetworkPhase phase, Action<NetworkObservation> callback, int? opcode=null)
    {
        ArgumentNullException.ThrowIfNull(callback);
        if(!Enum.IsDefined(phase) || opcode is <0 or >65535) throw new ArgumentOutOfRangeException(nameof(opcode));
        var item=new Subscription(phase,callback,opcode);
        lock(sync)
        {
            ObjectDisposedException.ThrowIf(disposed,this);
            Volatile.Write(ref subscriptions,[..subscriptions,item]);
        }
        return new ActionDisposable(()=>
        {
            // Already-entered callbacks may finish; cancellation cannot recall them.
            Interlocked.Exchange(ref item.Active,0);
            lock(sync) Volatile.Write(ref subscriptions,subscriptions.Where(x=>x!=item).ToArray());
        });
    }
    public byte Send(Func<byte> original, Func<Func<int,bool>,CaptureRead> capture)
    {
        var listeners=Volatile.Read(ref subscriptions);
        NetworkMessage? message=null;
        if(listeners.Any(s=>s.Matches(NetworkPhase.BeforeSend) || s.Matches(NetworkPhase.AfterSend)))
            message=Read(PacketDirection.Up,capture,opcode=>listeners.Any(s=>
                s.Matches(NetworkPhase.BeforeSend,opcode) || s.Matches(NetworkPhase.AfterSend,opcode)));
        if(message!=null)
        {
            Interlocked.Increment(ref beforeSend);
            Publish(listeners,new(message,NetworkPhase.BeforeSend));
        }
        // Outside observer/read exception handlers: never retry the native call.
        var result=original();
        Volatile.Write(ref lastSendResult,result);
        if(message!=null)
        {
            Interlocked.Increment(ref afterSend);
            Publish(listeners,new(message,NetworkPhase.AfterSend,result));
        }
        return result;
    }
    public byte Receive(Func<byte> original, Func<Func<int,bool>,CaptureRead> capture)
    {
        var result=original();
        if(result==0) return result; // Holder is unreadable on unsuccessful receive.
        var listeners=Volatile.Read(ref subscriptions);
        if(listeners.Any(s=>s.Matches(NetworkPhase.Received)))
        {
            var message=Read(PacketDirection.Down,capture,opcode=>listeners.Any(s=>s.Matches(NetworkPhase.Received,opcode)));
            if(message!=null)
            {
                Interlocked.Increment(ref received);
                Publish(listeners,new(message,NetworkPhase.Received,result));
            }
        }
        return result;
    }
    private NetworkMessage? Read(PacketDirection direction,Func<Func<int,bool>,CaptureRead> capture,Func<int,bool> accepts)
    {
        try
        {
            var read=capture(accepts);
            if(read.Error!=null) { ReadFailed(read.Error); return null; }
            if(read.Packet is not {} packet) return null;
            if(!packet.Complete) ReadFailed(packet.Issue ?? "载荷读取不完整");
            var origin=direction==PacketDirection.Up && ActiveSendScope.Active ? PacketOrigin.PluginActiveSend : PacketOrigin.NativeZone;
            return new(Interlocked.Increment(ref sequence),direction,version,packet,origin);
        }
        catch(Exception ex) { ReadFailed(ex.Message); return null; }
    }
    private void ReadFailed(string issue)
    {
        Interlocked.Increment(ref readErrors); Volatile.Write(ref lastIssue,issue);
    }
    [ThreadStatic] private static int callbackDepth;
    public static bool InCallback => callbackDepth > 0;
    private void Publish(Subscription[] listeners,NetworkObservation observation)
    {
        callbackDepth++;
        try
        {
            foreach(var listener in listeners)
            {
                if(!listener.Matches(observation.Phase,observation.Message.Opcode)) continue;
                try { listener.Callback(observation); }
                catch(Exception ex)
                {
                    Interlocked.Increment(ref observerErrors);
                    Volatile.Write(ref lastIssue,$"{observation.Phase} 订阅异常：{ex.Message}");
                }
            }
        }
        finally
        {
            if(callbackDepth>0) callbackDepth--;
        }
    }
    public void Dispose()
    {
        lock(sync)
        {
            disposed=true;
            foreach(var listener in subscriptions) Interlocked.Exchange(ref listener.Active,0);
            Volatile.Write(ref subscriptions,[]);
        }
    }
}