Add background worker to process replies asynchronously and idempoten…

app/idempotency.py

@@ -0,0 +1,34 @@
+import aioredis
+from app.config import settings
+
+class IdempotencyStore:
+    def __init__(self, redis_url: str | None = None):
+        self.redis_url = redis_url or settings.REDIS_URL
+        self._redis = None
+
+    async def _get_redis(self):
+        if self._redis is None:
+            # using redis-py's asyncio client (aioredis compatibility)
+            self._redis = await aioredis.from_url(self.redis_url)
+        return self._redis
+
+    async def seen_or_set(self, key: str, ttl: int = 24 * 60 * 60) -> bool:
+        """
+        Atomically set key with NX and expiry. Return True if the key was already present (i.e., duplicate),
+        False if we successfully set it (i.e., not duplicate).
+        """
+        redis = await self._get_redis()
+        # redis.set returns True if set, None if key exists
+        res = await redis.set(key, "1", ex=ttl, nx=True)
+        if res:
+            return False
+        return True
+
+    async def close(self):
+        if self._redis:
+            await self._redis.close()
+            self._redis = None
+
+
+# singleton
+idempotency = IdempotencyStore()

app/main.py

@@ -5,13 +5,28 @@
 from app.reply_generator import generate_reply
 from app.messaging import client as messaging_client
 from app.config import settings
+from app.worker import worker
+from app.idempotency import idempotency
 
 logging.basicConfig(level=logging.INFO)
 logger = logging.getLogger(__name__)
 
 app = FastAPI()
 store = ConvoStore(redis_url=settings.REDIS_URL)
 
+
+@app.on_event("startup")
+async def startup():
+    # start background worker
+    await worker.start()
+
+
+@app.on_event("shutdown")
+async def shutdown():
+    await worker.stop()
+    await idempotency.close()
+
+
 @app.get("/webhook")
 async def webhook_verify(request: Request):
     # Used for Meta/WhatsApp webhook verification
@@ -24,6 +39,7 @@ async def webhook_verify(request: Request):
         return int(challenge or 0)
     raise HTTPException(status_code=400, detail="Invalid verify token")
 
+
 @app.post("/webhook")
 async def webhook(request: Request):
     # Twilio posts form-encoded data; Meta posts JSON
@@ -38,6 +54,8 @@ async def webhook(request: Request):
     # parse depending on provider
     provider = getattr(settings, "WHATSAPP_PROVIDER", "twilio").lower()
 
+    provider_message_id = None
+
     if provider == "meta":
         try:
             payload = await request.json()
@@ -61,6 +79,7 @@ async def webhook(request: Request):
             message = messages[0]
             body = message.get("text", {}).get("body", "")
             from_number = "whatsapp:" + message.get("from", "")
+            provider_message_id = message.get("id") or message.get("message_id")
         except Exception as e:
             logger.exception("Failed to parse Meta payload")
             raise HTTPException(status_code=400, detail="Malformed payload")
@@ -69,6 +88,7 @@ async def webhook(request: Request):
         form = await request.form()
         body = form.get("Body") or ""
         from_number = form.get("From") or ""
+        provider_message_id = form.get("MessageSid") or form.get("SmsSid")
         params = dict(form)
         # Validate request
         valid = messaging_client.validate_request(public_url, params, headers)
@@ -79,21 +99,28 @@ async def webhook(request: Request):
     if not from_number:
         raise HTTPException(status_code=400, detail="Missing From")
 
+    # Idempotency check
+    if provider_message_id:
+        key = f"idempotency:{provider}:{provider_message_id}"
+        duplicate = await idempotency.seen_or_set(key)
+        if duplicate:
+            logger.info("Duplicate message received: %s", provider_message_id)
+            return {"status": "duplicate"}
+
     # Classify urgency
     urgency, score = classify_urgency(body)
+
     # Persist message and metadata
     await store.append_message(from_number, {"text": body, "urgency": urgency, "score": score})
-    # Decide response
+
+    # Enqueue processing to background worker (generate reply + send)
     if urgency == "high":
-        reply = "I detected this might be urgent. Do you want me to escalate this to support now?"
+        # short-circuit immediate response for high urgency is optional; here we still enqueue
+        reply_template = "I detected this might be urgent. Do you want me to escalate this to support now?"
+        await worker.enqueue({"from": from_number, "body": body, "history": [], "reply_override": reply_template})
+        return {"status": "ok", "urgency": urgency}
     else:
         history = await store.get_history(from_number, limit=10)
-        reply = await generate_reply(body, history)
-    # Send reply via provider client
-    try:
-        messaging_client.send(to=from_number, body=reply)
-    except Exception:
-        logger.exception("Failed to send message")
-        raise HTTPException(status_code=500, detail="Failed to send message")
+        await worker.enqueue({"from": from_number, "body": body, "history": history})
 
     return {"status": "ok", "urgency": urgency}

app/worker.py

@@ -0,0 +1,72 @@
+import asyncio
+import logging
+from typing import Any, Dict
+from app.reply_generator import generate_reply
+from app.messaging import client as messaging_client
+
+logger = logging.getLogger(__name__)
+
+class Worker:
+    def __init__(self):
+        self.queue: asyncio.Queue = asyncio.Queue()
+        self._task: asyncio.Task | None = None
+        self._running = False
+
+    async def start(self):
+        if self._running:
+            return
+        self._running = True
+        self._task = asyncio.create_task(self._run())
+        logger.info("Background worker started")
+
+    async def stop(self):
+        if not self._running:
+            return
+        self._running = False
+        # push sentinel to wake the worker and let it exit
+        await self.queue.put(None)
+        if self._task:
+            await self._task
+        logger.info("Background worker stopped")
+
+    async def _run(self):
+        while True:
+            item = await self.queue.get()
+            if item is None:
+                # sentinel to stop
+                break
+            try:
+                await self._process_item(item)
+            except Exception:
+                logger.exception("Worker failed to process item")
+            finally:
+                try:
+                    self.queue.task_done()
+                except Exception:
+                    pass
+
+    async def _process_item(self, item: Dict[str, Any]):
+        """Process a queued message: generate reply and send it."""
+        from_number = item.get("from")
+        body = item.get("body", "")
+        history = item.get("history", [])
+
+        # Generate reply (may internally call blocking code via run_in_executor)
+        try:
+            reply = await generate_reply(body, history)
+        except Exception:
+            logger.exception("Failed to generate reply; using fallback")
+            reply = "Thanks — I received your message. Can you tell me more?"
+
+        # Send message (messaging_client.send may be blocking -> run in thread)
+        try:
+            await asyncio.to_thread(messaging_client.send, to=from_number, body=reply)
+        except Exception:
+            logger.exception("Failed to send message to %s", from_number)
+
+    async def enqueue(self, item: Dict[str, Any]):
+        await self.queue.put(item)
+
+
+# singleton worker instance to be used by the app
+worker = Worker()