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

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>
/// Deferred register row <b>97</b> — <b>in-place reconfiguration of a LIVE data connection</b>.
///
/// <para>
/// <b>The gap these pin.</b> <c>DataConnectionManagerActor.HandleCreateConnection</c> is
/// create-only: it warns and returns for a name it already holds. So the Deployment Manager's
/// documented "a changed configuration re-issues the create so the DCL adopts it" was never true
/// on ANY node — a corrected endpoint or credential took effect only when the connection was next
/// built from scratch (node restart, singleton failover, or a remove-then-add), while the DM's
/// hash cache recorded it as applied. Discovered while validating rows 89/90 and recorded as
/// residual 1 of <c>docs/plans/2026-08-17-dcl-connection-arbitration-decision.md</c>.
/// </para>
///
/// <para>
/// <b>Why the fix is a swap and not a restart.</b> Stopping and re-creating the connection actor
/// is destructive: the subscription registry lives inside <c>DataConnectionActor</c>, and
/// <c>InstanceActor</c> cancels its per-connection subscribe-retry timer on the first success —
/// so nothing would ever re-subscribe. That trades "stale endpoint, data flowing" for "correct
/// endpoint, no data until redeploy". <see cref="ReconfigureConnectionCommand"/> instead reuses
/// the primary→backup failover machinery: dispose the adapter, build one from the new config,
/// bump the adapter generation, and let the reconnect cycle re-issue every subscription from the
/// registry that was never touched.
/// </para>
/// </summary>
[Collection("DataConnectionManagerActor")]
public class DataConnectionReconfigureTests : TestKit
{
    private const string OldEndpoint = "opc.tcp://old-plc:4840";
    private const string NewEndpoint = "opc.tcp://new-plc:4840";

    private readonly DataConnectionOptions _options;
    private readonly ISiteHealthCollector _healthCollector;
    private readonly IDataConnectionFactory _factory;

    public DataConnectionReconfigureTests()
        : base(@"akka.loglevel = DEBUG")
    {
        _healthCollector = Substitute.For<ISiteHealthCollector>();
        _factory = Substitute.For<IDataConnectionFactory>();
        _options = new DataConnectionOptions
        {
            ReconnectInterval = TimeSpan.FromMilliseconds(100),
            TagResolutionRetryInterval = TimeSpan.FromMilliseconds(200),
            WriteTimeout = TimeSpan.FromSeconds(5),
            SeedReadMaxAttempts = 1,
            SeedReadRetryDelay = TimeSpan.FromMilliseconds(10)
        };
    }

    private static Dictionary<string, string> Config(string endpoint) =>
        new() { ["Endpoint"] = endpoint };

    private IActorRef CreateConnectionActor(
        string name, IDataConnection adapter, IDictionary<string, string> primaryConfig) =>
        Sys.ActorOf(Props.Create(() =>
            new DataConnectionActor(
                name, adapter, _options, _healthCollector, _factory, "OpcUa",
                primaryConfig, null, 3)), name);

    /// <summary>An adapter that connects, subscribes and reads without incident.</summary>
    private static IDataConnection HealthyAdapter(string subscriptionId)
    {
        var adapter = Substitute.For<IDataConnection>();
        adapter.ConnectAsync(Arg.Any<IDictionary<string, string>>(), Arg.Any<CancellationToken>())
            .Returns(Task.CompletedTask);
        adapter.Status.Returns(ConnectionHealth.Connected);
        adapter.SubscribeAsync(Arg.Any<string>(), Arg.Any<SubscriptionCallback>(), Arg.Any<CancellationToken>())
            .Returns(subscriptionId);
        adapter.ReadAsync(Arg.Any<string>(), Arg.Any<CancellationToken>())
            .Returns(new ReadResult(false, null, null));
        return adapter;
    }

    private static void RaiseDisconnected(IDataConnection adapter) =>
        adapter.Disconnected += Raise.Event<Action>();

    private bool HasEnteredConnectedState(string connectionName) =>
        _healthCollector.ReceivedCalls().Any(c =>
            c.GetMethodInfo().Name == nameof(ISiteHealthCollector.UpdateConnectionHealth) &&
            c.GetArguments() is [string name, ConnectionHealth.Connected, ..] &&
            name == connectionName);

    private bool CreatedAdapterFor(string endpoint) =>
        _factory.ReceivedCalls().Any(c =>
            c.GetMethodInfo().Name == nameof(IDataConnectionFactory.Create) &&
            c.GetArguments()[1] is IDictionary<string, string> d &&
            d.TryGetValue("Endpoint", out var e) && e == endpoint);

    private static bool Subscribed(IDataConnection adapter) =>
        adapter.ReceivedCalls().Any(c => c.GetMethodInfo().Name == "SubscribeAsync");

    private static bool Connected(IDataConnection adapter) =>
        adapter.ReceivedCalls().Any(c => c.GetMethodInfo().Name == "ConnectAsync");

    // ── The connection actor ──

    /// <summary>
    /// The headline behaviour: a live, CONNECTED connection told to reconfigure builds a new
    /// adapter from the new configuration and re-issues the subscription it already holds against
    /// it — without the subscriber doing anything, and without the connection actor being stopped.
    ///
    /// <para>
    /// <b>Fails before the fix:</b> there was no reconfigure command at all, and the create the
    /// Deployment Manager re-issued was dropped by the manager's create-only guard, so the old
    /// adapter kept serving <c>old-plc</c> indefinitely.
    /// </para>
    /// </summary>
    [Fact]
    public void Reconfigure_WhileConnected_SwapsTheAdapter_AndReSubscribesExistingTags()
    {
        const string name = "reconf-connected";
        var oldAdapter = HealthyAdapter("sub-old");
        var newAdapter = HealthyAdapter("sub-new");
        _factory.Create("OpcUa", Arg.Is<IDictionary<string, string>>(d => d["Endpoint"] == NewEndpoint))
            .Returns(newAdapter);

        var actor = CreateConnectionActor(name, oldAdapter, Config(OldEndpoint));
        AwaitCondition(() => HasEnteredConnectedState(name), TimeSpan.FromSeconds(5),
            TimeSpan.FromMilliseconds(10));

        actor.Tell(new SubscribeTagsRequest(
            "c1", "inst1", name, ["sensor/temp"], DateTimeOffset.UtcNow));
        ExpectMsg<SubscribeTagsResponse>(TimeSpan.FromSeconds(5));
        AwaitCondition(() => Subscribed(oldAdapter), TimeSpan.FromSeconds(5));

        actor.Tell(new ReconfigureConnectionCommand(
            name, "OpcUa", Config(NewEndpoint), null, 3));

        AwaitCondition(() => CreatedAdapterFor(NewEndpoint), TimeSpan.FromSeconds(5),
            TimeSpan.FromMilliseconds(10));
        AwaitCondition(() => Connected(newAdapter), TimeSpan.FromSeconds(5));

        // The whole point: the subscription survives the swap and is re-issued for free.
        AwaitCondition(() => Subscribed(newAdapter), TimeSpan.FromSeconds(5),
            TimeSpan.FromMilliseconds(10));
    }

    /// <summary>
    /// The adapter generation must be bumped by the swap, exactly as it is on failover: the
    /// outgoing adapter's subscription callbacks captured Self and keep firing, and a value from
    /// the endpoint the operator has just corrected AWAY from must never be delivered as if it
    /// came from the new one.
    /// </summary>
    [Fact]
    public void Reconfigure_DropsValuesStillArrivingFromTheOldAdapter()
    {
        const string name = "reconf-generation";
        SubscriptionCallback? oldCallback = null;

        var oldAdapter = HealthyAdapter("sub-old");
        oldAdapter.SubscribeAsync("sensor/temp", Arg.Any<SubscriptionCallback>(), Arg.Any<CancellationToken>())
            .Returns(ci =>
            {
                oldCallback = (SubscriptionCallback)ci[1];
                return Task.FromResult("sub-old");
            });

        var newAdapter = HealthyAdapter("sub-new");
        _factory.Create("OpcUa", Arg.Is<IDictionary<string, string>>(d => d["Endpoint"] == NewEndpoint))
            .Returns(newAdapter);

        var actor = CreateConnectionActor(name, oldAdapter, Config(OldEndpoint));
        AwaitCondition(() => HasEnteredConnectedState(name), TimeSpan.FromSeconds(5),
            TimeSpan.FromMilliseconds(10));

        actor.Tell(new SubscribeTagsRequest(
            "c1", TestActor.Path.Name, name, ["sensor/temp"], DateTimeOffset.UtcNow));
        ExpectMsg<SubscribeTagsResponse>(TimeSpan.FromSeconds(5));
        AwaitCondition(() => oldCallback != null, TimeSpan.FromSeconds(5));

        actor.Tell(new ReconfigureConnectionCommand(
            name, "OpcUa", Config(NewEndpoint), null, 3));

        // The swap flips every subscribed tag to bad quality, same as a disconnect.
        ExpectMsg<ConnectionQualityChanged>(TimeSpan.FromSeconds(5));

        AwaitCondition(() => Subscribed(newAdapter), TimeSpan.FromSeconds(5));
        ExpectNoMsg(TimeSpan.FromMilliseconds(300));

        oldCallback!("sensor/temp", new TagValue(999, QualityCode.Good, DateTimeOffset.UtcNow));

        ExpectNoMsg(TimeSpan.FromSeconds(1));
    }

    /// <summary>
    /// <b>The state that matters most, and the one a stash would break.</b> A connection whose
    /// configured endpoint is wrong or dead never reaches Connected at all — it loops in the
    /// Connecting state retrying — which is EXACTLY when a corrected configuration is sent. If
    /// the reconfigure were stashed with the subscribe/write requests it would wait for the old
    /// (broken) configuration to succeed first, i.e. forever.
    /// </summary>
    [Fact]
    public void Reconfigure_WhileStillConnecting_IsHonoured_NotStashed()
    {
        const string name = "reconf-connecting";
        var deadAdapter = Substitute.For<IDataConnection>();
        deadAdapter.ConnectAsync(Arg.Any<IDictionary<string, string>>(), Arg.Any<CancellationToken>())
            .Returns(Task.FromException(new Exception("Connection refused")));

        var newAdapter = HealthyAdapter("sub-new");
        _factory.Create("OpcUa", Arg.Is<IDictionary<string, string>>(d => d["Endpoint"] == NewEndpoint))
            .Returns(newAdapter);

        var actor = CreateConnectionActor(name, deadAdapter, Config(OldEndpoint));
        AwaitCondition(() => Connected(deadAdapter), TimeSpan.FromSeconds(5));
        Assert.False(HasEnteredConnectedState(name));

        actor.Tell(new ReconfigureConnectionCommand(
            name, "OpcUa", Config(NewEndpoint), null, 3));

        AwaitCondition(() => CreatedAdapterFor(NewEndpoint), TimeSpan.FromSeconds(5),
            TimeSpan.FromMilliseconds(10));
        AwaitCondition(() => HasEnteredConnectedState(name), TimeSpan.FromSeconds(5),
            TimeSpan.FromMilliseconds(10));
    }

    /// <summary>
    /// The same, one state over: a connection that HAD been up and is now looping in the
    /// Reconnecting state. Reconnecting stashes subscribes and writes too, so the reconfigure
    /// must be handled there as well — and the subscriptions it re-issues are the ones the
    /// registry has been holding across the outage.
    /// </summary>
    [Fact]
    public void Reconfigure_WhileReconnecting_IsHonoured_AndReSubscribes()
    {
        const string name = "reconf-reconnecting";
        var connects = 0;
        var oldAdapter = HealthyAdapter("sub-old");
        oldAdapter.ConnectAsync(Arg.Any<IDictionary<string, string>>(), Arg.Any<CancellationToken>())
            .Returns(_ => Interlocked.Increment(ref connects) == 1
                ? Task.CompletedTask
                : Task.FromException(new Exception("still down")));

        var newAdapter = HealthyAdapter("sub-new");
        _factory.Create("OpcUa", Arg.Is<IDictionary<string, string>>(d => d["Endpoint"] == NewEndpoint))
            .Returns(newAdapter);

        var actor = CreateConnectionActor(name, oldAdapter, Config(OldEndpoint));
        AwaitCondition(() => HasEnteredConnectedState(name), TimeSpan.FromSeconds(5),
            TimeSpan.FromMilliseconds(10));

        actor.Tell(new SubscribeTagsRequest(
            "c1", "inst1", name, ["sensor/temp"], DateTimeOffset.UtcNow));
        ExpectMsg<SubscribeTagsResponse>(TimeSpan.FromSeconds(5));

        // Drop the link; with no backup configured it retries the same dead endpoint forever.
        RaiseDisconnected(oldAdapter);
        AwaitCondition(() => connects >= 2, TimeSpan.FromSeconds(5));

        actor.Tell(new ReconfigureConnectionCommand(
            name, "OpcUa", Config(NewEndpoint), null, 3));

        AwaitCondition(() => CreatedAdapterFor(NewEndpoint), TimeSpan.FromSeconds(5),
            TimeSpan.FromMilliseconds(10));
        AwaitCondition(() => Subscribed(newAdapter), TimeSpan.FromSeconds(5),
            TimeSpan.FromMilliseconds(10));
    }

    /// <summary>
    /// A reconfigure carrying the configuration already in force must cost nothing. The sender is
    /// hash-gated and does not send one, but an adapter swap drops the link and re-subscribes
    /// every tag, so a redundant command must never be able to do that.
    /// </summary>
    [Fact]
    public void Reconfigure_WithTheConfigurationAlreadyInForce_IsANoOp()
    {
        const string name = "reconf-identical";
        var adapter = HealthyAdapter("sub-old");

        var actor = CreateConnectionActor(name, adapter, Config(OldEndpoint));
        AwaitCondition(() => HasEnteredConnectedState(name), TimeSpan.FromSeconds(5),
            TimeSpan.FromMilliseconds(10));

        actor.Tell(new ReconfigureConnectionCommand(
            name, "OpcUa", Config(OldEndpoint), null, 3));

        // No adapter is ever built: the factory is only consulted by a swap.
        ExpectNoMsg(TimeSpan.FromMilliseconds(500));
        Assert.False(CreatedAdapterFor(OldEndpoint));
        Assert.DoesNotContain(_factory.ReceivedCalls(),
            c => c.GetMethodInfo().Name == nameof(IDataConnectionFactory.Create));
    }

    // ── The manager's routing ──

    /// <summary>
    /// The manager forwards a reconfigure to the connection actor that owns the name — which is
    /// the whole difference from <see cref="CreateConnectionCommand"/>, whose handler warns and
    /// returns for a name it already holds.
    /// </summary>
    [Fact]
    public void Manager_Reconfigure_ForALiveConnection_ReachesTheConnectionActor()
    {
        var oldAdapter = HealthyAdapter("sub-old");
        var newAdapter = HealthyAdapter("sub-new");
        _factory.Create("OpcUa", Arg.Is<IDictionary<string, string>>(d => d["Endpoint"] == OldEndpoint))
            .Returns(oldAdapter);
        _factory.Create("OpcUa", Arg.Is<IDictionary<string, string>>(d => d["Endpoint"] == NewEndpoint))
            .Returns(newAdapter);

        var manager = Sys.ActorOf(Props.Create(() =>
            new DataConnectionManagerActor(_factory, _options, _healthCollector)));

        manager.Tell(new CreateConnectionCommand("conn1", "OpcUa", Config(OldEndpoint), null, 3));
        AwaitCondition(() => Connected(oldAdapter), TimeSpan.FromSeconds(10));

        manager.Tell(new ReconfigureConnectionCommand("conn1", "OpcUa", Config(NewEndpoint), null, 3));

        AwaitCondition(() => Connected(newAdapter), TimeSpan.FromSeconds(10),
            TimeSpan.FromMilliseconds(10));
    }

    /// <summary>
    /// A reconfigure for a name the manager does not hold falls back to a create, so the command
    /// is safe whatever the manager's state — a sender whose view of which connections exist is
    /// stale still converges on the intended configuration in one message rather than leaving
    /// the connection missing until something else happens to create it.
    /// </summary>
    [Fact]
    public void Manager_Reconfigure_ForAnUnknownConnection_FallsBackToCreate()
    {
        var adapter = HealthyAdapter("sub-new");
        _factory.Create("OpcUa", Arg.Is<IDictionary<string, string>>(d => d["Endpoint"] == NewEndpoint))
            .Returns(adapter);

        var manager = Sys.ActorOf(Props.Create(() =>
            new DataConnectionManagerActor(_factory, _options, _healthCollector)));

        manager.Tell(new ReconfigureConnectionCommand("never-created", "OpcUa", Config(NewEndpoint), null, 3));

        AwaitCondition(() => Connected(adapter), TimeSpan.FromSeconds(10),
            TimeSpan.FromMilliseconds(10));

        // And it is a real, routable connection afterwards — not an orphan.
        manager.Tell(new SubscribeTagsRequest(
            "c1", "inst1", "never-created", ["sensor/temp"], DateTimeOffset.UtcNow));
        var response = ExpectMsg<SubscribeTagsResponse>(TimeSpan.FromSeconds(10));
        Assert.True(response.Success);
    }
}