ScadaBridge/src/ZB.MOM.WW.ScadaBridge.InboundAPI/HttpStandbyRequestForwarder.cs

using System.Net;
using System.Net.Http.Headers;
using Microsoft.AspNetCore.Http;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Options;

namespace ZB.MOM.WW.ScadaBridge.InboundAPI;

/// <summary>
/// Production <see cref="IStandbyRequestForwarder"/>: relays a standby-node inbound
/// API request to the active central node over plain HTTP using the named
/// <see cref="HttpClientName"/> client.
///
/// <para><b>Peer selection.</b> The standby does not know which peer is active — it
/// only knows it is not. It therefore tries
/// <see cref="InboundApiOptions.StandbyForwardPeerEndpoints"/> in order and accepts the
/// first response that is not itself a standby refusal. A peer that is also a standby
/// (mid-failover, or a misconfigured list that includes this node) answers 503 with the
/// <see cref="StandbyResponseHeader"/> marker, which the forwarder treats like a
/// connection failure: try the next peer. Connection errors and per-attempt timeouts
/// (<see cref="InboundApiOptions.StandbyForwardTimeout"/>) also move to the next peer.
/// When every peer fails the caller gets 503 <c>ACTIVE_NODE_UNREACHABLE</c>.</para>
///
/// <para><b>Loop guard.</b> Every relayed request carries
/// <see cref="ForwardedRequestHeader"/>. <see cref="InboundApiEndpointFilter"/> never
/// forwards a request that already carries it, so a pair in which neither node is
/// active (a transient during failover) bounces at most one hop, never ping-pongs.</para>
///
/// <para><b>Fidelity.</b> Method, path, query string, body and headers are relayed
/// verbatim (hop-by-hop headers and <c>Host</c> excepted), so the active node
/// authenticates the caller's own <c>Authorization</c>/<c>X-API-Key</c> credential and
/// applies its own body cap and method timeout. The active node's status, content type
/// and body are replayed to the caller unchanged. The body is buffered once on the
/// standby (bounded by <see cref="InboundApiOptions.MaxRequestBodyBytes"/>, which the
/// filter enforces before calling this class) so it can be re-sent to a second peer.</para>
/// </summary>
public sealed class HttpStandbyRequestForwarder : IStandbyRequestForwarder
{
    /// <summary>Name of the <see cref="IHttpClientFactory"/> client used for relays.</summary>
    public const string HttpClientName = "ScadaBridge.InboundApi.StandbyForward";

    /// <summary>
    /// Request header stamped on every relayed request (value: the forwarding node's
    /// host name when known, else <c>1</c>). The filter refuses to forward a request
    /// that already carries it.
    /// </summary>
    public const string ForwardedRequestHeader = "X-ScadaBridge-Forwarded";

    /// <summary>
    /// Response header the filter adds to its own 503 <c>STANDBY_NODE</c> reply so a
    /// forwarding peer can distinguish "that node is also standby — try the next one"
    /// from a genuine 503 the active node's method script produced.
    /// </summary>
    public const string StandbyResponseHeader = "X-ScadaBridge-Standby";

    /// <summary>Machine-readable <c>code</c> when no configured peer served the relay.</summary>
    public const string ActiveNodeUnreachableCode = "ACTIVE_NODE_UNREACHABLE";

    private static readonly HashSet<string> HopByHopRequestHeaders = new(StringComparer.OrdinalIgnoreCase)
    {
        "Host", "Content-Length", "Transfer-Encoding", "Connection", "Keep-Alive",
        "Proxy-Connection", "Upgrade", "TE", "Trailer", "Expect",
    };

    private readonly IHttpClientFactory _httpClientFactory;
    private readonly IOptionsMonitor<InboundApiOptions> _options;
    private readonly ILogger<HttpStandbyRequestForwarder> _logger;

    /// <summary>Initializes a new <see cref="HttpStandbyRequestForwarder"/>.</summary>
    /// <param name="httpClientFactory">Factory for the named relay client.</param>
    /// <param name="options">Live-reloadable inbound API options (peer list + timeout are read per relay).</param>
    /// <param name="logger">Logger for relay diagnostics.</param>
    public HttpStandbyRequestForwarder(
        IHttpClientFactory httpClientFactory,
        IOptionsMonitor<InboundApiOptions> options,
        ILogger<HttpStandbyRequestForwarder> logger)
    {
        _httpClientFactory = httpClientFactory;
        _options = options;
        _logger = logger;
    }

    /// <inheritdoc />
    public bool IsConfigured => _options.CurrentValue.StandbyForwardPeerEndpoints.Count > 0;

    /// <inheritdoc />
    public async Task<IResult> ForwardAsync(HttpContext context, CancellationToken cancellationToken)
    {
        var opts = _options.CurrentValue;
        var peers = opts.StandbyForwardPeerEndpoints;
        if (peers.Count == 0)
            return Unreachable();

        // Buffer the body once so it can be replayed to a second peer. The filter has
        // already enforced MaxRequestBodyBytes (Content-Length up front, and the
        // per-request Kestrel cap for chunked bodies), so this is bounded.
        byte[] body;
        using (var buffer = new MemoryStream())
        {
            await context.Request.Body.CopyToAsync(buffer, cancellationToken).ConfigureAwait(false);
            body = buffer.ToArray();
        }

        var client = _httpClientFactory.CreateClient(HttpClientName);
        var pathAndQuery = context.Request.Path.Value + context.Request.QueryString.Value;
        var forwardedBy = string.IsNullOrEmpty(context.Request.Host.Host) ? "1" : context.Request.Host.Host;

        foreach (var peer in peers)
        {
            cancellationToken.ThrowIfCancellationRequested();

            using var attemptCts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
            attemptCts.CancelAfter(opts.StandbyForwardTimeout);

            try
            {
                using var request = BuildRequest(context, peer, pathAndQuery, body, forwardedBy);
                var response = await client
                    .SendAsync(request, HttpCompletionOption.ResponseHeadersRead, attemptCts.Token)
                    .ConfigureAwait(false);

                if (response.StatusCode == HttpStatusCode.ServiceUnavailable
                    && response.Headers.Contains(StandbyResponseHeader))
                {
                    response.Dispose();
                    _logger.LogInformation(
                        "Inbound API relay: peer {Peer} is also a standby — trying the next peer", peer);
                    continue;
                }

                var responseBody = await response.Content
                    .ReadAsByteArrayAsync(attemptCts.Token)
                    .ConfigureAwait(false);
                var contentType = response.Content.Headers.ContentType?.ToString();
                var status = (int)response.StatusCode;
                response.Dispose();

                _logger.LogDebug(
                    "Inbound API relay: served {Path} via active peer {Peer} → HTTP {Status}",
                    context.Request.Path, peer, status);
                return new RelayedResult(status, contentType, responseBody);
            }
            catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
            {
                throw; // caller went away — nothing to answer
            }
            catch (OperationCanceledException)
            {
                _logger.LogWarning(
                    "Inbound API relay: peer {Peer} did not answer within {Timeout} — trying the next peer",
                    peer, opts.StandbyForwardTimeout);
            }
            catch (HttpRequestException ex)
            {
                _logger.LogWarning(ex,
                    "Inbound API relay: peer {Peer} unreachable — trying the next peer", peer);
            }
        }

        _logger.LogWarning(
            "Inbound API relay: no configured peer served {Path} ({PeerCount} tried)",
            context.Request.Path, peers.Count);
        return Unreachable();
    }

    private static HttpRequestMessage BuildRequest(
        HttpContext context, string peer, string pathAndQuery, byte[] body, string forwardedBy)
    {
        var request = new HttpRequestMessage(
            new HttpMethod(context.Request.Method),
            new Uri(new Uri(peer.TrimEnd('/') + "/"), pathAndQuery.TrimStart('/')));

        var content = new ByteArrayContent(body);
        foreach (var header in context.Request.Headers)
        {
            if (HopByHopRequestHeaders.Contains(header.Key)
                || string.Equals(header.Key, ForwardedRequestHeader, StringComparison.OrdinalIgnoreCase))
                continue;

            var values = header.Value.ToArray();
            if (!request.Headers.TryAddWithoutValidation(header.Key, values))
                content.Headers.TryAddWithoutValidation(header.Key, values);
        }

        if (content.Headers.ContentType == null
            && MediaTypeHeaderValue.TryParse(context.Request.ContentType, out var mediaType))
            content.Headers.ContentType = mediaType;

        request.Content = content;
        request.Headers.TryAddWithoutValidation(ForwardedRequestHeader, forwardedBy);
        return request;
    }

    private static IResult Unreachable() => Results.Json(
        new
        {
            error = "Inbound API request could not be relayed to the active central node",
            code = ActiveNodeUnreachableCode,
        },
        statusCode: StatusCodes.Status503ServiceUnavailable);

    /// <summary>
    /// Replays a peer's status, content type and body to the original caller.
    /// Exposed so tests can assert on what the relay produced without a server.
    /// </summary>
    public sealed class RelayedResult : IResult, IStatusCodeHttpResult
    {
        /// <summary>Initializes a new <see cref="RelayedResult"/>.</summary>
        /// <param name="statusCode">The peer's HTTP status.</param>
        /// <param name="contentType">The peer's <c>Content-Type</c>, if any.</param>
        /// <param name="body">The peer's response body.</param>
        public RelayedResult(int statusCode, string? contentType, byte[] body)
        {
            StatusCode = statusCode;
            ContentType = contentType;
            Body = body;
        }

        /// <inheritdoc />
        public int? StatusCode { get; }

        /// <summary>The peer's <c>Content-Type</c>, replayed verbatim.</summary>
        public string? ContentType { get; }

        /// <summary>The peer's response body, replayed verbatim.</summary>
        public byte[] Body { get; }

        /// <inheritdoc />
        public async Task ExecuteAsync(HttpContext httpContext)
        {
            httpContext.Response.StatusCode = StatusCode ?? StatusCodes.Status200OK;
            if (ContentType != null)
                httpContext.Response.ContentType = ContentType;
            httpContext.Response.ContentLength = Body.Length;
            await httpContext.Response.Body.WriteAsync(Body, httpContext.RequestAborted).ConfigureAwait(false);
        }
    }
}