Bugfix/dbc22 6256

src/backend/apps/consumer/processor.py

@@ -58,7 +58,6 @@
 S3_REGION = os.getenv("S3_REGION")
 S3_ACCESS_KEY = os.getenv("S3_ACCESS_KEY")
 S3_SECRET_KEY = os.getenv("S3_SECRET_KEY")
-RABBITMQ_URL = os.getenv("RABBITMQ_URL")
 S3_ENDPOINT_URL = os.getenv("S3_ENDPOINT_URL")
 QUEUE_NAME = os.getenv("RABBITMQ_QUEUE_NAME")
 QUEUE_MAX_BYTES = int(os.getenv("RABBITMQ_QUEUE_MAX_BYTES", "209715200"))
@@ -69,9 +68,6 @@
 RABBITMQ_TIMEOUT = int(os.getenv("RABBITMQ_TIMEOUT", "30")) 
 RABBITMQ_RECONNECT_INTERVAL = int(os.getenv("RABBITMQ_RECONNECT_INTERVAL", "5"))
 
-
-#boto3.set_stream_logger('botocore', logging.DEBUG)
-
 # S3 client configuration
 config = Config(
     signature_version='s3v4',
@@ -101,9 +97,6 @@
 last_camera_refresh = {}
 image_invalid = False
 
-last_activity = 0.0  # Track last message processing time
-health_check_interval = 60  # seconds
-max_idle_time = 300  # 5 minutes before warning
 
 tz_pst = 'America/Vancouver'
 
@@ -119,135 +112,153 @@ async def on_channel_close(ch, exc=None):
     logger.warning(f"RabbitMQ channel closed: {exc}")
 
 
-async def on_message(message: aio_pika.IncomingMessage):
-    """Process incoming RabbitMQ message."""
-    if stop_event.is_set():
-        logger.info("Stop requested. Dropping message.")
-        return
-
-    async with message.process(ignore_processed=True):
-        filename = message.headers.get("filename", "unknown.jpg")
-        camera_id = filename.split("_")[0].split(".")[0]
-        timestamp_utc = message.headers.get("timestamp", "unknown")
-        camera_status = calculate_camera_status(camera_id, timestamp_utc)
-
-        try:
-            timestamp_local = await sync_to_async(safe_db_call)(
-                generate_local_timestamp, db_data, camera_id, timestamp_utc
-            )
-            try:
-                await asyncio.wait_for(
-                    handle_image_message(camera_id, message.body, timestamp_local, camera_status),
-                    timeout=120
-                )
-            except asyncio.TimeoutError:
-                logger.error(f"Processing timeout for camera {camera_id}")
-
-            logger.info("Processed message for camera %s.", camera_id)
-        except Exception as e:
-            logger.exception("Failed processing message (camera %s): %s", camera_id, e)
-            await asyncio.sleep(2)
+async def setup_rabbitmq(rb_url: str, name: str):
+    connection = await aio_pika.connect_robust(
+        rb_url,
+        heartbeat=RABBITMQ_HEARTBEAT,
+        timeout=RABBITMQ_TIMEOUT,
+        reconnect_interval=RABBITMQ_RECONNECT_INTERVAL,
+        fail_fast=False,
+    )
+    logger.info(f"RabbitMQ connection established for {name}.")
+    connection.reconnect_callbacks.add(on_reconnect)
+    connection.close_callbacks.add(on_close)
+
+    channel = await connection.channel()
+    logger.info(f"RabbitMQ channel created for {name}.")
+    channel.close_callbacks.add(on_channel_close)
+
+    exchange = await channel.declare_exchange(
+        EXCHANGE_NAME,
+        type=aio_pika.ExchangeType.FANOUT,
+        durable=True,
+    )
 
-async def wait_for_timeout_or_signal(timeout_seconds: int):
-    """Wait for timeout or stop signal."""
-    try:
-        await asyncio.wait_for(stop_event.wait(), timeout=timeout_seconds)
-        return True  # Stop signal received
-    except asyncio.TimeoutError:
-        return False  # Timeout reached
+    queue = await channel.declare_queue(
+        QUEUE_NAME,
+        durable=True,
+        exclusive=False,
+        auto_delete=False,
+        arguments={
+            "x-max-length-bytes": QUEUE_MAX_BYTES,
+            "x-queue-type": "quorum"
+        },
+    )
 
-async def run_consumer():
-    """Long-running RabbitMQ consumer with automatic reconnection."""
-
+    await queue.bind(exchange)
+
+    return connection, queue
+
+async def consume_queue(queue, name: str):
+    async with queue.iterator() as queue_iter:
+        async for message in queue_iter:
+            if stop_event.is_set():
+                logger.info(f"Stop requested. Breaking consume loop for {name}.")
+                break
+
+            try:
+                await process_message(message)
+            except Exception as e:
+                logger.error(f"Error processing message from {name}: {e}")
+
+async def consume_from(rb_url: str, name: str):
     while not stop_event.is_set():
         connection = None
-        channel = None
-        queue = None
-
+
         try:
-            # 1. Connect to RabbitMQ (auto-reconnects on RabbitMQ restart)
-            connection = await aio_pika.connect_robust(
-                RABBITMQ_URL,
-                heartbeat=RABBITMQ_HEARTBEAT,
-                timeout=RABBITMQ_TIMEOUT,
-                reconnect_interval=RABBITMQ_RECONNECT_INTERVAL,
-                fail_fast=False,
-            )
-            logger.info("RabbitMQ connection established.")
-            connection.reconnect_callbacks.add(on_reconnect)
-            connection.close_callbacks.add(on_close)
-
-            # 2. Setup channel and queue
-            channel = await connection.channel()
-            logger.info("RabbitMQ channel created.")
-            channel.close_callbacks.add(on_channel_close)
-
-            exchange = await channel.declare_exchange(
-                EXCHANGE_NAME,
-                type=aio_pika.ExchangeType.FANOUT,
-                durable=True,
-            )
-            logger.info(f"RabbitMQ exchange '{EXCHANGE_NAME}' declared.")
-
-            # # Delete existing queue if wrong type, then recreate as quorum
-            # try:
-            #     await channel.queue_delete(QUEUE_NAME)
-            #     logger.info(f"Deleted existing queue '{QUEUE_NAME}' for recreation")
-            # except Exception:
-            #     pass
-
-            queue = await channel.declare_queue(
-                QUEUE_NAME,
-                durable=True,
-                exclusive=False,
-                auto_delete=False,
-                arguments={
-                    "x-max-length-bytes": QUEUE_MAX_BYTES,
-                    "x-queue-type": "quorum"
-                },
-            )
-            logger.info(f"RabbitMQ queue '{QUEUE_NAME}' declared.")
-
-            await queue.bind(exchange)
-            logger.info(f"RabbitMQ queue '{QUEUE_NAME}' bound to exchange '{EXCHANGE_NAME}'.")
-
-            logger.info("Starting message consumption...")
-
-            # 3. Consume messages using iterator (blocks waiting for messages)
-            async with queue.iterator() as queue_iter:
-                async for message in queue_iter:
-                    if stop_event.is_set():
-                        logger.info("Stop requested. Breaking consume loop.")
-                        break
-
-                    try:
-                        await process_message(message)
-                    except Exception as e:
-                        logger.error(f"Error processing message: {e}")
-                        # Continue to next message (processing errors don't trigger reconnect)
-                        continue
-
+            connection, queue = await setup_rabbitmq(rb_url, name)
+
+            logger.info(f"Starting message consumption from {name}...")
+            await consume_queue(queue, name)
+
         except asyncio.CancelledError:
-            logger.info("Consumer cancelled; shutting down.")
+            logger.info(f"Consumer cancelled from {name}; shutting down.")
             raise
-        except (aio_pika.exceptions.AMQPConnectionError,
-                aio_pika.exceptions.ConnectionClosed,
-                aio_pika.exceptions.ChannelClosed,
-                ChannelInvalidStateError) as e:
-            logger.warning(f"RabbitMQ connection/channel lost: {e}. Reconnecting...")
+
+        except (
+            aio_pika.exceptions.AMQPConnectionError,
+            aio_pika.exceptions.ConnectionClosed,
+            aio_pika.exceptions.ChannelClosed,
+            ChannelInvalidStateError,
+        ) as e:
+            logger.warning(f"RabbitMQ connection/channel lost from {name}: {e}. Reconnecting...")
+
         except Exception as e:
-            logger.exception(f"Unexpected error: {e}. Reconnecting...")
+            logger.exception(f"Unexpected error from {name}: {e}. Reconnecting...")
+
         finally:
             logger.info("Cleaning up RabbitMQ resources...")
             if connection and not connection.is_closed:
                 await connection.close()
-        
+
         if not stop_event.is_set():
             logger.info(f"Waiting {RABBITMQ_RECONNECT_INTERVAL}s before reconnecting...")
             await asyncio.sleep(RABBITMQ_RECONNECT_INTERVAL)
-    
+
     logger.info("Consumer stopped.")
-
+
+
+
+async def parse_rows(rb_url: str):
+    """
+    Loads camera rows from DB and prepares DB metadata for this RabbitMQ URL.
+    Raises an exception if no data is available.
+    """
+    logger.info("Fetching DB camera rows for %s", rb_url)
+
+    rows = await sync_to_async(get_all_from_db)()
+    db_data = process_camera_rows(rows)
+
+    if not db_data:
+        raise RuntimeError(
+            f"No camera data available for watermarking. Consumer ({rb_url}) exiting."
+        )
+
+    logger.info("DB camera data loaded for %s (%d rows).", rb_url, len(db_data))
+    return db_data
+
+
+async def run_consumer():
+    """
+    Launch consumers for Gold and GoldDR in parallel.
+    Each consumer listens to its own RabbitMQ instance.
+    """
+    gold_url = os.getenv("RABBITMQ_URL_GOLD")
+    golddr_url = os.getenv("RABBITMQ_URL_GOLDDR")
+
+    if not gold_url and not golddr_url:
+        raise RuntimeError("No RabbitMQ URLs configured. At least one is required.")
+
+    tasks = []
+
+    if gold_url:
+        logger.info("Starting GOLD consumer...")
+        tasks.append(asyncio.create_task(consume_from(gold_url, "GOLD")))
+        # pass
+
+    if golddr_url:
+        logger.info("Starting GOLDDR consumer...")
+        tasks.append(asyncio.create_task(consume_from(golddr_url, "GOLDDR")))
+        # pass
+
+    logger.info("All configured RabbitMQ consumers started.")
+
+    # Wait until stop_event is set (SIGTERM / SIGINT)
+    try:
+        await stop_event.wait()
+    finally:
+        logger.info("Stop event received; cancelling consumers...")
+        for task in tasks:
+            task.cancel()
+
+        await asyncio.gather(*tasks, return_exceptions=True)
+        logger.info("All consumers stopped.")
+
+def shutdown():
+    """Signal handler to gracefully stop the consumer."""
+    logger.info("Received shutdown signal.")
+    stop_event.set()
+
 async def process_message(message: aio_pika.IncomingMessage):
     """Process a single message from the queue."""
     async with message.process(ignore_processed=True):
@@ -277,10 +288,6 @@ def safe_db_call(func, *args, **kwargs):
     connection.ensure_connection()
     return func(*args, **kwargs)
 
-def shutdown():
-    """Signal handler to gracefully stop the consumer."""
-    logger.info("Received shutdown signal.")
-    stop_event.set()
 
 def process_camera_rows(rows):
     if not rows:
@@ -554,7 +561,7 @@ def update_webcam(camera_id, timestamp, webcam):
     camera_status = calculate_camera_status(camera_id, time_now_utc)
 
 
-    camera, created = Webcam.objects.update_or_create(
+    camera, _ = Webcam.objects.update_or_create(
         id=camera_id,
         defaults={
             "name": f"{webcam.get('cam_internet_name', '')}",