ScadaBridge/tests/ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Tests/Actors/DataConnectionSameAdapterReconnectTests.cs

using System.Collections.Concurrent;
using Akka.Actor;
using Akka.TestKit.Xunit2;
using NSubstitute;
using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Protocol;
using ZB.MOM.WW.ScadaBridge.Commons.Messages.DataConnection;
using ZB.MOM.WW.ScadaBridge.Commons.Types.Enums;
using ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Actors;
using ZB.MOM.WW.ScadaBridge.HealthMonitoring;

namespace ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Tests.Actors;

/// <summary>
/// The SAME-adapter twin of <see cref="DataConnectionStaleCompletionTests"/> (review 2026-10-21
/// 02/C1). A plain drop + reconnect keeps the adapter object — only the session underneath it
/// changes — so before the fix the adapter generation did not advance and an initial-subscribe
/// completion issued against the DEAD session passed every stale guard and was applied against the
/// new one:
/// <list type="bullet">
///   <item>(a) landing before the reconnect re-subscribe row for its tag, it parked the dead handle
///   under the tag's key, so the LIVE row then looked redundant and its handle was released — the
///   tag reads resolved and never delivers again;</item>
///   <item>(b) landing after it, its dead handle was released against the same adapter — and an
///   MxGateway handle is a session-scoped int the new session may already have handed to a
///   different tag;</item>
///   <item>a connection-level failure row in it knocked the freshly reconnected link back into
///   Reconnecting.</item>
/// </list>
/// </summary>
[Collection("DataConnectionManagerActor")]
public class DataConnectionSameAdapterReconnectTests : TestKit
{
    private const string Tag = "sensor/temp";
    private const string OtherTag = "sensor/pressure";

    private readonly ISiteHealthCollector _health = Substitute.For<ISiteHealthCollector>();
    private readonly IDataConnectionFactory _factory = Substitute.For<IDataConnectionFactory>();

    private readonly DataConnectionOptions _options = new()
    {
        ReconnectInterval = TimeSpan.FromMilliseconds(100),
        TagResolutionRetryInterval = TimeSpan.FromSeconds(30),
        WriteTimeout = TimeSpan.FromSeconds(5),
        SeedReadMaxAttempts = 1,
        SeedReadRetryDelay = TimeSpan.FromMilliseconds(10),
        SubscribeBatchDelay = TimeSpan.Zero
    };

    private IActorRef CreateConnectionActor(string name, IDataConnection adapter)
    {
        var config = new Dictionary<string, string> { ["Endpoint"] = "opc.tcp://plc:4840" };
        return Sys.ActorOf(Props.Create(() =>
            new DataConnectionActor(
                name, adapter, _options, _health, _factory, "OpcUa", config, null, 3)), name);
    }

    private int HealthTransitions(string connectionName, ConnectionHealth health) =>
        _health.ReceivedCalls().Count(c =>
            c.GetMethodInfo().Name == nameof(ISiteHealthCollector.UpdateConnectionHealth) &&
            c.GetArguments() is [string name, ConnectionHealth reported, ..] &&
            name == connectionName && reported == health);

    private void AwaitConnectedCount(string name, int count) =>
        AwaitCondition(() => HealthTransitions(name, ConnectionHealth.Connected) >= count,
            TimeSpan.FromSeconds(5), TimeSpan.FromMilliseconds(10));

    private void AwaitResolved(IActorRef actor, int resolved) =>
        AwaitCondition(() => actor.Ask<DataConnectionHealthReport>(
                new DataConnectionActor.GetHealthReport(), TimeSpan.FromSeconds(1)).Result.ResolvedTags == resolved,
            TimeSpan.FromSeconds(5), TimeSpan.FromMilliseconds(20));

    /// <summary>Drops the session and waits for the same adapter to reconnect.</summary>
    private void DropAndReconnect(string name, GatedAdapter adapter, int connectedCountAfter)
    {
        adapter.RaiseDisconnected();
        AwaitConnectedCount(name, connectedCountAfter);
    }

    /// <summary>
    /// (a) The stale completion lands BEFORE the re-subscribe row for its tag. The live handle
    /// must never be released, and the registry must still hold it afterwards.
    /// </summary>
    [Fact]
    public void StaleCompletion_BeforeResubscribeRow_DoesNotReleaseTheLiveHandle()
    {
        const string name = "same-adapter-a";
        var adapter = new GatedAdapter();
        var actor = CreateConnectionActor(name, adapter);
        AwaitConnectedCount(name, 1);

        var probe = CreateTestProbe();
        probe.Send(actor, new SubscribeTagsRequest("c1", "inst1", name, [Tag], DateTimeOffset.UtcNow));
        var initial = adapter.AwaitCall(0);

        DropAndReconnect(name, adapter, 2);
        var resubscribe = adapter.AwaitCall(1);
        var disconnectsAfterReconnect = HealthTransitions(name, ConnectionHealth.Disconnected);

        // Dead-session completion first …
        initial.Complete(Tag, "dead-1");
        probe.FishForMessage<SubscribeTagsResponse>(m => m.CorrelationId == "c1", TimeSpan.FromSeconds(5));
        // … then the live re-subscribe row.
        resubscribe.Complete(Tag, "live-2");

        // Let the live completion (and any release it would trigger) be applied.
        Thread.Sleep(500);

        Assert.DoesNotContain("live-2", adapter.ReleasedIds);
        Assert.Equal(disconnectsAfterReconnect, HealthTransitions(name, ConnectionHealth.Disconnected));

        // The registry holds the LIVE handle: tearing the instance down releases live-2.
        probe.Send(actor, new UnsubscribeTagsRequest("c2", "inst1", name, DateTimeOffset.UtcNow));
        AwaitCondition(() => adapter.ReleasedIds.Contains("live-2"), TimeSpan.FromSeconds(5));
    }

    /// <summary>
    /// (b) The stale completion lands AFTER the re-subscribe row. Its handle belongs to the dead
    /// session and must not be released against the same adapter: the new session may have reused
    /// that handle for a different, live tag (MxGateway handles are session-scoped ints).
    /// </summary>
    [Fact]
    public void StaleCompletion_AfterResubscribeRow_NeverReleasesAHandleTheNewSessionReused()
    {
        const string name = "same-adapter-b";
        var adapter = new GatedAdapter();
        var actor = CreateConnectionActor(name, adapter);
        AwaitConnectedCount(name, 1);

        var probe = CreateTestProbe();
        // inst2 owns OtherTag, fully established in the first session.
        probe.Send(actor, new SubscribeTagsRequest("o1", "inst2", name, [OtherTag], DateTimeOffset.UtcNow));
        adapter.AwaitCall(0).Complete(OtherTag, "h0");
        probe.FishForMessage<SubscribeTagsResponse>(m => m.CorrelationId == "o1" && m.Success, TimeSpan.FromSeconds(5));

        // inst1's subscribe is still in flight when the session drops.
        probe.Send(actor, new SubscribeTagsRequest("c1", "inst1", name, [Tag], DateTimeOffset.UtcNow));
        var initial = adapter.AwaitCall(1);

        DropAndReconnect(name, adapter, 2);
        var resubscribe = adapter.AwaitCall(2);
        var disconnectsAfterReconnect = HealthTransitions(name, ConnectionHealth.Disconnected);

        // The new session hands OtherTag the handle number "h1" …
        resubscribe.CompleteMany((OtherTag, "h1"), (Tag, "h2"));
        AwaitResolved(actor, 2);

        // … which is the very number the dead session gave inst1's in-flight subscribe.
        initial.Complete(Tag, "h1");
        probe.FishForMessage<SubscribeTagsResponse>(m => m.CorrelationId == "c1", TimeSpan.FromSeconds(5));
        actor.Tell(new DataConnectionActor.GetHealthReport(), probe.Ref);
        probe.FishForMessage<DataConnectionHealthReport>(_ => true, TimeSpan.FromSeconds(5));

        Assert.DoesNotContain("h1", adapter.ReleasedIds);
        Assert.DoesNotContain("h2", adapter.ReleasedIds);
        Assert.Equal(disconnectsAfterReconnect, HealthTransitions(name, ConnectionHealth.Disconnected));
    }

    /// <summary>
    /// A connection-level failure row in a dead-session completion must not knock the freshly
    /// reconnected link back into Reconnecting.
    /// </summary>
    [Fact]
    public void StaleConnectionLevelFailure_DoesNotDriveTheReconnectedLinkBackToReconnecting()
    {
        const string name = "same-adapter-connfail";
        var adapter = new GatedAdapter();
        var actor = CreateConnectionActor(name, adapter);
        AwaitConnectedCount(name, 1);

        var probe = CreateTestProbe();
        probe.Send(actor, new SubscribeTagsRequest("c1", "inst1", name, [Tag], DateTimeOffset.UtcNow));
        var initial = adapter.AwaitCall(0);

        DropAndReconnect(name, adapter, 2);
        adapter.AwaitCall(1).Complete(Tag, "live-2");
        var disconnectsAfterReconnect = HealthTransitions(name, ConnectionHealth.Disconnected);

        // InvalidOperationException is classified as connection-level (transport down).
        initial.Fail(new InvalidOperationException("adapter not connected"));
        probe.FishForMessage<SubscribeTagsResponse>(m => m.CorrelationId == "c1", TimeSpan.FromSeconds(5));
        actor.Tell(new DataConnectionActor.GetHealthReport(), probe.Ref);
        probe.FishForMessage<DataConnectionHealthReport>(_ => true, TimeSpan.FromSeconds(5));
        ExpectNoMsg(TimeSpan.FromMilliseconds(300));

        Assert.Equal(disconnectsAfterReconnect, HealthTransitions(name, ConnectionHealth.Disconnected));
        Assert.Equal(2, HealthTransitions(name, ConnectionHealth.Connected));
        Assert.DoesNotContain("live-2", adapter.ReleasedIds);
    }

    /// <summary>One gated <c>SubscribeBatchAsync</c> call.</summary>
    internal sealed class PendingBatch
    {
        private readonly TaskCompletionSource<IReadOnlyList<TagSubscribeResult>> _tcs =
            new(TaskCreationOptions.RunContinuationsAsynchronously);

        public PendingBatch(IReadOnlyList<string> tags) => Tags = tags;

        public IReadOnlyList<string> Tags { get; }

        public Task<IReadOnlyList<TagSubscribeResult>> Task => _tcs.Task;

        public void Complete(string tag, string id) => CompleteMany((tag, id));

        public void CompleteMany(params (string Tag, string Id)[] rows) =>
            _tcs.SetResult(rows.Select(r => new TagSubscribeResult(r.Tag, true, r.Id, null)).ToList());

        public void Fail(Exception ex) => _tcs.SetException(ex);
    }

    /// <summary>
    /// Batch-capable adapter whose every subscribe batch is held until the test completes it, so
    /// the arrival order of dead-session and live-session completions is fully controlled. The
    /// same instance is reused across reconnects, exactly like the shipped adapters.
    /// </summary>
    internal sealed class GatedAdapter : IDataConnection, IBatchSubscribableConnection
    {
        private readonly ConcurrentDictionary<int, PendingBatch> _calls = new();
        private int _callCount;

        public readonly ConcurrentQueue<string> ReleasedIds = new();

        public ConnectionHealth Status { get; private set; } = ConnectionHealth.Disconnected;

        public event Action? Disconnected;

        public void RaiseDisconnected()
        {
            Status = ConnectionHealth.Disconnected;
            Disconnected?.Invoke();
        }

        public PendingBatch AwaitCall(int index)
        {
            var deadline = DateTime.UtcNow + TimeSpan.FromSeconds(5);
            while (DateTime.UtcNow < deadline)
            {
                if (_calls.TryGetValue(index, out var call))
                    return call;
                Thread.Sleep(10);
            }

            throw new TimeoutException($"Subscribe batch #{index} was never issued.");
        }

        public Task ConnectAsync(IDictionary<string, string> connectionDetails, CancellationToken cancellationToken = default)
        {
            Status = ConnectionHealth.Connected;
            return System.Threading.Tasks.Task.CompletedTask;
        }

        public Task DisconnectAsync(CancellationToken cancellationToken = default) =>
            System.Threading.Tasks.Task.CompletedTask;

        public Task<IReadOnlyList<TagSubscribeResult>> SubscribeBatchAsync(
            IReadOnlyList<string> tagPaths, SubscriptionCallback callback, CancellationToken cancellationToken = default)
        {
            var call = new PendingBatch(tagPaths.ToList());
            _calls[Interlocked.Increment(ref _callCount) - 1] = call;
            return call.Task;
        }

        public Task UnsubscribeBatchAsync(IReadOnlyList<string> subscriptionIds, CancellationToken cancellationToken = default)
        {
            foreach (var id in subscriptionIds)
                ReleasedIds.Enqueue(id);
            return System.Threading.Tasks.Task.CompletedTask;
        }

        public Task<string> SubscribeAsync(string tagPath, SubscriptionCallback callback, CancellationToken cancellationToken = default) =>
            throw new NotSupportedException("batch-capable adapter");

        public Task UnsubscribeAsync(string subscriptionId, CancellationToken cancellationToken = default)
        {
            ReleasedIds.Enqueue(subscriptionId);
            return System.Threading.Tasks.Task.CompletedTask;
        }

        public Task<ReadResult> ReadAsync(string tagPath, CancellationToken cancellationToken = default) =>
            System.Threading.Tasks.Task.FromResult(new ReadResult(false, null, "no value"));

        public Task<IReadOnlyDictionary<string, ReadResult>> ReadBatchAsync(
            IEnumerable<string> tagPaths, CancellationToken cancellationToken = default) =>
            System.Threading.Tasks.Task.FromResult<IReadOnlyDictionary<string, ReadResult>>(
                new Dictionary<string, ReadResult>());

        public Task<WriteResult> WriteAsync(string tagPath, object? value, CancellationToken cancellationToken = default) =>
            System.Threading.Tasks.Task.FromResult(new WriteResult(true, null));

        public Task<IReadOnlyDictionary<string, WriteResult>> WriteBatchAsync(
            IDictionary<string, object?> values, CancellationToken cancellationToken = default) =>
            System.Threading.Tasks.Task.FromResult<IReadOnlyDictionary<string, WriteResult>>(
                values.ToDictionary(kv => kv.Key, _ => new WriteResult(true, null)));

        public ValueTask DisposeAsync() => ValueTask.CompletedTask;
    }
}