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 request presenting an OLDER generation — one whose authority snapshot was captured before a
    ///     deploy and which then lost the scheduling race — must not roll the cache stamp backward, and must
    ///     not purge the new generation's entries on its way past.
    /// </summary>
    /// <remarks>
    ///     <para>
    ///         The generation itself is monotonic by construction; what is not monotonic is the order
    ///         requests <i>arrive</i> holding it. Treating a straggler as a deploy inverts the stamp and
    ///         wipes shards that were populated correctly, so the post-deploy window — the coldest the cache
    ///         ever is, and the busiest repopulating — is precisely when it would thrash.
    ///     </para>
    ///     <para>
    ///         Nothing here uses <c>AdvanceGenerationForTest</c>: this is the ordinary lookup path,
    ///         observed. The straggler must miss (its snapshot is superseded, so recomputing is right) AND
    ///         leave everything else standing — a test asserting only the miss passes against the defect,
    ///         because the straggler misses either way.
    ///     </para>
    /// </remarks>
    [Fact]
    public void A_straggler_from_an_older_generation_neither_clears_nor_rolls_the_stamp_back()
    {
        var cache = new SessionAclDecisionCache();

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

        // The straggler: same question, superseded snapshot.
        cache.TryGet(generation: 1, Session, OpcUaOperation.Read, AddressSpaceRealm.Raw, Node, out _)
            .ShouldBeFalse("a superseded snapshot must recompute rather than read the newer generation");

        // Asserted BEFORE the generation-2 lookup below, which would itself restore a rolled-back stamp and
        // hide the defect.
        cache.Generation.ShouldBe(2, "the straggler must not have rolled the stamp backward");
        cache.Count.ShouldBe(1, "and must not have purged the current generation's entries");

        cache.TryGet(generation: 2, Session, OpcUaOperation.Read, AddressSpaceRealm.Raw, Node, out var decision)
            .ShouldBeTrue("the current generation's decision survived the straggler");
        decision.ShouldBeTrue();
    }

    /// <summary>
    ///     The store half of the same window: a <c>Set</c> that meets the previous generation's shard must
    ///     displace it, not fill it.
    /// </summary>
    /// <remarks>
    ///     Filing into an orphaned shard is never <i>wrong</i> — the read path's generation check means
    ///     nothing stored there can ever be served — which is exactly why it needs its own pin: the symptom
    ///     is silent, and it is a dead cache rather than a leaking one. Every store for that key would land
    ///     somewhere unreadable, consuming the key's whole per-shard budget before an eviction swap finally
    ///     minted a shard for the current generation. Twenty thousand wasted inserts per key per deployment,
    ///     with a hit rate of zero throughout and nothing in the counters to say so.
    /// </remarks>
    [Fact]
    public void A_store_meeting_the_previous_generations_shard_displaces_it()
    {
        var cache = new SessionAclDecisionCache();

        Ask(cache, generation: 1, compute: () => true).ShouldBeTrue();
        cache.AdvanceGenerationForTest(2);

        // Inside the window: the stamp says 2, the shard in the table was built for 1.
        cache.Set(generation: 2, Session, OpcUaOperation.Read, AddressSpaceRealm.Raw, Node, decision: false);

        cache.TryGet(generation: 2, Session, OpcUaOperation.Read, AddressSpaceRealm.Raw, Node, out var decision)
            .ShouldBeTrue("the store must have landed in a shard the current generation can read");
        decision.ShouldBeFalse("and it must be the new verdict, not the one it replaced");

        // The displaced shard was sealed and unbooked, so the survivor is the single new entry.
        cache.Count.ShouldBe(1);
        cache.TrackedCount.ShouldBe(1);
        cache.LiveShardCount.ShouldBe(1);
    }

    /// <summary>
    ///     The same rollover, arrived at the way production does — a generation moving under live traffic
    ///     rather than through the test seam. No hit may ever return a verdict stored under a different
    ///     generation, and the counters must still settle exactly.
    /// </summary>
    /// <remarks>
    ///     <para>
    ///         The generation is <b>encoded in the stored verdict</b> (<c>gen % 2 == 0</c>), which is what
    ///         makes a leak visible at all: a cache that answers with the right shape but the wrong
    ///         generation's data is indistinguishable from a correct one unless the data says which
    ///         generation it came from. A miss recomputes the encoding for the generation it was asked
    ///         about, so only a hit can ever disagree.
    ///     </para>
    ///     <para>
    ///         The bumps come from inside the worker mix rather than from a spinning thread of their own,
    ///         and are spaced far enough apart for entries to accumulate between them — a generation moving
    ///         faster than a lookup completes makes every call a miss, which is a cache that is never read
    ///         and therefore a test that pins nothing.
    ///     </para>
    /// </remarks>
    [Fact]
    public void No_hit_ever_returns_a_verdict_from_another_generation()
    {
        var cache = new SessionAclDecisionCache();
        const int workers = 6;
        const int writesEach = 30_000;
        const int bumpEvery = 500;
        const int rounds = 4;

        // A deliberately narrow node and key space: workers must keep landing on entries their peers
        // stored, and on shards a bump has just orphaned. Unique identifiers would make every call a miss.
        const int nodes = 64;
        const int keys = 2;

        for (var r = 0; r < rounds; r++)
        {
            var round = r;
            var generation = (round * 100L) + 1L;
            var mismatches = 0;
            var start = new Barrier(workers);
            var threads = new Thread[workers];

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

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

                    for (var i = 0; i < writesEach; i++)
                    {
                        // One worker advances the generation, so the bump lands in the middle of the others'
                        // traffic rather than beside it.
                        if (worker == 0 && i % bumpEvery == 0) Interlocked.Increment(ref generation);

                        var gen = Interlocked.Read(ref generation);
                        var expected = gen % 2 == 0;

                        var decided = cache.GetOrAdd(
                            gen, $"key-{worker % keys}", OpcUaOperation.Read, AddressSpaceRealm.Raw,
                            $"Plant/dev1/Tag{i % nodes}", () => expected);

                        if (decided != expected) Interlocked.Increment(ref mismatches);
                    }
                }) { IsBackground = true };
            }

            foreach (var thread in threads) thread.Start();
            foreach (var thread in threads) thread.Join();
            start.Dispose();

            mismatches.ShouldBe(
                0, $"round {round}: a hit served a verdict taken under a different generation");

            var survived = cache.Count;

            cache.LiveShardCount.ShouldBe(
                survived, $"round {round}: a shard's own counter drifted from its entries");
            cache.TrackedCount.ShouldBe(
                survived, $"round {round}: 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);
    }

    /// <summary>
    ///     The generation-rollover window, made deterministic. Publishing the new generation and purging the
    ///     shards are two steps, and a racer arriving between them sees a generation that already matches
    ///     and shards that have not been dropped yet — so it must not be able to read one.
    /// </summary>
    /// <remarks>
    ///     <para>
    ///         <c>AdvanceGenerationForTest</c> is the racer's view exactly: the stamp moves, nothing clears.
    ///         Before the fix this test served the generation-1 verdict to a generation-2 caller. Key
    ///         revocation and expiry are re-checked above the cache, so the leak was a stale <i>grant walk</i>
    ///         rather than a usable revoked key — real, but narrower than it first looks.
    ///     </para>
    ///     <para>
    ///         ⚠️ The assertion is a <b>miss</b>, not a corrected value: serving <c>false</c> from a shard
    ///         built under the old generation would be just as wrong, and a value assertion would pass on it.
    ///     </para>
    /// </remarks>
    [Fact]
    public void A_shard_built_under_the_previous_generation_is_never_served()
    {
        var cache = new SessionAclDecisionCache();

        // GetOrAdd, not a bare Set: Set drops a decision whose generation is not already the stamped one,
        // and a fresh cache is stamped -1. The lookup inside GetOrAdd is what publishes generation 1.
        Ask(cache, generation: 1, compute: () => true).ShouldBeTrue();
        cache.TryGet(generation: 1, Session, OpcUaOperation.Read, AddressSpaceRealm.Raw, Node, out var stored)
            .ShouldBeTrue("the entry must exist before the window opens, or the test proves nothing");
        stored.ShouldBeTrue();

        // Stand inside the window: the new generation is published, the shards are not yet purged.
        cache.AdvanceGenerationForTest(2);

        cache.TryGet(generation: 2, Session, OpcUaOperation.Read, AddressSpaceRealm.Raw, Node, out _)
            .ShouldBeFalse("a shard may only answer for the generation it was built under");
    }

    /// <summary>
    ///     A shard displaced by the per-shard eviction swap keeps taking inserts correctly, with the counters
    ///     still exact — the behaviour pin behind the capacity hint the swap now passes.
    /// </summary>
    /// <remarks>
    ///     <b>Pin, not red-first.</b> The hint sizes a dictionary's initial buckets, which is unobservable
    ///     through the public surface: no assertion here can distinguish a hinted shard from a default one,
    ///     and none tries to. What it pins is that passing it changed nothing about the refill — that the
    ///     replacement shard still accepts entries and still books them exactly once.
    /// </remarks>
    [Fact]
    public void A_shard_replaced_by_the_eviction_swap_still_refills_exactly()
    {
        var cache = new SessionAclDecisionCache();
        const int overshoot = 12;

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

        cache.Count.ShouldBe(overshoot);
        cache.TrackedCount.ShouldBe(overshoot);
        cache.LiveShardCount.ShouldBe(overshoot);

        // The refilled shard serves what it was given, and a fresh insert into it still books once.
        Ask(cache, generation: 1, session: "crawler-key", node: "Plant/dev1/Tag20011",
                compute: () => throw new InvalidOperationException("this decision was already memoized"))
            .ShouldBeTrue();
        Ask(cache, generation: 1, session: "crawler-key", node: "Plant/dev1/Fresh", compute: () => false)
            .ShouldBeFalse();

        cache.Count.ShouldBe(overshoot + 1);
        cache.TrackedCount.ShouldBe(cache.Count);
        cache.LiveShardCount.ShouldBe(cache.Count);
    }

    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);
}