lmxopcua/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/GalaxyDriverProbeWiringTests.cs

using System.Runtime.CompilerServices;
using System.Threading.Channels;
using Google.Protobuf.WellKnownTypes;
using Shouldly;
using Xunit;
using ZB.MOM.WW.MxGateway.Contracts.Proto;
using ZB.MOM.WW.MxGateway.Contracts.Proto.Galaxy;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
using ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Browse;
using ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Config;
using ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Runtime;

namespace ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests.Runtime;

/// <summary>
///     Review 2026-10-21 G-4 / G-7 / G-11: the per-platform <c>ScanState</c> probing was structurally
///     dead (its handles were never in the registry the pump resolves through, and a reconnect
///     never re-subscribed them), discovery fetched the full galaxy twice, and
///     <c>LastSuccessfulRead</c> never moved on data. These drive the real driver → pump → watcher
///     → aggregator path with a fake subscriber and a fake hierarchy.
/// </summary>
public sealed class GalaxyDriverProbeWiringTests
{
    private const int WaitMs = 10_000;

    private static GalaxyDriverOptions Opts() => new(
        new GalaxyGatewayOptions("https://mxgw.test:5001", "key"),
        new GalaxyMxAccessOptions("ProbeWiring"),
        new GalaxyRepositoryOptions(WatchDeployEvents: false),
        new GalaxyReconnectOptions());

    private sealed class CountingHierarchySource : IGalaxyHierarchySource
    {
        public int Calls;

        public Task<IReadOnlyList<GalaxyObject>> GetHierarchyAsync(CancellationToken cancellationToken)
        {
            Interlocked.Increment(ref Calls);
            var platform = new GalaxyObject { GobjectId = 1, TagName = "Plat01", ContainedName = "Plat01" };
            platform.TemplateChain.Add("$WinPlatform");
            var tank = new GalaxyObject { GobjectId = 2, TagName = "Tank01", ContainedName = "Tank01", ParentGobjectId = 1 };
            tank.TemplateChain.Add("$UserDefined");
            tank.Attributes.Add(new GalaxyAttribute { AttributeName = "Level", FullTagReference = "Tank01.Level", MxDataType = 5 });
            return Task.FromResult<IReadOnlyList<GalaxyObject>>([platform, tank]);
        }
    }

    private sealed class StreamingSubscriber : IGalaxySubscriber
    {
        private readonly Channel<MxEvent> _events = Channel.CreateUnbounded<MxEvent>();
        private int _nextHandle;

        public List<string> Subscribed { get; } = [];
        public Dictionary<string, int> HandleByRef { get; } = new(StringComparer.OrdinalIgnoreCase);

        public Task<IReadOnlyList<SubscribeResult>> SubscribeBulkAsync(
            IReadOnlyList<string> fullReferences, int bufferedUpdateIntervalMs, CancellationToken cancellationToken)
        {
            var results = new List<SubscribeResult>();
            lock (Subscribed)
            {
                foreach (var r in fullReferences)
                {
                    var h = Interlocked.Increment(ref _nextHandle);
                    Subscribed.Add(r);
                    HandleByRef[r] = h;
                    results.Add(new SubscribeResult { TagAddress = r, ItemHandle = h, WasSuccessful = true });
                }
            }
            return Task.FromResult<IReadOnlyList<SubscribeResult>>(results);
        }

        public Task UnsubscribeBulkAsync(IReadOnlyList<int> itemHandles, CancellationToken cancellationToken) => Task.CompletedTask;

        public async IAsyncEnumerable<MxEvent> StreamEventsAsync([EnumeratorCancellation] CancellationToken cancellationToken)
        {
            await foreach (var ev in _events.Reader.ReadAllAsync(cancellationToken).ConfigureAwait(false))
            {
                yield return ev;
            }
        }

        public ValueTask EmitBoolAsync(int handle, bool value) => _events.Writer.WriteAsync(new MxEvent
        {
            Family = MxEventFamily.OnDataChange,
            ItemHandle = handle,
            Value = new MxValue { BoolValue = value },
            Quality = 192,
            SourceTimestamp = Timestamp.FromDateTime(DateTime.UtcNow),
        });

        public ValueTask EmitDoubleAsync(int handle, double value) => _events.Writer.WriteAsync(new MxEvent
        {
            Family = MxEventFamily.OnDataChange,
            ItemHandle = handle,
            Value = new MxValue { DoubleValue = value },
            Quality = 192,
            SourceTimestamp = Timestamp.FromDateTime(DateTime.UtcNow),
        });
    }

    private sealed class NullBuilder : IAddressSpaceBuilder
    {
        public IAddressSpaceBuilder Folder(string browseName, string displayName) => this;
        public IVariableHandle Variable(string browseName, string displayName, DriverAttributeInfo attributeInfo) => new Handle(attributeInfo.FullName);
        public void AddProperty(string browseName, DriverDataType dataType, object? value) { }
        private sealed class Handle(string fullReference) : IVariableHandle
        {
            public string FullReference { get; } = fullReference;
            public IAlarmConditionSink MarkAsAlarmCondition(AlarmConditionInfo alarmInfo) => new NoopSink();
        }
        private sealed class NoopSink : IAlarmConditionSink
        {
            public void OnTransition(AlarmEventArgs args) { }
        }
    }

    private static async Task WaitUntilAsync(Func<bool> condition, string because)
    {
        var deadline = DateTime.UtcNow.AddMilliseconds(WaitMs);
        while (!condition())
        {
            if (DateTime.UtcNow > deadline) throw new ShouldAssertException(because);
            await Task.Delay(10);
        }
    }

    private static HostState? StateOf(GalaxyDriver driver, string host) =>
        driver.GetHostStatuses().FirstOrDefault(s => s.HostName == host)?.State;

    /// <summary>
    ///     A ScanState event on the probe handle updates GetHostStatuses; after a reconnect replay
    ///     the probe is re-subscribed and events on the NEW handle keep updating it, while the dead
    ///     pre-reconnect handle no longer does. Probe traffic never reaches public OnDataChange.
    /// </summary>
    [Fact]
    public async Task ScanState_event_updates_GetHostStatuses_and_survives_a_reconnect()
    {
        var sub = new StreamingSubscriber();
        var source = new CountingHierarchySource();
        await using var driver = new GalaxyDriver("drv", Opts(), hierarchySource: source, subscriber: sub);
        var publicRefs = new List<string>();
        driver.OnDataChange += (_, e) => { lock (publicRefs) publicRefs.Add(e.FullReference); };

        await driver.InitializeAsync("{}", CancellationToken.None);
        await driver.DiscoverAsync(new NullBuilder(), CancellationToken.None);
        // The pump starts on the first value subscription (the probe watcher does not start it).
        await driver.SubscribeAsync(["Tank01.Level"], TimeSpan.Zero, CancellationToken.None);

        var probeHandle = sub.HandleByRef["Plat01.ScanState"];
        await sub.EmitBoolAsync(probeHandle, true);
        await WaitUntilAsync(() => StateOf(driver, "Plat01") == HostState.Running,
            "a ScanState=true event must surface Plat01 as Running");

        // Reconnect replay: every pre-reconnect handle is dead; the probe must rebind.
        await driver.InvokeReplayForTestAsync(CancellationToken.None);
        var rebound = sub.HandleByRef["Plat01.ScanState"];
        rebound.ShouldNotBe(probeHandle, "replay must re-subscribe the probe and record the fresh handle");

        await sub.EmitBoolAsync(rebound, false);
        await WaitUntilAsync(() => StateOf(driver, "Plat01") == HostState.Stopped,
            "a ScanState=false event on the post-reconnect handle must surface Plat01 as Stopped");

        // The dead handle is no longer routed (positive control: the live one still is).
        await sub.EmitBoolAsync(probeHandle, true);
        await sub.EmitDoubleAsync(sub.HandleByRef["Tank01.Level"], 1.0);
        await WaitUntilAsync(() => { lock (publicRefs) return publicRefs.Contains("Tank01.Level"); }, "value traffic must still flow");
        StateOf(driver, "Plat01").ShouldBe(HostState.Stopped, "an event on the pre-reconnect handle must be dropped");

        lock (publicRefs) publicRefs.ShouldNotContain("Plat01.ScanState", "probe traffic is the watcher's, never the host's");
    }

    /// <summary>Discovery asks the gateway for the hierarchy once; the probe set is selected from that same sweep.</summary>
    [Fact]
    public async Task DiscoverAsync_fetches_the_hierarchy_once_and_still_syncs_the_probe_set()
    {
        var sub = new StreamingSubscriber();
        var source = new CountingHierarchySource();
        await using var driver = new GalaxyDriver("drv", Opts(), hierarchySource: source, subscriber: sub);
        await driver.InitializeAsync("{}", CancellationToken.None);

        await driver.DiscoverAsync(new NullBuilder(), CancellationToken.None);

        source.Calls.ShouldBe(1, "the probe selection must reuse the discoverer's sweep");
        sub.Subscribed.ShouldContain("Plat01.ScanState", "positive control: the platform probe was still synced");
        driver.ProbeWatcherForTests.ShouldNotBeNull().WatchedPlatforms.ShouldBe(["Plat01"]);
    }

    /// <summary>Inbound stream data advances LastSuccessfulRead (throttled); the throttle holds the stamp still.</summary>
    [Fact]
    public async Task Stream_data_advances_LastSuccessfulRead_under_a_throttle()
    {
        var sub = new StreamingSubscriber();
        await using var driver = new GalaxyDriver("drv", Opts(), hierarchySource: null, subscriber: sub)
        {
            LivenessStampThrottle = TimeSpan.FromHours(1),
        };
        await driver.InitializeAsync("{}", CancellationToken.None);
        var initStamp = driver.GetHealth().LastSuccessfulRead.ShouldNotBeNull();
        await Task.Delay(20, TestContext.Current.CancellationToken);

        var delivered = 0;
        driver.OnDataChange += (_, _) => Interlocked.Increment(ref delivered);
        await driver.SubscribeAsync(["Tank01.Level"], TimeSpan.Zero, CancellationToken.None);
        var handle = sub.HandleByRef["Tank01.Level"];

        await sub.EmitDoubleAsync(handle, 1.0);
        await WaitUntilAsync(() => Volatile.Read(ref delivered) == 1, "first event must be delivered");
        var first = driver.GetHealth().LastSuccessfulRead.ShouldNotBeNull();
        first.ShouldBeGreaterThan(initStamp, "data on the stream is liveness evidence");

        await Task.Delay(20, TestContext.Current.CancellationToken);
        await sub.EmitDoubleAsync(handle, 2.0);
        await WaitUntilAsync(() => Volatile.Read(ref delivered) == 2, "second event must be delivered");
        driver.GetHealth().LastSuccessfulRead.ShouldBe(first, "inside the throttle window the stamp holds still");
    }
}