fix(external): path-safe unlink + per-link staging + orphan repair (#859)

External downloads import IN PLACE, so the post-attach dedup-skip unlink could
delete a file that IS an ImageRecord's backing file — orphaning the record and
404-ing on playback. Two sources of that:

- Two links on the same post (same film from mega + gdrive) emitted the same
  filename into one external/<post_id>/ dir; the second overwrote the first.
  Stage per-LINK now (external/<post_id>/<link_id>/) so each file keeps its path.
- The duplicate_hash/duplicate_phash branch unlinked `f` unconditionally. Make it
  path-safe: only unlink when `f` is NOT the existing record's canonical file.

Plus an operator-triggered orphan-repair maintenance task
(prune_missing_file_records_task) to clean up records already orphaned by the
bug: scans ImageRecords, deletes those whose file is gone (cascade), with an
NFS-stall guard that aborts without deleting if a large sample is mostly missing.
Wired through POST /api/admin/maintenance/prune-missing-files and a
MissingFileRepairCard in the Maintenance panel.

Tests: refetch-same-link keeps the canonical file; orphan repair deletes only
real orphans and aborts on the mostly-missing guard.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

tests/test_external_worker.py

@@ -5,13 +5,14 @@ import zipfile
 
 import pytest
 from PIL import Image
-from sqlalchemy import func, select
+from sqlalchemy import func, select, update
 
 import backend.app.tasks.external as ext
 from backend.app.models import (
     Artist,
     ExternalLink,
     ImageProvenance,
+    ImageRecord,
     Post,
     PostAttachment,
     Source,
@@ -102,6 +103,55 @@ def test_fetch_external_link_downloads_and_attaches(db_sync, tmp_path, monkeypat
     assert len(atts) == 1
 
 
+def test_refetch_same_link_keeps_canonical_file(db_sync, tmp_path, monkeypatch):
+    """#859: re-fetching the SAME link (identical bytes into the same per-link
+    staging dir) must NOT unlink the canonical file out from under its record.
+    The first fetch imports the image in place; the second dedups on sha256, and
+    the path-safe unlink recognises the file IS the record's backing file and
+    keeps it — so the post never 404s on playback."""
+    _, link = _seed(db_sync, host="pixeldrain")
+    link_id = link.id
+
+    monkeypatch.setattr(ext, "IMAGES_ROOT", tmp_path)
+    monkeypatch.setattr(ext, "_redis", lambda: _FakeRedis())
+
+    def fake_fetch(host, url, dest_dir, *, timeout, should_stop=lambda: False):
+        dest_dir.mkdir(parents=True, exist_ok=True)
+        f = dest_dir / "clip.jpg"  # art → ImageRecord (imported in place)
+        f.write_bytes(_structured_jpeg_bytes())
+        return FetchResult(files=[f], bytes=f.stat().st_size)
+
+    monkeypatch.setattr(ext, "fetch_external", fake_fetch)
+
+    import backend.app.tasks.ml as ml_mod
+    import backend.app.tasks.thumbnail as thumb_mod
+    monkeypatch.setattr(ml_mod.tag_and_embed, "delay", lambda i: None)
+    monkeypatch.setattr(thumb_mod.generate_thumbnail, "delay", lambda i: None)
+
+    out = ext.fetch_external_link(link_id)
+    assert out["images"] == 1
+
+    db_sync.expire_all()
+    rec = db_sync.execute(select(ImageRecord)).scalars().one()
+    canonical = tmp_path / rec.path  # absolute path collapses the join
+    assert canonical.exists()
+
+    # Re-arm the SAME link and fetch again — same staging dir, identical bytes.
+    db_sync.execute(
+        update(ExternalLink).where(ExternalLink.id == link_id)
+        .values(status="pending", attempts=0, completed_at=None)
+    )
+    db_sync.commit()
+
+    out2 = ext.fetch_external_link(link_id)
+    assert out2["images"] == 0  # deduped, no new record
+
+    db_sync.expire_all()
+    # Exactly one record still, and its file survived the dedup unlink (the bug).
+    assert db_sync.execute(select(func.count(ImageRecord.id))).scalar() == 1
+    assert canonical.exists()
+
+
 def test_fetch_external_link_records_failure(db_sync, tmp_path, monkeypatch):
     _, link = _seed(db_sync)
     link_id = link.id

tests/test_tasks_admin.py

@@ -141,6 +141,80 @@ async def test_bulk_delete_images_task_removes_listed_images(
     assert surviving == 0
 
 
+# --- prune_missing_file_records_task (#859 orphan repair) -----------
+
+
+def _img(db_sync, artist, *, path, sha):
+    img = ImageRecord(
+        artist_id=artist.id, path=str(path),
+        sha256=sha, size_bytes=10, mime="image/jpeg",
+        origin="downloaded",
+    )
+    db_sync.add(img)
+    db_sync.flush()
+    return img.id
+
+
+@pytest.mark.asyncio
+async def test_prune_missing_file_records_deletes_only_orphans(db_sync, tmp_path):
+    """Records whose backing file is gone get deleted; records whose file is
+    present survive. Below the abort sample size, fraction is irrelevant."""
+    from backend.app.tasks.admin import prune_missing_file_records_task
+
+    a = Artist(name="Orph", slug="orph")
+    db_sync.add(a)
+    db_sync.flush()
+
+    present = []
+    for i in range(3):
+        f = tmp_path / f"present{i}.jpg"
+        f.write_bytes(b"real")
+        present.append(_img(db_sync, a, path=f, sha=f"{i:064x}"))
+    missing = [
+        _img(db_sync, a, path=tmp_path / f"gone{i}.jpg", sha=f"f{i:063x}")
+        for i in range(2)
+    ]
+    db_sync.commit()
+
+    result = prune_missing_file_records_task.delay().get()
+    assert result["checked"] == 5
+    assert result["missing"] == 2
+    assert result["deleted"] == 2
+
+    db_sync.expire_all()
+    surviving = db_sync.execute(
+        select(ImageRecord.id).where(ImageRecord.id.in_(present + missing))
+    ).scalars().all()
+    assert set(surviving) == set(present)
+
+
+@pytest.mark.asyncio
+async def test_prune_missing_file_records_aborts_when_mostly_missing(db_sync, tmp_path):
+    """NFS-stall guard: a large sample that's mostly missing means the
+    filesystem is unhealthy, not that the library evaporated — abort, delete
+    nothing."""
+    from backend.app.tasks.admin import prune_missing_file_records_task
+
+    a = Artist(name="Stall", slug="stall")
+    db_sync.add(a)
+    db_sync.flush()
+    ids = [
+        _img(db_sync, a, path=tmp_path / f"phantom{i}.jpg", sha=f"{i:064x}")
+        for i in range(55)  # >= _ORPHAN_MIN_SAMPLE, all missing -> frac 1.0
+    ]
+    db_sync.commit()
+
+    result = prune_missing_file_records_task.delay().get()
+    assert result["deleted"] == 0
+    assert "aborted" in result
+
+    db_sync.expire_all()
+    surviving = db_sync.execute(
+        select(func.count(ImageRecord.id)).where(ImageRecord.id.in_(ids))
+    ).scalar_one()
+    assert surviving == 55
+
+
 @pytest.mark.asyncio
 async def test_bulk_delete_images_task_idempotent_on_empty_list(db_sync):
     from backend.app.tasks.admin import bulk_delete_images_task