ScadaBridge/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/Grpc/SiteStreamGrpcClientTests.cs

using Google.Protobuf.WellKnownTypes;
using Grpc.Core;
using ZB.MOM.WW.ScadaBridge.Communication.Grpc;
using ZB.MOM.WW.ScadaBridge.Commons.Messages.Streaming;
using ZB.MOM.WW.ScadaBridge.Commons.Types.Enums;

namespace ZB.MOM.WW.ScadaBridge.Communication.Tests.Grpc;

public class SiteStreamGrpcClientTests
{
    /// <summary>Verifies that the connected-header timeout is fixed for the lifetime of a client.</summary>
    [Fact]
    public void ConnectedHeaderTimeout_IsImmutablePerClient()
    {
        var property = typeof(SiteStreamGrpcClient).GetProperty("ConnectedHeaderTimeout",
            System.Reflection.BindingFlags.NonPublic | System.Reflection.BindingFlags.Instance | System.Reflection.BindingFlags.Static)!;
        Assert.False(property.GetMethod!.IsStatic);
        Assert.Null(property.SetMethod);
    }

    /// <summary>Verifies that an attribute-changed stream event converts to the matching domain event.</summary>
    [Fact]
    public void ConvertToDomainEvent_AttributeChanged_MapsCorrectly()
    {
        var ts = DateTimeOffset.UtcNow;
        var evt = new SiteStreamEvent
        {
            CorrelationId = "corr-1",
            AttributeChanged = new AttributeValueUpdate
            {
                InstanceUniqueName = "Site1.Pump01",
                AttributePath = "Modules.IO",
                AttributeName = "Temperature",
                Value = "42.5",
                Quality = Quality.Good,
                Timestamp = Timestamp.FromDateTimeOffset(ts)
            }
        };

        var result = SiteStreamGrpcClient.ConvertToDomainEvent(evt);

        var attr = Assert.IsType<AttributeValueChanged>(result);
        Assert.Equal("Site1.Pump01", attr.InstanceUniqueName);
        Assert.Equal("Modules.IO", attr.AttributePath);
        Assert.Equal("Temperature", attr.AttributeName);
        Assert.Equal("42.5", attr.Value);
        Assert.Equal("Good", attr.Quality);
        Assert.Equal(ts, attr.Timestamp);
    }

    /// <summary>Verifies that an alarm-changed stream event converts to the matching domain event.</summary>
    [Fact]
    public void ConvertToDomainEvent_AlarmChanged_MapsCorrectly()
    {
        var ts = DateTimeOffset.UtcNow;
        var raiseTime = ts.AddMinutes(-30);
        var evt = new SiteStreamEvent
        {
            CorrelationId = "corr-2",
            AlarmChanged = new AlarmStateUpdate
            {
                InstanceUniqueName = "Site1.Motor01",
                AlarmName = "T01.Hi",
                State = AlarmStateEnum.AlarmStateActive,
                Priority = 850,
                Timestamp = Timestamp.FromDateTimeOffset(ts),
                Kind = "NativeOpcUa",
                Active = true,
                Acknowledged = false,
                Confirmed = false,
                ShelveState = "TimedShelved",
                Suppressed = true,
                SourceReference = "T01.Hi",
                AlarmTypeName = "AnalogLimit.Hi",
                Category = "Process",
                OperatorUser = "op2",
                OperatorComment = "shelved",
                OriginalRaiseTime = Timestamp.FromDateTimeOffset(raiseTime),
                CurrentValue = "120",
                LimitValue = "100"
            }
        };

        var result = SiteStreamGrpcClient.ConvertToDomainEvent(evt);

        var alarm = Assert.IsType<AlarmStateChanged>(result);
        Assert.Equal("Site1.Motor01", alarm.InstanceUniqueName);
        Assert.Equal("T01.Hi", alarm.AlarmName);
        Assert.Equal(AlarmState.Active, alarm.State);
        Assert.Equal(850, alarm.Priority);
        Assert.Equal(ts, alarm.Timestamp);

        // Native enrichment mapped back.
        Assert.Equal(AlarmKind.NativeOpcUa, alarm.Kind);
        Assert.True(alarm.Condition.Active);
        Assert.False(alarm.Condition.Acknowledged);
        Assert.Equal(AlarmShelveState.TimedShelved, alarm.Condition.Shelve);
        Assert.True(alarm.Condition.Suppressed);
        Assert.Equal(850, alarm.Condition.Severity);   // OPC UA: wire value is Part 9 Severity already
        Assert.Equal(850, alarm.Condition.Priority);
        Assert.Equal("T01.Hi", alarm.SourceReference);
        Assert.Equal("AnalogLimit.Hi", alarm.AlarmTypeName);
        Assert.Equal("Process", alarm.Category);
        Assert.Equal("op2", alarm.OperatorUser);
        Assert.Equal("shelved", alarm.OperatorComment);
        Assert.Equal(raiseTime, alarm.OriginalRaiseTime);
        Assert.Equal("120", alarm.CurrentValue);
        Assert.Equal("100", alarm.LimitValue);
    }

    /// <summary>
    /// Gitea #129: the wire carries the raw priority. MxAccess and computed rows band it onto a
    /// higher-is-more-urgent Severity (AVEVA priority 1 → 1000); the raw value stays on Priority.
    /// The wire meaning is unchanged, so this is also right for a site on an older build.
    /// </summary>
    [Theory]
    [InlineData("NativeMxAccess", 1, 1000)]
    [InlineData("NativeMxAccess", 400, 750)]
    [InlineData("NativeMxAccess", 950, 250)]
    [InlineData("Computed", 100, 1000)]
    [InlineData("Computed", 700, 500)]
    [InlineData("NativeOpcUa", 950, 950)]
    public void ConvertToDomainEvent_AlarmChanged_SeverityIsOneDirection(string kind, int wirePriority, int expectedSeverity)
    {
        var evt = new SiteStreamEvent
        {
            AlarmChanged = new AlarmStateUpdate
            {
                InstanceUniqueName = "Site1.Motor01",
                AlarmName = "A",
                State = AlarmStateEnum.AlarmStateActive,
                Priority = wirePriority,
                Timestamp = Timestamp.FromDateTimeOffset(DateTimeOffset.UnixEpoch),
                Kind = kind,
                Active = true
            }
        };

        var alarm = Assert.IsType<AlarmStateChanged>(SiteStreamGrpcClient.ConvertToDomainEvent(evt));

        Assert.Equal(expectedSeverity, alarm.Condition.Severity);
        Assert.Equal(wirePriority, alarm.Condition.Priority);
        Assert.Equal(wirePriority, alarm.Priority);
    }

    /// <summary>Verifies that an unknown stream event converts to null.</summary>
    [Fact]
    public void ConvertToDomainEvent_UnknownEvent_ReturnsNull()
    {
        var evt = new SiteStreamEvent
        {
            CorrelationId = "corr-3"
            // No oneof case set
        };

        var result = SiteStreamGrpcClient.ConvertToDomainEvent(evt);

        Assert.Null(result);
    }

    /// <summary>Verifies that each gRPC quality value maps to the expected quality string.</summary>
    /// <param name="input">The gRPC quality value.</param>
    /// <param name="expected">The expected quality string.</param>
    [Theory]
    [InlineData(Quality.Good, "Good")]
    [InlineData(Quality.Uncertain, "Uncertain")]
    [InlineData(Quality.Bad, "Bad")]
    [InlineData(Quality.Unspecified, "Unknown")]
    public void MapQuality_AllValues(Quality input, string expected)
    {
        var result = SiteStreamGrpcClient.MapQuality(input);
        Assert.Equal(expected, result);
    }

    /// <summary>Verifies that each gRPC alarm state maps to the expected domain alarm state.</summary>
    /// <param name="input">The gRPC alarm state.</param>
    /// <param name="expected">The expected domain alarm state.</param>
    [Theory]
    [InlineData(AlarmStateEnum.AlarmStateNormal, AlarmState.Normal)]
    [InlineData(AlarmStateEnum.AlarmStateActive, AlarmState.Active)]
    [InlineData(AlarmStateEnum.AlarmStateUnspecified, AlarmState.Normal)]
    public void MapAlarmState_AllValues(AlarmStateEnum input, AlarmState expected)
    {
        var result = SiteStreamGrpcClient.MapAlarmState(input);
        Assert.Equal(expected, result);
    }

    /// <summary>Verifies that each gRPC alarm level maps to the expected domain alarm level.</summary>
    /// <param name="input">The gRPC alarm level.</param>
    /// <param name="expected">The expected domain alarm level.</param>
    [Theory]
    [InlineData(AlarmLevelEnum.AlarmLevelNone, AlarmLevel.None)]
    [InlineData(AlarmLevelEnum.AlarmLevelLow, AlarmLevel.Low)]
    [InlineData(AlarmLevelEnum.AlarmLevelLowLow, AlarmLevel.LowLow)]
    [InlineData(AlarmLevelEnum.AlarmLevelHigh, AlarmLevel.High)]
    [InlineData(AlarmLevelEnum.AlarmLevelHighHigh, AlarmLevel.HighHigh)]
    public void MapAlarmLevel_AllValues(AlarmLevelEnum input, AlarmLevel expected)
    {
        var result = SiteStreamGrpcClient.MapAlarmLevel(input);
        Assert.Equal(expected, result);
    }

    /// <summary>Verifies that converting an alarm-changed event preserves the alarm level.</summary>
    [Fact]
    public void ConvertToDomainEvent_AlarmChanged_PreservesLevel()
    {
        // Round-trip: a HiLo alarm emitted at HighHigh must come through with Level intact.
        var evt = new SiteStreamEvent
        {
            CorrelationId = "test",
            AlarmChanged = new AlarmStateUpdate
            {
                InstanceUniqueName = "Pump1",
                AlarmName = "TempAlarm",
                State = AlarmStateEnum.AlarmStateActive,
                Priority = 900,
                Timestamp = Google.Protobuf.WellKnownTypes.Timestamp.FromDateTimeOffset(DateTimeOffset.UtcNow),
                Level = AlarmLevelEnum.AlarmLevelHighHigh
            }
        };

        var domain = SiteStreamGrpcClient.ConvertToDomainEvent(evt) as AlarmStateChanged;

        Assert.NotNull(domain);
        Assert.Equal(AlarmState.Active, domain.State);
        Assert.Equal(AlarmLevel.HighHigh, domain.Level);
        Assert.Equal(900, domain.Priority);
    }

    /// <summary>Verifies that unsubscribing cancels the subscription.</summary>
    [Fact]
    public void Unsubscribe_CancelsSubscription()
    {
        // We can't easily test the full Subscribe flow without a real gRPC server,
        // but we can test the Unsubscribe path by registering a CTS directly.
        // Use the internal AddSubscription helper for testability.
        var client = SiteStreamGrpcClient.CreateForTesting();

        var cts = new CancellationTokenSource();
        client.AddSubscriptionForTesting("corr-test", cts);

        Assert.False(cts.IsCancellationRequested);

        client.Unsubscribe("corr-test");

        Assert.True(cts.IsCancellationRequested);
    }

    /// <summary>Verifies that unsubscribing from a nonexistent subscription does not throw.</summary>
    [Fact]
    public void Unsubscribe_NonExistent_DoesNotThrow()
    {
        var client = SiteStreamGrpcClient.CreateForTesting();
        client.Unsubscribe("does-not-exist"); // Should not throw
    }

    /// <summary>Verifies that disposing the client cancels all subscriptions.</summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task DisposeAsync_CancelsAllSubscriptions()
    {
        var client = SiteStreamGrpcClient.CreateForTesting();

        var cts1 = new CancellationTokenSource();
        var cts2 = new CancellationTokenSource();
        client.AddSubscriptionForTesting("corr-1", cts1);
        client.AddSubscriptionForTesting("corr-2", cts2);

        await client.DisposeAsync();

        Assert.True(cts1.IsCancellationRequested);
        Assert.True(cts2.IsCancellationRequested);
    }

    // --- Site-wide (SubscribeSite) alarm-only stream ---

    /// <summary>Verifies that an alarm-changed event converts to the mapped alarm event.</summary>
    [Fact]
    public void ConvertToAlarmEvent_AlarmChanged_ReturnsMappedAlarm()
    {
        // An alarm event on the site-wide stream is delivered to onAlarmEvent with full enrichment.
        var ts = DateTimeOffset.UtcNow;
        var evt = new SiteStreamEvent
        {
            CorrelationId = "site-corr",
            AlarmChanged = new AlarmStateUpdate
            {
                InstanceUniqueName = "Site1.Motor01",
                AlarmName = "T01.Hi",
                State = AlarmStateEnum.AlarmStateActive,
                Priority = 700,
                Timestamp = Timestamp.FromDateTimeOffset(ts),
                Kind = "NativeOpcUa",
                Active = true,
                SourceReference = "T01.Hi"
            }
        };

        var alarm = SiteStreamGrpcClient.ConvertToAlarmEvent(evt);

        Assert.NotNull(alarm);
        Assert.Equal("Site1.Motor01", alarm!.InstanceUniqueName);
        Assert.Equal("T01.Hi", alarm.AlarmName);
        Assert.Equal(AlarmState.Active, alarm.State);
        Assert.Equal(AlarmKind.NativeOpcUa, alarm.Kind);
        Assert.Equal("T01.Hi", alarm.SourceReference);
    }

    /// <summary>Verifies that an attribute-changed event converts to no alarm event.</summary>
    [Fact]
    public void ConvertToAlarmEvent_AttributeChanged_ReturnsNull()
    {
        // Attribute events must never appear on the alarm-only site-wide stream; if one
        // somehow arrives it is defensively filtered out rather than delivered or thrown.
        var evt = new SiteStreamEvent
        {
            CorrelationId = "site-corr",
            AttributeChanged = new AttributeValueUpdate
            {
                InstanceUniqueName = "Site1.Pump01",
                AttributePath = "Modules.IO",
                AttributeName = "Temperature",
                Value = "42.5",
                Quality = Quality.Good,
                Timestamp = Timestamp.FromDateTimeOffset(DateTimeOffset.UtcNow)
            }
        };

        Assert.Null(SiteStreamGrpcClient.ConvertToAlarmEvent(evt));
    }

    /// <summary>Verifies that an unknown event converts to no alarm event.</summary>
    [Fact]
    public void ConvertToAlarmEvent_UnknownEvent_ReturnsNull()
    {
        var evt = new SiteStreamEvent { CorrelationId = "site-corr" };
        Assert.Null(SiteStreamGrpcClient.ConvertToAlarmEvent(evt));
    }

    /// <summary>Verifies that subscribing a site on a test-only client throws.</summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task SubscribeSiteAsync_OnTestOnlyClient_Throws()
    {
        // Guards against subscribing on a channel-less test double, mirroring SubscribeAsync.
        var client = SiteStreamGrpcClient.CreateForTesting();

        await Assert.ThrowsAsync<InvalidOperationException>(() =>
            client.SubscribeSiteAsync("corr", _ => { }, _ => { }, () => { }, CancellationToken.None));
    }

    // --- Graceful (status OK) stream completion is reported, not swallowed ---

    /// <summary>Verifies that a stream the server ends with an OK status invokes the completed callback and not the error callback.</summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task ConsumeStream_ServerEndsStreamWithOk_InvokesOnCompleted_NotOnError()
    {
        // The site server caps every stream at GrpcMaxStreamLifetime (4h) and then ends the
        // RPC with OK. That surfaces here as a read loop that simply runs out of events.
        var client = SiteStreamGrpcClient.CreateForTesting();
        var cts = new CancellationTokenSource();
        var events = new List<SiteStreamEvent>();
        Exception? error = null;
        var completed = 0;

        await client.ConsumeStreamAsync(
            "corr-ok",
            cts,
            () => FakeCall(new StubStreamReader(new SiteStreamEvent { CorrelationId = "corr-ok" })),
            events.Add,
            ex => error = ex,
            () => completed++);

        Assert.Single(events);
        Assert.Null(error);
        Assert.Equal(1, completed);
    }

    /// <summary>Verifies that a faulting stream invokes the error callback and not the completed callback.</summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task ConsumeStream_StreamFaults_InvokesOnError_NotOnCompleted()
    {
        var client = SiteStreamGrpcClient.CreateForTesting();
        var cts = new CancellationTokenSource();
        Exception? error = null;
        var completed = 0;

        await client.ConsumeStreamAsync(
            "corr-fault",
            cts,
            () => FakeCall(new StubStreamReader(
                new RpcException(new Status(StatusCode.Unavailable, "site gone")))),
            _ => { },
            ex => error = ex,
            () => completed++);

        Assert.IsType<RpcException>(error);
        Assert.Equal(0, completed);
    }

    /// <summary>Verifies that cancelling the client's own token invokes neither the error nor the completed callback.</summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task ConsumeStream_OwnCancellation_InvokesNeitherCallback()
    {
        // Our own Unsubscribe/reconnect is a teardown, not a fault and not a graceful end:
        // the caller has already moved on to a newer stream.
        var client = SiteStreamGrpcClient.CreateForTesting();
        var cts = new CancellationTokenSource();
        await cts.CancelAsync();
        Exception? error = null;
        var completed = 0;

        await client.ConsumeStreamAsync(
            "corr-cancel",
            cts,
            () => FakeCall(new StubStreamReader()),
            _ => { },
            ex => error = ex,
            () => completed++);

        Assert.Null(error);
        Assert.Equal(0, completed);
    }

    // --- Foreign vs. own Cancelled (review F2) ---

    /// <summary>Verifies that a cancellation not initiated by the client invokes the error callback.</summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task ConsumeStream_ForeignCancelled_InvokesOnError()
    {
        // A Cancelled status we did NOT ask for — the PEER cancelled, or the channel was
        // disposed underneath us. It used to be swallowed by an unguarded
        // `when (ex.StatusCode == StatusCode.Cancelled)` filter, firing none of
        // onError/onCompleted/onConnected: the consuming aggregator kept _streamDown=false,
        // so IsLive stayed true forever and the reconcile tick's reopen guard never fired.
        var client = SiteStreamGrpcClient.CreateForTesting();
        var cts = new CancellationTokenSource(); // deliberately NOT cancelled
        Exception? error = null;
        var completed = 0;

        await client.ConsumeStreamAsync(
            "corr-foreign-cancel",
            cts,
            () => FakeCall(new StubStreamReader(
                new RpcException(new Status(StatusCode.Cancelled, "cancelled by peer")))),
            _ => { },
            ex => error = ex,
            () => completed++);

        var rpc = Assert.IsType<RpcException>(error);
        Assert.Equal(StatusCode.Cancelled, rpc.StatusCode);
        Assert.Equal(0, completed);
        Assert.False(cts.IsCancellationRequested);
    }

    /// <summary>Verifies that a cancelled RPC exception caused by the client's own cancellation invokes neither callback.</summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task ConsumeStream_OwnCancelledRpcException_InvokesNeitherCallback()
    {
        // The other side of the same filter: OUR cancellation surfacing as
        // RpcException(Cancelled) is a teardown, so neither callback fires.
        var client = SiteStreamGrpcClient.CreateForTesting();
        var cts = new CancellationTokenSource();
        await cts.CancelAsync();
        Exception? error = null;
        var completed = 0;

        await client.ConsumeStreamAsync(
            "corr-own-cancel",
            cts,
            () => throw new RpcException(new Status(StatusCode.Cancelled, "we cancelled")),
            _ => { },
            ex => error = ex,
            () => completed++);

        Assert.Null(error);
        Assert.Equal(0, completed);
    }

    // --- onConnected is only ever raised on real proof of a live peer (review F3) ---

    /// <summary>Verifies that the connected callback is invoked exactly once when the response headers arrive.</summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task ConsumeStream_HeadersArrive_InvokesOnConnectedOnce()
    {
        var client = SiteStreamGrpcClient.CreateForTesting();
        var cts = new CancellationTokenSource();
        var connected = 0;

        await client.ConsumeStreamAsync(
            "corr-headers",
            cts,
            () => FakeCall(
                new StubStreamReader(
                    new SiteStreamEvent { CorrelationId = "corr-headers" },
                    new SiteStreamEvent { CorrelationId = "corr-headers" }),
                Task.FromResult(new Metadata())),
            _ => { },
            _ => { },
            () => { },
            () => connected++);

        // Once — the two events must not re-raise it.
        Assert.Equal(1, connected);
    }

    /// <summary>Verifies that a header timeout on a dead site never reports a connected state.</summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task ConsumeStream_HeaderTimeoutOnADeadSite_NeverReportsConnected()
    {
        // The bug: a header TIMEOUT used to be reported as connected. An unreachable/wedged
        // site produces exactly that shape, so the aggregator cleared _streamDown, consumed
        // its pending re-seed and fanned a full snapshot out at a site that never answered.
        var client = SiteStreamGrpcClient.CreateForTesting(new CommunicationOptions { ConnectedHeaderTimeout = TimeSpan.FromMilliseconds(50) });
        var cts = new CancellationTokenSource();
        var connected = 0;
        var neverArrives = new TaskCompletionSource<Metadata>();

        // No headers AND no events: the site is dead. The stream ends (status OK) with
        // no connected signal ever raised.
        await client.ConsumeStreamAsync(
            "corr-dead",
            cts,
            () => FakeCall(new StubStreamReader(), neverArrives.Task),
            _ => { },
            _ => { },
            () => { },
            () => connected++);

        Assert.Equal(0, connected);
    }

    /// <summary>Verifies that a header timeout followed by an event reports connected exactly once, triggered by the event.</summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task ConsumeStream_HeaderTimeoutThenEvent_ReportsConnectedOnceFromTheEvent()
    {
        // The case the timeout was originally added for — a peer that defers its headers
        // until the first message — is now covered by the event itself, which is real proof
        // of a live peer. Connected must precede the event's delivery and fire only once.
        var client = SiteStreamGrpcClient.CreateForTesting(new CommunicationOptions { ConnectedHeaderTimeout = TimeSpan.FromMilliseconds(50) });
        var cts = new CancellationTokenSource();
        var connected = 0;
        var connectedBeforeFirstEvent = false;
        var events = 0;
        var neverArrives = new TaskCompletionSource<Metadata>();

        await client.ConsumeStreamAsync(
            "corr-late-headers",
            cts,
            () => FakeCall(
                new StubStreamReader(
                    new SiteStreamEvent { CorrelationId = "corr-late-headers" },
                    new SiteStreamEvent { CorrelationId = "corr-late-headers" }),
                neverArrives.Task),
            _ =>
            {
                if (events == 0) connectedBeforeFirstEvent = connected == 1;
                events++;
            },
            _ => { },
            () => { },
            () => connected++);

        Assert.Equal(2, events);
        Assert.Equal(1, connected);
        Assert.True(connectedBeforeFirstEvent);
    }

    // ── R2: batch unpacking on the client ───────────────────────────────────────

    private static SiteStreamEvent Attr(string value, DateTimeOffset ts) => new()
    {
        AttributeChanged = new AttributeValueUpdate
        {
        InstanceUniqueName = "SiteA.Pump01",
        AttributePath = "Modules.IO",
        AttributeName = "Seq",
        Value = value,
        Quality = Quality.Good,
        Timestamp = Timestamp.FromDateTimeOffset(ts)
        }
    };

    /// <summary>Verifies that a plain frame is delivered unchanged.</summary>
    [Fact]
    public void ForEachEvent_PlainFrame_IsDeliveredAsIs()
    {
        // An OLD SITE (or any un-negotiated stream) sends one event per frame. The new
        // client's unpack path must pass it straight through — this is the new-central ↔
        // old-site skew direction.
        var frame = Attr("1", DateTimeOffset.UnixEpoch);
        var seen = new List<SiteStreamEvent>();

        SiteStreamGrpcClient.ForEachEvent(frame, seen.Add);

        Assert.Same(frame, Assert.Single(seen));
    }

    /// <summary>Verifies that a batch frame is unpacked in order with each event's own timestamp preserved.</summary>
    [Fact]
    public void ForEachEvent_BatchFrame_UnpacksInOrderPreservingPerEventTimestamps()
    {
        // Order and per-event Timestamp fidelity are the two properties the downstream
        // consumers (SiteAlarmAggregatorActor, DebugStreamBridgeActor) and the end-to-end
        // latency measurement depend on.
        var t0 = new DateTimeOffset(2026, 8, 15, 9, 0, 0, TimeSpan.Zero);
        var frame = new SiteStreamEvent
        {
            CorrelationId = "corr-batch",
            Batch = new SiteStreamEventBatch
            {
                Events =
                {
                    Attr("0", t0),
                    Attr("1", t0.AddMilliseconds(3)),
                    Attr("2", t0.AddMilliseconds(11))
                }
            }
        };

        var seen = new List<SiteStreamEvent>();
        SiteStreamGrpcClient.ForEachEvent(frame, seen.Add);

        Assert.Equal(["0", "1", "2"], seen.Select(e => e.AttributeChanged.Value));
        Assert.Equal(
            [t0, t0.AddMilliseconds(3), t0.AddMilliseconds(11)],
            seen.Select(e => e.AttributeChanged.Timestamp.ToDateTimeOffset()));
    }

    /// <summary>Verifies that nested batches and unknown inner cases are ignored.</summary>
    [Fact]
    public void ForEachEvent_IgnoresNestedAndUnknownInnerCases()
    {
        // The server never nests a batch inside a batch. A nested (or empty) inner frame
        // from a malformed or hostile peer must be skipped, not followed — unpacking is
        // deliberately non-recursive so a crafted frame cannot drive unbounded recursion.
        var frame = new SiteStreamEvent
        {
            CorrelationId = "corr-nested",
            Batch = new SiteStreamEventBatch
            {
                Events =
                {
                    Attr("0", DateTimeOffset.UnixEpoch),
                    new SiteStreamEvent { Batch = new SiteStreamEventBatch { Events = { Attr("hidden", DateTimeOffset.UnixEpoch) } } },
                    new SiteStreamEvent(),
                    Attr("1", DateTimeOffset.UnixEpoch)
                }
            }
        };

        var seen = new List<SiteStreamEvent>();
        SiteStreamGrpcClient.ForEachEvent(frame, seen.Add);

        Assert.Equal(["0", "1"], seen.Select(e => e.AttributeChanged.Value));
    }

    /// <summary>Verifies that a stream mixing batched and plain frames delivers every event in order.</summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task ConsumeStream_MixedBatchedAndPlainFrames_DeliverEveryEventInOrder()
    {
        // A reconnect can straddle a site upgrade, so one stream may legitimately carry
        // both frame shapes. Driving the real ConsumeStreamAsync with the real unpack
        // proves the combination is flat and ordered from the consumer's point of view.
        var client = SiteStreamGrpcClient.CreateForTesting();
        var cts = new CancellationTokenSource();
        var delivered = new List<string>();

        void Deliver(SiteStreamEvent e) => delivered.Add(e.AttributeChanged.Value);

        await client.ConsumeStreamAsync(
            "corr-mixed",
            cts,
            () => FakeCall(new StubStreamReader(
                Attr("0", DateTimeOffset.UnixEpoch),
                new SiteStreamEvent
                {
                    CorrelationId = "corr-mixed",
                    Batch = new SiteStreamEventBatch
                    {
                        Events = { Attr("1", DateTimeOffset.UnixEpoch), Attr("2", DateTimeOffset.UnixEpoch) }
                    }
                },
                Attr("3", DateTimeOffset.UnixEpoch))),
            frame => SiteStreamGrpcClient.ForEachEvent(frame, Deliver),
            _ => { },
            () => { });

        Assert.Equal(["0", "1", "2", "3"], delivered);
    }

    private static AsyncServerStreamingCall<SiteStreamEvent> FakeCall(StubStreamReader reader) =>
        FakeCall(reader, Task.FromResult(new Metadata()));

    private static AsyncServerStreamingCall<SiteStreamEvent> FakeCall(
        StubStreamReader reader, Task<Metadata> responseHeaders) =>
        new(reader,
            responseHeaders,
            () => Status.DefaultSuccess,
            () => new Metadata(),
            () => { });

    /// <summary>
    /// Server stream stand-in: yields the queued events, then either ends the stream (the
    /// status-OK completion the site's max-lifetime cap produces) or throws.
    /// </summary>
    private sealed class StubStreamReader : IAsyncStreamReader<SiteStreamEvent>
    {
        private readonly Queue<SiteStreamEvent> _events;
        private readonly Exception? _fault;

        /// <summary>Initializes the reader to yield the given events and then end the stream.</summary>
        /// <param name="events">The events to yield.</param>
        public StubStreamReader(params SiteStreamEvent[] events)
        {
            _events = new Queue<SiteStreamEvent>(events);
        }

        /// <summary>Initializes the reader to throw the given fault when read.</summary>
        /// <param name="fault">The exception to throw.</param>
        public StubStreamReader(Exception fault)
            : this()
        {
            _fault = fault;
        }

        /// <summary>Gets the event most recently read.</summary>
        public SiteStreamEvent Current { get; private set; } = null!;

        /// <summary>Advances to the next queued event, or faults or completes as configured.</summary>
        /// <param name="cancellationToken">A token to cancel the operation.</param>
        /// <returns>A task whose result is <see langword="true"/> if an event was read; otherwise <see langword="false"/>.</returns>
        public Task<bool> MoveNext(CancellationToken cancellationToken)
        {
            cancellationToken.ThrowIfCancellationRequested();

            if (_events.Count > 0)
            {
                Current = _events.Dequeue();
                return Task.FromResult(true);
            }

            return _fault is null ? Task.FromResult(false) : Task.FromException<bool>(_fault);
        }
    }

    // --- Subscription registration regression tests ---

    /// <summary>Verifies that reusing a correlation id cancels the prior token source without disposing it.</summary>
    [Fact]
    public void RegisterSubscription_ReusedCorrelationId_CancelsPriorCts_WithoutDisposingIt()
    {
        // Two SubscribeAsync calls briefly sharing a correlation ID (reconnect race).
        // Inserting the second must CANCEL the first — and must NOT dispose it (re-pinning the earlier
        // cancel-AND-dispose contract): the prior
        // handler's ConsumeStreamAsync is still running and still reads prior.Token, so
        // disposing from the replacing path throws ObjectDisposedException inside that loop.
        // The owning SubscribeAsync/SubscribeSiteAsync disposes it as its loop unwinds.
        var client = SiteStreamGrpcClient.CreateForTesting();

        using var first = new CancellationTokenSource();
        using var second = new CancellationTokenSource();

        client.RegisterSubscription("corr-shared", first);
        client.RegisterSubscription("corr-shared", second);

        Assert.True(first.IsCancellationRequested);
        // Still usable by the prior handler that owns it — no ObjectDisposedException.
        Assert.True(first.Token.IsCancellationRequested);

        // The second (live) CTS must remain intact.
        Assert.False(second.IsCancellationRequested);
    }

    /// <summary>Verifies that replacing an already-disposed prior token source does not throw into the new subscription.</summary>
    [Fact]
    public void RegisterSubscription_ReplacingAnAlreadyDisposedPriorCts_DoesNotThrowIntoTheNewSubscription()
    {
        // The prior handler's loop already unwound and disposed its own CTS while a replacement
        // subscription is registering. The unguarded prior.Cancel() threw ObjectDisposedException
        // straight out of the NEW SubscribeAsync, which reads it as a fault of the CURRENT
        // generation and burns a reconnect on a phantom failure.
        var client = SiteStreamGrpcClient.CreateForTesting();

        var prior = new CancellationTokenSource();
        client.RegisterSubscription("corr-shared", prior);
        prior.Dispose();

        using var replacement = new CancellationTokenSource();
        client.RegisterSubscription("corr-shared", replacement);

        // The replacement owns the slot and is untouched.
        Assert.False(replacement.IsCancellationRequested);
        client.Unsubscribe("corr-shared");
        Assert.True(replacement.IsCancellationRequested);
    }

    /// <summary>Verifies that unsubscribing after the client is disposed does nothing.</summary>
    [Fact]
    public void Unsubscribe_AfterClientDisposal_IsANoOp()
    {
        // Disposal cancels every live subscription and clears the map; a late Unsubscribe from
        // the consuming actor's teardown must not throw (an ObjectDisposedException here used to
        // escape into the caller's cleanup path).
        var client = SiteStreamGrpcClient.CreateForTesting();
        using var cts = new CancellationTokenSource();
        client.AddSubscriptionForTesting("corr-1", cts);

        client.Dispose();
        Assert.True(cts.IsCancellationRequested);

        client.Unsubscribe("corr-1");
        client.Unsubscribe("never-registered");
    }

    /// <summary>Verifies that disposing the client still cancels the remaining subscriptions when one token source is already disposed.</summary>
    [Fact]
    public void Dispose_WithAnAlreadyDisposedSubscriptionCts_StillCancelsTheRest()
    {
        // ReleaseResources iterated the live CTSes unguarded, so one already-disposed entry
        // aborted the whole loop — leaving later subscriptions uncancelled AND (via the
        // exception escaping into SiteStreamGrpcClientFactory's disposal loop) later clients'
        // channels undisposed.
        var client = SiteStreamGrpcClient.CreateForTesting();

        var disposedEarly = new CancellationTokenSource();
        using var live = new CancellationTokenSource();
        client.AddSubscriptionForTesting("corr-disposed", disposedEarly);
        client.AddSubscriptionForTesting("corr-live", live);
        disposedEarly.Dispose();

        // Deterministic regardless of dictionary iteration order: the disposed entry is always
        // visited, so the unguarded loop always threw out of Dispose().
        Assert.Null(Record.Exception(client.Dispose));

        Assert.True(live.IsCancellationRequested);
    }

    /// <summary>Verifies that removing a subscription removes only its own token source and not a replacement registered under the same id.</summary>
    [Fact]
    public void RemoveSubscription_OnlyRemovesOwnCts_NotAReplacement()
    {
        // First call's finally must NOT remove the second call's live entry.
        var client = SiteStreamGrpcClient.CreateForTesting();

        var first = new CancellationTokenSource();
        var second = new CancellationTokenSource();

        client.RegisterSubscription("corr-shared", first);
        // A racing second SubscribeAsync replaces the entry.
        client.RegisterSubscription("corr-shared", second);

        // The first call's finally runs and tries to remove its (already-replaced) entry.
        client.RemoveSubscription("corr-shared", first);

        // The live (second) subscription must still be cancellable via Unsubscribe.
        Assert.False(second.IsCancellationRequested);
        client.Unsubscribe("corr-shared");
        Assert.True(second.IsCancellationRequested);
    }
}