scan.py

from __future__ import annotations

import asyncio
from dataclasses import dataclass
from itertools import batched
from typing import Any, Final

import pydash
import socketio  # type: ignore
from rq import Worker
from rq.job import Job
from sqlalchemy.exc import IntegrityError

from adapters.services.screenscraper import reset_daily_quota as reset_ss_daily_quota
from config import DEV_MODE, REDIS_URL, SCAN_TIMEOUT, SCAN_WORKERS, TASK_RESULT_TTL
from config.config_manager import MetadataMediaType
from config.config_manager import config_manager as cm
from endpoints.responses import TaskType
from endpoints.responses.platform import PlatformSchema
from endpoints.responses.rom import SimpleRomSchema
from endpoints.sockets.activity import get_authenticated_user
from exceptions.fs_exceptions import (
    FOLDER_STRUCT_MSG,
    FirmwareNotFoundException,
    FolderStructureNotMatchException,
    RomsNotFoundException,
)
from exceptions.socket_exceptions import ScanStoppedException
from handler.auth.constants import Scope
from handler.database import db_firmware_handler, db_platform_handler, db_rom_handler
from handler.filesystem import (
    fs_firmware_handler,
    fs_platform_handler,
    fs_resource_handler,
    fs_rom_handler,
)
from handler.filesystem.roms_handler import FSRom
from handler.metadata import meta_gamelist_handler, meta_hltb_handler
from handler.metadata.ss_handler import add_ss_auth_to_url, get_preferred_media_types
from handler.redis_handler import get_job_func_name, high_prio_queue, redis_client
from handler.scan_handler import (
    MetadataSource,
    ScanType,
    persist_soundtrack_cover,
    scan_firmware,
    scan_platform,
    scan_rom,
)
from handler.socket_handler import socket_handler
from logger.formatter import BLUE, LIGHTYELLOW
from logger.formatter import highlight as hl
from logger.logger import log
from models.firmware import Firmware
from models.platform import Platform
from models.rom import Rom, RomFile, TrackMeta
from tasks.tasks import update_job_meta
from utils import emoji
from utils.context import initialize_context
from utils.gamelist_exporter import GamelistExporter
from utils.pegasus_exporter import PegasusExporter

STOP_SCAN_FLAG: Final = "scan:stop"


def _clone_track_meta(src: TrackMeta | None, rom_id: int) -> TrackMeta | None:
    """Build a fresh TrackMeta from a scanned (transient) one for a new RomFile."""
    if src is None:
        return None
    return TrackMeta(
        rom_id=rom_id,
        title=src.title,
        artist=src.artist,
        album=src.album,
        genre=src.genre,
        year=src.year,
        track=src.track,
        disc=src.disc,
        duration_seconds=src.duration_seconds,
        has_embedded_cover=src.has_embedded_cover,
        cover_path=src.cover_path,
    )


@dataclass
class ScanStats:
    total_platforms: int = 0
    total_roms: int = 0
    scanned_platforms: int = 0
    new_platforms: int = 0
    identified_platforms: int = 0
    scanned_roms: int = 0
    new_roms: int = 0
    identified_roms: int = 0
    scanned_firmware: int = 0
    new_firmware: int = 0

    def __post_init__(self):
        # Lock for thread-safe updates
        self._lock = asyncio.Lock()

    async def update(self, socket_manager: socketio.AsyncRedisManager, **kwargs):
        async with self._lock:
            for key, value in kwargs.items():
                if hasattr(self, key):
                    setattr(self, key, value)

            update_job_meta({"scan_stats": self.to_dict()})
            await socket_manager.emit("scan:update_stats", self.to_dict())

    async def increment(self, socket_manager: socketio.AsyncRedisManager, **kwargs):
        async with self._lock:
            for key, value in kwargs.items():
                if hasattr(self, key):
                    current_value = getattr(self, key)
                    setattr(self, key, current_value + value)

            update_job_meta({"scan_stats": self.to_dict()})
            await socket_manager.emit("scan:update_stats", self.to_dict())

    def to_dict(self) -> dict[str, Any]:
        return {
            "total_platforms": self.total_platforms,
            "total_roms": self.total_roms,
            "scanned_platforms": self.scanned_platforms,
            "new_platforms": self.new_platforms,
            "identified_platforms": self.identified_platforms,
            "scanned_roms": self.scanned_roms,
            "new_roms": self.new_roms,
            "identified_roms": self.identified_roms,
            "scanned_firmware": self.scanned_firmware,
            "new_firmware": self.new_firmware,
        }


def _get_socket_manager() -> socketio.AsyncRedisManager:
    """Connect to external socketio server"""
    return socketio.AsyncRedisManager(REDIS_URL, write_only=True)


async def _identify_firmware(
    platform: Platform,
    fs_fw: str,
) -> int:
    # Break early if the flag is set
    if redis_client.get(STOP_SCAN_FLAG):
        return 0

    firmware = db_firmware_handler.get_firmware_by_filename(platform.id, fs_fw)

    scanned_firmware = await scan_firmware(
        platform=platform,
        file_name=fs_fw,
        firmware=firmware,
    )

    is_verified = Firmware.verify_file_hashes(
        platform_slug=platform.slug,
        file_name=fs_fw,
        file_size_bytes=scanned_firmware.file_size_bytes,
        md5_hash=scanned_firmware.md5_hash,
        sha1_hash=scanned_firmware.sha1_hash,
        crc_hash=scanned_firmware.crc_hash,
    )

    scanned_firmware.missing_from_fs = False
    scanned_firmware.is_verified = is_verified
    db_firmware_handler.add_firmware(scanned_firmware)

    return 1 if not firmware else 0


def should_scan_rom(
    scan_type: ScanType,
    rom: Rom | None,
    roms_ids: list[int],
    metadata_sources: list[str],
) -> bool:
    """Decide if a rom should be scanned or not

    Args:
        scan_type (ScanType): Type of scan to be performed.
        rom (Rom | None): The rom to be scanned.
        roms_ids (list[int]): List of selected roms to be scanned.
        metadata_sources (list[str]): List of metadata sources to be used.
    """

    # When roms_ids is provided, the scan is scoped to those roms only
    if roms_ids:
        return bool(rom and rom.id in roms_ids)

    # This logic is tricky so only touch it if you know what you're doing"""
    should_scan = bool(
        # Any new roms should be scanned
        (scan_type in {ScanType.NEW_PLATFORMS, ScanType.QUICK} and not rom)
        # Complete rescan should scan all roms
        or (scan_type == ScanType.COMPLETE)
        # Hashes rescan should scan all roms to update the hashes
        or (scan_type == ScanType.HASHES)
        or (
            rom
            and (
                # Update scan should scan ROMs identified by the selected metadata sources
                (
                    scan_type == ScanType.UPDATE
                    and rom.is_identified
                    and any(
                        getattr(rom, f"{source}_id", None)
                        for source in metadata_sources
                    )
                )
                # Unmatched scan should scan ROMs that are not identified by the selected metadata sources
                or (
                    scan_type == ScanType.UNMATCHED
                    and any(
                        not getattr(rom, f"{source}_id", None)
                        for source in metadata_sources
                    )
                )
            )
        )
    )

    return should_scan


def _should_get_rom_files(
    scan_type: ScanType,
    rom: Rom,
    newly_added: bool,
    roms_ids: list[int],
) -> bool:
    """Decide if the files of a rom should be rebuilt or not

    Args:
        scan_type (ScanType): Type of scan to be performed.
        rom (Rom): The rom to be rebuilt.
        newly_added (bool): Whether the rom is newly added.
        roms_ids (list[int]): List of selected roms to be scanned.
    """

    return bool(
        newly_added
        or (scan_type == ScanType.COMPLETE)
        or (scan_type == ScanType.HASHES)
        or (rom and rom.id in roms_ids)
    )


# There's an order of operations here that is important:
# 1. Read the list of roms from the filesystem
# 2. Check if ROM should be scanned based on the scan type
# 3. Create a new ROM entry if it doesn't exist
# 4. Build the ROM files and calculate the hashes
# 4. Scan the ROM and update its metadata
async def _identify_rom(
    platform: Platform,
    fs_rom: FSRom,
    rom: Rom | None,
    scan_type: ScanType,
    roms_ids: list[int],
    metadata_sources: list[str],
    launchbox_remote_enabled: bool,
    playmatch_enabled: bool,
    socket_manager: socketio.AsyncRedisManager,
    scan_stats: ScanStats,
) -> None:
    # Break early if the flag is set
    if redis_client.get(STOP_SCAN_FLAG):
        return

    # Update properties that don't require metadata
    parsed_tags = fs_rom_handler.parse_tags(fs_rom["fs_name"])
    roms_path = fs_rom_handler.get_roms_fs_structure(platform.fs_slug)

    # Create the entry early so we have the ID
    newly_added: bool = rom is None
    if not rom:
        try:
            rom = db_rom_handler.add_rom(
                Rom(
                    fs_name=fs_rom["fs_name"],
                    fs_path=roms_path,
                    regions=parsed_tags.regions,
                    revision=parsed_tags.revision,
                    version=parsed_tags.version,
                    languages=parsed_tags.languages,
                    tags=parsed_tags.other_tags,
                    platform_id=platform.id,
                    name=fs_rom_handler.get_file_name_with_no_tags(fs_rom["fs_name"]),
                    url_cover="",
                    url_manual="",
                    url_screenshots=[],
                )
            )
        except IntegrityError:
            # A concurrent scan already created this ROM, so skip it here.
            log.debug(
                f"Skipping {hl(fs_rom['fs_name'])}: already created by a concurrent scan"
            )
            return

    # Build rom files object before scanning
    should_update_files = _should_get_rom_files(
        scan_type=scan_type,
        rom=rom,
        newly_added=newly_added,
        roms_ids=roms_ids,
    )
    if should_update_files:
        # Get hash calculation setting from config
        calculate_hashes = not cm.get_config().SKIP_HASH_CALCULATION
        if calculate_hashes:
            log.debug(f"Calculating file hashes for {rom.fs_name}...")

        parsed_rom_files = await fs_rom_handler.get_rom_files(
            rom, calculate_hashes=calculate_hashes
        )
        fs_rom.update(
            {
                "files": parsed_rom_files.rom_files,
                "crc_hash": parsed_rom_files.crc_hash,
                "md5_hash": parsed_rom_files.md5_hash,
                "sha1_hash": parsed_rom_files.sha1_hash,
                "ra_hash": parsed_rom_files.ra_hash,
            }
        )

    # For a COMPLETE rescan, wipe all downloaded resources before re-fetching so
    # stale files (e.g. a cover from the wrong region) can't be reused. The
    # post-scan download steps below skip downloads when a file already exists or
    # when the source URL is unchanged, so the on-disk files must be removed here.
    if not newly_added and scan_type == ScanType.COMPLETE:
        try:
            await fs_resource_handler.remove_cover(rom)
        except FileNotFoundError:
            pass

        try:
            await fs_resource_handler.remove_manual(rom)
        except FileNotFoundError:
            pass

        try:
            await fs_resource_handler.remove_directory(
                f"{rom.fs_resources_path}/screenshots"
            )
        except FileNotFoundError:
            pass

        for media_type in MetadataMediaType:
            try:
                await fs_resource_handler.remove_media_resources_path(
                    platform.id, rom.id, media_type
                )
            except FileNotFoundError:
                pass

    log.debug(f"Scanning {rom.fs_name}...")
    scanned_rom = await scan_rom(
        scan_type=scan_type,
        platform=platform,
        rom=rom,
        fs_rom=fs_rom,
        metadata_sources=metadata_sources,
        newly_added=newly_added,
        launchbox_remote_enabled=launchbox_remote_enabled,
        playmatch_enabled=playmatch_enabled,
        socket_manager=socket_manager,
    )

    await scan_stats.increment(
        socket_manager=socket_manager,
        scanned_roms=1,
        new_roms=1 if newly_added else 0,
        identified_roms=1 if scanned_rom.is_identified else 0,
    )

    _added_rom = db_rom_handler.add_rom(scanned_rom)

    if _added_rom.is_identified:
        await socket_manager.emit(
            "scan:scanning_rom",
            SimpleRomSchema.from_orm_with_factory(_added_rom).model_dump(
                exclude={
                    "created_at",
                    "updated_at",
                    "rom_user",
                    "last_modified",
                    "files",
                    "sibling_roms",
                }
            ),
        )

    if should_update_files:
        # Delete the existing rom files in the DB
        db_rom_handler.purge_rom_files(_added_rom.id)

        # Create each file entry for the rom
        new_rom_files = [
            RomFile(
                rom_id=_added_rom.id,
                file_name=file.file_name,
                file_path=file.file_path,
                file_size_bytes=file.file_size_bytes,
                last_modified=file.last_modified,
                category=file.category,
                track_meta=_clone_track_meta(file.track_meta, _added_rom.id),
                crc_hash=file.crc_hash,
                md5_hash=file.md5_hash,
                sha1_hash=file.sha1_hash,
                ra_hash=file.ra_hash,
                chd_sha1_hash=file.chd_sha1_hash,
            )
            for file in fs_rom["files"]
        ]
        for new_rom_file in new_rom_files:
            saved = db_rom_handler.add_rom_file(new_rom_file)
            persist_soundtrack_cover(saved, _added_rom)

    # Short circuit if the scan type is hashes
    if scan_type == ScanType.HASHES:
        return

    path_cover_s, path_cover_l = await fs_resource_handler.get_cover(
        entity=_added_rom,
        overwrite=_added_rom.url_cover != rom.url_cover,
        url_cover=add_ss_auth_to_url(_added_rom.url_cover),
    )

    path_manual = await fs_resource_handler.get_manual(
        rom=_added_rom,
        overwrite=_added_rom.url_manual != rom.url_manual,
        url_manual=add_ss_auth_to_url(_added_rom.url_manual),
    )

    screenshots_changed = pydash.xor(
        _added_rom.url_screenshots or [], rom.url_screenshots or []
    )
    url_screenshots = _added_rom.url_screenshots or []
    path_screenshots = await fs_resource_handler.get_rom_screenshots(
        rom=_added_rom,
        overwrite=bool(screenshots_changed),
        url_screenshots=[add_ss_auth_to_url(u) for u in url_screenshots],
    )

    _added_rom.path_cover_s = path_cover_s
    _added_rom.path_cover_l = path_cover_l
    _added_rom.path_screenshots = path_screenshots
    _added_rom.path_manual = path_manual

    # Update the scanned rom with the cover and screenshots paths and update database
    db_rom_handler.update_rom(
        _added_rom.id,
        {
            "path_cover_s": path_cover_s,
            "path_cover_l": path_cover_l,
            "path_screenshots": path_screenshots,
            "path_manual": path_manual,
        },
    )

    # Handle special media files from Screenscraper
    if _added_rom.ss_metadata and MetadataSource.SS in metadata_sources:
        preferred_media_types = get_preferred_media_types()
        for media_type in preferred_media_types:
            media_path = _added_rom.ss_metadata.get(f"{media_type.value}_path")
            media_url = _added_rom.ss_metadata.get(f"{media_type.value}_url")
            if media_path and media_url:
                await fs_resource_handler.store_media_file(
                    add_ss_auth_to_url(media_url),
                    media_path,
                )

    # Handle special media files from ES-DE gamelist.xml
    if _added_rom.gamelist_metadata and MetadataSource.GAMELIST in metadata_sources:
        preferred_media_types = get_preferred_media_types()
        for media_type in preferred_media_types:
            if _added_rom.gamelist_metadata.get(f"{media_type.value}_path"):
                await fs_resource_handler.store_media_file(
                    _added_rom.gamelist_metadata[f"{media_type.value}_url"],
                    _added_rom.gamelist_metadata[f"{media_type.value}_path"],
                )

    # Handle special media files from LaunchBox
    if _added_rom.launchbox_metadata and MetadataSource.LAUNCHBOX in metadata_sources:
        preferred_media_types = get_preferred_media_types()
        for media_type in preferred_media_types:
            if _added_rom.launchbox_metadata.get(f"{media_type.value}_path"):
                await fs_resource_handler.store_media_file(
                    _added_rom.launchbox_metadata[f"{media_type.value}_url"],
                    _added_rom.launchbox_metadata[f"{media_type.value}_path"],
                )

    # Store normal and locked badges
    if _added_rom.ra_metadata and MetadataSource.RA in metadata_sources:
        for ach in _added_rom.ra_metadata.get("achievements", []):
            badge_url_lock = ach.get("badge_url_lock", None)
            badge_path_lock = ach.get("badge_path_lock", None)
            if badge_url_lock and badge_path_lock:
                await fs_resource_handler.store_ra_badge(
                    badge_url_lock, badge_path_lock
                )
            badge_url = ach.get("badge_url", None)
            badge_path = ach.get("badge_path", None)
            if badge_url and badge_path:
                await fs_resource_handler.store_ra_badge(badge_url, badge_path)

    await socket_manager.emit(
        "scan:scanning_rom",
        SimpleRomSchema.from_orm_with_factory(_added_rom).model_dump(
            exclude={
                "created_at",
                "updated_at",
                "rom_user",
                "last_modified",
                "files",
                "sibling_roms",
            }
        ),
    )


async def _identify_platform(
    platform_slug: str,
    scan_type: ScanType,
    fs_platforms: list[str],
    roms_ids: list[int],
    metadata_sources: list[str],
    launchbox_remote_enabled: bool,
    playmatch_enabled: bool,
    socket_manager: socketio.AsyncRedisManager,
    scan_stats: ScanStats,
) -> ScanStats:
    # Stop the scan if the flag is set
    if redis_client.get(STOP_SCAN_FLAG):
        raise ScanStoppedException()

    platform = db_platform_handler.get_platform_by_fs_slug(platform_slug)
    if platform and scan_type == ScanType.NEW_PLATFORMS:
        return scan_stats

    scanned_platform = await scan_platform(platform_slug, fs_platforms)
    if platform:
        scanned_platform.id = platform.id

    await scan_stats.increment(
        socket_manager=socket_manager,
        scanned_platforms=1,
        new_platforms=1 if not platform else 0,
        identified_platforms=1 if scanned_platform.is_identified else 0,
    )

    platform = db_platform_handler.add_platform(scanned_platform)

    # Preparse the platform's gamelist.xml file and cache it
    if MetadataSource.GAMELIST in metadata_sources:
        await meta_gamelist_handler.populate_cache(platform)

    # Scanning firmware
    try:
        fs_firmware = await fs_firmware_handler.get_firmware(platform.fs_slug)
    except FirmwareNotFoundException:
        fs_firmware = []

    if len(fs_firmware) == 0:
        log.warning(
            f"{hl(emoji.EMOJI_WARNING, color=LIGHTYELLOW)} No firmware found for {hl(platform.custom_name or platform.name, color=BLUE)}[{hl(platform.fs_slug)}]"
        )
    else:
        log.info(f"{hl(str(len(fs_firmware)))} firmware files found")

    new_firmware = 0
    for fs_fw in fs_firmware:
        new_firmware += await _identify_firmware(
            platform=platform,
            fs_fw=fs_fw,
        )

    await socket_manager.emit(
        "scan:scanning_platform",
        PlatformSchema.model_validate(platform).model_dump(
            include={
                "id",
                "name",
                "display_name",
                "slug",
                "fs_slug",
                "is_identified",
                "firmware_count",
            }
        ),
    )

    # This reduces the number of socket emissions
    await scan_stats.increment(
        socket_manager=socket_manager,
        scanned_firmware=len(fs_firmware),
        new_firmware=new_firmware,
    )

    try:
        fs_roms = await fs_rom_handler.get_roms(platform)
    except RomsNotFoundException as e:
        log.error(e)
        return scan_stats

    if len(fs_roms) == 0:
        log.warning(
            f"{hl(emoji.EMOJI_WARNING, color=LIGHTYELLOW)} No roms found, verify that the folder structure is correct"
        )
    else:
        log.info(f"{hl(str(len(fs_roms)))} roms found in the file system")

    # Create semaphore to limit concurrent ROM scanning
    scan_semaphore = asyncio.Semaphore(SCAN_WORKERS)

    async def scan_rom_with_semaphore(fs_rom: FSRom, rom: Rom | None) -> None:
        """Scan a single ROM with semaphore limiting"""
        async with scan_semaphore:
            await _identify_rom(
                platform=platform,
                fs_rom=fs_rom,
                rom=rom,
                scan_type=scan_type,
                roms_ids=roms_ids,
                metadata_sources=metadata_sources,
                launchbox_remote_enabled=launchbox_remote_enabled,
                playmatch_enabled=playmatch_enabled,
                socket_manager=socket_manager,
                scan_stats=scan_stats,
            )

    for fs_roms_batch in batched(fs_roms, 200, strict=False):
        roms_by_fs_name = db_rom_handler.get_roms_by_fs_name(
            platform_id=platform.id,
            fs_names={fs_rom["fs_name"] for fs_rom in fs_roms_batch},
        )

        # Separate skipped ROMs from those that need scanning
        skipped_rom_ids: list[int] = []
        roms_to_scan: list[tuple[FSRom, Rom | None]] = []

        for fs_rom in fs_roms_batch:
            rom = roms_by_fs_name.get(fs_rom["fs_name"])
            if should_scan_rom(
                scan_type=scan_type,
                rom=rom,
                roms_ids=roms_ids,
                metadata_sources=metadata_sources,
            ):
                roms_to_scan.append((fs_rom, rom))
            elif rom:
                skipped_rom_ids.append(rom.id)

        # Bulk update all skipped ROMs in one query instead of per-ROM updates
        if skipped_rom_ids:
            db_rom_handler.bulk_mark_present(platform.id, skipped_rom_ids)
            await scan_stats.increment(
                socket_manager=socket_manager,
                scanned_roms=len(skipped_rom_ids),
            )

        # Process only ROMs that actually need scanning
        scan_tasks = [
            scan_rom_with_semaphore(fs_rom=fs_rom, rom=rom)
            for fs_rom, rom in roms_to_scan
        ]

        if scan_tasks:
            batched_results = await asyncio.gather(*scan_tasks, return_exceptions=True)
            for result, (fs_rom, _) in zip(batched_results, roms_to_scan, strict=False):
                if isinstance(result, Exception):
                    log.error(f"Error scanning ROM {fs_rom['fs_name']}: {result}")

    missing_roms = db_rom_handler.mark_missing_roms(
        platform.id, [rom["fs_name"] for rom in fs_roms]
    )
    if len(missing_roms) > 0:
        log.warning(f"{hl('Missing')} roms from filesystem:")
        for r in missing_roms:
            log.warning(f" - {r.fs_name}")

    missing_firmware = db_firmware_handler.mark_missing_firmware(
        platform.id, [fw for fw in fs_firmware]
    )
    if len(missing_firmware) > 0:
        log.warning(f"{hl('Missing')} firmware from filesystem:")
        for f in missing_firmware:
            log.warning(f" - {f}")

    return scan_stats


@initialize_context()
async def scan_platforms(
    platform_ids: list[int],
    metadata_sources: list[str],
    scan_type: ScanType = ScanType.QUICK,
    roms_ids: list[int] | None = None,
    launchbox_remote_enabled: bool = True,
    playmatch_enabled: bool = True,
    platform_fs_slugs: list[str] | None = None,
) -> ScanStats:
    """Scan all the listed platforms and fetch metadata from different sources

    Args:
        platform_ids (list[int]): List of platform ids to be scanned
        metadata_sources (list[str]): List of metadata sources to be used
        scan_type (ScanType): Type of scan to be performed.
        roms_ids (list[int], optional): List of selected roms to be scanned.
        platform_fs_slugs (list[str], optional): Filesystem slugs of folders to
            scan that have no database row yet (never-scanned platforms).
    """
    if not roms_ids:
        roms_ids = []

    if not platform_fs_slugs:
        platform_fs_slugs = []

    socket_manager = _get_socket_manager()
    scan_stats = ScanStats()

    # Reset the ScreenScraper daily-quota breaker so this scan re-evaluates the
    # quota instead of inheriting a tripped state from a previous scan.
    reset_ss_daily_quota()

    try:
        fs_platforms: list[str] = await fs_platform_handler.get_platforms()
    except FolderStructureNotMatchException as e:
        log.error(e)
        await socket_manager.emit("scan:done_ko", e.message)
        return scan_stats

    # Clear the gamelist cache to ensure we're using fresh gamelist.xml data
    meta_gamelist_handler.clear_cache()

    # Initialize HLTB handler (fetches current search endpoint and security token)
    if MetadataSource.HLTB in metadata_sources:
        meta_hltb_handler.initialize()

    # Resolve the platforms that will actually be scanned. When no platform ids
    # are provided, every filesystem platform is scanned.
    db_platforms = db_platform_handler.get_platforms()
    db_platforms_by_slug = {p.fs_slug: p for p in db_platforms}

    # Selected platforms arrive as database ids (existing platforms) and/or
    # filesystem slugs (a folder can be an existing platform or one on disk
    # without a database row yet). Both resolve to a folder slug the scanner
    # walks. Known database platforms are always accepted; a bare slug is only
    # accepted when it maps to a folder on disk. When nothing is selected,
    # every filesystem platform is scanned.
    selected_slugs = [p.fs_slug for p in db_platforms if p.id in platform_ids]
    for fs_slug in platform_fs_slugs:
        if fs_slug in selected_slugs:
            continue
        if fs_slug in db_platforms_by_slug or fs_slug in fs_platforms:
            selected_slugs.append(fs_slug)

    platform_list = sorted(selected_slugs or fs_platforms)

    # A "new platforms" scan skips platforms that already exist in the database,
    # so they must be excluded from the totals to keep the tracker accurate. This
    # mirrors the existence check done per-platform in _identify_platform, reusing
    # the platforms already fetched above instead of querying again per platform.
    platforms_to_scan = platform_list
    if scan_type == ScanType.NEW_PLATFORMS:
        platforms_to_scan = [
            platform_slug
            for platform_slug in platform_list
            if db_platforms_by_slug.get(platform_slug) is None
        ]

    total_roms = 0
    for platform_slug in platforms_to_scan:
        try:
            total_roms += await fs_rom_handler.count_roms(
                Platform(fs_slug=platform_slug)
            )
        except RomsNotFoundException as e:
            log.error(e)

    await scan_stats.update(
        socket_manager=socket_manager,
        total_platforms=len(platforms_to_scan),
        total_roms=total_roms,
    )

    async def stop_scan():
        log.info(f"{emoji.EMOJI_STOP_SIGN} Scan stopped manually")
        await socket_manager.emit("scan:done", scan_stats.to_dict())
        redis_client.delete(STOP_SCAN_FLAG)

    try:
        if len(platform_list) == 0:
            log.warning(
                f"{hl(emoji.EMOJI_WARNING, color=LIGHTYELLOW)} No platforms found, verify that the folder structure is right and the volume is mounted correctly."
                f"{FOLDER_STRUCT_MSG}"
            )
        else:
            log.info(
                f"Found {hl(str(len(platform_list)))} platforms in the file system"
            )

        for platform_slug in platform_list:
            scan_stats = await _identify_platform(
                platform_slug=platform_slug,
                scan_type=scan_type,
                fs_platforms=fs_platforms,
                roms_ids=roms_ids,
                metadata_sources=metadata_sources,
                launchbox_remote_enabled=launchbox_remote_enabled,
                playmatch_enabled=playmatch_enabled,
                socket_manager=socket_manager,
                scan_stats=scan_stats,
            )

        missed_platforms = db_platform_handler.mark_missing_platforms(fs_platforms)
        if len(missed_platforms) > 0:
            log.warning(f"{hl('Missing')} platforms from filesystem:")
            for p in missed_platforms:
                log.warning(f" - {p.slug} ({p.fs_slug})")

        log.info(f"{emoji.EMOJI_CHECK_MARK} Scan completed")

        # The library changed; drop cached filter values.
        db_rom_handler.invalidate_filter_values_cache()

        # Export metadata files if enabled in config
        config = cm.get_config()

        # Update the list of platforms after the scan to ensure we have the latest data
        db_platforms = db_platform_handler.get_platforms()
        db_platforms_by_slug = {p.fs_slug: p for p in db_platforms}

        if config.GAMELIST_AUTO_EXPORT_ON_SCAN:
            log.info("Auto-exporting gamelist.xml for all platforms...")
            gamelist_exporter = GamelistExporter(local_export=True)
            for platform_slug in platform_list:
                platform = db_platforms_by_slug.get(platform_slug)
                if platform:
                    export_success = await gamelist_exporter.export_platform_to_file(
                        platform.id,
                        request=None,
                    )
                    if export_success:
                        log.info(
                            f"Auto-exported gamelist.xml for platform {platform.name} after scan"
                        )
                    else:
                        log.warning(
                            f"Failed to auto-export gamelist.xml for platform {platform.name} after scan"
                        )
            log.info("Gamelist.xml auto-export completed.")

        if config.PEGASUS_AUTO_EXPORT_ON_SCAN:
            log.info("Auto-exporting metadata.pegasus.txt for all platforms...")
            pegasus_exporter = PegasusExporter(local_export=True)
            for platform_slug in platform_list:
                platform = db_platforms_by_slug.get(platform_slug)
                if platform:
                    export_success = await pegasus_exporter.export_platform_to_file(
                        platform.id,
                        request=None,
                    )
                    if export_success:
                        log.info(
                            f"Auto-exported metadata.pegasus.txt for platform {platform.name} after scan"
                        )
                    else:
                        log.warning(
                            f"Failed to auto-export metadata.pegasus.txt for platform {platform.name} after scan"
                        )
            log.info("Pegasus metadata auto-export completed.")

        await socket_manager.emit("scan:done", scan_stats.to_dict())
    except ScanStoppedException:
        await stop_scan()
    except Exception as e:
        log.error(f"Error in scan_platform: {e}")
        # Catch all exceptions and emit error to the client
        await socket_manager.emit("scan:done_ko", str(e))
        # Re-raise the exception to be caught by the error handler
        raise e

    return scan_stats


async def reject_unauthorized_scan(sid: str) -> bool:
    """Return ``True`` (and notify the caller) if the socket may not run scans.

    Scans are a privileged, destructive operation, so gate them on the same
    ``TASKS_RUN`` scope the REST task endpoints require, resolved from the
    server-side session (never from the client payload).
    """
    user = await get_authenticated_user(sid)
    if user is not None and Scope.TASKS_RUN in user.oauth_scopes:
        return False

    log.warning(f"{emoji.EMOJI_STOP_SIGN} Unauthorized scan request rejected")
    await socket_handler.socket_server.emit(
        "scan:done_ko",
        "You are not authorized to run scans",
        to=sid,
    )
    return True


@socket_handler.socket_server.on("scan")  # type: ignore
async def scan_handler(sid: str, options: dict[str, Any]):
    """Scan socket endpoint

    Args:
        options (dict): Socket options
    """

    if await reject_unauthorized_scan(sid):
        return

    log.info(f"{emoji.EMOJI_MAGNIFYING_GLASS_TILTED_RIGHT} Scanning")

    platform_ids = options.get("platforms", [])
    platform_fs_slugs = options.get("platform_fs_slugs", [])
    scan_type = ScanType[options.get("type", "quick").upper()]
    roms_ids = options.get("roms_ids", [])
    metadata_sources = options.get("apis", [])
    launchbox_remote_enabled = bool(options.get("launchbox_remote_enabled", True))
    playmatch_enabled = bool(options.get("playmatch_enabled", True))

    if DEV_MODE:
        return await scan_platforms(
            platform_ids=platform_ids,
            metadata_sources=metadata_sources,
            scan_type=scan_type,
            roms_ids=roms_ids,
            launchbox_remote_enabled=launchbox_remote_enabled,
            playmatch_enabled=playmatch_enabled,
            platform_fs_slugs=platform_fs_slugs,
        )

    return high_prio_queue.enqueue(
        scan_platforms,
        platform_ids=platform_ids,
        metadata_sources=metadata_sources,
        scan_type=scan_type,
        roms_ids=roms_ids,
        launchbox_remote_enabled=launchbox_remote_enabled,
        playmatch_enabled=playmatch_enabled,
        platform_fs_slugs=platform_fs_slugs,
        job_timeout=SCAN_TIMEOUT,  # Timeout (default of 4 hours)
        result_ttl=TASK_RESULT_TTL,
        meta={
            "task_name": f"{scan_type.value.capitalize()} Scan",
            "task_type": TaskType.SCAN,
        },
    )


@socket_handler.socket_server.on("scan:stop")  # type: ignore
async def stop_scan_handler(sid: str):
    """Stop scan socket endpoint"""

    if await reject_unauthorized_scan(sid):
        return

    log.info(f"{emoji.EMOJI_STOP_BUTTON} Stop scan requested...")

    async def cancel_job(job: Job):
        job.cancel()
        redis_client.set(STOP_SCAN_FLAG, 1)
        log.info(f"{emoji.EMOJI_STOP_BUTTON} Job found, stopping scan...")

    existing_jobs = high_prio_queue.get_jobs()
    for job in existing_jobs:
        if get_job_func_name(job) == "scan_platform" and job.is_started:
            return await cancel_job(job)

    workers = Worker.all(connection=redis_client)
    for worker in workers:
        current_job = worker.get_current_job()
        if (
            current_job
            and get_job_func_name(current_job)
            == "endpoints.sockets.scan.scan_platforms"
            and current_job.is_started
        ):
            return await cancel_job(current_job)

    log.info(f"{emoji.EMOJI_STOP_BUTTON} No running scan to stop")