jRay/Jellyfin.Plugin.JRay/Services/ManifestExchangeClient.cs

using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Globalization;
using System.Linq;
using System.Net.Http;
using System.Text.Json;
using System.Threading;
using System.Threading.Tasks;
using Jellyfin.Plugin.JRay.Configuration;
using Jellyfin.Plugin.JRay.Models;
using Jellyfin.Plugin.JRay.Services.Interfaces;
using Microsoft.Extensions.Logging;

namespace Jellyfin.Plugin.JRay.Services;

/// <summary>
/// Fetches actor-timeline manifests from the configured servers.
/// </summary>
/// <remarks>
/// Servers are an <b>ordered list</b>, and order is the user's trust ranking made
/// explicit: for a fetch, servers are tried in order and the <i>first acceptable</i>
/// result wins — acceptable meaning it clears the configured match tier.
/// <para>
/// First-match rather than best-match is deliberate. Querying every server for
/// every item multiplies egress, leaks the library to more parties, and the
/// ordering already encodes which source the admin prefers. Each configured
/// server multiplies the privacy exposure described in the server spec §9, so
/// later servers are queried only for what earlier ones lacked.
/// </para>
/// </remarks>
// TRACES: JR-025, JR-029, JR-030, JR-037 | PR-005, PR-006
public class ManifestExchangeClient : IManifestExchangeClient, IDisposable
{
    /// <summary>Server spec §9: a single manifest response is capped at 2 MiB.</summary>
    public const long MaxManifestBytes = 2 * 1024 * 1024;

    /// <summary>Server spec §9: a bundle response is capped at 25 MiB.</summary>
    public const long MaxBundleBytes = 25L * 1024 * 1024;

    private static readonly TimeSpan ConnectTimeout = TimeSpan.FromSeconds(5);
    private static readonly TimeSpan ReadTimeout = TimeSpan.FromSeconds(30);

    /// <summary>
    /// How long a server that failed is skipped for, doubling each consecutive
    /// failure. One dead server must never stall a library sweep.
    /// </summary>
    private static readonly TimeSpan BaseBackoff = TimeSpan.FromMinutes(1);
    private static readonly TimeSpan MaxBackoff = TimeSpan.FromHours(1);

    private readonly HttpClient _http;
    private readonly ILogger<ManifestExchangeClient> _logger;
    private readonly ConcurrentDictionary<string, ServerHealth> _health = new(StringComparer.Ordinal);
    private readonly bool _ownsClient;

    /// <summary>
    /// Initializes a new instance of the <see cref="ManifestExchangeClient"/> class.
    /// </summary>
    /// <param name="logger">Logger.</param>
    public ManifestExchangeClient(ILogger<ManifestExchangeClient> logger)
        : this(logger, null)
    {
    }

    /// <summary>
    /// Initializes a new instance of the <see cref="ManifestExchangeClient"/> class
    /// with an injected transport, for testing.
    /// </summary>
    /// <param name="logger">Logger.</param>
    /// <param name="httpClient">Transport to use, or null to build the default.</param>
    public ManifestExchangeClient(ILogger<ManifestExchangeClient> logger, HttpClient? httpClient)
    {
        _logger = logger;
        _ownsClient = httpClient is null;
        _http = httpClient ?? new HttpClient(new SocketsHttpHandler
        {
            ConnectTimeout = ConnectTimeout,
            // Certificate validation is never disabled: a plaintext or
            // unverified server would let any network intermediary rewrite
            // actor overlays.
            AutomaticDecompression = System.Net.DecompressionMethods.All,
        })
        {
            Timeout = ReadTimeout,
        };
    }

    /// <inheritdoc />
    public async Task<ManifestFetchOutcome?> FetchMovieAsync(
        IReadOnlyList<ManifestServer> servers,
        MatchTier minimumTier,
        TitleQuery query,
        CancellationToken cancellationToken)
    {
        ArgumentNullException.ThrowIfNull(servers);
        ArgumentNullException.ThrowIfNull(query);

        foreach (var server in Eligible(servers))
        {
            var url = BuildUrl(server, "manifests/movie", query);
            if (url is null)
            {
                continue;
            }

            var outcome = await TryFetchOneAsync(server, url, query, minimumTier, cancellationToken)
                .ConfigureAwait(false);
            if (outcome is not null)
            {
                // First acceptable result wins — no further servers are queried,
                // which is what bounds the privacy exposure.
                return outcome;
            }
        }

        return null;
    }

    /// <inheritdoc />
    public async Task<ManifestFetchOutcome?> FetchEpisodeAsync(
        IReadOnlyList<ManifestServer> servers,
        MatchTier minimumTier,
        TitleQuery query,
        CancellationToken cancellationToken)
    {
        ArgumentNullException.ThrowIfNull(servers);
        ArgumentNullException.ThrowIfNull(query);

        foreach (var server in Eligible(servers))
        {
            var url = BuildUrl(server, "manifests/episode", query);
            if (url is null)
            {
                continue;
            }

            var outcome = await TryFetchOneAsync(server, url, query, minimumTier, cancellationToken)
                .ConfigureAwait(false);
            if (outcome is not null)
            {
                return outcome;
            }
        }

        return null;
    }

    /// <inheritdoc />
    public IReadOnlyList<ServerStatus> GetStatus(IReadOnlyList<ManifestServer> servers)
    {
        ArgumentNullException.ThrowIfNull(servers);

        return servers.Select(s =>
        {
            _health.TryGetValue(s.Url, out var h);
            return new ServerStatus
            {
                Url = s.Url,
                Name = s.Name,
                Enabled = s.Enabled,
                Reachable = h is null || h.ConsecutiveFailures == 0,
                LastError = h?.LastError,
                SkippedUntil = h?.SkipUntil,
            };
        }).ToList();
    }

    /// <summary>
    /// Servers that are enabled and not currently in backoff, in configured order.
    /// </summary>
    private IEnumerable<ManifestServer> Eligible(IReadOnlyList<ManifestServer> servers)
    {
        var now = DateTimeOffset.UtcNow;
        foreach (var s in servers)
        {
            if (!s.Enabled || string.IsNullOrWhiteSpace(s.Url))
            {
                continue;
            }

            if (_health.TryGetValue(s.Url, out var h) && h.SkipUntil > now)
            {
                _logger.LogDebug("Skipping {Url} until {Until} after {Failures} failures", s.Url, h.SkipUntil, h.ConsecutiveFailures);
                continue;
            }

            yield return s;
        }
    }

    private async Task<ManifestFetchOutcome?> TryFetchOneAsync(
        ManifestServer server,
        Uri url,
        TitleQuery query,
        MatchTier minimumTier,
        CancellationToken cancellationToken)
    {
        try
        {
            using var response = await _http
                .GetAsync(url, HttpCompletionOption.ResponseHeadersRead, cancellationToken)
                .ConfigureAwait(false);

            if (response.StatusCode == System.Net.HttpStatusCode.NotFound)
            {
                // Not an error: this server simply does not hold it. The next
                // server in the list gets a turn.
                RecordSuccess(server.Url);
                return null;
            }

            if (!response.IsSuccessStatusCode)
            {
                RecordFailure(server.Url, $"HTTP {(int)response.StatusCode}");
                return null;
            }

            var json = await ReadCappedAsync(response, MaxManifestBytes, cancellationToken)
                .ConfigureAwait(false);
            if (json is null)
            {
                RecordFailure(server.Url, "response exceeded the size cap");
                return null;
            }

            var body = JsonSerializer.Deserialize<ManifestFetchResponse>(json);
            RecordSuccess(server.Url);

            if (body?.Manifest is null)
            {
                return null;
            }

            var tier = ParseTier(body.Match);
            if (tier is null || tier < minimumTier)
            {
                _logger.LogDebug(
                    "{Url} matched at {Tier}, below the configured minimum {Minimum}",
                    server.Url,
                    body.Match,
                    minimumTier);
                return null;
            }

            if (!ManifestValidator.TryValidate(body.Manifest, query.RuntimeSec, out var error))
            {
                // A manifest is never trusted merely because a server served it.
                _logger.LogWarning("Rejected manifest from {Url}: {Error}", server.Url, error);
                return null;
            }

            return new ManifestFetchOutcome
            {
                ServerUrl = server.Url,
                Tier = tier.Value,
                OffsetSec = body.OffsetSec,
                Manifest = body.Manifest,
            };
        }
        catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
        {
            throw;
        }
        catch (Exception ex) when (ex is HttpRequestException or TaskCanceledException or JsonException)
        {
            // A slow, unreachable or nonsense-returning server is skipped and
            // backed off; it must never stall the sweep or fail the whole fetch.
            RecordFailure(server.Url, ex.Message);
            return null;
        }
    }

    /// <summary>
    /// Reads a response body, aborting once it exceeds <paramref name="cap"/>.
    /// </summary>
    /// <remarks>
    /// Capped <b>while streaming</b> rather than after buffering: a hostile
    /// server can declare any <c>Content-Length</c> it likes, so reading to
    /// completion and then measuring is exactly the denial-of-service primitive
    /// the cap exists to prevent.
    /// </remarks>
    private static async Task<string?> ReadCappedAsync(
        HttpResponseMessage response,
        long cap,
        CancellationToken cancellationToken)
    {
        // The declared length is a cheap early rejection, never the enforcement.
        if (response.Content.Headers.ContentLength is { } declared && declared > cap)
        {
            return null;
        }

        using var stream = await response.Content.ReadAsStreamAsync(cancellationToken).ConfigureAwait(false);
        var buffer = new byte[8192];
        using var accumulated = new System.IO.MemoryStream();

        while (true)
        {
            var read = await stream.ReadAsync(buffer, cancellationToken).ConfigureAwait(false);
            if (read == 0)
            {
                break;
            }

            if (accumulated.Length + read > cap)
            {
                return null;
            }

            await accumulated.WriteAsync(buffer.AsMemory(0, read), cancellationToken).ConfigureAwait(false);
        }

        return System.Text.Encoding.UTF8.GetString(accumulated.ToArray());
    }

    /// <summary>
    /// Builds a fetch URL, or null when the server's URL is unusable.
    /// </summary>
    /// <remarks>
    /// <b>HTTPS is required for anything that is not loopback.</b> A plaintext
    /// community server would let any network intermediary rewrite actor
    /// overlays, and the overlay is displayed to the user as fact.
    /// </remarks>
    /// <param name="server">The configured server.</param>
    /// <param name="path">API path below <c>/api/v1/</c>.</param>
    /// <param name="query">Identity and cut parameters.</param>
    /// <returns>The URL to request, or null when the server URL is unusable.</returns>
    internal static Uri? BuildUrl(ManifestServer server, string path, TitleQuery query)
    {
        ArgumentNullException.ThrowIfNull(server);
        ArgumentNullException.ThrowIfNull(query);

        if (!Uri.TryCreate(server.Url, UriKind.Absolute, out var baseUri))
        {
            return null;
        }

        if (!IsTransportAcceptable(baseUri))
        {
            return null;
        }

        var q = query.ToQueryString();
        var trimmed = baseUri.AbsoluteUri.TrimEnd('/');
        return Uri.TryCreate($"{trimmed}/api/v1/{path}?{q}", UriKind.Absolute, out var built)
            ? built
            : null;
    }

    /// <summary>
    /// True when the URL may be used: HTTPS anywhere, or HTTP on loopback only.
    /// </summary>
    /// <param name="uri">The server base URL.</param>
    /// <returns><c>true</c> when the transport is acceptable.</returns>
    internal static bool IsTransportAcceptable(Uri uri)
    {
        ArgumentNullException.ThrowIfNull(uri);

        if (uri.Scheme == Uri.UriSchemeHttps)
        {
            return true;
        }

        if (uri.Scheme != Uri.UriSchemeHttp)
        {
            return false;
        }

        // Loopback is exempt because there is no network path to intercept.
        return uri.IsLoopback;
    }

    /// <summary>Parses a tier name the server reported.</summary>
    /// <param name="tier">The tier string.</param>
    /// <returns>The tier, or null when unrecognised.</returns>
    internal static MatchTier? ParseTier(string? tier) => tier switch
    {
        // `exact` is deliberately absent: the file-hash tier is withdrawn on
        // legal grounds (see MatchTier). A server cannot report it to us anyway,
        // since we send no `video_hash` — and if one did, treating it as
        // unrecognised means the manifest is declined rather than silently
        // accepted under a tier this plugin has no policy for.
        "audio" => MatchTier.Audio,
        "runtime" => MatchTier.Runtime,
        "loose" => MatchTier.Loose,
        _ => null,
    };

    private void RecordSuccess(string url) => _health.TryRemove(url, out _);

    private void RecordFailure(string url, string error)
    {
        var updated = _health.AddOrUpdate(
            url,
            _ => new ServerHealth { ConsecutiveFailures = 1, LastError = error, SkipUntil = DateTimeOffset.UtcNow + BaseBackoff },
            (_, existing) =>
            {
                var failures = existing.ConsecutiveFailures + 1;
                // Exponential, capped: a server that is down for a day should
                // not be retried every minute for that whole day.
                var delayTicks = Math.Min(
                    BaseBackoff.Ticks * (long)Math.Pow(2, Math.Min(failures - 1, 6)),
                    MaxBackoff.Ticks);
                return new ServerHealth
                {
                    ConsecutiveFailures = failures,
                    LastError = error,
                    SkipUntil = DateTimeOffset.UtcNow + TimeSpan.FromTicks(delayTicks),
                };
            });

        _logger.LogWarning(
            "Server {Url} failed ({Failures} consecutive): {Error}. Skipping until {Until}",
            url,
            updated.ConsecutiveFailures,
            error,
            updated.SkipUntil);
    }

    /// <inheritdoc />
    public void Dispose()
    {
        Dispose(true);
        GC.SuppressFinalize(this);
    }

    /// <summary>
    /// Releases the transport when this instance created it.
    /// </summary>
    /// <param name="disposing">Whether managed resources should be released.</param>
    protected virtual void Dispose(bool disposing)
    {
        if (disposing && _ownsClient)
        {
            _http.Dispose();
        }
    }

    private sealed class ServerHealth
    {
        public int ConsecutiveFailures { get; init; }

        public string? LastError { get; init; }

        public DateTimeOffset SkipUntil { get; init; }
    }
}