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), with the
///     role / grant verdicts shared per identity, instead of the SDK's per-item <c>QueueEvent(IFilterTarget)</c> and
///     per-item checks. The two paths must deliver the SAME events with the SAME fields: these run one scenario through
///     each (the node manager's <c>UseProjectedEventDelivery</c> switch, which also turns the verdict sharing off) and
///     compare every item's stream.
/// </summary>
public sealed class ProjectedEventDeliveryWireTests
{
    private const string EqA = ServerEventRig.EqA;

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

    /// <summary>
    ///     Folder and Server-object items; value-equal filters on distinct objects; three filters with the SAME select
    ///     list and different where clauses (they share a registry bucket, never a projection); an OfType where clause
    ///     with a field the event lacks; a Disabled item; a second session with other preferred locales; an anonymous
    ///     session (no grant under the ACL) on the Server object; and a filter modified half-way through each phase.
    /// </summary>
    [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, Narrow(severityAbove: 800));
        var severer = AddItem(subscription, eqA, Narrow(severityAbove: 900));
        var unfiltered = AddItem(subscription, eqA, Narrow(severityAbove: null));
        var ofType = AddItem(subscription, eqA, OfTypeAlarm(alternative: false));
        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);

        using var german = await rig.ConnectAsync(anonymous: false, ct, preferredLocales: ["de-DE", "fr"]);
        var germanEvents = new RecordedEvents();
        var germanSubscription = await RecordedEvents.SubscribeAsync(german, germanEvents, ct);
        var g1 = RecordedEvents.AddEventItem(germanSubscription, ServerEventRig.UnsNodeId(german, EqA));
        await germanSubscription.ApplyChangesAsync(ct);
        g1.Status.Error.ShouldBeNull();

        using var anonymous = await rig.ConnectAsync(anonymous: true, ct);
        var anonEvents = new RecordedEvents();
        var anonSubscription = await RecordedEvents.SubscribeAsync(anonymous, anonEvents, ct);
        var anonServer = RecordedEvents.AddEventItem(anonSubscription, ObjectIds.Server);
        await anonSubscription.ApplyChangesAsync(ct);
        anonServer.Status.Error.ShouldBeNull();

        async Task PhaseAsync(string prefix, bool projectedPath)
        {
            rig.Nm.UseProjectedEventDelivery = projectedPath;
            try
            {
                for (var i = 0; i < Severities.Length; i++)
                {
                    if (i == Severities.Length / 2)
                    {
                        ofType.Filter = OfTypeAlarm(alternative: true); // ModifyMonitoredItems mid-phase
                        await subscription.ApplyChangesAsync(ct);
                        ofType.Status.Error.ShouldBeNull();
                    }

                    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(a1, last) && events.Has(onServer, last) && germanEvents.Has(g1, last));
                await Task.Delay(TimeSpan.FromMilliseconds(400), ct);
            }
            finally
            {
                rig.Nm.UseProjectedEventDelivery = true;
                ofType.Filter = OfTypeAlarm(alternative: false);
                await subscription.ApplyChangesAsync(ct);
            }
        }

        var before = rig.Nm.ProjectedDeliveries;
        await PhaseAsync("p", projectedPath: true);
        rig.Nm.ProjectedDeliveries.ShouldBeGreaterThan(before, "the projected path must actually be taken by this rig");
        var afterProjected = rig.Nm.ProjectedDeliveries;
        await PhaseAsync("s", projectedPath: false);
        rig.Nm.ProjectedDeliveries.ShouldBe(afterProjected, "the SDK phase must not use the projected path");

        var streams = new (RecordedEvents Events, MonitoredItem Item)[]
        {
            (events, a1), (events, a2), (events, severe), (events, severer), (events, unfiltered), (events, ofType),
            (events, disabled), (events, onServer), (germanEvents, g1), (anonEvents, anonServer),
        };
        foreach (var (recorded, item) in streams)
        {
            Normalised(recorded.Of(item), "p").ShouldBe(
                Normalised(recorded.Of(item), "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));
        Normalised(events.Of(severer), "p").Count.ShouldBe(Severities.Count(v => v > 900), "same select list, other where clause");
        Normalised(events.Of(unfiltered), "p").Count.ShouldBe(Severities.Length);
        events.Of(disabled).ShouldBeEmpty("a Disabled item queues nothing on either path");
        if (enforced) anonEvents.Of(anonServer).Count(f => f[RecordedEvents.AckedField].TryGetValue(out bool _)).ShouldBe(0);
        else Normalised(anonEvents.Of(anonServer), "p").Count.ShouldBe(Severities.Length);
    }

    /// <summary>
    ///     A small queue overflowing while the client does not publish — DiscardOldest true and false — delivers the same
    ///     survivors and the same EventQueueOverflow event, in the same position, on both paths (each on a fresh server).
    /// </summary>
    [Theory]
    [InlineData(true)]
    [InlineData(false)]
    public async Task An_overflowing_small_queue_is_identical_on_both_paths(bool discardOldest)
    {
        var projected = await OverflowScenarioAsync(discardOldest, projectedPath: true);
        var sdk = await OverflowScenarioAsync(discardOldest, projectedPath: false);
        projected.ShouldBe(sdk);
        projected.ShouldContain(f => f.StartsWith("overflow", StringComparison.Ordinal), "the queue really overflowed");
    }

    private static async Task<List<string>> OverflowScenarioAsync(bool discardOldest, bool projectedPath)
    {
        var ct = TestContext.Current.CancellationToken;
        await using var rig = await ServerEventRig.StartAsync($"Projected855Overflow{projectedPath}", encrypted: false, ct);
        rig.KeyRoles = [];
        rig.MaterialiseNative();
        rig.Enforce(false, [ServerEventRig.Key()], KeyEquipmentGrant(EqA));
        rig.Nm.UseProjectedEventDelivery = projectedPath;

        using var session = await rig.ConnectAsync(anonymous: false, ct);
        var events = new RecordedEvents();
        var subscription = await RecordedEvents.SubscribeAsync(session, events, ct);
        var item = RecordedEvents.AddEventItem(subscription, ServerEventRig.UnsNodeId(session, EqA));
        item.QueueSize = 3;
        item.DiscardOldest = discardOldest;
        await subscription.ApplyChangesAsync(ct);
        item.Status.Error.ShouldBeNull();
        await subscription.SetPublishingModeAsync(false, ct);

        for (var i = 0; i < Severities.Length; i++)
        {
            rig.Project(native: true, new AlarmConditionSnapshot(
                true, false, false, true, AlarmShelvingKind.Unshelved, Severities[i], $"o{i}"));
        }

        await subscription.SetPublishingModeAsync(true, ct);
        await RecordedEvents.WaitUntilAsync(() =>
            events.Of(item).Any(f => f[RecordedEvents.EventTypeField].GetNodeId() == ObjectTypeIds.EventQueueOverflowEventType));
        await Task.Delay(TimeSpan.FromMilliseconds(400), ct);
        return events.Of(item)
            .Select(f => f[RecordedEvents.EventTypeField].GetNodeId() == ObjectTypeIds.EventQueueOverflowEventType
                ? "overflow"
                : $"{RecordedEvents.Message(f)} sev={f[RecordedEvents.SeverityField]}")
            .ToList();
    }

    /// <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)");

    /// <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 — with the optional where clause Severity &gt; <paramref name="severityAbove"/>.</summary>
    private static EventFilter Narrow(int? severityAbove)
    {
        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();
        if (severityAbove is { } floor)
        {
            where.Push(FilterOperator.GreaterThan,
                Variant.FromStructure(new SimpleAttributeOperand(ObjectTypeIds.BaseEventType, QualifiedName.From(BrowseNames.Severity)) { AttributeId = Attributes.Value }),
                Variant.FromStructure(new LiteralOperand(new Variant((ushort)floor))));
        }

        filter.WhereClause = where;
        return filter;
    }

    /// <summary>OfType AlarmConditionType: EventId, Message, Retain, a field the event lacks (or, alternatively, Severity).</summary>
    private static EventFilter OfTypeAlarm(bool alternative)
    {
        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));
        if (alternative) filter.AddSelectClause(new NodeId(ObjectTypes.BaseEventType), QualifiedName.From(BrowseNames.Severity));
        else 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,
    };
}