feat(thumb-backfill): backfill_thumbnails planner task — keyset-paginates ImageRecord, NULLs bad thumb paths, enqueues generate_thumbnail for NULL/missing/corrupt

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

backend/app/tasks/thumbnail.py

@@ -74,3 +74,58 @@ def generate_thumbnail(self, image_id: int) -> dict:
         session.add(record)
         session.commit()
         return {"status": "ok", "image_id": image_id, "path": str(result.path)}
+
+
+@celery.task(
+    name="backend.app.tasks.thumbnail.backfill_thumbnails",
+    bind=True,
+)
+def backfill_thumbnails(self) -> dict:
+    """Scan ImageRecord and enqueue generate_thumbnail for rows whose
+    thumbnail is missing, gone from disk, or has wrong magic bytes.
+
+    Keyset paginates by id ASC, page size 500. NULLs out thumbnail_path for
+    rows that point at a missing or corrupt file before enqueueing — keeps
+    the DB self-consistent on partial runs and makes re-runs safe.
+
+    Returns {"enqueued": N, "ok": M, "regenerated": K} where:
+      - enqueued = total generate_thumbnail.delay() calls
+      - ok       = rows whose existing thumbnail file is valid (skipped)
+      - regenerated = subset of enqueued that had a non-NULL thumbnail_path
+                      cleared (i.e. missing + corrupt)
+    """
+    from sqlalchemy import select, update
+
+    SessionLocal = _sync_session_factory()
+    enqueued = 0
+    ok = 0
+    regenerated = 0
+    last_id = 0
+    with SessionLocal() as session:
+        while True:
+            rows = session.execute(
+                select(ImageRecord.id, ImageRecord.thumbnail_path)
+                .where(ImageRecord.id > last_id)
+                .order_by(ImageRecord.id.asc())
+                .limit(500)
+            ).all()
+            if not rows:
+                break
+            for image_id, thumb_path in rows:
+                if thumb_path is None:
+                    generate_thumbnail.delay(image_id)
+                    enqueued += 1
+                elif _thumb_is_valid(Path(thumb_path)):
+                    ok += 1
+                else:
+                    session.execute(
+                        update(ImageRecord)
+                        .where(ImageRecord.id == image_id)
+                        .values(thumbnail_path=None)
+                    )
+                    generate_thumbnail.delay(image_id)
+                    enqueued += 1
+                    regenerated += 1
+            session.commit()
+            last_id = rows[-1][0]
+    return {"enqueued": enqueued, "ok": ok, "regenerated": regenerated}

tests/test_backfill_thumbnails.py

@@ -1,7 +1,10 @@
 """Thumbnail backfill: _thumb_is_valid helper + backfill_thumbnails planner."""
 
+from pathlib import Path
+
 import pytest
 
+from backend.app.models import ImageRecord
 from backend.app.tasks.thumbnail import _thumb_is_valid
 
 pytestmark = pytest.mark.integration
@@ -33,3 +36,207 @@ def test_thumb_is_valid_wrong_magic(tmp_path):
 
 def test_thumb_is_valid_missing_file(tmp_path):
     assert _thumb_is_valid(tmp_path / "nope") is False
+
+
+# --- backfill_thumbnails planner tests ------------------------------------
+
+
+class _Ctx:
+    def __init__(self, s):
+        self.s = s
+
+    def __enter__(self):
+        return self.s
+
+    def __exit__(self, *a):
+        return False
+
+
+def _sf(db_sync):
+    """sessionmaker-like returning the test's bound session, matching the
+    pattern in tests/test_backfill_phash.py."""
+    class _SM:
+        def __call__(self):
+            return _Ctx(db_sync)
+
+    return _SM()
+
+
+def _sha(prefix: str) -> str:
+    return f"{prefix}".ljust(64, "0")[:64]
+
+
+def _rec(db_sync, path, *, sha, thumb_path=None, mime="image/jpeg"):
+    rec = ImageRecord(
+        path=str(path), sha256=sha, size_bytes=1, mime=mime,
+        width=64, height=64, origin="imported_filesystem",
+        integrity_status="unknown",
+        thumbnail_path=str(thumb_path) if thumb_path is not None else None,
+    )
+    db_sync.add(rec)
+    db_sync.flush()
+    return rec
+
+
+def _write_jpeg(p: Path) -> Path:
+    p.parent.mkdir(parents=True, exist_ok=True)
+    p.write_bytes(b"\xff\xd8\xff\xe0" + b"\x00" * 100)
+    return p
+
+
+def _write_png(p: Path) -> Path:
+    p.parent.mkdir(parents=True, exist_ok=True)
+    p.write_bytes(b"\x89PNG\r\n\x1a\n" + b"\x00" * 100)
+    return p
+
+
+def _write_garbage(p: Path) -> Path:
+    p.parent.mkdir(parents=True, exist_ok=True)
+    p.write_bytes(b"\x00" * 12)
+    return p
+
+
+def test_backfill_null_path_enqueued(db_sync, tmp_path, monkeypatch):
+    from backend.app.tasks import thumbnail as m
+
+    src = tmp_path / "a.bin"
+    src.write_bytes(b"x")
+    rec = _rec(db_sync, src, sha=_sha("a"), thumb_path=None)
+    db_sync.commit()
+
+    monkeypatch.setattr(m, "_sync_session_factory", lambda: _sf(db_sync))
+    delayed: list[int] = []
+    monkeypatch.setattr(
+        m.generate_thumbnail, "delay", lambda image_id: delayed.append(image_id)
+    )
+
+    result = m.backfill_thumbnails()
+    assert result == {"enqueued": 1, "ok": 0, "regenerated": 0}
+    assert delayed == [rec.id]
+
+
+def test_backfill_missing_file_clears_and_enqueues(db_sync, tmp_path, monkeypatch):
+    from backend.app.tasks import thumbnail as m
+
+    src = tmp_path / "b.bin"
+    src.write_bytes(b"x")
+    rec = _rec(
+        db_sync, src, sha=_sha("b"),
+        thumb_path=tmp_path / "thumbs" / "missing.jpg",
+    )
+    db_sync.commit()
+
+    monkeypatch.setattr(m, "_sync_session_factory", lambda: _sf(db_sync))
+    delayed: list[int] = []
+    monkeypatch.setattr(
+        m.generate_thumbnail, "delay", lambda image_id: delayed.append(image_id)
+    )
+
+    result = m.backfill_thumbnails()
+    db_sync.expire_all()
+    assert result == {"enqueued": 1, "ok": 0, "regenerated": 1}
+    assert delayed == [rec.id]
+    assert db_sync.get(ImageRecord, rec.id).thumbnail_path is None
+
+
+def test_backfill_valid_jpeg_skipped(db_sync, tmp_path, monkeypatch):
+    from backend.app.tasks import thumbnail as m
+
+    src = tmp_path / "c.bin"
+    src.write_bytes(b"x")
+    thumb = _write_jpeg(tmp_path / "thumbs" / "c.jpg")
+    rec = _rec(db_sync, src, sha=_sha("c"), thumb_path=thumb)
+    db_sync.commit()
+
+    monkeypatch.setattr(m, "_sync_session_factory", lambda: _sf(db_sync))
+    delayed: list[int] = []
+    monkeypatch.setattr(
+        m.generate_thumbnail, "delay", lambda image_id: delayed.append(image_id)
+    )
+
+    result = m.backfill_thumbnails()
+    db_sync.expire_all()
+    assert result == {"enqueued": 0, "ok": 1, "regenerated": 0}
+    assert delayed == []
+    assert db_sync.get(ImageRecord, rec.id).thumbnail_path == str(thumb)
+
+
+def test_backfill_valid_png_skipped(db_sync, tmp_path, monkeypatch):
+    from backend.app.tasks import thumbnail as m
+
+    src = tmp_path / "d.bin"
+    src.write_bytes(b"x")
+    thumb = _write_png(tmp_path / "thumbs" / "d.png")
+    _rec(db_sync, src, sha=_sha("d"), thumb_path=thumb)
+    db_sync.commit()
+
+    monkeypatch.setattr(m, "_sync_session_factory", lambda: _sf(db_sync))
+    delayed: list[int] = []
+    monkeypatch.setattr(
+        m.generate_thumbnail, "delay", lambda image_id: delayed.append(image_id)
+    )
+
+    result = m.backfill_thumbnails()
+    assert result == {"enqueued": 0, "ok": 1, "regenerated": 0}
+    assert delayed == []
+
+
+def test_backfill_corrupt_magic_clears_and_enqueues(db_sync, tmp_path, monkeypatch):
+    from backend.app.tasks import thumbnail as m
+
+    src = tmp_path / "e.bin"
+    src.write_bytes(b"x")
+    thumb = _write_garbage(tmp_path / "thumbs" / "e.jpg")
+    rec = _rec(db_sync, src, sha=_sha("e"), thumb_path=thumb)
+    db_sync.commit()
+
+    monkeypatch.setattr(m, "_sync_session_factory", lambda: _sf(db_sync))
+    delayed: list[int] = []
+    monkeypatch.setattr(
+        m.generate_thumbnail, "delay", lambda image_id: delayed.append(image_id)
+    )
+
+    result = m.backfill_thumbnails()
+    db_sync.expire_all()
+    assert result == {"enqueued": 1, "ok": 0, "regenerated": 1}
+    assert delayed == [rec.id]
+    assert db_sync.get(ImageRecord, rec.id).thumbnail_path is None
+
+
+def test_backfill_mixed_aggregate(db_sync, tmp_path, monkeypatch):
+    from backend.app.tasks import thumbnail as m
+
+    src_null = tmp_path / "src_null.bin"
+    src_null.write_bytes(b"x")
+    src_jpeg = tmp_path / "src_jpeg.bin"
+    src_jpeg.write_bytes(b"x")
+    src_png = tmp_path / "src_png.bin"
+    src_png.write_bytes(b"x")
+    src_missing = tmp_path / "src_missing.bin"
+    src_missing.write_bytes(b"x")
+    src_bad = tmp_path / "src_bad.bin"
+    src_bad.write_bytes(b"x")
+
+    jpeg = _write_jpeg(tmp_path / "thumbs" / "ok.jpg")
+    png = _write_png(tmp_path / "thumbs" / "ok.png")
+    bad = _write_garbage(tmp_path / "thumbs" / "bad.jpg")
+
+    r_null = _rec(db_sync, src_null, sha=_sha("aa"), thumb_path=None)
+    _rec(db_sync, src_jpeg, sha=_sha("bb"), thumb_path=jpeg)
+    _rec(db_sync, src_png, sha=_sha("cc"), thumb_path=png)
+    r_missing = _rec(
+        db_sync, src_missing, sha=_sha("dd"),
+        thumb_path=tmp_path / "thumbs" / "missing.jpg",
+    )
+    r_bad = _rec(db_sync, src_bad, sha=_sha("ee"), thumb_path=bad)
+    db_sync.commit()
+
+    monkeypatch.setattr(m, "_sync_session_factory", lambda: _sf(db_sync))
+    delayed: list[int] = []
+    monkeypatch.setattr(
+        m.generate_thumbnail, "delay", lambda image_id: delayed.append(image_id)
+    )
+
+    result = m.backfill_thumbnails()
+    assert result == {"enqueued": 3, "ok": 2, "regenerated": 2}
+    assert sorted(delayed) == sorted([r_null.id, r_missing.id, r_bad.id])

tests/test_tasks_register.py

@@ -23,3 +23,7 @@ def test_import_media_file_registered():
 
 def test_generate_thumbnail_registered():
     assert "backend.app.tasks.thumbnail.generate_thumbnail" in celery.tasks
+
+
+def test_backfill_thumbnails_registered():
+    assert "backend.app.tasks.thumbnail.backfill_thumbnails" in celery.tasks