lmxopcua/tests/Server/ZB.MOM.WW.OtOpcUa.Host.IntegrationTests/LocalDb/LocalDbPairConvergenceTests.cs

using Xunit;
using ZB.MOM.WW.LocalDb;
using ZB.MOM.WW.OtOpcUa.Runtime.DeploymentCache;

namespace ZB.MOM.WW.OtOpcUa.Host.IntegrationTests.LocalDb;

/// <summary>
///     LocalDb Phase 1 (Task 12) — two driver nodes replicating the deployment-artifact cache over a
///     real loopback h2c transport, through the real fail-closed interceptor.
/// </summary>
/// <remarks>
///     This is the test the whole phase exists to pass: does a redundant driver pair actually share
///     the cached configuration a node boots from when central SQL is down? Every upstream piece —
///     the schema, the DI wiring, the interceptor, the chunked cache — can be individually green
///     while the pair still fails to converge.
/// </remarks>
[Collection("LocalDbPairConvergence")]
public sealed class LocalDbPairConvergenceTests
{
    private const string Cluster = "cluster-1";

    /// <summary>
    ///     A deterministic artifact large enough to force multiple 128 KiB chunks (≈ 3 here), so the
    ///     tests exercise the chunk-split/reassemble path rather than a single-row shortcut.
    /// </summary>
    private static byte[] MultiChunkArtifact(int seed, int size = 300 * 1024)
    {
        var bytes = new byte[size];
        new Random(seed).NextBytes(bytes);
        return bytes;
    }

    private static async Task<byte[]?> ReadArtifactAsync(IDeploymentArtifactCache cache, string cluster)
    {
        var cached = await cache.GetCurrentAsync(cluster, TestContext.Current.CancellationToken);
        return cached?.Artifact;
    }

    [Fact]
    public async Task ArtifactStoredOnA_ConvergesToB_ByteIdentical()
    {
        await using var harness = new LocalDbPairHarness();
        await harness.StartAsync();

        var deploymentId = Guid.NewGuid().ToString("N");
        var artifact = MultiChunkArtifact(seed: 1);
        await harness.CacheA.StoreAsync(Cluster, deploymentId, "rev-1", artifact,
            TestContext.Current.CancellationToken);

        // B's cache reassembles byte-for-byte what A stored — proving the chunk rows, chunk_count,
        // and pointer all replicated intact through the real transport + interceptor.
        await LocalDbPairHarness.WaitUntilAsync(
            async () =>
            {
                var onB = await ReadArtifactAsync(harness.CacheB, Cluster);
                return onB is not null && onB.AsSpan().SequenceEqual(artifact);
            },
            "the artifact stored on node A to be byte-identical on node B");

        // The pointer row on B carries A's origin HLC + node id, not a locally re-derived stamp: the
        // row_version dumps for the pointer table are identical on both nodes. This is what
        // distinguishes "B replicated A's row" from "both nodes happened to write equal content".
        await LocalDbPairHarness.WaitUntilAsync(
            async () =>
                await LocalDbPairHarness.DumpRowVersionAsync(harness.A, DeploymentCacheSchema.PointerTable)
                == await LocalDbPairHarness.DumpRowVersionAsync(harness.B, DeploymentCacheSchema.PointerTable),
            "both nodes' pointer row_version (hlc + origin node id) to match");
    }

    [Fact]
    public async Task RetentionPruneOnA_TombstonesReachB()
    {
        await using var harness = new LocalDbPairHarness();
        await harness.StartAsync();

        // Three deployments for one cluster. The cache retains the newest two, so the first is pruned
        // on A — and that prune must replicate to B as tombstones, not linger as live chunks.
        var dep1 = Guid.NewGuid().ToString("N");
        var dep2 = Guid.NewGuid().ToString("N");
        var dep3 = Guid.NewGuid().ToString("N");

        await harness.CacheA.StoreAsync(Cluster, dep1, "rev-1", MultiChunkArtifact(1),
            TestContext.Current.CancellationToken);
        await harness.CacheA.StoreAsync(Cluster, dep2, "rev-2", MultiChunkArtifact(2),
            TestContext.Current.CancellationToken);

        // dep1's chunks are still present until the third store prunes them — confirm they reached B
        // first, so the later disappearance is a real replicated prune rather than never-arrived.
        await LocalDbPairHarness.WaitUntilAsync(
            async () => await LocalDbPairHarness.ChunkCountAsync(harness.B, dep1) > 0,
            "dep1's chunks to reach node B before the prune");

        await harness.CacheA.StoreAsync(Cluster, dep3, "rev-3", MultiChunkArtifact(3),
            TestContext.Current.CancellationToken);

        // A pruned dep1 locally; B must follow via replicated deletes: dep1 gone, tombstones present,
        // and the surviving newest (dep3) intact.
        await LocalDbPairHarness.WaitUntilAsync(
            async () =>
                await LocalDbPairHarness.ChunkCountAsync(harness.B, dep1) == 0
                && await LocalDbPairHarness.TombstoneCountAsync(harness.B, DeploymentCacheSchema.ArtifactsTable) > 0
                && await LocalDbPairHarness.ChunkCountAsync(harness.B, dep3) > 0,
            "node B to prune dep1 (with tombstones) while keeping dep3");
    }

    [Fact]
    public async Task WipedNode_IsBackFilledByItsPeer_WithoutAnyNewDeploy()
    {
        // The Phase 1 live gate's check 4, as an offline scenario. A node whose LocalDb volume is
        // lost used to come back permanently empty: the pair had already converged, so every oplog
        // row was acked and pruned, and the healthy node had no delta left to send. Nothing short of
        // a fresh deploy could repopulate it — which is the worst possible moment to depend on
        // central being reachable, since the cache exists precisely for when it is not.
        //
        // Fixed in ZB.MOM.WW.LocalDb 0.1.2 (snapshot resync now measures the peer's gap against
        // last_acked_seq when the oplog is empty). Pin the payoff at this level, not just the
        // library's: what matters here is that the deployment artifact itself comes back.
        await using var harness = new LocalDbPairHarness();
        await harness.StartAsync();

        var deploymentId = Guid.NewGuid().ToString("N");
        var artifact = MultiChunkArtifact(seed: 20);
        await harness.CacheA.StoreAsync(Cluster, deploymentId, "rev-1", artifact,
            TestContext.Current.CancellationToken);

        await LocalDbPairHarness.WaitUntilAsync(
            async () => await ReadArtifactAsync(harness.CacheB, Cluster) is { } b && b.AsSpan().SequenceEqual(artifact),
            "the artifact to converge before the wipe");

        // Convergence alone is not the precondition — an emptied oplog on A is. That is the state
        // that used to make back-fill impossible, so assert it rather than assume it.
        await LocalDbPairHarness.WaitUntilAsync(
            async () => await LocalDbPairHarness.OplogDepthAsync(harness.A) == 0,
            "node A's oplog to be fully acked and pruned");

        await harness.WipePassiveAsync();
        Assert.Null(await ReadArtifactAsync(harness.CacheB, Cluster));

        await harness.RestartPairAsync();

        // No new deploy, no central: the rebuilt node gets its cached configuration back from its
        // peer alone, byte-identical, and can boot from cache again.
        await LocalDbPairHarness.WaitUntilAsync(
            async () =>
            {
                var cached = await harness.CacheB.GetCurrentAsync(Cluster, TestContext.Current.CancellationToken);
                return cached is not null
                       && cached.DeploymentId == deploymentId
                       && cached.Artifact.AsSpan().SequenceEqual(artifact);
            },
            "the wiped node to be back-filled from its peer");
    }

    [Fact]
    public async Task WritesWhileTransportDown_SurviveRejoin()
    {
        await using var harness = new LocalDbPairHarness();
        await harness.StartAsync();

        var dep1 = Guid.NewGuid().ToString("N");
        var first = MultiChunkArtifact(seed: 10);
        await harness.CacheA.StoreAsync(Cluster, dep1, "rev-1", first,
            TestContext.Current.CancellationToken);
        await LocalDbPairHarness.WaitUntilAsync(
            async () => await ReadArtifactAsync(harness.CacheB, Cluster) is { } b && b.AsSpan().SequenceEqual(first),
            "the first artifact to converge before the outage");

        // B goes offline; A keeps deploying. B's database survives the host teardown (pre-constructed
        // instance), so the rejoin is genuine rather than a fresh node.
        await harness.StopPassiveAsync();

        var dep2 = Guid.NewGuid().ToString("N");
        var second = MultiChunkArtifact(seed: 11);
        await harness.CacheA.StoreAsync(Cluster, dep2, "rev-2", second,
            TestContext.Current.CancellationToken);

        await harness.RestartPairAsync();

        // Everything written during the outage catches up: B now serves the newer deployment.
        await LocalDbPairHarness.WaitUntilAsync(
            async () =>
            {
                var cached = await harness.CacheB.GetCurrentAsync(Cluster, TestContext.Current.CancellationToken);
                return cached is not null
                       && cached.DeploymentId == dep2
                       && cached.Artifact.AsSpan().SequenceEqual(second);
            },
            "node B to catch up on the deployment written while it was offline");
    }

    [Fact]
    public async Task WrongApiKey_NeverConverges_WithMatchingKeyPositiveControl()
    {
        // Negative half: A and B hold different keys, so B's interceptor fail-closes on A's stream.
        await using (var mismatched = new LocalDbPairHarness(
            apiKeyA: LocalDbPairHarness.DefaultApiKey, apiKeyB: "a-different-key-entirely"))
        {
            await mismatched.StartAsync();

            var deploymentId = Guid.NewGuid().ToString("N");
            await mismatched.CacheA.StoreAsync(Cluster, deploymentId, "rev-1", MultiChunkArtifact(20),
                TestContext.Current.CancellationToken);

            var converged = await LocalDbPairHarness.HeldWithinAsync(
                async () => await ReadArtifactAsync(mismatched.CacheB, Cluster) is not null,
                TimeSpan.FromSeconds(5));

            Assert.False(converged,
                "a key mismatch must stop the pair converging — the interceptor is fail-closed");
        }

        // Positive control: the SAME scenario with matching keys DOES converge. Without this, the
        // negative assertion above would pass even if convergence were broken for an unrelated reason.
        await using var matched = new LocalDbPairHarness();
        await matched.StartAsync();

        var controlId = Guid.NewGuid().ToString("N");
        var controlArtifact = MultiChunkArtifact(seed: 21);
        await matched.CacheA.StoreAsync(Cluster, controlId, "rev-1", controlArtifact,
            TestContext.Current.CancellationToken);

        await LocalDbPairHarness.WaitUntilAsync(
            async () => await ReadArtifactAsync(matched.CacheB, Cluster) is { } b && b.AsSpan().SequenceEqual(controlArtifact),
            "the matching-key control pair to converge");
    }
}