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; /// /// Fetches actor-timeline manifests from the configured servers. /// /// /// Servers are an ordered list, and order is the user's trust ranking made /// explicit: for a fetch, servers are tried in order and the first acceptable /// result wins — acceptable meaning it clears the configured match tier. /// /// 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. /// /// // TRACES: JR-025, JR-029, JR-030, JR-037 | PR-005, PR-006 public class ManifestExchangeClient : IManifestExchangeClient, IDisposable { /// Server spec §9: a single manifest response is capped at 2 MiB. public const long MaxManifestBytes = 2 * 1024 * 1024; /// Server spec §9: a bundle response is capped at 25 MiB. public const long MaxBundleBytes = 25L * 1024 * 1024; private static readonly TimeSpan ConnectTimeout = TimeSpan.FromSeconds(5); private static readonly TimeSpan ReadTimeout = TimeSpan.FromSeconds(30); /// /// How long a server that failed is skipped for, doubling each consecutive /// failure. One dead server must never stall a library sweep. /// private static readonly TimeSpan BaseBackoff = TimeSpan.FromMinutes(1); private static readonly TimeSpan MaxBackoff = TimeSpan.FromHours(1); private readonly HttpClient _http; private readonly ILogger _logger; private readonly ConcurrentDictionary _health = new(StringComparer.Ordinal); private readonly bool _ownsClient; /// /// Initializes a new instance of the class. /// /// Logger. public ManifestExchangeClient(ILogger logger) : this(logger, null) { } /// /// Initializes a new instance of the class /// with an injected transport, for testing. /// /// Logger. /// Transport to use, or null to build the default. public ManifestExchangeClient(ILogger 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, }; } /// public async Task FetchMovieAsync( IReadOnlyList 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; } /// public async Task FetchEpisodeAsync( IReadOnlyList 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; } /// public IReadOnlyList GetStatus(IReadOnlyList 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(); } /// /// Servers that are enabled and not currently in backoff, in configured order. /// private IEnumerable Eligible(IReadOnlyList 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 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(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; } } /// /// Reads a response body, aborting once it exceeds . /// /// /// Capped while streaming rather than after buffering: a hostile /// server can declare any Content-Length it likes, so reading to /// completion and then measuring is exactly the denial-of-service primitive /// the cap exists to prevent. /// // TRACES: JR-028 | SR-004 private static async Task 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()); } /// /// Builds a fetch URL, or null when the server's URL is unusable. /// /// /// HTTPS is required for anything that is not loopback. A plaintext /// community server would let any network intermediary rewrite actor /// overlays, and the overlay is displayed to the user as fact. /// /// The configured server. /// API path below /api/v1/. /// Identity and cut parameters. /// The URL to request, or null when the server URL is unusable. 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; } /// /// True when the URL may be used: HTTPS anywhere, or HTTP on loopback only. /// /// The server base URL. /// true when the transport is acceptable. 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; } /// Parses a tier name the server reported. /// The tier string. /// The tier, or null when unrecognised. 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); } /// public void Dispose() { Dispose(true); GC.SuppressFinalize(this); } /// /// Releases the transport when this instance created it. /// /// Whether managed resources should be released. 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; } } }