src/main.py

@@ -16,7 +16,6 @@ from uvicorn_logger_config import LOGGING_CONFIG
 from secret_manager import cleanup_secret_files
 
 logger = setup_logger(__name__)
-producer = RabbitMQProducer()
 api_key_header_internal = APIKeyHeader(name="X-API-Key-Internal")
 
 class Notification(BaseModel):
@@ -34,6 +33,9 @@ async def lifespan(app: FastAPI):
 	start_healthcheck_thread()
 	logger.info("MySQL healthcheck thread started.")
 
+	producer = RabbitMQProducer()
+
+	try:
 		logger.info("Starting RabbitMQ producer...")
 		await producer.connect()
 		app.state.rmq_producer = producer
@@ -48,6 +50,7 @@ async def lifespan(app: FastAPI):
 		cleanup_secret_files(SECRET_PATHS)
 
 		yield
+	finally:
 		logger.info("Closing RabbitMQ producer...")
 		await producer.close()
 

src/rabbitmq_handler.py

@@ -5,45 +5,46 @@ from secret_manager import return_credentials
 import os
 from simple_logger_handler import setup_logger
 import json
+from urllib.parse import quote
 
 logger = setup_logger(__name__)
 
-rmq_username = return_credentials("/etc/secrets/rmq_username")
-rmq_password = return_credentials("/etc/secrets/rmq_password")
-rmq_host = os.getenv("BACKEND_API_INTERNAL_RMQ_HOST", "localhost")
-rmq_vhost = os.getenv("BACKEND_API_INTERNAL_RMQ_VHOST", "app_notifications")
-rmq_exchange = os.getenv("BACKEND_API_INTERNAL_RMQ_EXCHANGE", "app_notifications")
-
-RABBITMQ_URL = f"amqp://{rmq_username}:{rmq_password}@{rmq_host}/{rmq_vhost}"
-
-
 class RabbitMQProducer:
-	def __init__(self, url=RABBITMQ_URL, exchange_name=rmq_exchange,rmq_host=rmq_host):
-		self.url = url
-		self.rmq_host = rmq_host
-		self.exchange_name = exchange_name
+	def __init__(self):
+		self.rmq_host = os.getenv("BACKEND_API_INTERNAL_RMQ_HOST", "localhost")
+		self.rmq_vhost = os.getenv("BACKEND_API_INTERNAL_RMQ_VHOST", "app_notifications")
+		self.rmq_exchange = os.getenv("BACKEND_API_INTERNAL_RMQ_EXCHANGE", "app_notifications")
+
 		self.connection: aio_pika.RobustConnection | None = None
 		self.channel: aio_pika.RobustChannel | None = None
 		self.exchange: aio_pika.Exchange | None = None
-		self._queue: asyncio.Queue[tuple[int, dict]] = asyncio.Queue()
+		self._queue: asyncio.Queue[tuple[str, dict]] = asyncio.Queue()
 		self._flush_task: asyncio.Task | None = None
 		self._closing = False
 		self._ready = asyncio.Event()
 
 	async def connect(self):
+		username = return_credentials("/etc/secrets/rmq_username")
+		password = return_credentials("/etc/secrets/rmq_password")
+		url = (
+			f"amqp://{quote(username, safe='')}:{quote(password, safe='')}"
+			f"@{self.rmq_host}/{quote(self.rmq_vhost, safe='')}"
+		)
+
 		logger.info(f"[RabbitMQ] Connecting to {self.rmq_host}...")
-		self.connection = await aio_pika.connect_robust(self.url)
+		self.connection = await aio_pika.connect_robust(url)
 		self.channel = await self.connection.channel(publisher_confirms=True)
 		self.exchange = await self.channel.declare_exchange(
-			self.exchange_name, aio_pika.ExchangeType.TOPIC, durable=True
+			self.rmq_exchange, aio_pika.ExchangeType.TOPIC, durable=True
 		)
-		logger.info(f"[RabbitMQ] Connected and exchange '{self.exchange_name}' ready.")
+		logger.info(f"[RabbitMQ] Connected and exchange '{self.rmq_exchange}' ready.")
 		self._ready.set()
 		self._flush_task = asyncio.create_task(self._flush_queue_loop())
 		logger.debug("[RabbitMQ] Flush queue loop started")
 
 	async def publish(self, routing_key: int, message: dict):
-		await self._queue.put((routing_key, message))
+		await self._queue.put((f"notify.user.{routing_key}", message))
+		await self._queue.put((f"notify.windows.{routing_key}", message))
 		logger.debug(f"[RabbitMQ] Queued message for {routing_key}")
 
 	async def _flush_queue_loop(self):
@@ -59,12 +60,13 @@ class RabbitMQProducer:
 						content_type="application/json",
 						delivery_mode=aio_pika.DeliveryMode.PERSISTENT
 					),
-					routing_key=f"notify.user.{routing_key}",
+					routing_key=routing_key,
 					mandatory=True
 				)
 				logger.debug(f"[RabbitMQ] Successfully published to notify.user.{routing_key}")
+
 			except (AMQPException, OSError) as e:
-				logger.warning(f"[RabbitMQ] Publish failed, requeuing message: {e}", exc_info=True)
+				logger.warning(f"[RabbitMQ] Failed to publish to {routing_key}: {e}. Requeuing message.", exc_info=True)
 				await asyncio.sleep(2)
 				await self._queue.put((routing_key, message))
 			finally: