lmxopcua/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/ProjectedEventDeliveryWireTests.cs

using Opc.Ua;
using Opc.Ua.Client;
using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.Commons.OpcUa;
using ZB.MOM.WW.OtOpcUa.Configuration.Entities;
using ZB.MOM.WW.OtOpcUa.Configuration.Enums;
using ZB.MOM.WW.OtOpcUa.Core.Authorization;
using MonitoredItem = Opc.Ua.Client.MonitoredItem;

namespace ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests;

/// <summary>
///     #855 perf: condition events are queued to many items through one shared projection per (filter, locales) instead
///     of the SDK's per-item <c>QueueEvent(IFilterTarget)</c>. The two paths must deliver the SAME events with the SAME
///     fields: this runs one scenario through each (the node manager's switch) and compares every item's stream —
///     folder and Server-object items, value-equal filters on distinct objects, a where clause that rejects some events,
///     a where clause on the type, a narrower select list, and a Disabled item.
/// </summary>
public sealed class ProjectedEventDeliveryWireTests
{
    private const string EqA = ServerEventRig.EqA;

    [Theory]
    [InlineData(false)]
    [InlineData(true)]
    public async Task Projected_delivery_queues_exactly_what_the_sdk_queues(bool enforced)
    {
        var ct = TestContext.Current.CancellationToken;
        await using var rig = await ServerEventRig.StartAsync("Projected855", encrypted: false, ct);
        rig.KeyRoles = [];
        rig.MaterialiseNative();
        rig.Enforce(enforced, [ServerEventRig.Key()], KeyEquipmentGrant(EqA));

        using var session = await rig.ConnectAsync(anonymous: false, ct);
        var events = new RecordedEvents();
        var subscription = await RecordedEvents.SubscribeAsync(session, events, ct);
        var eqA = ServerEventRig.UnsNodeId(session, EqA);
        var a1 = RecordedEvents.AddEventItem(subscription, eqA);
        var a2 = RecordedEvents.AddEventItem(subscription, eqA); // a value-equal filter on a distinct object
        var severe = AddItem(subscription, eqA, SevereOnly());
        var ofType = AddItem(subscription, eqA, OfTypeAlarm());
        var disabled = RecordedEvents.AddEventItem(subscription, eqA);
        var onServer = RecordedEvents.AddEventItem(subscription, ObjectIds.Server);
        await subscription.ApplyChangesAsync(ct);
        foreach (var item in subscription.MonitoredItems) item.Status.Error.ShouldBeNull();
        await subscription.SetMonitoringModeAsync(MonitoringMode.Disabled, [disabled], ct);

        var before = ProjectedEventDelivery.ProjectedQueued;
        await PhaseAsync(rig, events, a1, onServer, "p", projectedPath: true);
        ProjectedEventDelivery.ProjectedQueued.ShouldBeGreaterThan(before, "the projected path must actually be taken");
        await PhaseAsync(rig, events, a1, onServer, "s", projectedPath: false);

        foreach (var item in new[] { a1, a2, severe, ofType, disabled, onServer })
        {
            var p = Normalised(events.Of(item), "p");
            var s = Normalised(events.Of(item), "s");
            p.ShouldBe(s, $"{item.DisplayName}: the projected path must deliver exactly the SDK's events and fields");
        }

        Normalised(events.Of(a1), "p").Count.ShouldBe(Severities.Length);
        Normalised(events.Of(severe), "p").Count.ShouldBe(Severities.Count(v => v > 800), "the where clause rejects some");
        Normalised(events.Of(ofType), "p").Count.ShouldBe(Severities.Length);
        events.Of(disabled).ShouldBeEmpty("a Disabled item queues nothing on either path");
    }

    /// <summary>
    ///     The projected path rests on PRIVATE members of the pinned SDK's <c>MonitoredItem</c> (its lock, its event queue
    ///     handler, the filter in use). If an SDK upgrade drops one, delivery silently falls back to the SDK's per-item
    ///     <c>QueueEvent</c> (correct, slower) — this fails first, so the upgrade is looked at.
    /// </summary>
    [Fact]
    public void The_sdk_members_the_projected_path_reads_are_still_resolvable() =>
        ProjectedEventDelivery.Available.ShouldBeTrue(
            "MonitoredItem.m_lock / m_eventQueueHandler / FilterToUse changed — re-check ProjectedEventDelivery.TryQueue against the new SDK's QueueEvent(IFilterTarget)");

    private static readonly ushort[] Severities = [700, 900, 300, 850, 801, 800, 999];

    private static async Task PhaseAsync(
        ServerEventRig rig, RecordedEvents events, MonitoredItem probe, MonitoredItem onServer, string prefix, bool projectedPath)
    {
        rig.Nm.UseProjectedEventDelivery = projectedPath;
        try
        {
            for (var i = 0; i < Severities.Length; i++)
            {
                rig.Project(native: true, new AlarmConditionSnapshot(
                    i % 2 == 0, i % 3 == 0, false, true, AlarmShelvingKind.Unshelved, Severities[i], $"{prefix}{i}"));
            }

            var last = $"{prefix}{Severities.Length - 1}";
            await RecordedEvents.WaitUntilAsync(() => events.Has(probe, last) && events.Has(onServer, last));
            await Task.Delay(TimeSpan.FromMilliseconds(400));
        }
        finally
        {
            rig.Nm.UseProjectedEventDelivery = true;
        }
    }

    /// <summary>One phase's events of an item, with the per-event identity (EventId, the phase prefix) removed.</summary>
    private static List<string> Normalised(IReadOnlyList<ArrayOf<Variant>> stream, string prefix) =>
        stream
            .Where(f => RecordedEvents.Message(f)?.StartsWith(prefix, StringComparison.Ordinal) == true)
            .Select(f => string.Join(" | ", f.ToArray()!.Select((v, i) => i switch
            {
                RecordedEvents.EventIdField => v.IsNull ? "null" : "id",
                RecordedEvents.MessageField => RecordedEvents.Message(f)![prefix.Length..],
                _ => v.ToString(),
            })))
            .ToList();

    private static MonitoredItem AddItem(Subscription subscription, NodeId source, EventFilter filter)
    {
        var item = new MonitoredItem(subscription.DefaultItem)
        {
            StartNodeId = source,
            AttributeId = Attributes.EventNotifier,
            Filter = filter,
            QueueSize = 1000,
            SamplingInterval = 0,
        };
        subscription.AddItem(item);
        return item;
    }

    /// <summary>EventId, Message, Severity — only events with Severity &gt; 800.</summary>
    private static EventFilter SevereOnly()
    {
        var filter = new EventFilter();
        filter.AddSelectClause(new NodeId(ObjectTypes.BaseEventType), QualifiedName.From(BrowseNames.EventId));
        filter.AddSelectClause(new NodeId(ObjectTypes.BaseEventType), QualifiedName.From(BrowseNames.Message));
        filter.AddSelectClause(new NodeId(ObjectTypes.BaseEventType), QualifiedName.From(BrowseNames.Severity));
        var where = new ContentFilter();
        where.Push(FilterOperator.GreaterThan,
            Variant.FromStructure(new SimpleAttributeOperand(ObjectTypeIds.BaseEventType, QualifiedName.From(BrowseNames.Severity)) { AttributeId = Attributes.Value }),
            Variant.FromStructure(new LiteralOperand(new Variant((ushort)800))));
        filter.WhereClause = where;
        return filter;
    }

    /// <summary>EventId, Message, Retain, a field the event does not have — OfType AlarmConditionType.</summary>
    private static EventFilter OfTypeAlarm()
    {
        var filter = new EventFilter();
        filter.AddSelectClause(new NodeId(ObjectTypes.BaseEventType), QualifiedName.From(BrowseNames.EventId));
        filter.AddSelectClause(new NodeId(ObjectTypes.BaseEventType), QualifiedName.From(BrowseNames.Message));
        filter.AddSelectClause(ObjectTypeIds.ConditionType, QualifiedName.From(BrowseNames.Retain));
        filter.AddSelectClause(ObjectTypeIds.LimitAlarmType, QualifiedName.From(BrowseNames.HighLimit));
        var where = new ContentFilter();
        where.Push(FilterOperator.OfType, Variant.FromStructure(new LiteralOperand(new Variant(ObjectTypeIds.AlarmConditionType))));
        filter.WhereClause = where;
        return filter;
    }

    private static NodeAcl KeyEquipmentGrant(string equipment) => new()
    {
        NodeAclId = $"acl-key-{equipment}",
        ClusterId = ServerEventRig.Cluster,
        Principal = ServerEventRig.KeyPrincipal,
        ScopeKind = NodeAclScopeKind.Equipment,
        ScopeId = $"eq-{equipment}",
        PermissionFlags = NodePermissions.ReadOnly,
    };
}