using System.Collections.Frozen;
using System.Globalization;
using System.Net.WebSockets;
using System.Security.Claims;
using System.Text.Json;
using DodoSSH.Api.Authorization;
using DodoSSH.Api.Setup;
using DodoSSH.Contracts;
using DodoSSH.Domain.Authorization;
using FastEndpoints;
using Microsoft.Extensions.Options;

namespace DodoSSH.Api.Features.Events;

/// <summary>
/// The socket that says "pull now" so a client does not have to wait for its timer.
/// </summary>
/// <remarks>
/// <para>
/// Everything this endpoint sends is a <em>notice</em>. It never carries an item, a payload or a
/// cursor: the client's answer to a notice is the delta pull it would have run on its own anyway, so
/// there is exactly one code path that applies a change and this is not it. See ADR 0012 for why
/// pushing the items themselves is refused.
/// </para>
/// <para>
/// The bearer token authorises the upgrade, unlike the relay's ticket in ADR 0004. The relay's socket
/// is a byte pipe whose whole authorization decision is made before it opens; this one is a view of
/// the caller's own vault list and has to keep answering "what may this account read" for as long as
/// it is held. Its two bounds on that — the token's own expiry, and a periodic re-resolve — are in
/// <see cref="MindAsync"/>.
/// </para>
/// </remarks>
internal sealed class VaultEventsEndpoint(
    IServiceScopeFactory scopes,
    VaultEventHub hub,
    IOptions<EventsOptions> options,
    TimeProvider clock,
    IHostApplicationLifetime lifetime,
    ILogger<VaultEventsEndpoint> logger)
    : EndpointWithoutRequest
{
    /// <summary>
    /// The largest message this endpoint will read from a client.
    /// </summary>
    /// <remarks>
    /// A client sends nothing but <c>ping</c>, so the cap is three orders of magnitude of headroom and
    /// still small enough that a hostile client cannot make the server buffer anything worth having.
    /// </remarks>
    private const int MaxInboundFrameBytes = 4 * 1024;

    /// <summary>How long to wait for the close handshake before dropping the socket.</summary>
    private static readonly TimeSpan CloseTimeout = TimeSpan.FromSeconds(5);

    /// <inheritdoc />
    public override void Configure()
    {
        Get(VaultEvents.Path);

        // Enrolled, matching sync. A caller with no identity key holds no vault key either, so every
        // notice this socket could send is about ciphertext they cannot read.
        Policies(Auth.EnrolledPolicy);

        Description(b => b
            .WithName("VaultEvents")
            .WithSummary("Pushes a notice when a vault the caller can read has changed.")
            .WithTags("Events"));
    }

    /// <inheritdoc />
    public override async Task HandleAsync(CancellationToken ct)
    {
        if (await RefusedAsync().ConfigureAwait(false))
        {
            return;
        }

        var (userId, vaults) = await ResolveAccessAsync(ct).ConfigureAwait(false);

        // Admitted before the upgrade so a refusal costs nothing, but answered *through* the socket
        // rather than as an HTTP status: a constrained WebSocket client cannot read the status of a
        // failed upgrade, and "you have too many open" is precisely the case where the client needs
        // to know to back off rather than retry. Same reasoning as ADR 0004 on request headers.
        var connection = hub.TryAdmit(userId, vaults);

        try
        {
            using var socket = await HttpContext.WebSockets
                .AcceptWebSocketAsync(new WebSocketAcceptContext { SubProtocol = VaultEvents.SubProtocol })
                .ConfigureAwait(false);

            if (connection is null)
            {
                await CloseAsync(
                        socket,
                        new Closure(
                            VaultEvents.TooManyConnectionsCloseCode, "Too many open event sockets."))
                    .ConfigureAwait(false);

                return;
            }

            await PumpAsync(socket, connection, ct).ConfigureAwait(false);
        }
        finally
        {
            // Inside a try that starts *before* the upgrade, because an accept that throws — a client
            // that abandoned the handshake — would otherwise leave an admitted connection in the hub
            // for the life of the process, counting against this account's cap and taking a slot from
            // the sockets that did open.
            if (connection is not null)
            {
                hub.Remove(connection);
            }
        }
    }

    /// <summary>
    /// Answers the requests that are not an event socket at all, as ordinary HTTP.
    /// </summary>
    /// <returns>Whether a response was sent and the handler should stop.</returns>
    /// <remarks>
    /// All three answers are problem documents rather than bare statuses, because each one has a
    /// different remedy and a client that cannot tell them apart would retry the two that will never
    /// succeed. Answered before the upgrade, so a caller that got the handshake wrong reads why in a
    /// body rather than inferring it from a socket that closed.
    /// </remarks>
    private async Task<bool> RefusedAsync()
    {
        if (!options.Value.Enabled)
        {
            // 404 rather than 501: the feature is absent from this deployment, and /api/v1/meta does
            // not advertise it. A client that dialled anyway keeps polling, which is correct.
            await Send.ResultAsync(Problems.Coded(
                    StatusCodes.Status404NotFound,
                    ProblemCodes.EventsUnavailable,
                    "This server does not push vault changes. Synchronise on a timer instead; "
                    + "GET /api/v1/meta lists the features it does offer."))
                .ConfigureAwait(false);

            return true;
        }

        if (!HttpContext.WebSockets.IsWebSocketRequest)
        {
            await Send.ResultAsync(Problems.Coded(
                    StatusCodes.Status400BadRequest,
                    ProblemCodes.MalformedRequest,
                    "This endpoint is a WebSocket. Send an upgrade request offering the "
                    + $"'{VaultEvents.SubProtocol}' subprotocol."))
                .ConfigureAwait(false);

            return true;
        }

        // The subprotocol is this API's version negotiation for the socket, so an upgrade that does
        // not offer it is refused rather than accepted and answered in a dialect the caller may not
        // read. See VaultEvents.SubProtocol.
        if (!HttpContext.WebSockets.WebSocketRequestedProtocols
                .Contains(VaultEvents.SubProtocol, StringComparer.Ordinal))
        {
            await Send.ResultAsync(Problems.Coded(
                    StatusCodes.Status400BadRequest,
                    ProblemCodes.MalformedRequest,
                    $"This server speaks '{VaultEvents.SubProtocol}', which the upgrade request did "
                    + "not offer."))
                .ConfigureAwait(false);

            return true;
        }

        return false;
    }

    /// <summary>
    /// Reads who the caller is and which vaults they may follow, in a scope of its own.
    /// </summary>
    /// <remarks>
    /// A fresh scope, disposed at once, rather than services injected into this endpoint — which is
    /// the lesson ADR 0004 records paying for on the relay. This handler runs for as long as the
    /// socket is open, so anything scoped it held would be a <c>DbContext</c> alive for hours, and a
    /// few hundred of those exhaust the connection pool. The database is touched here and in
    /// <see cref="RefreshAsync"/>, briefly, and nowhere else.
    /// </remarks>
    private async Task<(Guid UserId, FrozenSet<Guid> Vaults)> ResolveAccessAsync(
        CancellationToken cancellationToken)
    {
        var scope = scopes.CreateAsyncScope();
        await using var _ = scope.ConfigureAwait(false);

        var currentUser = scope.ServiceProvider.GetRequiredService<ICurrentUserContext>();
        var vaultAccess = scope.ServiceProvider.GetRequiredService<IVaultAccessService>();

        var user = await currentUser.GetOrProvisionAsync(cancellationToken).ConfigureAwait(false);

        return (user.Id, await ReachableAsync(vaultAccess, user.Id, cancellationToken).ConfigureAwait(false));
    }

    private static async Task<FrozenSet<Guid>> ReachableAsync(
        IVaultAccessService vaultAccess,
        Guid userId,
        CancellationToken cancellationToken)
    {
        var accessible = await vaultAccess.ListAsync(userId, cancellationToken).ConfigureAwait(false);

        return accessible
            .Where(access => access.Vault is not null
                && access.Permissions.HasFlag(PermissionFlags.Read))
            .Select(access => access.Vault!.Id)
            .ToFrozenSet();
    }

    /// <summary>Runs the socket until something ends it, then closes it politely.</summary>
    /// <remarks>
    /// <para>
    /// Three loops rather than one: reading a socket and writing to it are independent waits, and the
    /// clock is a third. They are joined by <see cref="Task.WhenAny(Task[])"/>, and the sender is
    /// finished before the close is written, because a close frame racing a notice frame is a protocol
    /// violation that presents as a client dropping its connection for no visible reason.
    /// </para>
    /// <para>
    /// <b>No operation on the WebSocket is ever cancelled.</b> Cancelling a pending receive or send on
    /// the socket Kestrel hands over aborts it, and an aborted socket cannot send the close frame that
    /// is the only way a client learns <em>why</em> it was dropped — 4401 "reconnect with a fresh
    /// token" against 1001 "the server is going away" against a failure to back off from. This was
    /// once the case here, and it went unnoticed because <c>TestServer</c>'s in-memory socket survives
    /// the cancellation that a real one does not. So the token stops the queue and the clock; the
    /// sender is let finish its last write; the close goes out; and the receiver ends on the client's
    /// answering close. Each wait is bounded, and the socket is aborted only when a peer has stopped
    /// answering altogether. The relay's pump does the same, for the same reason.
    /// </para>
    /// </remarks>
    private async Task PumpAsync(
        WebSocket socket,
        VaultEventConnection connection,
        CancellationToken requestAborted)
    {
        var settings = options.Value;

        using var pump = CancellationTokenSource.CreateLinkedTokenSource(
            requestAborted, lifetime.ApplicationStopping);

        var closure = new Closure(
            (int)WebSocketCloseStatus.NormalClosure, string.Empty);

        connection.TryEnqueue(new VaultEvent(
            VaultEventKinds.Hello,
            ServerTime: clock.GetUtcNow(),
            HeartbeatSeconds: (int)settings.HeartbeatInterval.TotalSeconds,
            VaultCount: connection.VaultCount));

        var sending = SendAsync(socket, connection, pump.Token);
        var receiving = ReceiveAsync(socket, connection);
        var minding = MindAsync(connection, closure, pump.Token);

        await Task.WhenAny(sending, receiving, minding).ConfigureAwait(false);

        await pump.CancelAsync().ConfigureAwait(false);

        // Nothing may still be mid-send when the close frame goes out. A send that does not finish is
        // a client that has stopped reading, which will not read a close frame either.
        if (!await SettlesAsync(sending).ConfigureAwait(false))
        {
            socket.Abort();
        }

        await SettlesAsync(minding).ConfigureAwait(false);

        if (lifetime.ApplicationStopping.IsCancellationRequested)
        {
            // 1001 "going away", so a client knows to reconnect immediately rather than treating a
            // rolling deployment as a server that has broken.
            closure.Set((int)WebSocketCloseStatus.EndpointUnavailable, "The server is shutting down.");
        }

        EventsLog.ClosingConnection(logger, connection.UserId, closure.Reason);

        await CloseAsync(socket, closure).ConfigureAwait(false);

        // The client's answering close is what ends the receiver. Returning before it arrives would
        // dispose the socket under a close frame still in flight.
        if (!await SettlesAsync(receiving).ConfigureAwait(false))
        {
            socket.Abort();
            await SettlesAsync(receiving).ConfigureAwait(false);
        }
    }

    /// <summary>
    /// Writes queued frames to the socket, one at a time.
    /// </summary>
    /// <remarks>
    /// The <em>only</em> writer, which is what makes concurrent sends impossible without a lock: the
    /// heartbeat and the pong both go into the same queue rather than to the socket. A WebSocket
    /// permits one send at a time and faults permanently on a second, so this is not a tidiness
    /// preference.
    /// </remarks>
    private async Task SendAsync(
        WebSocket socket,
        VaultEventConnection connection,
        CancellationToken cancellationToken)
    {
        await foreach (var frame in connection.Outbound
            .ReadAllAsync(cancellationToken)
            .ConfigureAwait(false))
        {
            // Re-resolved before the notice is forwarded, not after: the client's answer to this frame
            // is to re-read its vault list, and the point of a newly shared vault is that the *next*
            // change to it produces a notice too. A socket that forwarded first would not follow the
            // new vault until its next periodic refresh.
            if (string.Equals(frame.Kind, VaultEventKinds.VaultsChanged, StringComparison.Ordinal))
            {
                await RefreshAsync(connection, cancellationToken).ConfigureAwait(false);
            }

            var bytes = JsonSerializer.SerializeToUtf8Bytes(frame, DodoSshJsonContext.Default.VaultEvent);

            // Not cancellable, deliberately. See PumpAsync.
            await socket
                .SendAsync(bytes, WebSocketMessageType.Text, endOfMessage: true, CancellationToken.None)
                .ConfigureAwait(false);
        }
    }

    /// <summary>
    /// Reads what the client sends, which in this version is heartbeats and a close.
    /// </summary>
    /// <remarks>
    /// <para>
    /// A client cannot ask to follow a vault, and that is deliberate rather than unfinished: a
    /// <c>subscribe(vaultId)</c> frame is an existence oracle for vault ids, which is the disclosure
    /// <c>SyncPullEndpoint</c> answers 404 rather than 403 to avoid. Subscription is decided from the
    /// caller's access and nothing else.
    /// </para>
    /// <para>
    /// It takes no token: it ends on the client's close, a frame it refuses, or the socket failing —
    /// and when the server is the one ending the connection, the close it sends is what prompts the
    /// first of those. See <see cref="PumpAsync"/>.
    /// </para>
    /// </remarks>
    private static async Task ReceiveAsync(WebSocket socket, VaultEventConnection connection)
    {
        var buffer = new byte[MaxInboundFrameBytes];

        while (true)
        {
            // Not cancellable, deliberately. See PumpAsync.
            var received = await socket.ReceiveAsync(buffer, CancellationToken.None).ConfigureAwait(false);

            if (received.MessageType == WebSocketMessageType.Close)
            {
                return;
            }

            // Oversized, or split across frames. Nothing this protocol sends is either, so the client
            // is broken or probing; ending the socket is cheaper than reassembling for it.
            if (!received.EndOfMessage)
            {
                return;
            }

            // Binary is unused in v1 and skipped rather than refused, because ADR 0012 reserves it for
            // shared-session data — an older server meeting a newer client must ignore those, not
            // close on them.
            if (received.MessageType != WebSocketMessageType.Text)
            {
                continue;
            }

            if (Kind(buffer.AsSpan(0, received.Count)) is VaultEventKinds.Ping)
            {
                connection.TryEnqueue(new VaultEvent(VaultEventKinds.Pong));
            }
        }
    }

    /// <summary>
    /// Reads a frame's kind, or null if it is not one this server understands.
    /// </summary>
    /// <remarks>
    /// A frame that will not parse is skipped rather than closing the socket. This is a control
    /// channel whose failure mode is "the client polls instead", so tolerating a frame from a newer
    /// client costs nothing and refusing one costs that client its push for the whole session.
    /// </remarks>
    private static string? Kind(ReadOnlySpan<byte> utf8)
    {
        try
        {
            return JsonSerializer.Deserialize(utf8, DodoSshJsonContext.Default.VaultEvent)?.Kind;
        }
        catch (JsonException)
        {
            return null;
        }
    }

    /// <summary>
    /// Keeps the heartbeat going, the vault set current, and the socket inside its token's lifetime.
    /// </summary>
    /// <remarks>
    /// <para>
    /// The deadline is the earlier of the access token's <c>exp</c> and a hard cap on how long any one
    /// socket may live. Closing on expiry is what keeps a long-lived connection from outliving the
    /// short-lived credential that authorised it; the client answers by reconnecting with a fresh
    /// token, which is a sub-second gap in a channel that degrades to polling anyway.
    /// </para>
    /// <para>
    /// The wait is the shorter of the heartbeat and the time left, so the deadline is met to within a
    /// tick rather than to within a heartbeat.
    /// </para>
    /// </remarks>
    private async Task MindAsync(
        VaultEventConnection connection,
        Closure closure,
        CancellationToken cancellationToken)
    {
        var settings = options.Value;
        var started = clock.GetUtcNow();

        var deadline = TokenExpiry() is { } expiry && expiry < started + settings.MaxConnectionDuration
            ? (Expiry: expiry, ForToken: true)
            : (Expiry: started + settings.MaxConnectionDuration, ForToken: false);

        var refreshed = started;

        while (!cancellationToken.IsCancellationRequested)
        {
            var now = clock.GetUtcNow();
            var remaining = deadline.Expiry - now;

            if (remaining <= TimeSpan.Zero)
            {
                closure.Set(
                    deadline.ForToken
                        ? VaultEvents.TokenExpiredCloseCode
                        : (int)WebSocketCloseStatus.NormalClosure,
                    deadline.ForToken
                        ? "The access token has expired. Reconnect with a fresh one."
                        : "This connection reached its maximum lifetime.");

                return;
            }

            var wait = settings.HeartbeatInterval < remaining ? settings.HeartbeatInterval : remaining;

            await Task.Delay(wait, clock, cancellationToken).ConfigureAwait(false);

            now = clock.GetUtcNow();

            if (now - refreshed >= settings.AccessRefreshInterval)
            {
                // The backstop for a grant withdrawn while this socket was open. What it bounds is
                // metadata — that a vault changed — because that is all a notice carries and reading
                // the vault still needs a key this server has never held. See ADR 0012.
                await RefreshAsync(connection, cancellationToken).ConfigureAwait(false);
                refreshed = now;
            }

            connection.TryEnqueue(new VaultEvent(VaultEventKinds.Ping, ServerTime: now));
        }
    }

    /// <summary>Re-reads which vaults this socket may follow.</summary>
    /// <remarks>
    /// A failure is logged and swallowed. The alternative is dropping a working socket because one
    /// database call timed out, which would trade an occasionally stale vault set for an outage.
    /// </remarks>
    private async Task RefreshAsync(VaultEventConnection connection, CancellationToken cancellationToken)
    {
        try
        {
            var scope = scopes.CreateAsyncScope();
            await using var _ = scope.ConfigureAwait(false);

            var vaultAccess = scope.ServiceProvider.GetRequiredService<IVaultAccessService>();

            connection.Resubscribe(
                await ReachableAsync(vaultAccess, connection.UserId, cancellationToken)
                    .ConfigureAwait(false));
        }
        catch (OperationCanceledException)
        {
            // The socket is closing.
        }
        catch (Exception exception) when (exception is not OutOfMemoryException)
        {
            EventsLog.AccessRefreshFailed(logger, connection.UserId, exception);
        }
    }

    /// <remarks>
    /// <c>MapInboundClaims</c> is off — see <see cref="Auth"/> — so the claim is spelled as the
    /// provider issued it rather than as a WS-Federation URI. Null is treated as "no bound from the
    /// token", which the bearer handler's <c>RequireExpirationTime</c> should make unreachable; the
    /// lifetime cap covers it either way.
    /// </remarks>
    private DateTimeOffset? TokenExpiry() =>
        long.TryParse(
            HttpContext.User.FindFirstValue("exp"),
            NumberStyles.Integer,
            CultureInfo.InvariantCulture,
            out var seconds)
            ? DateTimeOffset.FromUnixTimeSeconds(seconds)
            : null;

    private static async Task CloseAsync(WebSocket socket, Closure closure)
    {
        if (socket.State is not (WebSocketState.Open or WebSocketState.CloseReceived))
        {
            return;
        }

        using var timeout = new CancellationTokenSource(CloseTimeout);

        try
        {
            await socket
                .CloseOutputAsync((WebSocketCloseStatus)closure.Code, closure.Reason, timeout.Token)
                .ConfigureAwait(false);
        }
        catch (Exception exception)
            when (exception is OperationCanceledException or WebSocketException or ObjectDisposedException)
        {
            // The peer is already gone. There is nothing to tell it and nothing to recover.
        }
    }

    /// <summary>
    /// Waits for a pump loop to finish, for at most the close timeout.
    /// </summary>
    /// <returns>Whether it finished.</returns>
    /// <remarks>
    /// Every one of these loops ends by being cancelled or by the socket going away, so an exception
    /// here is the expected shape of "this connection is over" rather than a fault to propagate — and
    /// propagating it would skip the close frame the other side is waiting for.
    /// </remarks>
    private static async Task<bool> SettlesAsync(Task loop)
    {
        try
        {
            await loop.WaitAsync(CloseTimeout).ConfigureAwait(false);

            return true;
        }
        catch (TimeoutException)
        {
            return false;
        }
        catch (Exception exception)
            when (exception is OperationCanceledException or WebSocketException or ObjectDisposedException)
        {
            return true;
        }
    }

    /// <summary>Why the socket is being closed, decided by whichever loop ended first.</summary>
    /// <remarks>
    /// Mutable and shared, and safe without a lock for one specific reason: it is written only by
    /// <see cref="MindAsync"/>, and read only after that loop has been awaited, which is a memory
    /// barrier. If that loop outlived the close timeout, a late write races the read, and either
    /// answer is still a true one.
    /// </remarks>
    private sealed class Closure(int code, string reason)
    {
        internal int Code { get; private set; } = code;

        internal string Reason { get; private set; } = reason;

        internal void Set(int code, string reason)
        {
            Code = code;
            Reason = reason;
        }
    }
}