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

using System.Net;
using Microsoft.AspNetCore.Builder;
using Microsoft.AspNetCore.Hosting;
using Microsoft.AspNetCore.Hosting.Server;
using Microsoft.AspNetCore.Hosting.Server.Features;
using Microsoft.AspNetCore.Server.Kestrel.Core;
using Microsoft.Extensions.Configuration;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Hosting;
using Microsoft.Extensions.Logging.Abstractions;
using Xunit;
using ZB.MOM.WW.LocalDb;
using ZB.MOM.WW.LocalDb.Replication;
using ZB.MOM.WW.OtOpcUa.Host.Configuration;
using ZB.MOM.WW.OtOpcUa.Runtime.DeploymentCache;

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

/// <summary>
///     Serializes the two-node convergence tests against one another: each stands up a real Kestrel
///     h2c listener plus two SQLite files, and running them concurrently under CI contention is a
///     flakiness risk. The rest of the assembly parallelizes normally.
/// </summary>
[CollectionDefinition("LocalDbPairConvergence")]
public sealed class LocalDbPairConvergenceCollection;

/// <summary>
///     Two driver-role OtOpcUa nodes replicating the deployment-artifact cache over a REAL loopback
///     gRPC transport (Kestrel h2c on 127.0.0.1), through the REAL fail-closed
///     <see cref="LocalDbSyncAuthInterceptor"/>. Node A is the initiator (it dials the peer); node B
///     is passive (it hosts <c>MapZbLocalDbSync</c>).
/// </summary>
/// <remarks>
///     <para>
///         Both databases are initialised through the production <see cref="LocalDbSetup.OnReady"/> —
///         a hand-written schema here would prove only that the test agrees with itself. The tables,
///         their primary keys, and the DDL→<c>RegisterReplicated</c> ordering under test are the ones
///         the host actually runs.
///     </para>
///     <para>
///         The two <see cref="ILocalDb"/> instances are owned by the harness and registered into the
///         hosts as pre-constructed singletons. MS.DI does not dispose instances it did not create,
///         so tearing a host down (the transport-loss scenario) leaves the databases intact and
///         writable — which is exactly what lets node A accumulate writes while B is down.
///     </para>
///     <para>
///         A real loopback socket (not an in-memory TestServer) is deliberate: killing the passive
///         host disposes the server and closes the socket, faulting the initiator's active stream
///         promptly. An in-memory TestServer does not model a connection drop and leaves the client
///         hanging.
///     </para>
///     <para>
///         The API keys are per-node so the wrong-key scenario can hand A and B different keys and
///         assert non-convergence — with a matching-key positive control proving the same harness
///         does converge when the keys agree (an absence assertion without a positive control passed
///         vacuously in ScadaBridge).
///     </para>
///     <para>Offline: no docker, no external services.</para>
/// </remarks>
public sealed class LocalDbPairHarness : IAsyncDisposable
{
    /// <summary>The key both nodes share unless a test overrides one of them.</summary>
    public const string DefaultApiKey = "otopcua-localdb-pair-convergence-key";

    /// <summary>How long a scenario waits for the pair to agree (or to stay diverged) before deciding.</summary>
    public static readonly TimeSpan ConvergeTimeout = TimeSpan.FromSeconds(30);

    private readonly string _apiKeyA;
    private readonly string _apiKeyB;

    private readonly string _pathA = Path.Combine(Path.GetTempPath(), $"otopcua-pairA-{Guid.NewGuid():N}.db");

    // Not readonly: WipePassiveAsync replaces node B's database wholesale, which is what a rebuilt
    // node with a lost volume actually is — a new file, and with it a new node id.
    private string _pathB = Path.Combine(Path.GetTempPath(), $"otopcua-pairB-{Guid.NewGuid():N}.db");

    private readonly ServiceProvider _dbProviderA;
    private ServiceProvider _dbProviderB;

    private IHost? _serverHost;      // node B — passive
    private IHost? _initiatorHost;   // node A — dials the peer

    static LocalDbPairHarness() =>
        // Grpc.Net.Client dials the loopback server over HTTP/2 cleartext (h2c).
        AppContext.SetSwitch("System.Net.Http.SocketsHttpHandler.Http2UnencryptedSupport", true);

    /// <param name="apiKeyA">Node A's replication key. Defaults to <see cref="DefaultApiKey"/>.</param>
    /// <param name="apiKeyB">Node B's replication key. Defaults to <see cref="DefaultApiKey"/>.</param>
    public LocalDbPairHarness(string? apiKeyA = null, string? apiKeyB = null)
    {
        _apiKeyA = apiKeyA ?? DefaultApiKey;
        _apiKeyB = apiKeyB ?? DefaultApiKey;

        _dbProviderA = BuildDatabaseProvider(_pathA);
        _dbProviderB = BuildDatabaseProvider(_pathB);
    }

    /// <summary>Node A — the initiator, which dials the peer.</summary>
    public ILocalDb A => _dbProviderA.GetRequiredService<ILocalDb>();

    /// <summary>Node B — the passive node, which listens.</summary>
    public ILocalDb B => _dbProviderB.GetRequiredService<ILocalDb>();

    /// <summary>The production artifact cache over node A's database.</summary>
    public IDeploymentArtifactCache CacheA =>
        new LocalDbDeploymentArtifactCache(A, NullLogger<LocalDbDeploymentArtifactCache>.Instance);

    /// <summary>The production artifact cache over node B's database.</summary>
    public IDeploymentArtifactCache CacheB =>
        new LocalDbDeploymentArtifactCache(B, NullLogger<LocalDbDeploymentArtifactCache>.Instance);

    // ---- lifecycle --------------------------------------------------------------------------

    /// <summary>Forces both databases to construct (running OnReady) then brings the pair online.</summary>
    public async Task StartAsync()
    {
        _ = A;
        _ = B;

        await StartPassiveAsync();
        await StartInitiatorAsync();
    }

    /// <summary>Takes node B's listener down, leaving its database intact and writable.</summary>
    public async Task StopPassiveAsync()
    {
        await StopHostAsync(_serverHost);
        _serverHost = null;
    }

    /// <summary>
    ///     Destroys node B's database and replaces it with an empty one, leaving the pairing config
    ///     untouched — the rebuilt-node case: a lost volume, a re-imaged host, a fresh container.
    /// </summary>
    /// <remarks>
    ///     A new file, not a truncated one, because that is what the real thing is: the node id lives
    ///     in the database, so a rebuilt node comes back with a new identity and a zero watermark. The
    ///     caller follows with <see cref="RestartPairAsync"/> to bring the wiped node back online.
    /// </remarks>
    public async Task WipePassiveAsync()
    {
        await StopPassiveAsync();
        await _dbProviderB.DisposeAsync();

        // Pooled connections outlive the provider; without this the delete silently no-ops on
        // Windows and the "wiped" node would still hold its rows.
        Microsoft.Data.Sqlite.SqliteConnection.ClearAllPools();
        DeleteDatabaseFiles(_pathB);

        _pathB = Path.Combine(Path.GetTempPath(), $"otopcua-pairB-{Guid.NewGuid():N}.db");
        _dbProviderB = BuildDatabaseProvider(_pathB);
        _ = B;
    }

    /// <summary>
    ///     Brings node B back on a NEW loopback port and re-dials from A. The initiator re-reads the
    ///     peer address on each reconnect, so this is a genuine rejoin over the same databases.
    /// </summary>
    public async Task RestartPairAsync()
    {
        await StartPassiveAsync();
        await StopHostAsync(_initiatorHost);
        _initiatorHost = null;
        await StartInitiatorAsync();
    }

    // ---- convergence helpers ----------------------------------------------------------------

    /// <summary>Polls <paramref name="condition"/> until true or the deadline passes; fails otherwise.</summary>
    public static async Task WaitUntilAsync(Func<Task<bool>> condition, string because)
    {
        var deadline = DateTime.UtcNow + ConvergeTimeout;
        while (DateTime.UtcNow < deadline)
        {
            if (await condition())
                return;
            await Task.Delay(50);
        }

        Assert.Fail($"Timed out after {ConvergeTimeout.TotalSeconds:0}s waiting for: {because}");
    }

    /// <summary>Waits out a bounded window and returns whether <paramref name="condition"/> ever held.</summary>
    /// <remarks>
    ///     For the negative half of the wrong-key scenario: it must NOT converge. A short, fixed
    ///     window keeps the test quick while still giving a matching-key control ample time to
    ///     converge (the control uses <see cref="WaitUntilAsync"/> with the full timeout).
    /// </remarks>
    public static async Task<bool> HeldWithinAsync(Func<Task<bool>> condition, TimeSpan window)
    {
        var deadline = DateTime.UtcNow + window;
        while (DateTime.UtcNow < deadline)
        {
            if (await condition())
                return true;
            await Task.Delay(50);
        }

        return false;
    }

    /// <summary>
    ///     Dumps the <c>__localdb_row_version</c> rows for one table, ordered by pk, as
    ///     <c>pk_json|hlc|node_id|is_tombstone</c>. Equal dumps on both nodes prove B holds A's
    ///     origin-stamped row (same HLC + node id), not a locally re-derived one.
    /// </summary>
    public static async Task<string> DumpRowVersionAsync(ILocalDb db, string table)
    {
        var rows = await db.QueryAsync(
            """
            SELECT pk_json, hlc, node_id, is_tombstone
            FROM __localdb_row_version
            WHERE table_name = @Table
            ORDER BY pk_json
            """,
            static r => $"{r.GetString(0)}|{r.GetInt64(1)}|{r.GetString(2)}|{r.GetInt64(3)}",
            new { Table = table });
        return string.Join("\n", rows);
    }

    /// <summary>Counts <c>deployment_artifacts</c> chunk rows for one deployment on a node.</summary>
    public static async Task<long> ChunkCountAsync(ILocalDb db, string deploymentId)
    {
        var rows = await db.QueryAsync(
            "SELECT COUNT(*) FROM deployment_artifacts WHERE deployment_id = @DeploymentId",
            static r => r.GetInt64(0),
            new { DeploymentId = deploymentId });
        return rows[0];
    }

    /// <summary>
    ///     Counts oplog rows on a node. Zero means every local change has been acked by the peer and
    ///     pruned — the steady state of a converged pair, and the state from which a wiped peer can
    ///     only be healed by a snapshot.
    /// </summary>
    public static async Task<long> OplogDepthAsync(ILocalDb db)
    {
        var rows = await db.QueryAsync("SELECT COUNT(*) FROM __localdb_oplog", static r => r.GetInt64(0));
        return rows[0];
    }

    /// <summary>Counts tombstone rows in <c>__localdb_row_version</c> for one table on a node.</summary>
    public static async Task<long> TombstoneCountAsync(ILocalDb db, string table)
    {
        var rows = await db.QueryAsync(
            """
            SELECT COUNT(*) FROM __localdb_row_version
            WHERE table_name = @Table AND is_tombstone = 1
            """,
            static r => r.GetInt64(0),
            new { Table = table });
        return rows[0];
    }

    // ---- fixture internals ------------------------------------------------------------------

    /// <summary>
    ///     A provider owning one deployment-cache database, initialised through the host's own
    ///     <see cref="LocalDbSetup.OnReady"/> — same schema, same registration order.
    /// </summary>
    private static ServiceProvider BuildDatabaseProvider(string path)
    {
        var config = new ConfigurationBuilder()
            .AddInMemoryCollection(new Dictionary<string, string?> { ["LocalDb:Path"] = path })
            .Build();

        return new ServiceCollection()
            .AddZbLocalDb(config, db => LocalDbSetup.OnReady(db, config))
            .BuildServiceProvider();
    }

    private IConfiguration ReplicationConfig(string apiKey, string? peerAddress)
    {
        var values = new Dictionary<string, string?>
        {
            // Tight flush + bounded reconnect backoff so convergence is observable well inside the
            // poll deadline. The 60 s production default would let the doubling backoff overrun it
            // after a peer outage.
            ["LocalDb:Replication:FlushInterval"] = "00:00:00.050",
            ["LocalDb:Replication:ReconnectBackoffMax"] = "00:00:02",
            ["LocalDb:Replication:ApiKey"] = apiKey,
        };

        if (peerAddress is not null)
            values["LocalDb:Replication:PeerAddress"] = peerAddress;

        return new ConfigurationBuilder().AddInMemoryCollection(values).Build();
    }

    /// <summary>Starts node B, the passive listener, behind the real auth interceptor.</summary>
    private async Task StartPassiveAsync()
    {
        var config = ReplicationConfig(_apiKeyB, peerAddress: null);

        _serverHost = await new HostBuilder()
            .ConfigureWebHost(web =>
            {
                web.UseKestrel(o =>
                    o.Listen(IPAddress.Loopback, 0, listen => listen.Protocols = HttpProtocols.Http2));
                web.ConfigureServices(services =>
                {
                    services.AddLogging();
                    services.AddRouting();
                    // The REAL interceptor, not a stand-in. If it rejected legitimate peer traffic,
                    // every matching-key scenario would fail — which is the point.
                    services.AddGrpc(o => o.Interceptors.Add<LocalDbSyncAuthInterceptor>());
                    services.AddSingleton(B);
                    services.AddZbLocalDbReplication(config);
                })
                .Configure(app =>
                {
                    app.UseRouting();
                    app.UseEndpoints(e => e.MapZbLocalDbSync());
                });
            })
            .StartAsync();
    }

    /// <summary>Starts node A, which dials the passive node.</summary>
    private async Task StartInitiatorAsync()
    {
        var config = ReplicationConfig(_apiKeyA, PassiveAddress());

        _initiatorHost = await new HostBuilder()
            .ConfigureServices(services =>
            {
                services.AddLogging();
                services.AddSingleton(A);
                services.AddZbLocalDbReplication(config);
            })
            .StartAsync();
    }

    private string PassiveAddress()
        => _serverHost!.Services.GetRequiredService<IServer>()
            .Features.Get<IServerAddressesFeature>()!.Addresses.Single();

    private static async Task StopHostAsync(IHost? host)
    {
        if (host is null)
            return;
        try { await host.StopAsync(TimeSpan.FromSeconds(5)); } catch { /* teardown */ }
        host.Dispose();
    }

    public async ValueTask DisposeAsync()
    {
        await StopHostAsync(_initiatorHost);
        await StopHostAsync(_serverHost);
        await _dbProviderA.DisposeAsync();
        await _dbProviderB.DisposeAsync();

        Microsoft.Data.Sqlite.SqliteConnection.ClearAllPools();
        DeleteDatabaseFiles(_pathA);
        DeleteDatabaseFiles(_pathB);
    }

    /// <summary>Deletes a SQLite database and its WAL/shm sidecars, best effort.</summary>
    private static void DeleteDatabaseFiles(string path)
    {
        foreach (var suffix in new[] { "", "-wal", "-shm" })
        {
            try { File.Delete(path + suffix); } catch { /* best effort */ }
        }
    }
}