scadaproj/ZB.MOM.WW.LocalDb/tests/ZB.MOM.WW.LocalDb.Tests/GrpcAdapterTests.cs

using Grpc.Core;
using Grpc.Core.Interceptors;
using Grpc.Net.Client;
using Microsoft.AspNetCore.Builder;
using Microsoft.AspNetCore.Hosting;
using Microsoft.AspNetCore.TestHost;
using Microsoft.Extensions.Configuration;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Hosting;
using ZB.MOM.WW.LocalDb.Contracts;
using ZB.MOM.WW.LocalDb.Replication;

namespace ZB.MOM.WW.LocalDb.Tests;

public sealed class GrpcAdapterTests : IAsyncLifetime
{
    private static readonly TimeSpan RunTimeout = TimeSpan.FromSeconds(30);

    private readonly List<string> _paths = [];
    private readonly List<IHost> _hosts = [];
    private readonly List<ServiceProvider> _providers = [];

    public Task InitializeAsync() => Task.CompletedTask;

    public async Task DisposeAsync()
    {
        foreach (var provider in _providers)
            await provider.DisposeAsync();
        foreach (var host in _hosts)
        {
            try { await host.StopAsync(TimeSpan.FromSeconds(5)); } catch { /* teardown */ }
            host.Dispose();
        }
        Microsoft.Data.Sqlite.SqliteConnection.ClearAllPools();
        foreach (var p in _paths)
        {
            if (File.Exists(p)) File.Delete(p);
            if (File.Exists(p + "-wal")) File.Delete(p + "-wal");
            if (File.Exists(p + "-shm")) File.Delete(p + "-shm");
        }
    }

    // ---- end-to-end wire tests -------------------------------------------------------------

    [Fact]
    public async Task EndToEnd_TwoDbs_InsertOnA_AppearsOnB()
    {
        var serverHost = await BuildServerAsync(NewDbPath(), apiKey: null, sink: null);
        var serverDb = serverHost.Services.GetRequiredService<ILocalDb>();
        await serverDb.ExecuteAsync("INSERT INTO orders (id, sku, qty) VALUES (2, 'FROM_SERVER', 20)");
        var handler = serverHost.GetTestServer().CreateHandler();

        var (clientProvider, bg) = BuildClient(NewDbPath(), "http://localhost", apiKey: null, () => ChannelOver(handler));
        var clientDb = clientProvider.GetRequiredService<ILocalDb>();
        await clientDb.ExecuteAsync("INSERT INTO orders (id, sku, qty) VALUES (1, 'FROM_CLIENT', 10)");

        using var cts = new CancellationTokenSource(RunTimeout);
        await bg.StartAsync(cts.Token);

        await WaitForAsync(
            async () => (await ReadOrders(clientDb)).Count == 2 && (await ReadOrders(serverDb)).Count == 2,
            RunTimeout);

        var clientRows = await ReadOrders(clientDb);
        var serverRows = await ReadOrders(serverDb);
        Assert.Equal(clientRows, serverRows);
        Assert.Equal([(1L, "FROM_CLIENT", (long?)10), (2L, "FROM_SERVER", (long?)20)], clientRows);

        await bg.StopAsync(CancellationToken.None);
    }

    [Fact]
    public async Task SecondConcurrentStream_Rejected()
    {
        var serverHost = await BuildServerAsync(NewDbPath(), apiKey: null, sink: null);
        var serverDb = serverHost.Services.GetRequiredService<ILocalDb>();
        var handler = serverHost.GetTestServer().CreateHandler();

        var (clientProvider, bg) = BuildClient(NewDbPath(), "http://localhost", apiKey: null, () => ChannelOver(handler));
        var clientDb = clientProvider.GetRequiredService<ILocalDb>();
        await clientDb.ExecuteAsync("INSERT INTO orders (id, sku, qty) VALUES (1, 'C', 10)");

        using var cts = new CancellationTokenSource(RunTimeout);
        await bg.StartAsync(cts.Token);

        // Convergence proves the first stream's server handler is active and holding the single-stream guard.
        await WaitForAsync(async () => (await ReadOrders(serverDb)).Count == 1, RunTimeout);

        var client = new LocalDbSync.LocalDbSyncClient(ChannelOver(handler));
        using var call2 = client.Sync(cancellationToken: cts.Token);
        var ex = await Assert.ThrowsAsync<RpcException>(() => call2.ResponseStream.MoveNext(cts.Token));
        Assert.Equal(StatusCode.AlreadyExists, ex.StatusCode);

        await bg.StopAsync(CancellationToken.None);
    }

    // Shipped variant: the channel factory throws on its first call, then returns a live channel.
    // This deterministically drives the backoff/retry loop (no flaky server restart in TestHost)
    // and proves convergence after recovery.
    [Fact]
    public async Task ClientReconnects_AfterTransientFault()
    {
        var serverHost = await BuildServerAsync(NewDbPath(), apiKey: null, sink: null);
        var serverDb = serverHost.Services.GetRequiredService<ILocalDb>();
        await serverDb.ExecuteAsync("INSERT INTO orders (id, sku, qty) VALUES (2, 'S', 20)");
        var handler = serverHost.GetTestServer().CreateHandler();

        var calls = 0;
        GrpcChannel Factory()
        {
            if (Interlocked.Increment(ref calls) == 1)
                throw new InvalidOperationException("transient dial failure");
            return ChannelOver(handler);
        }

        var (clientProvider, bg) = BuildClient(NewDbPath(), "http://localhost", apiKey: null, Factory);
        var clientDb = clientProvider.GetRequiredService<ILocalDb>();
        await clientDb.ExecuteAsync("INSERT INTO orders (id, sku, qty) VALUES (1, 'C', 10)");

        using var cts = new CancellationTokenSource(RunTimeout);
        await bg.StartAsync(cts.Token);

        await WaitForAsync(
            async () => (await ReadOrders(clientDb)).Count == 2 && (await ReadOrders(serverDb)).Count == 2,
            RunTimeout);
        Assert.True(bg.ConnectionAttempts >= 2, $"expected >= 2 connection attempts, got {bg.ConnectionAttempts}");

        await bg.StopAsync(CancellationToken.None);
    }

    [Fact]
    public async Task ApiKeyHeader_SentWhenConfigured()
    {
        var sink = new AuthCaptureSink();
        var serverHost = await BuildServerAsync(NewDbPath(), apiKey: "test-key", sink: sink);
        var handler = serverHost.GetTestServer().CreateHandler();

        var (clientProvider, bg) = BuildClient(NewDbPath(), "http://localhost", apiKey: "test-key", () => ChannelOver(handler));
        var clientDb = clientProvider.GetRequiredService<ILocalDb>();
        await clientDb.ExecuteAsync("INSERT INTO orders (id, sku, qty) VALUES (1, 'C', 10)");

        using var cts = new CancellationTokenSource(RunTimeout);
        await bg.StartAsync(cts.Token);

        await WaitForAsync(() => Task.FromResult(sink.Authorization is not null), RunTimeout);
        Assert.Equal("Bearer test-key", sink.Authorization);

        await bg.StopAsync(CancellationToken.None);
    }

    // Shipped variant: in-proc against GrpcSyncDuplex with a flow-control-stalled fake writer
    // (WriteAsync never completes until its token fires) — pins requirement 2 (Send honors its
    // CancellationToken even when transport-blocked) deterministically; a true stalled-gRPC-peer
    // rig in TestHost cannot force flow-control blockage reliably.
    [Fact]
    public async Task Send_StalledOnFlowControl_CancelUnblocksPromptly()
    {
        using var cts = new CancellationTokenSource();
        var (duplex, readerTask) = Replication.Internal.GrpcSyncDuplex.Create(
            new HangingReader(), new HangingWriter(), cts.Token);

        var send = duplex.Send(new SyncMessage(), cts.Token);
        Assert.False(send.IsCompleted);

        var sw = System.Diagnostics.Stopwatch.StartNew();
        cts.Cancel();
        await Assert.ThrowsAnyAsync<OperationCanceledException>(() => send);
        // The reader pump must also tear down: its fault travels into the inbox completion.
        await readerTask.WaitAsync(TimeSpan.FromSeconds(5));
        await Assert.ThrowsAnyAsync<OperationCanceledException>(
            async () => await duplex.Inbox.Completion);
        sw.Stop();
        Assert.True(sw.Elapsed < TimeSpan.FromSeconds(3), $"teardown took {sw.Elapsed}");
    }

    // ---- inbox backpressure ------------------------------------------------------------------

    [Fact]
    public async Task Inbox_SlowReader_StopsDrainingTheStream_AndLosesNothing()
    {
        // The pump must STOP pulling from the gRPC stream once the inbox is full — that stall is
        // what closes the peer's flow-control window and bounds this process's memory at
        // InboxCapacity x MaxBatchBytes instead of "whatever the peer decides to send" (a full
        // snapshot of a large dataset, on both nodes at once, when a pair baselines late).
        const int capacity = Replication.Internal.GrpcSyncDuplex.InboxCapacity;
        using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(15));
        var reader = new CountingReader();
        var (duplex, readerTask) = Replication.Internal.GrpcSyncDuplex.Create(
            reader, new HangingWriter(), cts.Token);

        // Nothing reads the inbox: the pump fills it and blocks.
        await WaitForAsync(() => Task.FromResult(reader.Produced >= capacity), TimeSpan.FromSeconds(5));
        await Task.Delay(200, cts.Token);
        // capacity in the channel + at most one in the pump's hand, blocked on WriteAsync.
        Assert.True(reader.Produced <= capacity + 1,
            $"the reader pump kept draining an already-full inbox: {reader.Produced} messages read for a {capacity}-message bound");

        // Draining releases it, and every message arrives exactly once in stream order — TryWrite
        // on a full bounded channel would have dropped them silently instead.
        for (long expected = 1; expected <= capacity + 1; expected++)
        {
            var msg = await ReadInboxAsync(duplex, cts.Token);
            Assert.Equal(expected, msg.DeltaAck.AppliedThruSeq);
        }
        await WaitForAsync(() => Task.FromResult(reader.Produced > capacity + 1), TimeSpan.FromSeconds(5));

        cts.Cancel();
        try { await readerTask.WaitAsync(TimeSpan.FromSeconds(5)); }
        catch (OperationCanceledException) { }
    }

    private static async Task<SyncMessage> ReadInboxAsync(
        Replication.Internal.SyncDuplex duplex, CancellationToken ct)
    {
        await duplex.Inbox.WaitToReadAsync(ct);
        duplex.Inbox.TryRead(out var message);
        return message!;
    }

    // Produces messages as fast as the inbox will take them, numbering each one so the consumer can
    // prove nothing was dropped or reordered.
    private sealed class CountingReader : IAsyncStreamReader<SyncMessage>
    {
        private long _produced;

        public long Produced => Interlocked.Read(ref _produced);

        public SyncMessage Current { get; private set; } = new();

        public Task<bool> MoveNext(CancellationToken cancellationToken)
        {
            cancellationToken.ThrowIfCancellationRequested();
            Current = new SyncMessage
            {
                DeltaAck = new DeltaAck { AppliedThruSeq = Interlocked.Increment(ref _produced) },
            };
            return Task.FromResult(true);
        }
    }

    // ---- validator -------------------------------------------------------------------------

    [Fact]
    public void Validator_InvalidPeerAddress_Fails()
    {
        var result = new ReplicationOptionsValidator().Validate(
            null, new ReplicationOptions { PeerAddress = "not-a-uri" });

        Assert.True(result.Failed);
        Assert.Contains(result.Failures, f => f.Contains("PeerAddress", StringComparison.Ordinal));
    }

    [Fact]
    public void Validator_NonPositiveIntervals_Fail()
    {
        var result = new ReplicationOptionsValidator().Validate(null, new ReplicationOptions
        {
            FlushInterval = TimeSpan.Zero,
            MaxBatchSize = 0,
            MaxOplogRows = 0,
            MaxOplogAge = TimeSpan.Zero,
            TombstoneRetention = TimeSpan.Zero,
            ReconnectBackoffMax = TimeSpan.Zero,
            MaxHlcDriftAhead = TimeSpan.Zero,
        });

        Assert.True(result.Failed);
        Assert.Contains(result.Failures, f => f.Contains("FlushInterval", StringComparison.Ordinal));
        Assert.Contains(result.Failures, f => f.Contains("MaxBatchSize", StringComparison.Ordinal));
        Assert.Contains(result.Failures, f => f.Contains("MaxOplogRows", StringComparison.Ordinal));
        Assert.Contains(result.Failures, f => f.Contains("MaxOplogAge", StringComparison.Ordinal));
        Assert.Contains(result.Failures, f => f.Contains("TombstoneRetention", StringComparison.Ordinal));
        Assert.Contains(result.Failures, f => f.Contains("ReconnectBackoffMax", StringComparison.Ordinal));
        Assert.Contains(result.Failures, f => f.Contains("MaxHlcDriftAhead", StringComparison.Ordinal));
    }

    [Fact]
    public void Validator_EmptyPeerAddress_Allowed()
    {
        // A passive node leaves PeerAddress empty; the validator can't know the role, so it only
        // checks format when present. All other defaults are valid.
        var result = new ReplicationOptionsValidator().Validate(null, new ReplicationOptions { PeerAddress = "" });

        Assert.True(result.Succeeded);
    }

    [Fact]
    public void Validator_ReconnectBackoffMax_CappedAtOneDay()
    {
        var atCeiling = new ReplicationOptionsValidator().Validate(
            null, new ReplicationOptions { ReconnectBackoffMax = TimeSpan.FromDays(1) });
        Assert.True(atCeiling.Succeeded);

        var overCeiling = new ReplicationOptionsValidator().Validate(
            null, new ReplicationOptions { ReconnectBackoffMax = TimeSpan.FromDays(1) + TimeSpan.FromTicks(1) });
        Assert.True(overCeiling.Failed);
        Assert.Contains(overCeiling.Failures, f => f.Contains("ReconnectBackoffMax", StringComparison.Ordinal));
    }

    // ---- DI --------------------------------------------------------------------------------

    [Fact]
    public async Task AddZbLocalDbReplication_PassiveConfig_InitiatorIdles()
    {
        var passive = new ServiceCollection();
        passive.AddLogging();
        var passiveCfg = ConfigFor(NewDbPath(), peerAddress: null, apiKey: null);
        passive.AddZbLocalDb(passiveCfg, OnReady);
        passive.AddZbLocalDbReplication(passiveCfg);
        var p1 = passive.BuildServiceProvider();
        _providers.Add(p1);

        Assert.NotNull(p1.GetService<LocalDbSyncService>());

        // The hosted service is always registered; with no PeerAddress it starts, idles, and
        // completes without dialing anything.
        var bg = p1.GetServices<IHostedService>().OfType<SyncBackgroundService>().Single();
        await bg.StartAsync(CancellationToken.None);
        await bg.ExecuteTask!.WaitAsync(TimeSpan.FromSeconds(5));
        Assert.True(bg.ExecuteTask.IsCompletedSuccessfully);
        Assert.Equal(0, bg.ConnectionAttempts);
        await bg.StopAsync(CancellationToken.None);

        var initiator = new ServiceCollection();
        initiator.AddLogging();
        var initiatorCfg = ConfigFor(NewDbPath(), peerAddress: "http://localhost", apiKey: null);
        initiator.AddZbLocalDb(initiatorCfg, OnReady);
        initiator.AddZbLocalDbReplication(initiatorCfg);
        var p2 = initiator.BuildServiceProvider();
        _providers.Add(p2);

        Assert.Contains(p2.GetServices<IHostedService>(), s => s is SyncBackgroundService);
    }

    // ---- harness ---------------------------------------------------------------------------

    private string NewDbPath()
    {
        var path = Path.Combine(Path.GetTempPath(), Guid.NewGuid() + ".db");
        _paths.Add(path);
        return path;
    }

    private static void OnReady(ILocalDb db)
    {
        using var conn = db.CreateConnection();
        using var cmd = conn.CreateCommand();
        cmd.CommandText = "CREATE TABLE orders (id INTEGER PRIMARY KEY, sku TEXT, qty INTEGER)";
        cmd.ExecuteNonQuery();
        db.RegisterReplicated("orders");
    }

    private static IConfiguration ConfigFor(string path, string? peerAddress, string? apiKey)
    {
        var dict = new Dictionary<string, string?>
        {
            ["LocalDb:Path"] = path,
            ["LocalDb:Replication:FlushInterval"] = "00:00:00.020",
        };
        if (peerAddress is not null) dict["LocalDb:Replication:PeerAddress"] = peerAddress;
        if (apiKey is not null) dict["LocalDb:Replication:ApiKey"] = apiKey;
        return new ConfigurationBuilder().AddInMemoryCollection(dict).Build();
    }

    private static GrpcChannel ChannelOver(HttpMessageHandler handler) =>
        GrpcChannel.ForAddress("http://localhost", new GrpcChannelOptions { HttpHandler = handler });

    private async Task<IHost> BuildServerAsync(string path, string? apiKey, AuthCaptureSink? sink)
    {
        var config = ConfigFor(path, peerAddress: null, apiKey: apiKey);
        var host = await new HostBuilder()
            .ConfigureWebHost(web =>
            {
                web.UseTestServer();
                web.ConfigureServices(services =>
                {
                    services.AddRouting();
                    if (sink is not null)
                    {
                        services.AddSingleton(sink);
                        services.AddGrpc(o => o.Interceptors.Add<AuthCaptureInterceptor>());
                    }
                    else
                    {
                        services.AddGrpc();
                    }

                    services.AddZbLocalDb(config, OnReady);
                    services.AddZbLocalDbReplication(config);
                });
                web.Configure(app =>
                {
                    app.UseRouting();
                    app.UseEndpoints(e => e.MapZbLocalDbSync());
                });
            })
            .StartAsync();
        _hosts.Add(host);
        return host;
    }

    private (ServiceProvider Provider, SyncBackgroundService Bg) BuildClient(
        string path, string peerAddress, string? apiKey, Func<GrpcChannel> channelFactory)
    {
        var config = ConfigFor(path, peerAddress, apiKey);
        var services = new ServiceCollection();
        services.AddLogging();
        services.AddZbLocalDb(config, OnReady);
        services.AddZbLocalDbReplication(config);
        var provider = services.BuildServiceProvider();
        _providers.Add(provider);

        var bg = provider.GetServices<IHostedService>().OfType<SyncBackgroundService>().Single();
        bg.ChannelFactory = channelFactory;
        return (provider, bg);
    }

    private static async Task<List<(long Id, string? Sku, long? Qty)>> ReadOrders(ILocalDb db)
    {
        var rows = await db.QueryAsync(
            "SELECT id, sku, qty FROM orders ORDER BY id",
            x => (x.GetInt64(0), x.IsDBNull(1) ? null : x.GetString(1), (long?)(x.IsDBNull(2) ? null : x.GetInt64(2))));
        return rows.ToList();
    }

    private static async Task WaitForAsync(Func<Task<bool>> predicate, TimeSpan timeout)
    {
        var deadline = DateTime.UtcNow + timeout;
        while (DateTime.UtcNow < deadline)
        {
            if (await predicate()) return;
            await Task.Delay(20);
        }
        throw new TimeoutException("Condition not reached within " + timeout);
    }

    private sealed class HangingReader : IAsyncStreamReader<SyncMessage>
    {
        public SyncMessage Current => throw new InvalidOperationException("no message");

        public async Task<bool> MoveNext(CancellationToken cancellationToken)
        {
            await Task.Delay(Timeout.Infinite, cancellationToken);
            return false;
        }
    }

    private sealed class HangingWriter : IAsyncStreamWriter<SyncMessage>
    {
        public WriteOptions? WriteOptions { get; set; }

        public Task WriteAsync(SyncMessage message) => Task.Delay(Timeout.Infinite);

        public Task WriteAsync(SyncMessage message, CancellationToken cancellationToken) =>
            Task.Delay(Timeout.Infinite, cancellationToken);
    }

    private sealed class AuthCaptureSink
    {
        public volatile string? Authorization;
    }

    private sealed class AuthCaptureInterceptor(AuthCaptureSink sink) : Interceptor
    {
        public override Task DuplexStreamingServerHandler<TRequest, TResponse>(
            IAsyncStreamReader<TRequest> requestStream,
            IServerStreamWriter<TResponse> responseStream,
            ServerCallContext context,
            DuplexStreamingServerMethod<TRequest, TResponse> continuation)
        {
            var auth = context.RequestHeaders.GetValue("authorization");
            if (auth is not null)
                sink.Authorization = auth;
            return continuation(requestStream, responseStream, context);
        }
    }
}