using Grpc.Core;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Options;
using ZB.MOM.WW.LocalDb.Contracts;
using ZB.MOM.WW.LocalDb.Internal;
using ZB.MOM.WW.LocalDb.Replication.Internal;

namespace ZB.MOM.WW.LocalDb.Replication;

/// <summary>
/// The passive (server) side of replication: a gRPC service that accepts an inbound bidirectional
/// <c>Sync</c> stream and runs the symmetric <see cref="SyncSession"/> over it. Registered as a
/// singleton so its single-stream guard spans concurrent calls (a node syncs with exactly one peer).
/// </summary>
public sealed class LocalDbSyncService : LocalDbSync.LocalDbSyncBase
{
    private readonly SqliteLocalDb _db;
    private readonly ReplicationOptions _options;
    private readonly ILoggerFactory _loggerFactory;
    private readonly ILogger<LocalDbSyncService> _logger;
    private readonly LocalDbMetrics? _metrics;
    private readonly SyncStatus? _status;

    // The stream currently holding the guard, or null when idle. Guarded by _gate (singleton service).
    private readonly object _gate = new();
    private Registration? _active;

    /// <summary>
    /// How long a displacing call waits for the incumbent it evicted to actually finish before
    /// giving up and refusing itself. Test seam; the default is generous because eviction only has
    /// to outrun the operator, not the network.
    /// </summary>
    internal TimeSpan TakeoverTimeout { get; set; } = TimeSpan.FromSeconds(10);

    /// <summary>One in-flight <c>Sync</c> handler: its cancellation handle plus a completion signal
    /// a displacing handler can await, so two sessions never run against the database at once.</summary>
    private sealed class Registration(CancellationTokenSource cts)
    {
        public CancellationTokenSource Cts { get; } = cts;

        public TaskCompletionSource Finished { get; } =
            new(TaskCreationOptions.RunContinuationsAsynchronously);
    }

    /// <summary>
    /// Creates the passive sync service. <paramref name="status"/> is taken as the public
    /// <see cref="ISyncStatus"/> (SyncStatus is internal and cannot appear in a public signature);
    /// DI always supplies the concrete SyncStatus the ctor recovers by cast.
    /// </summary>
    public LocalDbSyncService(
        ILocalDb db, IOptions<ReplicationOptions> options, ILoggerFactory loggerFactory,
        LocalDbMetrics? metrics = null, ISyncStatus? status = null)
    {
        _db = SyncSessionFactory.RequireSqlite(db, nameof(LocalDbSyncService));
        _options = options.Value;
        _loggerFactory = loggerFactory;
        _logger = loggerFactory.CreateLogger<LocalDbSyncService>();
        _metrics = metrics;
        _status = status as SyncStatus;
    }

    /// <summary>
    /// Accepts one inbound sync stream, evicting any stream already holding the guard.
    /// </summary>
    /// <remarks>
    /// <para>
    /// <b>Newest wins.</b> The guard exists to SERIALIZE sessions — a node syncs with exactly one
    /// peer, and two sessions against one database would fight over the same watermarks — not to
    /// fence out an impostor; authentication is the interceptor's job, and an authenticated caller
    /// on this key IS the peer by definition. Refusing the newcomer therefore protected nothing and
    /// cost everything in the case that actually happens: the dialling node crashes or is
    /// partitioned, this handler stays parked forever inside <c>MoveNext</c> on a half-open stream
    /// that will never produce another byte, and every reconnect from the rebooted peer is rejected
    /// <c>AlreadyExists</c> — potentially for as long as the pair stays idle, while this side keeps
    /// reporting itself connected. The newest dialler is the one with a live link, so it takes over.
    /// </para>
    /// <para>
    /// Eviction is cancel-then-await, not cancel-and-go: the incumbent's cancellation unwinds its
    /// session and reader pump, and the newcomer waits for that to complete before starting its own,
    /// so the "one session at a time" property the guard was built for still holds. An incumbent
    /// that cannot be evicted within <see cref="TakeoverTimeout"/> is the one case where the old
    /// answer is still right — the newcomer refuses itself rather than double-running.
    /// </para>
    /// </remarks>
    public override async Task Sync(
        IAsyncStreamReader<SyncMessage> requestStream,
        IServerStreamWriter<SyncMessage> responseStream,
        ServerCallContext context)
    {
        // The linked source exists so the reader pump is deterministically torn down on EVERY exit
        // path — including a session fault for a local (non-transport) reason, where the call token
        // never fires and the pump would otherwise keep reading a stream whose handler has exited.
        // It is also the handle a later call cancels to displace this one.
        using var linkedCts = CancellationTokenSource.CreateLinkedTokenSource(context.CancellationToken);
        var mine = new Registration(linkedCts);

        try
        {
            // Inside the try so the finally below always releases the guard — including when the
            // takeover itself is abandoned because THIS caller gave up mid-handover.
            await TakeOverAsync(mine, context.CancellationToken);

            var (duplex, readerTask) = GrpcSyncDuplex.Create(requestStream, responseStream, linkedCts.Token);
            var session = SyncSessionFactory.Create(_db, _options, _loggerFactory, _metrics, _status);
            _status?.SessionStarted();
            try
            {
                await session.RunAsync(duplex, linkedCts.Token);
            }
            finally
            {
                _status?.SessionEnded();
                linkedCts.Cancel();
                try
                {
                    await readerTask;
                }
                catch (Exception ex)
                {
                    _logger.LogDebug(ex, "Sync reader pump ended during teardown.");
                }
            }
        }
        catch (OperationCanceledException) when (context.CancellationToken.IsCancellationRequested)
        {
            // Client disconnect / server shutdown: the normal end of a long-lived streaming RPC.
        }
        catch (OperationCanceledException)
        {
            // Displaced by a newer stream: an orderly handover, not a failure of this node.
            _logger.LogInformation("Sync stream ended: displaced by a newer stream from the peer.");
        }
        catch (LocalDbSchemaMismatchException ex)
        {
            throw new RpcException(new Status(StatusCode.FailedPrecondition, ex.Message));
        }
        finally
        {
            lock (_gate)
            {
                if (ReferenceEquals(_active, mine))
                    _active = null;
            }
            mine.Finished.TrySetResult();
        }
    }

    /// <summary>
    /// Claims the single-stream guard for <paramref name="mine"/>, cancelling and awaiting whatever
    /// held it. Throws <c>AlreadyExists</c> only when the incumbent refuses to end.
    /// </summary>
    private async Task TakeOverAsync(Registration mine, CancellationToken ct)
    {
        Registration? displaced;
        lock (_gate)
        {
            displaced = _active;
            _active = mine;
        }
        if (displaced is null)
            return;

        _logger.LogWarning(
            "A new sync stream arrived while one was already active; displacing the incumbent (newest wins).");
        try
        {
            // The incumbent may already be disposing its own source between its finally and here.
            displaced.Cts.Cancel();
        }
        catch (ObjectDisposedException) { }

        try
        {
            await displaced.Finished.Task.WaitAsync(TakeoverTimeout, ct);
        }
        catch (TimeoutException)
        {
            // Could not evict it: two live sessions over one database is worse than a refused dial,
            // so fall back to the historical answer — and hand the guard back to the incumbent.
            lock (_gate)
            {
                if (ReferenceEquals(_active, mine))
                    _active = displaced;
            }
            throw new RpcException(new Status(
                StatusCode.AlreadyExists,
                "a sync stream is already active and did not yield within the takeover timeout"));
        }
    }
}