diff --git a/src/AvParser.Core/Collecting/ICollectRunner.cs b/src/AvParser.Core/Collecting/ICollectRunner.cs new file mode 100644 index 0000000..a8a04d8 --- /dev/null +++ b/src/AvParser.Core/Collecting/ICollectRunner.cs @@ -0,0 +1,34 @@ +using AvParser.Core.Parsing; + +namespace AvParser.Core.Collecting; + +/// How to run one collection. +public sealed record CollectOptions +{ + /// How many downloads may be in flight at once, across all hosts. + /// + /// The per-host cap sits underneath this and is usually what actually binds: a run against a + /// single origin is limited by politeness to that origin, not by this number. + /// + public int MaxConcurrentDownloads { get; init; } = 4; + + /// Ignore the journal and fetch every address again. + public bool ForceRefetch { get; init; } +} + +/// Runs a source end to end: discover, download, store. +/// +/// Yields the same stream the page already knows how to consume, so +/// the UI needs no notion of channels, workers or journals. +/// +public interface ICollectRunner +{ + /// Collects everything finds for . + IAsyncEnumerable> RunAsync( + IMediaSource source, + MediaQuery query, + CollectOptions options, + IProgress? progress, + CancellationToken cancellationToken + ); +} diff --git a/src/AvParser.Core/Collecting/IMediaSource.cs b/src/AvParser.Core/Collecting/IMediaSource.cs new file mode 100644 index 0000000..2b38554 --- /dev/null +++ b/src/AvParser.Core/Collecting/IMediaSource.cs @@ -0,0 +1,75 @@ +using AvParser.Core.Parsing; + +namespace AvParser.Core.Collecting; + +/// +/// Closed, non-generic facade over for media discovery. +/// +/// +/// Closed for the same reason the text facade was: an open generic interface cannot be resolved as +/// IEnumerable<T> by the container, so every source implements this and gets registered +/// under it. Adding a source stays a one-line change. +/// +public interface IMediaSource : IParser; + +/// Read-only view over every registered media source. +public interface IMediaSourceCatalog +{ + /// All registered sources, ordered by . + IReadOnlyList Sources { get; } + + /// The source used when nothing has been chosen yet. + IMediaSource DefaultSource { get; } + + /// Finds a source by its stable id; when unknown. + IMediaSource? Find(string? id); + + /// Finds a source by id, falling back to . + IMediaSource FindOrDefault(string? id) => Find(id) ?? DefaultSource; +} + +/// +public sealed class MediaSourceCatalog : IMediaSourceCatalog +{ + private readonly Dictionary _byId; + + /// Builds a catalog from every source the container resolved. + /// Registered sources. + /// + /// Which source to land on. Named explicitly rather than taken as "first alphabetically", + /// because that would make the landing page depend on a display name — and would put a network + /// source there, so the app would open behind the proxy gate for no reason. + /// + /// No sources were registered, or two share an id. + public MediaSourceCatalog(IEnumerable sources, string? defaultId = null) + { + ArgumentNullException.ThrowIfNull(sources); + + Sources = [.. sources.OrderBy(source => source.DisplayName, StringComparer.OrdinalIgnoreCase)]; + + if (Sources.Count == 0) + { + throw new ArgumentException("At least one media source must be registered.", nameof(sources)); + } + + _byId = new Dictionary(StringComparer.OrdinalIgnoreCase); + foreach (var source in Sources) + { + if (!_byId.TryAdd(source.Id, source)) + { + throw new ArgumentException($"Duplicate media source id '{source.Id}'.", nameof(sources)); + } + } + + DefaultSource = Find(defaultId) ?? Sources[0]; + } + + /// + public IReadOnlyList Sources { get; } + + /// + public IMediaSource DefaultSource { get; } + + /// + public IMediaSource? Find(string? id) => id is not null && _byId.TryGetValue(id, out var source) ? source : null; +} diff --git a/src/AvParser.Core/Collecting/Sources/UrlListSource.cs b/src/AvParser.Core/Collecting/Sources/UrlListSource.cs new file mode 100644 index 0000000..a729828 --- /dev/null +++ b/src/AvParser.Core/Collecting/Sources/UrlListSource.cs @@ -0,0 +1,122 @@ +using System.Runtime.CompilerServices; +using AvParser.Core.Parsing; + +namespace AvParser.Core.Collecting.Sources; + +/// +/// Reads addresses the user pasted in, one per line. +/// +/// +/// The only source that touches no network at all, which is why it lives in the domain and why +/// RequiresNetwork stays false: it discovers nothing, it just reads what it was handed. +/// Downloading those addresses is the fetcher's business and is gated separately. +/// +public sealed class UrlListSource : IMediaSource +{ + /// + public string Id => "url-list"; + + /// + public string DisplayName => "URL list"; + + /// + public string Description => "One address per line. Blank lines and lines starting with '#' are ignored."; + + /// + public bool CanParse(MediaQuery input) => + input is not null && TextLines.Split(input.Text).Any(line => TryParse(line, out _)); + + /// + public async IAsyncEnumerable> ParseAsync( + MediaQuery input, + IProgress? progress, + [EnumeratorCancellation] CancellationToken cancellationToken + ) + { + ArgumentNullException.ThrowIfNull(input); + + var lines = TextLines.Split(input.Text); + var ordinal = 0; + var yielded = 0; + + for (var index = 0; index < lines.Length; index++) + { + cancellationToken.ThrowIfCancellationRequested(); + + var line = lines[index]; + + if (TextLines.IsSkippable(line)) + { + continue; + } + + ordinal++; + + if (TryParse(line, out var url)) + { + yield return ParseOutcome.Success( + new MediaCandidate(url!) { SourceId = Id, Ordinal = ordinal } + ); + + yielded++; + + if (input.HasLimit && yielded >= input.Limit) + { + break; + } + } + else + { + // Named rather than silently dropped: a mistyped address in a pasted list of two + // hundred is otherwise impossible to find. The line goes in twice on purpose — as + // the English fallback message, and as the argument the translation substitutes. + var offending = line.Trim(); + + yield return ParseOutcome.Failure( + ParseError.Create( + ordinal, + "NotAnAddress", + $"'{offending}' is not an http or https address.", + offending + ) + ); + } + + if (ordinal % TextLines.ProgressInterval == 0) + { + progress?.Report(new ParseProgress(index + 1, lines.Length)); + } + + if (ordinal % TextLines.YieldInterval == 0) + { + await Task.Yield(); + } + } + + progress?.Report(new ParseProgress(lines.Length, lines.Length)); + } + + /// Accepts only absolute HTTP addresses. + /// + /// file: and data: are rejected here as well as at redirect time: a pasted list is + /// as likely to have come from somewhere else as to have been typed. + /// + private static bool TryParse(string line, out Uri? url) + { + url = null; + var text = line.Trim(); + + if (text.Length == 0 || !Uri.TryCreate(text, UriKind.Absolute, out var parsed)) + { + return false; + } + + if (parsed.Scheme is not ("http" or "https")) + { + return false; + } + + url = parsed; + return true; + } +} diff --git a/src/AvParser.Core/Parsing/Samples/TextLines.cs b/src/AvParser.Core/Collecting/TextLines.cs similarity index 89% rename from src/AvParser.Core/Parsing/Samples/TextLines.cs rename to src/AvParser.Core/Collecting/TextLines.cs index 6b30fca..7291306 100644 --- a/src/AvParser.Core/Parsing/Samples/TextLines.cs +++ b/src/AvParser.Core/Collecting/TextLines.cs @@ -1,6 +1,6 @@ -namespace AvParser.Core.Parsing.Samples; +namespace AvParser.Core.Collecting; -/// Line-splitting helpers shared by the sample parsers. +/// Line-splitting helpers for anything that reads a pasted list. internal static class TextLines { /// How many records to process between progress reports. diff --git a/src/AvParser.Core/DependencyInjection/CoreServiceCollectionExtensions.cs b/src/AvParser.Core/DependencyInjection/CoreServiceCollectionExtensions.cs index fb29a29..dc6b5d5 100644 --- a/src/AvParser.Core/DependencyInjection/CoreServiceCollectionExtensions.cs +++ b/src/AvParser.Core/DependencyInjection/CoreServiceCollectionExtensions.cs @@ -1,3 +1,5 @@ +using AvParser.Core.Collecting; +using AvParser.Core.Collecting.Sources; using AvParser.Core.Parsing; using AvParser.Core.Parsing.Samples; using Microsoft.Extensions.DependencyInjection; @@ -22,6 +24,15 @@ public static class CoreServiceCollectionExtensions services.AddSingleton(); services.AddSingleton(); + services.AddSingleton(); + return services; } + + /// Id of the source the collector opens on. + /// + /// Named rather than left to alphabetical order, which would land on the network source and + /// open the page behind the proxy gate before the user has asked for anything. + /// + public const string DefaultMediaSourceId = "url-list"; } diff --git a/src/AvParser.Core/Parsing/Samples/DelimitedTextParser.cs b/src/AvParser.Core/Parsing/Samples/DelimitedTextParser.cs index f0c66d5..0e8b561 100644 --- a/src/AvParser.Core/Parsing/Samples/DelimitedTextParser.cs +++ b/src/AvParser.Core/Parsing/Samples/DelimitedTextParser.cs @@ -1,4 +1,5 @@ using System.Runtime.CompilerServices; +using AvParser.Core.Collecting; namespace AvParser.Core.Parsing.Samples; diff --git a/src/AvParser.Core/Parsing/Samples/KeyValueTextParser.cs b/src/AvParser.Core/Parsing/Samples/KeyValueTextParser.cs index 826ffed..ae05e8d 100644 --- a/src/AvParser.Core/Parsing/Samples/KeyValueTextParser.cs +++ b/src/AvParser.Core/Parsing/Samples/KeyValueTextParser.cs @@ -1,4 +1,5 @@ using System.Runtime.CompilerServices; +using AvParser.Core.Collecting; namespace AvParser.Core.Parsing.Samples; diff --git a/src/AvParser.Desktop/App.axaml.cs b/src/AvParser.Desktop/App.axaml.cs index dc7af12..aeadda4 100644 --- a/src/AvParser.Desktop/App.axaml.cs +++ b/src/AvParser.Desktop/App.axaml.cs @@ -2,6 +2,7 @@ using Avalonia; using Avalonia.Controls; using Avalonia.Controls.ApplicationLifetimes; using Avalonia.Markup.Xaml; +using AvParser.Core.Collecting; using AvParser.Core.Settings; using AvParser.Infrastructure.Proxies; using AvParser.UI; @@ -73,6 +74,10 @@ public partial class App : Application // would be the wrong trade. It never throws, so there is nothing to observe. _ = proxies.EnsureLoadedAsync(); + // Creates the media schema and sweeps anything a killed process left staged. Awaited, + // unlike the pool: it is local, it is quick, and every collect run assumes it has happened. + _services.GetRequiredService().InitialiseAsync().GetAwaiter().GetResult(); + base.OnFrameworkInitializationCompleted(); } diff --git a/src/AvParser.Infrastructure/Collecting/CollectRunner.cs b/src/AvParser.Infrastructure/Collecting/CollectRunner.cs new file mode 100644 index 0000000..48b5fba --- /dev/null +++ b/src/AvParser.Infrastructure/Collecting/CollectRunner.cs @@ -0,0 +1,393 @@ +using System.Runtime.CompilerServices; +using System.Threading.Channels; +using AvParser.Core.Collecting; +using AvParser.Core.Parsing; +using AvParser.Core.Settings; +using Microsoft.Extensions.Logging; + +namespace AvParser.Infrastructure.Collecting; + +/// +/// Runs a source end to end: discover, skip what is already settled, download, store. +/// +/// +/// +/// Discovery and downloading run at completely different speeds — a listing page arrives in one +/// request while its contents take a second each — so they are decoupled by a bounded channel. +/// Bounded rather than unbounded on purpose: a listing of two hundred thousand items must not +/// materialise in memory just because the workers are slower than the source. +/// +/// +/// The runner owns its workers and waits for them before returning, including when the user +/// cancels. Without that a stopped run keeps writing to the store after the page has said it +/// stopped, which is the classic bug in this shape. +/// +/// +public sealed class CollectRunner( + IMediaFetcher fetcher, + IMediaStore store, + ISettingsService settings, + ILogger logger +) : ICollectRunner +{ + /// How many addresses to check against the journal in one query. + private const int JournalBatch = 200; + + /// Report progress at least this often, however slow the items are. + private static readonly TimeSpan ProgressInterval = TimeSpan.FromMilliseconds(500); + + private readonly IMediaFetcher _fetcher = fetcher ?? throw new ArgumentNullException(nameof(fetcher)); + private readonly IMediaStore _store = store ?? throw new ArgumentNullException(nameof(store)); + private readonly ISettingsService _settings = settings ?? throw new ArgumentNullException(nameof(settings)); + private readonly ILogger _logger = logger ?? throw new ArgumentNullException(nameof(logger)); + + /// + public async IAsyncEnumerable> RunAsync( + IMediaSource source, + MediaQuery query, + CollectOptions options, + IProgress? progress, + [EnumeratorCancellation] CancellationToken cancellationToken + ) + { + ArgumentNullException.ThrowIfNull(source); + ArgumentNullException.ThrowIfNull(query); + ArgumentNullException.ThrowIfNull(options); + + var workers = Math.Clamp(options.MaxConcurrentDownloads, 1, 32); + var runId = await _store.BeginRunAsync(source.Id, cancellationToken).ConfigureAwait(false); + var tombstones = await _store.LoadTombstonesAsync(cancellationToken).ConfigureAwait(false); + var fetchOptions = BuildFetchOptions(tombstones); + + var work = Channel.CreateBounded( + new BoundedChannelOptions(workers * 4) { FullMode = BoundedChannelFullMode.Wait } + ); + var results = Channel.CreateUnbounded>(); + + using var run = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken); + var counters = new Counters(); + + var producer = Task.Run( + () => ProduceAsync(source, query, options, work.Writer, results.Writer, counters, progress, run.Token), + CancellationToken.None + ); + + var consumers = Enumerable + .Range(0, workers) + .Select(_ => + Task.Run( + () => ConsumeAsync(source, runId, fetchOptions, work.Reader, results.Writer, counters, run.Token), + CancellationToken.None + ) + ) + .ToArray(); + + var completion = Task.Run( + async () => + { + try + { + await producer.ConfigureAwait(false); + await Task.WhenAll(consumers).ConfigureAwait(false); + } + finally + { + results.Writer.TryComplete(); + } + }, + CancellationToken.None + ); + + var cancelled = false; + + try + { + await foreach (var outcome in results.Reader.ReadAllAsync(cancellationToken).ConfigureAwait(false)) + { + yield return outcome; + } + } + finally + { + cancelled = cancellationToken.IsCancellationRequested; + + // Stop anything still in flight, then wait for it. Returning while a worker is still + // writing to the store would leave the page reporting a finished run that is not. + await run.CancelAsync().ConfigureAwait(false); + work.Writer.TryComplete(); + + try + { + await completion.ConfigureAwait(false); + } + catch (Exception ex) when (ex is OperationCanceledException or ChannelClosedException) + { + // Expected on the cancellation path. + } + + await _store + .CompleteRunAsync(runId, counters.ToSummary(cancelled), CancellationToken.None) + .ConfigureAwait(false); + + _logger.LogInformation( + "Collected from {Source}: {Stored} new, {Duplicates} duplicate, {Skipped} skipped, {Failed} failed", + source.Id, + counters.Stored, + counters.Duplicates, + counters.Skipped, + counters.Failed + ); + } + } + + /// Reads the source and feeds the workers, skipping what the journal already settled. + private async Task ProduceAsync( + IMediaSource source, + MediaQuery query, + CollectOptions options, + ChannelWriter work, + ChannelWriter> results, + Counters counters, + IProgress? progress, + CancellationToken token + ) + { + var batch = new List(JournalBatch); + + try + { + await foreach ( + var discovered in source.ParseAsync(query, null, token).WithCancellation(token).ConfigureAwait(false) + ) + { + if (!discovered.IsSuccess) + { + counters.Fail(); + await results + .WriteAsync(ParseOutcome.Failure(discovered.Error), token) + .ConfigureAwait(false); + continue; + } + + counters.Discover(); + batch.Add(discovered.Value!); + + if (batch.Count >= JournalBatch) + { + await DispatchAsync(source, options, batch, work, results, counters, token).ConfigureAwait(false); + } + + progress?.Report(new ParseProgress(counters.Processed, 0)); + } + + await DispatchAsync(source, options, batch, work, results, counters, token).ConfigureAwait(false); + } + catch (OperationCanceledException) + { + // The run was stopped; nothing further to discover. + } + catch (Exception ex) + { + _logger.LogError(ex, "The source {Source} failed while listing", source.Id); + await results + .WriteAsync( + ParseOutcome.Failure(ParseError.Create(0, "SourceFailed", ex.Message, ex.Message)), + CancellationToken.None + ) + .ConfigureAwait(false); + } + finally + { + work.TryComplete(); + } + } + + /// Checks a batch against the journal and queues whatever still needs fetching. + private async Task DispatchAsync( + IMediaSource source, + CollectOptions options, + List batch, + ChannelWriter work, + ChannelWriter> results, + Counters counters, + CancellationToken token + ) + { + if (batch.Count == 0) + { + return; + } + + var seen = options.ForceRefetch + ? new Dictionary(StringComparer.Ordinal) + : await _store + .GetSeenAsync(source.Id, [.. batch.Select(c => c.Url.AbsoluteUri)], token) + .ConfigureAwait(false); + + foreach (var candidate in batch) + { + if (seen.TryGetValue(candidate.Url.AbsoluteUri, out var outcome) && SeenOutcomes.IsTerminal(outcome)) + { + // Settled by a previous run: no request, no bytes, no proxy. + counters.Skip(); + await results + .WriteAsync( + ParseOutcome.Success( + new CollectedItem(candidate, Placeholder, CollectStatus.Skipped) + ), + token + ) + .ConfigureAwait(false); + + continue; + } + + await work.WriteAsync(candidate, token).ConfigureAwait(false); + } + + batch.Clear(); + } + + /// Downloads and stores whatever the producer queues. + private async Task ConsumeAsync( + IMediaSource source, + string runId, + FetchOptions fetchOptions, + ChannelReader work, + ChannelWriter> results, + Counters counters, + CancellationToken token + ) + { + try + { + await foreach (var candidate in work.ReadAllAsync(token).ConfigureAwait(false)) + { + var outcome = await CollectOneAsync(source, runId, candidate, fetchOptions, counters, token) + .ConfigureAwait(false); + + await results.WriteAsync(outcome, token).ConfigureAwait(false); + } + } + catch (OperationCanceledException) + { + // Stopped mid-run; the staged file has already been cleaned up by the fetcher. + } + } + + private async Task> CollectOneAsync( + IMediaSource source, + string runId, + MediaCandidate candidate, + FetchOptions fetchOptions, + Counters counters, + CancellationToken token + ) + { + var result = await _fetcher.FetchAsync(candidate, fetchOptions, token).ConfigureAwait(false); + + if (!result.IsSuccess || result.TempPath is null || result.Blob is null) + { + counters.Fail(); + + await _store + .RecordSeenAsync( + source.Id, + candidate.Url.AbsoluteUri, + result.Outcome, + result.HttpStatus == 0 ? null : result.HttpStatus, + result.ErrorCode, + token + ) + .ConfigureAwait(false); + + return ParseOutcome.Failure( + ParseError.Create( + candidate.Ordinal, + result.ErrorCode ?? "Failed", + result.ErrorDetail ?? candidate.Url.AbsoluteUri, + result.ErrorDetail ?? candidate.Url.AbsoluteUri + ) + ); + } + + var request = new MediaStoreRequest( + candidate, + result.Blob, + result.TempPath, + runId, + result.FinalUrl, + result.HttpStatus + ) + { + ContentType = result.ContentType, + ProxyKey = result.ProxyKey, + }; + + var stored = await _store.StoreAsync(request, token).ConfigureAwait(false); + + if (stored.Status == CollectStatus.Stored) + { + counters.Store(result.Blob.Length); + } + else + { + counters.Duplicate(); + } + + return ParseOutcome.Success(stored with { Elapsed = result.Elapsed }); + } + + private FetchOptions BuildFetchOptions(IReadOnlySet tombstones) + { + var current = _settings.Current; + + return new FetchOptions + { + // Mirrors the parser gate: proxy-only unless the user has said direct is acceptable. + RequireProxy = !current.AllowDirectConnection, + Tombstones = tombstones, + }; + } + + /// Stands in for content on a skipped item, which by definition has none. + private static MediaBlob Placeholder { get; } = MediaBlob.Create(new string('0', 64), MediaKind.Unknown, 0); + + /// Run totals, written from several workers at once. + private sealed class Counters + { + private int _discovered; + private int _stored; + private int _duplicates; + private int _skipped; + private int _failed; + private long _bytes; + + public int Stored => Volatile.Read(ref _stored); + + public int Duplicates => Volatile.Read(ref _duplicates); + + public int Skipped => Volatile.Read(ref _skipped); + + public int Failed => Volatile.Read(ref _failed); + + public int Processed => Stored + Duplicates + Skipped + Failed; + + public void Discover() => Interlocked.Increment(ref _discovered); + + public void Skip() => Interlocked.Increment(ref _skipped); + + public void Fail() => Interlocked.Increment(ref _failed); + + public void Store(long bytes) + { + Interlocked.Increment(ref _stored); + Interlocked.Add(ref _bytes, bytes); + } + + public void Duplicate() => Interlocked.Increment(ref _duplicates); + + public RunSummary ToSummary(bool cancelled) => + new(Volatile.Read(ref _discovered), Stored, Duplicates, Failed, Interlocked.Read(ref _bytes), cancelled); + } +} diff --git a/src/AvParser.Infrastructure/Collecting/OwnServiceSource.cs b/src/AvParser.Infrastructure/Collecting/OwnServiceSource.cs new file mode 100644 index 0000000..43a4230 --- /dev/null +++ b/src/AvParser.Infrastructure/Collecting/OwnServiceSource.cs @@ -0,0 +1,293 @@ +using System.Globalization; +using System.Runtime.CompilerServices; +using System.Text.Json; +using AvParser.Core.Collecting; +using AvParser.Core.Parsing; +using AvParser.Core.Settings; +using AvParser.Infrastructure.Proxies; +using Microsoft.Extensions.Logging; + +namespace AvParser.Infrastructure.Collecting; + +/// +/// Lists what a service you run holds, by asking it. +/// +/// +/// +/// A service that knows its own contents can simply say so, which is why this is a listing client +/// and not a crawler: one request per page returns exactly what is there, in order, with metadata, +/// and stops when the pages run out. +/// +/// +/// The expected response is deliberately loose, because the service on the other end is the user's +/// own and should not have to be rewritten to match us. Either of these works: +/// +/// +/// { "items": [ { "url": "...", "id": "...", "name": "...", "published": "...", "size": 1234, +/// "tags": ["a"] } ], "next": "cursor" } +/// [ "https://host/one.png", "https://host/two.gif" ] +/// +/// +/// Paging follows next until it is absent. A page that repeats a cursor stops the walk +/// rather than looping for ever. +/// +/// +public sealed class OwnServiceSource( + IProxiedHttpClientFactory clients, + ISettingsService settings, + ILogger logger +) : IMediaSource +{ + /// Stops a service that keeps handing back pages from running the collector for ever. + private const int MaxPages = 10_000; + + private readonly IProxiedHttpClientFactory _clients = clients ?? throw new ArgumentNullException(nameof(clients)); + private readonly ISettingsService _settings = settings ?? throw new ArgumentNullException(nameof(settings)); + private readonly ILogger _logger = logger ?? throw new ArgumentNullException(nameof(logger)); + + /// + public string Id => "own-service"; + + /// + public string DisplayName => "Own service"; + + /// + public string Description => "Reads the listing endpoint of a service you run."; + + /// + public bool RequiresNetwork => true; + + /// + public bool CanParse(MediaQuery input) => + input?.Endpoint is { IsAbsoluteUri: true } endpoint && endpoint.Scheme is "http" or "https"; + + /// + public async IAsyncEnumerable> ParseAsync( + MediaQuery input, + IProgress? progress, + [EnumeratorCancellation] CancellationToken cancellationToken + ) + { + ArgumentNullException.ThrowIfNull(input); + + if (!CanParse(input)) + { + yield return ParseOutcome.Failure( + ParseError.Create(0, "NoEndpoint", "No listing endpoint was configured.") + ); + yield break; + } + + var requireProxy = !_settings.Current.AllowDirectConnection; + + using var leased = await _clients + .LeaseAsync(HttpClientTimeouts.Default, requireProxy, cancellationToken) + .ConfigureAwait(false); + + var cursor = input.Cursor; + var seenCursors = new HashSet(StringComparer.Ordinal); + var ordinal = 0; + + for (var page = 0; page < MaxPages; page++) + { + cancellationToken.ThrowIfCancellationRequested(); + + var url = WithCursor(input.Endpoint!, cursor); + + // Read and parse inside the try, report outside it: a yield cannot live in a catch. + Page parsed = default; + ParseError? error = null; + + try + { + var body = await leased.Client.GetStringAsync(url, cancellationToken).ConfigureAwait(false); + leased.Lease?.ReportSuccess(); + + parsed = ReadPage(body, input.Endpoint!); + } + catch (Exception ex) when (ex is HttpRequestException or IOException) + { + leased.Lease?.ReportFailure(ex.Message); + _logger.LogWarning(ex, "Could not read the listing at {Url}", url); + error = ParseError.Create(ordinal, "ListingFailed", ex.Message, ex.Message); + } + catch (JsonException ex) + { + // The transport was fine; the service answered with something we cannot read. + error = ParseError.Create(ordinal, "ListingMalformed", ex.Message, ex.Message); + } + + if (error is not null) + { + yield return ParseOutcome.Failure(error); + yield break; + } + + foreach (var candidate in parsed.Items) + { + ordinal++; + + yield return ParseOutcome.Success(candidate with { SourceId = Id, Ordinal = ordinal }); + + if (input.HasLimit && ordinal >= input.Limit) + { + yield break; + } + } + + // Total is unknown until the last page, so progress stays indeterminate and only the + // running count moves. + progress?.Report(new ParseProgress(ordinal, 0)); + + if (string.IsNullOrEmpty(parsed.Next) || !seenCursors.Add(parsed.Next)) + { + yield break; + } + + cursor = parsed.Next; + } + + _logger.LogWarning("Stopped paging {Endpoint} after {Pages} pages", input.Endpoint, MaxPages); + } + + /// One page of a listing. + /// Candidates on this page. + /// Cursor for the following page, or null when this was the last. + internal readonly record struct Page(IReadOnlyList Items, string? Next); + + private static Uri WithCursor(Uri endpoint, string? cursor) + { + if (string.IsNullOrEmpty(cursor)) + { + return endpoint; + } + + var separator = string.IsNullOrEmpty(endpoint.Query) ? '?' : '&'; + + return new Uri($"{endpoint.AbsoluteUri}{separator}cursor={Uri.EscapeDataString(cursor)}"); + } + + /// + /// Reads either shape of listing. + /// + /// + /// Hand-parsed with rather than deserialised into a type: the two + /// accepted shapes and the string-or-object item would need a converter each, and being lenient + /// about a contract the user controls is the whole point. + /// + internal static Page ReadPage(string json, Uri baseAddress) + { + using var document = JsonDocument.Parse(json); + var root = document.RootElement; + + var items = new List(); + string? next = null; + + var array = root.ValueKind switch + { + JsonValueKind.Array => root, + JsonValueKind.Object when root.TryGetProperty("items", out var found) => found, + _ => default, + }; + + if (root.ValueKind == JsonValueKind.Object && root.TryGetProperty("next", out var cursor)) + { + next = cursor.ValueKind == JsonValueKind.String ? cursor.GetString() : null; + } + + if (array.ValueKind != JsonValueKind.Array) + { + return new Page(items, next); + } + + foreach (var element in array.EnumerateArray()) + { + if (ReadItem(element, baseAddress) is { } candidate) + { + items.Add(candidate); + } + } + + return new Page(items, next); + } + + private static MediaCandidate? ReadItem(JsonElement element, Uri baseAddress) + { + if (element.ValueKind == JsonValueKind.String) + { + return Resolve(element.GetString(), baseAddress) is { } bare ? new MediaCandidate(bare) : null; + } + + if (element.ValueKind != JsonValueKind.Object) + { + return null; + } + + var url = Resolve(ReadString(element, "url") ?? ReadString(element, "href"), baseAddress); + + if (url is null) + { + return null; + } + + var candidate = new MediaCandidate(url) + { + ExternalId = ReadString(element, "id"), + SuggestedName = ReadString(element, "name") ?? ReadString(element, "filename"), + }; + + if ( + ReadString(element, "published") is { } published + && DateTimeOffset.TryParse( + published, + CultureInfo.InvariantCulture, + DateTimeStyles.RoundtripKind, + out var when + ) + ) + { + candidate = candidate with { PublishedUtc = when }; + } + + if (element.TryGetProperty("size", out var size) && size.TryGetInt64(out var length)) + { + candidate = candidate with { ExpectedLength = length }; + } + + if (element.TryGetProperty("tags", out var tags) && tags.ValueKind == JsonValueKind.Array) + { + candidate = candidate with + { + Tags = + [ + .. tags.EnumerateArray() + .Where(tag => tag.ValueKind == JsonValueKind.String) + .Select(tag => tag.GetString()!), + ], + }; + } + + return candidate; + } + + private static string? ReadString(JsonElement element, string name) => + element.TryGetProperty(name, out var value) && value.ValueKind == JsonValueKind.String + ? value.GetString() + : null; + + /// Resolves a listed address, allowing relative paths against the endpoint. + private static Uri? Resolve(string? value, Uri baseAddress) + { + if (string.IsNullOrWhiteSpace(value)) + { + return null; + } + + if (!Uri.TryCreate(baseAddress, value.Trim(), out var resolved)) + { + return null; + } + + return resolved.Scheme is "http" or "https" ? resolved : null; + } +} diff --git a/src/AvParser.Infrastructure/DependencyInjection/InfrastructureServiceCollectionExtensions.cs b/src/AvParser.Infrastructure/DependencyInjection/InfrastructureServiceCollectionExtensions.cs index 4f623a4..bf2c82b 100644 --- a/src/AvParser.Infrastructure/DependencyInjection/InfrastructureServiceCollectionExtensions.cs +++ b/src/AvParser.Infrastructure/DependencyInjection/InfrastructureServiceCollectionExtensions.cs @@ -1,5 +1,9 @@ +using AvParser.Core.Collecting; +using AvParser.Core.DependencyInjection; using AvParser.Core.Proxies; using AvParser.Core.Settings; +using AvParser.Infrastructure.Collecting; +using AvParser.Infrastructure.Media; using AvParser.Infrastructure.Proxies; using AvParser.Infrastructure.Settings; using AvParser.Infrastructure.Storage; @@ -34,6 +38,54 @@ public static class InfrastructureServiceCollectionExtensions )); services.AddAvParserProxies(); + services.AddAvParserCollecting(); + + return services; + } + + /// Registers the media store, the download pipeline and the network sources. + /// + /// Everything here is a singleton because everything here owns something shared: a database + /// connection pool, a per-host throttle whose whole purpose is being common to all workers, and + /// a blob directory that must have exactly one owner deciding what is complete. + /// + public static IServiceCollection AddAvParserCollecting(this IServiceCollection services) + { + ArgumentNullException.ThrowIfNull(services); + + services.AddSingleton(); + services.AddSingleton(); + + // Explicit factories: both take an optional parameter, and letting the container pick a + // constructor by registration order is how that goes wrong quietly. + services.AddSingleton(sp => new ShowcaseLinker( + sp.GetRequiredService(), + sp.GetRequiredService(), + sp.GetRequiredService>() + )); + + services.AddSingleton(); + services.AddSingleton(sp => sp.GetRequiredService()); + + // Two concurrent requests per origin, a quarter of a second apart. These become settings in + // their own right; until then the defaults are the polite ones. + services.AddSingleton(sp => new HostThrottle( + maxConcurrentPerHost: 2, + minimumInterval: TimeSpan.FromMilliseconds(250), + sp.GetRequiredService>() + )); + + services.AddSingleton(); + services.AddSingleton(); + + services.AddSingleton(); + + // Resolves sources from both assemblies: the container gathers every IMediaSource + // registration regardless of which project declared it. + services.AddSingleton(sp => new MediaSourceCatalog( + sp.GetServices(), + CoreServiceCollectionExtensions.DefaultMediaSourceId + )); return services; } diff --git a/tests/AvParser.Core.Tests/Collecting/UrlListSourceTests.cs b/tests/AvParser.Core.Tests/Collecting/UrlListSourceTests.cs new file mode 100644 index 0000000..82d38b3 --- /dev/null +++ b/tests/AvParser.Core.Tests/Collecting/UrlListSourceTests.cs @@ -0,0 +1,198 @@ +using AvParser.Core.Collecting; +using AvParser.Core.Collecting.Sources; +using AvParser.Core.Parsing; + +namespace AvParser.Core.Tests.Collecting; + +public class UrlListSourceTests +{ + private static readonly IMediaSource Source = new UrlListSource(); + + private static async Task>> RunAsync(string text, int limit = 0) + { + var results = new List>(); + + await foreach ( + var outcome in Source.ParseAsync( + new MediaQuery(text, Limit: limit), + null, + TestContext.Current.CancellationToken + ) + ) + { + results.Add(outcome); + } + + return results; + } + + [Fact] + public async Task Each_line_becomes_a_candidate() + { + var results = await RunAsync("https://example.test/a.png\nhttps://example.test/b.gif"); + + results.Count.ShouldBe(2); + results.ShouldAllBe(r => r.IsSuccess); + results[0].Value!.Url.AbsoluteUri.ShouldBe("https://example.test/a.png"); + results[1].Value!.Ordinal.ShouldBe(2); + } + + [Fact] + public async Task Blank_lines_and_comments_are_ignored() + { + var results = await RunAsync( + """ + # my list + https://example.test/a.png + + # indented comment + https://example.test/b.png + """ + ); + + results.Count.ShouldBe(2); + } + + [Fact] + public async Task Whitespace_around_an_address_is_forgiven() + { + var results = await RunAsync(" https://example.test/a.png "); + + results.ShouldHaveSingleItem().IsSuccess.ShouldBeTrue(); + } + + [Fact] + public async Task A_bad_line_is_named_rather_than_dropped() + { + // In a paste of two hundred addresses a silently skipped typo is unfindable. + var results = await RunAsync("https://example.test/a.png\nnot an address\nhttps://example.test/b.png"); + + results.Count.ShouldBe(3); + results[1].IsSuccess.ShouldBeFalse(); + results[1].Error!.Code.ShouldBe("NotAnAddress"); + results[1].Error!.Arguments.ShouldContain("not an address"); + } + + [Theory] + [InlineData("file:///etc/passwd")] + [InlineData("data:image/png;base64,AAAA")] + [InlineData("ftp://example.test/a.png")] + [InlineData("javascript:alert(1)")] + public async Task Only_http_addresses_are_accepted(string line) + { + // A pasted list is as likely to have come from somewhere else as to have been typed. + var results = await RunAsync(line); + + results.ShouldHaveSingleItem().IsSuccess.ShouldBeFalse(); + } + + [Fact] + public async Task A_limit_stops_the_listing_early() + { + var text = string.Join('\n', Enumerable.Range(0, 50).Select(i => $"https://example.test/{i}.png")); + + var results = await RunAsync(text, limit: 5); + + results.Count(r => r.IsSuccess).ShouldBe(5); + } + + [Fact] + public void The_source_needs_no_network_of_its_own() + { + // It discovers nothing; downloading what it read is the fetcher's job and is gated there. + Source.RequiresNetwork.ShouldBeFalse(); + } + + [Theory] + [InlineData("https://example.test/a.png", true)] + [InlineData("# only a comment", false)] + [InlineData("", false)] + [InlineData("nonsense", false)] + public void CanParse_answers_without_doing_any_work(string text, bool expected) => + Source.CanParse(new MediaQuery(text)).ShouldBe(expected); + + [Fact] + public async Task Cancellation_is_honoured() + { + using var cancellation = new CancellationTokenSource(); + await cancellation.CancelAsync(); + + await Should.ThrowAsync(async () => + { + await foreach (var _ in Source.ParseAsync(new MediaQuery("https://a.test/x.png"), null, cancellation.Token)) + { + // Draining is the point; the first move must already throw. + } + }); + } +} + +public class MediaSourceCatalogTests +{ + [Fact] + public void Sources_are_ordered_by_display_name() + { + var catalog = new MediaSourceCatalog([new FakeSource("z", "Zebra"), new FakeSource("a", "Aardvark")]); + + catalog.Sources.Select(s => s.Id).ShouldBe(["a", "z"]); + } + + [Fact] + public void The_named_default_wins_over_alphabetical_order() + { + // Otherwise the landing page depends on a display name, and would open on the network + // source — behind the proxy gate — before the user has asked for anything. + var catalog = new MediaSourceCatalog([new FakeSource("a", "Aardvark"), new FakeSource("z", "Zebra")], "z"); + + catalog.DefaultSource.Id.ShouldBe("z"); + } + + [Fact] + public void An_unknown_default_falls_back_rather_than_throwing() + { + var catalog = new MediaSourceCatalog([new FakeSource("a", "Aardvark")], "removed-in-a-past-version"); + + catalog.DefaultSource.Id.ShouldBe("a"); + } + + [Fact] + public void Lookup_ignores_case_and_reports_a_miss() + { + IMediaSourceCatalog catalog = new MediaSourceCatalog([new FakeSource("url-list", "URL list")]); + + catalog.Find("URL-LIST").ShouldNotBeNull(); + catalog.Find("nope").ShouldBeNull(); + catalog.FindOrDefault("nope").Id.ShouldBe("url-list"); + } + + [Fact] + public void An_empty_registration_is_rejected() => + Should.Throw(() => new MediaSourceCatalog([])); + + [Fact] + public void Two_sources_sharing_an_id_are_rejected() => + Should.Throw(() => + new MediaSourceCatalog([new FakeSource("same", "One"), new FakeSource("same", "Two")]) + ); + + private sealed class FakeSource(string id, string name) : IMediaSource + { + public string Id => id; + + public string DisplayName => name; + + public string Description => string.Empty; + + public bool CanParse(MediaQuery input) => true; + + public async IAsyncEnumerable> ParseAsync( + MediaQuery input, + IProgress? progress, + [System.Runtime.CompilerServices.EnumeratorCancellation] CancellationToken cancellationToken + ) + { + await Task.Yield(); + yield break; + } + } +} diff --git a/tests/AvParser.Infrastructure.Tests/Collecting/CollectRunnerTests.cs b/tests/AvParser.Infrastructure.Tests/Collecting/CollectRunnerTests.cs new file mode 100644 index 0000000..bc77bae --- /dev/null +++ b/tests/AvParser.Infrastructure.Tests/Collecting/CollectRunnerTests.cs @@ -0,0 +1,416 @@ +using System.Runtime.CompilerServices; +using System.Security.Cryptography; +using System.Text; +using AvParser.Core.Collecting; +using AvParser.Core.Parsing; +using AvParser.Core.Settings; +using AvParser.Infrastructure.Collecting; +using AvParser.Infrastructure.Media; +using AvParser.Infrastructure.Storage; +using Microsoft.Extensions.Logging.Abstractions; +using ReactiveUI.Primitives.Signals; + +namespace AvParser.Infrastructure.Tests.Collecting; + +/// A source that hands back exactly the addresses the test names. +internal sealed class ListSource(params string[] urls) : IMediaSource +{ + public string Id => "test-source"; + + public string DisplayName => "Test source"; + + public string Description => string.Empty; + + public int Listings { get; private set; } + + public bool CanParse(MediaQuery input) => true; + + public async IAsyncEnumerable> ParseAsync( + MediaQuery input, + IProgress? progress, + [EnumeratorCancellation] CancellationToken cancellationToken + ) + { + Listings++; + + for (var index = 0; index < urls.Length; index++) + { + cancellationToken.ThrowIfCancellationRequested(); + await Task.Yield(); + + yield return ParseOutcome.Success( + new MediaCandidate(new Uri(urls[index])) { SourceId = Id, Ordinal = index + 1 } + ); + } + } +} + +/// A fetcher that stages bytes from a table instead of using a network. +internal sealed class ScriptedFetcher(BlobStore blobs) : IMediaFetcher +{ + public Dictionary Content { get; } = new(StringComparer.Ordinal); + + public HashSet Failing { get; } = new(StringComparer.Ordinal); + + public List Fetched { get; } = []; + + public TimeSpan Delay { get; set; } + + public async Task FetchAsync( + MediaCandidate candidate, + FetchOptions options, + CancellationToken cancellationToken = default + ) + { + var url = candidate.Url.AbsoluteUri; + + lock (Fetched) + { + Fetched.Add(url); + } + + if (Delay > TimeSpan.Zero) + { + await Task.Delay(Delay, cancellationToken); + } + + cancellationToken.ThrowIfCancellationRequested(); + + if (Failing.Contains(url) || !Content.TryGetValue(url, out var bytes)) + { + return new FetchResult(SeenOutcome.Failed, candidate.Url) { ErrorCode = "RequestFailed", HttpStatus = 500 }; + } + + var temp = blobs.CreateTempPath(); + await File.WriteAllBytesAsync(temp, bytes, cancellationToken); + + var hash = Convert.ToHexStringLower(SHA256.HashData(bytes)); + + return new FetchResult(SeenOutcome.Stored, candidate.Url) + { + Blob = MediaBlob.Create(hash, MediaKind.Png, bytes.Length), + TempPath = temp, + HttpStatus = 200, + }; + } +} + +/// In-memory settings, so the runner never reads the developer's real profile. +internal sealed class FixedSettings(AppSettings? initial = null) : ISettingsService, IDisposable +{ + private readonly BehaviorSignal _current = new(initial ?? new AppSettings()); + + public AppSettings Current => _current.Value; + + public IObservable Changes => _current; + + public void Update(Func mutate) => _current.OnNext(mutate(_current.Value)); + + public Task FlushAsync(CancellationToken cancellationToken = default) => Task.CompletedTask; + + public void Dispose() => _current.Dispose(); +} + +public sealed class CollectRunnerTests : IAsyncLifetime +{ + private readonly string _root = Path.Combine(Path.GetTempPath(), "AvParserTests", Guid.NewGuid().ToString("N")); + + private AppPaths _paths = null!; + private SqliteMediaIndex _index = null!; + private BlobStore _blobs = null!; + private MediaStore _store = null!; + private ScriptedFetcher _fetcher = null!; + private FixedSettings _settings = null!; + private CollectRunner _runner = null!; + + public async ValueTask InitializeAsync() + { + _paths = new AppPaths(_root); + _paths.EnsureCreated(); + + _index = new SqliteMediaIndex(_paths, NullLogger.Instance); + _blobs = new BlobStore(_paths, NullLogger.Instance); + _store = new MediaStore( + _index, + _blobs, + new ShowcaseLinker(_paths, _blobs, NullLogger.Instance), + NullLogger.Instance + ); + + // Direct is allowed here: these tests are about the runner, not the proxy gate. + _settings = new FixedSettings(new AppSettings { AllowDirectConnection = true }); + _fetcher = new ScriptedFetcher(_blobs); + _runner = new CollectRunner(_fetcher, _store, _settings, NullLogger.Instance); + + await _store.InitialiseAsync(TestContext.Current.CancellationToken); + } + + public ValueTask DisposeAsync() + { + _settings.Dispose(); + _index.Dispose(); + Microsoft.Data.Sqlite.SqliteConnection.ClearAllPools(); + + if (Directory.Exists(_root)) + { + try + { + Directory.Delete(_root, recursive: true); + } + catch (IOException) + { + // Not worth failing a green test over. + } + } + + return ValueTask.CompletedTask; + } + + private static byte[] Image(string seed) => [.. Samples.Png(), .. Encoding.UTF8.GetBytes(seed)]; + + private async Task>> RunAsync(IMediaSource source, CollectOptions? options = null) + { + var results = new List>(); + + await foreach ( + var outcome in _runner.RunAsync( + source, + new MediaQuery(), + options ?? new CollectOptions(), + null, + TestContext.Current.CancellationToken + ) + ) + { + results.Add(outcome); + } + + return results; + } + + [Fact] + public async Task Everything_the_source_finds_is_downloaded_and_stored() + { + var source = new ListSource("https://a.test/1.png", "https://a.test/2.png"); + _fetcher.Content["https://a.test/1.png"] = Image("one"); + _fetcher.Content["https://a.test/2.png"] = Image("two"); + + var results = await RunAsync(source); + + results.Count.ShouldBe(2); + results.ShouldAllBe(r => r.IsSuccess); + results.ShouldAllBe(r => r.Value!.Status == CollectStatus.Stored); + + var stats = await _store.GetStatsAsync(TestContext.Current.CancellationToken); + stats.BlobCount.ShouldBe(2); + } + + [Fact] + public async Task A_second_run_over_the_same_list_fetches_nothing() + { + // The journal exists precisely so that re-running a list costs no requests and no proxy. + var source = new ListSource("https://a.test/1.png", "https://a.test/2.png"); + _fetcher.Content["https://a.test/1.png"] = Image("one"); + _fetcher.Content["https://a.test/2.png"] = Image("two"); + + await RunAsync(source); + _fetcher.Fetched.Clear(); + + var second = await RunAsync(source); + + _fetcher.Fetched.ShouldBeEmpty(); + second.Count.ShouldBe(2); + second.ShouldAllBe(r => r.Value!.Status == CollectStatus.Skipped); + } + + [Fact] + public async Task Forcing_a_refetch_ignores_the_journal() + { + var source = new ListSource("https://a.test/1.png"); + _fetcher.Content["https://a.test/1.png"] = Image("one"); + + await RunAsync(source); + _fetcher.Fetched.Clear(); + + await RunAsync(source, new CollectOptions { ForceRefetch = true }); + + _fetcher.Fetched.ShouldHaveSingleItem(); + } + + [Fact] + public async Task A_failure_is_reported_and_retried_next_time() + { + // Failures describe the moment, not the resource; a flaky network must not lose content. + var source = new ListSource("https://a.test/1.png"); + _fetcher.Failing.Add("https://a.test/1.png"); + + var first = await RunAsync(source); + first.ShouldHaveSingleItem().IsSuccess.ShouldBeFalse(); + + _fetcher.Failing.Clear(); + _fetcher.Content["https://a.test/1.png"] = Image("recovered"); + _fetcher.Fetched.Clear(); + + var second = await RunAsync(source); + + _fetcher.Fetched.ShouldHaveSingleItem(); + second.ShouldHaveSingleItem().Value!.Status.ShouldBe(CollectStatus.Stored); + } + + [Fact] + public async Task The_same_bytes_at_two_addresses_are_stored_once() + { + var source = new ListSource("https://a.test/1.png", "https://a.test/2.png"); + var shared = Image("identical"); + _fetcher.Content["https://a.test/1.png"] = shared; + _fetcher.Content["https://a.test/2.png"] = shared; + + var results = await RunAsync(source, new CollectOptions { MaxConcurrentDownloads = 1 }); + + results.Count(r => r.Value!.Status == CollectStatus.Stored).ShouldBe(1); + results.Count(r => r.Value!.Status == CollectStatus.Duplicate).ShouldBe(1); + (await _store.GetStatsAsync(TestContext.Current.CancellationToken)).BlobCount.ShouldBe(1); + } + + [Fact] + public async Task A_source_that_throws_does_not_take_the_run_down() + { + var results = await RunAsync(new ThrowingSource()); + + results.ShouldHaveSingleItem().Error!.Code.ShouldBe("SourceFailed"); + } + + [Fact] + public async Task Cancelling_stops_the_run_and_nothing_writes_afterwards() + { + // The runner owns its workers and waits for them; otherwise a stopped run keeps writing to + // the store after the page has already said it stopped. + var urls = Enumerable.Range(0, 40).Select(i => $"https://a.test/{i}.png").ToArray(); + var source = new ListSource(urls); + + foreach (var url in urls) + { + _fetcher.Content[url] = Image(url); + } + + _fetcher.Delay = TimeSpan.FromMilliseconds(50); + + using var cancellation = new CancellationTokenSource(); + var results = new List>(); + + await Should.ThrowAsync(async () => + { + await foreach ( + var outcome in _runner.RunAsync( + source, + new MediaQuery(), + new CollectOptions(), + null, + cancellation.Token + ) + ) + { + results.Add(outcome); + + if (results.Count == 3) + { + await cancellation.CancelAsync(); + } + } + }); + + var afterReturn = await _store.GetStatsAsync(TestContext.Current.CancellationToken); + await Task.Delay(200, TestContext.Current.CancellationToken); + var later = await _store.GetStatsAsync(TestContext.Current.CancellationToken); + + later.ItemCount.ShouldBe(afterReturn.ItemCount); + Directory.EnumerateFiles(_paths.MediaTempDirectory).ShouldBeEmpty(); + } + + private sealed class ThrowingSource : IMediaSource + { + public string Id => "throwing"; + + public string DisplayName => "Throwing source"; + + public string Description => string.Empty; + + public bool CanParse(MediaQuery input) => true; + + public async IAsyncEnumerable> ParseAsync( + MediaQuery input, + IProgress? progress, + [EnumeratorCancellation] CancellationToken cancellationToken + ) + { + await Task.Yield(); + throw new InvalidOperationException("the listing endpoint moved"); +#pragma warning disable CS0162 + yield break; +#pragma warning restore CS0162 + } + } +} + +public class OwnServiceListingTests +{ + private static readonly Uri Endpoint = new("https://own.test/api/list"); + + [Fact] + public void An_object_with_items_is_read() + { + var page = OwnServiceSource.ReadPage( + """ + { "items": [ { "url": "https://own.test/a.png", "id": "42", "name": "kitten", + "published": "2026-08-13T10:00:00Z", "size": 4096, "tags": ["cats"] } ], + "next": "page2" } + """, + Endpoint + ); + + var item = page.Items.ShouldHaveSingleItem(); + item.Url.AbsoluteUri.ShouldBe("https://own.test/a.png"); + item.ExternalId.ShouldBe("42"); + item.SuggestedName.ShouldBe("kitten"); + item.ExpectedLength.ShouldBe(4096); + item.Tags.ShouldBe(["cats"]); + page.Next.ShouldBe("page2"); + } + + [Fact] + public void A_bare_array_of_addresses_is_read() + { + // The service on the other end is the user's own; it should not have to be rewritten to + // match a schema we invented. + var page = OwnServiceSource.ReadPage("""["https://own.test/a.png", "https://own.test/b.gif"]""", Endpoint); + + page.Items.Count.ShouldBe(2); + page.Next.ShouldBeNull(); + } + + [Fact] + public void Relative_addresses_resolve_against_the_endpoint() + { + var page = OwnServiceSource.ReadPage("""{"items":[{"url":"/files/a.png"}]}""", Endpoint); + + page.Items.ShouldHaveSingleItem().Url.AbsoluteUri.ShouldBe("https://own.test/files/a.png"); + } + + [Fact] + public void Entries_without_a_usable_address_are_dropped() + { + var page = OwnServiceSource.ReadPage( + """{"items":[{"name":"no url"}, {"url":"data:image/png;base64,AA"}, {"url":"https://own.test/ok.png"}]}""", + Endpoint + ); + + page.Items.ShouldHaveSingleItem().Url.AbsoluteUri.ShouldBe("https://own.test/ok.png"); + } + + [Fact] + public void An_empty_listing_is_not_an_error() + { + OwnServiceSource.ReadPage("""{"items":[]}""", Endpoint).Items.ShouldBeEmpty(); + OwnServiceSource.ReadPage("[]", Endpoint).Items.ShouldBeEmpty(); + } +}