lmxopcua/tests/Server/ZB.MOM.WW.OtOpcUa.AdminUI.Tests/Telemetry/TelemetryDialDeadlineTests.cs

using System.Collections.Concurrent;
using Akka.Actor;
using Akka.TestKit.Xunit2;
using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.AdminUI.Hubs;
using ZB.MOM.WW.OtOpcUa.AdminUI.Telemetry;
using ZB.MOM.WW.OtOpcUa.Cluster;
using ZB.MOM.WW.OtOpcUa.Commons.Messages.Alerts;
using ZB.MOM.WW.OtOpcUa.Commons.Messages.Drivers;
using ZB.MOM.WW.OtOpcUa.Commons.Messages.Logging;

namespace ZB.MOM.WW.OtOpcUa.AdminUI.Tests.Telemetry;

/// <summary>
///     Review 02 U2: <c>TelemetryDial:CallTimeoutSeconds</c> was bound, defaulted and documented, and consumed
///     nowhere. It now bounds a dial's <b>establishment</b>: a dial that has delivered no event by the deadline is
///     cut and re-dialled with a failure streak. A live node answers a fresh subscribe at once (the node-local hub
///     replays its health/resilience snapshots), so the deadline bites only a dial that never connects — a
///     black-holed host, where the underlying socket connect has no timeout of its own. An established stream
///     is deliberately NOT bounded: HTTP/2 keepalive pings own its liveness, and the wire has no heartbeat, so an
///     idle timeout would churn every quiet node.
/// </summary>
public sealed class TelemetryDialDeadlineTests : TestKit
{
    /// <summary>Explicit AwaitCondition poll interval: Akka's default is max / 10 of the dilated budget, i.e.
    /// seconds under CI's timefactor (lmxopcua#775).</summary>
    private static readonly TimeSpan PollInterval = TimeSpan.FromMilliseconds(20);

    private static readonly TimeSpan Timeout = TimeSpan.FromSeconds(8);

    [Fact]
    public void A_dial_that_delivers_no_event_by_the_deadline_is_cancelled_and_redialled()
    {
        var dial = new FakeDialLoop();
        Spawn(dial, callTimeoutSeconds: 1);

        var first = dial.WaitForCount(1, Timeout).Single();
        first.Ct.IsCancellationRequested.ShouldBeFalse("the deadline has not elapsed yet");

        // The stalled dial is cut at the deadline...
        AwaitCondition(() => first.Ct.IsCancellationRequested, Timeout, PollInterval);
        // ...and the supervisor re-dials (a stalled dial is a transport failure, never a silent end).
        var invocations = dial.WaitForCount(2, Timeout);
        invocations[1].CorrelationId.ShouldBe("central-node-1-2");
    }

    [Fact]
    public async Task A_dial_that_delivers_an_event_before_the_deadline_is_left_open()
    {
        var dial = new FakeDialLoop { DeliverOnOpen = true };
        var sinks = new Sinks();
        Spawn(dial, callTimeoutSeconds: 1, sinks);

        var first = dial.WaitForCount(1, Timeout).Single();
        AwaitCondition(() => sinks.HealthStore.Upserts.Count == 1, Timeout, PollInterval);

        // Positive control: well past the deadline, the established stream is untouched.
        await Task.Delay(TimeSpan.FromSeconds(2.5));
        first.Ct.IsCancellationRequested.ShouldBeFalse("an established stream must never be cut by the dial deadline");
        dial.Invocations.Count.ShouldBe(1);
    }

    private IActorRef Spawn(FakeDialLoop dial, int callTimeoutSeconds, Sinks? sinks = null)
    {
        sinks ??= new Sinks();
        var options = new TelemetryDialOptions { ContactRefreshSeconds = 3600, CallTimeoutSeconds = callTimeoutSeconds };
        return Sys.ActorOf(TelemetryDialSupervisor.Props(
            () => Task.FromResult<IReadOnlyList<TelemetryDialTarget>>(
                new[] { new TelemetryDialTarget("node-1", "http://h1:5300") }),
            dial.Loop,
            sinks.AlarmBroadcaster,
            sinks.ScriptBroadcaster,
            sinks.HealthStore,
            sinks.ResilienceStore,
            options));
    }

    private static DriverHealthChanged Health(string id) =>
        new("C1", id, "Healthy", DateTime.UtcNow, null, 0, DateTime.UtcNow);

    private sealed class DialInvocation
    {
        private readonly TaskCompletionSource _completion = new(TaskCreationOptions.RunContinuationsAsynchronously);
        public required string CorrelationId { get; init; }
        public required CancellationToken Ct { get; init; }
        public Task Completion => _completion.Task;
        public void Complete() => _completion.TrySetResult();
    }

    /// <summary>Records each invocation; optionally delivers one health event on open; ends only when cancelled.</summary>
    private sealed class FakeDialLoop
    {
        private readonly ConcurrentQueue<DialInvocation> _invocations = new();

        public bool DeliverOnOpen { get; init; }

        public IReadOnlyList<DialInvocation> Invocations => _invocations.ToArray();

        public TelemetryDialLoop Loop => async (_, correlationId, onMapped, _, ct) =>
        {
            var invocation = new DialInvocation { CorrelationId = correlationId, Ct = ct };
            _invocations.Enqueue(invocation);
            if (DeliverOnOpen) onMapped(Health("drv-1"));
            await using var reg = ct.Register(() => invocation.Complete());
            await invocation.Completion;
        };

        public IReadOnlyList<DialInvocation> WaitForCount(int count, TimeSpan timeout)
        {
            var deadline = DateTime.UtcNow + timeout;
            while (DateTime.UtcNow < deadline)
            {
                var snapshot = _invocations.ToArray();
                if (snapshot.Length >= count) return snapshot;
                Thread.Sleep(10);
            }

            throw new Xunit.Sdk.XunitException(
                $"Expected at least {count} dial invocations within {timeout}, saw {_invocations.Count}.");
        }
    }

    private sealed class Sinks
    {
        public NullBroadcaster<AlarmTransitionEvent> AlarmBroadcaster { get; } = new();
        public NullBroadcaster<ScriptLogEntry> ScriptBroadcaster { get; } = new();
        public RecordingHealthStore HealthStore { get; } = new();
        public NullResilienceStore ResilienceStore { get; } = new();
    }

    private sealed class NullBroadcaster<T> : IInProcessBroadcaster<T>
    {
        public event Action<T>? Received;
        public event Action<bool>? ConnectionStateChanged;
        public bool IsConnected { get; private set; }
        public void Publish(T item) => Received?.Invoke(item);
        public void SetConnected(bool connected) { IsConnected = connected; ConnectionStateChanged?.Invoke(connected); }
    }

    private sealed class RecordingHealthStore : IDriverStatusSnapshotStore
    {
        public ConcurrentQueue<DriverHealthChanged> Upserts { get; } = new();
        public event Action<DriverHealthChanged>? SnapshotChanged;
        public void Upsert(DriverHealthChanged snapshot) { Upserts.Enqueue(snapshot); SnapshotChanged?.Invoke(snapshot); }
        public bool TryGet(string driverInstanceId, out DriverHealthChanged snapshot) { snapshot = null!; return false; }
        public IReadOnlyCollection<DriverHealthChanged> GetAll() => Upserts.ToArray();
        public int EvictUnconfiguredStale(IReadOnlySet<string> configuredInstanceIds, DateTime olderThanUtc) => 0;
    }

    private sealed class NullResilienceStore : IDriverResilienceStatusStore
    {
        public event Action<DriverResilienceStatusChanged>? SnapshotChanged;
        public void Upsert(DriverResilienceStatusChanged snapshot) => SnapshotChanged?.Invoke(snapshot);
        public IReadOnlyList<DriverResilienceStatusChanged> GetForInstance(string driverInstanceId) => Array.Empty<DriverResilienceStatusChanged>();
        public IReadOnlyCollection<DriverResilienceStatusChanged> GetAll() => Array.Empty<DriverResilienceStatusChanged>();
        public int EvictUnconfiguredStale(IReadOnlySet<string> configuredInstanceIds, DateTime olderThanUtc) => 0;
    }
}