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.Abstractions;
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>
    ///     A walk that reaches one notifier twice — a Line folder above two equipment folders the native alarm is wired into
    ///     (alarm → E1 → Line, alarm → E2 → Line) — must queue each event to the Line's item once, as the SDK's duplicate
    ///     scan did. Both paths, compared; no EventId may repeat.
    /// </summary>
    [Fact]
    public async Task A_notifier_reached_twice_in_one_walk_gets_each_event_once_on_both_paths()
    {
        var ct = TestContext.Current.CancellationToken;
        await using var rig = await ServerEventRig.StartAsync("Projected855Twice", encrypted: false, ct);
        rig.KeyRoles = [];
        rig.Enforce(false, [ServerEventRig.Key()], KeyEquipmentGrant(EqA));
        var sink = rig.Sink;
        sink.EnsureFolder(ServerEventRig.RawDeviceFolder, parentNodeId: null, displayName: "dev1", realm: AddressSpaceRealm.Raw);
        sink.MaterialiseAlarmCondition(ServerEventRig.RawAlarmPath, ServerEventRig.RawDeviceFolder, "temp_hi", "OffNormalAlarm", 700, AddressSpaceRealm.Raw, isNative: true);
        sink.EnsureFolder("L1", parentNodeId: null, displayName: "L1", realm: AddressSpaceRealm.Uns);
        sink.EnsureFolder("E1", parentNodeId: "L1", displayName: "E1", realm: AddressSpaceRealm.Uns);
        sink.EnsureFolder("E2", parentNodeId: "L1", displayName: "E2", realm: AddressSpaceRealm.Uns);
        // A condition directly under L1 makes L1 an event notifier (only folders a condition hangs under, or is wired into, are).
        sink.MaterialiseAlarmCondition("line-alarm", "L1", "line", "AlarmCondition", 500, AddressSpaceRealm.Uns, isNative: false);
        sink.WireAlarmNotifiers(ServerEventRig.RawAlarmPath, AddressSpaceRealm.Raw, ["E1", "E2"], AddressSpaceRealm.Uns);

        using var session = await rig.ConnectAsync(anonymous: false, ct);
        var events = new RecordedEvents();
        var subscription = await RecordedEvents.SubscribeAsync(session, events, ct);
        var onLine = RecordedEvents.AddEventItem(subscription, ServerEventRig.UnsNodeId(session, "L1"));
        var onE1 = RecordedEvents.AddEventItem(subscription, ServerEventRig.UnsNodeId(session, "E1"));
        await subscription.ApplyChangesAsync(ct);
        onLine.Status.Error.ShouldBeNull();

        foreach (var (prefix, projected) in new[] { ("p", true), ("s", false) })
        {
            rig.Nm.UseProjectedEventDelivery = projected;
            for (var i = 0; i < 5; i++) rig.Project(native: true, State($"{prefix}{i}"));
            await RecordedEvents.WaitUntilAsync(() => events.Has(onLine, $"{prefix}4") && events.Has(onE1, $"{prefix}4"));
            await Task.Delay(TimeSpan.FromMilliseconds(300), ct);
            rig.Nm.UseProjectedEventDelivery = true;
        }

        var ids = events.Of(onLine)
            .Where(f => f[RecordedEvents.AckedField].TryGetValue(out bool _))
            .Select(f => Convert.ToHexString(f[RecordedEvents.EventIdField].GetByteString().ToArray()!))
            .ToList();
        ids.Count.ShouldBe(10);
        ids.Distinct().Count().ShouldBe(ids.Count, "each event once per item, though the walk reaches L1 twice");
        Normalised(events.Of(onLine), "p").ShouldBe(Normalised(events.Of(onLine), "s"));
    }

    /// <summary>
    ///     Events still QUEUED when the item is modified (publishing off, then a new select list) are re-projected by the
    ///     SDK from each queued event's Handle; the projected path's per-item <see cref="EventFieldList"/> must take part in
    ///     that exactly like the SDK's own (each on a fresh server).
    /// </summary>
    [Fact]
    public async Task Queued_events_re_projected_by_a_modify_are_identical_on_both_paths()
    {
        var projected = await ModifyScenarioAsync(projectedPath: true);
        var sdk = await ModifyScenarioAsync(projectedPath: false);
        projected.ShouldBe(sdk);
        projected.Count.ShouldBe(Severities.Length);
    }

    private static async Task<List<string>> ModifyScenarioAsync(bool projectedPath)
    {
        var ct = TestContext.Current.CancellationToken;
        await using var rig = await ServerEventRig.StartAsync($"Projected855Modify{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 = AddItem(subscription, ServerEventRig.UnsNodeId(session, EqA), OfTypeAlarm(alternative: false));
        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], $"m{i}"));
        }

        item.Filter = OfTypeAlarm(alternative: true); // a new select list while every event is still queued
        await subscription.ApplyChangesAsync(ct);
        item.Status.Error.ShouldBeNull();
        await subscription.SetPublishingModeAsync(true, ct);
        await RecordedEvents.WaitUntilAsync(() => events.Has(item, $"m{Severities.Length - 1}"));
        await Task.Delay(TimeSpan.FromMilliseconds(300), ct);
        return Normalised(events.Of(item), "m");
    }

    /// <summary>
    ///     The two keys the grouping rests on: the locale key is unambiguous (null, empty, one empty locale, and a locale
    ///     containing the separator are all distinct), and a where clause hashes by value (two value-equal filters land in
    ///     one bucket and share a projection).
    /// </summary>
    [Fact]
    public void Locale_keys_are_unambiguous_and_where_clauses_hash_by_value()
    {
        string?[] keys =
        [
            ProjectedEventDelivery.LocalesKey(null), ProjectedEventDelivery.LocalesKey([]),
            ProjectedEventDelivery.LocalesKey([""]), ProjectedEventDelivery.LocalesKey(["a\u0001b"]),
            ProjectedEventDelivery.LocalesKey(["a", "b"]), ProjectedEventDelivery.LocalesKey(["a;b"]),
            ProjectedEventDelivery.LocalesKey(["a", ";b"]),
        ];
        keys.Distinct().Count().ShouldBe(keys.Length);
        Narrow(800).WhereClause.GetHashCode().ShouldBe(Narrow(800).WhereClause.GetHashCode());
        Narrow(800).WhereClause.GetHashCode().ShouldNotBe(Narrow(900).WhereClause.GetHashCode());
    }

    /// <summary>
    ///     Sharing really happens (not only equivalence): three items with value-equal filters on one session and one on a
    ///     session with other locales cost TWO projections per report (one per locale set), not four.
    /// </summary>
    [Fact]
    public async Task Value_equal_filters_share_one_projection_per_locale_set()
    {
        var ct = TestContext.Current.CancellationToken;
        await using var rig = await ServerEventRig.StartAsync("Projected855Share", encrypted: false, ct);
        rig.KeyRoles = [];
        rig.MaterialiseNative();
        rig.Enforce(false, [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 items = Enumerable.Range(0, 3).Select(_ => RecordedEvents.AddEventItem(subscription, eqA)).ToList();
        await subscription.ApplyChangesAsync(ct);

        using var german = await rig.ConnectAsync(anonymous: false, ct, preferredLocales: ["de-DE"]);
        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);

        var before = rig.Nm.ProjectionsCreated;
        const int Reports = 5;
        for (var i = 0; i < Reports; i++) rig.Project(native: true, State($"x{i}"));
        (rig.Nm.ProjectionsCreated - before).ShouldBe(2 * Reports, "one projection per (filter group, locale set) per report");
        await RecordedEvents.WaitUntilAsync(() => items.All(m => events.Has(m, $"x{Reports - 1}")) && germanEvents.Has(g1, $"x{Reports - 1}"));
    }

    private static AlarmConditionSnapshot State(string message) =>
        new(true, false, false, true, AlarmShelvingKind.Unshelved, 700, message);

    /// <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,
    };
}