diff --git a/Directory.Packages.props b/Directory.Packages.props index 4f63d97..0cee536 100644 --- a/Directory.Packages.props +++ b/Directory.Packages.props @@ -10,6 +10,9 @@ 24.1.0 7.1.1 10.0.11 + + 10.0.11 @@ -46,6 +49,12 @@ + + + + + diff --git a/src/AvParser.Core/Collecting/CollectedItem.cs b/src/AvParser.Core/Collecting/CollectedItem.cs new file mode 100644 index 0000000..ced96b0 --- /dev/null +++ b/src/AvParser.Core/Collecting/CollectedItem.cs @@ -0,0 +1,94 @@ +namespace AvParser.Core.Collecting; + +/// What happened to one candidate. +public enum CollectStatus +{ + /// New content; a blob was written. + Stored = 0, + + /// The bytes were already in the store; only provenance was recorded. + Duplicate = 1, + + /// Not fetched at all, because a previous run already settled this address. + Skipped = 2, +} + +/// One candidate carried all the way through to the store. +/// What the source found. +/// The content that came back. +/// Whether it was new, already held, or never fetched. +public sealed record CollectedItem(MediaCandidate Candidate, MediaBlob Blob, CollectStatus Status) +{ + /// How long the fetch took, including redirects. + public TimeSpan Elapsed { get; init; } + + /// Path of the showcase entry, relative to the showcase root; null when none was made. + public string? ShowcasePath { get; init; } +} + +/// +/// The recorded fate of an address, so a later run need not ask again. +/// +/// +/// Split into terminal and retryable by . 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. +/// +public enum SeenOutcome +{ + /// Downloaded and stored. + Stored = 0, + + /// Downloaded; the bytes were already held. + Duplicate = 1, + + /// Content matched a known dead-link placeholder. + Placeholder = 2, + + /// Bigger than the configured ceiling. + TooLarge = 3, + + /// Smaller than the configured floor — tracking pixels and spacers. + TooSmall = 4, + + /// A recognised format that the user excluded. + UnsupportedType = 5, + + /// The bytes matched no known signature. + NotMedia = 6, + + /// The origin said it is permanently gone. + Gone = 7, + + /// Failed for a reason that may not recur. + Failed = 8, + + /// Deferred by the origin's own rate limiting. + RateLimited = 9, + + /// Timed out. + Timeout = 10, +} + +/// Helpers over . +public static class SeenOutcomes +{ + /// + /// Whether the outcome settles the address for good, so a re-run can skip it without asking. + /// + /// + /// Deliberately excludes , + /// and : those describe the moment, not the resource, and a + /// journal that treated them as final would make a flaky network permanently lose content. + /// + 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; +} diff --git a/src/AvParser.Core/Collecting/IMediaStore.cs b/src/AvParser.Core/Collecting/IMediaStore.cs new file mode 100644 index 0000000..fa618db --- /dev/null +++ b/src/AvParser.Core/Collecting/IMediaStore.cs @@ -0,0 +1,137 @@ +namespace AvParser.Core.Collecting; + +/// A finished download waiting to be taken into the store. +/// What the source found. +/// Hash, kind and size of the content that came back. +/// +/// Complete, verified file to take ownership of. The store moves or deletes it and the caller must +/// not touch it afterwards. +/// +/// Run this belongs to. +/// Address after redirects; is the one asked for. +/// Status of the final response. +public sealed record MediaStoreRequest( + MediaCandidate Candidate, + MediaBlob Blob, + string TempFilePath, + string RunId, + Uri FinalUrl, + int HttpStatus +) +{ + /// What the origin claimed the type was. Kept precisely to catch the ones that lie. + public string? ContentType { get; init; } + + /// Proxy the content came through, or null when the connection was direct. + public string? ProxyKey { get; init; } +} + +/// What to remove when purging. +/// Only content attributed to this source is considered. +public sealed record PurgeOptions(string SourceId) +{ + /// + /// Also forget every address this source ever visited. + /// + /// + /// 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. + /// + public bool ForgetSeenUrls { get; init; } +} + +/// Outcome of a purge. +/// Provenance rows deleted. +/// Blobs that lost their last reference and were deleted. +/// Bytes reclaimed on disk. +public readonly record struct PurgeResult(int ItemsRemoved, int BlobsRemoved, long BytesFreed); + +/// Totals for the dashboard. +/// Distinct byte sequences held. +/// Provenance rows across every source. +/// Sum of blob sizes. +/// Distinct sources that contributed. +public readonly record struct MediaStoreStats(int BlobCount, int ItemCount, long TotalBytes, int SourceCount); + +/// How a run ended. +/// Candidates the source produced. +/// Candidates that produced new content. +/// Candidates whose bytes were already held. +/// Candidates that produced an error. +/// Bytes newly written. +/// Whether the user stopped it. +public readonly record struct RunSummary( + int Discovered, + int Stored, + int Duplicates, + int Failed, + long Bytes, + bool Cancelled +); + +/// +/// Everything the collector persists: the blobs, their provenance, and the journal of addresses +/// already visited. +/// +/// +/// The abstraction lives in the domain and the implementation in the infrastructure, the same way +/// ISettingsService and IProxySource do, so that the domain stays runnable from a +/// CLI or a test without dragging in a database. +/// +public interface IMediaStore +{ + /// Creates or upgrades the schema. Safe to call repeatedly. + Task InitialiseAsync(CancellationToken cancellationToken = default); + + /// Opens a run and returns its id. + Task BeginRunAsync(string sourceId, CancellationToken cancellationToken = default); + + /// Closes a run with its totals. + Task CompleteRunAsync(string runId, RunSummary summary, CancellationToken cancellationToken = default); + + /// + /// Looks up what previous runs made of these addresses. + /// + /// + /// Batched deliberately: a listing of a hundred thousand addresses must cost a handful of + /// queries, not a hundred thousand of them. + /// + Task> GetSeenAsync( + string sourceId, + IReadOnlyCollection urls, + CancellationToken cancellationToken = default + ); + + /// Records the fate of an address that produced no content. + Task RecordSeenAsync( + string sourceId, + string url, + SeenOutcome outcome, + int? httpStatus = null, + string? errorCode = null, + CancellationToken cancellationToken = default + ); + + /// Reads every hash known to be a dead-link placeholder. + /// Loaded once per run: the set is tiny and the check runs on every download. + Task> LoadTombstonesAsync(CancellationToken cancellationToken = default); + + /// + /// Marks a hash as a placeholder and removes every copy already held. + /// + /// How many stored items were removed. + Task TombstoneAsync(string sha256, string? reason, CancellationToken cancellationToken = default); + + /// Takes ownership of a completed download and records it. + Task StoreAsync(MediaStoreRequest request, CancellationToken cancellationToken = default); + + /// Removes everything attributed to one source. + Task PurgeAsync(PurgeOptions options, CancellationToken cancellationToken = default); + + /// Rebuilds the browsable showcase for a source from the recorded provenance. + /// How many entries were linked. + Task RebuildShowcaseAsync(string sourceId, CancellationToken cancellationToken = default); + + /// Totals across the whole store. + Task GetStatsAsync(CancellationToken cancellationToken = default); +} diff --git a/src/AvParser.Core/Collecting/MediaBlob.cs b/src/AvParser.Core/Collecting/MediaBlob.cs new file mode 100644 index 0000000..f2f5fd5 --- /dev/null +++ b/src/AvParser.Core/Collecting/MediaBlob.cs @@ -0,0 +1,40 @@ +namespace AvParser.Core.Collecting; + +/// +/// One distinct byte sequence in the store, identified by its content. +/// +/// +/// 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 decoded bytes — so the same image +/// served gzipped at one URL and plain at another still collapses to a single blob. +/// +/// Lowercase hex SHA-256 of the content, 64 characters. +/// What the signature said it is. +/// Extension derived from , leading dot included. +/// Size in bytes of the decoded content. +public sealed record MediaBlob(string Sha256, MediaKind Kind, string Extension, long Length) +{ + /// Pixel width, when it was cheap to read from the header. + public int? Width { get; init; } + + /// Pixel height, when it was cheap to read from the header. + public int? Height { get; init; } + + /// Whether the content has more than one frame. + public bool IsAnimated { get; init; } + + /// Creates a blob, deriving the extension from the kind. + public static MediaBlob Create(string sha256, MediaKind kind, long length) => + new(sha256, kind, MediaKinds.ExtensionFor(kind), length); + + /// + /// Path of this blob relative to the blob root, using two levels of two hex characters. + /// + /// + /// 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. + /// + public string RelativePath => Path.Combine(Sha256[..2], Sha256.Substring(2, 2), string.Concat(Sha256, Extension)); +} diff --git a/src/AvParser.Core/Collecting/MediaCandidate.cs b/src/AvParser.Core/Collecting/MediaCandidate.cs new file mode 100644 index 0000000..2b0b4ae --- /dev/null +++ b/src/AvParser.Core/Collecting/MediaCandidate.cs @@ -0,0 +1,38 @@ +namespace AvParser.Core.Collecting; + +/// +/// Something a source found and thinks is worth downloading. +/// +/// +/// 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. +/// +/// Where to fetch it from. +public sealed record MediaCandidate(Uri Url) +{ + /// Id of the source that found it. Stamped by the runner when a source leaves it blank. + public string SourceId { get; init; } = string.Empty; + + /// Page to send as Referer; some origins refuse a bare request. + public Uri? Referer { get; init; } + + /// Identifier in the originating service, when it has one. + public string? ExternalId { get; init; } + + /// Name to prefer in the showcase, before sanitising. + public string? SuggestedName { get; init; } + + /// When the origin says it was published. + public DateTimeOffset? PublishedUtc { get; init; } + + /// Size the listing claimed, when it said. Advisory only — never trusted for limits. + public long? ExpectedLength { get; init; } + + /// Tags carried over from the source. + public IReadOnlyList Tags { get; init; } = []; + + /// 1-based position within the listing, used to report failures against something. + public int Ordinal { get; init; } +} diff --git a/src/AvParser.Core/Collecting/MediaKind.cs b/src/AvParser.Core/Collecting/MediaKind.cs new file mode 100644 index 0000000..bc22774 --- /dev/null +++ b/src/AvParser.Core/Collecting/MediaKind.cs @@ -0,0 +1,62 @@ +namespace AvParser.Core.Collecting; + +/// What a downloaded byte sequence actually turned out to be. +/// +/// Decided by the file's own signature, never by its URL, extension or Content-Type — all +/// three are routinely wrong, and two of them are attacker-controlled. +/// +public enum MediaKind +{ + /// Nothing recognised. Never stored. + Unknown = 0, + + /// JPEG image. + Jpeg = 1, + + /// PNG image, possibly animated (APNG). + Png = 2, + + /// GIF image, possibly animated. + Gif = 3, + + /// WebP image, possibly animated. + WebP = 4, + + /// AVIF image. + Avif = 5, + + /// MP4 video. What most sites serve when they say "GIF". + Mp4 = 6, + + /// WebM video. + WebM = 7, +} + +/// Helpers over . +public static class MediaKinds +{ + /// File extension for a kind, leading dot included. + /// + /// 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. + /// + 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", + }; + + /// Whether the kind is a still or animated picture rather than a video container. + public static bool IsImage(MediaKind kind) => + kind is MediaKind.Jpeg or MediaKind.Png or MediaKind.Gif or MediaKind.WebP or MediaKind.Avif; + + /// Whether the kind is a video container. + public static bool IsVideo(MediaKind kind) => kind is MediaKind.Mp4 or MediaKind.WebM; +} diff --git a/src/AvParser.Core/Collecting/MediaQuery.cs b/src/AvParser.Core/Collecting/MediaQuery.cs new file mode 100644 index 0000000..aa4311b --- /dev/null +++ b/src/AvParser.Core/Collecting/MediaQuery.cs @@ -0,0 +1,23 @@ +namespace AvParser.Core.Collecting; + +/// +/// What to ask a source for. +/// +/// +/// 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. +/// +/// Free text, typically a pasted list of addresses, one per line. +/// Listing endpoint, for sources that ask a service what it holds. +/// Stop after this many candidates; 0 means no cap. +/// Where to resume a paginated listing. +public sealed record MediaQuery(string Text = "", Uri? Endpoint = null, int Limit = 0, string? Cursor = null) +{ + /// Source-specific extras, so a new source needs no change to this type. + public IReadOnlyDictionary Options { get; init; } = + new Dictionary(StringComparer.OrdinalIgnoreCase); + + /// Whether the caller asked for a bounded number of candidates. + public bool HasLimit => Limit > 0; +} diff --git a/src/AvParser.Core/Collecting/ShowcaseMode.cs b/src/AvParser.Core/Collecting/ShowcaseMode.cs new file mode 100644 index 0000000..c62b7f7 --- /dev/null +++ b/src/AvParser.Core/Collecting/ShowcaseMode.cs @@ -0,0 +1,25 @@ +namespace AvParser.Core.Collecting; + +/// +/// How a showcase entry points at the blob it shows. +/// +/// +/// 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. +/// +public enum ShowcaseMode +{ + /// No browsable entry; only the content-addressed blob exists. + None = 0, + + /// A second directory entry for the same content. Costs no extra space. + HardLink = 1, + + /// A symbolic link. Needs Developer Mode or elevation on Windows, so it is opt-in. + SymbolicLink = 2, + + /// An independent copy. Doubles the space used. + Copy = 3, +} diff --git a/src/AvParser.Infrastructure/AvParser.Infrastructure.csproj b/src/AvParser.Infrastructure/AvParser.Infrastructure.csproj index 2414d0d..8080aa7 100644 --- a/src/AvParser.Infrastructure/AvParser.Infrastructure.csproj +++ b/src/AvParser.Infrastructure/AvParser.Infrastructure.csproj @@ -1,6 +1,11 @@ AvParser.Infrastructure + + true @@ -14,6 +19,7 @@ + diff --git a/src/AvParser.Infrastructure/Media/BlobStore.cs b/src/AvParser.Infrastructure/Media/BlobStore.cs new file mode 100644 index 0000000..91059d6 --- /dev/null +++ b/src/AvParser.Infrastructure/Media/BlobStore.cs @@ -0,0 +1,209 @@ +using AvParser.Core.Collecting; +using AvParser.Infrastructure.Storage; +using Microsoft.Extensions.Logging; + +namespace AvParser.Infrastructure.Media; + +/// Whether a promotion actually wrote anything. +public enum BlobPromotion +{ + /// The file was moved into the store; these bytes were new. + Written = 0, + + /// The content was already held, so the temporary file was discarded. + AlreadyPresent = 1, +} + +/// +/// The files on disk: content-addressed, sharded, and written only once complete. +/// +/// +/// The invariant this type exists to hold is that blobs/ 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. +/// +public sealed class BlobStore(IAppPaths paths, ILogger logger) +{ + private readonly IAppPaths _paths = paths ?? throw new ArgumentNullException(nameof(paths)); + private readonly ILogger _logger = logger ?? throw new ArgumentNullException(nameof(logger)); + + /// Absolute path this content would live at. + public string PathFor(MediaBlob blob) + { + ArgumentNullException.ThrowIfNull(blob); + + return Path.Combine(_paths.BlobDirectory, blob.RelativePath); + } + + /// Absolute path for a hash and extension already known. + 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) + ); + } + + /// Whether this content is already held. + public bool Exists(MediaBlob blob) => File.Exists(PathFor(blob)); + + /// Reserves a path for an in-progress download. + /// + /// The .part 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. + /// + public string CreateTempPath() + { + Directory.CreateDirectory(_paths.MediaTempDirectory); + + return Path.Combine(_paths.MediaTempDirectory, $"{Guid.NewGuid():N}.part"); + } + + /// + /// Takes ownership of a completed temporary file and moves it into the store. + /// + /// + /// 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. + /// + 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; + } + } + + /// Removes content from disk. Missing is success — the caller wants it gone. + 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; + } + + /// + /// Deletes temporary files left behind by a previous process. + /// + /// How many were removed. + /// + /// Only files older than are touched, so a sweep at startup + /// cannot delete a download that another instance of the app is running right now. + /// + 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; + } + } + + /// Removes the two shard directories once the last file in them is gone. + 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; + } + } + } +} diff --git a/src/AvParser.Infrastructure/Media/HardLink.cs b/src/AvParser.Infrastructure/Media/HardLink.cs new file mode 100644 index 0000000..f228291 --- /dev/null +++ b/src/AvParser.Infrastructure/Media/HardLink.cs @@ -0,0 +1,78 @@ +using System.Runtime.InteropServices; + +namespace AvParser.Infrastructure.Media; + +/// +/// Creates a second directory entry for an existing file. +/// +/// +/// The BCL has 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. +/// +internal static partial class HardLink +{ + /// Windows: the source and destination are on different volumes. + private const int ErrorNotSameDevice = 17; + + /// Unix: EXDEV, a cross-device link. + private const int CrossDeviceLink = 18; + + /// + /// Creates a hard link at pointing at . + /// + /// on success; otherwise with a reason. + 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); +} diff --git a/src/AvParser.Infrastructure/Media/MediaStore.cs b/src/AvParser.Infrastructure/Media/MediaStore.cs new file mode 100644 index 0000000..3309da2 --- /dev/null +++ b/src/AvParser.Infrastructure/Media/MediaStore.cs @@ -0,0 +1,189 @@ +using AvParser.Core.Collecting; +using Microsoft.Extensions.Logging; + +namespace AvParser.Infrastructure.Media; + +/// +/// The store as the rest of the app sees it: files, index and showcase behind one interface. +/// +/// +/// 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. +/// +public sealed class MediaStore( + SqliteMediaIndex index, + BlobStore blobs, + ShowcaseLinker showcase, + ILogger 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 _logger = logger ?? throw new ArgumentNullException(nameof(logger)); + + /// How showcase entries should point at their blobs. + /// + /// Settable rather than injected for the same reason IProxyPool.Configure is: it changes + /// while the app runs, and threading a snapshot through the container would freeze it at startup. + /// + public ShowcaseMode ShowcaseMode { get; private set; } = ShowcaseMode.HardLink; + + /// Applies the user's showcase preference. + public void Configure(ShowcaseMode mode) => ShowcaseMode = mode; + + /// + 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)); + } + + /// + public Task BeginRunAsync(string sourceId, CancellationToken cancellationToken = default) => + _index.BeginRunAsync(sourceId, cancellationToken); + + /// + public Task CompleteRunAsync(string runId, RunSummary summary, CancellationToken cancellationToken = default) => + _index.CompleteRunAsync(runId, summary, cancellationToken); + + /// + public Task> GetSeenAsync( + string sourceId, + IReadOnlyCollection urls, + CancellationToken cancellationToken = default + ) => _index.GetSeenAsync(sourceId, urls, cancellationToken); + + /// + 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); + + /// + public Task> LoadTombstonesAsync(CancellationToken cancellationToken = default) => + _index.LoadTombstonesAsync(cancellationToken); + + /// + public async Task 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; + } + + /// + public async Task 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 }; + } + + /// + public async Task 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); + } + + /// + public async Task 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; + } + + /// + public Task GetStatsAsync(CancellationToken cancellationToken = default) => + _index.GetStatsAsync(cancellationToken); + + /// Removes the files a committed removal orphaned. + /// Bytes reclaimed. + 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; + } +} diff --git a/src/AvParser.Infrastructure/Media/ShowcaseLinker.cs b/src/AvParser.Infrastructure/Media/ShowcaseLinker.cs new file mode 100644 index 0000000..dc4ccb2 --- /dev/null +++ b/src/AvParser.Infrastructure/Media/ShowcaseLinker.cs @@ -0,0 +1,292 @@ +using System.Globalization; +using AvParser.Core.Collecting; +using AvParser.Infrastructure.Storage; +using Microsoft.Extensions.Logging; + +namespace AvParser.Infrastructure.Media; + +/// A browsable entry pointing at a blob. +/// Path under the showcase root, with forward slashes. +/// What was actually achieved, which may be less than what was asked for. +public readonly record struct ShowcaseEntry(string RelativePath, ShowcaseMode Mode); + +/// Creates a hard link, reporting failure rather than throwing. +/// File that already exists. +/// Second name to create for it. +/// Why it failed, when it did. +/// Whether the link was created. +/// +/// 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. +/// +public delegate bool LinkStrategy(string existingPath, string linkPath, out string? error); + +/// +/// Builds a human-browsable tree over the content-addressed blobs. +/// +/// +/// +/// Content addressing is right for storage and useless for looking at: nobody wants to browse +/// blobs/a3/f1/a3f1…png. 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. +/// +/// +/// 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. +/// +/// +public sealed class ShowcaseLinker( + IAppPaths paths, + BlobStore blobs, + ILogger logger, + LinkStrategy? hardLink = null +) +{ + /// Kept well under MAX_PATH once the dated directories and the sequence are added. + 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 _logger = logger ?? throw new ArgumentNullException(nameof(logger)); + private readonly LinkStrategy _hardLink = hardLink ?? HardLink.TryCreate; + + /// Creates a showcase entry for content already in the blob store. + /// The entry, or when none was wanted or possible. + 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); + } + + /// Removes one showcase entry. The blob it pointed at is untouched. + 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); + } + } + + /// Removes an entire source's showcase subtree. + 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); + } + } + + /// + /// Turns whatever the origin suggested into something safe to write to disk. + /// + /// + /// 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. + /// + 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"; + } + + /// Numbers entries so browsing order matches collection order. + 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}"); + } + + /// + /// Links, degrading rather than failing. + /// + /// + /// 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. + /// + 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; + } + } +} diff --git a/src/AvParser.Infrastructure/Media/SqliteMediaIndex.cs b/src/AvParser.Infrastructure/Media/SqliteMediaIndex.cs new file mode 100644 index 0000000..6f9f98f --- /dev/null +++ b/src/AvParser.Infrastructure/Media/SqliteMediaIndex.cs @@ -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; + +/// A blob that lost its last reference and whose file can now be deleted. +/// Content hash. +/// Stored extension, needed to rebuild the path. +/// Size, for reporting how much was freed. +public readonly record struct OrphanedBlob(string Sha256, string Extension, long Length); + +/// Rows removed by a purge or a tombstone, with what the caller must unlink. +/// Provenance rows deleted. +/// Blobs whose last reference went with them. +/// Showcase entries to remove. +public sealed record RemovalPlan( + int ItemsRemoved, + IReadOnlyList Orphans, + IReadOnlyList ShowcasePaths +); + +/// One recorded item, as needed to rebuild the showcase. +/// Content hash. +/// Stored extension. +/// Name the origin suggested, if any. +/// When it was collected — the showcase is dated by this. +public readonly record struct IndexedItem( + string Sha256, + string Extension, + string? SuggestedName, + DateTimeOffset CollectedUtc +); + +/// +/// The SQLite half of the store: what is held, where it came from, and what has been tried. +/// +/// +/// +/// 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 BEGIN IMMEDIATE makes +/// contention fail at the start of a transaction rather than halfway through it. +/// +/// +/// 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. +/// +/// +public sealed class SqliteMediaIndex : IDisposable +{ + /// Bumped when the schema changes in a way that needs migrating. + private const int CurrentSchemaVersion = 1; + + /// SQLite has a parameter ceiling; batching well under it keeps the query planner happy. + private const int MaxParametersPerQuery = 400; + + private readonly string _connectionString; + private readonly ILogger _logger; + private readonly SemaphoreSlim _writeGate = new(1, 1); + + /// Creates the index over the configured database file. + public SqliteMediaIndex(IAppPaths paths, ILogger 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(); + } + + /// Creates the schema if it is not there yet. + 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(); + } + } + + /// Opens a run row and returns its id. + public async Task 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; + } + + /// Closes a run row with its totals. + 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 + ); + } + + /// Looks up recorded outcomes for a batch of addresses. + public async Task> GetSeenAsync( + string sourceId, + IReadOnlyCollection urls, + CancellationToken cancellationToken = default + ) + { + ArgumentException.ThrowIfNullOrWhiteSpace(sourceId); + ArgumentNullException.ThrowIfNull(urls); + + var found = new Dictionary(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; + } + + /// Records the fate of an address that produced no stored content. + 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 + ); + } + + /// Reads every hash marked as a dead-link placeholder. + public async Task> LoadTombstonesAsync(CancellationToken cancellationToken = default) + { + var hashes = new HashSet(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; + } + + /// Marks a hash as a placeholder and plans the removal of every copy held. + public async Task 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; + } + + /// + /// Records a stored blob and its provenance in one transaction. + /// + /// + /// when these bytes were new to the index, otherwise + /// . + /// + public async Task 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; + } + + /// Removes everything attributed to one source and plans the file deletions. + public async Task 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; + } + + /// Reads everything one source contributed, for rebuilding the showcase. + public async IAsyncEnumerable 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) + ); + } + } + + /// Updates one item's showcase entry after a rebuild. + 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 + ); + + /// Totals across the whole store. + public async Task 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)); + } + + /// Recomputes every reference count and reports how many were wrong. + /// + /// ref_count 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. + /// + public async Task 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; + } + + /// + public void Dispose() => _writeGate.Dispose(); + + private static async Task> 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(); + + 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> ReadShowcasePathsAsync( + SqliteConnection connection, + SqliteTransaction transaction, + string sql, + (string Name, string Value) parameter, + CancellationToken cancellationToken + ) + { + var paths = new List(); + + 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 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 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 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; + """; +} diff --git a/src/AvParser.Infrastructure/Storage/AppPaths.cs b/src/AvParser.Infrastructure/Storage/AppPaths.cs index 2fe3618..69dd49a 100644 --- a/src/AvParser.Infrastructure/Storage/AppPaths.cs +++ b/src/AvParser.Infrastructure/Storage/AppPaths.cs @@ -23,6 +23,30 @@ public interface IAppPaths /// Directory holding rolling log files. string LogDirectory { get; } + + /// Root of the collected media, holding the blobs, the showcase and the index. + /// + /// 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 and gets the rest for free. + /// + string MediaDirectory => Path.Combine(DataDirectory, "media"); + + /// Content-addressed store: one file per distinct byte sequence. + string BlobDirectory => Path.Combine(MediaDirectory, "blobs"); + + /// Browsable view over the blobs, organised by source and date. + string ShowcaseDirectory => Path.Combine(MediaDirectory, "showcase"); + + /// Where downloads are staged before they are promoted into . + /// + /// 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. + /// + string MediaTempDirectory => Path.Combine(MediaDirectory, "tmp"); + + /// Full path of the SQLite index describing everything collected. + string MediaIndexFile => Path.Combine(MediaDirectory, "index.db"); } /// @@ -47,7 +71,13 @@ public sealed class AppPaths : IAppPaths ) { } /// Creates paths under an explicit root. Used by tests. - public AppPaths(string dataDirectory) + /// Root for settings, logs and proxy state. + /// + /// Where collected media goes. Defaults to a subdirectory of . + /// A separate parameter because a collection outgrows the profile directory quickly, and + /// pointing it at a second drive is the first thing anyone does. + /// + 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"); } /// @@ -73,10 +111,28 @@ public sealed class AppPaths : IAppPaths /// public string LogDirectory { get; } + /// + public string MediaDirectory { get; } + + /// + public string BlobDirectory { get; } + + /// + public string ShowcaseDirectory { get; } + + /// + public string MediaTempDirectory { get; } + + /// + public string MediaIndexFile { get; } + /// Creates every directory this instance points at. public void EnsureCreated() { Directory.CreateDirectory(DataDirectory); Directory.CreateDirectory(LogDirectory); + Directory.CreateDirectory(BlobDirectory); + Directory.CreateDirectory(ShowcaseDirectory); + Directory.CreateDirectory(MediaTempDirectory); } } diff --git a/tests/AvParser.Infrastructure.Tests/Media/MediaStoreTests.cs b/tests/AvParser.Infrastructure.Tests/Media/MediaStoreTests.cs new file mode 100644 index 0000000..c239611 --- /dev/null +++ b/tests/AvParser.Infrastructure.Tests/Media/MediaStoreTests.cs @@ -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.Instance); + _blobs = new BlobStore(_paths, NullLogger.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.Instance, hardLink), + NullLogger.Instance + ); + + /// Writes bytes to a staged temp file and returns the blob describing them. + 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 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(); + } +} diff --git a/tests/AvParser.Infrastructure.Tests/Media/ShowcaseNameTests.cs b/tests/AvParser.Infrastructure.Tests/Media/ShowcaseNameTests.cs new file mode 100644 index 0000000..9913b7a --- /dev/null +++ b/tests/AvParser.Infrastructure.Tests/Media/ShowcaseNameTests.cs @@ -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"); + } +}