using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.Commons.OpcUa;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
using ZB.MOM.WW.OtOpcUa.OpcUaServer.Security;

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

/// <summary>
///     #520 stage 6 — the decision cache behind the request-rate seams.
/// </summary>
/// <remarks>
///     The risk in a cache like this is not that it is slow; it is that it answers a question slightly
///     different from the one asked. Each test below pins one dimension the key must discriminate on,
///     because a key that collides across any of them hands one session another's authority.
/// </remarks>
public sealed class SessionAclDecisionCacheTests
{
    private const string Session = "key-1";
    private const string Node = "Plant/Modbus/dev1/Speed";

    /// <summary>The second identical question is answered from the cache.</summary>
    [Fact]
    public void A_repeated_question_is_not_recomputed()
    {
        var cache = new SessionAclDecisionCache();
        var calls = 0;

        for (var i = 0; i < 5; i++)
            Ask(cache, generation: 1, compute: () => { calls++; return true; }).ShouldBeTrue();

        calls.ShouldBe(1);
    }

    /// <summary>
    ///     A new generation drops everything. This is the invalidation the whole design rests on: grants
    ///     only change by deployment, and a deployment always advances the generation.
    /// </summary>
    [Fact]
    public void A_new_generation_invalidates_every_decision()
    {
        var cache = new SessionAclDecisionCache();

        Ask(cache, generation: 1, compute: () => true).ShouldBeTrue();
        cache.Count.ShouldBe(1);

        // The same question, one deployment later, must be re-decided — and the new answer must win.
        Ask(cache, generation: 2, compute: () => false).ShouldBeFalse();
        cache.Generation.ShouldBe(2);
    }

    /// <summary>
    ///     Two sessions asking about the same node get their own answers. A key that omitted the session
    ///     would let the first caller's verdict decide for everyone.
    /// </summary>
    [Fact]
    public void Two_sessions_do_not_share_a_decision()
    {
        var cache = new SessionAclDecisionCache();

        Ask(cache, generation: 1, session: "granted-key", compute: () => true).ShouldBeTrue();
        Ask(cache, generation: 1, session: "ungranted-key", compute: () => false).ShouldBeFalse();
    }

    /// <summary>Read and Write on one node are separate questions.</summary>
    [Fact]
    public void Two_operations_do_not_share_a_decision()
    {
        var cache = new SessionAclDecisionCache();

        Ask(cache, generation: 1, operation: OpcUaOperation.Read, compute: () => true).ShouldBeTrue();
        Ask(cache, generation: 1, operation: OpcUaOperation.WriteOperate, compute: () => false).ShouldBeFalse();
    }

    /// <summary>
    ///     The two realms are separate namespaces that can carry the SAME <c>s=</c> identifier, so the realm
    ///     is part of the key. Without it, a raw grant would answer for the UNS node of the same path —
    ///     which is exactly the cross-realm confusion the trie itself is careful to avoid.
    /// </summary>
    [Fact]
    public void The_two_realms_do_not_share_a_decision()
    {
        var cache = new SessionAclDecisionCache();

        Ask(cache, generation: 1, realm: AddressSpaceRealm.Raw, compute: () => true).ShouldBeTrue();
        Ask(cache, generation: 1, realm: AddressSpaceRealm.Uns, compute: () => false).ShouldBeFalse();
    }

    /// <summary>Different nodes are different questions.</summary>
    [Fact]
    public void Two_nodes_do_not_share_a_decision()
    {
        var cache = new SessionAclDecisionCache();

        Ask(cache, generation: 1, node: "Plant/Modbus/dev1/Speed", compute: () => true).ShouldBeTrue();
        Ask(cache, generation: 1, node: "Plant/Modbus/dev1/Torque", compute: () => false).ShouldBeFalse();
    }

    /// <summary>
    ///     Node identifiers are path segments, so they compare ordinally — <c>Speed</c> and <c>speed</c> can
    ///     both exist under one device and must not collide in the cache when they do not collide in the
    ///     trie.
    /// </summary>
    [Fact]
    public void Node_identifiers_are_case_sensitive()
    {
        var cache = new SessionAclDecisionCache();

        Ask(cache, generation: 1, node: "Plant/dev1/Speed", compute: () => true).ShouldBeTrue();
        Ask(cache, generation: 1, node: "Plant/dev1/speed", compute: () => false).ShouldBeFalse();
    }

    /// <summary>
    ///     The global bound still holds when no single session is anywhere near its own ceiling. Spreading
    ///     the entries over many sessions is the case the per-session cap cannot catch, so the global
    ///     counter is the only thing standing between a large fleet and unbounded growth.
    /// </summary>
    [Fact]
    public void The_global_bound_still_holds()
    {
        var cache = new SessionAclDecisionCache();

        // Well under MaxEntriesPerSession, so every clear observed here is the global one.
        const int perSession = 5_000;
        var total = SessionAclDecisionCache.MaxEntries + perSession;

        for (var i = 0; i < total; i++)
            Ask(
                cache, generation: 1, session: $"key-{i / perSession}", node: $"Plant/dev1/Tag{i}",
                compute: () => true);

        // The global counter is approximate under concurrency, so the contract allows one shard's worth of
        // slack; single-threaded it is exact, which is the stronger fact pinned here.
        cache.Count.ShouldBeLessThanOrEqualTo(SessionAclDecisionCache.MaxEntries);
    }

    /// <summary>
    ///     One session filling its own shard evicts only itself. This is the whole point of sharding: a
    ///     machine consumer crawling a 50k-node tree used to blow away every other session's decisions on
    ///     its way past the ceiling, so the sessions doing ordinary work paid for the crawl.
    /// </summary>
    [Fact]
    public void A_full_session_shard_clears_only_that_session()
    {
        var cache = new SessionAclDecisionCache();
        const string crawler = "crawler-key";
        const string neighbour = "neighbour-key";
        var neighbourCalls = 0;

        Ask(cache, generation: 1, session: neighbour, compute: () => { neighbourCalls++; return true; })
            .ShouldBeTrue();
        neighbourCalls.ShouldBe(1);

        for (var i = 0; i <= SessionAclDecisionCache.MaxEntriesPerSession + 10; i++)
            Ask(cache, generation: 1, session: crawler, node: $"Plant/dev1/Tag{i}", compute: () => true);

        // The crawler's own shard was cleared and restarted rather than growing without limit.
        cache.Count.ShouldBeLessThanOrEqualTo(SessionAclDecisionCache.MaxEntriesPerSession);

        // The neighbour's single decision survived the crawl — its walk is not re-run.
        Ask(cache, generation: 1, session: neighbour, compute: () => { neighbourCalls++; return true; })
            .ShouldBeTrue();
        neighbourCalls.ShouldBe(1);
    }

    /// <summary>
    ///     After a shard is evicted the global counter equals what actually survived — the arithmetic of
    ///     the swap, pinned single-threaded where it is exact and every off-by-a-shard is visible.
    /// </summary>
    [Fact]
    public void An_eviction_subtracts_exactly_what_it_displaced()
    {
        var cache = new SessionAclDecisionCache();
        const int overshoot = 25;

        for (var i = 0; i < SessionAclDecisionCache.MaxEntriesPerSession + overshoot; i++)
            Ask(cache, generation: 1, session: "crawler-key", node: $"Plant/dev1/Tag{i}", compute: () => true);

        // The cap is reached on the entry that would have been the 20,001st; everything after it is the
        // refilled shard. Subtracting anything but the displaced shard's own count would show up here.
        cache.Count.ShouldBe(overshoot);
        cache.TrackedCount.ShouldBe(cache.Count);
        cache.LiveShardCount.ShouldBe(cache.Count);
    }

    /// <summary>
    ///     The eviction race. Many writers hammer ONE shard past its ceiling while others insert into it,
    ///     so evictions and inserts interleave repeatedly; at quiescence the global counter must equal the
    ///     entries that actually survived.
    /// </summary>
    /// <remarks>
    ///     <para>
    ///         The defect this pins is not reachable single-threaded. Eviction used to empty the shard's
    ///         dictionary and then reset the counters as two steps: a writer blocked inside
    ///         <c>TryAdd</c> for the duration of the <c>Clear</c> resumes the moment the table's locks are
    ///         released — squarely in the window before the counter reset — books an increment, and has it
    ///         subtracted a second time by the reset that follows. The drift is negative-only, so it never
    ///         changed a verdict; it made the cache trim progressively earlier than it should.
    ///     </para>
    ///     <para>
    ///         ⚠️ <b>The assertion is equality, not a bound.</b> A ceiling check
    ///         (<c>TrackedCount &lt;= Count</c>, or <c>&gt;= 0</c>) passes against the broken code, because
    ///         under-counting is exactly what the broken code does. Only <c>TrackedCount == Count</c>
    ///         falsifies it — verified by reverting the fix, which turns this test red on the first round.
    ///     </para>
    /// </remarks>
    [Fact]
    public void Concurrent_eviction_and_insertion_leave_the_counter_exact()
    {
        var cache = new SessionAclDecisionCache();
        const string crawler = "crawler-key";
        const int crawlers = 8;
        const int neighbours = 2;
        const int perCrawler = 40_000;
        const int perNeighbour = 500;

        // Sized for exposure, not for coverage: one round drives the shard past its ceiling ~16 times, and
        // the interleaving that used to corrupt the counter needs a writer to land inside the few
        // nanoseconds between the clear and the counter reset. Fewer rounds let the reverted code pass.
        const int rounds = 8;

        for (var round = 0; round < rounds; round++)
        {
            var currentRound = round;
            var start = new Barrier(crawlers + neighbours);
            var workers = new Thread[crawlers + neighbours];

            for (var w = 0; w < workers.Length; w++)
            {
                var worker = w;

                // The first `crawlers` threads share ONE shard and drive it past its ceiling many times
                // over; the rest hold their own shards, so the audit also covers the shards an eviction
                // must not touch.
                var session = worker < crawlers ? crawler : $"neighbour-{worker}";
                var writes = worker < crawlers ? perCrawler : perNeighbour;

                workers[w] = new Thread(() =>
                {
                    start.SignalAndWait();

                    // Node identifiers are unique per (round, worker, index), so every store is a genuine
                    // insert and nothing is lost to a TryAdd collision instead of to the race.
                    for (var i = 0; i < writes; i++)
                        cache.GetOrAdd(
                            generation: 1, session, OpcUaOperation.Read, AddressSpaceRealm.Raw,
                            $"Plant/dev1/R{currentRound}-W{worker}-T{i}", static () => true);
                }) { IsBackground = true };
            }

            foreach (var worker in workers) worker.Start();
            foreach (var worker in workers) worker.Join();
            start.Dispose();

            var survived = cache.Count;

            cache.TrackedCount.ShouldBeGreaterThanOrEqualTo(
                0, $"round {currentRound}: the counter must never be driven negative");
            cache.LiveShardCount.ShouldBe(
                survived, $"round {currentRound}: a shard's own counter drifted from its entries");
            cache.TrackedCount.ShouldBe(
                survived, $"round {currentRound}: the global counter drifted from the live shards");
        }
    }

    /// <summary>
    ///     A decision taken while a deployment lands mid-walk is returned but NOT stored, so the new
    ///     generation never inherits a verdict its own trie did not make.
    /// </summary>
    [Fact]
    public void A_decision_overtaken_by_a_deploy_is_not_stored()
    {
        var cache = new SessionAclDecisionCache();

        // The compute callback simulates the deploy landing while the walk is in flight.
        var decided = cache.GetOrAdd(
            generation: 1, Session, OpcUaOperation.Read, AddressSpaceRealm.Raw, Node,
            () =>
            {
                Ask(cache, generation: 2, node: "unrelated", compute: () => true);
                return true;
            });

        decided.ShouldBeTrue("the caller still gets the answer its own walk produced");

        // Nothing was filed under the superseded generation — the only entry is the unrelated one the
        // callback itself made.
        cache.Generation.ShouldBe(2);
        cache.Count.ShouldBe(1);
    }

    private static bool Ask(
        SessionAclDecisionCache cache,
        long generation,
        Func<bool> compute,
        string session = Session,
        OpcUaOperation operation = OpcUaOperation.Read,
        AddressSpaceRealm realm = AddressSpaceRealm.Raw,
        string node = Node)
        => cache.GetOrAdd(generation, session, operation, realm, node, compute);
}