idempotancy integration

routes/api/whatsapp_webhook.py

@@ -3,11 +3,18 @@ from fastapi import APIRouter, Request, HTTPException, status
 from fastapi.responses import JSONResponse
 import logging
 import json
+import os
 from typing import Tuple, Optional
 
+# Redis for idempotency
+import redis
+
 # Classifier
 from components.LLMs.Classifier import ZeroShotClassifier
 
+# User storage (as you referenced it)
+from storage.users import ensure_user_from_webhook
+
 # Handlers
 from components.handlers.whatsapp_handlers import (
     handle_headlines,
@@ -20,9 +27,60 @@ from components.handlers.whatsapp_handlers import (
 )
 
 logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
+
 router = APIRouter()
 classifier = ZeroShotClassifier()
 
+# ---- Redis client (idempotency) ----
+REDIS_URL = os.environ.get("UPSTASH_REDIS_URL", "redis://localhost:6379")
+redis_client = redis.Redis.from_url(REDIS_URL, decode_responses=True)
+IDEMPOTENCY_TTL_SECONDS = 10 * 60  # 10 minutes
+
+
+def _mark_if_new(message_id: str) -> bool:
+    """
+    Returns True if message_id is new (we should process),
+    False if we've already seen it (skip).
+    """
+    try:
+        return bool(
+            redis_client.set(
+                f"wa:handled:{message_id}",
+                "1",
+                nx=True,  # only set if not exists
+                ex=IDEMPOTENCY_TTL_SECONDS,  # expire automatically
+            )
+        )
+    except Exception as e:
+        # Fail open if Redis hiccups (process message)
+        logging.warning(f"Idempotency Redis error: {e}")
+        return True
+
+
+def _get_incoming_message_id(payload: dict) -> Optional[str]:
+    """
+    Returns the WhatsApp message id (wamid...) if present, else None.
+    Only for real user 'messages' events (ignores statuses/billing).
+    """
+    try:
+        entries = payload.get("entry", [])
+        for entry in entries:
+            changes = entry.get("changes", [])
+            for change in changes:
+                if change.get("field") != "messages":
+                    continue
+                value = change.get("value", {}) or {}
+                msgs = value.get("messages", []) or []
+                if not msgs:
+                    continue
+                # Only look at the first message
+                msg = msgs[0]
+                return msg.get("id")
+    except Exception as e:
+        logging.debug(f"_get_incoming_message_id error: {e}")
+    return None
+
+
 def _extract_from_number_and_text(payload: dict) -> Tuple[Optional[str], Optional[str]]:
     """
     Extracts (from_number, message_text) from a WhatsApp webhook payload.
@@ -39,14 +97,10 @@ def _extract_from_number_and_text(payload: dict) -> Tuple[Optional[str], Optiona
                     continue
 
                 value = change.get("value", {}) or {}
-                # Ignore pure status/billing updates that don't have 'messages'
+                # Ignore updates without 'messages'
                 messages_list = value.get("messages", [])
                 if not messages_list:
-                    # Helpful debug so you can see why it was ignored
-                    statuses = value.get("statuses", [])
-                    if statuses:
-                        # it's a status webhook, not a user message
-                        pass
+                    # Likely a status/billing event
                     continue
 
                 # Only look at the first message in the list
@@ -87,9 +141,19 @@ def _extract_from_number_and_text(payload: dict) -> Tuple[Optional[str], Optiona
         logging.exception(f"_extract_from_number_and_text error: {e}")
         return None, None
 
+
 @router.post("/message-received")
 async def whatsapp_webhook_receiver(request: Request):
+    """
+    Receives incoming messages from WhatsApp/Gupshup webhook.
+    - Parses JSON
+    - Idempotency check on wamid
+    - Extracts from_number/message_text
+    - Upserts user profile
+    - Classifies and routes to handlers
+    """
     try:
+        # Read + parse body
         body_str = (await request.body()).decode("utf-8")
         logging.info(f"Raw webhook body received: {body_str}")
 
@@ -99,6 +163,15 @@ async def whatsapp_webhook_receiver(request: Request):
             logging.error("❌ Failed to decode webhook body as JSON")
             return JSONResponse(status_code=400, content={"error": "Invalid JSON format"})
 
+        # Idempotency: skip duplicates early
+        msg_id = _get_incoming_message_id(incoming_message)
+        if msg_id:
+            is_new = _mark_if_new(msg_id)
+            if not is_new:
+                logging.info(f"Duplicate inbound message {msg_id} ignored.")
+                return JSONResponse(status_code=200, content={"status": "ignored", "message": "Duplicate message"})
+
+        # Extract sender + message
         from_number, message_text = _extract_from_number_and_text(incoming_message)
         if not from_number or not message_text:
             logging.info("Ignoring non-message webhook or missing sender/text.")
@@ -106,10 +179,18 @@ async def whatsapp_webhook_receiver(request: Request):
 
         logging.info(f"Message from {from_number}: {message_text}")
 
+        # Upsert user (profile + last_seen + last_msg_id)
+        try:
+            ensure_user_from_webhook(incoming_message, from_number, msg_id)
+        except Exception as e:
+            logging.warning(f"ensure_user_from_webhook failed (continuing): {e}")
+
+        # Legacy fast-path (still supported)
         normalized = message_text.lower().strip().replace("’", "'")
         if normalized == "view today's headlines":
             return handle_headlines(from_number)
 
+        # Intent classification
         try:
             label, meta = classifier.classify(message_text)
             logging.info(f"Intent classified: {label} | meta={meta}")
@@ -119,6 +200,7 @@ async def whatsapp_webhook_receiver(request: Request):
                 return handle_headlines(from_number)
             return handle_help(from_number)
 
+        # Routing
         if label == "headlines_request":
             return handle_headlines(from_number)
         elif label == "preferences_update":
@@ -143,12 +225,21 @@ async def whatsapp_webhook_receiver(request: Request):
 
 @router.get("/message-received")
 async def whatsapp_webhook_verify(request: Request):
+    """
+    Webhook verification endpoint (GET with 'hub.mode' and 'hub.challenge').
+    """
     mode = request.query_params.get("hub.mode")
     challenge = request.query_params.get("hub.challenge")
 
     if mode == "subscribe" and challenge:
         logging.info(f"Webhook verification successful. Challenge: {challenge}")
-        return JSONResponse(status_code=200, content=int(challenge))
+        # Challenge often needs to be returned as an integer
+        try:
+            challenge_int = int(challenge)
+        except ValueError:
+            # If not an int, return as string to avoid 500s
+            return JSONResponse(status_code=200, content=challenge)
+        return JSONResponse(status_code=200, content=challenge_int)
     else:
         logging.warning(f"Webhook verification failed. Mode: {mode}, Challenge: {challenge}")
         raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Verification failed")

storage/users.py

@@ -0,0 +1,173 @@
+# components/store/users.py
+import os
+import time
+import json
+import logging
+from typing import Any, Dict, List, Optional
+
+import redis
+
+# Reuse your Upstash Redis
+REDIS_URL = os.environ.get("UPSTASH_REDIS_URL", "redis://localhost:6379")
+redis_client = redis.Redis.from_url(REDIS_URL, decode_responses=True)
+
+# Key helpers
+def _user_key(wa_id: str) -> str:
+    return f"user:{wa_id}"
+
+def _prefs_key(wa_id: str) -> str:
+    return f"user:{wa_id}:prefs"
+
+# ------------- Core CRUD -------------
+
+def get_user(wa_id: str) -> Optional[Dict[str, Any]]:
+    try:
+        data = redis_client.hgetall(_user_key(wa_id))
+        if not data:
+            return None
+        # Normalize types
+        if "created_at" in data:
+            data["created_at"] = float(data["created_at"])
+        if "updated_at" in data:
+            data["updated_at"] = float(data["updated_at"])
+        if "last_seen" in data:
+            data["last_seen"] = float(data["last_seen"])
+        if "last_msg_ts" in data:
+            data["last_msg_ts"] = float(data["last_msg_ts"])
+        # Preferences
+        prefs = redis_client.get(_prefs_key(wa_id))
+        data["preferences"] = json.loads(prefs) if prefs else {"topics": [], "regions": [], "time": None}
+        return data
+    except Exception as e:
+        logging.exception(f"get_user error for {wa_id}: {e}")
+        return None
+
+
+def upsert_user(
+    wa_id: str,
+    display_name: Optional[str] = None,
+    phone_e164: Optional[str] = None,
+    source: str = "whatsapp",
+    extra: Optional[Dict[str, Any]] = None,
+) -> Dict[str, Any]:
+    """
+    Create or update a user profile. Non-destructive: only updates provided fields.
+    """
+    now = time.time()
+    key = _user_key(wa_id)
+    # pipe for atomic-ish writes
+    p = redis_client.pipeline()
+    try:
+        exists = redis_client.exists(key)
+        fields = {
+            "wa_id": wa_id,
+            "source": source,
+            "updated_at": now,
+        }
+        if display_name is not None:
+            fields["display_name"] = display_name
+        if phone_e164 is not None:
+            fields["phone_e164"] = phone_e164
+        if extra:
+            # store a small JSON for misc signals (e.g., locale)
+            fields["extra_json"] = json.dumps(extra)[:4096]
+
+        if not exists:
+            fields["created_at"] = now
+            fields.setdefault("consent", "unknown")  # 'opt_in'|'opt_out'|'unknown'
+        p.hset(key, mapping={k: str(v) for k, v in fields.items()})
+        p.execute()
+        user = get_user(wa_id) or {}
+        return user
+    except Exception as e:
+        logging.exception(f"upsert_user error for {wa_id}: {e}")
+        return {"wa_id": wa_id, "error": str(e)}
+
+
+def set_last_seen(wa_id: str, ts: Optional[float] = None) -> None:
+    try:
+        redis_client.hset(_user_key(wa_id), mapping={
+            "last_seen": str(ts if ts is not None else time.time())
+        })
+    except Exception as e:
+        logging.exception(f"set_last_seen error for {wa_id}: {e}")
+
+
+def record_inbound_message(wa_id: str, message_id: str, ts: Optional[float] = None) -> None:
+    try:
+        redis_client.hset(_user_key(wa_id), mapping={
+            "last_msg_id": message_id,
+            "last_msg_ts": str(ts if ts is not None else time.time())
+        })
+    except Exception as e:
+        logging.exception(f"record_inbound_message error for {wa_id}: {e}")
+
+
+def set_preferences(wa_id: str, topics: Optional[List[str]] = None,
+                    regions: Optional[List[str]] = None, time_pref: Optional[str] = None) -> Dict[str, Any]:
+    """
+    Persist user preferences as a small JSON blob.
+    - topics: e.g., ["india","finance"]
+    - regions: e.g., ["IN","EU"]
+    - time_pref: e.g., "09:00"
+    """
+    try:
+        current = get_user(wa_id) or {}
+        prefs = current.get("preferences", {}) or {}
+        if topics is not None:
+            prefs["topics"] = [t.strip().lower() for t in topics if t.strip()]
+        if regions is not None:
+            prefs["regions"] = [r.strip().upper() for r in regions if r.strip()]
+        if time_pref is not None:
+            prefs["time"] = time_pref
+        redis_client.set(_prefs_key(wa_id), json.dumps(prefs))
+        # Touch updated_at
+        redis_client.hset(_user_key(wa_id), mapping={"updated_at": str(time.time())})
+        current["preferences"] = prefs
+        return current
+    except Exception as e:
+        logging.exception(f"set_preferences error for {wa_id}: {e}")
+        return {"wa_id": wa_id, "error": str(e)}
+
+# ------------- Webhook helpers -------------
+
+def _extract_contact_name(payload: Dict[str, Any]) -> Optional[str]:
+    """
+    Try multiple shapes:
+    value.contacts[0].profile.name
+    """
+    try:
+        for entry in payload.get("entry", []):
+            for change in entry.get("changes", []):
+                if change.get("field") != "messages":
+                    continue
+                val = change.get("value", {}) or {}
+                contacts = val.get("contacts", []) or []
+                if contacts:
+                    prof = contacts[0].get("profile", {}) or {}
+                    name = prof.get("name")
+                    if name:
+                        return name
+    except Exception:
+        pass
+    return None
+
+
+def ensure_user_from_webhook(payload: Dict[str, Any], from_number: str, message_id: Optional[str]) -> Dict[str, Any]:
+    """
+    Ensure the user exists/updated using webhook payload signals.
+    - from_number: WhatsApp 'from' (wa_id / E.164)
+    - message_id: wamid... (for last message stamping)
+    """
+    display_name = _extract_contact_name(payload)
+    user = upsert_user(
+        wa_id=from_number,
+        display_name=display_name,
+        phone_e164=from_number,
+        source="whatsapp",
+        extra=None,
+    )
+    set_last_seen(from_number)
+    if message_id:
+        record_inbound_message(from_number, message_id)
+    return user