Add the media store: content-addressed blobs, SQLite index, showcase

First half of replacing the stub text domain with a media collector. Nothing
references this yet - the store is standalone and fully tested before anything
depends on it.

Blobs are addressed by SHA-256 and sharded two levels deep, so the same picture
re-uploaded at a dozen addresses costs one file. Downloads stage in a sibling
temp directory on the same volume and are promoted by rename, which is what
keeps blobs/ free of truncated files: a crash leaves a stray .part that the next
startup sweeps, never a half-image indistinguishable from a real one.

The SQLite index holds provenance separately from content, so purging one source
leaves blobs another source still references - that is what ref_count buys, and
it is recomputed rather than incremented because the item upsert can replace a
row pointing at a different blob. The seen_url journal deliberately outlives a
purge: without that, the next run downloads again exactly what the user just
deleted. Terminal outcomes are split from retryable ones so a flaky network does
not permanently lose content.

The showcase gives every item a dated, named path via hard links - a second name
for one file, not a second file. Hard links are a filesystem privilege rather
than a guarantee, so it degrades to copying and records which it achieved; the
UI has to be able to admit that. Names suggested by the origin are treated as
hostile: only the last path segment survives, Windows device names are pushed
aside, and the extension comes from the sniffed kind, never from the remote.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
Leonid Pershin
2026-08-13 20:46:21 +03:00
co-authored by Claude Opus 5
parent 44fb0d3a5f
commit 181f974a37
17 changed files with 2556 additions and 1 deletions
+9
View File
@@ -10,6 +10,9 @@
<ReactiveUIVersion>24.1.0</ReactiveUIVersion>
<ReactiveUIPrimitivesVersion>7.1.1</ReactiveUIPrimitivesVersion>
<MicrosoftExtensionsVersion>10.0.11</MicrosoftExtensionsVersion>
<!-- Ships on the EF Core release line, not the Extensions one. They happen to match today;
they will not always, so this is deliberately its own knob. -->
<MicrosoftDataSqliteVersion>10.0.11</MicrosoftDataSqliteVersion>
</PropertyGroup>
<ItemGroup Label="Avalonia">
@@ -46,6 +49,12 @@
<PackageVersion Include="Microsoft.Extensions.Logging.Abstractions" Version="$(MicrosoftExtensionsVersion)" />
</ItemGroup>
<ItemGroup Label="Data">
<!-- Not Microsoft.Data.Sqlite.Core: we want the bundled e_sqlite3 native, so nothing extra
has to be shipped per runtime identifier. -->
<PackageVersion Include="Microsoft.Data.Sqlite" Version="$(MicrosoftDataSqliteVersion)" />
</ItemGroup>
<ItemGroup Label="Logging">
<PackageVersion Include="Serilog" Version="4.4.0" />
<PackageVersion Include="Serilog.Extensions.Logging" Version="10.0.0" />
@@ -0,0 +1,94 @@
namespace AvParser.Core.Collecting;
/// <summary>What happened to one candidate.</summary>
public enum CollectStatus
{
/// <summary>New content; a blob was written.</summary>
Stored = 0,
/// <summary>The bytes were already in the store; only provenance was recorded.</summary>
Duplicate = 1,
/// <summary>Not fetched at all, because a previous run already settled this address.</summary>
Skipped = 2,
}
/// <summary>One candidate carried all the way through to the store.</summary>
/// <param name="Candidate">What the source found.</param>
/// <param name="Blob">The content that came back.</param>
/// <param name="Status">Whether it was new, already held, or never fetched.</param>
public sealed record CollectedItem(MediaCandidate Candidate, MediaBlob Blob, CollectStatus Status)
{
/// <summary>How long the fetch took, including redirects.</summary>
public TimeSpan Elapsed { get; init; }
/// <summary>Path of the showcase entry, relative to the showcase root; null when none was made.</summary>
public string? ShowcasePath { get; init; }
}
/// <summary>
/// The recorded fate of an address, so a later run need not ask again.
/// </summary>
/// <remarks>
/// Split into terminal and retryable by <see cref="SeenOutcomes.IsTerminal"/>. That split is the
/// entire value of the journal: without it a re-run either re-downloads everything or gives up on
/// addresses that failed once for a transient reason.
/// </remarks>
public enum SeenOutcome
{
/// <summary>Downloaded and stored.</summary>
Stored = 0,
/// <summary>Downloaded; the bytes were already held.</summary>
Duplicate = 1,
/// <summary>Content matched a known dead-link placeholder.</summary>
Placeholder = 2,
/// <summary>Bigger than the configured ceiling.</summary>
TooLarge = 3,
/// <summary>Smaller than the configured floor — tracking pixels and spacers.</summary>
TooSmall = 4,
/// <summary>A recognised format that the user excluded.</summary>
UnsupportedType = 5,
/// <summary>The bytes matched no known signature.</summary>
NotMedia = 6,
/// <summary>The origin said it is permanently gone.</summary>
Gone = 7,
/// <summary>Failed for a reason that may not recur.</summary>
Failed = 8,
/// <summary>Deferred by the origin's own rate limiting.</summary>
RateLimited = 9,
/// <summary>Timed out.</summary>
Timeout = 10,
}
/// <summary>Helpers over <see cref="SeenOutcome"/>.</summary>
public static class SeenOutcomes
{
/// <summary>
/// Whether the outcome settles the address for good, so a re-run can skip it without asking.
/// </summary>
/// <remarks>
/// Deliberately excludes <see cref="SeenOutcome.Failed"/>, <see cref="SeenOutcome.RateLimited"/>
/// and <see cref="SeenOutcome.Timeout"/>: those describe the moment, not the resource, and a
/// journal that treated them as final would make a flaky network permanently lose content.
/// </remarks>
public static bool IsTerminal(SeenOutcome outcome) =>
outcome
is SeenOutcome.Stored
or SeenOutcome.Duplicate
or SeenOutcome.Placeholder
or SeenOutcome.TooLarge
or SeenOutcome.TooSmall
or SeenOutcome.UnsupportedType
or SeenOutcome.NotMedia
or SeenOutcome.Gone;
}
+137
View File
@@ -0,0 +1,137 @@
namespace AvParser.Core.Collecting;
/// <summary>A finished download waiting to be taken into the store.</summary>
/// <param name="Candidate">What the source found.</param>
/// <param name="Blob">Hash, kind and size of the content that came back.</param>
/// <param name="TempFilePath">
/// Complete, verified file to take ownership of. The store moves or deletes it and the caller must
/// not touch it afterwards.
/// </param>
/// <param name="RunId">Run this belongs to.</param>
/// <param name="FinalUrl">Address after redirects; <see cref="MediaCandidate.Url"/> is the one asked for.</param>
/// <param name="HttpStatus">Status of the final response.</param>
public sealed record MediaStoreRequest(
MediaCandidate Candidate,
MediaBlob Blob,
string TempFilePath,
string RunId,
Uri FinalUrl,
int HttpStatus
)
{
/// <summary>What the origin claimed the type was. Kept precisely to catch the ones that lie.</summary>
public string? ContentType { get; init; }
/// <summary>Proxy the content came through, or null when the connection was direct.</summary>
public string? ProxyKey { get; init; }
}
/// <summary>What to remove when purging.</summary>
/// <param name="SourceId">Only content attributed to this source is considered.</param>
public sealed record PurgeOptions(string SourceId)
{
/// <summary>
/// Also forget every address this source ever visited.
/// </summary>
/// <remarks>
/// Off by default, and the default is the interesting one: the journal outliving a purge is
/// what stops the next run from downloading again exactly what the user just deleted.
/// </remarks>
public bool ForgetSeenUrls { get; init; }
}
/// <summary>Outcome of a purge.</summary>
/// <param name="ItemsRemoved">Provenance rows deleted.</param>
/// <param name="BlobsRemoved">Blobs that lost their last reference and were deleted.</param>
/// <param name="BytesFreed">Bytes reclaimed on disk.</param>
public readonly record struct PurgeResult(int ItemsRemoved, int BlobsRemoved, long BytesFreed);
/// <summary>Totals for the dashboard.</summary>
/// <param name="BlobCount">Distinct byte sequences held.</param>
/// <param name="ItemCount">Provenance rows across every source.</param>
/// <param name="TotalBytes">Sum of blob sizes.</param>
/// <param name="SourceCount">Distinct sources that contributed.</param>
public readonly record struct MediaStoreStats(int BlobCount, int ItemCount, long TotalBytes, int SourceCount);
/// <summary>How a run ended.</summary>
/// <param name="Discovered">Candidates the source produced.</param>
/// <param name="Stored">Candidates that produced new content.</param>
/// <param name="Duplicates">Candidates whose bytes were already held.</param>
/// <param name="Failed">Candidates that produced an error.</param>
/// <param name="Bytes">Bytes newly written.</param>
/// <param name="Cancelled">Whether the user stopped it.</param>
public readonly record struct RunSummary(
int Discovered,
int Stored,
int Duplicates,
int Failed,
long Bytes,
bool Cancelled
);
/// <summary>
/// Everything the collector persists: the blobs, their provenance, and the journal of addresses
/// already visited.
/// </summary>
/// <remarks>
/// The abstraction lives in the domain and the implementation in the infrastructure, the same way
/// <c>ISettingsService</c> and <c>IProxySource</c> do, so that the domain stays runnable from a
/// CLI or a test without dragging in a database.
/// </remarks>
public interface IMediaStore
{
/// <summary>Creates or upgrades the schema. Safe to call repeatedly.</summary>
Task InitialiseAsync(CancellationToken cancellationToken = default);
/// <summary>Opens a run and returns its id.</summary>
Task<string> BeginRunAsync(string sourceId, CancellationToken cancellationToken = default);
/// <summary>Closes a run with its totals.</summary>
Task CompleteRunAsync(string runId, RunSummary summary, CancellationToken cancellationToken = default);
/// <summary>
/// Looks up what previous runs made of these addresses.
/// </summary>
/// <remarks>
/// Batched deliberately: a listing of a hundred thousand addresses must cost a handful of
/// queries, not a hundred thousand of them.
/// </remarks>
Task<IReadOnlyDictionary<string, SeenOutcome>> GetSeenAsync(
string sourceId,
IReadOnlyCollection<string> urls,
CancellationToken cancellationToken = default
);
/// <summary>Records the fate of an address that produced no content.</summary>
Task RecordSeenAsync(
string sourceId,
string url,
SeenOutcome outcome,
int? httpStatus = null,
string? errorCode = null,
CancellationToken cancellationToken = default
);
/// <summary>Reads every hash known to be a dead-link placeholder.</summary>
/// <remarks>Loaded once per run: the set is tiny and the check runs on every download.</remarks>
Task<IReadOnlySet<string>> LoadTombstonesAsync(CancellationToken cancellationToken = default);
/// <summary>
/// Marks a hash as a placeholder and removes every copy already held.
/// </summary>
/// <returns>How many stored items were removed.</returns>
Task<int> TombstoneAsync(string sha256, string? reason, CancellationToken cancellationToken = default);
/// <summary>Takes ownership of a completed download and records it.</summary>
Task<CollectedItem> StoreAsync(MediaStoreRequest request, CancellationToken cancellationToken = default);
/// <summary>Removes everything attributed to one source.</summary>
Task<PurgeResult> PurgeAsync(PurgeOptions options, CancellationToken cancellationToken = default);
/// <summary>Rebuilds the browsable showcase for a source from the recorded provenance.</summary>
/// <returns>How many entries were linked.</returns>
Task<int> RebuildShowcaseAsync(string sourceId, CancellationToken cancellationToken = default);
/// <summary>Totals across the whole store.</summary>
Task<MediaStoreStats> GetStatsAsync(CancellationToken cancellationToken = default);
}
+40
View File
@@ -0,0 +1,40 @@
namespace AvParser.Core.Collecting;
/// <summary>
/// One distinct byte sequence in the store, identified by its content.
/// </summary>
/// <remarks>
/// Identity is the hash, not the URL: the same picture is re-uploaded and re-hosted constantly,
/// and deduplicating by address would store it once per address. Because the proxy handler
/// decompresses transparently, the hash is over the <b>decoded</b> bytes — so the same image
/// served gzipped at one URL and plain at another still collapses to a single blob.
/// </remarks>
/// <param name="Sha256">Lowercase hex SHA-256 of the content, 64 characters.</param>
/// <param name="Kind">What the signature said it is.</param>
/// <param name="Extension">Extension derived from <paramref name="Kind"/>, leading dot included.</param>
/// <param name="Length">Size in bytes of the decoded content.</param>
public sealed record MediaBlob(string Sha256, MediaKind Kind, string Extension, long Length)
{
/// <summary>Pixel width, when it was cheap to read from the header.</summary>
public int? Width { get; init; }
/// <summary>Pixel height, when it was cheap to read from the header.</summary>
public int? Height { get; init; }
/// <summary>Whether the content has more than one frame.</summary>
public bool IsAnimated { get; init; }
/// <summary>Creates a blob, deriving the extension from the kind.</summary>
public static MediaBlob Create(string sha256, MediaKind kind, long length) =>
new(sha256, kind, MediaKinds.ExtensionFor(kind), length);
/// <summary>
/// Path of this blob relative to the blob root, using two levels of two hex characters.
/// </summary>
/// <remarks>
/// 65 536 leaf directories keeps any one of them to a few hundred entries at a million blobs,
/// which matters because some filesystems and most file managers degrade badly on flat
/// directories of that size.
/// </remarks>
public string RelativePath => Path.Combine(Sha256[..2], Sha256.Substring(2, 2), string.Concat(Sha256, Extension));
}
@@ -0,0 +1,38 @@
namespace AvParser.Core.Collecting;
/// <summary>
/// Something a source found and thinks is worth downloading.
/// </summary>
/// <remarks>
/// A discovery, not the bytes. Keeping the two apart is the whole shape of the collector: a source
/// only has to enumerate addresses, so it stays cheap and mostly testable without a network, while
/// everything hard about downloading — redirects, timeouts, sniffing, throttling — lives in one
/// place instead of being reimplemented per source.
/// </remarks>
/// <param name="Url">Where to fetch it from.</param>
public sealed record MediaCandidate(Uri Url)
{
/// <summary>Id of the source that found it. Stamped by the runner when a source leaves it blank.</summary>
public string SourceId { get; init; } = string.Empty;
/// <summary>Page to send as <c>Referer</c>; some origins refuse a bare request.</summary>
public Uri? Referer { get; init; }
/// <summary>Identifier in the originating service, when it has one.</summary>
public string? ExternalId { get; init; }
/// <summary>Name to prefer in the showcase, before sanitising.</summary>
public string? SuggestedName { get; init; }
/// <summary>When the origin says it was published.</summary>
public DateTimeOffset? PublishedUtc { get; init; }
/// <summary>Size the listing claimed, when it said. Advisory only — never trusted for limits.</summary>
public long? ExpectedLength { get; init; }
/// <summary>Tags carried over from the source.</summary>
public IReadOnlyList<string> Tags { get; init; } = [];
/// <summary>1-based position within the listing, used to report failures against something.</summary>
public int Ordinal { get; init; }
}
+62
View File
@@ -0,0 +1,62 @@
namespace AvParser.Core.Collecting;
/// <summary>What a downloaded byte sequence actually turned out to be.</summary>
/// <remarks>
/// Decided by the file's own signature, never by its URL, extension or <c>Content-Type</c> — all
/// three are routinely wrong, and two of them are attacker-controlled.
/// </remarks>
public enum MediaKind
{
/// <summary>Nothing recognised. Never stored.</summary>
Unknown = 0,
/// <summary>JPEG image.</summary>
Jpeg = 1,
/// <summary>PNG image, possibly animated (APNG).</summary>
Png = 2,
/// <summary>GIF image, possibly animated.</summary>
Gif = 3,
/// <summary>WebP image, possibly animated.</summary>
WebP = 4,
/// <summary>AVIF image.</summary>
Avif = 5,
/// <summary>MP4 video. What most sites serve when they say "GIF".</summary>
Mp4 = 6,
/// <summary>WebM video.</summary>
WebM = 7,
}
/// <summary>Helpers over <see cref="MediaKind"/>.</summary>
public static class MediaKinds
{
/// <summary>File extension for a kind, leading dot included.</summary>
/// <remarks>
/// The single source of truth for what a stored file is called. Deriving it from the URL would
/// mean a host could dictate the extension of a file written to the user's disk.
/// </remarks>
public static string ExtensionFor(MediaKind kind) =>
kind switch
{
MediaKind.Jpeg => ".jpg",
MediaKind.Png => ".png",
MediaKind.Gif => ".gif",
MediaKind.WebP => ".webp",
MediaKind.Avif => ".avif",
MediaKind.Mp4 => ".mp4",
MediaKind.WebM => ".webm",
_ => ".bin",
};
/// <summary>Whether the kind is a still or animated picture rather than a video container.</summary>
public static bool IsImage(MediaKind kind) =>
kind is MediaKind.Jpeg or MediaKind.Png or MediaKind.Gif or MediaKind.WebP or MediaKind.Avif;
/// <summary>Whether the kind is a video container.</summary>
public static bool IsVideo(MediaKind kind) => kind is MediaKind.Mp4 or MediaKind.WebM;
}
@@ -0,0 +1,23 @@
namespace AvParser.Core.Collecting;
/// <summary>
/// What to ask a source for.
/// </summary>
/// <remarks>
/// The collector's equivalent of the text that used to be pasted into the parse page. A record
/// rather than a bare string because sources differ in what they need — one reads a pasted list,
/// another walks a paginated listing — and widening a record beats a second interface per source.
/// </remarks>
/// <param name="Text">Free text, typically a pasted list of addresses, one per line.</param>
/// <param name="Endpoint">Listing endpoint, for sources that ask a service what it holds.</param>
/// <param name="Limit">Stop after this many candidates; 0 means no cap.</param>
/// <param name="Cursor">Where to resume a paginated listing.</param>
public sealed record MediaQuery(string Text = "", Uri? Endpoint = null, int Limit = 0, string? Cursor = null)
{
/// <summary>Source-specific extras, so a new source needs no change to this type.</summary>
public IReadOnlyDictionary<string, string> Options { get; init; } =
new Dictionary<string, string>(StringComparer.OrdinalIgnoreCase);
/// <summary>Whether the caller asked for a bounded number of candidates.</summary>
public bool HasLimit => Limit > 0;
}
@@ -0,0 +1,25 @@
namespace AvParser.Core.Collecting;
/// <summary>
/// How a showcase entry points at the blob it shows.
/// </summary>
/// <remarks>
/// Doubles as the user's preference and as the record of what was actually achieved for one entry.
/// They differ more often than you would like: hard links are a filesystem privilege, not a
/// guarantee, and a media root on exFAT, on a network share or across volumes silently has to fall
/// back to copying. Storing the achieved mode per entry is what lets the UI admit that.
/// </remarks>
public enum ShowcaseMode
{
/// <summary>No browsable entry; only the content-addressed blob exists.</summary>
None = 0,
/// <summary>A second directory entry for the same content. Costs no extra space.</summary>
HardLink = 1,
/// <summary>A symbolic link. Needs Developer Mode or elevation on Windows, so it is opt-in.</summary>
SymbolicLink = 2,
/// <summary>An independent copy. Doubles the space used.</summary>
Copy = 3,
}
@@ -1,6 +1,11 @@
<Project Sdk="Microsoft.NET.Sdk">
<PropertyGroup>
<RootNamespace>AvParser.Infrastructure</RootNamespace>
<!-- [LibraryImport] generates pointer-based marshalling, so the source generator needs this.
It is here only for the hard-link P/Invokes in Media/HardLink.cs; the BCL has no wrapper
for CreateHardLink/link(2). Using [DllImport] instead would trip SYSLIB1054, which is an
error under TreatWarningsAsErrors. -->
<AllowUnsafeBlocks>true</AllowUnsafeBlocks>
</PropertyGroup>
<ItemGroup>
@@ -14,6 +19,7 @@
</ItemGroup>
<ItemGroup>
<PackageReference Include="Microsoft.Data.Sqlite" />
<PackageReference Include="Microsoft.Extensions.DependencyInjection.Abstractions" />
<PackageReference Include="Microsoft.Extensions.Http" />
<PackageReference Include="Microsoft.Extensions.Logging.Abstractions" />
@@ -0,0 +1,209 @@
using AvParser.Core.Collecting;
using AvParser.Infrastructure.Storage;
using Microsoft.Extensions.Logging;
namespace AvParser.Infrastructure.Media;
/// <summary>Whether a promotion actually wrote anything.</summary>
public enum BlobPromotion
{
/// <summary>The file was moved into the store; these bytes were new.</summary>
Written = 0,
/// <summary>The content was already held, so the temporary file was discarded.</summary>
AlreadyPresent = 1,
}
/// <summary>
/// The files on disk: content-addressed, sharded, and written only once complete.
/// </summary>
/// <remarks>
/// The invariant this type exists to hold is that <c>blobs/</c> contains nothing but complete,
/// hashed content. Downloads are staged in a sibling temporary directory on the same volume and
/// promoted by rename, so a crash mid-download leaves a stray temporary file rather than a
/// truncated image that would then be indistinguishable from a real one for ever.
/// </remarks>
public sealed class BlobStore(IAppPaths paths, ILogger<BlobStore> logger)
{
private readonly IAppPaths _paths = paths ?? throw new ArgumentNullException(nameof(paths));
private readonly ILogger<BlobStore> _logger = logger ?? throw new ArgumentNullException(nameof(logger));
/// <summary>Absolute path this content would live at.</summary>
public string PathFor(MediaBlob blob)
{
ArgumentNullException.ThrowIfNull(blob);
return Path.Combine(_paths.BlobDirectory, blob.RelativePath);
}
/// <summary>Absolute path for a hash and extension already known.</summary>
public string PathFor(string sha256, string extension)
{
ArgumentException.ThrowIfNullOrWhiteSpace(sha256);
return Path.Combine(
_paths.BlobDirectory,
sha256[..2],
sha256.Substring(2, 2),
string.Concat(sha256, extension)
);
}
/// <summary>Whether this content is already held.</summary>
public bool Exists(MediaBlob blob) => File.Exists(PathFor(blob));
/// <summary>Reserves a path for an in-progress download.</summary>
/// <remarks>
/// The <c>.part</c> suffix is load-bearing for the sweep below: it is how a leftover from a
/// killed process is told apart from a file some other part of the app is using.
/// </remarks>
public string CreateTempPath()
{
Directory.CreateDirectory(_paths.MediaTempDirectory);
return Path.Combine(_paths.MediaTempDirectory, $"{Guid.NewGuid():N}.part");
}
/// <summary>
/// Takes ownership of a completed temporary file and moves it into the store.
/// </summary>
/// <remarks>
/// The temporary file is consumed either way: moved when the content is new, deleted when it
/// was already held. Callers must not touch it afterwards.
/// </remarks>
public BlobPromotion Promote(MediaBlob blob, string tempFilePath)
{
ArgumentNullException.ThrowIfNull(blob);
ArgumentException.ThrowIfNullOrWhiteSpace(tempFilePath);
var target = PathFor(blob);
if (File.Exists(target))
{
// Identical hash means identical bytes; there is nothing to compare and nothing to write.
TryDelete(tempFilePath);
return BlobPromotion.AlreadyPresent;
}
Directory.CreateDirectory(Path.GetDirectoryName(target)!);
try
{
// Deliberately without overwrite: if a concurrent worker just wrote the same content,
// its file is as good as ours and clobbering it would break any reader mid-stream.
File.Move(tempFilePath, target);
return BlobPromotion.Written;
}
catch (IOException) when (File.Exists(target))
{
TryDelete(tempFilePath);
return BlobPromotion.AlreadyPresent;
}
}
/// <summary>Removes content from disk. Missing is success — the caller wants it gone.</summary>
public bool Delete(string sha256, string extension)
{
var path = PathFor(sha256, extension);
if (!File.Exists(path))
{
return false;
}
if (!TryDelete(path))
{
return false;
}
TrimEmptyShards(Path.GetDirectoryName(path));
return true;
}
/// <summary>
/// Deletes temporary files left behind by a previous process.
/// </summary>
/// <returns>How many were removed.</returns>
/// <remarks>
/// Only files older than <paramref name="olderThan"/> are touched, so a sweep at startup
/// cannot delete a download that another instance of the app is running right now.
/// </remarks>
public int SweepTemp(TimeSpan olderThan)
{
if (!Directory.Exists(_paths.MediaTempDirectory))
{
return 0;
}
var cutoff = DateTime.UtcNow - olderThan;
var removed = 0;
foreach (var file in Directory.EnumerateFiles(_paths.MediaTempDirectory, "*.part"))
{
try
{
if (File.GetLastWriteTimeUtc(file) < cutoff && TryDelete(file))
{
removed++;
}
}
catch (Exception ex) when (ex is IOException or UnauthorizedAccessException)
{
// Being unable to tidy up is not a reason to fail whatever asked for the sweep.
}
}
if (removed > 0)
{
_logger.LogInformation("Removed {Count} unfinished download(s) from a previous run", removed);
}
return removed;
}
private bool TryDelete(string path)
{
try
{
File.Delete(path);
return true;
}
catch (Exception ex) when (ex is IOException or UnauthorizedAccessException)
{
_logger.LogWarning(ex, "Could not delete {Path}", path);
return false;
}
}
/// <summary>Removes the two shard directories once the last file in them is gone.</summary>
private void TrimEmptyShards(string? leaf)
{
for (var directory = leaf; directory is not null; directory = Path.GetDirectoryName(directory))
{
if (
string.Equals(
Path.GetFullPath(directory).TrimEnd(Path.DirectorySeparatorChar),
Path.GetFullPath(_paths.BlobDirectory).TrimEnd(Path.DirectorySeparatorChar),
StringComparison.OrdinalIgnoreCase
)
)
{
return;
}
try
{
if (Directory.EnumerateFileSystemEntries(directory).Any())
{
return;
}
Directory.Delete(directory);
}
catch (Exception ex) when (ex is IOException or UnauthorizedAccessException)
{
return;
}
}
}
}
@@ -0,0 +1,78 @@
using System.Runtime.InteropServices;
namespace AvParser.Infrastructure.Media;
/// <summary>
/// Creates a second directory entry for an existing file.
/// </summary>
/// <remarks>
/// The BCL has <see cref="File.CreateSymbolicLink(string, string)"/> but nothing for hard links, so
/// this is two P/Invokes. Never throws for an ordinary filesystem refusal — a hard link is a
/// privilege the volume may simply not grant, and the caller's job is to fall back, not to crash.
/// </remarks>
internal static partial class HardLink
{
/// <summary>Windows: the source and destination are on different volumes.</summary>
private const int ErrorNotSameDevice = 17;
/// <summary>Unix: <c>EXDEV</c>, a cross-device link.</summary>
private const int CrossDeviceLink = 18;
/// <summary>
/// Creates a hard link at <paramref name="linkPath"/> pointing at <paramref name="existingPath"/>.
/// </summary>
/// <returns><see langword="true"/> on success; otherwise <see langword="false"/> with a reason.</returns>
public static bool TryCreate(string existingPath, string linkPath, out string? error)
{
ArgumentException.ThrowIfNullOrWhiteSpace(existingPath);
ArgumentException.ThrowIfNullOrWhiteSpace(linkPath);
try
{
// The two platforms take their arguments in opposite orders: Windows names the new
// link first, POSIX names the existing file first. Getting this backwards fails in a
// way that looks like a permissions problem, so it is spelled out at both call sites.
var ok = OperatingSystem.IsWindows()
? CreateHardLinkW(linkPath, existingPath, IntPtr.Zero)
: Link(existingPath, linkPath) == 0;
if (ok)
{
error = null;
return true;
}
var code = Marshal.GetLastPInvokeError();
error = code switch
{
ErrorNotSameDevice when OperatingSystem.IsWindows() => "cross-volume",
CrossDeviceLink when !OperatingSystem.IsWindows() => "cross-volume",
_ => $"error {code.ToString(System.Globalization.CultureInfo.InvariantCulture)}",
};
return false;
}
catch (Exception ex) when (ex is DllNotFoundException or EntryPointNotFoundException)
{
// A platform without either entry point. Nothing to do but fall back to copying.
error = "unsupported";
return false;
}
}
[LibraryImport(
"kernel32.dll",
EntryPoint = "CreateHardLinkW",
SetLastError = true,
StringMarshalling = StringMarshalling.Utf16
)]
[return: MarshalAs(UnmanagedType.Bool)]
private static partial bool CreateHardLinkW(
string lpFileName,
string lpExistingFileName,
IntPtr lpSecurityAttributes
);
[LibraryImport("libc", EntryPoint = "link", SetLastError = true, StringMarshalling = StringMarshalling.Utf8)]
private static partial int Link(string oldpath, string newpath);
}
@@ -0,0 +1,189 @@
using AvParser.Core.Collecting;
using Microsoft.Extensions.Logging;
namespace AvParser.Infrastructure.Media;
/// <summary>
/// The store as the rest of the app sees it: files, index and showcase behind one interface.
/// </summary>
/// <remarks>
/// Ordering is the whole job of this type. The file lands first and the row second, so an
/// interruption leaves an unreferenced file — which the next verify reclaims — rather than a row
/// pointing at nothing, which nothing could repair. Deletions run the other way for the same
/// reason: commit the removal, then unlink.
/// </remarks>
public sealed class MediaStore(
SqliteMediaIndex index,
BlobStore blobs,
ShowcaseLinker showcase,
ILogger<MediaStore> logger
) : IMediaStore
{
private readonly SqliteMediaIndex _index = index ?? throw new ArgumentNullException(nameof(index));
private readonly BlobStore _blobs = blobs ?? throw new ArgumentNullException(nameof(blobs));
private readonly ShowcaseLinker _showcase = showcase ?? throw new ArgumentNullException(nameof(showcase));
private readonly ILogger<MediaStore> _logger = logger ?? throw new ArgumentNullException(nameof(logger));
/// <summary>How showcase entries should point at their blobs.</summary>
/// <remarks>
/// Settable rather than injected for the same reason <c>IProxyPool.Configure</c> is: it changes
/// while the app runs, and threading a snapshot through the container would freeze it at startup.
/// </remarks>
public ShowcaseMode ShowcaseMode { get; private set; } = ShowcaseMode.HardLink;
/// <summary>Applies the user's showcase preference.</summary>
public void Configure(ShowcaseMode mode) => ShowcaseMode = mode;
/// <inheritdoc />
public async Task InitialiseAsync(CancellationToken cancellationToken = default)
{
await _index.InitialiseAsync(cancellationToken).ConfigureAwait(false);
// Anything still staged belongs to a process that is no longer running.
_blobs.SweepTemp(TimeSpan.FromHours(6));
}
/// <inheritdoc />
public Task<string> BeginRunAsync(string sourceId, CancellationToken cancellationToken = default) =>
_index.BeginRunAsync(sourceId, cancellationToken);
/// <inheritdoc />
public Task CompleteRunAsync(string runId, RunSummary summary, CancellationToken cancellationToken = default) =>
_index.CompleteRunAsync(runId, summary, cancellationToken);
/// <inheritdoc />
public Task<IReadOnlyDictionary<string, SeenOutcome>> GetSeenAsync(
string sourceId,
IReadOnlyCollection<string> urls,
CancellationToken cancellationToken = default
) => _index.GetSeenAsync(sourceId, urls, cancellationToken);
/// <inheritdoc />
public Task RecordSeenAsync(
string sourceId,
string url,
SeenOutcome outcome,
int? httpStatus = null,
string? errorCode = null,
CancellationToken cancellationToken = default
) => _index.RecordSeenAsync(sourceId, url, outcome, httpStatus, errorCode, cancellationToken);
/// <inheritdoc />
public Task<IReadOnlySet<string>> LoadTombstonesAsync(CancellationToken cancellationToken = default) =>
_index.LoadTombstonesAsync(cancellationToken);
/// <inheritdoc />
public async Task<int> TombstoneAsync(string sha256, string? reason, CancellationToken cancellationToken = default)
{
var plan = await _index.TombstoneAsync(sha256, reason, cancellationToken).ConfigureAwait(false);
Unlink(plan);
_logger.LogInformation("Marked {Hash} as a placeholder; removed {Count} copy(ies)", sha256, plan.ItemsRemoved);
return plan.ItemsRemoved;
}
/// <inheritdoc />
public async Task<CollectedItem> StoreAsync(
MediaStoreRequest request,
CancellationToken cancellationToken = default
)
{
ArgumentNullException.ThrowIfNull(request);
var promotion = _blobs.Promote(request.Blob, request.TempFilePath);
var entry = _showcase.Create(
request.Candidate.SourceId,
request.Blob,
request.Candidate.SuggestedName,
DateTimeOffset.UtcNow,
ShowcaseMode
);
var status = await _index.RecordStoredAsync(request, entry, cancellationToken).ConfigureAwait(false);
// The file and the index can disagree: the bytes may already have been on disk from an
// interrupted run while the index never learned about them. The index is the authority on
// what counts as new, so its verdict wins and the file promotion is only an optimisation.
if (promotion == BlobPromotion.AlreadyPresent && status == CollectStatus.Stored)
{
_logger.LogDebug("Blob {Hash} was on disk but missing from the index; adopted it", request.Blob.Sha256);
}
return new CollectedItem(request.Candidate, request.Blob, status) { ShowcasePath = entry?.RelativePath };
}
/// <inheritdoc />
public async Task<PurgeResult> PurgeAsync(PurgeOptions options, CancellationToken cancellationToken = default)
{
ArgumentNullException.ThrowIfNull(options);
var plan = await _index.PurgeAsync(options, cancellationToken).ConfigureAwait(false);
var freed = Unlink(plan);
_showcase.RemoveSource(options.SourceId);
_logger.LogInformation(
"Purged {Items} item(s) from {Source}; {Blobs} blob(s) freed",
plan.ItemsRemoved,
options.SourceId,
plan.Orphans.Count
);
return new PurgeResult(plan.ItemsRemoved, plan.Orphans.Count, freed);
}
/// <inheritdoc />
public async Task<int> RebuildShowcaseAsync(string sourceId, CancellationToken cancellationToken = default)
{
ArgumentException.ThrowIfNullOrWhiteSpace(sourceId);
_showcase.RemoveSource(sourceId);
var linked = 0;
await foreach (var item in _index.ReadItemsAsync(sourceId, cancellationToken).ConfigureAwait(false))
{
var blob = MediaBlob.Create(item.Sha256, MediaKind.Unknown, 0) with { Extension = item.Extension };
var entry = _showcase.Create(sourceId, blob, item.SuggestedName, item.CollectedUtc, ShowcaseMode);
await _index.SetShowcaseAsync(sourceId, item.Sha256, entry, cancellationToken).ConfigureAwait(false);
if (entry is not null)
{
linked++;
}
}
_logger.LogInformation("Rebuilt the showcase for {Source}: {Count} entry(ies)", sourceId, linked);
return linked;
}
/// <inheritdoc />
public Task<MediaStoreStats> GetStatsAsync(CancellationToken cancellationToken = default) =>
_index.GetStatsAsync(cancellationToken);
/// <summary>Removes the files a committed removal orphaned.</summary>
/// <returns>Bytes reclaimed.</returns>
private long Unlink(RemovalPlan plan)
{
foreach (var path in plan.ShowcasePaths)
{
_showcase.Remove(path);
}
var freed = 0L;
foreach (var orphan in plan.Orphans)
{
if (_blobs.Delete(orphan.Sha256, orphan.Extension))
{
freed += orphan.Length;
}
}
return freed;
}
}
@@ -0,0 +1,292 @@
using System.Globalization;
using AvParser.Core.Collecting;
using AvParser.Infrastructure.Storage;
using Microsoft.Extensions.Logging;
namespace AvParser.Infrastructure.Media;
/// <summary>A browsable entry pointing at a blob.</summary>
/// <param name="RelativePath">Path under the showcase root, with forward slashes.</param>
/// <param name="Mode">What was actually achieved, which may be less than what was asked for.</param>
public readonly record struct ShowcaseEntry(string RelativePath, ShowcaseMode Mode);
/// <summary>Creates a hard link, reporting failure rather than throwing.</summary>
/// <param name="existingPath">File that already exists.</param>
/// <param name="linkPath">Second name to create for it.</param>
/// <param name="error">Why it failed, when it did.</param>
/// <returns>Whether the link was created.</returns>
/// <remarks>
/// A delegate rather than a direct call so the copy fallback can be exercised on a machine whose
/// filesystem happily supports hard links — otherwise that path is only ever reached in the field.
/// </remarks>
public delegate bool LinkStrategy(string existingPath, string linkPath, out string? error);
/// <summary>
/// Builds a human-browsable tree over the content-addressed blobs.
/// </summary>
/// <remarks>
/// <para>
/// Content addressing is right for storage and useless for looking at: nobody wants to browse
/// <c>blobs/a3/f1/a3f1…png</c>. The showcase gives every item a dated, named path without storing
/// the bytes twice — a hard link is a second name for one file, not a second file.
/// </para>
/// <para>
/// The consequence has to be stated wherever this is surfaced: editing a showcase entry edits the
/// blob, and deleting one frees nothing until the last name for that content is gone.
/// </para>
/// </remarks>
public sealed class ShowcaseLinker(
IAppPaths paths,
BlobStore blobs,
ILogger<ShowcaseLinker> logger,
LinkStrategy? hardLink = null
)
{
/// <summary>Kept well under MAX_PATH once the dated directories and the sequence are added.</summary>
private const int MaxNameLength = 80;
private static readonly string[] ReservedNames =
[
"CON",
"PRN",
"AUX",
"NUL",
"COM1",
"COM2",
"COM3",
"COM4",
"COM5",
"COM6",
"COM7",
"COM8",
"COM9",
"LPT1",
"LPT2",
"LPT3",
"LPT4",
"LPT5",
"LPT6",
"LPT7",
"LPT8",
"LPT9",
];
private readonly IAppPaths _paths = paths ?? throw new ArgumentNullException(nameof(paths));
private readonly BlobStore _blobs = blobs ?? throw new ArgumentNullException(nameof(blobs));
private readonly ILogger<ShowcaseLinker> _logger = logger ?? throw new ArgumentNullException(nameof(logger));
private readonly LinkStrategy _hardLink = hardLink ?? HardLink.TryCreate;
/// <summary>Creates a showcase entry for content already in the blob store.</summary>
/// <returns>The entry, or <see langword="null"/> when none was wanted or possible.</returns>
public ShowcaseEntry? Create(
string sourceId,
MediaBlob blob,
string? suggestedName,
DateTimeOffset when,
ShowcaseMode preferred
)
{
ArgumentException.ThrowIfNullOrWhiteSpace(sourceId);
ArgumentNullException.ThrowIfNull(blob);
if (preferred == ShowcaseMode.None)
{
return null;
}
var target = _blobs.PathFor(blob);
if (!File.Exists(target))
{
_logger.LogWarning("No blob at {Path} to link into the showcase", target);
return null;
}
var directory = Path.Combine(
_paths.ShowcaseDirectory,
SanitiseSegment(sourceId),
when.ToString("yyyy", CultureInfo.InvariantCulture),
when.ToString("MM", CultureInfo.InvariantCulture),
when.ToString("dd", CultureInfo.InvariantCulture)
);
Directory.CreateDirectory(directory);
var name = SanitiseName(suggestedName, blob.Sha256);
var linkPath = NextFreePath(directory, name, blob.Extension);
var achieved = Link(target, linkPath, preferred);
if (achieved == ShowcaseMode.None)
{
return null;
}
var relative = Path.GetRelativePath(_paths.ShowcaseDirectory, linkPath).Replace('\\', '/');
return new ShowcaseEntry(relative, achieved);
}
/// <summary>Removes one showcase entry. The blob it pointed at is untouched.</summary>
public void Remove(string relativePath)
{
if (string.IsNullOrWhiteSpace(relativePath))
{
return;
}
var full = Path.Combine(_paths.ShowcaseDirectory, relativePath.Replace('/', Path.DirectorySeparatorChar));
try
{
if (File.Exists(full))
{
File.Delete(full);
}
}
catch (Exception ex) when (ex is IOException or UnauthorizedAccessException)
{
_logger.LogWarning(ex, "Could not remove showcase entry {Path}", full);
}
}
/// <summary>Removes an entire source's showcase subtree.</summary>
public void RemoveSource(string sourceId)
{
ArgumentException.ThrowIfNullOrWhiteSpace(sourceId);
var directory = Path.Combine(_paths.ShowcaseDirectory, SanitiseSegment(sourceId));
try
{
if (Directory.Exists(directory))
{
Directory.Delete(directory, recursive: true);
}
}
catch (Exception ex) when (ex is IOException or UnauthorizedAccessException)
{
_logger.LogWarning(ex, "Could not remove showcase directory {Path}", directory);
}
}
/// <summary>
/// Turns whatever the origin suggested into something safe to write to disk.
/// </summary>
/// <remarks>
/// The name comes from a remote server, so it is treated as hostile: separators are stripped
/// rather than escaped so that no suggestion can walk out of the directory, Windows device
/// names are pushed out of the way, and anything left unusable falls back to the hash.
/// </remarks>
internal static string SanitiseName(string? suggested, string sha256)
{
var fallback = sha256.Length >= 12 ? sha256[..12] : sha256;
if (string.IsNullOrWhiteSpace(suggested))
{
return fallback;
}
// Take the last segment only: a suggestion of "../../etc/passwd" must not escape.
var candidate = suggested.Replace('\\', '/');
var slash = candidate.LastIndexOf('/');
if (slash >= 0)
{
candidate = candidate[(slash + 1)..];
}
candidate = Path.GetFileNameWithoutExtension(candidate);
var cleaned = new string([
.. candidate.Select(ch => Path.GetInvalidFileNameChars().Contains(ch) || ch < ' ' ? '_' : ch),
]);
cleaned = cleaned.Trim().Trim('.', ' ');
if (cleaned.Length > MaxNameLength)
{
cleaned = cleaned[..MaxNameLength].TrimEnd('.', ' ');
}
if (cleaned.Length == 0 || cleaned.All(ch => ch == '_'))
{
return fallback;
}
// "CON.png" is still the console device on Windows, extension or not.
return ReservedNames.Contains(cleaned, StringComparer.OrdinalIgnoreCase) ? $"_{cleaned}" : cleaned;
}
private static string SanitiseSegment(string segment)
{
var cleaned = new string([.. segment.Select(ch => Path.GetInvalidFileNameChars().Contains(ch) ? '_' : ch)]);
return cleaned.Trim().Trim('.', ' ') is { Length: > 0 } trimmed ? trimmed : "unknown";
}
/// <summary>Numbers entries so browsing order matches collection order.</summary>
private static string NextFreePath(string directory, string name, string extension)
{
var sequence = Directory.EnumerateFiles(directory).Count() + 1;
for (var attempt = 0; attempt < 1000; attempt++)
{
var suffix = attempt == 0 ? string.Empty : $"-{attempt.ToString(CultureInfo.InvariantCulture)}";
var candidate = Path.Combine(
directory,
$"{sequence.ToString("D4", CultureInfo.InvariantCulture)}-{name}{suffix}{extension}"
);
if (!File.Exists(candidate))
{
return candidate;
}
}
return Path.Combine(directory, $"{Guid.NewGuid():N}{extension}");
}
/// <summary>
/// Links, degrading rather than failing.
/// </summary>
/// <remarks>
/// Trying and catching beats inspecting the filesystem type: the answer depends on the volume,
/// the share, the container layer and the user's privileges, and every one of those can be
/// wrong in a way only the actual syscall knows about.
/// </remarks>
private ShowcaseMode Link(string target, string linkPath, ShowcaseMode preferred)
{
if (preferred == ShowcaseMode.HardLink)
{
if (_hardLink(target, linkPath, out var error))
{
return ShowcaseMode.HardLink;
}
_logger.LogDebug("Hard link failed ({Error}); copying instead", error);
}
if (preferred == ShowcaseMode.SymbolicLink)
{
try
{
File.CreateSymbolicLink(linkPath, target);
return ShowcaseMode.SymbolicLink;
}
catch (Exception ex)
when (ex is IOException or UnauthorizedAccessException or PlatformNotSupportedException)
{
_logger.LogDebug(ex, "Symbolic link failed; copying instead");
}
}
try
{
File.Copy(target, linkPath, overwrite: false);
return ShowcaseMode.Copy;
}
catch (Exception ex) when (ex is IOException or UnauthorizedAccessException)
{
_logger.LogWarning(ex, "Could not create showcase entry {Path}", linkPath);
return ShowcaseMode.None;
}
}
}
@@ -0,0 +1,870 @@
using System.Globalization;
using AvParser.Core.Collecting;
using AvParser.Infrastructure.Storage;
using Microsoft.Data.Sqlite;
using Microsoft.Extensions.Logging;
namespace AvParser.Infrastructure.Media;
/// <summary>A blob that lost its last reference and whose file can now be deleted.</summary>
/// <param name="Sha256">Content hash.</param>
/// <param name="Extension">Stored extension, needed to rebuild the path.</param>
/// <param name="Length">Size, for reporting how much was freed.</param>
public readonly record struct OrphanedBlob(string Sha256, string Extension, long Length);
/// <summary>Rows removed by a purge or a tombstone, with what the caller must unlink.</summary>
/// <param name="ItemsRemoved">Provenance rows deleted.</param>
/// <param name="Orphans">Blobs whose last reference went with them.</param>
/// <param name="ShowcasePaths">Showcase entries to remove.</param>
public sealed record RemovalPlan(
int ItemsRemoved,
IReadOnlyList<OrphanedBlob> Orphans,
IReadOnlyList<string> ShowcasePaths
);
/// <summary>One recorded item, as needed to rebuild the showcase.</summary>
/// <param name="Sha256">Content hash.</param>
/// <param name="Extension">Stored extension.</param>
/// <param name="SuggestedName">Name the origin suggested, if any.</param>
/// <param name="CollectedUtc">When it was collected — the showcase is dated by this.</param>
public readonly record struct IndexedItem(
string Sha256,
string Extension,
string? SuggestedName,
DateTimeOffset CollectedUtc
);
/// <summary>
/// The SQLite half of the store: what is held, where it came from, and what has been tried.
/// </summary>
/// <remarks>
/// <para>
/// Writes are serialised behind one semaphore, the same convention the settings and proxy state
/// use. WAL admits one writer and any number of readers, and <c>BEGIN IMMEDIATE</c> makes
/// contention fail at the start of a transaction rather than halfway through it.
/// </para>
/// <para>
/// This type never touches files. Deleting a blob is the caller's job, and it happens after the
/// transaction commits — a crash in between leaves a file with no row, which the store reclaims,
/// rather than a row with no file, which it could not.
/// </para>
/// </remarks>
public sealed class SqliteMediaIndex : IDisposable
{
/// <summary>Bumped when the schema changes in a way that needs migrating.</summary>
private const int CurrentSchemaVersion = 1;
/// <summary>SQLite has a parameter ceiling; batching well under it keeps the query planner happy.</summary>
private const int MaxParametersPerQuery = 400;
private readonly string _connectionString;
private readonly ILogger<SqliteMediaIndex> _logger;
private readonly SemaphoreSlim _writeGate = new(1, 1);
/// <summary>Creates the index over the configured database file.</summary>
public SqliteMediaIndex(IAppPaths paths, ILogger<SqliteMediaIndex> logger)
{
ArgumentNullException.ThrowIfNull(paths);
_logger = logger ?? throw new ArgumentNullException(nameof(logger));
Directory.CreateDirectory(Path.GetDirectoryName(paths.MediaIndexFile)!);
_connectionString = new SqliteConnectionStringBuilder
{
DataSource = paths.MediaIndexFile,
Pooling = true,
ForeignKeys = true,
}.ToString();
}
/// <summary>Creates the schema if it is not there yet.</summary>
public async Task InitialiseAsync(CancellationToken cancellationToken = default)
{
await _writeGate.WaitAsync(cancellationToken).ConfigureAwait(false);
try
{
await using var connection = await OpenAsync(cancellationToken).ConfigureAwait(false);
await ExecuteAsync(connection, Schema, cancellationToken).ConfigureAwait(false);
var version = Convert.ToInt32(
await ScalarAsync(
connection,
"SELECT COALESCE(MAX(version), 0) FROM schema_version;",
cancellationToken
)
.ConfigureAwait(false),
CultureInfo.InvariantCulture
);
if (version == 0)
{
await ExecuteAsync(
connection,
$"INSERT INTO schema_version(version) VALUES({CurrentSchemaVersion});",
cancellationToken
)
.ConfigureAwait(false);
}
else if (version > CurrentSchemaVersion)
{
_logger.LogWarning(
"Media index was written by a newer version ({Found} > {Known}); reading it anyway",
version,
CurrentSchemaVersion
);
}
}
finally
{
_writeGate.Release();
}
}
/// <summary>Opens a run row and returns its id.</summary>
public async Task<string> BeginRunAsync(string sourceId, CancellationToken cancellationToken = default)
{
ArgumentException.ThrowIfNullOrWhiteSpace(sourceId);
var runId = Guid.NewGuid().ToString("N");
await WriteAsync(
async connection =>
{
await using var command = connection.CreateCommand();
command.CommandText = """
INSERT INTO run(id, source_id, started_utc) VALUES($id, $source, $started);
""";
command.Parameters.AddWithValue("$id", runId);
command.Parameters.AddWithValue("$source", sourceId);
command.Parameters.AddWithValue("$started", Now());
await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false);
},
cancellationToken
)
.ConfigureAwait(false);
return runId;
}
/// <summary>Closes a run row with its totals.</summary>
public Task CompleteRunAsync(string runId, RunSummary summary, CancellationToken cancellationToken = default)
{
ArgumentException.ThrowIfNullOrWhiteSpace(runId);
return WriteAsync(
async connection =>
{
await using var command = connection.CreateCommand();
command.CommandText = """
UPDATE run SET finished_utc = $finished, discovered = $discovered, stored = $stored,
duplicates = $duplicates, failed = $failed, bytes = $bytes,
cancelled = $cancelled
WHERE id = $id;
""";
command.Parameters.AddWithValue("$id", runId);
command.Parameters.AddWithValue("$finished", Now());
command.Parameters.AddWithValue("$discovered", summary.Discovered);
command.Parameters.AddWithValue("$stored", summary.Stored);
command.Parameters.AddWithValue("$duplicates", summary.Duplicates);
command.Parameters.AddWithValue("$failed", summary.Failed);
command.Parameters.AddWithValue("$bytes", summary.Bytes);
command.Parameters.AddWithValue("$cancelled", summary.Cancelled ? 1 : 0);
await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false);
},
cancellationToken
);
}
/// <summary>Looks up recorded outcomes for a batch of addresses.</summary>
public async Task<IReadOnlyDictionary<string, SeenOutcome>> GetSeenAsync(
string sourceId,
IReadOnlyCollection<string> urls,
CancellationToken cancellationToken = default
)
{
ArgumentException.ThrowIfNullOrWhiteSpace(sourceId);
ArgumentNullException.ThrowIfNull(urls);
var found = new Dictionary<string, SeenOutcome>(urls.Count, StringComparer.Ordinal);
if (urls.Count == 0)
{
return found;
}
await using var connection = await OpenAsync(cancellationToken).ConfigureAwait(false);
foreach (var batch in urls.Chunk(MaxParametersPerQuery))
{
await using var command = connection.CreateCommand();
var placeholders = string.Join(
',',
batch.Select((_, index) => $"$u{index.ToString(CultureInfo.InvariantCulture)}")
);
command.CommandText =
$"SELECT url, outcome FROM seen_url WHERE source_id = $source AND url IN ({placeholders});";
command.Parameters.AddWithValue("$source", sourceId);
for (var index = 0; index < batch.Length; index++)
{
command.Parameters.AddWithValue($"$u{index.ToString(CultureInfo.InvariantCulture)}", batch[index]);
}
await using var reader = await command.ExecuteReaderAsync(cancellationToken).ConfigureAwait(false);
while (await reader.ReadAsync(cancellationToken).ConfigureAwait(false))
{
found[reader.GetString(0)] = (SeenOutcome)reader.GetInt32(1);
}
}
return found;
}
/// <summary>Records the fate of an address that produced no stored content.</summary>
public Task RecordSeenAsync(
string sourceId,
string url,
SeenOutcome outcome,
int? httpStatus,
string? errorCode,
CancellationToken cancellationToken = default
)
{
ArgumentException.ThrowIfNullOrWhiteSpace(sourceId);
ArgumentException.ThrowIfNullOrWhiteSpace(url);
return WriteAsync(
async connection =>
{
await using var command = connection.CreateCommand();
command.CommandText = SeenUpsert;
BindSeen(command, sourceId, url, outcome, null, httpStatus, errorCode);
await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false);
},
cancellationToken
);
}
/// <summary>Reads every hash marked as a dead-link placeholder.</summary>
public async Task<IReadOnlySet<string>> LoadTombstonesAsync(CancellationToken cancellationToken = default)
{
var hashes = new HashSet<string>(StringComparer.Ordinal);
await using var connection = await OpenAsync(cancellationToken).ConfigureAwait(false);
await using var command = connection.CreateCommand();
command.CommandText = "SELECT sha256 FROM tombstone;";
await using var reader = await command.ExecuteReaderAsync(cancellationToken).ConfigureAwait(false);
while (await reader.ReadAsync(cancellationToken).ConfigureAwait(false))
{
hashes.Add(reader.GetString(0));
}
return hashes;
}
/// <summary>Marks a hash as a placeholder and plans the removal of every copy held.</summary>
public async Task<RemovalPlan> TombstoneAsync(
string sha256,
string? reason,
CancellationToken cancellationToken = default
)
{
ArgumentException.ThrowIfNullOrWhiteSpace(sha256);
RemovalPlan plan = new(0, [], []);
await WriteAsync(
async connection =>
{
await using var transaction = await connection
.BeginTransactionAsync(cancellationToken)
.ConfigureAwait(false);
await using (var insert = connection.CreateCommand())
{
insert.Transaction = (SqliteTransaction)transaction;
insert.CommandText = """
INSERT INTO tombstone(sha256, reason, added_utc, hit_count)
VALUES($sha, $reason, $added, 0)
ON CONFLICT(sha256) DO UPDATE SET hit_count = tombstone.hit_count + 1;
""";
insert.Parameters.AddWithValue("$sha", sha256);
insert.Parameters.AddWithValue("$reason", (object?)reason ?? DBNull.Value);
insert.Parameters.AddWithValue("$added", Now());
await insert.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false);
}
var showcase = await ReadShowcasePathsAsync(
connection,
(SqliteTransaction)transaction,
"SELECT showcase_path FROM item WHERE sha256 = $sha AND showcase_path IS NOT NULL;",
("$sha", sha256),
cancellationToken
)
.ConfigureAwait(false);
int removed;
await using (var delete = connection.CreateCommand())
{
delete.Transaction = (SqliteTransaction)transaction;
delete.CommandText = "DELETE FROM item WHERE sha256 = $sha;";
delete.Parameters.AddWithValue("$sha", sha256);
removed = await delete.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false);
}
var orphans = await CollectOrphansAsync(
connection,
(SqliteTransaction)transaction,
cancellationToken
)
.ConfigureAwait(false);
// The journal keeps the address so a re-run recognises it without downloading.
await using (var mark = connection.CreateCommand())
{
mark.Transaction = (SqliteTransaction)transaction;
mark.CommandText =
$"UPDATE seen_url SET outcome = {(int)SeenOutcome.Placeholder} WHERE sha256 = $sha;";
mark.Parameters.AddWithValue("$sha", sha256);
await mark.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false);
}
await transaction.CommitAsync(cancellationToken).ConfigureAwait(false);
plan = new RemovalPlan(removed, orphans, showcase);
},
cancellationToken
)
.ConfigureAwait(false);
return plan;
}
/// <summary>
/// Records a stored blob and its provenance in one transaction.
/// </summary>
/// <returns>
/// <see cref="CollectStatus.Stored"/> when these bytes were new to the index, otherwise
/// <see cref="CollectStatus.Duplicate"/>.
/// </returns>
public async Task<CollectStatus> RecordStoredAsync(
MediaStoreRequest request,
ShowcaseEntry? showcase,
CancellationToken cancellationToken = default
)
{
ArgumentNullException.ThrowIfNull(request);
var status = CollectStatus.Duplicate;
await WriteAsync(
async connection =>
{
await using var transaction = await connection
.BeginTransactionAsync(cancellationToken)
.ConfigureAwait(false);
var blob = request.Blob;
await using (var insertBlob = connection.CreateCommand())
{
insertBlob.Transaction = (SqliteTransaction)transaction;
insertBlob.CommandText = """
INSERT INTO blob(sha256, kind, extension, length, width, height, is_animated,
first_seen_utc, ref_count)
VALUES($sha, $kind, $ext, $len, $w, $h, $anim, $seen, 0)
ON CONFLICT(sha256) DO NOTHING;
""";
insertBlob.Parameters.AddWithValue("$sha", blob.Sha256);
insertBlob.Parameters.AddWithValue("$kind", (int)blob.Kind);
insertBlob.Parameters.AddWithValue("$ext", blob.Extension);
insertBlob.Parameters.AddWithValue("$len", blob.Length);
insertBlob.Parameters.AddWithValue("$w", (object?)blob.Width ?? DBNull.Value);
insertBlob.Parameters.AddWithValue("$h", (object?)blob.Height ?? DBNull.Value);
insertBlob.Parameters.AddWithValue("$anim", blob.IsAnimated ? 1 : 0);
insertBlob.Parameters.AddWithValue("$seen", Now());
var inserted = await insertBlob.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false);
status = inserted > 0 ? CollectStatus.Stored : CollectStatus.Duplicate;
}
await using (var insertItem = connection.CreateCommand())
{
insertItem.Transaction = (SqliteTransaction)transaction;
insertItem.CommandText = """
INSERT INTO item(sha256, source_id, run_id, request_url, url, referer, external_id,
suggested_name, published_utc, collected_utc, http_status,
content_type, proxy_key, showcase_path, showcase_mode)
VALUES($sha, $source, $run, $request, $url, $referer, $external, $name, $published,
$collected, $status, $type, $proxy, $showcase, $mode)
ON CONFLICT(source_id, request_url) DO UPDATE SET
sha256 = excluded.sha256, url = excluded.url, run_id = excluded.run_id,
collected_utc = excluded.collected_utc, http_status = excluded.http_status,
content_type = excluded.content_type, proxy_key = excluded.proxy_key,
showcase_path = excluded.showcase_path, showcase_mode = excluded.showcase_mode;
""";
insertItem.Parameters.AddWithValue("$sha", blob.Sha256);
insertItem.Parameters.AddWithValue("$source", request.Candidate.SourceId);
insertItem.Parameters.AddWithValue("$run", request.RunId);
insertItem.Parameters.AddWithValue("$request", request.Candidate.Url.AbsoluteUri);
insertItem.Parameters.AddWithValue("$url", request.FinalUrl.AbsoluteUri);
insertItem.Parameters.AddWithValue(
"$referer",
(object?)request.Candidate.Referer?.AbsoluteUri ?? DBNull.Value
);
insertItem.Parameters.AddWithValue(
"$external",
(object?)request.Candidate.ExternalId ?? DBNull.Value
);
insertItem.Parameters.AddWithValue(
"$name",
(object?)request.Candidate.SuggestedName ?? DBNull.Value
);
insertItem.Parameters.AddWithValue(
"$published",
(object?)Format(request.Candidate.PublishedUtc) ?? DBNull.Value
);
insertItem.Parameters.AddWithValue("$collected", Now());
insertItem.Parameters.AddWithValue("$status", request.HttpStatus);
insertItem.Parameters.AddWithValue("$type", (object?)request.ContentType ?? DBNull.Value);
insertItem.Parameters.AddWithValue("$proxy", (object?)request.ProxyKey ?? DBNull.Value);
insertItem.Parameters.AddWithValue(
"$showcase",
(object?)showcase?.RelativePath ?? DBNull.Value
);
insertItem.Parameters.AddWithValue("$mode", (int)(showcase?.Mode ?? ShowcaseMode.None));
await insertItem.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false);
}
// Recomputed rather than incremented: the item upsert above may have replaced a
// row that pointed at a different blob, and a blind +1 would drift for ever.
await using (var recount = connection.CreateCommand())
{
recount.Transaction = (SqliteTransaction)transaction;
recount.CommandText = """
UPDATE blob SET ref_count = (SELECT COUNT(*) FROM item WHERE item.sha256 = blob.sha256)
WHERE sha256 = $sha;
""";
recount.Parameters.AddWithValue("$sha", blob.Sha256);
await recount.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false);
}
await using (var seen = connection.CreateCommand())
{
seen.Transaction = (SqliteTransaction)transaction;
seen.CommandText = SeenUpsert;
BindSeen(
seen,
request.Candidate.SourceId,
request.Candidate.Url.AbsoluteUri,
status == CollectStatus.Stored ? SeenOutcome.Stored : SeenOutcome.Duplicate,
blob.Sha256,
request.HttpStatus,
null
);
await seen.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false);
}
await transaction.CommitAsync(cancellationToken).ConfigureAwait(false);
},
cancellationToken
)
.ConfigureAwait(false);
return status;
}
/// <summary>Removes everything attributed to one source and plans the file deletions.</summary>
public async Task<RemovalPlan> PurgeAsync(PurgeOptions options, CancellationToken cancellationToken = default)
{
ArgumentNullException.ThrowIfNull(options);
RemovalPlan plan = new(0, [], []);
await WriteAsync(
async connection =>
{
await using var transaction = await connection
.BeginTransactionAsync(cancellationToken)
.ConfigureAwait(false);
var showcase = await ReadShowcasePathsAsync(
connection,
(SqliteTransaction)transaction,
"SELECT showcase_path FROM item WHERE source_id = $source AND showcase_path IS NOT NULL;",
("$source", options.SourceId),
cancellationToken
)
.ConfigureAwait(false);
int removed;
await using (var delete = connection.CreateCommand())
{
delete.Transaction = (SqliteTransaction)transaction;
delete.CommandText = "DELETE FROM item WHERE source_id = $source;";
delete.Parameters.AddWithValue("$source", options.SourceId);
removed = await delete.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false);
}
if (options.ForgetSeenUrls)
{
await using var forget = connection.CreateCommand();
forget.Transaction = (SqliteTransaction)transaction;
forget.CommandText = "DELETE FROM seen_url WHERE source_id = $source;";
forget.Parameters.AddWithValue("$source", options.SourceId);
await forget.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false);
}
var orphans = await CollectOrphansAsync(
connection,
(SqliteTransaction)transaction,
cancellationToken
)
.ConfigureAwait(false);
await transaction.CommitAsync(cancellationToken).ConfigureAwait(false);
plan = new RemovalPlan(removed, orphans, showcase);
},
cancellationToken
)
.ConfigureAwait(false);
return plan;
}
/// <summary>Reads everything one source contributed, for rebuilding the showcase.</summary>
public async IAsyncEnumerable<IndexedItem> ReadItemsAsync(
string sourceId,
[System.Runtime.CompilerServices.EnumeratorCancellation] CancellationToken cancellationToken = default
)
{
await using var connection = await OpenAsync(cancellationToken).ConfigureAwait(false);
await using var command = connection.CreateCommand();
command.CommandText = """
SELECT i.sha256, b.extension, i.suggested_name, i.collected_utc
FROM item i JOIN blob b ON b.sha256 = i.sha256
WHERE i.source_id = $source
ORDER BY i.collected_utc, i.id;
""";
command.Parameters.AddWithValue("$source", sourceId);
await using var reader = await command.ExecuteReaderAsync(cancellationToken).ConfigureAwait(false);
while (await reader.ReadAsync(cancellationToken).ConfigureAwait(false))
{
yield return new IndexedItem(
reader.GetString(0),
reader.GetString(1),
reader.IsDBNull(2) ? null : reader.GetString(2),
DateTimeOffset.Parse(reader.GetString(3), CultureInfo.InvariantCulture, DateTimeStyles.RoundtripKind)
);
}
}
/// <summary>Updates one item's showcase entry after a rebuild.</summary>
public Task SetShowcaseAsync(
string sourceId,
string sha256,
ShowcaseEntry? entry,
CancellationToken cancellationToken = default
) =>
WriteAsync(
async connection =>
{
await using var command = connection.CreateCommand();
command.CommandText = """
UPDATE item SET showcase_path = $path, showcase_mode = $mode
WHERE source_id = $source AND sha256 = $sha;
""";
command.Parameters.AddWithValue("$source", sourceId);
command.Parameters.AddWithValue("$sha", sha256);
command.Parameters.AddWithValue("$path", (object?)entry?.RelativePath ?? DBNull.Value);
command.Parameters.AddWithValue("$mode", (int)(entry?.Mode ?? ShowcaseMode.None));
await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false);
},
cancellationToken
);
/// <summary>Totals across the whole store.</summary>
public async Task<MediaStoreStats> GetStatsAsync(CancellationToken cancellationToken = default)
{
await using var connection = await OpenAsync(cancellationToken).ConfigureAwait(false);
await using var command = connection.CreateCommand();
command.CommandText = """
SELECT (SELECT COUNT(*) FROM blob),
(SELECT COUNT(*) FROM item),
(SELECT COALESCE(SUM(length), 0) FROM blob),
(SELECT COUNT(DISTINCT source_id) FROM item);
""";
await using var reader = await command.ExecuteReaderAsync(cancellationToken).ConfigureAwait(false);
if (!await reader.ReadAsync(cancellationToken).ConfigureAwait(false))
{
return default;
}
return new MediaStoreStats(reader.GetInt32(0), reader.GetInt32(1), reader.GetInt64(2), reader.GetInt32(3));
}
/// <summary>Recomputes every reference count and reports how many were wrong.</summary>
/// <remarks>
/// <c>ref_count</c> is denormalised so that purging can find orphans by index instead of
/// scanning. Denormalised state drifts, so the maintenance path that proves it has not is part
/// of the design rather than an afterthought.
/// </remarks>
public async Task<int> VerifyReferenceCountsAsync(CancellationToken cancellationToken = default)
{
var corrected = 0;
await WriteAsync(
async connection =>
{
await using var command = connection.CreateCommand();
command.CommandText = """
UPDATE blob SET ref_count = (SELECT COUNT(*) FROM item WHERE item.sha256 = blob.sha256)
WHERE ref_count <> (SELECT COUNT(*) FROM item WHERE item.sha256 = blob.sha256);
""";
corrected = await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false);
},
cancellationToken
)
.ConfigureAwait(false);
if (corrected > 0)
{
_logger.LogWarning("Corrected {Count} drifted reference count(s) in the media index", corrected);
}
return corrected;
}
/// <inheritdoc />
public void Dispose() => _writeGate.Dispose();
private static async Task<IReadOnlyList<OrphanedBlob>> CollectOrphansAsync(
SqliteConnection connection,
SqliteTransaction transaction,
CancellationToken cancellationToken
)
{
await using (var recount = connection.CreateCommand())
{
recount.Transaction = transaction;
recount.CommandText = """
UPDATE blob SET ref_count = (SELECT COUNT(*) FROM item WHERE item.sha256 = blob.sha256);
""";
await recount.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false);
}
var orphans = new List<OrphanedBlob>();
await using (var select = connection.CreateCommand())
{
select.Transaction = transaction;
select.CommandText = "SELECT sha256, extension, length FROM blob WHERE ref_count <= 0;";
await using var reader = await select.ExecuteReaderAsync(cancellationToken).ConfigureAwait(false);
while (await reader.ReadAsync(cancellationToken).ConfigureAwait(false))
{
orphans.Add(new OrphanedBlob(reader.GetString(0), reader.GetString(1), reader.GetInt64(2)));
}
}
if (orphans.Count > 0)
{
await using var delete = connection.CreateCommand();
delete.Transaction = transaction;
delete.CommandText = "DELETE FROM blob WHERE ref_count <= 0;";
await delete.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false);
}
return orphans;
}
private static async Task<IReadOnlyList<string>> ReadShowcasePathsAsync(
SqliteConnection connection,
SqliteTransaction transaction,
string sql,
(string Name, string Value) parameter,
CancellationToken cancellationToken
)
{
var paths = new List<string>();
await using var command = connection.CreateCommand();
command.Transaction = transaction;
command.CommandText = sql;
command.Parameters.AddWithValue(parameter.Name, parameter.Value);
await using var reader = await command.ExecuteReaderAsync(cancellationToken).ConfigureAwait(false);
while (await reader.ReadAsync(cancellationToken).ConfigureAwait(false))
{
paths.Add(reader.GetString(0));
}
return paths;
}
private static void BindSeen(
SqliteCommand command,
string sourceId,
string url,
SeenOutcome outcome,
string? sha256,
int? httpStatus,
string? errorCode
)
{
command.Parameters.AddWithValue("$source", sourceId);
command.Parameters.AddWithValue("$url", url);
command.Parameters.AddWithValue("$outcome", (int)outcome);
command.Parameters.AddWithValue("$sha", (object?)sha256 ?? DBNull.Value);
command.Parameters.AddWithValue("$status", (object?)httpStatus ?? DBNull.Value);
command.Parameters.AddWithValue("$error", (object?)errorCode ?? DBNull.Value);
command.Parameters.AddWithValue("$now", Now());
}
private static string Now() => DateTimeOffset.UtcNow.ToString("o", CultureInfo.InvariantCulture);
private static string? Format(DateTimeOffset? value) => value?.ToString("o", CultureInfo.InvariantCulture);
private async Task<SqliteConnection> OpenAsync(CancellationToken cancellationToken)
{
var connection = new SqliteConnection(_connectionString);
await connection.OpenAsync(cancellationToken).ConfigureAwait(false);
await ExecuteAsync(connection, Pragmas, cancellationToken).ConfigureAwait(false);
return connection;
}
private async Task WriteAsync(Func<SqliteConnection, Task> work, CancellationToken cancellationToken)
{
await _writeGate.WaitAsync(cancellationToken).ConfigureAwait(false);
try
{
await using var connection = await OpenAsync(cancellationToken).ConfigureAwait(false);
await work(connection).ConfigureAwait(false);
}
finally
{
_writeGate.Release();
}
}
private static async Task ExecuteAsync(SqliteConnection connection, string sql, CancellationToken cancellationToken)
{
await using var command = connection.CreateCommand();
command.CommandText = sql;
await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false);
}
private static async Task<object?> ScalarAsync(
SqliteConnection connection,
string sql,
CancellationToken cancellationToken
)
{
await using var command = connection.CreateCommand();
command.CommandText = sql;
return await command.ExecuteScalarAsync(cancellationToken).ConfigureAwait(false);
}
private const string Pragmas = """
PRAGMA journal_mode = WAL;
PRAGMA synchronous = NORMAL;
PRAGMA foreign_keys = ON;
PRAGMA busy_timeout = 5000;
""";
private const string SeenUpsert = """
INSERT INTO seen_url(source_id, url, outcome, sha256, http_status, error_code, attempts,
first_seen_utc, last_seen_utc)
VALUES($source, $url, $outcome, $sha, $status, $error, 1, $now, $now)
ON CONFLICT(source_id, url) DO UPDATE SET
outcome = excluded.outcome, sha256 = excluded.sha256, http_status = excluded.http_status,
error_code = excluded.error_code, attempts = seen_url.attempts + 1,
last_seen_utc = excluded.last_seen_utc;
""";
private const string Schema = """
CREATE TABLE IF NOT EXISTS schema_version (version INTEGER NOT NULL) STRICT;
CREATE TABLE IF NOT EXISTS blob (
sha256 TEXT PRIMARY KEY NOT NULL,
kind INTEGER NOT NULL,
extension TEXT NOT NULL,
length INTEGER NOT NULL,
width INTEGER NULL,
height INTEGER NULL,
is_animated INTEGER NOT NULL DEFAULT 0,
first_seen_utc TEXT NOT NULL,
ref_count INTEGER NOT NULL DEFAULT 0
) STRICT;
CREATE INDEX IF NOT EXISTS ix_blob_kind ON blob(kind);
CREATE INDEX IF NOT EXISTS ix_blob_orphan ON blob(ref_count) WHERE ref_count <= 0;
CREATE TABLE IF NOT EXISTS item (
id INTEGER PRIMARY KEY AUTOINCREMENT,
sha256 TEXT NOT NULL REFERENCES blob(sha256) ON DELETE CASCADE,
source_id TEXT NOT NULL,
run_id TEXT NOT NULL,
request_url TEXT NOT NULL,
url TEXT NOT NULL,
referer TEXT NULL,
external_id TEXT NULL,
suggested_name TEXT NULL,
published_utc TEXT NULL,
collected_utc TEXT NOT NULL,
http_status INTEGER NOT NULL,
content_type TEXT NULL,
proxy_key TEXT NULL,
showcase_path TEXT NULL,
showcase_mode INTEGER NOT NULL DEFAULT 0
) STRICT;
CREATE UNIQUE INDEX IF NOT EXISTS ux_item_source_request ON item(source_id, request_url);
CREATE INDEX IF NOT EXISTS ix_item_sha ON item(sha256);
CREATE INDEX IF NOT EXISTS ix_item_source_date ON item(source_id, collected_utc);
CREATE INDEX IF NOT EXISTS ix_item_run ON item(run_id);
CREATE TABLE IF NOT EXISTS seen_url (
source_id TEXT NOT NULL,
url TEXT NOT NULL,
outcome INTEGER NOT NULL,
sha256 TEXT NULL,
http_status INTEGER NULL,
error_code TEXT NULL,
attempts INTEGER NOT NULL DEFAULT 1,
first_seen_utc TEXT NOT NULL,
last_seen_utc TEXT NOT NULL,
PRIMARY KEY (source_id, url)
) WITHOUT ROWID, STRICT;
CREATE INDEX IF NOT EXISTS ix_seen_outcome ON seen_url(source_id, outcome);
CREATE TABLE IF NOT EXISTS tombstone (
sha256 TEXT PRIMARY KEY NOT NULL,
reason TEXT NULL,
added_utc TEXT NOT NULL,
hit_count INTEGER NOT NULL DEFAULT 0
) STRICT;
CREATE TABLE IF NOT EXISTS run (
id TEXT PRIMARY KEY NOT NULL,
source_id TEXT NOT NULL,
started_utc TEXT NOT NULL,
finished_utc TEXT NULL,
discovered INTEGER NOT NULL DEFAULT 0,
stored INTEGER NOT NULL DEFAULT 0,
duplicates INTEGER NOT NULL DEFAULT 0,
failed INTEGER NOT NULL DEFAULT 0,
bytes INTEGER NOT NULL DEFAULT 0,
cancelled INTEGER NOT NULL DEFAULT 0
) STRICT;
""";
}
@@ -23,6 +23,30 @@ public interface IAppPaths
/// <summary>Directory holding rolling log files.</summary>
string LogDirectory { get; }
/// <summary>Root of the collected media, holding the blobs, the showcase and the index.</summary>
/// <remarks>
/// The media members are default implementations rather than plain members so that adding them
/// does not break every existing implementor — the test doubles in particular. Anything that
/// wants them somewhere else overrides <see cref="MediaDirectory"/> and gets the rest for free.
/// </remarks>
string MediaDirectory => Path.Combine(DataDirectory, "media");
/// <summary>Content-addressed store: one file per distinct byte sequence.</summary>
string BlobDirectory => Path.Combine(MediaDirectory, "blobs");
/// <summary>Browsable view over the blobs, organised by source and date.</summary>
string ShowcaseDirectory => Path.Combine(MediaDirectory, "showcase");
/// <summary>Where downloads are staged before they are promoted into <see cref="BlobDirectory"/>.</summary>
/// <remarks>
/// Deliberately a sibling of the blobs so that promotion is a rename on the same volume rather
/// than a cross-volume copy of a file that may be tens of megabytes.
/// </remarks>
string MediaTempDirectory => Path.Combine(MediaDirectory, "tmp");
/// <summary>Full path of the SQLite index describing everything collected.</summary>
string MediaIndexFile => Path.Combine(MediaDirectory, "index.db");
}
/// <inheritdoc />
@@ -47,7 +71,13 @@ public sealed class AppPaths : IAppPaths
) { }
/// <summary>Creates paths under an explicit root. Used by tests.</summary>
public AppPaths(string dataDirectory)
/// <param name="dataDirectory">Root for settings, logs and proxy state.</param>
/// <param name="mediaDirectory">
/// Where collected media goes. Defaults to a subdirectory of <paramref name="dataDirectory"/>.
/// A separate parameter because a collection outgrows the profile directory quickly, and
/// pointing it at a second drive is the first thing anyone does.
/// </param>
public AppPaths(string dataDirectory, string? mediaDirectory = null)
{
ArgumentException.ThrowIfNullOrWhiteSpace(dataDirectory);
@@ -56,6 +86,14 @@ public sealed class AppPaths : IAppPaths
CustomProxiesFile = Path.Combine(dataDirectory, "proxies.custom.json");
ProxyStateFile = Path.Combine(dataDirectory, "proxies.state.json");
LogDirectory = Path.Combine(dataDirectory, "logs");
MediaDirectory = string.IsNullOrWhiteSpace(mediaDirectory)
? Path.Combine(dataDirectory, "media")
: mediaDirectory;
BlobDirectory = Path.Combine(MediaDirectory, "blobs");
ShowcaseDirectory = Path.Combine(MediaDirectory, "showcase");
MediaTempDirectory = Path.Combine(MediaDirectory, "tmp");
MediaIndexFile = Path.Combine(MediaDirectory, "index.db");
}
/// <inheritdoc />
@@ -73,10 +111,28 @@ public sealed class AppPaths : IAppPaths
/// <inheritdoc />
public string LogDirectory { get; }
/// <inheritdoc />
public string MediaDirectory { get; }
/// <inheritdoc />
public string BlobDirectory { get; }
/// <inheritdoc />
public string ShowcaseDirectory { get; }
/// <inheritdoc />
public string MediaTempDirectory { get; }
/// <inheritdoc />
public string MediaIndexFile { get; }
/// <summary>Creates every directory this instance points at.</summary>
public void EnsureCreated()
{
Directory.CreateDirectory(DataDirectory);
Directory.CreateDirectory(LogDirectory);
Directory.CreateDirectory(BlobDirectory);
Directory.CreateDirectory(ShowcaseDirectory);
Directory.CreateDirectory(MediaTempDirectory);
}
}
@@ -0,0 +1,351 @@
using System.Security.Cryptography;
using System.Text;
using AvParser.Core.Collecting;
using AvParser.Infrastructure.Media;
using AvParser.Infrastructure.Storage;
using Microsoft.Extensions.Logging.Abstractions;
namespace AvParser.Infrastructure.Tests.Media;
public sealed class MediaStoreTests : 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!;
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 = Build();
await _store.InitialiseAsync(TestContext.Current.CancellationToken);
}
public ValueTask DisposeAsync()
{
_index.Dispose();
// Pooled SQLite connections keep the file open; without this the delete races the pool.
Microsoft.Data.Sqlite.SqliteConnection.ClearAllPools();
if (Directory.Exists(_root))
{
try
{
Directory.Delete(_root, recursive: true);
}
catch (IOException)
{
// A leaked handle is not worth failing a green test over.
}
}
return ValueTask.CompletedTask;
}
private MediaStore Build(LinkStrategy? hardLink = null) =>
new(
_index,
_blobs,
new ShowcaseLinker(_paths, _blobs, NullLogger<ShowcaseLinker>.Instance, hardLink),
NullLogger<MediaStore>.Instance
);
/// <summary>Writes bytes to a staged temp file and returns the blob describing them.</summary>
private (MediaBlob Blob, string TempPath) Stage(string content, MediaKind kind = MediaKind.Png)
{
var bytes = Encoding.UTF8.GetBytes(content);
var hash = Convert.ToHexStringLower(SHA256.HashData(bytes));
var temp = _blobs.CreateTempPath();
File.WriteAllBytes(temp, bytes);
return (MediaBlob.Create(hash, kind, bytes.Length), temp);
}
private async Task<CollectedItem> StoreAsync(
string content,
string sourceId = "url-list",
string? url = null,
string? suggestedName = null,
MediaStore? store = null
)
{
var (blob, temp) = Stage(content);
var candidate = new MediaCandidate(new Uri(url ?? $"https://example.test/{content}.png"))
{
SourceId = sourceId,
SuggestedName = suggestedName,
};
var request = new MediaStoreRequest(candidate, blob, temp, "run-1", candidate.Url, 200);
return await (store ?? _store).StoreAsync(request, TestContext.Current.CancellationToken);
}
[Fact]
public async Task Storing_writes_one_blob_and_reports_it_as_new()
{
var item = await StoreAsync("alpha");
item.Status.ShouldBe(CollectStatus.Stored);
File.Exists(_blobs.PathFor(item.Blob)).ShouldBeTrue();
var stats = await _store.GetStatsAsync(TestContext.Current.CancellationToken);
stats.BlobCount.ShouldBe(1);
stats.ItemCount.ShouldBe(1);
}
[Fact]
public async Task The_same_bytes_at_two_addresses_produce_one_blob_and_two_provenance_rows()
{
// The point of hashing the content instead of the address: the web re-uploads constantly.
var first = await StoreAsync("same", url: "https://example.test/a.png");
var second = await StoreAsync("same", url: "https://example.test/b.png");
first.Status.ShouldBe(CollectStatus.Stored);
second.Status.ShouldBe(CollectStatus.Duplicate);
first.Blob.Sha256.ShouldBe(second.Blob.Sha256);
var stats = await _store.GetStatsAsync(TestContext.Current.CancellationToken);
stats.BlobCount.ShouldBe(1);
stats.ItemCount.ShouldBe(2);
}
[Fact]
public async Task Re_collecting_the_same_address_does_not_add_a_second_row()
{
await StoreAsync("alpha", url: "https://example.test/a.png");
await StoreAsync("alpha", url: "https://example.test/a.png");
(await _store.GetStatsAsync(TestContext.Current.CancellationToken)).ItemCount.ShouldBe(1);
}
[Fact]
public async Task A_stored_address_is_journalled_so_a_re_run_can_skip_it()
{
await StoreAsync("alpha", url: "https://example.test/a.png");
var seen = await _store.GetSeenAsync(
"url-list",
["https://example.test/a.png"],
TestContext.Current.CancellationToken
);
seen["https://example.test/a.png"].ShouldBe(SeenOutcome.Stored);
SeenOutcomes.IsTerminal(seen["https://example.test/a.png"]).ShouldBeTrue();
}
[Fact]
public async Task Repeated_failures_at_one_address_count_up_without_becoming_terminal()
{
// A failure describes the moment, not the resource; treating it as final would make a
// flaky network permanently lose content.
for (var attempt = 0; attempt < 3; attempt++)
{
await _store.RecordSeenAsync(
"url-list",
"https://example.test/flaky.png",
SeenOutcome.Failed,
cancellationToken: TestContext.Current.CancellationToken
);
}
var seen = await _store.GetSeenAsync(
"url-list",
["https://example.test/flaky.png"],
TestContext.Current.CancellationToken
);
seen["https://example.test/flaky.png"].ShouldBe(SeenOutcome.Failed);
SeenOutcomes.IsTerminal(SeenOutcome.Failed).ShouldBeFalse();
}
[Fact]
public async Task Purging_a_source_leaves_a_blob_that_another_source_still_references()
{
// This is exactly what the reference count buys; without it the second source would find
// its own index rows pointing at a file that the first source's purge deleted.
await StoreAsync("shared", sourceId: "url-list", url: "https://example.test/a.png");
await StoreAsync("shared", sourceId: "own-service", url: "https://own.test/a.png");
var result = await _store.PurgeAsync(new PurgeOptions("url-list"), TestContext.Current.CancellationToken);
result.ItemsRemoved.ShouldBe(1);
result.BlobsRemoved.ShouldBe(0);
var stats = await _store.GetStatsAsync(TestContext.Current.CancellationToken);
stats.BlobCount.ShouldBe(1);
stats.ItemCount.ShouldBe(1);
}
[Fact]
public async Task Purging_the_last_reference_deletes_the_file()
{
var item = await StoreAsync("lonely", sourceId: "url-list");
var path = _blobs.PathFor(item.Blob);
var result = await _store.PurgeAsync(new PurgeOptions("url-list"), TestContext.Current.CancellationToken);
result.BlobsRemoved.ShouldBe(1);
result.BytesFreed.ShouldBe(item.Blob.Length);
File.Exists(path).ShouldBeFalse();
}
[Fact]
public async Task Purging_keeps_the_journal_unless_asked_to_forget_it()
{
await StoreAsync("alpha", url: "https://example.test/a.png");
await _store.PurgeAsync(new PurgeOptions("url-list"), TestContext.Current.CancellationToken);
// Deliberate: otherwise the next run downloads again exactly what was just deleted.
var kept = await _store.GetSeenAsync(
"url-list",
["https://example.test/a.png"],
TestContext.Current.CancellationToken
);
kept.ShouldContainKey("https://example.test/a.png");
await _store.PurgeAsync(
new PurgeOptions("url-list") { ForgetSeenUrls = true },
TestContext.Current.CancellationToken
);
var forgotten = await _store.GetSeenAsync(
"url-list",
["https://example.test/a.png"],
TestContext.Current.CancellationToken
);
forgotten.ShouldBeEmpty();
}
[Fact]
public async Task Marking_a_placeholder_removes_every_copy_already_held()
{
var first = await StoreAsync("placeholder", url: "https://example.test/a.png");
await StoreAsync("placeholder", url: "https://example.test/b.png");
var path = _blobs.PathFor(first.Blob);
var removed = await _store.TombstoneAsync(
first.Blob.Sha256,
"dead link image",
TestContext.Current.CancellationToken
);
removed.ShouldBe(2);
File.Exists(path).ShouldBeFalse();
(await _store.LoadTombstonesAsync(TestContext.Current.CancellationToken)).ShouldContain(first.Blob.Sha256);
// And the addresses are remembered as placeholders, so a re-run does not fetch them again.
var seen = await _store.GetSeenAsync(
"url-list",
["https://example.test/a.png"],
TestContext.Current.CancellationToken
);
seen["https://example.test/a.png"].ShouldBe(SeenOutcome.Placeholder);
}
[Fact]
public async Task A_showcase_entry_is_the_same_content_and_deleting_it_leaves_the_blob()
{
var item = await StoreAsync("linked", suggestedName: "kitten");
item.ShowcasePath.ShouldNotBeNull();
var showcasePath = Path.Combine(
_paths.ShowcaseDirectory,
item.ShowcasePath!.Replace('/', Path.DirectorySeparatorChar)
);
File.ReadAllBytes(showcasePath).ShouldBe(File.ReadAllBytes(_blobs.PathFor(item.Blob)));
item.ShowcasePath.ShouldContain("kitten");
File.Delete(showcasePath);
File.Exists(_blobs.PathFor(item.Blob)).ShouldBeTrue();
}
[Fact]
public async Task A_filesystem_without_hard_links_falls_back_to_copying()
{
// Forced, because the developer's own disk supports hard links and this path would
// otherwise only ever run on a user's exFAT drive or network share.
var store = Build(
(string _, string _, out string? error) =>
{
error = "cross-volume";
return false;
}
);
var item = await StoreAsync("copied", store: store);
item.ShowcasePath.ShouldNotBeNull();
var showcasePath = Path.Combine(
_paths.ShowcaseDirectory,
item.ShowcasePath!.Replace('/', Path.DirectorySeparatorChar)
);
File.Exists(showcasePath).ShouldBeTrue();
File.ReadAllBytes(showcasePath).ShouldBe(File.ReadAllBytes(_blobs.PathFor(item.Blob)));
}
[Fact]
public async Task The_showcase_can_be_rebuilt_after_the_user_deletes_it()
{
await StoreAsync("one", url: "https://example.test/1.png", suggestedName: "one");
await StoreAsync("two", url: "https://example.test/2.png", suggestedName: "two");
Directory.Delete(_paths.ShowcaseDirectory, recursive: true);
var linked = await _store.RebuildShowcaseAsync("url-list", TestContext.Current.CancellationToken);
linked.ShouldBe(2);
Directory.EnumerateFiles(_paths.ShowcaseDirectory, "*", SearchOption.AllDirectories).Count().ShouldBe(2);
}
[Fact]
public async Task Reference_counts_do_not_drift()
{
await StoreAsync("a", url: "https://example.test/1.png");
await StoreAsync("a", url: "https://example.test/2.png");
await StoreAsync("b", url: "https://example.test/3.png");
await _store.PurgeAsync(new PurgeOptions("nobody"), TestContext.Current.CancellationToken);
(await _index.VerifyReferenceCountsAsync(TestContext.Current.CancellationToken)).ShouldBe(0);
}
[Fact]
public async Task A_run_records_what_it_did()
{
var runId = await _store.BeginRunAsync("url-list", TestContext.Current.CancellationToken);
runId.ShouldNotBeNullOrWhiteSpace();
await _store.CompleteRunAsync(
runId,
new RunSummary(10, 7, 2, 1, 4096, Cancelled: false),
TestContext.Current.CancellationToken
);
}
[Fact]
public async Task An_unfinished_download_from_a_previous_process_is_swept()
{
var stale = _blobs.CreateTempPath();
await File.WriteAllTextAsync(stale, "half a picture", TestContext.Current.CancellationToken);
File.SetLastWriteTimeUtc(stale, DateTime.UtcNow.AddDays(-1));
var fresh = _blobs.CreateTempPath();
await File.WriteAllTextAsync(fresh, "in progress right now", TestContext.Current.CancellationToken);
_blobs.SweepTemp(TimeSpan.FromHours(6)).ShouldBe(1);
File.Exists(stale).ShouldBeFalse();
File.Exists(fresh).ShouldBeTrue();
}
}
@@ -0,0 +1,76 @@
using AvParser.Infrastructure.Media;
namespace AvParser.Infrastructure.Tests.Media;
public class ShowcaseNameTests
{
private const string Hash = "a3f19b7c5d2e8f0142536475869708192a3b4c5d6e7f8091a2b3c4d5e6f70819";
[Fact]
public void A_plain_name_is_kept()
{
ShowcaseLinker.SanitiseName("kitten", Hash).ShouldBe("kitten");
}
[Fact]
public void An_absent_name_falls_back_to_the_hash()
{
ShowcaseLinker.SanitiseName(null, Hash).ShouldBe(Hash[..12]);
ShowcaseLinker.SanitiseName(" ", Hash).ShouldBe(Hash[..12]);
}
[Theory]
[InlineData("../../etc/passwd")]
[InlineData("..\\..\\windows\\system32\\config")]
[InlineData("/absolute/path/name")]
public void A_name_cannot_walk_out_of_its_directory(string suggested)
{
// The suggestion comes from a remote server, so it is treated as hostile rather than
// merely untidy: only the last segment survives, and separators never do.
var name = ShowcaseLinker.SanitiseName(suggested, Hash);
name.ShouldNotContain("/");
name.ShouldNotContain("\\");
name.ShouldNotContain("..");
}
[Fact]
public void The_extension_the_origin_suggested_is_dropped()
{
// The extension is derived from the sniffed kind, never from what the origin claimed.
ShowcaseLinker.SanitiseName("photo.exe", Hash).ShouldBe("photo");
}
[Theory]
[InlineData("CON")]
[InlineData("nul")]
[InlineData("COM1")]
[InlineData("lpt9")]
public void Windows_device_names_are_pushed_out_of_the_way(string reserved)
{
ShowcaseLinker.SanitiseName(reserved, Hash).ShouldBe($"_{reserved}");
}
[Fact]
public void An_over_long_name_is_trimmed()
{
var name = ShowcaseLinker.SanitiseName(new string('x', 400), Hash);
name.Length.ShouldBeLessThanOrEqualTo(80);
}
[Fact]
public void A_name_of_nothing_but_junk_falls_back_to_the_hash()
{
// Dots rather than, say, '?': the set of invalid filename characters differs by platform,
// but a name that trims away to nothing does so everywhere.
ShowcaseLinker.SanitiseName("...", Hash).ShouldBe(Hash[..12]);
ShowcaseLinker.SanitiseName("\t\t", Hash).ShouldBe(Hash[..12]);
}
[Fact]
public void Control_characters_are_replaced_but_spaces_survive()
{
ShowcaseLinker.SanitiseName("a b\tc", Hash).ShouldBe("a b_c");
}
}