Add media sources and the collect runner, alongside the old parsers
Third step: the collector becomes wireable. Both catalogs coexist for exactly this one step, so ParseViewModel and every existing test stay green while the new domain is proven. IMediaSource reuses the closed-generic trick ITextParser used, and for the same reason - the container cannot resolve an open generic as IEnumerable<T>, so adding a source stays a one-line registration. Its input is a MediaQuery rather than text, because a source that walks a paginated listing needs an endpoint and a cursor, not a string. Sources discover; they do not download. That split is why UrlListSource lives in the domain with no network at all, and why everything hard about fetching lives in one place instead of once per source. The catalog takes an explicit default id. Left to alphabetical order the landing source would be the network one, so the app would open behind the proxy gate before the user had asked for anything. The runner decouples discovery from downloading with a bounded channel - a listing of two hundred thousand items must not materialise because the workers are slower than the source - and owns its workers, waiting for them even when cancelled. Without that a stopped run keeps writing to the store after the page has said it stopped. The own-service listing is read leniently: the service on the other end is the user's own and should not have to be rewritten to match a schema we invented, so both a bare array of addresses and an object with items and a cursor work. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
Leonid Pershin
co-authored by
Claude Opus 5
Claude Opus 5
parent
1742c094e9
commit
6909884851
@@ -0,0 +1,34 @@
|
||||
using AvParser.Core.Parsing;
|
||||
|
||||
namespace AvParser.Core.Collecting;
|
||||
|
||||
/// <summary>How to run one collection.</summary>
|
||||
public sealed record CollectOptions
|
||||
{
|
||||
/// <summary>How many downloads may be in flight at once, across all hosts.</summary>
|
||||
/// <remarks>
|
||||
/// 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.
|
||||
/// </remarks>
|
||||
public int MaxConcurrentDownloads { get; init; } = 4;
|
||||
|
||||
/// <summary>Ignore the journal and fetch every address again.</summary>
|
||||
public bool ForceRefetch { get; init; }
|
||||
}
|
||||
|
||||
/// <summary>Runs a source end to end: discover, download, store.</summary>
|
||||
/// <remarks>
|
||||
/// Yields the same <see cref="ParseOutcome{T}"/> stream the page already knows how to consume, so
|
||||
/// the UI needs no notion of channels, workers or journals.
|
||||
/// </remarks>
|
||||
public interface ICollectRunner
|
||||
{
|
||||
/// <summary>Collects everything <paramref name="source"/> finds for <paramref name="query"/>.</summary>
|
||||
IAsyncEnumerable<ParseOutcome<CollectedItem>> RunAsync(
|
||||
IMediaSource source,
|
||||
MediaQuery query,
|
||||
CollectOptions options,
|
||||
IProgress<ParseProgress>? progress,
|
||||
CancellationToken cancellationToken
|
||||
);
|
||||
}
|
||||
@@ -0,0 +1,75 @@
|
||||
using AvParser.Core.Parsing;
|
||||
|
||||
namespace AvParser.Core.Collecting;
|
||||
|
||||
/// <summary>
|
||||
/// Closed, non-generic facade over <see cref="IParser{TInput, TOutput}"/> for media discovery.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// Closed for the same reason the text facade was: an open generic interface cannot be resolved as
|
||||
/// <c>IEnumerable<T></c> by the container, so every source implements this and gets registered
|
||||
/// under it. Adding a source stays a one-line change.
|
||||
/// </remarks>
|
||||
public interface IMediaSource : IParser<MediaQuery, MediaCandidate>;
|
||||
|
||||
/// <summary>Read-only view over every registered media source.</summary>
|
||||
public interface IMediaSourceCatalog
|
||||
{
|
||||
/// <summary>All registered sources, ordered by <see cref="IParser{TInput,TOutput}.DisplayName"/>.</summary>
|
||||
IReadOnlyList<IMediaSource> Sources { get; }
|
||||
|
||||
/// <summary>The source used when nothing has been chosen yet.</summary>
|
||||
IMediaSource DefaultSource { get; }
|
||||
|
||||
/// <summary>Finds a source by its stable id; <see langword="null"/> when unknown.</summary>
|
||||
IMediaSource? Find(string? id);
|
||||
|
||||
/// <summary>Finds a source by id, falling back to <see cref="DefaultSource"/>.</summary>
|
||||
IMediaSource FindOrDefault(string? id) => Find(id) ?? DefaultSource;
|
||||
}
|
||||
|
||||
/// <inheritdoc cref="IMediaSourceCatalog" />
|
||||
public sealed class MediaSourceCatalog : IMediaSourceCatalog
|
||||
{
|
||||
private readonly Dictionary<string, IMediaSource> _byId;
|
||||
|
||||
/// <summary>Builds a catalog from every source the container resolved.</summary>
|
||||
/// <param name="sources">Registered sources.</param>
|
||||
/// <param name="defaultId">
|
||||
/// 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.
|
||||
/// </param>
|
||||
/// <exception cref="ArgumentException">No sources were registered, or two share an id.</exception>
|
||||
public MediaSourceCatalog(IEnumerable<IMediaSource> 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<string, IMediaSource>(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];
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
public IReadOnlyList<IMediaSource> Sources { get; }
|
||||
|
||||
/// <inheritdoc />
|
||||
public IMediaSource DefaultSource { get; }
|
||||
|
||||
/// <inheritdoc />
|
||||
public IMediaSource? Find(string? id) => id is not null && _byId.TryGetValue(id, out var source) ? source : null;
|
||||
}
|
||||
@@ -0,0 +1,122 @@
|
||||
using System.Runtime.CompilerServices;
|
||||
using AvParser.Core.Parsing;
|
||||
|
||||
namespace AvParser.Core.Collecting.Sources;
|
||||
|
||||
/// <summary>
|
||||
/// Reads addresses the user pasted in, one per line.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// The only source that touches no network at all, which is why it lives in the domain and why
|
||||
/// <c>RequiresNetwork</c> stays false: it discovers nothing, it just reads what it was handed.
|
||||
/// Downloading those addresses is the fetcher's business and is gated separately.
|
||||
/// </remarks>
|
||||
public sealed class UrlListSource : IMediaSource
|
||||
{
|
||||
/// <inheritdoc />
|
||||
public string Id => "url-list";
|
||||
|
||||
/// <inheritdoc />
|
||||
public string DisplayName => "URL list";
|
||||
|
||||
/// <inheritdoc />
|
||||
public string Description => "One address per line. Blank lines and lines starting with '#' are ignored.";
|
||||
|
||||
/// <inheritdoc />
|
||||
public bool CanParse(MediaQuery input) =>
|
||||
input is not null && TextLines.Split(input.Text).Any(line => TryParse(line, out _));
|
||||
|
||||
/// <inheritdoc />
|
||||
public async IAsyncEnumerable<ParseOutcome<MediaCandidate>> ParseAsync(
|
||||
MediaQuery input,
|
||||
IProgress<ParseProgress>? 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<MediaCandidate>.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<MediaCandidate>.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));
|
||||
}
|
||||
|
||||
/// <summary>Accepts only absolute HTTP addresses.</summary>
|
||||
/// <remarks>
|
||||
/// <c>file:</c> and <c>data:</c> 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.
|
||||
/// </remarks>
|
||||
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;
|
||||
}
|
||||
}
|
||||
+2
-2
@@ -1,6 +1,6 @@
|
||||
namespace AvParser.Core.Parsing.Samples;
|
||||
namespace AvParser.Core.Collecting;
|
||||
|
||||
/// <summary>Line-splitting helpers shared by the sample parsers.</summary>
|
||||
/// <summary>Line-splitting helpers for anything that reads a pasted list.</summary>
|
||||
internal static class TextLines
|
||||
{
|
||||
/// <summary>How many records to process between progress reports.</summary>
|
||||
@@ -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<ITextParser, KeyValueTextParser>();
|
||||
services.AddSingleton<IParserCatalog, ParserCatalog>();
|
||||
|
||||
services.AddSingleton<IMediaSource, UrlListSource>();
|
||||
|
||||
return services;
|
||||
}
|
||||
|
||||
/// <summary>Id of the source the collector opens on.</summary>
|
||||
/// <remarks>
|
||||
/// 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.
|
||||
/// </remarks>
|
||||
public const string DefaultMediaSourceId = "url-list";
|
||||
}
|
||||
|
||||
@@ -1,4 +1,5 @@
|
||||
using System.Runtime.CompilerServices;
|
||||
using AvParser.Core.Collecting;
|
||||
|
||||
namespace AvParser.Core.Parsing.Samples;
|
||||
|
||||
|
||||
@@ -1,4 +1,5 @@
|
||||
using System.Runtime.CompilerServices;
|
||||
using AvParser.Core.Collecting;
|
||||
|
||||
namespace AvParser.Core.Parsing.Samples;
|
||||
|
||||
|
||||
@@ -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<IMediaStore>().InitialiseAsync().GetAwaiter().GetResult();
|
||||
|
||||
base.OnFrameworkInitializationCompleted();
|
||||
}
|
||||
|
||||
|
||||
@@ -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;
|
||||
|
||||
/// <summary>
|
||||
/// Runs a source end to end: discover, skip what is already settled, download, store.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// <para>
|
||||
/// 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.
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// 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.
|
||||
/// </para>
|
||||
/// </remarks>
|
||||
public sealed class CollectRunner(
|
||||
IMediaFetcher fetcher,
|
||||
IMediaStore store,
|
||||
ISettingsService settings,
|
||||
ILogger<CollectRunner> logger
|
||||
) : ICollectRunner
|
||||
{
|
||||
/// <summary>How many addresses to check against the journal in one query.</summary>
|
||||
private const int JournalBatch = 200;
|
||||
|
||||
/// <summary>Report progress at least this often, however slow the items are.</summary>
|
||||
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<CollectRunner> _logger = logger ?? throw new ArgumentNullException(nameof(logger));
|
||||
|
||||
/// <inheritdoc />
|
||||
public async IAsyncEnumerable<ParseOutcome<CollectedItem>> RunAsync(
|
||||
IMediaSource source,
|
||||
MediaQuery query,
|
||||
CollectOptions options,
|
||||
IProgress<ParseProgress>? 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<MediaCandidate>(
|
||||
new BoundedChannelOptions(workers * 4) { FullMode = BoundedChannelFullMode.Wait }
|
||||
);
|
||||
var results = Channel.CreateUnbounded<ParseOutcome<CollectedItem>>();
|
||||
|
||||
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
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>Reads the source and feeds the workers, skipping what the journal already settled.</summary>
|
||||
private async Task ProduceAsync(
|
||||
IMediaSource source,
|
||||
MediaQuery query,
|
||||
CollectOptions options,
|
||||
ChannelWriter<MediaCandidate> work,
|
||||
ChannelWriter<ParseOutcome<CollectedItem>> results,
|
||||
Counters counters,
|
||||
IProgress<ParseProgress>? progress,
|
||||
CancellationToken token
|
||||
)
|
||||
{
|
||||
var batch = new List<MediaCandidate>(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<CollectedItem>.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<CollectedItem>.Failure(ParseError.Create(0, "SourceFailed", ex.Message, ex.Message)),
|
||||
CancellationToken.None
|
||||
)
|
||||
.ConfigureAwait(false);
|
||||
}
|
||||
finally
|
||||
{
|
||||
work.TryComplete();
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>Checks a batch against the journal and queues whatever still needs fetching.</summary>
|
||||
private async Task DispatchAsync(
|
||||
IMediaSource source,
|
||||
CollectOptions options,
|
||||
List<MediaCandidate> batch,
|
||||
ChannelWriter<MediaCandidate> work,
|
||||
ChannelWriter<ParseOutcome<CollectedItem>> results,
|
||||
Counters counters,
|
||||
CancellationToken token
|
||||
)
|
||||
{
|
||||
if (batch.Count == 0)
|
||||
{
|
||||
return;
|
||||
}
|
||||
|
||||
var seen = options.ForceRefetch
|
||||
? new Dictionary<string, SeenOutcome>(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<CollectedItem>.Success(
|
||||
new CollectedItem(candidate, Placeholder, CollectStatus.Skipped)
|
||||
),
|
||||
token
|
||||
)
|
||||
.ConfigureAwait(false);
|
||||
|
||||
continue;
|
||||
}
|
||||
|
||||
await work.WriteAsync(candidate, token).ConfigureAwait(false);
|
||||
}
|
||||
|
||||
batch.Clear();
|
||||
}
|
||||
|
||||
/// <summary>Downloads and stores whatever the producer queues.</summary>
|
||||
private async Task ConsumeAsync(
|
||||
IMediaSource source,
|
||||
string runId,
|
||||
FetchOptions fetchOptions,
|
||||
ChannelReader<MediaCandidate> work,
|
||||
ChannelWriter<ParseOutcome<CollectedItem>> 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<ParseOutcome<CollectedItem>> 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<CollectedItem>.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<CollectedItem>.Success(stored with { Elapsed = result.Elapsed });
|
||||
}
|
||||
|
||||
private FetchOptions BuildFetchOptions(IReadOnlySet<string> 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,
|
||||
};
|
||||
}
|
||||
|
||||
/// <summary>Stands in for content on a skipped item, which by definition has none.</summary>
|
||||
private static MediaBlob Placeholder { get; } = MediaBlob.Create(new string('0', 64), MediaKind.Unknown, 0);
|
||||
|
||||
/// <summary>Run totals, written from several workers at once.</summary>
|
||||
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);
|
||||
}
|
||||
}
|
||||
@@ -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;
|
||||
|
||||
/// <summary>
|
||||
/// Lists what a service you run holds, by asking it.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// <para>
|
||||
/// 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.
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// 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:
|
||||
/// </para>
|
||||
/// <code>
|
||||
/// { "items": [ { "url": "...", "id": "...", "name": "...", "published": "...", "size": 1234,
|
||||
/// "tags": ["a"] } ], "next": "cursor" }
|
||||
/// [ "https://host/one.png", "https://host/two.gif" ]
|
||||
/// </code>
|
||||
/// <para>
|
||||
/// Paging follows <c>next</c> until it is absent. A page that repeats a cursor stops the walk
|
||||
/// rather than looping for ever.
|
||||
/// </para>
|
||||
/// </remarks>
|
||||
public sealed class OwnServiceSource(
|
||||
IProxiedHttpClientFactory clients,
|
||||
ISettingsService settings,
|
||||
ILogger<OwnServiceSource> logger
|
||||
) : IMediaSource
|
||||
{
|
||||
/// <summary>Stops a service that keeps handing back pages from running the collector for ever.</summary>
|
||||
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<OwnServiceSource> _logger = logger ?? throw new ArgumentNullException(nameof(logger));
|
||||
|
||||
/// <inheritdoc />
|
||||
public string Id => "own-service";
|
||||
|
||||
/// <inheritdoc />
|
||||
public string DisplayName => "Own service";
|
||||
|
||||
/// <inheritdoc />
|
||||
public string Description => "Reads the listing endpoint of a service you run.";
|
||||
|
||||
/// <inheritdoc />
|
||||
public bool RequiresNetwork => true;
|
||||
|
||||
/// <inheritdoc />
|
||||
public bool CanParse(MediaQuery input) =>
|
||||
input?.Endpoint is { IsAbsoluteUri: true } endpoint && endpoint.Scheme is "http" or "https";
|
||||
|
||||
/// <inheritdoc />
|
||||
public async IAsyncEnumerable<ParseOutcome<MediaCandidate>> ParseAsync(
|
||||
MediaQuery input,
|
||||
IProgress<ParseProgress>? progress,
|
||||
[EnumeratorCancellation] CancellationToken cancellationToken
|
||||
)
|
||||
{
|
||||
ArgumentNullException.ThrowIfNull(input);
|
||||
|
||||
if (!CanParse(input))
|
||||
{
|
||||
yield return ParseOutcome<MediaCandidate>.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<string>(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<MediaCandidate>.Failure(error);
|
||||
yield break;
|
||||
}
|
||||
|
||||
foreach (var candidate in parsed.Items)
|
||||
{
|
||||
ordinal++;
|
||||
|
||||
yield return ParseOutcome<MediaCandidate>.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);
|
||||
}
|
||||
|
||||
/// <summary>One page of a listing.</summary>
|
||||
/// <param name="Items">Candidates on this page.</param>
|
||||
/// <param name="Next">Cursor for the following page, or null when this was the last.</param>
|
||||
internal readonly record struct Page(IReadOnlyList<MediaCandidate> 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)}");
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Reads either shape of listing.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// Hand-parsed with <see cref="JsonDocument"/> 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.
|
||||
/// </remarks>
|
||||
internal static Page ReadPage(string json, Uri baseAddress)
|
||||
{
|
||||
using var document = JsonDocument.Parse(json);
|
||||
var root = document.RootElement;
|
||||
|
||||
var items = new List<MediaCandidate>();
|
||||
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;
|
||||
|
||||
/// <summary>Resolves a listed address, allowing relative paths against the endpoint.</summary>
|
||||
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;
|
||||
}
|
||||
}
|
||||
+52
@@ -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;
|
||||
}
|
||||
|
||||
/// <summary>Registers the media store, the download pipeline and the network sources.</summary>
|
||||
/// <remarks>
|
||||
/// 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.
|
||||
/// </remarks>
|
||||
public static IServiceCollection AddAvParserCollecting(this IServiceCollection services)
|
||||
{
|
||||
ArgumentNullException.ThrowIfNull(services);
|
||||
|
||||
services.AddSingleton<BlobStore>();
|
||||
services.AddSingleton<SqliteMediaIndex>();
|
||||
|
||||
// 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<IAppPaths>(),
|
||||
sp.GetRequiredService<BlobStore>(),
|
||||
sp.GetRequiredService<ILogger<ShowcaseLinker>>()
|
||||
));
|
||||
|
||||
services.AddSingleton<MediaStore>();
|
||||
services.AddSingleton<IMediaStore>(sp => sp.GetRequiredService<MediaStore>());
|
||||
|
||||
// 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<ILogger<HostThrottle>>()
|
||||
));
|
||||
|
||||
services.AddSingleton<IMediaFetcher, MediaFetcher>();
|
||||
services.AddSingleton<ICollectRunner, CollectRunner>();
|
||||
|
||||
services.AddSingleton<IMediaSource, OwnServiceSource>();
|
||||
|
||||
// Resolves sources from both assemblies: the container gathers every IMediaSource
|
||||
// registration regardless of which project declared it.
|
||||
services.AddSingleton<IMediaSourceCatalog>(sp => new MediaSourceCatalog(
|
||||
sp.GetServices<IMediaSource>(),
|
||||
CoreServiceCollectionExtensions.DefaultMediaSourceId
|
||||
));
|
||||
|
||||
return services;
|
||||
}
|
||||
|
||||
@@ -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<List<ParseOutcome<MediaCandidate>>> RunAsync(string text, int limit = 0)
|
||||
{
|
||||
var results = new List<ParseOutcome<MediaCandidate>>();
|
||||
|
||||
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<OperationCanceledException>(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<ArgumentException>(() => new MediaSourceCatalog([]));
|
||||
|
||||
[Fact]
|
||||
public void Two_sources_sharing_an_id_are_rejected() =>
|
||||
Should.Throw<ArgumentException>(() =>
|
||||
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<ParseOutcome<MediaCandidate>> ParseAsync(
|
||||
MediaQuery input,
|
||||
IProgress<ParseProgress>? progress,
|
||||
[System.Runtime.CompilerServices.EnumeratorCancellation] CancellationToken cancellationToken
|
||||
)
|
||||
{
|
||||
await Task.Yield();
|
||||
yield break;
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -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;
|
||||
|
||||
/// <summary>A source that hands back exactly the addresses the test names.</summary>
|
||||
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<ParseOutcome<MediaCandidate>> ParseAsync(
|
||||
MediaQuery input,
|
||||
IProgress<ParseProgress>? progress,
|
||||
[EnumeratorCancellation] CancellationToken cancellationToken
|
||||
)
|
||||
{
|
||||
Listings++;
|
||||
|
||||
for (var index = 0; index < urls.Length; index++)
|
||||
{
|
||||
cancellationToken.ThrowIfCancellationRequested();
|
||||
await Task.Yield();
|
||||
|
||||
yield return ParseOutcome<MediaCandidate>.Success(
|
||||
new MediaCandidate(new Uri(urls[index])) { SourceId = Id, Ordinal = index + 1 }
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>A fetcher that stages bytes from a table instead of using a network.</summary>
|
||||
internal sealed class ScriptedFetcher(BlobStore blobs) : IMediaFetcher
|
||||
{
|
||||
public Dictionary<string, byte[]> Content { get; } = new(StringComparer.Ordinal);
|
||||
|
||||
public HashSet<string> Failing { get; } = new(StringComparer.Ordinal);
|
||||
|
||||
public List<string> Fetched { get; } = [];
|
||||
|
||||
public TimeSpan Delay { get; set; }
|
||||
|
||||
public async Task<FetchResult> 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,
|
||||
};
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>In-memory settings, so the runner never reads the developer's real profile.</summary>
|
||||
internal sealed class FixedSettings(AppSettings? initial = null) : ISettingsService, IDisposable
|
||||
{
|
||||
private readonly BehaviorSignal<AppSettings> _current = new(initial ?? new AppSettings());
|
||||
|
||||
public AppSettings Current => _current.Value;
|
||||
|
||||
public IObservable<AppSettings> Changes => _current;
|
||||
|
||||
public void Update(Func<AppSettings, AppSettings> 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<SqliteMediaIndex>.Instance);
|
||||
_blobs = new BlobStore(_paths, NullLogger<BlobStore>.Instance);
|
||||
_store = new MediaStore(
|
||||
_index,
|
||||
_blobs,
|
||||
new ShowcaseLinker(_paths, _blobs, NullLogger<ShowcaseLinker>.Instance),
|
||||
NullLogger<MediaStore>.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<CollectRunner>.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<List<ParseOutcome<CollectedItem>>> RunAsync(IMediaSource source, CollectOptions? options = null)
|
||||
{
|
||||
var results = new List<ParseOutcome<CollectedItem>>();
|
||||
|
||||
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<ParseOutcome<CollectedItem>>();
|
||||
|
||||
await Should.ThrowAsync<OperationCanceledException>(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<ParseOutcome<MediaCandidate>> ParseAsync(
|
||||
MediaQuery input,
|
||||
IProgress<ParseProgress>? 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();
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user