diff --git a/app/idempotency.py b/app/idempotency.py new file mode 100644 index 0000000..05dca3a --- /dev/null +++ b/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() diff --git a/app/main.py b/app/main.py index 44025f6..c73b6be 100644 --- a/app/main.py +++ b/app/main.py @@ -5,6 +5,8 @@ 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__) @@ -12,6 +14,19 @@ 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} diff --git a/app/worker.py b/app/worker.py new file mode 100644 index 0000000..edb8dd8 --- /dev/null +++ b/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()