feat(retroachievements): flag which ROM version RA actually supports

backend/alembic/versions/0108_roms_ra_hash_match.py

@@ -0,0 +1,39 @@
+"""Record whether RetroAchievements knows a ROM's own hash
+
+`ra_id` is the RetroAchievements *game*, and every version of a game resolves
+to the same one: on the Hasheous path it comes from the matched game's metadata
+list, so a bad dump and a good dump of the same title are indistinguishable by
+it. Achievements, though, only unlock when the file's RA hash is in RA's hash
+list, which is a per-file fact RomM computed during scans and then threw away.
+
+This column keeps it. NULL means never checked -- no RA hash, a platform RA
+doesn't cover, or a ROM last scanned before this existed -- so it stays
+distinguishable from a checked-and-absent hash.
+
+Revision ID: 0108_roms_ra_hash_match
+Revises: 0107_roms_dedup_cover_index
+Create Date: 2026-08-05 00:00:00.000000
+
+"""
+
+import sqlalchemy as sa
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision = "0108_roms_ra_hash_match"
+down_revision = "0107_roms_dedup_cover_index"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    with op.batch_alter_table("roms", schema=None) as batch_op:
+        batch_op.add_column(
+            sa.Column("ra_hash_match", sa.Boolean(), nullable=True),
+            if_not_exists=True,
+        )
+
+
+def downgrade() -> None:
+    with op.batch_alter_table("roms", schema=None) as batch_op:
+        batch_op.drop_column("ra_hash_match", if_exists=True)

backend/endpoints/responses/rom.py

@@ -331,6 +331,7 @@ class RomSchema(BaseModel):
     md5_hash: str | None
     sha1_hash: str | None
     ra_hash: str | None
+    ra_hash_match: bool | None
 
     has_simple_single_file: bool
     has_nested_single_file: bool
@@ -381,6 +382,9 @@ class SiblingRomSchema(BaseModel):
     fs_name_no_tags: str
     fs_name_no_ext: str
     is_main_sibling: bool
+    # Per-file, unlike `ra_id`, so the version switcher can flag which
+    # sibling's achievements will actually unlock.
+    ra_hash_match: bool | None
 
     @computed_field  # type: ignore
     @property
@@ -402,6 +406,7 @@ def from_rom(cls, rom: Rom, *, is_main_sibling: bool = False) -> SiblingRomSchem
             fs_name_no_tags=rom.fs_name_no_tags,
             fs_name_no_ext=rom.fs_name_no_ext,
             is_main_sibling=is_main_sibling,
+            ra_hash_match=rom.ra_hash_match,
         )
 
 

backend/handler/database/roms_handler.py

@@ -363,6 +363,7 @@ def wrapper(*args, **kwargs):
                     Rom.platform_id,
                     Rom.fs_name_no_tags,
                     Rom.fs_name_no_ext,
+                    Rom.ra_hash_match,
                 ),
             ),
             selectinload(Rom.collections),
@@ -408,6 +409,7 @@ def wrapper(*args, **kwargs):
                     Rom.platform_id,
                     Rom.fs_name_no_tags,
                     Rom.fs_name_no_ext,
+                    Rom.ra_hash_match,
                 ),
             ),
             selectinload(Rom.notes),
@@ -537,6 +539,7 @@ def get_siblings_for_roms(
                     Rom.name,
                     Rom.fs_name_no_tags,
                     Rom.fs_name_no_ext,
+                    Rom.ra_hash_match,
                 )
             )
         )
@@ -856,6 +859,10 @@ def _filter_by_missing_from_fs(self, query: Query, value: bool) -> Query:
         return query.filter(Rom.missing_from_fs == (true() if value else false()))
 
     def _filter_by_verified(self, query: Query, value: bool) -> Query:
+        # Databases Hasheous answers for. RetroAchievements is handled
+        # separately below: RomM asks RA directly, and that answer outranks
+        # Hasheous' RA signature coverage in both directions. Keep in step
+        # with `romVerification.ts` on the frontend.
         keys_to_check = [
             "tosec_match",
             "mame_arcade_match",
@@ -864,7 +871,6 @@ def _filter_by_verified(self, query: Query, value: bool) -> Query:
             "redump_match",
             "mame_redump_match",
             "whdload_match",
-            "ra_match",
             "fbneo_match",
             "puredos_match",
         ]
@@ -879,13 +885,25 @@ def _filter_by_verified(self, query: Query, value: bool) -> Query:
                 f"COALESCE((hasheous_metadata->>'{key}')::boolean, false)"
                 for key in keys_to_check
             )
+            # RA verified when its own hash list says so, or when it was
+            # never asked (NULL) and Hasheous saw the hash.
+            conditions += (
+                " OR ra_hash_match IS TRUE"
+                " OR (ra_hash_match IS NULL AND"
+                " COALESCE((hasheous_metadata->>'ra_match')::boolean, false))"
+            )
             predicate = text(f"({conditions})")
             if not value:
                 predicate = text(f"NOT ({conditions})")
             return query.filter(predicate)
         else:
             predicate = or_(
-                *(Rom.hasheous_metadata[key].as_boolean() for key in keys_to_check)
+                *(Rom.hasheous_metadata[key].as_boolean() for key in keys_to_check),
+                Rom.ra_hash_match.is_(true()),
+                and_(
+                    Rom.ra_hash_match.is_(None),
+                    Rom.hasheous_metadata["ra_match"].as_boolean(),
+                ),
             )
             if not value:
                 predicate = not_(predicate)

backend/handler/metadata/ra_handler.py

@@ -26,6 +26,17 @@
 # Regex to detect RetroAchievements ID tags in filenames like (ra-12345)
 RA_TAG_REGEX = re.compile(r"\(ra-(\d+)\)", re.IGNORECASE)
 
+# How long an in-process copy of a platform's hash list stays usable.
+# Long enough to serve a whole scan from one parse, short enough that the
+# on-disk refresh still gets picked up.
+HASH_INDEX_CACHE_TTL_SECONDS = 900
+
+# A hash absent from a list this old (in days) is refetched before being
+# reported as unsupported. REFRESH_RETROACHIEVEMENTS_CACHE_DAYS governs when
+# the list is refreshed for matching, and defaults to 30: far too coarse to
+# record "RetroAchievements doesn't have this dump" against.
+NEGATIVE_MAX_LIST_AGE_DAYS = 1
+
 
 class RAGamesPlatform(TypedDict):
     slug: str
@@ -59,6 +70,7 @@ class RAMetadata(TypedDict):
 class RAGameRom(BaseRom):
     ra_id: int | None
     ra_metadata: NotRequired[RAMetadata]
+    ra_hash_match: NotRequired[bool | None]
 
 
 class EarnedAchievement(TypedDict):
@@ -128,6 +140,8 @@ class RAHandler(MetadataHandler):
     def __init__(self) -> None:
         self.ra_service = RetroAchievementsService()
         self.HASHES_FILE_NAME = "ra_hashes_v2.json"
+        # platform id -> (monotonic expiry, hash index)
+        self._hash_index_cache: dict[int, tuple[float, dict[str, int]]] = {}
 
     @classmethod
     def is_enabled(cls) -> bool:
@@ -173,9 +187,18 @@ async def _days_since_last_cache_file_update(self, platform_id: int) -> int:
         file_stat = await AnyioPath(str(full_path)).stat()
         return int((time.time() - file_stat.st_mtime) / (24 * 3600))
 
-    async def _search_rom(self, rom: Rom, ra_hash: str) -> int | None:
-        if not rom.platform.ra_id:
-            return None
+    async def _get_hash_index(self, rom: Rom) -> dict[str, int]:
+        """RetroAchievements' hash list for the platform: hash -> game ID.
+
+        Memoised per platform: a scan asks for the same index once per
+        ROM, and re-reading a multi-megabyte JSON blob every time
+        dominated the RA leg of large scans. The TTL keeps a long-lived
+        worker from pinning a stale index, since the memo also skips the
+        staleness check below.
+        """
+        cached = self._hash_index_cache.get(rom.platform.id)
+        if cached is not None and cached[0] > time.monotonic():
+            return cached[1]
 
         # hash_index maps lowercase hash -> game ID for O(1) lookups
         hash_index: dict[str, int]
@@ -184,34 +207,91 @@ async def _search_rom(self, rom: Rom, ra_hash: str) -> int | None:
             <= await self._days_since_last_cache_file_update(rom.platform.id)
             or not await self._exists_cache_file(rom.platform.id)
         ):
-            # Fetch all games (including those without achievements) and build index
-            roms = await self.ra_service.get_game_list(
-                system_id=rom.platform.ra_id,
-                only_games_with_achievements=False,
-                include_hashes=True,
-            )
+            return await self._refresh_hash_index(rom)
+
+        # Read the hash index from the JSON file
+        json_file_bytes = await fs_resource_handler.read_file(
+            self._get_hashes_file_path(rom.platform.id)
+        )
+        hash_index = json.loads(json_file_bytes.decode("utf-8"))
 
-            hash_index = {h.lower(): r["ID"] for r in roms for h in r.get("Hashes", ())}
+        self._memoise_hash_index(rom.platform.id, hash_index)
+        return hash_index
 
-            platform_resources_path = fs_resource_handler.get_platform_resources_path(
-                rom.platform.id
-            )
+    def _memoise_hash_index(self, platform_id: int, hash_index: dict[str, int]) -> None:
+        self._hash_index_cache[platform_id] = (
+            time.monotonic() + HASH_INDEX_CACHE_TTL_SECONDS,
+            hash_index,
+        )
 
-            json_file = json.dumps(hash_index, indent=4)
-            await fs_resource_handler.write_file(
-                json_file.encode("utf-8"),
-                platform_resources_path,
-                self.HASHES_FILE_NAME,
-            )
-        else:
-            # Read the hash index from the JSON file
-            json_file_bytes = await fs_resource_handler.read_file(
-                self._get_hashes_file_path(rom.platform.id)
-            )
-            hash_index = json.loads(json_file_bytes.decode("utf-8"))
+    async def _refresh_hash_index(self, rom: Rom) -> dict[str, int]:
+        """Rebuild the platform's hash list from RetroAchievements and store it."""
+        # Fetch all games (including those without achievements) and build index
+        roms = await self.ra_service.get_game_list(
+            system_id=rom.platform.ra_id,
+            only_games_with_achievements=False,
+            include_hashes=True,
+        )
+
+        hash_index = {h.lower(): r["ID"] for r in roms for h in r.get("Hashes", ())}
+
+        platform_resources_path = fs_resource_handler.get_platform_resources_path(
+            rom.platform.id
+        )
+
+        json_file = json.dumps(hash_index, indent=4)
+        await fs_resource_handler.write_file(
+            json_file.encode("utf-8"),
+            platform_resources_path,
+            self.HASHES_FILE_NAME,
+        )
 
+        self._memoise_hash_index(rom.platform.id, hash_index)
+        return hash_index
+
+    async def _search_rom(self, rom: Rom, ra_hash: str) -> int | None:
+        if not rom.platform.ra_id:
+            return None
+
+        hash_index = await self._get_hash_index(rom)
         return hash_index.get(ra_hash.lower())
 
+    async def hash_is_known(self, rom: Rom, ra_hash: str) -> bool | None:
+        """Whether RetroAchievements recognises this exact file.
+
+        Asked of RetroAchievements' own hash list, never of another
+        provider: a sibling that Hasheous identifies inherits the game's
+        `ra_id` whether or not RA has ever seen that dump. Returns None
+        when there is nothing to check against.
+
+        RetroAchievements adds hashes continuously, so a miss against a list
+        this old is not evidence of anything. Refetch before reporting one,
+        which costs a single request per platform per scan: the fresh list is
+        both memoised and written to disk, so the next miss reads it as
+        current.
+        """
+        if not rom.platform.ra_id or not ra_hash:
+            return None
+
+        needle = ra_hash.lower()
+        try:
+            hash_index = await self._get_hash_index(rom)
+            if needle in hash_index:
+                return True
+
+            stale_days = await self._days_since_last_cache_file_update(rom.platform.id)
+            if stale_days >= NEGATIVE_MAX_LIST_AGE_DAYS:
+                hash_index = await self._refresh_hash_index(rom)
+        except Exception as exc:
+            log.error(
+                "Couldn't read the RetroAchievements hash list, "
+                "leaving hash support unknown: %s",
+                exc,
+            )
+            return None
+
+        return needle in hash_index
+
     def get_platform(self, slug: str) -> RAGamesPlatform:
         if slug not in RA_PLATFORM_LIST:
             return RAGamesPlatform(ra_id=None, slug=slug)

backend/handler/scan_handler.py

@@ -793,6 +793,13 @@ async def fetch_ra_rom(hasheous_rom: HasheousRom) -> RAGameRom:
                 )
             )
         ):
+            # Asked of RA's own hash list whatever identified the game, since
+            # an `ra_id` from Hasheous or a filename tag says nothing about
+            # this file.
+            ra_hash_match = await meta_ra_handler.hash_is_known(
+                rom=rom, ra_hash=rom_attrs["ra_hash"]
+            )
+
             # Use Hasheous match to get the RA ID
             h_ra_id = hasheous_rom.get("ra_id")
             if h_ra_id:
@@ -801,17 +808,19 @@ async def fetch_ra_rom(hasheous_rom: HasheousRom) -> RAGameRom:
                     f"{hl(str(h_ra_id), color=BLUE)} {emoji.EMOJI_ALIEN_MONSTER}",
                     extra=LOGGER_MODULE_NAME,
                 )
-                return await meta_ra_handler.get_rom_by_id(rom=rom, ra_id=h_ra_id)
-
-            if (scan_type == ScanType.UPDATE and rom.ra_id) or (
+                ra_rom = await meta_ra_handler.get_rom_by_id(rom=rom, ra_id=h_ra_id)
+            elif (scan_type == ScanType.UPDATE and rom.ra_id) or (
                 scan_type == ScanType.UNMATCHED and rom.ra_id and not rom.ra_metadata
             ):
-                return await meta_ra_handler.get_rom_by_id(rom=rom, ra_id=rom.ra_id)
+                ra_rom = await meta_ra_handler.get_rom_by_id(rom=rom, ra_id=rom.ra_id)
             else:
-                return await meta_ra_handler.get_rom(
+                ra_rom = await meta_ra_handler.get_rom(
                     rom=rom, ra_hash=rom_attrs["ra_hash"]
                 )
 
+            ra_rom["ra_hash_match"] = ra_hash_match
+            return ra_rom
+
         return RAGameRom(ra_id=None)
 
     async def fetch_hasheous_rom(hasheous_rom: HasheousRom) -> HasheousRom:
@@ -1006,6 +1015,14 @@ async def fetch_hasheous_rom(hasheous_rom: HasheousRom) -> HasheousRom:
             if field_value:
                 rom_attrs[key] = field_value
 
+    # Tri-state, so it can ride neither the truthy-only loop above (False would
+    # be dropped, reading as never-checked) nor its `ra_id` gate (Hasheous can
+    # supply an id for a file RA has never seen). None means the question went
+    # unanswered, e.g. no hash to check or an unreadable hash list, so keep
+    # whatever the last scan that did get an answer recorded.
+    if ra_handler_rom.get("ra_hash_match") is not None:
+        rom_attrs["ra_hash_match"] = ra_handler_rom["ra_hash_match"]
+
     # Artwork sources are prioritized separately, and each field can carry its
     # own override on top of the shared artwork priority.
     for field in ["url_cover", "url_screenshots", "url_manual"]:

backend/models/rom.py

@@ -441,6 +441,10 @@ class Rom(BaseModel):
     md5_hash: Mapped[str | None] = mapped_column(String(length=100))
     sha1_hash: Mapped[str | None] = mapped_column(String(length=100))
     ra_hash: Mapped[str | None] = mapped_column(String(length=100))
+    # Whether `ra_hash` is in RetroAchievements' own hash list, i.e. whether
+    # achievements unlock for THIS file. `ra_id` is the game, shared by every
+    # version. NULL means never checked.
+    ra_hash_match: Mapped[bool | None] = mapped_column(default=None)
 
     missing_from_fs: Mapped[bool] = mapped_column(default=False, nullable=False)