feat(firmware): let missing firmware be found, cleaned up, and kept out of the player

backend/endpoints/firmware.py

@@ -1,6 +1,6 @@
 from typing import Annotated
 
-from fastapi import Body, File, HTTPException, Request, UploadFile, status
+from fastapi import Body, File, HTTPException, Query, Request, UploadFile, status
 from fastapi.responses import FileResponse
 
 from config import DISABLE_DOWNLOAD_ENDPOINT_AUTH
@@ -124,6 +124,10 @@ async def add_firmware(
 def get_platform_firmware(
     request: Request,
     platform_id: int | None = None,
+    missing: Annotated[
+        bool | None,
+        Query(description="Whether the firmware is missing from the filesystem."),
+    ] = None,
 ) -> list[FirmwareSchema]:
     """Get firmware endpoint
 
@@ -138,6 +142,7 @@ def get_platform_firmware(
         FirmwareSchema.model_validate(f)
         for f in db_firmware_handler.list_firmware(
             platform_id=platform_id,
+            missing=missing,
             hidden_platform_ids=perms.hidden_platform_ids,
         )
     ]

backend/endpoints/responses/firmware.py

@@ -9,6 +9,7 @@ class FirmwareSchema(BaseModel):
     model_config = ConfigDict(from_attributes=True)
 
     id: int
+    platform_id: int
 
     file_name: str
     file_name_no_tags: str

backend/endpoints/responses/platform.py

@@ -46,10 +46,12 @@ class PlatformSchema(BaseModel):
     def display_name(self) -> str:
         return self.custom_name or self.name
 
+    # Missing entries stay in `firmware` so they can be cleaned up, but they
+    # aren't usable BIOS, so they don't count.
     @computed_field  # type: ignore
     @property
     def firmware_count(self) -> int:
-        return len(self.firmware)
+        return len([f for f in self.firmware if not f.missing_from_fs])
 
     @field_validator("firmware")
     def sort_files(cls, v: list[FirmwareSchema]) -> list[FirmwareSchema]:

backend/endpoints/tasks.py

@@ -33,6 +33,7 @@
     low_prio_queue,
     redis_client,
 )
+from tasks.manual.cleanup_missing_firmware import cleanup_missing_firmware_task
 from tasks.manual.cleanup_missing_roms import cleanup_missing_roms_task
 from tasks.manual.recompute_save_content_hashes import (
     recompute_save_content_hashes_task,
@@ -119,6 +120,13 @@ class ManualTask(ScheduledTask):
             "task": cleanup_missing_roms_task,
         }
     ),
+    ManualTask(
+        {
+            "name": "cleanup_missing_firmware",
+            "type": TaskType.CLEANUP,
+            "task": cleanup_missing_firmware_task,
+        }
+    ),
     ManualTask(
         {
             "name": "sync_folder_scan",

backend/handler/database/firmware_handler.py

@@ -32,6 +32,7 @@ def list_firmware(
         self,
         *,
         platform_id: int | None = None,
+        missing: bool | None = None,
         only_fields: Sequence[QueryableAttribute] | None = None,
         hidden_platform_ids: Sequence[int] | None = None,
         session: Session = None,  # type: ignore
@@ -41,6 +42,9 @@ def list_firmware(
         if platform_id:
             query = query.filter_by(platform_id=platform_id)
 
+        if missing is not None:
+            query = query.filter(Firmware.missing_from_fs == missing)
+
         # Firmware inherits its platform's visibility: hide firmware whose
         # platform an admin has hidden from the caller.
         if hidden_platform_ids:

backend/tasks/manual/cleanup_missing_firmware.py

@@ -0,0 +1,83 @@
+from dataclasses import dataclass
+
+from handler.database import db_firmware_handler
+from logger.logger import log
+from tasks.tasks import Task, TaskType, update_job_meta
+from utils.context import initialize_context
+
+
+@dataclass
+class CleanupMissingFirmwareStats:
+    """Statistics for missing firmware cleanup operations."""
+
+    platform_id: int | None = None
+    firmware_found: int = 0
+    firmware_deleted: int = 0
+    errors: int = 0
+
+    def update(self, **kwargs) -> None:
+        for key, value in kwargs.items():
+            if hasattr(self, key):
+                setattr(self, key, value)
+
+        update_job_meta({"cleanup_stats": self.to_dict()})
+
+    def to_dict(self) -> dict:
+        return {
+            "platform_id": self.platform_id,
+            "firmware_found": self.firmware_found,
+            "firmware_deleted": self.firmware_deleted,
+            "errors": self.errors,
+        }
+
+
+class CleanupMissingFirmwareTask(Task):
+    def __init__(self):
+        super().__init__(
+            title="Cleanup missing firmware",
+            description="Delete all firmware flagged as missing from the filesystem from the database",
+            task_type=TaskType.CLEANUP,
+            enabled=True,
+            manual_run=True,
+            cron_string=None,
+        )
+
+    @initialize_context()
+    async def run(self, platform_id: int | None = None) -> dict:
+        """Clean up firmware that is flagged as missing from the filesystem."""
+        log.info(f"Starting {self.title} task...")
+
+        stats = CleanupMissingFirmwareStats(platform_id=platform_id)
+
+        missing_firmware = db_firmware_handler.list_firmware(
+            platform_id=platform_id, missing=True
+        )
+
+        stats.update(firmware_found=len(missing_firmware))
+        log.info(
+            f"Found {len(missing_firmware)} missing firmware file(s) to clean up"
+            + (f" for platform ID {platform_id}" if platform_id else "")
+        )
+
+        # The row is stale because the file is already gone, so there is
+        # nothing to remove from disk here.
+        for firmware in missing_firmware:
+            try:
+                log.info(
+                    f"Deleting missing firmware '{firmware.file_name}' [ID: {firmware.id}] from database"
+                )
+                db_firmware_handler.delete_firmware(firmware.id)
+            except Exception as e:
+                log.error(f"Failed to delete missing firmware {firmware.id}: {e}")
+                stats.update(errors=stats.errors + 1)
+                continue
+
+            stats.update(firmware_deleted=stats.firmware_deleted + 1)
+
+        log.info(
+            f"Cleanup of missing firmware completed: {stats.firmware_deleted} deleted, {stats.errors} error(s)"
+        )
+        return stats.to_dict()
+
+
+cleanup_missing_firmware_task = CleanupMissingFirmwareTask()

backend/tests/conftest.py

@@ -14,6 +14,7 @@
 from handler.auth import auth_handler
 from handler.auth.base_handler import ALGORITHM, oct_key
 from handler.database import (
+    db_firmware_handler,
     db_permission_handler,
     db_platform_handler,
     db_rom_handler,
@@ -26,6 +27,7 @@
 from models.client_token import ClientToken
 from models.device import Device
 from models.device_save_sync import DeviceSaveSync
+from models.firmware import Firmware
 from models.platform import Platform
 from models.play_session import PlaySession
 from models.rom import Rom, RomFile
@@ -100,6 +102,7 @@ def clear_database():
         s.query(Screenshot).delete(synchronize_session="evaluate")
         s.query(RomFile).delete(synchronize_session="evaluate")
         s.query(Rom).delete(synchronize_session="evaluate")
+        s.query(Firmware).delete(synchronize_session="evaluate")
         s.query(Platform).delete(synchronize_session="evaluate")
         s.query(User).delete(synchronize_session="evaluate")
 
@@ -124,6 +127,35 @@ def platform():
     return db_platform_handler.add_platform(platform)
 
 
+def _build_firmware(platform: Platform, file_name: str, missing: bool) -> Firmware:
+    return Firmware(
+        platform_id=platform.id,
+        file_name=file_name,
+        file_path=f"{platform.fs_slug}/bios",
+        file_size_bytes=1024,
+        crc_hash="crc",
+        md5_hash="md5",
+        sha1_hash="sha1",
+        missing_from_fs=missing,
+    )
+
+
+@pytest.fixture
+def firmware(platform: Platform):
+    """Firmware whose file is still on disk."""
+    return db_firmware_handler.add_firmware(
+        _build_firmware(platform, "present.bin", missing=False)
+    )
+
+
+@pytest.fixture
+def missing_firmware(platform: Platform):
+    """Firmware flagged by a scan as gone from the filesystem."""
+    return db_firmware_handler.add_firmware(
+        _build_firmware(platform, "gone.bin", missing=True)
+    )
+
+
 @pytest.fixture
 def rom(admin_user: User, platform: Platform):
     rom = Rom(

backend/tests/endpoints/test_firmware.py

@@ -0,0 +1,78 @@
+"""Tests for `GET /api/firmware`.
+
+Issue #4075: the endpoint returned every row with no way to select on
+`missing_from_fs`, so the player's BIOS list offered (and auto-selected)
+firmware whose file was gone, and nothing could list missing firmware
+library-wide.
+"""
+
+from fastapi import status
+
+
+def test_get_firmware_requires_auth(client, firmware):
+    response = client.get("/api/firmware")
+    assert response.status_code == status.HTTP_401_UNAUTHORIZED
+
+
+def test_get_firmware_returns_everything_by_default(
+    client, access_token, firmware, missing_firmware
+):
+    response = client.get(
+        "/api/firmware", headers={"Authorization": f"Bearer {access_token}"}
+    )
+    assert response.status_code == status.HTTP_200_OK
+
+    names = [f["file_name"] for f in response.json()]
+    assert sorted(names) == ["gone.bin", "present.bin"]
+
+
+def test_get_firmware_missing_true_returns_only_missing(
+    client, access_token, firmware, missing_firmware
+):
+    response = client.get(
+        "/api/firmware",
+        params={"missing": "true"},
+        headers={"Authorization": f"Bearer {access_token}"},
+    )
+    assert response.status_code == status.HTTP_200_OK
+
+    body = response.json()
+    assert [f["file_name"] for f in body] == ["gone.bin"]
+    assert body[0]["missing_from_fs"] is True
+
+
+def test_get_firmware_missing_false_excludes_missing(
+    client, access_token, firmware, missing_firmware
+):
+    response = client.get(
+        "/api/firmware",
+        params={"missing": "false"},
+        headers={"Authorization": f"Bearer {access_token}"},
+    )
+    assert response.status_code == status.HTTP_200_OK
+
+    body = response.json()
+    assert [f["file_name"] for f in body] == ["present.bin"]
+    assert body[0]["missing_from_fs"] is False
+
+
+def test_get_firmware_exposes_its_platform(client, access_token, platform, firmware):
+    """The library-wide missing view groups by platform, so the row has to
+    carry one without a second round trip per entry."""
+    response = client.get(
+        "/api/firmware", headers={"Authorization": f"Bearer {access_token}"}
+    )
+    assert response.status_code == status.HTTP_200_OK
+    assert response.json()[0]["platform_id"] == platform.id
+
+
+def test_get_firmware_missing_filter_stacks_with_platform_id(
+    client, access_token, platform, firmware, missing_firmware
+):
+    response = client.get(
+        "/api/firmware",
+        params={"platform_id": platform.id, "missing": "true"},
+        headers={"Authorization": f"Bearer {access_token}"},
+    )
+    assert response.status_code == status.HTTP_200_OK
+    assert [f["file_name"] for f in response.json()] == ["gone.bin"]

backend/tests/endpoints/test_platform.py

@@ -176,3 +176,21 @@ def test_update_platform_description_requires_write_scope(client, platform):
         json={"description": "Nope"},
     )
     assert response.status_code == status.HTTP_401_UNAUTHORIZED
+
+
+def test_firmware_count_excludes_missing_firmware(
+    client, access_token, platform, firmware, missing_firmware
+):
+    """A platform whose only BIOS file was deleted from disk shouldn't still
+    advertise it (issue #4075)."""
+    response = client.get(
+        f"/api/platforms/{platform.id}",
+        headers={"Authorization": f"Bearer {access_token}"},
+    )
+    assert response.status_code == status.HTTP_200_OK
+
+    body = response.json()
+    assert body["firmware_count"] == 1
+    # The rows themselves still ship so the Firmware tab can strike them
+    # through and offer a cleanup.
+    assert len(body["firmware"]) == 2

backend/tests/endpoints/test_tasks.py

@@ -183,6 +183,19 @@ def test_list_tasks_empty(self, client, access_token):
         assert data["watcher"][0]["enabled"] is False
         assert "10 minute delay" in data["watcher"][0]["description"]
 
+    def test_missing_firmware_cleanup_is_registered(self, client, access_token):
+        """Unpatched registry: the Missing tab runs this task by name, so a
+        missing registration is a 404 at the point of use (issue #4075)."""
+        response = client.get(
+            "/api/tasks", headers={"Authorization": f"Bearer {access_token}"}
+        )
+
+        assert response.status_code == status.HTTP_200_OK
+        manual = {t["name"]: t for t in response.json()["manual"]}
+        assert "cleanup_missing_firmware" in manual
+        assert manual["cleanup_missing_firmware"]["manual_run"] is True
+        assert manual["cleanup_missing_firmware"]["type"] == TaskType.CLEANUP.value
+
     def test_list_tasks_unauthorized(self, client):
         """Test that unauthorized requests are rejected"""
         response = client.get("/api/tasks")