lmxopcua/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorSourceRemovedTests.cs

using Akka.Actor;
using Akka.TestKit;
using Microsoft.EntityFrameworkCore;
using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.Commons.Engines;
using ZB.MOM.WW.OtOpcUa.Commons.Messages.Admin;
using ZB.MOM.WW.OtOpcUa.Commons.OpcUa;
using ZB.MOM.WW.OtOpcUa.Commons.Types;
using ZB.MOM.WW.OtOpcUa.Configuration;
using ZB.MOM.WW.OtOpcUa.Runtime.Drivers;
using ZB.MOM.WW.OtOpcUa.Runtime.Tests.Harness;
using ZB.MOM.WW.OtOpcUa.Runtime.VirtualTags;

namespace ZB.MOM.WW.OtOpcUa.Runtime.Tests.Drivers;

/// <summary>
///     The dependency mux retains a ref's last value for replay-opting subscribers and evicts a driver's values on
///     <see cref="DependencyMuxActor.SourceRemoved"/>. Before this, nothing in production sent that message, so a
///     driver removed by a deploy kept its last Good in the mux and replayed it to the next replay-opting
///     subscriber as if it were current. The host now sends it wherever a driver OBJECT ends (deploy removal,
///     config-change respawn, operator Restart), never on a supervision restart, and drops value bursts from the
///     child incarnations it has already stopped so a late burst cannot re-retain what was evicted. All through a
///     REAL mux and real driver children.
/// </summary>
public sealed class DriverHostActorSourceRemovedTests : RuntimeActorTestBase
{
    private static readonly NodeId TestNode = NodeId.Parse("driver-sourceremoved-test");
    private static readonly RevisionHash RevA = RevisionHash.Parse(new string('a', 64));
    private static readonly RevisionHash RevB = RevisionHash.Parse(new string('b', 64));
    private const string Path1 = "Plant/Modbus/dev1/speed";
    private const string Path2 = "Plant/Modbus2/dev2/flow";
    private static readonly (string, string, string, string, string?)[] NoRefs = [];

    private static readonly (string Driver, string DriverName, string Device, string Tag, string DataType, bool Writable)[] BothDrivers =
    [
        ("DI-1", "Modbus", "dev1", "speed", "Double", false),
        ("DI-2", "Modbus2", "dev2", "flow", "Double", false),
    ];

    private IActorRef _host = ActorRefs.Nobody;
    private IActorRef _mux = ActorRefs.Nobody;
    private TestProbe _coordinator = null!;
    private TestProbe _holder = null!;
    private RecordingSubscribableDriverFactory _factory = null!;
    private IDbContextFactory<OtOpcUaConfigDbContext> _db = null!;

    /// <summary>Spawns a host over a real mux with two enabled drivers, waits for both children to subscribe,
    /// registers a replay-opting holder (so the mux retains both refs) and publishes one value per driver.</summary>
    private void ArrangeTwoDriversWithRetainedValues()
    {
        _db = NewInMemoryDbFactory();
        var revA = V3ArtifactSeeder.SeedV3Deployment(_db, RevA, BothDrivers, NoRefs, enabled: true);
        _factory = new RecordingSubscribableDriverFactory("Modbus");
        _mux = Sys.ActorOf(DependencyMuxActor.Props());
        _coordinator = CreateTestProbe();
        _holder = CreateTestProbe();
        _host = Sys.ActorOf(DriverHostActor.Props(
            _db, TestNode, _coordinator.Ref,
            driverFactory: _factory,
            localRoles: new HashSet<string> { "driver" },
            dependencyMux: _mux,
            virtualTagEvaluator: NullVirtualTagEvaluator.Instance,
            virtualTagHostOverride: CreateTestProbe().Ref));
        V3ArtifactSeeder.Dispatch(_host, _coordinator, revA, RevA);
        AwaitAssert(() => _factory.Created.Count.ShouldBe(2), PresenceBudget);
        // The spawn hands a child the PREVIOUS apply's set (empty on the first deploy) and the bulk push then
        // re-subscribes with the real one; a value fired before that lands is lost, so wait for the real set.
        AwaitSubscribed("DI-1", Path1);
        AwaitSubscribed("DI-2", Path2);

        _mux.Tell(new DependencyMuxActor.RegisterInterest(new[] { Path1, Path2 }, _holder.Ref, ReplayRetained: true));
        Driver("DI-1").FireDataChange(Path1, 11.0);
        Driver("DI-2").FireDataChange(Path2, 22.0);
        // One fish for both: the two drivers' bursts arrive in either order, and fishing for one at a time would
        // discard the other.
        var pending = new HashSet<string>(StringComparer.Ordinal) { Path1, Path2 };
        _holder.FishForMessage<VirtualTagActor.DependencyValueBatch>(b =>
        {
            foreach (var i in b.Items)
                if ((i.TagId == Path1 && Equals(i.Value, 11.0)) || (i.TagId == Path2 && Equals(i.Value, 22.0)))
                    pending.Remove(i.TagId);
            return pending.Count == 0;
        }, PresenceBudget);
    }

    /// <summary>Waits until the holder has been delivered <paramref name="value"/> for <paramref name="path"/>.</summary>
    private void AwaitHolder(string path, object value) =>
        _holder.FishForMessage<VirtualTagActor.DependencyValueBatch>(
            b => b.Items.Any(i => i.TagId == path && Equals(i.Value, value)), PresenceBudget);

    private void AwaitSubscribed(string driverId, string path) =>
        AwaitAssert(() => (Driver(driverId).LastSubscribedRefs ?? []).ShouldContain(path), PresenceBudget);

    private RecordingSubscribableDriver Driver(string id) => _factory.Created.Last(d => d.DriverInstanceId == id);

    /// <summary>A fresh replay-opting subscriber for both refs; returns what it was replayed, keyed by ref.</summary>
    private Dictionary<string, VirtualTagActor.DependencyValueChanged> Replayed()
    {
        var probe = CreateTestProbe();
        _mux.Tell(new DependencyMuxActor.RegisterInterest(new[] { Path1, Path2 }, probe.Ref, ReplayRetained: true));
        return probe.ExpectMsg<VirtualTagActor.DependencyValueBatch>(PresenceBudget)
            .Items.ToDictionary(i => i.TagId, StringComparer.Ordinal);
    }

    /// <summary>The live child actor for <paramref name="driverId"/> (named <c>drv-{id}-g{generation}</c>).</summary>
    private IActorRef ChildOf(string driverId)
    {
        var probe = CreateTestProbe();
        Sys.ActorSelection(_host.Path.ToStringWithoutAddress() + "/drv-" + driverId + "-g*").Tell(new Identify(driverId), probe.Ref);
        return probe.ExpectMsg<ActorIdentity>(PresenceBudget).Subject.ShouldNotBeNull();
    }

    private static void ShouldNotBeTheOldGood(Dictionary<string, VirtualTagActor.DependencyValueChanged> replay, string path, object oldValue)
    {
        if (!replay.TryGetValue(path, out var item)) return;
        (Equals(item.Value, oldValue) && OpcUaStatus.IsGood(item.StatusCode))
            .ShouldBeFalse($"the stopped incarnation's last Good for {path} was replayed as current");
    }

    [Fact]
    public void A_driver_removed_by_a_deploy_is_no_longer_replayed_while_a_still_present_driver_is()
    {
        ArrangeTwoDriversWithRetainedValues();

        // Positive control before the removal: both values replay.
        Replayed().Keys.ShouldBe(new[] { Path1, Path2 }, ignoreOrder: true);

        // Deploy B drops DI-1 (StopChild).
        var revB = V3ArtifactSeeder.SeedV3Deployment(_db, RevB, [("DI-2", "Modbus2", "dev2", "flow", "Double", false)], NoRefs, enabled: true);
        V3ArtifactSeeder.Dispatch(_host, _coordinator, revB, RevB);

        AwaitAssert(
            () => Replayed().Keys.ToArray().ShouldBe(new[] { Path2 }, "DI-1 was removed: its last Good must not replay; DI-2's must"),
            PresenceBudget, cancellationToken: TestContext.Current.CancellationToken);
    }

    /// <summary>Review I-1: a burst the removed child sent after the eviction (it stops asynchronously, so it keeps
    /// publishing through the rest of the apply) must be dropped by the host, not forwarded to re-retain the value.</summary>
    [Fact]
    public void A_late_burst_from_a_removed_child_does_not_re_retain_its_value()
    {
        ArrangeTwoDriversWithRetainedValues();
        var oldChild = ChildOf("DI-1");
        // Hold the removed child in its PostStop so it has NOT terminated when the stale burst lands; a terminated
        // actor cannot send, and the host forgets a retired child on Terminated.
        var gate = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
        Driver("DI-1").ShutdownGate = gate;

        var revB = V3ArtifactSeeder.SeedV3Deployment(_db, RevB, [("DI-2", "Modbus2", "dev2", "flow", "Double", false)], NoRefs, enabled: true);
        V3ArtifactSeeder.Dispatch(_host, _coordinator, revB, RevB);

        // The stale burst, sent AS the stopped child.
        _host.Tell(ValuePublishes.Single("DI-1", Path1, 11.0, OpcUaStatus.Good, DateTime.UtcNow), oldChild);
        // Positive control / ordering fence: a live driver's value sent after it is delivered, so the stale burst
        // has been processed by the host by the time we look.
        Driver("DI-2").FireDataChange(Path2, 23.0);
        AwaitHolder(Path2, 23.0);

        Replayed().Keys.ToArray().ShouldBe(new[] { Path2 }, "the removed child's late burst re-retained its last Good");
        gate.TrySetResult();
    }

    /// <summary>Review m-3: a config edit respawns the driver (stop + spawn); the OLD object's value is evicted, and
    /// the NEW driver's value replays (positive control).</summary>
    [Fact]
    public void A_config_change_respawn_evicts_the_old_value_and_the_new_drivers_value_replays()
    {
        ArrangeTwoDriversWithRetainedValues();

        // Adding a tag to DI-1 changes its merged DriverConfig, which respawns it.
        var revB = V3ArtifactSeeder.SeedV3Deployment(_db, RevB,
            [.. BothDrivers, ("DI-1", "Modbus", "dev1", "temp", "Double", false)], NoRefs, enabled: true);
        V3ArtifactSeeder.Dispatch(_host, _coordinator, revB, RevB);
        AwaitAssert(() => _factory.Created.Count.ShouldBe(3), PresenceBudget, cancellationToken: TestContext.Current.CancellationToken);
        var respawned = Driver("DI-1");
        // The spawn hands the child the PREVIOUS apply's set, then the bulk push re-subscribes with the new one
        // (which names the added tag); a value fired between the two can be lost, so wait for the final set.
        AwaitAssert(() => (respawned.LastSubscribedRefs ?? []).ShouldContain("Plant/Modbus/dev1/temp"), PresenceBudget, cancellationToken: TestContext.Current.CancellationToken);

        ShouldNotBeTheOldGood(Replayed(), Path1, 11.0);

        respawned.FireDataChange(Path1, 33.0);
        AwaitHolder(Path1, 33.0);
        var replay = Replayed();
        replay[Path1].Value.ShouldBe(33.0, "the respawned driver's own value replays");
        replay[Path2].Value.ShouldBe(22.0, "the untouched driver is unaffected");
    }

    /// <summary>Review m-3: an operator Restart builds a new driver object; the old one's value is evicted and the
    /// new one's replays.</summary>
    [Fact]
    public void An_operator_restart_evicts_the_old_value_and_the_new_drivers_value_replays()
    {
        ArrangeTwoDriversWithRetainedValues();

        _host.Tell(new RestartDriver("c1", "DI-1", "tester", Guid.NewGuid()));
        AwaitAssert(() => _factory.Created.Count.ShouldBe(3), PresenceBudget, cancellationToken: TestContext.Current.CancellationToken);
        var restarted = Driver("DI-1");
        AwaitSubscribed("DI-1", Path1);

        ShouldNotBeTheOldGood(Replayed(), Path1, 11.0);

        restarted.FireDataChange(Path1, 44.0);
        AwaitHolder(Path1, 44.0);
        Replayed()[Path1].Value.ShouldBe(44.0);
    }

    [Fact]
    public void A_supervision_restart_does_not_evict_the_drivers_retained_value()
    {
        ArrangeTwoDriversWithRetainedValues();
        var di1 = Driver("DI-1");

        // Crash DI-1's child: Akka restarts it around the SAME driver object, and the host reprovisions it.
        di1.FireMalformedDataChange(Path1);
        AwaitAssert(() => di1.InitializeCount.ShouldBe(2), PresenceBudget, cancellationToken: TestContext.Current.CancellationToken);
        _factory.Created.Count.ShouldBe(2, "a supervision restart re-uses the driver; no new one is created");

        var replay = Replayed();
        replay.Keys.ShouldBe(new[] { Path1, Path2 }, ignoreOrder: true,
            "a restart is the same driver instance: its last value stays replayable");
    }
}