fix(scan): let update metadata scans refresh artwork, and record uploads explicitly

backend/alembic/versions/0108_roms_locked_fields.py

@@ -0,0 +1,80 @@
+"""Record which rom fields a user supplied by hand
+
+Uploading artwork stores the file and clears `url_cover`, so a stored cover
+path with no url was the only thing separating a hand-supplied cover from a
+scraped one. That signal cannot survive: `get_cover` returns no path when the
+file is missing, and the scan writes that straight back to `path_cover_s`, so a
+single scan run while the resources volume is unavailable erases it. The next
+scan then reads the row as having no cover at all, adopts the provider url, and
+replaces the user's cover with provider art once storage returns.
+
+`locked_fields` records the same fact durably, independent of what is on disk.
+
+The backfill is the load-bearing part: existing uploads are recognisable only by
+the old inferred marker, and reading it once here is the last chance to do so.
+Without it the first scan after upgrading would replace every uploaded cover in
+every library.
+
+Manuals are deliberately not backfilled. An uploaded manual and a scraped one
+share a path and neither clears `url_manual`, so nothing distinguishes them and
+any guess would be wrong for half the rows. Manuals stay pinned by the scan, and
+uploads made from here on are marked as they happen.
+
+Revision ID: 0108_roms_locked_fields
+Revises: 0107_roms_dedup_cover_index
+Create Date: 2026-08-08 00:00:00.000000
+
+"""
+
+import sqlalchemy as sa
+from alembic import op
+
+from utils.database import CustomJSON
+
+# revision identifiers, used by Alembic.
+revision = "0108_roms_locked_fields"
+down_revision = "0107_roms_dedup_cover_index"
+branch_labels = None
+depends_on = None
+
+
+def _roms_table() -> sa.TableClause:
+    return sa.table(
+        "roms",
+        sa.column("path_cover_s", sa.Text),
+        sa.column("url_cover", sa.Text),
+        sa.column("locked_fields", CustomJSON()),
+    )
+
+
+def upgrade() -> None:
+    with op.batch_alter_table("roms", schema=None) as batch_op:
+        batch_op.add_column(
+            sa.Column("locked_fields", CustomJSON(), nullable=True),
+            if_not_exists=True,
+        )
+
+    roms = _roms_table()
+    connection = op.get_bind()
+
+    connection.execute(roms.update().values(locked_fields=[]))
+
+    # A stored cover path with an empty url is the pre-migration marker for an
+    # upload. Values go through the JSON type rather than a literal so each
+    # dialect serialises its own way.
+    connection.execute(
+        roms.update()
+        .where(
+            sa.and_(
+                roms.c.path_cover_s.isnot(None),
+                roms.c.path_cover_s != "",
+                sa.or_(roms.c.url_cover.is_(None), roms.c.url_cover == ""),
+            )
+        )
+        .values(locked_fields=["url_cover"])
+    )
+
+
+def downgrade() -> None:
+    with op.batch_alter_table("roms", schema=None) as batch_op:
+        batch_op.drop_column("locked_fields", if_exists=True)

backend/endpoints/roms/__init__.py

@@ -1845,9 +1845,17 @@ async def update_rom(
             }
         )
 
+    # The cover and manual blocks below both take and release locks, so they
+    # share one running set that is written back once. Deriving each change from
+    # rom.locked_fields instead would let the later block discard the earlier
+    # block's change, since both would start from the same pre-update state.
+    locked_fields = set(rom.locked_fields or [])
+
     if remove_cover:
         cleaned_data.update(await fs_resource_handler.remove_cover(rom))
+        # Dropping the hand-supplied cover hands the slot back to the providers.
         cleaned_data.update({"url_cover": ""})
+        locked_fields.discard("url_cover")
     else:
         if artwork is not None and artwork.filename is not None:
             file_ext = validate_image_upload(artwork, label="Artwork")
@@ -1857,13 +1865,16 @@ async def update_rom(
                 path_cover_s,
             ) = await fs_resource_handler.store_artwork(rom, artwork_content, file_ext)
 
+            # Supplying a file is the explicit act that locks the cover; the
+            # lock outlives the file, so losing it to a scan can't unlock it.
             cleaned_data.update(
                 {
                     "url_cover": "",
                     "path_cover_s": path_cover_s,
                     "path_cover_l": path_cover_l,
                 }
             )
+            locked_fields.add("url_cover")
         else:
             url_cover = (
                 form_data.url_cover if "url_cover" in provided_fields else rom.url_cover
@@ -1881,6 +1892,12 @@ async def update_rom(
                         "path_cover_l": path_cover_l,
                     }
                 )
+                # Naming a different source url is a handover back to the
+                # providers. Testing that it changed, not merely that one was
+                # sent, matters because the client posts the stored url on every
+                # save, so a plain save must not release the lock.
+                if url_cover and url_cover != rom.url_cover:
+                    locked_fields.discard("url_cover")
             except ValidationError as e:
                 log.error(f"Invalid cover URL in update_rom: {str(e)}")
                 raise HTTPException(status_code=400, detail=str(e)) from e
@@ -1900,10 +1917,17 @@ async def update_rom(
                 "path_manual": path_manual,
             }
         )
+        # Same handover as the cover. An upload leaves url_manual untouched, so
+        # unlike the cover this url is often already set on a locked manual,
+        # which is exactly why only an actual change may release it.
+        if url_manual and url_manual != rom.url_manual:
+            locked_fields.discard("url_manual")
     except ValidationError as e:
         log.error(f"Invalid manual URL in update_rom: {str(e)}")
         raise HTTPException(status_code=400, detail=str(e)) from e
 
+    cleaned_data["locked_fields"] = sorted(locked_fields)
+
     # Handle RetroAchievements badges when the ID has changed
     if cleaned_data["ra_id"] and int(cleaned_data["ra_id"]) != rom.ra_id:
         for ach in cleaned_data.get("ra_metadata", {}).get("achievements", []):

backend/endpoints/roms/manual.py

@@ -120,10 +120,13 @@ def cleanup_partial_file():
             detail="There was an error uploading the manual",
         ) from exc
 
+    # An uploaded manual and a scraped one share this path and neither clears
+    # url_manual, so the lock is the only thing that will tell them apart.
     db_rom_handler.update_rom(
         id,
         {
             "path_manual": f"{manuals_path}/{rom.id}{ext}",
+            "locked_fields": rom.locked_fields_with("url_manual"),
         },
     )
 
@@ -163,7 +166,14 @@ async def redownload_rom_manual(
             overwrite=True,
             url_manual=str(rom.url_manual),
         )
-        db_rom_handler.update_rom(id, {"path_manual": path_manual})
+        # Asking for the provider's manual back is a handover.
+        db_rom_handler.update_rom(
+            id,
+            {
+                "path_manual": path_manual,
+                "locked_fields": rom.locked_fields_without("url_manual"),
+            },
+        )
         log.info(
             f"Re-downloaded manual for {hl(rom.name or 'ROM', color=BLUE)} "
             f"[{hl(rom.fs_name)}]"
@@ -384,6 +394,7 @@ async def delete_rom_manuals(
             {
                 "path_manual": "",
                 "url_manual": "",
+                "locked_fields": rom.locked_fields_without("url_manual"),
             },
         )
 
@@ -400,6 +411,7 @@ async def delete_rom_manuals(
             {
                 "path_manual": "",
                 "url_manual": "",
+                "locked_fields": rom.locked_fields_without("url_manual"),
             },
         )
     except Exception as exc:

backend/handler/filesystem/resources_handler.py

@@ -473,6 +473,11 @@ async def _store_screenshot(self, rom: Rom, url_screenhot: str, idx: int):
                 log.error(f"Unable to write screenshot for {url_screenhot}: {str(exc)}")
                 return None
 
+    def _stored_screenshot_count(self, rom: Rom) -> int:
+        """How many screenshot files this rom actually has on disk."""
+        full_path = self.validate_path(f"{rom.fs_resources_path}/screenshots")
+        return sum(1 for _ in full_path.glob("*.jpg"))
+
     def screenshots_exist(self, rom: Rom) -> bool:
         """Check if rom screenshots exist in filesystem
 
@@ -507,16 +512,24 @@ async def get_rom_screenshots(
         Returns
             List of paths to screenshots
         """
-        # Return existing screenshots if no URLs provided
-        # Or if not overwriting and screenshots already exist
-        if not url_screenshots or (not overwrite and self.screenshots_exist(rom)):
+        if not url_screenshots:
+            return rom.path_screenshots or []
+
+        # Count what is on disk rather than what was recorded: fewer files than
+        # urls means an earlier run lost some, and the url set alone would never
+        # signal that, so fall through and fetch them again.
+        if not overwrite and self._stored_screenshot_count(rom) >= len(url_screenshots):
             return rom.path_screenshots or []
 
         # Download and store new screenshots
         path_screenshots: list[str] = []
         for idx, url_screenshot in enumerate(url_screenshots):
             await self._store_screenshot(rom, url_screenshot, idx)
-            path_screenshots.append(self._get_screenshot_path(rom, str(idx)))
+            path = self._get_screenshot_path(rom, str(idx))
+            # A failed download leaves nothing behind, and recording its path
+            # anyway points the database at a file that isn't there.
+            if await self.file_exists(path):
+                path_screenshots.append(path)
 
         return path_screenshots
 

backend/handler/scan_handler.py

@@ -442,6 +442,7 @@ async def scan_rom(
                 "path_cover_l": rom.path_cover_l,
                 "path_screenshots": rom.path_screenshots,
                 "path_manual": rom.path_manual,
+                "locked_fields": rom.locked_fields,
                 "igdb_id": rom.igdb_id,
                 "moby_id": rom.moby_id,
                 "ss_id": rom.ss_id,
@@ -996,7 +997,10 @@ async def fetch_hasheous_rom(hasheous_rom: HasheousRom) -> HasheousRom:
                 if fields["metadata_field"]:
                     rom_attrs[fields["metadata_field"]] = {}
 
-        # Reset artwork fields so stale values are cleared when no source supplies them
+        # Reset artwork fields so stale values are cleared when no source supplies them.
+        # The locks go with them: a complete rescan deletes the resource files, so
+        # keeping one would leave a cover locked to a file that no longer exists and
+        # block any replacement from being fetched.
         rom_attrs.update(
             {
                 "url_cover": "",
@@ -1006,6 +1010,7 @@ async def fetch_hasheous_rom(hasheous_rom: HasheousRom) -> HasheousRom:
                 "path_cover_l": "",
                 "path_screenshots": [],
                 "path_manual": "",
+                "locked_fields": [],
             }
         )
 
@@ -1064,16 +1069,22 @@ async def fetch_hasheous_rom(hasheous_rom: HasheousRom) -> HasheousRom:
             {
                 "name": existing_name or matched_name or fs_name_no_tags or None,
                 "summary": rom.summary or rom_attrs.get("summary") or None,
-                # Don't overwrite existing manually uploaded cover image
+                # A locked cover was supplied by hand and stays put. Anything
+                # else came from a provider, and the freshly resolved url
+                # replaces it so a changed source or source priority reaches the
+                # download step.
                 "url_cover": (
-                    rom.url_cover
-                    if rom.path_cover_s
+                    ""
+                    if rom.is_field_locked("url_cover")
                     else rom_attrs.get("url_cover") or None
                 ),
+                # A manual keeps its stored url because an uploaded manual and a
+                # scraped one share a path and neither clears the url, so there
+                # is nothing here to tell them apart.
                 "url_manual": rom.url_manual or rom_attrs.get("url_manual") or None,
-                "url_screenshots": rom.url_screenshots
-                or rom_attrs.get("url_screenshots")
-                or [],
+                # Screenshots have no upload path, so every stored url came from
+                # a provider and the freshly resolved set wins.
+                "url_screenshots": rom_attrs.get("url_screenshots") or [],
             }
         )
 

backend/models/rom.py

@@ -482,6 +482,12 @@ class Rom(BaseModel):
         CustomJSON(), default=[], doc="URLs to screenshots stored in IGDB"
     )
 
+    locked_fields: Mapped[list[str] | None] = mapped_column(
+        CustomJSON(),
+        default=[],
+        doc="Fields a user supplied by hand, which scans must not overwrite",
+    )
+
     revision: Mapped[str | None] = mapped_column(String(length=100))
     version: Mapped[str | None] = mapped_column(String(length=100))
     regions: Mapped[list[str] | None] = mapped_column(CustomJSON(), default=[])
@@ -651,6 +657,18 @@ def path_video(self) -> str | None:
             or (self.launchbox_metadata or {}).get("video_path")
         )
 
+    def is_field_locked(self, field: str) -> bool:
+        """Whether a user supplied this field by hand, so scans must leave it."""
+        return field in (self.locked_fields or [])
+
+    def locked_fields_with(self, field: str) -> list[str]:
+        """This rom's locks plus ``field``, for handing to an update."""
+        return sorted({*(self.locked_fields or []), field})
+
+    def locked_fields_without(self, field: str) -> list[str]:
+        """This rom's locks minus ``field``, for handing to an update."""
+        return [f for f in (self.locked_fields or []) if f != field]
+
     @property
     def is_unidentified(self) -> bool:
         return (

backend/tests/endpoints/roms/test_manual.py

@@ -89,6 +89,9 @@ def test_upload_manual_to_resources_success(
     assert written.read_bytes() == PDF_BYTES
     refreshed = db_rom_handler.get_rom(rom.id)
     assert refreshed.path_manual == f"{rom.fs_resources_path}/manual/{rom.id}.pdf"
+    # An uploaded manual and a scraped one land at the same path and neither
+    # clears url_manual, so the lock is the only thing telling them apart.
+    assert refreshed.locked_fields == ["url_manual"]
 
 
 def test_upload_markdown_manual_to_resources_preserves_extension(
@@ -252,7 +255,11 @@ def test_redownload_manual_success(
     monkeypatch: pytest.MonkeyPatch,
 ):
     db_rom_handler.update_rom(
-        rom.id, {"url_manual": "https://screenscraper.fr/api/manual.pdf"}
+        rom.id,
+        {
+            "url_manual": "https://screenscraper.fr/api/manual.pdf",
+            "locked_fields": ["url_manual"],
+        },
     )
     fake_path = f"{rom.fs_resources_path}/manual/{rom.id}.pdf"
     monkeypatch.setattr(
@@ -269,6 +276,8 @@ def test_redownload_manual_success(
     assert response.status_code == status.HTTP_200_OK
     refreshed = db_rom_handler.get_rom(rom.id)
     assert refreshed.path_manual == fake_path
+    # Asking for the provider's manual back is a handover.
+    assert refreshed.locked_fields == []
 
 
 # ---------- DELETE /api/roms/{id}/manuals (resources) ----------
@@ -305,6 +314,7 @@ def test_delete_manual_success(
         {
             "path_manual": f"{rom.fs_resources_path}/manual/{rom.id}.pdf",
             "url_manual": "https://screenscraper.fr/api/manual.pdf",
+            "locked_fields": ["url_manual"],
         },
     )
     monkeypatch.setattr(
@@ -325,6 +335,8 @@ def test_delete_manual_success(
     refreshed = db_rom_handler.get_rom(rom.id)
     assert refreshed.path_manual == ""
     assert refreshed.url_manual == ""
+    # Deleting the hand-supplied manual hands the slot back to the providers.
+    assert refreshed.locked_fields == []
 
 
 # ---------- DELETE /api/roms/{id}/manuals/files/{file_id} ----------