feat(worker): Crawl each storage config as a work item, with scope routing (#281)

CLAUDE.md

@@ -132,7 +132,7 @@ do not always match entity names (`AssetInteraction` → `Interactions`, `UserSt
 - `User` — owns storage configs and interactions
 - `RefreshToken` — issued and revoked by the API, expired rows deleted by the Worker's `TokenCleanup` work item; `TokenHash` unique
 - `Invite` — admin-issued registration tokens; `TokenHash` unique, and the only entity carrying an optimistic-concurrency token (`xmin`)
-- `WorkItem` — one durable unit of Worker work (`Kind`, jsonb `Payload`, optional `DedupKey`, status, lease, attempt, progress snapshot, result or error); see "Work queue"
+- `WorkItem` — one durable unit of Worker work (`Kind`, jsonb `Payload`, optional `DedupKey`, optional `StorageConfigId` scope that cascades on config delete, status, lease, attempt, progress snapshot, result or error); see "Work queue"
 
 ### Key EF Core / Database Decisions
 
@@ -159,8 +159,8 @@ The API reads files only through two narrow stores in `Anichron.Infrastructure/S
 
 ### Worker Media Processing
 
-- Crawls NAS on a configurable interval (`Worker:CrawlIntervalHours`, default 4 h; immediate crawl on startup)
-- The crawl walks folder by folder and never enters `CrawlExclusions`: any dot-folder or dot-file (macOS/Linux/QNAP/ZFS/sync-tool metadata), NAS and OS system folders (`@eaDir`, `#recycle`, `@Recycle`, `$RECYCLE.BIN`, `lost+found`, Nextcloud `files_trashbin`/`files_versions`, …) and app library packages (`*.photoslibrary`, `*.lrdata`, …, which hold thumbnails and renders). `Worker:ExcludedDirectories` adds comma-separated folder names (not paths; rejected at startup) on top. An unreadable subfolder is logged and skipped; an unreadable or missing root fails that storage config's crawl, which is logged as an error, and the Worker moves on to the next config (#282)
+- Crawls each storage config as its own `Crawl` work item (#281), due every `Worker:CrawlIntervalHours` (default 4 h) after the last one finished; the first crawl after a start comes from due-ness, so a restart that outlasted the interval crawls straight away. Progress is items found, processed and failed (a Live Photo pair is one item), with no total until the walk ends
+- The crawl walks folder by folder and never enters `CrawlExclusions`: any dot-folder or dot-file (macOS/Linux/QNAP/ZFS/sync-tool metadata), NAS and OS system folders (`@eaDir`, `#recycle`, `@Recycle`, `$RECYCLE.BIN`, `lost+found`, Nextcloud `files_trashbin`/`files_versions`, …) and app library packages (`*.photoslibrary`, `*.lrdata`, …, which hold thumbnails and renders). `Worker:ExcludedDirectories` adds comma-separated folder names (not paths; rejected at startup) on top. An unreadable subfolder is logged and skipped; an unreadable or missing root fails that storage config's `Crawl` item with the error, and the other configs' items still run (#282)
 - Computes XXHash64 `ContentHash` → extracts EXIF via MetadataExtractor → writes `MediaAsset` + `Metadata`
 - Generates proxy types: `Thumbnail`, `FullPreview`, `WebVideo` (the `ProxyType` enum, persisted as text), and a BlurHash string for images, stored in `Metadata.Blurhash`
 - Image proxies are auto-oriented once after decoding: upright pixels, orientation tag 1. `Metadata.Orientation` applies only when serving originals — never rotate a proxy by it
@@ -169,7 +169,7 @@ The API reads files only through two narrow stores in `Anichron.Infrastructure/S
 - FFmpeg transcodes video with runtime GPU detection: QuickSync (`h264_qsv`) → NVENC (`h264_nvenc`) → AMF (`h264_amf`) → software (`libx264`)
 - Burst detection (**planned, not yet implemented**): will group rapid-fire sequences and assign a `PrimaryAssetId` cover. The `Burst` entity and schema exist; no detection code is wired into the Worker yet.
 - Reconciliation (**planned, Epic 4**): a periodic scan soft-deletes assets whose file is gone (interactions survive) and restores them when it reappears (#33, #34, #37). Moves are already handled at ingest (#35): when the hash matches an active row, `KnownFileReconciler` refreshes `LastSeenOnNas` if the path is unchanged (`Unchanged`), moves the row to the new path if its stored file is gone (`Moved`), and leaves it alone if the stored file still exists (`DuplicateCopy`, reported by #36). A Live Photo's MOV row gets the same treatment by its own hash. A move onto a path another row already holds, soft-deleted rows included (`PathTaken`), is logged as a warning and skipped; if a concurrent consumer takes the path after classification, the save fails and the item is retried on the next crawl. A file **edited in place** (hash unknown, but an active row holds the path: `ContentChanged`) is processed like new content and `ContentChangedWriter` updates that row: new hash, capture date, media type, `Metadata` and proxy rows (updated by type), while `Id`, interactions, `BurstId` and `PairedAssetId` are kept. A soft-deleted row at the path still blocks it (restore, #34). An edited Live Photo **MOV** keeps its stored row unchanged for now
-- Duplicate content is **reported, never merged** (#36). `SharedContentReporter` runs after each crawl cycle over the configs just crawled: it logs one `Information` summary per user (copies within one config, counted by `DuplicateCopyTally` during the crawl, and pairs of that user's active rows sharing a `ContentHash` across configs), with each copy and pair at `Debug`. Duplicates across different users are expected and never reported. Two new identical copies racing in one crawl are first reported on the next crawl
+- Duplicate content is **reported, never merged** (#36). `SharedContentReporter` runs at the end of each config's `Crawl` item: it logs one `Information` summary for the crawled config (copies within it, counted by `DuplicateCopyTally` during that crawl in this Worker, and pairs of its active rows sharing a `ContentHash` with the same user's other active configs), with each pair at `Debug`, and the counts go into the crawl's result. A pair across two configs is reported by each of their crawls. A failed report is logged and does not fail the crawl. Duplicates across different users are expected and never reported. Two new identical copies racing in one crawl are first reported on the next crawl
 - Failure ownership: `ProxyStagingWriter` is the single place a proxy reaches its final name (architecture rule `proxies-reach-final-name-only-in-staging-writer`) — it writes under a per-attempt `.<attempt id>.tmp` suffix (so concurrent attempts on the same content never share a temporary file), renames into place once complete, and registers the proxy on the `IngestionContext` at that moment; the proxy middlewares only supply the bytes. If a step throws, `IngestionPipelineRunner` deletes exactly those registered proxies plus its own temporary files, and removes the directory only if it is left empty — never recursively, since a concurrent attempt on the same content shares it until one of them persists. An attempt that loses a uniqueness race (`LostIngestionRace`: the active content-hash index or the one-proxy-per-asset-and-type index) first releases its paired MOV's registrations, because the winner owns those content-derived files. A clean shutdown is treated differently: the in-flight temporary file is dropped, but proxies already completed in that run are kept, and the consumer loop re-throws the cancellation instead of recording an item failure and draining the queue. Each attempt has its own `IngestionContext.AttemptId`, which names temporary files and scopes compensation; `IngestionContext.AssetId` is the row that owns the proxies, set by classification (the stored row for `ContentChanged`, the attempt id for new content). A hard kill, the storage-config cascade delete, and the old content directory of a file edited in place are out of reach in-process and belong to the orphan sweeper (#174); the old directory is deliberately not deleted after commit, because a copy of the old content elsewhere can be ingested into it in the same crawl.
 
 ### Flashback Interaction Rules
@@ -194,12 +194,13 @@ Each user has a private set of `UserStorageConfig` entries pointing to NAS roots
 
 ### Work queue
 
-Durable work for the Worker lives in `WorkItems` (#276, ADR 0008). Today its kinds are periodic maintenance (`TokenCleanup`, `WorkItemRetention`); the crawl (#281) and admin requests (#174) follow.
+Durable work for the Worker lives in `WorkItems` (#276, ADR 0008). Its kinds are the crawl (`Crawl`, one per storage config) and periodic maintenance (`TokenCleanup`, `WorkItemRetention`); admin requests (#174) follow.
 
 - **Producing:** `IWorkQueue.EnqueueAsync` (`EfWorkQueue`, Core) is the only insert. A payload with a `DedupKey` coalesces: while an item with that key is `Queued`, enqueueing returns it (a partial unique index absorbs racing inserts).
-- **Scheduling:** every Worker runs `WorkScheduler`. Each poll (`Worker:WorkPollIntervalSeconds`, 5 s) it recovers expired leases, then asks each `IWorkSchedule` whether its kind is due: nothing queued or running with its key and nothing finished within its interval. No timer state is kept, so overdue work runs right after downtime.
+- **Scope:** a scoped item carries the `StorageConfigId` it belongs to (`IWorkPayload.StorageConfigId`). A per-user Worker (`WORKER__USER`) claims unscoped items and items of its user's configs; an all-user Worker claims everything. The served user is `WorkerState.ResolvedUserId`, read at each claim and tick, never cached at construction.
+- **Scheduling:** every Worker runs `WorkScheduler`. Each poll (`Worker:WorkPollIntervalSeconds`, 5 s) it recovers expired leases, then asks each `IWorkScheduleSource` for its schedules (a fixed `WorkSchedule<TPayload>` is its own source; `CrawlSchedules` returns one per active config the Worker serves) and whether each is due: nothing queued or running with its key and nothing finished within its interval. No timer state is kept, so overdue work runs right after downtime.
 - **Executing:** `WorkExecutor` claims the oldest queued item (`FOR UPDATE SKIP LOCKED`), one at a time, under a lease (`WorkLeaseSeconds`, 60 s) that a heartbeat (`WorkHeartbeatSeconds`, 2 s) extends while writing the progress snapshot. Every write about a running item is fenced by `(Id, Attempt)`. A fenced heartbeat cancels the handler and writes nothing more; a shutdown releases the item without a stall; a handler exception fails it with no retry. An expired lease requeues the item and counts a stall; the `WorkMaxStalls`th (3) fails it as stalled. An item handed back (release or recovery) while another with its key is already queued ends `Cancelled` as superseded, since the queued one does the same work. A Worker claims only the kinds it knows, so a newer kind waits for a newer Worker. `WorkItemRetention` deletes items finished more than `WorkItemRetentionDays` (30) ago.
-- **Adding a kind:** a `WorkKind` value, a payload record in Core implementing `IWorkPayload` (static `Kind`, optional `DedupKey`), an `IWorkHandler<TPayload>` in the Worker registered with `AddWorkHandler`, and for a periodic kind a `WorkSchedule<TPayload>` subclass. `WorkHandlerRegistry` fails startup unless every kind has exactly one payload and one handler; nothing switches on the kind.
+- **Adding a kind:** a `WorkKind` value, a payload record in Core implementing `IWorkPayload` (static `Kind`, optional `DedupKey`), an `IWorkHandler<TPayload>` in the Worker registered with `AddWorkHandler`, and for a periodic kind a `WorkSchedule<TPayload>` subclass registered as an `IWorkScheduleSource` (or a source that builds several). `WorkHandlerRegistry` fails startup unless every kind has exactly one payload and one handler; nothing switches on the kind.
 - **Handler contract:** delivery is at least once. Each run starts from scratch and converges, side effects are safe to repeat, and the returned summary becomes `Result`.
 
 ### Worker Processing Model

docs/schema.sql

@@ -64,6 +64,7 @@ CREATE TABLE "WorkItems" (
     "Kind" text NOT NULL,
     "Payload" jsonb NOT NULL,
     "DedupKey" text,
+    "StorageConfigId" uuid,
     "RequestedBy" uuid,
     "RequestedAt" timestamp with time zone NOT NULL,
     "Status" text NOT NULL,
@@ -80,12 +81,14 @@ CREATE TABLE "WorkItems" (
     "Result" jsonb,
     "Error" text,
     CONSTRAINT "PK_WorkItems" PRIMARY KEY ("Id"),
+    CONSTRAINT "FK_WorkItems_StorageConfigs_StorageConfigId" FOREIGN KEY ("StorageConfigId") REFERENCES "StorageConfigs" ("Id") ON DELETE CASCADE,
     CONSTRAINT "FK_WorkItems_Users_RequestedBy" FOREIGN KEY ("RequestedBy") REFERENCES "Users" ("Id") ON DELETE SET NULL
 );
 COMMENT ON TABLE "WorkItems" IS 'Durable work for the Worker: periodic maintenance today, admin requests later. Only IWorkQueue inserts rows; every later state change belongs to the Worker''s executor. Delivery is at least once, so handlers start from scratch and converge.';
 COMMENT ON COLUMN "WorkItems"."Kind" IS 'WorkKind, persisted as text. Picks the payload type and the handler through the registry, never through a switch.';
 COMMENT ON COLUMN "WorkItems"."Payload" IS 'The kind''s IWorkPayload, serialised with System.Text.Json and NodaTime settings.';
 COMMENT ON COLUMN "WorkItems"."DedupKey" IS 'Opt-in coalescing. While an item with this key is Queued, enqueueing the same key returns that item instead of adding a second; the partial unique index enforces it.';
+COMMENT ON COLUMN "WorkItems"."StorageConfigId" IS 'The storage config the work belongs to, or null for unscoped work. A per-user Worker claims only its user''s configs and unscoped work; deleting the config deletes its items.';
 COMMENT ON COLUMN "WorkItems"."RequestedBy" IS 'The admin who asked for the work; null when a Worker''s scheduler enqueued it.';
 COMMENT ON COLUMN "WorkItems"."Status" IS 'Queued, Running, Succeeded, Failed or Cancelled, persisted as text.';
 COMMENT ON COLUMN "WorkItems"."Attempt" IS 'Incremented by every claim. Every write a Worker makes while running an item is fenced by (Id, Attempt), so an attempt that lost its lease changes nothing.';
@@ -256,6 +259,9 @@ CREATE INDEX ix_workitems_lease ON "WorkItems" ("LeaseUntil") WHERE "Status" = '
 CREATE INDEX "IX_WorkItems_RequestedBy" ON "WorkItems" ("RequestedBy");
 
 
+CREATE INDEX "IX_WorkItems_StorageConfigId" ON "WorkItems" ("StorageConfigId");
+
+
 CREATE INDEX "IX_WorkItems_UpdatedAt" ON "WorkItems" ("UpdatedAt");
 
 

src/Anichron.Core.Tests.Unit/Work/PayloadsTests.cs

@@ -9,6 +9,23 @@ private static (WorkKind Kind, string? DedupKey) Describe<TPayload>(TPayload pay
         where TPayload : IWorkPayload
         => (TPayload.Kind, payload.DedupKey);
 
+    private static Guid? ScopeOf<TPayload>(TPayload payload)
+        where TPayload : IWorkPayload
+        => payload.StorageConfigId;
+
+    [Fact]
+    public void Crawl_IsOnePeriodicItemPerConfigScopedToIt()
+    {
+        var configId = Guid.Parse("77777777-7777-7777-7777-777777777777");
+
+        Describe(new CrawlPayload(configId)).Should().Be((WorkKind.Crawl, "crawl:77777777-7777-7777-7777-777777777777"));
+        ScopeOf(new CrawlPayload(configId)).Should().Be(configId);
+    }
+
+    [Fact]
+    public void MaintenancePayloads_AreUnscoped()
+        => (ScopeOf(new TokenCleanupPayload()), ScopeOf(new WorkItemRetentionPayload())).Should().Be((null, null));
+
     [Fact]
     public void TokenCleanup_IsOnePeriodicItemAtATime()
         => Describe(new TokenCleanupPayload()).Should().Be((WorkKind.TokenCleanup, "token-cleanup"));

src/Anichron.Core.Tests.Unit/Work/WorkQueueTests.cs

@@ -22,6 +22,13 @@ private sealed record Keyed(string Key) : IWorkPayload
         string IWorkPayload.DedupKey => Key;
     }
 
+    private sealed record Scoped(Guid ConfigId) : IWorkPayload
+    {
+        public static WorkKind Kind => WorkKind.TokenCleanup;
+
+        Guid? IWorkPayload.StorageConfigId => ConfigId;
+    }
+
     private static readonly Instant Now = Instant.FromUtc(2026, 10, 18, 9, 0);
     private static readonly Guid AdminId = Guid.Parse("11111111-1111-1111-1111-111111111111");
 
@@ -90,7 +97,18 @@ public async Task EnqueueAsync_Unkeyed_TakesTheKindFromThePayloadTypeAndKeepsNoK
         await EnqueueAsync(new Unkeyed(1));
 
         var item = (await ItemsAsync()).Should().ContainSingle().Which;
-        (item.Kind, item.DedupKey, item.RequestedBy).Should().Be((WorkKind.WorkItemRetention, null, null));
+        (item.Kind, item.DedupKey, item.RequestedBy, item.StorageConfigId)
+            .Should().Be((WorkKind.WorkItemRetention, null, null, null));
+    }
+
+    [Fact]
+    public async Task EnqueueAsync_ScopedPayload_StoresItsStorageConfigAsync()
+    {
+        var configId = Guid.Parse("55555555-5555-5555-5555-555555555555");
+
+        await EnqueueAsync(new Scoped(configId));
+
+        (await ItemsAsync()).Should().ContainSingle().Which.StorageConfigId.Should().Be(configId);
     }
 
     [Fact]

src/Anichron.Core/Data/AnichronDbContext.cs

@@ -327,6 +327,9 @@ protected override void OnModelCreating(ModelBuilder modelBuilder)
             entity.Property(w => w.DedupKey).HasComment(
                 "Opt-in coalescing. While an item with this key is Queued, enqueueing the same key "
                 + "returns that item instead of adding a second; the partial unique index enforces it.");
+            entity.Property(w => w.StorageConfigId).HasComment(
+                "The storage config the work belongs to, or null for unscoped work. A per-user Worker "
+                + "claims only its user's configs and unscoped work; deleting the config deletes its items.");
             entity.Property(w => w.RequestedBy).HasComment(
                 "The admin who asked for the work; null when a Worker's scheduler enqueued it.");
             entity.Property(w => w.Status).HasConversion<string>().HasComment(
@@ -363,6 +366,11 @@ protected override void OnModelCreating(ModelBuilder modelBuilder)
                   .HasDatabaseName("ix_workitems_lease");
             entity.HasIndex(w => w.UpdatedAt);
 
+            entity.HasOne<UserStorageConfig>()
+                  .WithMany()
+                  .HasForeignKey(w => w.StorageConfigId)
+                  .OnDelete(DeleteBehavior.Cascade);
+
             entity.HasOne<User>()
                   .WithMany()
                   .HasForeignKey(w => w.RequestedBy)

src/Anichron.Core/Domain/WorkItem.cs

@@ -6,6 +6,7 @@ public class WorkItem
     public WorkKind Kind { get; set; }
     public required string Payload { get; set; }
     public string? DedupKey { get; set; }
+    public Guid? StorageConfigId { get; set; }
     public Guid? RequestedBy { get; set; }
     public Instant RequestedAt { get; set; }
 

src/Anichron.Core/Domain/WorkKind.cs

@@ -4,4 +4,5 @@ public enum WorkKind
 {
     TokenCleanup = 1,
     WorkItemRetention = 2,
+    Crawl = 3,
 }