refactor(notification): use job to send discord and telegram notifications

Author: M03ED

app/__init__.py

@@ -66,7 +66,7 @@ async def lifespan(app: FastAPI):
     openapi_url="/openapi.json" if DOCS else None,
 )
 
-scheduler = AsyncIOScheduler(job_defaults={"max_instances": 20}, timezone="UTC")
+scheduler = AsyncIOScheduler(job_defaults={"max_instances": 30}, timezone="UTC")
 logger = get_logger()
 
 

app/jobs/process_notification_queues.py

@@ -0,0 +1,32 @@
+import asyncio
+
+from app import scheduler
+from app.notification.client import process_telegram_queue, process_discord_queue
+from app.utils.logger import get_logger
+from config import JOB_SEND_NOTIFICATIONS_INTERVAL
+
+logger = get_logger("process-notification-queues")
+
+
+async def process_all_notification_queues():
+    """
+    Process both Telegram and Discord notification queues concurrently.
+    Each queue sends messages one by one internally.
+    """
+    logger.debug("Processing notification queues")
+
+    await asyncio.gather(
+        process_telegram_queue(),
+        process_discord_queue(),
+        return_exceptions=True
+    )
+
+
+# Schedule the job to run at the same interval as webhook notifications
+scheduler.add_job(
+    process_all_notification_queues,
+    "interval",
+    seconds=JOB_SEND_NOTIFICATIONS_INTERVAL,
+    max_instances=1,
+    coalesce=True
+)

app/notification/client.py

@@ -5,6 +5,14 @@
 from app.settings import notification_settings
 from app.utils.logger import get_logger
 from app import on_startup
+from app.notification.queue_manager import (
+    telegram_queue,
+    discord_queue,
+    enqueue_telegram,
+    enqueue_discord,
+    TelegramNotification,
+    DiscordNotification,
+)
 
 
 client = None
@@ -30,51 +38,59 @@ async def define_client():
 logger = get_logger("Notification")
 
 
-async def send_discord_webhook(json_data, webhook):
-    max_retries = (await notification_settings()).max_retries
+async def _send_discord_webhook_direct(json_data, webhook, max_retries: int) -> bool:
+    """
+    Internal function to send Discord webhook with proper retry_after handling.
+    Returns True if successful, False otherwise.
+    """
     retries = 0
     while retries < max_retries:
         try:
             response = await client.post(webhook, json=json_data)
             if response.status_code in [200, 204]:
                 logger.debug(f"Discord webhook payload delivered successfully, code {response.status_code}.")
-                return
+                return True
             elif response.status_code == 429:
                 retries += 1
                 if retries < max_retries:
-                    await asyncio.sleep(0.5)
+                    # Extract retry_after from response
+                    try:
+                        retry_after = response.json().get("retry_after", 0.5)
+                    except Exception:
+                        retry_after = 0.5
+                    logger.warning(f"Discord rate limit hit, waiting {retry_after}s (attempt {retries}/{max_retries})")
+                    await asyncio.sleep(retry_after)
                     continue
             else:
                 response_text = response.text
                 logger.error(f"Discord webhook failed: {response.status_code} - {response_text}")
-                return
+                return False
         except Exception as err:
             logger.error(f"Discord webhook failed Exception: {str(err)}")
-            return
+            return False
 
     logger.error(f"Discord webhook failed after {max_retries} retries")
+    return False
 
 
-async def send_telegram_message(
-    message, chat_id: int | None = None, channel_id: int | None = None, topic_id: int | None = None
-):
+async def send_discord_webhook(json_data, webhook):
+    """Enqueue Discord notification for processing"""
+    await enqueue_discord(json_data, webhook)
+
+
+async def _send_telegram_message_direct(
+    message: str,
+    chat_id: int | None,
+    channel_id: int | None,
+    topic_id: int | None,
+    max_retries: int,
+    telegram_api_token: str
+) -> bool:
     """
-    Send a message to Telegram based on the available IDs.
-    Args:
-        message (str): The message to send
-        chat_id (int, optional): The chat ID for direct messages
-        channel_id (int, optional): The channel ID for channel messages
-        topic_id (int, optional): The topic ID for forum topics in channels
-    Returns:
-        bool: True if message was sent successfully, False otherwise
+    Internal function to send Telegram message with proper retry_after handling.
+    Returns True if successful, False otherwise.
     """
-    # Ensure TELEGRAM_API_TOKEN is available
-    settings: NotificationSettings = await notification_settings()
-    if not settings.telegram_api_token:
-        logger.error("TELEGRAM_API_TOKEN is not defined")
-        return
-
-    base_url = f"https://api.telegram.org/bot{settings.telegram_api_token}/sendMessage"
+    base_url = f"https://api.telegram.org/bot{telegram_api_token}/sendMessage"
     payload = {"parse_mode": "HTML", "text": message}
 
     # Determine the target chat/channel/topic
@@ -87,27 +103,114 @@ async def send_telegram_message(
         payload["chat_id"] = chat_id
     else:
         logger.error("At least one of chat_id, channel_id must be provided")
-        return
+        return False
 
-    max_retries = settings.max_retries
     retries = 0
     while retries < max_retries:
         try:
             response = await client.post(base_url, data=payload)
             if response.status_code == 200:
                 logger.debug(f"Telegram message sent successfully, code {response.status_code}.")
-                return
+                return True
             elif response.status_code == 429:
                 retries += 1
                 if retries < max_retries:
-                    await asyncio.sleep(0.5)
+                    # Extract retry_after from Telegram response
+                    try:
+                        retry_after = response.json().get("parameters", {}).get("retry_after", 0.5)
+                    except Exception:
+                        retry_after = 0.5
+                    logger.warning(f"Telegram rate limit hit, waiting {retry_after}s (attempt {retries}/{max_retries})")
+                    await asyncio.sleep(retry_after)
                     continue
             else:
                 response_text = response.text
                 logger.error(f"Telegram message failed: {response.status_code} - {response_text}")
-                return
+                return False
         except Exception as err:
             logger.error(f"Telegram message failed: {str(err)}")
-            return
+            return False
 
     logger.error(f"Telegram message failed after {max_retries} retries")
+    return False
+
+
+async def send_telegram_message(
+    message, chat_id: int | None = None, channel_id: int | None = None, topic_id: int | None = None
+):
+    """
+    Enqueue a Telegram message for processing.
+    Args:
+        message (str): The message to send
+        chat_id (int, optional): The chat ID for direct messages
+        channel_id (int, optional): The channel ID for channel messages
+        topic_id (int, optional): The topic ID for forum topics in channels
+    """
+    await enqueue_telegram(message, chat_id, channel_id, topic_id)
+
+
+async def process_telegram_queue():
+    """
+    Process Telegram notification queue, sending messages one by one.
+    """
+    settings: NotificationSettings = await notification_settings()
+    if not settings.telegram_api_token:
+        return
+
+    processed = 0
+    failed = 0
+
+    while not telegram_queue.empty():
+        try:
+            notification: TelegramNotification = await telegram_queue.get()
+
+            success = await _send_telegram_message_direct(
+                message=notification.message,
+                chat_id=notification.chat_id,
+                channel_id=notification.channel_id,
+                topic_id=notification.topic_id,
+                max_retries=settings.max_retries,
+                telegram_api_token=settings.telegram_api_token
+            )
+
+            if success:
+                processed += 1
+            else:
+                failed += 1
+        except Exception as err:
+            logger.error(f"Error processing Telegram notification: {str(err)}")
+            failed += 1
+
+    if processed > 0 or failed > 0:
+        logger.info(f"Telegram queue processed: {processed} sent, {failed} failed")
+
+
+async def process_discord_queue():
+    """
+    Process Discord notification queue, sending webhooks one by one.
+    """
+    settings: NotificationSettings = await notification_settings()
+
+    processed = 0
+    failed = 0
+
+    while not discord_queue.empty():
+        try:
+            notification: DiscordNotification = await discord_queue.get()
+
+            success = await _send_discord_webhook_direct(
+                json_data=notification.json_data,
+                webhook=notification.webhook,
+                max_retries=settings.max_retries
+            )
+
+            if success:
+                processed += 1
+            else:
+                failed += 1
+        except Exception as err:
+            logger.error(f"Error processing Discord notification: {str(err)}")
+            failed += 1
+
+    if processed > 0 or failed > 0:
+        logger.info(f"Discord queue processed: {processed} sent, {failed} failed")

app/notification/queue_manager.py

@@ -0,0 +1,49 @@
+import asyncio
+from typing import Optional
+from pydantic import BaseModel, Field
+
+
+class TelegramNotification(BaseModel):
+    """Model for Telegram notification queue items"""
+    message: str
+    chat_id: Optional[int] = Field(default=None)
+    channel_id: Optional[int] = Field(default=None)
+    topic_id: Optional[int] = Field(default=None)
+    tries: int = Field(default=0)
+
+
+class DiscordNotification(BaseModel):
+    """Model for Discord notification queue items"""
+    json_data: dict
+    webhook: str
+    tries: int = Field(default=0)
+
+
+# Global queues for Telegram and Discord notifications
+telegram_queue: asyncio.Queue[TelegramNotification] = asyncio.Queue()
+discord_queue: asyncio.Queue[DiscordNotification] = asyncio.Queue()
+
+
+async def enqueue_telegram(
+    message: str,
+    chat_id: Optional[int] = None,
+    channel_id: Optional[int] = None,
+    topic_id: Optional[int] = None
+) -> None:
+    """Add a Telegram notification to the queue"""
+    notification = TelegramNotification(
+        message=message,
+        chat_id=chat_id,
+        channel_id=channel_id,
+        topic_id=topic_id
+    )
+    await telegram_queue.put(notification)
+
+
+async def enqueue_discord(json_data: dict, webhook: str) -> None:
+    """Add a Discord notification to the queue"""
+    notification = DiscordNotification(
+        json_data=json_data,
+        webhook=webhook
+    )
+    await discord_queue.put(notification)