Add watchdog and watch roms

backend/config/__init__.py

@@ -70,7 +70,7 @@
     os.environ.get("ENABLE_RESCAN_ON_FILESYSTEM_CHANGE", "false") == "true"
 )
 RESCAN_ON_FILESYSTEM_CHANGE_DELAY: Final = int(
-    os.environ.get("RESCAN_ON_FILESYSTEM_CHANGE_DELAY", 5)  # 5 seconds
+    os.environ.get("RESCAN_ON_FILESYSTEM_CHANGE_DELAY", 5)  # 5 minutes
 )
 ENABLE_SCHEDULED_RESCAN: Final = (
     os.environ.get("ENABLE_SCHEDULED_RESCAN", "false") == "true"

backend/endpoints/scan.py

@@ -13,7 +13,9 @@
 
 
 async def scan_platforms(
-    platform_slugs: list[str], complete_rescan: bool, selected_roms: list[str]
+    platform_slugs: list[str],
+    complete_rescan: bool = False,
+    selected_roms: list[str] = (),
 ):
     # Connect to external socketio server
     sm = (

backend/tasks/scan_library.py

@@ -4,6 +4,8 @@
     ENABLE_SCHEDULED_RESCAN,
     SCHEDULED_RESCAN_CRON,
 )
+from endpoints.scan import scan_platforms
+from utils.redis import low_prio_queue
 from .exceptions import SchedulerException
 from . import tasks_scheduler
 
@@ -17,16 +19,14 @@ def _get_existing_job():
     return None
 
 
-async def run():
+def run():
     if not ENABLE_SCHEDULED_RESCAN:
         log.info("Scheduled library scan not enabled, unscheduling...")
         unschedule()
         return
 
-    from endpoints.scan import scan_platforms
-
     log.info("Scheduled library scan started...")
-    await scan_platforms("", False)
+    low_prio_queue.enqueue(scan_platforms, platform_slugs=[])
     log.info("Scheduled library scan done.")
 
 

backend/watcher.py

@@ -0,0 +1,87 @@
+import os
+from datetime import timedelta
+from watchdog.observers import Observer
+from watchdog.events import FileSystemEventHandler
+
+from endpoints.scan import scan_platforms
+from utils.redis import low_prio_queue
+from logger.logger import log
+
+from config import (
+    HIGH_PRIO_STRUCTURE_PATH,
+    LIBRARY_BASE_PATH,
+    ENABLE_RESCAN_ON_FILESYSTEM_CHANGE,
+    RESCAN_ON_FILESYSTEM_CHANGE_DELAY,
+)
+
+path = (
+    HIGH_PRIO_STRUCTURE_PATH
+    if os.path.exists(HIGH_PRIO_STRUCTURE_PATH)
+    else LIBRARY_BASE_PATH
+)
+
+
+class EventHandler(FileSystemEventHandler):
+    def on_any_event(self, event):
+        if not ENABLE_RESCAN_ON_FILESYSTEM_CHANGE:
+            return
+
+        # Ignore .DS_Store files
+        if event.src_path.endswith(".DS_Store"):
+            return
+
+        # Ignore modified events
+        if event.event_type == "modified":
+            return
+
+        event_src = event.src_path.split(path)[-1]
+        platform_slug = event_src.split("/")[1]
+        time_delta = timedelta(minutes=RESCAN_ON_FILESYSTEM_CHANGE_DELAY)
+
+        log.info(f"Filesystem event: {event.event_type} {event_src}")
+
+        low_prio_queue.scheduled_job_registry.remove_jobs()
+
+        # Skip if a scan is already scheduled
+        for job_id in low_prio_queue.scheduled_job_registry.get_job_ids():
+            job = low_prio_queue.fetch_job(job_id)
+            if (
+                job
+                and job.is_scheduled
+                and job.func_name == "endpoints.scan.scan_platforms"
+            ):
+                if job.args[0] == []:
+                    log.info("Full rescan already scheduled")
+                    return
+
+                if platform_slug in job.args[0]:
+                    log.info(f"Scan already scheduled for {platform_slug}")
+                    return
+
+        rescan_in_msg = f"rescanning in {RESCAN_ON_FILESYSTEM_CHANGE_DELAY} minutes."
+
+        # # Any change to a platform directory should trigger a full rescan
+        if event.is_directory and event_src.count("/") == 1:
+            log.info(f"Platform directory changed, {rescan_in_msg}")
+            return low_prio_queue.enqueue_in(time_delta, scan_platforms, [])
+
+        # Otherwise trigger a rescan for the specific platform
+        log.info(f"Change detected in {platform_slug} folder, {rescan_in_msg}")
+        return low_prio_queue.enqueue_in(
+            time_delta,
+            scan_platforms,
+            [platform_slug],
+        )
+
+
+if __name__ == "__main__":
+    observer = Observer()
+    observer.schedule(EventHandler(), path, recursive=True)
+    observer.start()
+
+    try:
+        while observer.is_alive():
+            observer.join(1)
+    finally:
+        observer.stop()
+        observer.join()

docker/init_scripts/init

@@ -12,6 +12,9 @@
 # Start rq scheduler
 /init_scheduler &
 
+# Start watchdog
+/init_watchdog &
+
 # Wait for any process to exit
 wait -n
 

docker/init_scripts/init_scheduler

@@ -1,4 +1,4 @@
 #!/bin/bash
 
 cd /back
-[[ ${ENABLE_EXPERIMENTAL_REDIS} == "true" ]] && rqscheduler --host ${REDIS_HOST} --port ${REDIS_PORT} || sleep infinity
+[[ ${ENABLE_EXPERIMENTAL_REDIS} == "true" ]] && python3 scheduler.py || sleep infinity

docker/init_scripts/init_watchdog

@@ -0,0 +1,4 @@
+#!/bin/bash
+
+cd /back
+python3 watcher.py

docker/init_scripts/init_worker

@@ -1,4 +1,4 @@
 #!/bin/bash
 
 cd /back
-[[ ${ENABLE_EXPERIMENTAL_REDIS} == "true" ]] && rq worker high default low --url redis://${REDIS_HOST}:${REDIS_PORT}/0 || sleep infinity
+[[ ${ENABLE_EXPERIMENTAL_REDIS} == "true" ]] && python3 worker.py || sleep infinity

pyproject.toml

@@ -44,6 +44,7 @@ httpx = "^0.24.1"
 python-multipart = "^0.0.6"
 types-python-jose = "^3.3.4.8"
 types-passlib = "^1.7.7.13"
+watchdog = "^3.0.0"
 
 [build-system]
 requires = ["poetry-core"]