ScadaBridge/tests/ZB.MOM.WW.ScadaBridge.LoadHarness/Probes/WireStreamSubscriber.cs

using ZB.MOM.WW.ScadaBridge.Commons.Messages.Streaming;
using ZB.MOM.WW.ScadaBridge.Communication.Grpc;
using ZB.MOM.WW.ScadaBridge.LoadHarness.Metrics;

namespace ZB.MOM.WW.ScadaBridge.LoadHarness.Probes;

/// <summary>
/// The central half of the wire-path stream probe: one instance subscription opened
/// through the PRODUCTION <see cref="SiteStreamGrpcClient"/> — the exact
/// <c>SubscribeAsync</c> path central's <c>DebugStreamBridgeActor</c> uses, including
/// the R2 batch negotiation (<c>BatchingSupported = true</c>) and the client-side
/// batch unpack + proto→domain conversion.
///
/// <para>
/// Latency is recorded in <c>onEvent</c>, i.e. AFTER wire deserialization, batch
/// unpacking and domain mapping — the same "DCL boundary → subscriber" measurement as
/// <see cref="StreamSubscriberProbe"/>, extended across the real socket. The emit
/// instant travels verbatim on the proto <c>Timestamp</c>, so the subtraction stays a
/// genuine end-to-end figure (same process, same clock).
/// </para>
/// </summary>
public sealed class WireStreamSubscriber
{
    private Task _streamTask = Task.CompletedTask;
    private volatile LatencyHistogram? _latency;
    private long _received;
    private long _faults;
    private long _completions;

    /// <summary>Probe name (also the stream's correlation id).</summary>
    public string Name { get; }

    /// <summary>Domain events delivered by the client callback.</summary>
    public long ReceivedEvents => Interlocked.Read(ref _received);

    /// <summary>Stream faults reported via <c>onError</c>. Expected 0 for a healthy run.</summary>
    public long Faults => Interlocked.Read(ref _faults);

    /// <summary>Graceful server-side completions via <c>onCompleted</c>. Expected 0 inside a run shorter than the 4 h stream lifetime.</summary>
    public long GracefulCompletions => Interlocked.Read(ref _completions);

    private WireStreamSubscriber(string name, LatencyHistogram? latency)
    {
        Name = name;
        _latency = latency;
    }

    /// <summary>
    /// Opens one instance stream on the shared per-site client. The subscription task
    /// runs until <paramref name="ct"/> cancels — the harness deliberately does NOT
    /// reconnect on fault (a fault inside a 25-minute loopback run is a finding, not
    /// an operational event to paper over), it counts it.
    /// </summary>
    /// <param name="client">The site's shared production gRPC client (one channel per site, streams multiplexed).</param>
    /// <param name="instanceUniqueName">Instance whose events this probe receives.</param>
    /// <param name="name">Probe/correlation name (must be a valid Akka path element).</param>
    /// <param name="latency">Histogram fed with end-to-end event latency.</param>
    /// <param name="log">Progress sink for fault reporting.</param>
    /// <param name="ct">Stops the subscription.</param>
    /// <returns>The attached subscriber.</returns>
    public static WireStreamSubscriber Attach(
        SiteStreamGrpcClient client,
        string instanceUniqueName,
        string name,
        LatencyHistogram? latency,
        Action<string> log,
        CancellationToken ct)
    {
        var self = new WireStreamSubscriber(name, latency);

        self._streamTask = client.SubscribeAsync(
            correlationId: name,
            instanceUniqueName: instanceUniqueName,
            onEvent: evt =>
            {
                Interlocked.Increment(ref self._received);
                if (self._latency is { } histogram && evt is AttributeValueChanged changed)
                    histogram.Record(DateTimeOffset.UtcNow - changed.Timestamp);
            },
            onError: ex =>
            {
                Interlocked.Increment(ref self._faults);
                log($"  wire probe {name} stream fault: {ex.GetType().Name}: {ex.Message}");
            },
            onCompleted: () => Interlocked.Increment(ref self._completions),
            ct);

        return self;
    }

    /// <summary>Repoints the latency histogram (steady-state window separation) — mirrors <see cref="StreamSubscriberProbe.RetargetLatency"/>.</summary>
    /// <param name="latency">The histogram to record into from now on, or null to stop recording.</param>
    public void RetargetLatency(LatencyHistogram? latency) => _latency = latency;

    /// <summary>Awaits the subscription task after its token has been cancelled.</summary>
    /// <returns>A task that completes when the stream loop has exited.</returns>
    public async Task WaitForShutdownAsync()
    {
        try
        {
            await _streamTask;
        }
        catch (OperationCanceledException)
        {
            // Normal teardown.
        }
    }
}