ScadaBridge/tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests/Streaming/SiteStreamManagerTests.cs

using Akka.Actor;
using Akka.TestKit.Xunit2;
using Microsoft.Extensions.Logging.Abstractions;
using ZB.MOM.WW.ScadaBridge.Commons.Messages.Streaming;
using ZB.MOM.WW.ScadaBridge.Commons.Types.Enums;
using ZB.MOM.WW.ScadaBridge.SiteRuntime.Streaming;

namespace ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests.Streaming;

/// <summary>
/// WP-23: Site-Wide Akka Stream tests.
/// WP-25: Debug View Backend tests (subscribe/unsubscribe).
/// </summary>
public class SiteStreamManagerTests : TestKit, IDisposable
{
    private readonly SiteStreamManager _streamManager;

    public SiteStreamManagerTests()
    {
        var options = new SiteRuntimeOptions { StreamBufferSize = 100 };
        _streamManager = new SiteStreamManager(
            options, NullLogger<SiteStreamManager>.Instance);
        _streamManager.Initialize(Sys);
    }

    void IDisposable.Dispose()
    {
        Shutdown();
    }

    [Fact]
    public void Subscribe_CreatesSubscription()
    {
        var probe = CreateTestProbe();
        var id = _streamManager.Subscribe("Pump1", probe.Ref);

        Assert.NotNull(id);
        Assert.Equal(1, _streamManager.SubscriptionCount);
    }

    [Fact]
    public void Unsubscribe_RemovesSubscription()
    {
        var probe = CreateTestProbe();
        var id = _streamManager.Subscribe("Pump1", probe.Ref);

        Assert.True(_streamManager.Unsubscribe(id));
        Assert.Equal(0, _streamManager.SubscriptionCount);
    }

    [Fact]
    public void Unsubscribe_InvalidId_ReturnsFalse()
    {
        Assert.False(_streamManager.Unsubscribe("nonexistent"));
    }

    [Fact]
    public void PublishAttributeValueChanged_ForwardsToSubscriber()
    {
        var probe = CreateTestProbe();
        _streamManager.Subscribe("Pump1", probe.Ref);

        var changed = new AttributeValueChanged(
            "Pump1", "Temperature", "Temperature", "100", "Good", DateTimeOffset.UtcNow);
        _streamManager.PublishAttributeValueChanged(changed);

        var received = probe.ExpectMsg<AttributeValueChanged>(TimeSpan.FromSeconds(3));
        Assert.Equal("Pump1", received.InstanceUniqueName);
        Assert.Equal("Temperature", received.AttributeName);
    }

    [Fact]
    public void PublishAlarmStateChanged_ForwardsToSubscriber()
    {
        var probe = CreateTestProbe();
        _streamManager.Subscribe("Pump1", probe.Ref);

        var changed = new AlarmStateChanged(
            "Pump1", "HighTemp", AlarmState.Active, 1, DateTimeOffset.UtcNow);
        _streamManager.PublishAlarmStateChanged(changed);

        var received = probe.ExpectMsg<AlarmStateChanged>(TimeSpan.FromSeconds(3));
        Assert.Equal("Pump1", received.InstanceUniqueName);
        Assert.Equal(AlarmState.Active, received.State);
    }

    [Fact]
    public void PublishAttributeValueChanged_FiltersbyInstance()
    {
        var probe1 = CreateTestProbe();
        var probe2 = CreateTestProbe();
        _streamManager.Subscribe("Pump1", probe1.Ref);
        _streamManager.Subscribe("Pump2", probe2.Ref);

        var changed = new AttributeValueChanged(
            "Pump1", "Temperature", "Temperature", "100", "Good", DateTimeOffset.UtcNow);
        _streamManager.PublishAttributeValueChanged(changed);

        // Pump1 subscriber should receive
        probe1.ExpectMsg<AttributeValueChanged>(TimeSpan.FromSeconds(3));

        // Pump2 subscriber should NOT receive
        probe2.ExpectNoMsg(TimeSpan.FromMilliseconds(500));
    }

    [Fact]
    public void SubscribeSiteAlarms_ForwardsAlarmsForAllInstances_ButNotAttributes()
    {
        var probe = CreateTestProbe();
        _streamManager.SubscribeSiteAlarms(probe.Ref);

        // Alarm events from two different instances — both should arrive.
        _streamManager.PublishAlarmStateChanged(new AlarmStateChanged(
            "Pump1", "HighTemp", AlarmState.Active, 1, DateTimeOffset.UtcNow));
        _streamManager.PublishAlarmStateChanged(new AlarmStateChanged(
            "Pump2", "LowFlow", AlarmState.Active, 2, DateTimeOffset.UtcNow));

        // Attribute events must be filtered out entirely.
        _streamManager.PublishAttributeValueChanged(new AttributeValueChanged(
            "Pump1", "Temperature", "Temperature", "100", "Good", DateTimeOffset.UtcNow));
        _streamManager.PublishAttributeValueChanged(new AttributeValueChanged(
            "Pump3", "Flow", "Flow", "42", "Good", DateTimeOffset.UtcNow));

        // Collect the two alarms (order across instances is not guaranteed).
        var received = new[]
        {
            probe.ExpectMsg<AlarmStateChanged>(TimeSpan.FromSeconds(3)),
            probe.ExpectMsg<AlarmStateChanged>(TimeSpan.FromSeconds(3)),
        };

        var instances = received.Select(a => a.InstanceUniqueName).OrderBy(n => n).ToArray();
        Assert.Equal(new[] { "Pump1", "Pump2" }, instances);

        // No attribute events (nor any further alarm) should be delivered.
        probe.ExpectNoMsg(TimeSpan.FromMilliseconds(500));
    }

    [Fact]
    public void SubscribeSiteAlarms_SubscriptionRemovableViaUnsubscribe()
    {
        var probe = CreateTestProbe();
        var id = _streamManager.SubscribeSiteAlarms(probe.Ref);

        Assert.NotNull(id);
        Assert.Equal(1, _streamManager.SubscriptionCount);
        Assert.True(_streamManager.Unsubscribe(id));
        Assert.Equal(0, _streamManager.SubscriptionCount);
    }

    [Fact]
    public void RemoveSubscriber_RemovesAllSubscriptionsForActor()
    {
        var probe = CreateTestProbe();
        _streamManager.Subscribe("Pump1", probe.Ref);
        _streamManager.Subscribe("Pump2", probe.Ref);

        Assert.Equal(2, _streamManager.SubscriptionCount);

        _streamManager.RemoveSubscriber(probe.Ref);
        Assert.Equal(0, _streamManager.SubscriptionCount);
    }

    // ── WP2.6d: separate alarm publish path, drop counter, skip-at-zero-subscribers ──

    /// <summary>
    /// WP2.6d: alarm state changes now travel a dedicated publish source, isolated from
    /// the (far higher-volume) attribute path — a burst of attribute events for OTHER
    /// instances, interleaved with an alarm-only subscriber's events, must not cause any
    /// alarm to be lost.
    /// </summary>
    [Fact]
    public void PublishAlarmStateChanged_SurvivesConcurrentAttributeStorm_ForUnrelatedInstances()
    {
        var alarmProbe = CreateTestProbe();
        _streamManager.SubscribeSiteAlarms(alarmProbe.Ref);

        // A storm of attribute events for a DIFFERENT instance — none of which the
        // alarm subscriber is even listening to — interleaved with alarm events. Before
        // WP2.6d these shared one upstream buffer; now they are fully separate sources.
        for (var i = 0; i < 500; i++)
        {
            _streamManager.PublishAttributeValueChanged(new AttributeValueChanged(
                "NoisyPump", "Temperature", "Temperature", i.ToString(), "Good", DateTimeOffset.UtcNow));
        }

        _streamManager.PublishAlarmStateChanged(new AlarmStateChanged(
            "Pump1", "HighTemp", AlarmState.Active, 1, DateTimeOffset.UtcNow));

        var received = alarmProbe.ExpectMsg<AlarmStateChanged>(TimeSpan.FromSeconds(3));
        Assert.Equal("Pump1", received.InstanceUniqueName);
    }

    /// <summary>
    /// WP2.6d: with zero subscribers of any kind, PublishAlarmStateChanged/
    /// PublishAttributeValueChanged must be no-ops — no exception, and (for alarms) the
    /// event never reaches the bounded hand-off queue, so
    /// <see cref="SiteStreamManager.AlarmPublishDroppedCount"/> stays at zero rather than
    /// counting events nobody could ever have received anyway.
    /// </summary>
    [Fact]
    public void Publish_WithNoSubscribers_IsNoOp_AndDoesNotCountAsDropped()
    {
        _streamManager.PublishAlarmStateChanged(new AlarmStateChanged(
            "Pump1", "HighTemp", AlarmState.Active, 1, DateTimeOffset.UtcNow));
        _streamManager.PublishAttributeValueChanged(new AttributeValueChanged(
            "Pump1", "Temperature", "Temperature", "1", "Good", DateTimeOffset.UtcNow));

        Assert.Equal(0, _streamManager.AlarmPublishDroppedCount);

        // Publishing resumes working normally once a subscriber exists.
        var probe = CreateTestProbe();
        _streamManager.SubscribeSiteAlarms(probe.Ref);
        _streamManager.PublishAlarmStateChanged(new AlarmStateChanged(
            "Pump1", "HighTemp", AlarmState.Active, 2, DateTimeOffset.UtcNow));

        probe.ExpectMsg<AlarmStateChanged>(TimeSpan.FromSeconds(3));
    }

    /// <summary>
    /// WP2.6d: once the bounded alarm hand-off queue (the exact factory
    /// <see cref="SiteStreamManager.Initialize"/> wires into the drop-count callback) is
    /// full, a further write evicts the oldest queued item and fires the drop callback.
    /// Exercised directly against the queue factory — deterministic, no actor
    /// system/async pump race involved (see the factory's own doc comment).
    /// </summary>
    [Fact]
    public void CreateAlarmPublishQueue_OverCapacity_DropsOldestAndInvokesCallback()
    {
        var dropped = 0;
        var queue = SiteStreamManager.CreateAlarmPublishQueue(capacity: 2, () => Interlocked.Increment(ref dropped));

        // Nothing reads from this queue, so all five writes stay purely upstream —
        // capacity 2 means the 3rd/4th/5th writes must each evict the oldest.
        for (var i = 0; i < 5; i++)
        {
            var wrote = queue.Writer.TryWrite(new AlarmStateChanged(
                "Pump1", "HighTemp", AlarmState.Active, i, DateTimeOffset.UtcNow));
            Assert.True(wrote); // DropOldest TryWrite always succeeds once capacity > 0
        }

        Assert.Equal(3, dropped);
        Assert.Equal(2, queue.Reader.Count);
    }
}