feat: add webhook idempotency check

- Created 'processed_updates' table to record incoming Telegram update IDs.
- Modified main.py to immediately return 200 OK for duplicate update IDs.
- Added daily cleanup routine in cron.py to prune updates older than 48 hours.
- Prevents redundant command execution when Cloud Run processes responses slowly and Telegram sends retries.

Author: rozen03

cron.py

@@ -3,9 +3,12 @@
 import logging
 import traceback
 import resource
+import datetime
 from telegram.ext import Application
 from tg_ids import ROZEN_CHATID
 from bot_logic import _update_groups, felizdia, actualizarRiver
+from handlers.db import get_session
+from models import ProcessedUpdate
 
 def main():
     """Runs the update_groups command."""
@@ -17,6 +20,15 @@ def main():
     application = Application.builder().token(os.environ["TELEGRAM_BOT_TOKEN"]).build()
 
     async def run_update():
+        try:
+            # Clean up old processed updates (older than 48h) to save DB space
+            with get_session() as session:
+                forty_eight_hours_ago = datetime.datetime.utcnow() - datetime.timedelta(hours=48)
+                deleted = session.query(ProcessedUpdate).filter(ProcessedUpdate.timestamp < forty_eight_hours_ago).delete()
+                logging.info(f"Cleaned up {deleted} old processed updates from DB")
+        except Exception as e:
+            logging.error(f"Error cleaning up processed updates: {e}", exc_info=True)
+
         try:
             await felizdia(application)
         except Exception as e:

main.py

@@ -11,6 +11,8 @@
 
 from bot_logic import COMMANDS, button
 from tg_ids import ROZEN_CHATID
+from handlers.db import get_session
+from models import ProcessedUpdate
 
 async def error_handler(update, context):
     """Log the error and send a telegram message to notify the developer."""
@@ -117,6 +119,18 @@ async def telegram_webhook(token: str, request: Request):
     try:
         data = await request.json()
         update = telegram.Update.de_json(data=data, bot=application.bot)
+        
+        # Idempotency check: Have we processed this update_id already?
+        with get_session() as session:
+            existing = session.query(ProcessedUpdate).filter_by(update_id=update.update_id).first()
+            if existing:
+                logging.getLogger("DCUBABOT").info(f"Skipping duplicate update_id {update.update_id}")
+                return Response(status_code=200)
+            
+            # If not, mark as processed and commit immediately
+            session.add(ProcessedUpdate(update_id=update.update_id))
+            session.commit()
+
         # Await the processing SYNCHRONOUSLY before returning 200 OK
         # This prevents Cloud Run from throttling the CPU while the bot is doing work
         # TODO: If user traffic scales significantly and Telegram throws `RetryAfter` for

models.py

@@ -39,6 +39,11 @@ class Lock(Base):
     key = Column(String, primary_key=True)
     expires_at = Column(DateTime, nullable=False)
 
+class ProcessedUpdate(Base):
+    __tablename__ = 'processed_updates'
+    update_id = Column(Integer, primary_key=True)
+    timestamp = Column(DateTime, nullable=False, default=datetime.datetime.utcnow)
+
 class SentMessage(Base):
     __tablename__ = 'sent_messages'
     id = Column(Integer, primary_key=True)