mxaccessgw/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Dashboard/DashboardSnapshotPublisherTests.cs

using System.Runtime.CompilerServices;
using Microsoft.AspNetCore.SignalR;
using Microsoft.Extensions.Logging.Abstractions;
using ZB.MOM.WW.MxGateway.Server.Dashboard;
using ZB.MOM.WW.MxGateway.Server.Dashboard.Hubs;

namespace ZB.MOM.WW.MxGateway.Tests.Gateway.Dashboard;

public sealed class DashboardSnapshotPublisherTests
{
    private static readonly TimeSpan TestTimeout = TimeSpan.FromSeconds(5);
    private static readonly TimeSpan IdlePollInterval = TimeSpan.FromMilliseconds(10);

    /// <summary>
    /// A transient failure inside
    /// <see cref="IDashboardSnapshotService.WatchSnapshotsAsync"/> must not
    /// end the BackgroundService; the publisher must wait the configured
    /// reconnect delay and then re-open the subscription. Before the fix,
    /// the publisher exited on the first non-cancellation exception and
    /// the dashboard's snapshot stream went silent until process restart.
    /// </summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task ExecuteAsync_WhenSnapshotServiceThrowsOnce_ReconnectsAfterDelay()
    {
        ThrowOnceThenYieldSnapshotService snapshotService = new();
        RecordingHubContext hubContext = new();
        TimeSpan reconnectDelay = TimeSpan.FromMilliseconds(50);
        DashboardSnapshotPublisher publisher = new(
            snapshotService,
            hubContext,
            ConnectedCounter(),
            NullLogger<DashboardSnapshotPublisher>.Instance,
            reconnectDelay,
            IdlePollInterval);

        using CancellationTokenSource cts = new();
        Task execute = publisher.StartAsync(cts.Token);
        await execute.WaitAsync(TestTimeout);

        // The publisher's first WatchSnapshotsAsync call throws; the second
        // call yields one snapshot. We block here until the publisher has
        // made the second subscribe attempt AND broadcast its first
        // snapshot — proving the publisher did NOT exit on the throw.
        await WaitUntilAsync(() => snapshotService.SubscribeCount >= 2);
        await WaitUntilAsync(() => hubContext.SendCount >= 1);

        DateTimeOffset firstThrowAt = snapshotService.FirstThrowAt
            ?? throw new InvalidOperationException("First subscribe did not record a throw timestamp.");
        DateTimeOffset secondSubscribeAt = snapshotService.SecondSubscribeAt
            ?? throw new InvalidOperationException("Second subscribe did not record a timestamp.");

        await cts.CancelAsync();
        await publisher.StopAsync(CancellationToken.None);

        Assert.True(snapshotService.SubscribeCount >= 2,
            $"Expected at least 2 subscribe calls, got {snapshotService.SubscribeCount}.");
        Assert.True(hubContext.SendCount >= 1);

        // The gap is measured from the moment the first subscribe actually
        // threw (inside the fake) to the moment the second subscribe began
        // (also inside the fake). This isolates the publisher's
        // Task.Delay(reconnectDelay) — no StartAsync / scheduling overhead in
        // the baseline. The 10ms slack absorbs Task.Delay's coarse Windows
        // timer quantum (~15ms) when the underlying scheduler wakes early.
        TimeSpan gap = secondSubscribeAt - firstThrowAt;
        Assert.True(gap >= reconnectDelay - TimeSpan.FromMilliseconds(10),
            $"Expected reconnect gap >= {reconnectDelay.TotalMilliseconds}ms; got {gap.TotalMilliseconds}ms.");
    }

    /// <summary>
    /// Sanity: a normal completion of WatchSnapshotsAsync (no exception)
    /// also reconnects after the delay — exits only on host shutdown.
    /// </summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task ExecuteAsync_WhenSnapshotServiceCompletes_ReconnectsAfterDelay()
    {
        CompleteImmediatelySnapshotService snapshotService = new();
        RecordingHubContext hubContext = new();
        TimeSpan reconnectDelay = TimeSpan.FromMilliseconds(50);
        DashboardSnapshotPublisher publisher = new(
            snapshotService,
            hubContext,
            ConnectedCounter(),
            NullLogger<DashboardSnapshotPublisher>.Instance,
            reconnectDelay,
            IdlePollInterval);

        using CancellationTokenSource cts = new();
        Task execute = publisher.StartAsync(cts.Token);
        await execute.WaitAsync(TestTimeout);

        await WaitUntilAsync(() => snapshotService.SubscribeCount >= 2);

        await cts.CancelAsync();
        await publisher.StopAsync(CancellationToken.None);

        Assert.True(snapshotService.SubscribeCount >= 2);
    }

    /// <summary>
    /// With no dashboard connected there is nobody to broadcast to, so the publisher must
    /// not advance the snapshot enumerator at all — every pull costs a registry snapshot and
    /// sort, a locked metrics dictionary copy, and periodically a SQLite key-table read.
    /// The first viewer to connect resumes the tick.
    /// </summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task ExecuteAsync_WhenNoHubConnections_DoesNotPullSnapshots()
    {
        CountingSnapshotService snapshotService = new();
        RecordingHubContext hubContext = new();
        DashboardSnapshotHubConnectionCounter connectionCounter = new();
        DashboardSnapshotPublisher publisher = new(
            snapshotService,
            hubContext,
            connectionCounter,
            NullLogger<DashboardSnapshotPublisher>.Instance,
            TimeSpan.FromMilliseconds(50),
            IdlePollInterval);

        using CancellationTokenSource cts = new();
        await publisher.StartAsync(cts.Token).WaitAsync(TestTimeout);

        // Long enough for many idle polls at IdlePollInterval.
        await Task.Delay(TimeSpan.FromMilliseconds(250));

        Assert.Equal(0, snapshotService.PullCount);
        Assert.Equal(0, hubContext.SendCount);

        connectionCounter.Increment();
        await WaitUntilAsync(() => hubContext.SendCount >= 1);

        await cts.CancelAsync();
        await publisher.StopAsync(CancellationToken.None);

        Assert.True(snapshotService.PullCount >= 1);
    }

    /// <summary>Creates a connection counter that already has one live viewer.</summary>
    /// <returns>A counter reporting a single connection.</returns>
    private static DashboardSnapshotHubConnectionCounter ConnectedCounter()
    {
        DashboardSnapshotHubConnectionCounter counter = new();
        counter.Increment();
        return counter;
    }

    private static async Task WaitUntilAsync(Func<bool> predicate)
    {
        using CancellationTokenSource cancellationTokenSource = new(TestTimeout);
        while (!predicate())
        {
            await Task.Delay(TimeSpan.FromMilliseconds(5), cancellationTokenSource.Token);
        }
    }

    private sealed class ThrowOnceThenYieldSnapshotService : IDashboardSnapshotService
    {
        /// <summary>Gets the number of subscription attempts.</summary>
        public int SubscribeCount { get; private set; }

        /// <summary>
        /// The wall-clock instant the first <c>WatchSnapshotsAsync</c> throws.
        /// The reconnect-gap assertion is measured against this timestamp (NOT the
        /// pre-<c>StartAsync</c> wall clock) so scheduling overhead is not baselined
        /// into the lower bound.
        /// </summary>
        public DateTimeOffset? FirstThrowAt { get; private set; }

        /// <summary>Gets the wall-clock instant of the second subscription attempt.</summary>
        public DateTimeOffset? SecondSubscribeAt { get; private set; }

        /// <inheritdoc />
        public DashboardSnapshot GetSnapshot()
        {
            return null!;
        }

        /// <inheritdoc />
        public async IAsyncEnumerable<DashboardSnapshot> WatchSnapshotsAsync(
            [EnumeratorCancellation] CancellationToken cancellationToken)
        {
            SubscribeCount++;
            int call = SubscribeCount;

            if (call == 1)
            {
                // First call: throw after a brief yield so the publisher
                // observes us as a live producer that failed.
                await Task.Yield();
                FirstThrowAt = DateTimeOffset.UtcNow;
                throw new InvalidOperationException("simulated transient snapshot failure");
            }

            SecondSubscribeAt = DateTimeOffset.UtcNow;
            yield return GetSnapshot();

            // Stay open until cancelled so the publisher's inner await
            // foreach doesn't immediately re-loop.
            try
            {
                await Task.Delay(Timeout.InfiniteTimeSpan, cancellationToken).ConfigureAwait(false);
            }
            catch (OperationCanceledException)
            {
            }
        }
    }

    private sealed class CompleteImmediatelySnapshotService : IDashboardSnapshotService
    {
        /// <summary>Gets the number of subscription attempts.</summary>
        public int SubscribeCount { get; private set; }

        /// <inheritdoc />
        public DashboardSnapshot GetSnapshot()
        {
            return null!;
        }

        /// <inheritdoc />
#pragma warning disable CS1998 // async without await — IAsyncEnumerable contract requires async signature
        public async IAsyncEnumerable<DashboardSnapshot> WatchSnapshotsAsync(
            [EnumeratorCancellation] CancellationToken cancellationToken)
#pragma warning restore CS1998
        {
            SubscribeCount++;
            // Yield nothing and complete immediately — simulates a transient
            // upstream disconnect that completes cleanly.
            yield break;
        }
    }

    private sealed class CountingSnapshotService : IDashboardSnapshotService
    {
        private int _pullCount;

        /// <summary>Gets the number of snapshots the publisher pulled from this source.</summary>
        public int PullCount => Volatile.Read(ref _pullCount);

        /// <inheritdoc />
        public DashboardSnapshot GetSnapshot()
        {
            return null!;
        }

        /// <inheritdoc />
        public async IAsyncEnumerable<DashboardSnapshot> WatchSnapshotsAsync(
            [EnumeratorCancellation] CancellationToken cancellationToken)
        {
            while (!cancellationToken.IsCancellationRequested)
            {
                // Short cadence so a listening publisher pulls quickly; the counter only
                // moves when the publisher actually advances the enumerator.
                await Task.Delay(TimeSpan.FromMilliseconds(5), cancellationToken).ConfigureAwait(false);
                Interlocked.Increment(ref _pullCount);
                yield return GetSnapshot();
            }
        }
    }

    private sealed class RecordingHubContext : IHubContext<DashboardSnapshotHub>
    {
        private readonly RecordingHubClients _clients = new();

        /// <summary>Gets the hub clients.</summary>
        public IHubClients Clients => _clients;

        /// <summary>Gets the group manager.</summary>
        public IGroupManager Groups { get; } = new NoopGroupManager();

        /// <summary>Gets the number of send calls recorded.</summary>
        public int SendCount => _clients.AllProxy.SendCount;
    }

    private sealed class RecordingHubClients : IHubClients
    {
        /// <summary>Gets the recording client proxy for all clients.</summary>
        public RecordingClientProxy AllProxy { get; } = new();

        /// <summary>Gets a client proxy targeting all clients.</summary>
        public IClientProxy All => AllProxy;

        /// <summary>Gets a client proxy excluding specified connections.</summary>
        /// <param name="excludedConnectionIds">Connection identifiers to exclude.</param>
        /// <returns>The recording client proxy shared by this fake.</returns>
        public IClientProxy AllExcept(IReadOnlyList<string> excludedConnectionIds) => AllProxy;

        /// <summary>Gets a client proxy for a specific connection.</summary>
        /// <param name="connectionId">The connection identifier.</param>
        /// <returns>The recording client proxy shared by this fake.</returns>
        public IClientProxy Client(string connectionId) => AllProxy;

        /// <summary>Gets a client proxy for specified connections.</summary>
        /// <param name="connectionIds">The connection identifiers.</param>
        /// <returns>The recording client proxy shared by this fake.</returns>
        public IClientProxy Clients(IReadOnlyList<string> connectionIds) => AllProxy;

        /// <summary>Gets a client proxy for a group.</summary>
        /// <param name="groupName">The group name.</param>
        /// <returns>The recording client proxy shared by this fake.</returns>
        public IClientProxy Group(string groupName) => AllProxy;

        /// <summary>Gets a client proxy for a group excluding specified connections.</summary>
        /// <param name="groupName">The group name.</param>
        /// <param name="excludedConnectionIds">Connection identifiers to exclude.</param>
        /// <returns>The recording client proxy shared by this fake.</returns>
        public IClientProxy GroupExcept(string groupName, IReadOnlyList<string> excludedConnectionIds) => AllProxy;

        /// <summary>Gets a client proxy for specified groups.</summary>
        /// <param name="groupNames">The group names.</param>
        /// <returns>The recording client proxy shared by this fake.</returns>
        public IClientProxy Groups(IReadOnlyList<string> groupNames) => AllProxy;

        /// <summary>Gets a client proxy for a specific user.</summary>
        /// <param name="userId">The user identifier.</param>
        /// <returns>The recording client proxy shared by this fake.</returns>
        public IClientProxy User(string userId) => AllProxy;

        /// <summary>Gets a client proxy for specified users.</summary>
        /// <param name="userIds">The user identifiers.</param>
        /// <returns>The recording client proxy shared by this fake.</returns>
        public IClientProxy Users(IReadOnlyList<string> userIds) => AllProxy;
    }

    private sealed class RecordingClientProxy : IClientProxy
    {
        private int _sendCount;

        /// <summary>Gets the number of send calls recorded.</summary>
        public int SendCount => Volatile.Read(ref _sendCount);

        /// <summary>Records a send call and completes asynchronously.</summary>
        /// <param name="method">The SignalR method name.</param>
        /// <param name="args">The method arguments.</param>
        /// <param name="cancellationToken">Token to observe for cancellation.</param>
        /// <returns>A task that represents the asynchronous operation.</returns>
        public Task SendCoreAsync(string method, object?[] args, CancellationToken cancellationToken = default)
        {
            Interlocked.Increment(ref _sendCount);
            return Task.CompletedTask;
        }
    }

    private sealed class NoopGroupManager : IGroupManager
    {
        /// <summary>Completes immediately without performing group addition.</summary>
        /// <param name="connectionId">The connection identifier.</param>
        /// <param name="groupName">The group name.</param>
        /// <param name="cancellationToken">Token to observe for cancellation.</param>
        /// <returns>A task that represents the asynchronous operation.</returns>
        public Task AddToGroupAsync(string connectionId, string groupName, CancellationToken cancellationToken = default)
            => Task.CompletedTask;

        /// <summary>Completes immediately without performing group removal.</summary>
        /// <param name="connectionId">The connection identifier.</param>
        /// <param name="groupName">The group name.</param>
        /// <param name="cancellationToken">Token to observe for cancellation.</param>
        /// <returns>A task that represents the asynchronous operation.</returns>
        public Task RemoveFromGroupAsync(string connectionId, string groupName, CancellationToken cancellationToken = default)
            => Task.CompletedTask;
    }
}