feat: low queue for mass updates (#1526)

When there is a massive update on the Open Food Facts server, put them in lower priority queues, to keep user submissions in high priorities, and be able to ask them question quickly.

fixes: #1519

---------

Co-authored-by: prasanna7codes <prasannalit098@gmail.com>

Author: alexgarel

robotoff/workers/queues.py

@@ -56,6 +56,11 @@ def get_high_queue(product_id: Optional[ProductIdentifier] = None) -> Queue:
     return high_queues[queue_idx]
 
 
+def get_low_queue() -> Queue:
+    """Return the low-priority queue."""
+    return low_queue
+
+
 def enqueue_in_job(
     func: Callable,
     queue: Queue,

robotoff/workers/update_listener.py

@@ -7,7 +7,12 @@
 from robotoff import settings
 from robotoff.types import ProductIdentifier, ServerType
 from robotoff.utils.logger import get_logger
-from robotoff.workers.queues import enqueue_in_job, enqueue_job, get_high_queue
+from robotoff.workers.queues import (
+    enqueue_in_job,
+    enqueue_job,
+    get_high_queue,
+    get_low_queue,
+)
 from robotoff.workers.tasks import delete_product_insights_job
 from robotoff.workers.tasks.import_image import run_import_image_job
 from robotoff.workers.tasks.product_updated import update_insights_job
@@ -16,8 +21,7 @@
 
 
 def get_redis_client():
-    """Get the Redis client where Product Opener publishes it's product
-    update."""
+    """Get the Redis client where Product Opener publishes its product updates."""
     return Redis(
         host=settings.REDIS_UPDATE_HOST,
         port=settings.REDIS_UPDATE_PORT,
@@ -36,11 +40,22 @@ def process_redis_update(self, redis_update: RedisUpdate):
         action = redis_update.action
         server_type = ServerType.from_product_type(redis_update.product_type)
         product_id = ProductIdentifier(redis_update.code, server_type)
+
+        # Check if the update was triggered by scanbot or specific mass update accounts
+        is_scanbot_or_mass_update = redis_update.user_id in [
+            "scanbot",
+            "update_all_products",
+        ]
+        # Select queue based on triggering actor
+        selected_queue = (
+            get_low_queue() if is_scanbot_or_mass_update else get_high_queue(product_id)
+        )
+
         if action == "deleted":
             logger.info("Product %s has been deleted", redis_update.code)
             enqueue_job(
                 delete_product_insights_job,
-                get_high_queue(product_id),
+                selected_queue,
                 job_kwargs={"result_ttl": 0},
                 product_id=product_id,
             )
@@ -69,7 +84,7 @@ def process_redis_update(self, redis_update: RedisUpdate):
                 )
                 enqueue_job(
                     run_import_image_job,
-                    get_high_queue(product_id),
+                    selected_queue,
                     job_kwargs={"result_ttl": 0},
                     product_id=product_id,
                     image_url=image_url,
@@ -79,7 +94,7 @@ def process_redis_update(self, redis_update: RedisUpdate):
                 logger.info("Product %s has been updated", redis_update.code)
                 enqueue_in_job(
                     update_insights_job,
-                    get_high_queue(product_id),
+                    selected_queue,
                     settings.UPDATED_PRODUCT_WAIT,
                     job_kwargs={"result_ttl": 0},
                     product_id=product_id,
@@ -91,7 +106,7 @@ def run_update_listener():
     """Run the update import daemon.
 
     This daemon listens to the Redis stream containing information about
-    product updates, and triggers
+    product updates and triggers appropriate actions.
     """
     redis_client = get_redis_client()
     update_listener = UpdateListener(