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; }
}
}