lmxopcua/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/OpcUa/OpcUaPublishActorOffThreadPlanTests.cs

using System.Collections.Concurrent;
using System.Text.Json;
using Akka.Actor;
using Microsoft.Extensions.Logging.Abstractions;
using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.Commons.OpcUa;
using ZB.MOM.WW.OtOpcUa.Commons.Types;
using ZB.MOM.WW.OtOpcUa.OpcUaServer;
using ZB.MOM.WW.OtOpcUa.Runtime.OpcUa;
using ZB.MOM.WW.OtOpcUa.Runtime.Tests.Harness;

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

/// <summary>
///     Review 02 P5 — a rebuild's artifact parse + plan no longer run on the pinned OPC UA thread. The
///     observable: value batches sent right after a large rebuild land on the sink BEFORE that rebuild's first
///     materialise call, because the actor is free while the thread pool parses and plans; two rebuilds sent
///     back to back still apply in arrival order; a plan that throws still answers the requester degraded.
/// </summary>
public sealed class OpcUaPublishActorOffThreadPlanTests : RuntimeActorTestBase
{
    [Fact]
    public void Value_batches_keep_flowing_while_a_large_plan_computes()
    {
        var sink = new OrderedSink();
        var applier = new AddressSpaceApplier(sink, NullLogger<AddressSpaceApplier>.Instance);
        var actor = Sys.ActorOf(OpcUaPublishActor.PropsForTests(sink: sink, applier: applier));
        // A value node the batch names; the sink is a recorder, so it need not exist.
        var batch = new OpcUaPublishActor.AttributeValueBatchUpdate(
            [new ValueWrite("Plant/Modbus/dev1/live", 1, OpcUaStatus.Good, DateTime.UtcNow, AddressSpaceRealm.Raw)]);

        actor.Tell(new OpcUaPublishActor.RebuildAddressSpace(CorrelationId.NewId(), DeploymentId.NewId(), RawTagArtifact(50_000), ReportOutcome: true));
        actor.Tell(batch);

        var completed = ExpectMsg<OpcUaPublishActor.RebuildCompleted>(TimeSpan.FromSeconds(120), cancellationToken: TestContext.Current.CancellationToken);
        completed.Degraded.ShouldBeFalse(completed.DegradedReason);

        var calls = sink.Calls.ToList();
        var batchIndex = calls.IndexOf("WV:Plant/Modbus/dev1/live");
        var firstMaterialise = calls.FindIndex(c => c.StartsWith("EV:", StringComparison.Ordinal) || c.StartsWith("EF:", StringComparison.Ordinal));
        batchIndex.ShouldBeGreaterThanOrEqualTo(0, "the batch must reach the sink");
        firstMaterialise.ShouldBeGreaterThanOrEqualTo(0, "the plan must still be applied (positive control)");
        batchIndex.ShouldBeLessThan(firstMaterialise, "the batch sent after the rebuild must land before the rebuild's materialisation — the parse/plan step no longer holds the actor thread");
        calls.Count(c => c.StartsWith("EV:", StringComparison.Ordinal)).ShouldBe(50_000);
    }

    [Fact]
    public void Back_to_back_rebuilds_apply_in_arrival_order()
    {
        var sink = new OrderedSink();
        var applier = new AddressSpaceApplier(sink, NullLogger<AddressSpaceApplier>.Instance);
        var actor = Sys.ActorOf(OpcUaPublishActor.PropsForTests(sink: sink, applier: applier));

        // The first (large) plan is still computing when the second (tiny) arrives; the second must wait for
        // it — its plan is a diff against what the first applies.
        actor.Tell(new OpcUaPublishActor.RebuildAddressSpace(CorrelationId.NewId(), DeploymentId.NewId(), RawTagArtifact(20_000), ReportOutcome: true));
        actor.Tell(new OpcUaPublishActor.RebuildAddressSpace(CorrelationId.NewId(), DeploymentId.NewId(), RawTagArtifact(2), ReportOutcome: true));

        var first = ExpectMsg<OpcUaPublishActor.RebuildCompleted>(TimeSpan.FromSeconds(120), cancellationToken: TestContext.Current.CancellationToken);
        var second = ExpectMsg<OpcUaPublishActor.RebuildCompleted>(TimeSpan.FromSeconds(120), cancellationToken: TestContext.Current.CancellationToken);
        first.Degraded.ShouldBeFalse(first.DegradedReason);
        second.Degraded.ShouldBeFalse(second.DegradedReason);

        // The second composition is a strict subset (2 of the same tags), so it diffs as a PureRemove of the
        // other 19,998 — which only works if it was planned AFTER the first apply, against its composition.
        sink.Calls.Count(c => c.StartsWith("RV:", StringComparison.Ordinal)).ShouldBe(19_998);
    }

    [Fact]
    public void Unreadable_bytes_answer_degraded_and_keep_the_served_space()
    {
        var sink = new OrderedSink();
        var applier = new AddressSpaceApplier(sink, NullLogger<AddressSpaceApplier>.Instance);
        var actor = Sys.ActorOf(OpcUaPublishActor.PropsForTests(sink: sink, applier: applier));

        actor.Tell(new OpcUaPublishActor.RebuildAddressSpace(CorrelationId.NewId(), DeploymentId.NewId(), "not json"u8.ToArray(), ReportOutcome: true));

        var completed = ExpectMsg<OpcUaPublishActor.RebuildCompleted>(PresenceBudget, cancellationToken: TestContext.Current.CancellationToken);
        completed.Degraded.ShouldBeTrue();
        completed.DegradedReason.ShouldNotBeNull().ShouldContain("threw");
        sink.Calls.ShouldBeEmpty("nothing may be torn down for bytes that could not be read");
        sink.Completed.ShouldBe(1, "the apply is still closed");
    }

    private static byte[] RawTagArtifact(int tagCount) =>
        JsonSerializer.SerializeToUtf8Bytes(new
        {
            UnsAreas = Array.Empty<object>(),
            UnsLines = Array.Empty<object>(),
            Equipment = Array.Empty<object>(),
            RawFolders = new[] { new { RawFolderId = "rf-1", ParentRawFolderId = (string?)null, Name = "Plant", ClusterId = "c1" } },
            DriverInstances = new[] { new { DriverInstanceId = "drv-1", DriverType = "Modbus", DriverConfig = "{}", Name = "Modbus", RawFolderId = "rf-1", ClusterId = "c1" } },
            Devices = new[] { new { DeviceId = "dev-1", DriverInstanceId = "drv-1", Name = "dev1", DeviceConfig = "{}" } },
            TagGroups = Array.Empty<object>(),
            Tags = Enumerable.Range(0, tagCount).Select(i => new
            {
                TagId = $"t-{i}", DeviceId = "dev-1", TagGroupId = (string?)null, Name = $"T{i}",
                DataType = "Float", AccessLevel = 0, TagConfig = "{}",
            }).ToArray(),
            UnsTagReferences = Array.Empty<object>(),
            ScriptedAlarms = Array.Empty<object>(),
        });

    private sealed class OrderedSink : IOpcUaAddressSpaceSink, ISurgicalAddressSpaceSink
    {
        public ConcurrentQueue<string> Calls { get; } = new();
        public int Completed;

        public void WriteValue(string nodeId, object? value, uint statusCode, DateTime ts, AddressSpaceRealm realm) => Calls.Enqueue($"WV:{nodeId}");
        public void WriteValues(IReadOnlyList<ValueWrite> values) { foreach (var v in values) Calls.Enqueue($"WV:{v.NodeId}"); }
        public void WriteAlarmQuality(string alarmNodeId, uint statusCode, DateTime ts, AddressSpaceRealm realm) { }
        public void WriteAlarmCondition(string alarmNodeId, AlarmConditionSnapshot state, DateTime ts, AddressSpaceRealm realm) => Calls.Enqueue($"WA:{alarmNodeId}");
        public void MaterialiseAlarmCondition(string alarmNodeId, string equipmentNodeId, string displayName, string alarmType, int severity, AddressSpaceRealm realm, bool isNative = false) => Calls.Enqueue($"MA:{alarmNodeId}");
        public void EnsureFolder(string folderNodeId, string? parentNodeId, string displayName, AddressSpaceRealm realm) => Calls.Enqueue($"EF:{folderNodeId}");
        public void EnsureVariable(string variableNodeId, string? parentFolderNodeId, string displayName, string dataType, bool writable, AddressSpaceRealm realm, string? historianTagname = null, bool isArray = false, uint? arrayLength = null, bool refuseArrayOnScalar = false) => Calls.Enqueue($"EV:{variableNodeId}");
        public void RebuildAddressSpace() => Calls.Enqueue("RB");
        public void CompleteAddressSpaceApply() => Interlocked.Increment(ref Completed);
        public void RaiseNodesAddedModelChange(string affectedNodeId, AddressSpaceRealm realm) => Calls.Enqueue($"NA:{affectedNodeId}");
        public void WireAlarmNotifiers(string alarmNodeId, AddressSpaceRealm alarmRealm, IReadOnlyList<string> notifierFolderNodeIds, AddressSpaceRealm notifierFolderRealm) { }
        public void AddReference(string sourceNodeId, AddressSpaceRealm sourceRealm, string targetNodeId, AddressSpaceRealm targetRealm, string referenceType = "Organizes") { }
        public bool UpdateTagAttributes(string variableNodeId, bool writable, string? historianTagname, string dataType, bool isArray, uint? arrayLength, AddressSpaceRealm realm) { Calls.Enqueue($"UT:{variableNodeId}"); return true; }
        public bool UpdateFolderDisplayName(string folderNodeId, string displayName, AddressSpaceRealm realm) => true;
        public bool RemoveVariableNode(string variableNodeId, AddressSpaceRealm realm) { Calls.Enqueue($"RV:{variableNodeId}"); return true; }
        public bool RemoveAlarmConditionNode(string alarmNodeId, AddressSpaceRealm realm) { Calls.Enqueue($"RA:{alarmNodeId}"); return true; }
        public bool RemoveEquipmentSubtree(string equipmentNodeId, AddressSpaceRealm realm) { Calls.Enqueue($"RE:{equipmentNodeId}"); return true; }
    }
}