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

using Akka.Actor;
using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.Commons.OpcUa;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
using ZB.MOM.WW.OtOpcUa.Runtime.Drivers;
using ZB.MOM.WW.OtOpcUa.Runtime.Tests.Harness;

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

/// <summary>
///     Covers the ingress seam of the value-pipeline batching work: <see cref="DriverInstanceActor"/>
///     forwards data changes to its parent as <see cref="DriverInstanceActor.AttributeValueBatchPublished"/>
///     — ONE message per driver production burst.
///     <para>A driver implementing <see cref="IBatchSubscribable"/> gets the batch handler attached and the
///     per-tag <see cref="ISubscribable.OnDataChange"/> handler NOT attached: such a driver raises BOTH
///     events per the contract, so attaching both would double-deliver every value. A per-tag-only
///     <see cref="ISubscribable"/> driver keeps working unchanged — each raise is wrapped as a
///     single-item batch, so the parent sees exactly one message shape either way.</para>
///     <para>Status codes ride verbatim (#519); timestamps come from the driver's snapshot
///     (<c>SourceTimestampUtc ?? ServerTimestampUtc</c>), never from the actor's clock.</para>
/// </summary>
public sealed class DriverInstanceActorBatchForwardTests : RuntimeActorTestBase
{
    /// <summary>
    ///     A batch-capable driver's burst of 3 changed tags reaches the parent as exactly ONE
    ///     <see cref="DriverInstanceActor.AttributeValueBatchPublished"/> carrying all 3 items, each mapped
    ///     verbatim (reference / value / status code) with the timestamp derived from the snapshot.
    /// </summary>
    [Fact]
    public async Task Batch_driver_forwards_one_batch_message()
    {
        var driver = new BatchSubscribableStubDriver();
        var parent = CreateTestProbe();
        // #477: DriverInstanceActor Tells the parent ConnectivityChanged on each connect/disconnect;
        // this suite asserts on the value forwards only, so ignore it.
        parent.IgnoreMessages(m => m is DriverInstanceActor.ConnectivityChanged);
        var actor = parent.ChildActorOf(DriverInstanceActor.Props(driver));

        actor.Tell(new DriverInstanceActor.InitializeRequested("{}"));
        AwaitCondition(() => driver.InitializeCount > 0, PresenceBudget);

        await actor.Ask<DriverInstanceActor.SubscriptionEstablished>(
            new DriverInstanceActor.Subscribe(new[] { "tag-a", "tag-b", "tag-c" }, TimeSpan.FromMilliseconds(100)),
            PresenceBudget);

        var sourceTs = new DateTime(2026, 8, 15, 10, 0, 0, DateTimeKind.Utc);
        var serverTs = new DateTime(2026, 8, 15, 10, 0, 5, DateTimeKind.Utc);
        driver.FireBatch(includePerTagRaises: false,
            new DataChangeItem("tag-a", new DataValueSnapshot(3.14, OpcUaStatus.Good, sourceTs, serverTs)),
            new DataChangeItem("tag-b", new DataValueSnapshot(7, 0x80740000u /* BadTypeMismatch */, sourceTs.AddSeconds(1), serverTs)),
            // Null source timestamp — the server timestamp is the fallback, per OnDataChangeForward.
            new DataChangeItem("tag-c", new DataValueSnapshot(null, OpcUaStatus.BadNoCommunication, null, serverTs)));

        var batch = parent.ExpectMsg<DriverInstanceActor.AttributeValueBatchPublished>(PresenceBudget);
        batch.DriverInstanceId.ShouldBe(driver.DriverInstanceId);
        batch.Items.Count.ShouldBe(3);

        batch.Items[0].FullReference.ShouldBe("tag-a");
        batch.Items[0].Value.ShouldBe(3.14);
        batch.Items[0].StatusCode.ShouldBe(OpcUaStatus.Good);
        batch.Items[0].TimestampUtc.ShouldBe(sourceTs);

        batch.Items[1].FullReference.ShouldBe("tag-b");
        batch.Items[1].Value.ShouldBe(7);
        // #519: the sub-code rides through, not a collapsed OpcUaStatus.Bad.
        batch.Items[1].StatusCode.ShouldBe(0x80740000u);
        batch.Items[1].TimestampUtc.ShouldBe(sourceTs.AddSeconds(1));

        batch.Items[2].FullReference.ShouldBe("tag-c");
        batch.Items[2].Value.ShouldBeNull();
        batch.Items[2].StatusCode.ShouldBe(OpcUaStatus.BadNoCommunication);
        batch.Items[2].TimestampUtc.ShouldBe(serverTs, "a null SourceTimestampUtc falls back to the server timestamp");

        // Exactly ONE message for the burst — no per-item trailer.
        parent.ExpectNoMsg(TimeSpan.FromMilliseconds(300));
    }

    /// <summary>
    ///     An <see cref="IBatchSubscribable"/> driver raises BOTH the batch event and the per-tag
    ///     <see cref="ISubscribable.OnDataChange"/> for every item (that is the documented driver contract,
    ///     kept for legacy consumers). The actor must attach ONE handler only, so the parent sees the batch
    ///     and nothing else — attaching both would publish every value twice.
    /// </summary>
    [Fact]
    public async Task Batch_driver_per_tag_event_not_double_forwarded()
    {
        var driver = new BatchSubscribableStubDriver();
        var parent = CreateTestProbe();
        parent.IgnoreMessages(m => m is DriverInstanceActor.ConnectivityChanged);
        var actor = parent.ChildActorOf(DriverInstanceActor.Props(driver));

        actor.Tell(new DriverInstanceActor.InitializeRequested("{}"));
        AwaitCondition(() => driver.InitializeCount > 0, PresenceBudget);

        await actor.Ask<DriverInstanceActor.SubscriptionEstablished>(
            new DriverInstanceActor.Subscribe(new[] { "tag-a", "tag-b" }, TimeSpan.FromMilliseconds(100)),
            PresenceBudget);

        // Deterministic half of the assertion: the per-tag handler was never attached at all. ExpectNoMsg
        // below only proves nothing arrived within its window; this proves it structurally cannot.
        driver.OnDataChangeSubscriberCount.ShouldBe(0, "a batch-capable driver must NOT get the per-tag handler");

        var ts = new DateTime(2026, 8, 15, 11, 0, 0, DateTimeKind.Utc);
        driver.FireBatch(includePerTagRaises: true,
            new DataChangeItem("tag-a", new DataValueSnapshot(1, OpcUaStatus.Good, ts, ts)),
            new DataChangeItem("tag-b", new DataValueSnapshot(2, OpcUaStatus.Good, ts, ts)));

        var batch = parent.ExpectMsg<DriverInstanceActor.AttributeValueBatchPublished>(PresenceBudget);
        batch.Items.Count.ShouldBe(2);

        // The per-tag raises for the SAME burst produced nothing extra — no second batch, no
        // AttributeValuePublished. Anything here means both handlers were attached.
        parent.ExpectNoMsg(TimeSpan.FromMilliseconds(400));
    }

    /// <summary>
    ///     A driver that implements only <see cref="ISubscribable"/> keeps its per-tag handler and each raise
    ///     arrives at the parent as a single-item batch — the whole fleet minus the batch-capable drivers
    ///     stays on one message shape.
    /// </summary>
    [Fact]
    public async Task Per_tag_driver_forwards_single_item_batches()
    {
        var driver = new SubscribableStubDriver();
        var parent = CreateTestProbe();
        parent.IgnoreMessages(m => m is DriverInstanceActor.ConnectivityChanged);
        var actor = parent.ChildActorOf(DriverInstanceActor.Props(driver));

        actor.Tell(new DriverInstanceActor.InitializeRequested("{}"));
        AwaitCondition(() => driver.InitializeCount > 0, PresenceBudget);

        await actor.Ask<DriverInstanceActor.SubscriptionEstablished>(
            new DriverInstanceActor.Subscribe(new[] { "tag-a" }, TimeSpan.FromMilliseconds(100)),
            PresenceBudget);

        driver.FireDataChange("tag-a", value: 3.14, statusCode: OpcUaStatus.Good);

        var batch = parent.ExpectMsg<DriverInstanceActor.AttributeValueBatchPublished>(PresenceBudget);
        batch.DriverInstanceId.ShouldBe(driver.DriverInstanceId);
        batch.Items.Count.ShouldBe(1);
        batch.Items[0].FullReference.ShouldBe("tag-a");
        batch.Items[0].Value.ShouldBe(3.14);
        batch.Items[0].StatusCode.ShouldBe(OpcUaStatus.Good);
    }

    /// <summary>
    ///     The detach is load-bearing — the <see cref="IDriver"/> object outlives the actor across respawns,
    ///     so a batch handler left attached would push into a dead/stale actor. After Unsubscribe, a burst
    ///     produces nothing at the parent.
    /// </summary>
    [Fact]
    public async Task Detach_removes_batch_handler()
    {
        var driver = new BatchSubscribableStubDriver();
        var parent = CreateTestProbe();
        parent.IgnoreMessages(m => m is DriverInstanceActor.ConnectivityChanged);
        var actor = parent.ChildActorOf(DriverInstanceActor.Props(driver));

        actor.Tell(new DriverInstanceActor.InitializeRequested("{}"));
        AwaitCondition(() => driver.InitializeCount > 0, PresenceBudget);

        await actor.Ask<DriverInstanceActor.SubscriptionEstablished>(
            new DriverInstanceActor.Subscribe(new[] { "tag-a" }, TimeSpan.FromMilliseconds(100)),
            PresenceBudget);
        driver.OnDataChangeBatchSubscriberCount.ShouldBe(1, "the batch handler is attached for an IBatchSubscribable driver");

        actor.Tell(new DriverInstanceActor.Unsubscribe());

        AwaitCondition(() => driver.OnDataChangeBatchSubscriberCount == 0, PresenceBudget);

        var ts = new DateTime(2026, 8, 15, 12, 0, 0, DateTimeKind.Utc);
        driver.FireBatch(includePerTagRaises: true,
            new DataChangeItem("tag-a", new DataValueSnapshot(42, OpcUaStatus.Good, ts, ts)));

        parent.ExpectNoMsg(TimeSpan.FromMilliseconds(400));
    }

    /// <summary>
    ///     Comms loss (<see cref="DriverInstanceActor.DisconnectObserved"/>, Connected → Reconnecting) detaches
    ///     the batch handler too. The <see cref="IDriver"/> object outlives the actor across a host respawn, so a
    ///     handler surviving the disconnect would push a burst produced by the dead transport into a stale actor.
    /// </summary>
    [Fact]
    public async Task DisconnectObserved_detaches_batch_handler_so_late_bursts_are_dropped()
    {
        var driver = new BatchSubscribableStubDriver();
        var parent = CreateTestProbe();
        parent.IgnoreMessages(m => m is DriverInstanceActor.ConnectivityChanged);
        var actor = parent.ChildActorOf(DriverInstanceActor.Props(driver, reconnectInterval: TimeSpan.FromSeconds(30)));

        actor.Tell(new DriverInstanceActor.InitializeRequested("{}"));
        AwaitCondition(() => driver.InitializeCount > 0, PresenceBudget);

        await actor.Ask<DriverInstanceActor.SubscriptionEstablished>(
            new DriverInstanceActor.Subscribe(new[] { "tag-a" }, TimeSpan.FromMilliseconds(100)),
            PresenceBudget);
        driver.OnDataChangeBatchSubscriberCount.ShouldBe(1);

        actor.Tell(new DriverInstanceActor.DisconnectObserved("backend went away"));

        // Race window — once the disconnect is processed the handler is gone and a late burst is inert.
        AwaitCondition(() => driver.OnDataChangeBatchSubscriberCount == 0, PresenceBudget);

        var ts = new DateTime(2026, 8, 15, 13, 0, 0, DateTimeKind.Utc);
        driver.FireBatch(includePerTagRaises: true,
            new DataChangeItem("tag-a", new DataValueSnapshot(42, OpcUaStatus.Good, ts, ts)));

        parent.ExpectNoMsg(TimeSpan.FromMilliseconds(300));
    }

    /// <summary>
    ///     Stopping the actor detaches the batch handler in <c>PostStop</c> — the same leak guard as the
    ///     per-tag path, and the one that matters when the host respawns a child around the same
    ///     <see cref="IDriver"/> instance.
    /// </summary>
    [Fact]
    public async Task PostStop_detaches_batch_handler()
    {
        var driver = new BatchSubscribableStubDriver();
        var parent = CreateTestProbe();
        parent.IgnoreMessages(m => m is DriverInstanceActor.ConnectivityChanged);
        var actor = parent.ChildActorOf(DriverInstanceActor.Props(driver));

        actor.Tell(new DriverInstanceActor.InitializeRequested("{}"));
        AwaitCondition(() => driver.InitializeCount > 0, PresenceBudget);

        await actor.Ask<DriverInstanceActor.SubscriptionEstablished>(
            new DriverInstanceActor.Subscribe(new[] { "tag-a" }, TimeSpan.FromMilliseconds(100)),
            PresenceBudget);
        driver.OnDataChangeBatchSubscriberCount.ShouldBe(1);

        Sys.Stop(actor);

        AwaitCondition(() => driver.OnDataChangeBatchSubscriberCount == 0, PresenceBudget);
    }

    /// <summary>
    ///     A malformed raise from a buggy driver — a null payload, or one carrying no items — is dropped at
    ///     the handler and never reaches the mailbox. Unguarded, the null would throw inside the event
    ///     handler, restart the actor, and take the driver transport down with it; the still-attached handler
    ///     (count unchanged) and the burst that lands afterwards prove neither happened.
    /// </summary>
    [Fact]
    public async Task Malformed_batch_raise_is_dropped_without_restarting_the_actor()
    {
        var driver = new BatchSubscribableStubDriver();
        var parent = CreateTestProbe();
        parent.IgnoreMessages(m => m is DriverInstanceActor.ConnectivityChanged);
        var actor = parent.ChildActorOf(DriverInstanceActor.Props(driver));

        actor.Tell(new DriverInstanceActor.InitializeRequested("{}"));
        AwaitCondition(() => driver.InitializeCount > 0, PresenceBudget);

        await actor.Ask<DriverInstanceActor.SubscriptionEstablished>(
            new DriverInstanceActor.Subscribe(new[] { "tag-a" }, TimeSpan.FromMilliseconds(100)),
            PresenceBudget);

        driver.FireBatch(includePerTagRaises: false /* no items — an empty burst */);
        driver.FireNullBatch();

        parent.ExpectNoMsg(TimeSpan.FromMilliseconds(300));
        driver.OnDataChangeBatchSubscriberCount.ShouldBe(1,
            "a restart would have detached the handler in PostStop");

        // …and the subscription is still live: a well-formed burst still lands.
        var ts = new DateTime(2026, 8, 15, 14, 0, 0, DateTimeKind.Utc);
        driver.FireBatch(includePerTagRaises: false,
            new DataChangeItem("tag-a", new DataValueSnapshot(5, OpcUaStatus.Good, ts, ts)));

        var batch = parent.ExpectMsg<DriverInstanceActor.AttributeValueBatchPublished>(PresenceBudget);
        batch.Items.Count.ShouldBe(1);
        batch.Items[0].FullReference.ShouldBe("tag-a");
        batch.Items[0].Value.ShouldBe(5);
    }

    /// <summary>
    ///     A <see cref="StubDriver"/> that implements <see cref="IBatchSubscribable"/> — it raises
    ///     <see cref="OnDataChangeBatch"/> once per burst AND (optionally, per the documented driver
    ///     contract) the per-tag <see cref="OnDataChange"/> for each item of that same burst.
    /// </summary>
    private sealed class BatchSubscribableStubDriver : StubDriver, IBatchSubscribable
    {
        private readonly StubHandle _handle = new();

        /// <summary>Occurs when a single attribute changes (legacy per-tag contract).</summary>
        public event EventHandler<DataChangeEventArgs>? OnDataChange;

        /// <summary>Occurs once per production burst with every attribute that changed in it.</summary>
        public event EventHandler<DataChangeBatchEventArgs>? OnDataChangeBatch;

        /// <summary>Gets the number of handlers attached to <see cref="OnDataChange"/>.</summary>
        public int OnDataChangeSubscriberCount => OnDataChange?.GetInvocationList().Length ?? 0;

        /// <summary>Gets the number of handlers attached to <see cref="OnDataChangeBatch"/>.</summary>
        public int OnDataChangeBatchSubscriberCount => OnDataChangeBatch?.GetInvocationList().Length ?? 0;

        /// <summary>Subscribes to the specified full references.</summary>
        /// <param name="fullReferences">The full references to subscribe to.</param>
        /// <param name="publishingInterval">The publishing interval.</param>
        /// <param name="cancellationToken">Cancellation token for the operation.</param>
        public Task<ISubscriptionHandle> SubscribeAsync(
            IReadOnlyList<string> fullReferences, TimeSpan publishingInterval, CancellationToken cancellationToken) =>
            Task.FromResult<ISubscriptionHandle>(_handle);

        /// <summary>Unsubscribes from the specified subscription handle.</summary>
        /// <param name="handle">The subscription handle.</param>
        /// <param name="cancellationToken">Cancellation token for the operation.</param>
        public Task UnsubscribeAsync(ISubscriptionHandle handle, CancellationToken cancellationToken) =>
            Task.CompletedTask;

        /// <summary>Raises one production burst.</summary>
        /// <param name="includePerTagRaises">
        ///     When true, also raises the per-tag <see cref="OnDataChange"/> for every item — the documented
        ///     contract a real batch-capable driver honours for legacy consumers.
        /// </param>
        /// <param name="items">The changed attributes of this burst.</param>
        public void FireBatch(bool includePerTagRaises, params DataChangeItem[] items)
        {
            OnDataChangeBatch?.Invoke(this, new DataChangeBatchEventArgs(_handle, items));
            if (!includePerTagRaises) return;
            foreach (var item in items)
            {
                OnDataChange?.Invoke(this, new DataChangeEventArgs(_handle, item.FullReference, item.Snapshot));
            }
        }

        /// <summary>Raises the batch event with a null payload — what a buggy driver does, and what the
        /// handler's guard exists to absorb.</summary>
        public void FireNullBatch() => OnDataChangeBatch?.Invoke(this, null!);
    }
}