mvp fix

app/ai/routes/chat.py

@@ -3,16 +3,21 @@
 AIDA Chat Endpoint - Brain-Based AI Agent
 """
 
-from fastapi import APIRouter, HTTPException
+from fastapi import APIRouter, HTTPException, Request
+from app.middleware import limiter, AI_RATE_LIMIT
 from pydantic import BaseModel
 from typing import Optional, Dict, Any
 from structlog import get_logger
 from uuid import uuid4
 from datetime import datetime
+import hashlib
+import json
+import asyncio
 from langgraph.types import Command
 
 from app.ai.agent.graph import get_aida_graph
 from app.ai.agent.schemas import AgentResponse
+from app.ai.config import redis_client
 
 logger = get_logger(__name__)
 
@@ -44,8 +49,13 @@ class AskBody(BaseModel):
 # MAIN CHAT ENDPOINT - LANGGRAPH POWERED (FIXED)
 # ============================================================
 
+# Concurrency Control: Max 5 concurrent AI requests
+# This prevents the server from being overwhelmed by CPU-heavy processing
+ai_semaphore = asyncio.Semaphore(5)
+
 @router.post("/ask", response_model=AgentResponse)
-async def ask_ai(body: AskBody) -> AgentResponse:
+@limiter.limit(AI_RATE_LIMIT)
+async def ask_ai(request: Request, body: AskBody) -> AgentResponse:
     """
     Main chat endpoint using LangGraph state machine.
     
@@ -58,19 +68,15 @@ async def ask_ai(body: AskBody) -> AgentResponse:
     Flow:
     1. Validate input
     2. Build input dict
-    3. Invoke graph with dict input and high recursion_limit
-    4. Extract final_response from returned dict
-    5. Return to client
+    3. Run graph.ainvoke() (Thread-safe)
+    4. Extract and clean response
     """
     
-    logger.info("🚀 Chat request received", message_len=len(body.message))
+    # ============================================================
+    # STEP 1: Validate input
+    # ============================================================
     
-    try:
-        # ============================================================
-        # STEP 1: Validate input
-        # ============================================================
-        
-        if not body.message or not body.message.strip():
+    if not body.message or not body.message.strip():
             logger.warning("❌ Empty message received")
             return AgentResponse(
                 success=False,
@@ -79,51 +85,76 @@ async def ask_ai(body: AskBody) -> AgentResponse:
                 error="Empty message",
             )
         
-        message = body.message.strip()
-        session_id = body.session_id or str(uuid4())
-        user_id = body.user_id or f"anonymous_{uuid4()}"
-        user_role = body.user_role or "renter"
-        
-        logger.info(
-            "📋 User session info",
-            user_id=user_id,
-            session_id=session_id,
-            user_role=user_role,
-            message_len=len(message),
-        )
-        
-        # ============================================================
-        # STEP 2: Build input dict for graph
-        # ============================================================
-        # ✅ CRITICAL: Pass dict, not AgentState
-        
-        input_dict = {
-            "user_id": user_id,
-            "session_id": session_id,
-            "user_role": user_role,
-            "user_name": body.user_name,
-            "user_location": body.user_location,
-            "last_user_message": message,
-            # "conversation_history": [],  <-- REMOVED: Do not overwrite history!
-            "language_detected": "en",
-            "start_new_session": body.start_new_session or False,
-            "is_voice_message": body.is_voice_message or False,  # Track voice input
-            "source": body.source or "default",
-            # Store reply_context in temp_data so brain can access it
-            "temp_data": {"reply_context": body.reply_context} if body.reply_context else {},
-        }
+    message = body.message.strip()
+    session_id = body.session_id or str(uuid4())
+    user_id = body.user_id or f"anonymous_{uuid4()}"
+    user_role = body.user_role or "renter"
+    
+    logger.info(
+        "📋 User session info",
+        user_id=user_id,
+        session_id=session_id,
+        user_role=user_role,
+        message_len=len(message),
+    )
+    
+    # ============================================================
+    # STEP 2: Build input dict for graph
+    # ============================================================
+    # ✅ CRITICAL: Pass dict, not AgentState
+    
+    input_dict = {
+        "user_id": user_id,
+        "session_id": session_id,
+        "user_role": user_role,
+        "user_name": body.user_name,
+        "user_location": body.user_location,
+        "last_user_message": message,
+        # "conversation_history": [],  <-- REMOVED: Do not overwrite history!
+        "language_detected": "en",
+        "start_new_session": body.start_new_session or False,
+        "is_voice_message": body.is_voice_message or False,  # Track voice input
+        "source": body.source or "default",
+        # Store reply_context in temp_data so brain can access it
+        "temp_data": {"reply_context": body.reply_context} if body.reply_context else {},
+    }
         
         # Only initialize history if starting new session
-        if body.start_new_session:
-            input_dict["conversation_history"] = []
-            input_dict["provided_fields"] = {}
-            input_dict["missing_required_fields"] = []
-            logger.info("🆕 Starting NEW session (clearing state)")
+    # Only initialize history if starting new session
+    if body.start_new_session:
+        input_dict["conversation_history"] = []
+        input_dict["provided_fields"] = {}
+        input_dict["missing_required_fields"] = []
+        logger.info("🆕 Starting NEW session (clearing state)")
+    
+    logger.info("📦 Input dict prepared", keys=list(input_dict.keys()))
         
-        logger.info("📦 Input dict prepared", keys=list(input_dict.keys()))
+    # ============================================================
+    # STEP 3: Smart Caching Check (Cost Saver)
+    # ============================================================
+    
+    # Generate cache key: ai_cache:{role}:{hash_of_message}
+    # Only cache if not a new session (context matters) and message is substantial (>5 chars)
+    cache_key = None
+    should_cache = False
+    
+    if not body.start_new_session and len(body.message) > 5 and redis_client:
+        msg_hash = hashlib.md5(body.message.lower().strip().encode()).hexdigest()
+        cache_key = f"ai_cache:{user_role}:{msg_hash}"
+        should_cache = True
+        
+        try:
+            cached_data = await redis_client.get(cache_key)
+            if cached_data:
+                logger.info(f"⚡ CACHE HIT: Serving cached response for {cache_key}")
+                cached_response = json.loads(cached_data)
+                # Return as AgentResponse object
+                return AgentResponse(**cached_response)
+        except Exception as e:
+            logger.warning(f"⚠️ Cache read failed: {e}")
         
         # ============================================================
-        # STEP 3: Get graph and validate
+        # STEP 4: Get graph and validate
         # ============================================================
         
         try:
@@ -158,15 +189,21 @@ async def ask_ai(body: AskBody) -> AgentResponse:
         try:
             # ✅ CRITICAL FIX: Pass recursion_limit config to prevent infinite loops
             # ✅ CRITICAL FIX: Pass thread_id for persistence
-            config = {
-                "recursion_limit": 50,
-                "configurable": {"thread_id": session_id}
-            }
             
-            final_state_dict = await graph.ainvoke(
-                input_dict,
-                config=config
-            )
+            # CONCURRENCY CONTROL: Wait here if server is busy
+            if ai_semaphore.locked():
+                 logger.warning(f"⚠️ AI Semaphore FULL/LOCKED - Request for {user_id} waiting...")
+            
+            async with ai_semaphore:
+                config = {
+                    "recursion_limit": 50,
+                    "configurable": {"thread_id": session_id}
+                }
+                
+                final_state_dict = await graph.ainvoke(
+                    input_dict,
+                    config=config
+                )
             
             # ✅ CRITICAL: final_state_dict is a DICT, not AgentState!
             # Access with dict keys: ['key'], not .attribute
@@ -256,18 +293,24 @@ async def ask_ai(body: AskBody) -> AgentResponse:
         
         logger.info("✅ Fallback response built", action=response.action)
         
+        # CACHE WRITE (Task 2.2)
+        if should_cache and cache_key and response.success:
+            try:
+                # Cache for 1 hour
+                await redis_client.setex(
+                    cache_key,
+                    3600, 
+                    json.dumps(response.dict())
+                )
+                logger.info(f"💾 Response cached: {cache_key}")
+            except Exception as e:
+                logger.warning(f"⚠️ Cache write warning: {e}")
+        
         return response
     
-    except HTTPException:
-        raise
-    except Exception as e:
-        logger.error("❌ Chat endpoint critical error", exc_info=e)
-        return AgentResponse(
-            success=False,
-            text="An unexpected error occurred. Please try again.",
-            action="error",
-            error=str(e),
-        )
+    
+    # End of ask_ai
+
 
 
 # ============================================================

app/database.py

@@ -19,7 +19,14 @@ db = DatabaseConnection()
 async def connect_db():
     """Connect to MongoDB"""
     try:
-        db.client = AsyncClient(settings.MONGODB_URL)
+        db.client = AsyncClient(
+            settings.MONGODB_URL,
+            maxPoolSize=100,           # Max connections (Scale: ~300 per replica member)
+            minPoolSize=10,            # Keep warm connections ready
+            maxIdleTimeMS=45000,       # Close idle connections after 45s
+            waitQueueTimeoutMS=10000,  # Fail fast if pool is exhausted (10s)
+            serverSelectionTimeoutMS=5000, # Fail fast if DB is down (5s)
+        )
         db.database = db.client[settings.MONGODB_DATABASE]
         
         # Test connection

app/middleware/__init__.py

@@ -0,0 +1,29 @@
+# ============================================================
+# app/middleware/__init__.py - Middleware Package
+# ============================================================
+
+from .rate_limiter import (
+    limiter,
+    get_limiter,
+    rate_limit_exceeded_handler,
+    AI_RATE_LIMIT,
+    AUTH_RATE_LIMIT,
+    SEARCH_RATE_LIMIT,
+    STANDARD_RATE_LIMIT,
+    WEBSOCKET_RATE_LIMIT,
+    HEAVY_RATE_LIMIT,
+    is_exempt,
+)
+
+__all__ = [
+    "limiter",
+    "get_limiter",
+    "rate_limit_exceeded_handler",
+    "AI_RATE_LIMIT",
+    "AUTH_RATE_LIMIT",
+    "SEARCH_RATE_LIMIT",
+    "STANDARD_RATE_LIMIT",
+    "WEBSOCKET_RATE_LIMIT",
+    "HEAVY_RATE_LIMIT",
+    "is_exempt",
+]

app/middleware/rate_limiter.py

@@ -0,0 +1,156 @@
+# ============================================================
+# app/middleware/rate_limiter.py - API Rate Limiting
+# ============================================================
+# 
+# Production-grade rate limiting using slowapi with Redis backend
+# for distributed rate limiting across multiple server instances.
+#
+# Rate Limits:
+# - AI endpoints (/ai/*): 10 requests/minute (expensive LLM calls)
+# - Auth endpoints (/api/auth/*): 20 requests/minute (security)
+# - Standard API: 100 requests/minute
+# - WebSocket connections: 10/minute per IP
+# ============================================================
+
+from slowapi import Limiter
+from slowapi.util import get_remote_address
+from slowapi.errors import RateLimitExceeded
+from slowapi.middleware import SlowAPIMiddleware
+from fastapi import Request
+from fastapi.responses import JSONResponse
+import logging
+import os
+
+logger = logging.getLogger(__name__)
+
+# ============================================================
+# Configuration
+# ============================================================
+
+# Use Redis for distributed rate limiting in production
+# Falls back to in-memory for development
+REDIS_URL = os.getenv("REDIS_URL", None)
+
+def get_client_identifier(request: Request) -> str:
+    """
+    Get a unique identifier for the client.
+    Priority:
+    1. Authenticated user ID (if available in request state)
+    2. X-Forwarded-For header (for proxied requests)
+    3. Client IP address
+    """
+    # Try to get authenticated user ID
+    if hasattr(request.state, "user_id") and request.state.user_id:
+        return f"user:{request.state.user_id}"
+    
+    # Check for forwarded IP (behind load balancer/proxy)
+    forwarded = request.headers.get("X-Forwarded-For")
+    if forwarded:
+        # X-Forwarded-For can contain multiple IPs, get the first one
+        return forwarded.split(",")[0].strip()
+    
+    # Fall back to direct client IP
+    return get_remote_address(request)
+
+
+# ============================================================
+# Limiter Instance
+# ============================================================
+
+# Configure storage backend
+if REDIS_URL:
+    storage_uri = REDIS_URL
+    logger.info(f"Rate limiter using Redis backend")
+else:
+    storage_uri = "memory://"
+    logger.warning("Rate limiter using in-memory backend (not suitable for production)")
+
+limiter = Limiter(
+    key_func=get_client_identifier,
+    default_limits=["100/minute"],  # Default for all endpoints
+    storage_uri=storage_uri,
+    strategy="fixed-window",  # Simple and efficient
+    headers_enabled=True,  # Include X-RateLimit-* headers in response
+)
+
+
+# ============================================================
+# Custom Rate Limit Exceeded Handler
+# ============================================================
+
+async def rate_limit_exceeded_handler(request: Request, exc: RateLimitExceeded):
+    """
+    Custom handler for rate limit exceeded errors.
+    Returns a user-friendly JSON response with retry information.
+    """
+    logger.warning(
+        f"Rate limit exceeded for {get_client_identifier(request)}: "
+        f"{request.method} {request.url.path}"
+    )
+    
+    # Parse the retry-after from the exception
+    retry_after = getattr(exc, "retry_after", 60)
+    
+    return JSONResponse(
+        status_code=429,
+        content={
+            "success": False,
+            "error_code": "RATE_LIMIT_EXCEEDED",
+            "message": "Too many requests. Please slow down.",
+            "retry_after_seconds": retry_after,
+            "detail": str(exc.detail) if hasattr(exc, "detail") else "Rate limit exceeded",
+        },
+        headers={
+            "Retry-After": str(retry_after),
+            "X-RateLimit-Limit": str(getattr(exc, "limit", "unknown")),
+        }
+    )
+
+
+# ============================================================
+# Rate Limit Decorators for Specific Endpoints
+# ============================================================
+
+# AI endpoints - expensive, limit strictly
+AI_RATE_LIMIT = "10/minute"
+
+# Auth endpoints - security-sensitive
+AUTH_RATE_LIMIT = "20/minute"
+
+# Search endpoints - moderately expensive
+SEARCH_RATE_LIMIT = "30/minute"
+
+# Standard API endpoints
+STANDARD_RATE_LIMIT = "100/minute"
+
+# WebSocket connections
+WEBSOCKET_RATE_LIMIT = "10/minute"
+
+# Heavy operations (file uploads, etc.)
+HEAVY_RATE_LIMIT = "5/minute"
+
+
+# ============================================================
+# Helper function to apply rate limiting
+# ============================================================
+
+def get_limiter():
+    """Get the configured limiter instance."""
+    return limiter
+
+
+# ============================================================
+# Exempt paths (no rate limiting)
+# ============================================================
+
+EXEMPT_PATHS = {
+    "/health",
+    "/",
+    "/docs",
+    "/openapi.json",
+    "/redoc",
+}
+
+def is_exempt(path: str) -> bool:
+    """Check if a path is exempt from rate limiting."""
+    return path in EXEMPT_PATHS

app/routes/auth.py

@@ -1,5 +1,7 @@
 import logging
 from fastapi import APIRouter, Header, HTTPException, status, Depends, Request
+from app.middleware import limiter, AUTH_RATE_LIMIT
+
 from app.schemas.auth import (
     SignupDto,
     VerifySignupOtpDto,
@@ -43,13 +45,13 @@ async def check_auth_rate_limit(identifier: str, operation: str, request: Reques
 # ============================================================
 
 @router.post("/signup", status_code=status.HTTP_200_OK)
+@limiter.limit(AUTH_RATE_LIMIT)
 async def signup(signup_dto: SignupDto, request: Request):
     """
     Step 1: Initiate Signup
     Create account and send OTP to email or phone
     """
     identifier = signup_dto.email or signup_dto.phone
-    await check_auth_rate_limit(identifier, "signup", request)
     logger.info("Signup request")
     return await auth_service.signup(signup_dto)
 
@@ -68,12 +70,12 @@ async def verify_signup_otp(dto: VerifySignupOtpDto, request: Request):
 # ============================================================
 
 @router.post("/login", status_code=status.HTTP_200_OK)
+@limiter.limit(AUTH_RATE_LIMIT)
 async def login(login_dto: LoginDto, request: Request):
     """
     User Login
     Authenticate with email or phone and password. Returns JWT token.
     """
-    await check_auth_rate_limit(login_dto.identifier, "login", request)
     logger.info(f"Login request: {login_dto.identifier}")
     return await auth_service.login(login_dto)
 

app/routes/websocket_chat.py

@@ -371,24 +371,77 @@ class ChatConnectionManager:
         """Get online status for multiple users"""
         return {uid: self.is_user_online(uid) for uid in user_ids}
     
-    async def send_to_user(self, user_id: str, message: dict):
-        """Send message to all of a user's connections"""
+    async def initialize(self):
+        """Initialize Redis Pub/Sub listener"""
+        try:
+            from app.services.redis_pubsub import redis_pubsub
+            await redis_pubsub.start_listening(self._handle_redis_message)
+            logger.info("[Chat WS] Redis Pub/Sub listener initialized")
+        except Exception as e:
+            logger.warning(f"[Chat WS] Failed to init Redis Pub/Sub: {e}")
+
+    async def _handle_redis_message(self, payload: dict):
+        """Process messages received from other servers via Redis"""
+        from app.services.redis_pubsub import redis_pubsub
+        
+        # Ignore messages from self
+        if payload.get("source_server") == id(redis_pubsub):
+            return
+
+        event_type = payload.get("type")
+        data = payload.get("data", {})
+        
+        if event_type == "direct":
+            user_id = data.get("target_user")
+            message = data.get("message")
+            await self._send_local(user_id, message)
+            
+        elif event_type == "broadcast":
+            participants = data.get("participants", [])
+            message = data.get("message")
+            sender_id = data.get("sender_id")
+            
+            for user_id in participants:
+                # Don't send back to sender (they assume it sent ok)
+                if user_id != sender_id:
+                    await self._send_local(user_id, message)
+
+    async def _send_local(self, user_id: str, message: dict) -> bool:
+        """Send message ONLY to locally connected users"""
         if user_id not in self.user_connections:
             return False
         
         disconnected = set()
+        sent = False
         for websocket in self.user_connections[user_id]:
             try:
                 await websocket.send_json(message)
+                sent = True
             except Exception as e:
                 logger.warning(f"[Chat WS] Error sending to {user_id}: {e}")
                 disconnected.add(websocket)
         
-        # Clean up disconnected
         for ws in disconnected:
             self.disconnect(ws)
+            
+        return sent
+
+    async def send_to_user(self, user_id: str, message: dict):
+        """Send message to a user (local + distributed)"""
+        # 1. Try sending locally
+        sent_local = await self._send_local(user_id, message)
         
-        return True
+        # 2. Publish to Redis for other servers
+        try:
+            from app.services.redis_pubsub import redis_pubsub
+            await redis_pubsub.publish("direct", {
+                "target_user": user_id,
+                "message": message
+            })
+        except Exception:
+            pass
+            
+        return sent_local
     
     async def broadcast_to_conversation(
         self, 
@@ -399,17 +452,27 @@ class ChatConnectionManager:
         message_id: str = None
     ):
         """
-        Send message to all participants in a conversation.
-        Tracks delivery - when recipient is online and receives it, updates delivered_to.
+        Send message to all participants in a conversation (local + distributed).
         """
+        # 1. Send to all LOCAL participants
         delivered_users = []
-        
         for user_id in participants:
-            sent_ok = await self.send_to_user(user_id, message)
+            sent_ok = await self._send_local(user_id, message)
             if sent_ok and user_id != sender_id:
                 delivered_users.append(user_id)
         
-        # If we have message_id and delivered users, update delivered_to in DB
+        # 2. Publish broadcast event to Redis for other servers
+        try:
+            from app.services.redis_pubsub import redis_pubsub
+            await redis_pubsub.publish("broadcast", {
+                "participants": participants,
+                "message": message,
+                "sender_id": sender_id
+            })
+        except Exception as e:
+            logger.warning(f"[Chat WS] Redis publish failed: {e}")
+        
+        # 3. Track delivery (only for local users currently)
         if message_id and delivered_users and sender_id:
             try:
                 from app.database import get_db
@@ -421,7 +484,7 @@ class ChatConnectionManager:
                     {"$addToSet": {"delivered_to": {"$each": delivered_users}}}
                 )
                 
-                # Notify sender that message was delivered
+                # Notify sender
                 delivery_event = {
                     "action": "message_delivered",
                     "conversation_id": conversation_id,
@@ -430,7 +493,6 @@ class ChatConnectionManager:
                 }
                 await self.send_to_user(sender_id, delivery_event)
                 
-                logger.info(f"[Chat WS] Message {message_id} delivered to {len(delivered_users)} users")
             except Exception as e:
                 logger.warning(f"[Chat WS] Failed to track delivery: {e}")
 

app/services/conversation_parts/__init__.py

@@ -0,0 +1,4 @@
+from .crud_mixin import ConversationCRUDMixin
+from .message_mixin import ConversationMessageMixin
+from .action_mixin import ConversationActionMixin
+from .ai_mixin import ConversationAIMixin

app/services/conversation_parts/action_mixin.py

@@ -0,0 +1,475 @@
+import logging
+from datetime import datetime
+from bson import ObjectId
+from fastapi import HTTPException, status
+
+from app.database import get_db
+
+logger = logging.getLogger(__name__)
+
+class ConversationActionMixin:
+    """
+    Handling actions on messages: Edit, Delete, Reactions, Clear Chat
+    """
+    
+    async def edit_message(
+        self,
+        conversation_id: str,
+        message_id: str,
+        user_id: str,
+        new_content: str,
+    ) -> dict:
+        """
+        Edit a message content.
+        Only the sender can edit, within 24 hours, text messages only.
+        """
+        db = await get_db()
+        
+        # Validate IDs
+        if not ObjectId.is_valid(conversation_id) or not ObjectId.is_valid(message_id):
+            raise HTTPException(
+                status_code=status.HTTP_400_BAD_REQUEST,
+                detail="Invalid ID format"
+            )
+        
+        # Get the message
+        message_doc = await db.messages.find_one({"_id": ObjectId(message_id)})
+        if not message_doc:
+            raise HTTPException(
+                status_code=status.HTTP_404_NOT_FOUND,
+                detail="Message not found"
+            )
+        
+        # Verify sender
+        if message_doc.get("sender_id") != user_id:
+            raise HTTPException(
+                status_code=status.HTTP_403_FORBIDDEN,
+                detail="You can only edit your own messages"
+            )
+        
+        # Verify message is in the correct conversation
+        if message_doc.get("conversation_id") != conversation_id:
+            raise HTTPException(
+                status_code=status.HTTP_400_BAD_REQUEST,
+                detail="Message does not belong to this conversation"
+            )
+        
+        # Only allow editing text messages
+        if message_doc.get("message_type") != "text":
+            raise HTTPException(
+                status_code=status.HTTP_400_BAD_REQUEST,
+                detail="Only text messages can be edited"
+            )
+        
+        # Check 15-minute edit window
+        created_at = message_doc.get("created_at")
+        if created_at:
+            minutes_since = (datetime.utcnow() - created_at).total_seconds() / 60
+            if minutes_since > 15:
+                raise HTTPException(
+                    status_code=status.HTTP_400_BAD_REQUEST,
+                    detail="Edit window expired (15 minutes)"
+                )
+        
+        # Cannot edit deleted messages
+        if message_doc.get("is_deleted"):
+            raise HTTPException(
+                status_code=status.HTTP_400_BAD_REQUEST,
+                detail="Cannot edit a deleted message"
+            )
+        
+        # Update the message
+        now = datetime.utcnow()
+        await db.messages.update_one(
+            {"_id": ObjectId(message_id)},
+            {
+                "$set": {
+                    "content": new_content.strip(),
+                    "is_edited": True,
+                    "edited_at": now,
+                }
+            }
+        )
+        
+        # Broadcast message_edited event to all participants via WebSocket
+        try:
+            from app.routes.websocket_chat import chat_manager
+            
+            conversation = await db.conversations.find_one({"_id": ObjectId(conversation_id)})
+            if conversation:
+                edit_event = {
+                    "action": "message_edited",
+                    "conversation_id": conversation_id,
+                    "message_id": message_id,
+                    "new_content": new_content.strip(),
+                    "edited_at": now.isoformat(),
+                    "edited_by": user_id,
+                }
+                
+                await chat_manager.broadcast_to_conversation(
+                    conversation_id,
+                    conversation.get("participants", []),
+                    edit_event
+                )
+                
+                logger.info(f"Broadcasted message_edited event for message {message_id}")
+        except Exception as e:
+            logger.warning(f"Failed to broadcast message_edited event: {e}")
+        
+        logger.info(f"Message {message_id} edited by {user_id}")
+        
+        return {
+            "success": True,
+            "message_id": message_id,
+            "new_content": new_content.strip(),
+            "edited_at": now.isoformat(),
+        }
+    
+    async def delete_message(
+        self,
+        conversation_id: str,
+        message_id: str,
+        user_id: str,
+        delete_for: str = "me",  # "everyone" or "me"
+    ) -> dict:
+        """
+        Delete a message.
+        - "everyone": Only sender, within 1 hour
+        - "me": Any participant
+        """
+        db = await get_db()
+        
+        # Validate IDs
+        if not ObjectId.is_valid(conversation_id) or not ObjectId.is_valid(message_id):
+            raise HTTPException(
+                status_code=status.HTTP_400_BAD_REQUEST,
+                detail="Invalid ID format"
+            )
+        
+        # Get the message
+        message_doc = await db.messages.find_one({"_id": ObjectId(message_id)})
+        if not message_doc:
+            raise HTTPException(
+                status_code=status.HTTP_404_NOT_FOUND,
+                detail="Message not found"
+            )
+        
+        # Verify message is in the correct conversation
+        if message_doc.get("conversation_id") != conversation_id:
+            raise HTTPException(
+                status_code=status.HTTP_400_BAD_REQUEST,
+                detail="Message does not belong to this conversation"
+            )
+        
+        # Verify user is a participant
+        conversation = await db.conversations.find_one({"_id": ObjectId(conversation_id)})
+        if not conversation or user_id not in conversation.get("participants", []):
+            raise HTTPException(
+                status_code=status.HTTP_403_FORBIDDEN,
+                detail="You are not a participant in this conversation"
+            )
+        
+        now = datetime.utcnow()
+        
+        if delete_for == "everyone":
+            # Only sender can delete for everyone
+            if message_doc.get("sender_id") != user_id:
+                raise HTTPException(
+                    status_code=status.HTTP_403_FORBIDDEN,
+                    detail="Only the sender can delete for everyone"
+                )
+            
+            # Check 1-hour delete window
+            created_at = message_doc.get("created_at")
+            if created_at:
+                hours_since = (datetime.utcnow() - created_at).total_seconds() / 3600
+                if hours_since > 1:
+                    raise HTTPException(
+                        status_code=status.HTTP_400_BAD_REQUEST,
+                        detail="Delete-for-everyone window expired (1 hour)"
+                    )
+            
+            # Mark as deleted for everyone
+            await db.messages.update_one(
+                {"_id": ObjectId(message_id)},
+                {
+                    "$set": {
+                        "is_deleted": True,
+                        "deleted_at": now,
+                    }
+                }
+            )
+            
+            # Broadcast message_deleted event to all participants via WebSocket
+            try:
+                from app.routes.websocket_chat import chat_manager
+                
+                delete_event = {
+                    "action": "message_deleted",
+                    "conversation_id": conversation_id,
+                    "message_id": message_id,
+                    "deleted_for": "everyone",
+                    "deleted_at": now.isoformat(),
+                    "deleted_by": user_id,
+                }
+                
+                await chat_manager.broadcast_to_conversation(
+                    conversation_id,
+                    conversation.get("participants", []),
+                    delete_event
+                )
+                
+                logger.info(f"Broadcasted message_deleted event for message {message_id}")
+            except Exception as e:
+                logger.warning(f"Failed to broadcast message_deleted event: {e}")
+            
+            logger.info(f"Message {message_id} deleted for everyone by {user_id}")
+            
+            return {
+                "success": True,
+                "message_id": message_id,
+                "deleted_for": "everyone",
+                "deleted_at": now.isoformat(),
+            }
+        
+        else:  # delete_for == "me"
+            # Add user to deleted_for list
+            await db.messages.update_one(
+                {"_id": ObjectId(message_id)},
+                {"$addToSet": {"deleted_for": user_id}}
+            )
+            
+            logger.info(f"Message {message_id} deleted for {user_id} only")
+            
+            return {
+                "success": True,
+                "message_id": message_id,
+                "deleted_for": "me",
+                "deleted_at": now.isoformat(),
+            }
+    
+    async def add_reaction(
+        self,
+        conversation_id: str,
+        message_id: str,
+        user_id: str,
+        emoji: str,
+    ) -> dict:
+        """Add an emoji reaction to a message."""
+        db = await get_db()
+        
+        # Validate IDs
+        if not ObjectId.is_valid(conversation_id) or not ObjectId.is_valid(message_id):
+            raise HTTPException(
+                status_code=status.HTTP_400_BAD_REQUEST,
+                detail="Invalid ID format"
+            )
+        
+        # Get the message
+        message_doc = await db.messages.find_one({"_id": ObjectId(message_id)})
+        if not message_doc:
+            raise HTTPException(
+                status_code=status.HTTP_404_NOT_FOUND,
+                detail="Message not found"
+            )
+        
+        # Verify message is in the correct conversation
+        if message_doc.get("conversation_id") != conversation_id:
+            raise HTTPException(
+                status_code=status.HTTP_400_BAD_REQUEST,
+                detail="Message does not belong to this conversation"
+            )
+        
+        # Verify user is a participant
+        conversation = await db.conversations.find_one({"_id": ObjectId(conversation_id)})
+        if not conversation or user_id not in conversation.get("participants", []):
+            raise HTTPException(
+                status_code=status.HTTP_403_FORBIDDEN,
+                detail="You are not a participant in this conversation"
+            )
+        
+        # Cannot react to deleted messages
+        if message_doc.get("is_deleted"):
+            raise HTTPException(
+                status_code=status.HTTP_400_BAD_REQUEST,
+                detail="Cannot react to a deleted message"
+            )
+        
+        # Add reaction
+        await db.messages.update_one(
+            {"_id": ObjectId(message_id)},
+            {"$addToSet": {f"reactions.{emoji}": user_id}}
+        )
+        
+        # Broadcast reaction_added event to all participants via WebSocket
+        try:
+            from app.routes.websocket_chat import chat_manager
+            
+            reaction_event = {
+                "action": "reaction_added",
+                "conversation_id": conversation_id,
+                "message_id": message_id,
+                "emoji": emoji,
+                "user_id": user_id,
+            }
+            
+            await chat_manager.broadcast_to_conversation(
+                conversation_id,
+                conversation.get("participants", []),
+                reaction_event
+            )
+            
+            logger.info(f"Broadcasted reaction_added event for message {message_id}")
+        except Exception as e:
+            logger.warning(f"Failed to broadcast reaction_added event: {e}")
+        
+        logger.info(f"Reaction {emoji} added to message {message_id} by {user_id}")
+        
+        return {
+            "success": True,
+            "message_id": message_id,
+            "emoji": emoji,
+            "user_id": user_id,
+        }
+    
+    async def remove_reaction(
+        self,
+        conversation_id: str,
+        message_id: str,
+        user_id: str,
+        emoji: str,
+    ) -> dict:
+        """Remove an emoji reaction from a message."""
+        db = await get_db()
+        
+        # Validate IDs
+        if not ObjectId.is_valid(conversation_id) or not ObjectId.is_valid(message_id):
+            raise HTTPException(
+                status_code=status.HTTP_400_BAD_REQUEST,
+                detail="Invalid ID format"
+            )
+        
+        # Get the message
+        message_doc = await db.messages.find_one({"_id": ObjectId(message_id)})
+        if not message_doc:
+            raise HTTPException(
+                status_code=status.HTTP_404_NOT_FOUND,
+                detail="Message not found"
+            )
+        
+        # Verify message is in the correct conversation
+        if message_doc.get("conversation_id") != conversation_id:
+            raise HTTPException(
+                status_code=status.HTTP_400_BAD_REQUEST,
+                detail="Message does not belong to this conversation"
+            )
+        
+        # Verify user is a participant
+        conversation = await db.conversations.find_one({"_id": ObjectId(conversation_id)})
+        if not conversation or user_id not in conversation.get("participants", []):
+            raise HTTPException(
+                status_code=status.HTTP_403_FORBIDDEN,
+                detail="You are not a participant in this conversation"
+            )
+        
+        # Remove reaction
+        await db.messages.update_one(
+            {"_id": ObjectId(message_id)},
+            {"$pull": {f"reactions.{emoji}": user_id}}
+        )
+        
+        # Clean up empty reaction arrays
+        updated_msg = await db.messages.find_one({"_id": ObjectId(message_id)})
+        if updated_msg:
+            reactions = updated_msg.get("reactions", {})
+            if emoji in reactions and len(reactions[emoji]) == 0:
+                await db.messages.update_one(
+                    {"_id": ObjectId(message_id)},
+                    {"$unset": {f"reactions.{emoji}": ""}}
+                )
+        
+        # Broadcast reaction_removed event to all participants via WebSocket
+        try:
+            from app.routes.websocket_chat import chat_manager
+            
+            reaction_event = {
+                "action": "reaction_removed",
+                "conversation_id": conversation_id,
+                "message_id": message_id,
+                "emoji": emoji,
+                "user_id": user_id,
+            }
+            
+            await chat_manager.broadcast_to_conversation(
+                conversation_id,
+                conversation.get("participants", []),
+                reaction_event
+            )
+            
+            logger.info(f"Broadcasted reaction_removed event for message {message_id}")
+        except Exception as e:
+            logger.warning(f"Failed to broadcast reaction_removed event: {e}")
+        
+        logger.info(f"Reaction {emoji} removed from message {message_id} by {user_id}")
+        
+        return {
+            "success": True,
+            "message_id": message_id,
+            "emoji": emoji,
+            "user_id": user_id,
+        }
+    
+    async def clear_chat(
+        self,
+        conversation_id: str,
+        user_id: str,
+    ) -> dict:
+        """
+        Clear all messages in a conversation for the current user only.
+        
+        Stores a cleared_at timestamp on the conversation document.
+        Messages with created_at <= cleared_at won't be shown to this user,
+        even after logout/login or on a new device.
+        Other participants still see all messages.
+        """
+        db = await get_db()
+        
+        # Validate ID
+        if not ObjectId.is_valid(conversation_id):
+            raise HTTPException(
+                status_code=status.HTTP_400_BAD_REQUEST,
+                detail="Invalid conversation ID format"
+            )
+        
+        # Verify user is a participant
+        conversation = await db.conversations.find_one({"_id": ObjectId(conversation_id)})
+        if not conversation or user_id not in conversation.get("participants", []):
+            raise HTTPException(
+                status_code=status.HTTP_403_FORBIDDEN,
+                detail="You are not a participant in this conversation"
+            )
+        
+        now = datetime.utcnow()
+        
+        # Store cleared_at timestamp on conversation for persistent filtering
+        # This is the key change - ensures clear persists across sessions/devices
+        await db.conversations.update_one(
+            {"_id": ObjectId(conversation_id)},
+            {"$set": {f"cleared_at.{user_id}": now}}
+        )
+        
+        # Also mark existing messages with deleted_for (backwards compatibility)
+        # This provides immediate UI response and works with existing code
+        result = await db.messages.update_many(
+            {"conversation_id": conversation_id},
+            {"$addToSet": {"deleted_for": user_id}}
+        )
+        
+        logger.info(f"Chat {conversation_id} cleared for {user_id} at {now.isoformat()} ({result.modified_count} messages)")
+        
+        return {
+            "success": True,
+            "conversation_id": conversation_id,
+            "cleared_count": result.modified_count,
+            "cleared_at": now.isoformat(),
+        }

app/services/conversation_parts/ai_mixin.py

@@ -0,0 +1,223 @@
+import logging
+import asyncio
+from typing import Optional
+from datetime import datetime
+from bson import ObjectId
+
+from app.models.message import Message
+from app.routes.websocket_chat import chat_manager
+
+logger = logging.getLogger(__name__)
+
+class ConversationAIMixin:
+    """
+    Handling AI (AIDA) interactions
+    """
+    
+    async def _process_aida_response(
+        self,
+        db,
+        conversation_id: str,
+        conversation: dict,
+        user_id: str,
+        user_message: str,
+        user_message_id: str,
+        reply_context: Optional[dict] = None,
+        audio_url: Optional[str] = None,
+    ):
+        """
+        Process user message through AIDA AI brain and send response.
+        Runs asynchronously so user sees their message immediately.
+        """
+        try:
+            logger.info(f"Processing AIDA response for user {user_id}")
+            
+            # 1. Handle Voice Transcription if needed
+            if audio_url:
+                try:
+                    from app.services.voice_service import voice_service
+                    logger.info(f"Transcribing voice message: {audio_url}")
+                    transcript, lang = await voice_service.transcribe_audio(audio_url)
+                    
+                    if transcript:
+                        user_message = transcript
+                        logger.info(f"Voice transcribed: '{user_message}' ({lang})")
+                        
+                        # Optionally update the original message with transcript?
+                        # For now, we just let AIDA know the content
+                    else:
+                        logger.warning("Empty transcription result")
+                        user_message = "(Inaudible voice message)"
+                        
+                except Exception as e:
+                    logger.error(f"Transcription failed: {e}")
+                    # Don't crash, just let AIDA handle failure
+                    user_message = "(Voice message could not be transcribed)"
+            
+            # Import specialized DM brain
+            from app.ai.agent.dm_brain import DmBrain
+            
+            # Build context message with reply info
+            context_message = user_message
+            if reply_context:
+                property_card = reply_context.get("property_card")
+                replied_content = reply_context.get("message_content", "")
+                metadata = reply_context.get("metadata", {})
+                alert_title = metadata.get("alert_title") if metadata else None
+                
+                if property_card:
+                    listing_id = property_card.get("listing_id")
+                    listing_title = property_card.get("title")
+                    listing_location = property_card.get("location")
+                    
+                    context_message = f"""{user_message}
+
+[System Context: User is replying to property listing:
+- ID: {listing_id}
+- Title: {listing_title}
+- Location: {listing_location}
+
+When user says "this", "it", "this property", they mean this listing.
+If user says they found what they wanted, use delete_alert tool with location matching this listing.]"""
+                
+                elif alert_title:
+                    context_message = f"""{user_message}
+
+[System Context: User is replying to alert notification for: "{alert_title}"
+If user says they found what they wanted or wants to stop notifications, use delete_alert tool.]"""
+                
+                elif replied_content:
+                    context_message = f"""{user_message}
+
+[System Context: User is replying to AIDA's previous message: "{replied_content}"]"""
+            
+            # Call specialized DM brain (architecture separation)
+            brain = DmBrain()
+            result = await brain.process(
+                message=context_message,
+                user_id=user_id,
+                source="dm",
+            )
+            
+            # Extract response text and metadata
+            response_text = result.get("text", "I'm sorry, I couldn't process that. Please try again.")
+            response_metadata = result.get("metadata", {})
+            
+            # Check for property cards or alert results
+            property_card = None
+            if response_metadata.get("property_card"):
+                property_card = response_metadata.get("property_card")
+            
+            # ============================================================
+            # VOICE RESPONSE: If user sent a voice note, AIDA responds with voice too
+            # ============================================================
+            aida_audio_url = None
+            aida_audio_duration = None
+            message_type = "text"  # Default to text
+            
+            if audio_url:
+                # User sent voice, so AIDA should respond with voice
+                try:
+                    from app.services.voice_service import voice_service
+                    
+                    # Determine language from transcription or default to English
+                    response_language = "en"
+                    
+                    # Generate AIDA's voice response
+                    logger.info(f"Generating AIDA voice response for DM...")
+                    voice_result = await voice_service.generate_aida_voice_response(
+                        text=response_text,
+                        language=response_language
+                    )
+                    
+                    if voice_result:
+                        aida_audio_url = voice_result.get("audio_url")
+                        aida_audio_duration = voice_result.get("duration")
+                        message_type = "voice"  # Change message type to voice
+                        logger.info(f"✅ AIDA voice response generated: {aida_audio_url}")
+                        
+                except Exception as e:
+                    logger.error(f"Failed to generate AIDA voice response: {e}")
+                    # Fall back to text response
+                    message_type = "text"
+            
+            # Send AIDA's response as a message in the conversation
+            aida_message_doc = Message.create_document(
+                conversation_id=conversation_id,
+                sender_id="AIDA_BOT",
+                sender_name="AIDA",
+                sender_avatar="https://imagedelivery.net/0utJlkqgAVuawL5OpMWxgw/3922956f-b69d-4cb3-97b9-3a185abec900/public",
+                message_type=message_type,
+                content=response_text,
+                property_card=property_card,
+                replied_to_message_id=user_message_id if reply_context else None,
+                replied_to_content=user_message if reply_context else None,
+                replied_to_sender="User" if reply_context else None,
+            )
+            
+            # Add voice-specific fields if this is a voice response
+            if aida_audio_url:
+                aida_message_doc["audio_url"] = aida_audio_url
+                aida_message_doc["audio_duration"] = aida_audio_duration
+            
+            # Add metadata for rich content (alert results, etc.)
+            if response_metadata:
+                aida_message_doc["metadata"] = response_metadata
+            
+            # Insert AIDA's message
+            aida_result = await db.messages.insert_one(aida_message_doc)
+            aida_message_id = str(aida_result.inserted_id)
+            aida_message_doc["_id"] = aida_result.inserted_id
+            
+            logger.info(f"AIDA response {aida_message_id} sent in conversation {conversation_id}")
+            
+            # Update conversation's last_message (AIDA's response)
+            update_data = {
+                "last_message": {
+                    "text": response_text[:100] if response_text else "[AI Response]",
+                    "sender_id": "AIDA_BOT",
+                    "timestamp": aida_message_doc["created_at"],
+                },
+                "updated_at": datetime.utcnow(),
+            }
+            
+            # Increment unread count for user (AIDA just sent a message)
+            update_data[f"unread_count.{user_id}"] = conversation.get("unread_count", {}).get(user_id, 0) + 1
+            
+            await db.conversations.update_one(
+                {"_id": ObjectId(conversation_id)},
+                {"$set": update_data}
+            )
+            
+            # Broadcast AIDA's response via WebSocket
+            try:
+                formatted_message = Message.format_response(aida_message_doc)
+                broadcast_message = {
+                    "action": "new_message",
+                    "conversation_id": conversation_id,
+                    "message": formatted_message,
+                }
+                await chat_manager.broadcast_to_conversation(
+                    conversation_id,
+                    conversation["participants"],
+                    broadcast_message
+                )
+                logger.info(f"AIDA response broadcast via WebSocket")
+            except Exception as e:
+                logger.warning(f"Failed to broadcast AIDA response: {e}")
+                
+        except Exception as e:
+            logger.error(f"Error processing AIDA response: {e}")
+            # Optionally send error message to user
+            try:
+                error_message_doc = Message.create_document(
+                    conversation_id=conversation_id,
+                    sender_id="AIDA_BOT",
+                    sender_name="AIDA",
+                    sender_avatar="https://imagedelivery.net/0utJlkqgAVuawL5OpMWxgw/3922956f-b69d-4cb3-97b9-3a185abec900/public",
+                    message_type="text",
+                    content="I'm having trouble processing that right now. Please try again in a moment. 😅",
+                )
+                await db.messages.insert_one(error_message_doc)
+            except:
+                pass

app/services/conversation_parts/crud_mixin.py

@@ -0,0 +1,485 @@
+import logging
+from datetime import datetime
+from typing import Optional, List, Dict, Any, Tuple
+from bson import ObjectId
+from fastapi import HTTPException, status
+
+from app.database import get_db
+from app.models.conversation import Conversation
+from app.models.listing import Listing
+
+logger = logging.getLogger(__name__)
+
+class ConversationCRUDMixin:
+    """Basic CRUD operations for conversations"""
+    
+    async def start_or_get_conversation(
+        self,
+        listing_id: str,
+        current_user_id: str,
+        initial_message: Optional[str] = None,
+    ) -> dict:
+        """
+        Start a new conversation or get existing one between two users.
+        """
+        db = await get_db()
+        
+        # 1. Validate listing exists and get owner
+        if not ObjectId.is_valid(listing_id):
+            raise HTTPException(
+                status_code=status.HTTP_400_BAD_REQUEST,
+                detail="Invalid listing ID format"
+            )
+        
+        listing = await db.listings.find_one({"_id": ObjectId(listing_id)})
+        if not listing:
+            raise HTTPException(
+                status_code=status.HTTP_404_NOT_FOUND,
+                detail="Listing not found"
+            )
+        
+        owner_id = listing.get("user_id")
+        if not owner_id:
+            raise HTTPException(
+                status_code=status.HTTP_400_BAD_REQUEST,
+                detail="Listing has no owner"
+            )
+        
+        # 2. Check if user is trying to message themselves
+        if owner_id == current_user_id:
+            return {
+                "success": False,
+                "error": "self_chat",
+                "message": "Oops! You can't chat with yourself. This is your own listing! 😊",
+            }
+        
+        # 3. Prepare property card data (always needed for frontend prompt)
+        property_card = {
+            "listing_id": listing_id,
+            "title": listing.get("title", ""),
+            "price": listing.get("price", 0),
+            "currency": listing.get("currency", "NGN"),
+            "bedrooms": listing.get("bedrooms", 0),
+            "bathrooms": listing.get("bathrooms", 0),
+            "location": listing.get("location", ""),
+            "image_url": listing.get("images", [None])[0] if listing.get("images") else None,
+            "listing_type": listing.get("listing_type", ""),
+        }
+        
+        # 4. Check if conversation already exists BETWEEN THESE TWO USERS (regardless of listing)
+        participants = sorted([owner_id, current_user_id])  # Sort for consistent ordering
+        participants_key = "::".join(participants)
+        
+        existing_conversation = await db.conversations.find_one({
+            "participants_key": participants_key
+        })
+        
+        if existing_conversation:
+            logger.info(f"Found existing conversation between users: {existing_conversation['_id']}")
+            return {
+                "success": True,
+                "is_new": False,
+                "conversation": Conversation.format_response(existing_conversation),
+                "property_card": property_card,  # Always include so frontend can prompt to send
+            }
+        
+        # 5. Create new conversation (first time these two users chat)
+        conversation_doc = Conversation.create_document(
+            listing_id=listing_id,  # Store the first listing that started the conversation
+            participants=participants,
+            listing_title=listing.get("title", "Property"),
+            listing_image=listing.get("images", [None])[0] if listing.get("images") else None,
+        )
+        
+        result = await db.conversations.insert_one(conversation_doc)
+        conversation_id = str(result.inserted_id)
+        
+        logger.info(f"Created new conversation: {conversation_id}")
+        
+        # 6. Get conversation with ID
+        conversation_doc["_id"] = result.inserted_id
+        
+        return {
+            "success": True,
+            "is_new": True,
+            "conversation": Conversation.format_response(conversation_doc),
+            "property_card": property_card,  # Frontend will show confirmation dialog
+        }
+
+    async def start_or_get_aida_conversation(
+        self,
+        current_user_id: str,
+    ) -> dict:
+        """
+        Start or get an AIDA DM conversation.
+        """
+        db = await get_db()
+        
+        AIDA_BOT_ID = "AIDA_BOT"
+        LEGACY_AIDA_ID = "ai_assistant"  # Legacy ID that might exist in old conversations
+        
+        # 1. Find existing conversation between AIDA and User
+        participants = sorted([AIDA_BOT_ID, current_user_id])
+        participants_key = "::".join(participants)
+        
+        # First, try newer format with participants_key
+        existing_conversation = await db.conversations.find_one({
+            "participants_key": participants_key
+        })
+        
+        # Fallback: Try legacy ai_assistant ID
+        if not existing_conversation:
+            legacy_participants = sorted([LEGACY_AIDA_ID, current_user_id])
+            legacy_key = "::".join(legacy_participants)
+            existing_conversation = await db.conversations.find_one({
+                "participants_key": legacy_key
+            })
+            if existing_conversation:
+                logger.info(f"Found legacy AIDA conversation with ai_assistant ID")
+        
+        # Fallback: Try querying by participants array
+        if not existing_conversation:
+            existing_conversation = await db.conversations.find_one({
+                "participants": {"$all": [AIDA_BOT_ID, current_user_id]}
+            })
+            if existing_conversation:
+                logger.info(f"Found AIDA conversation by participants array")
+        
+        # Fallback: Try legacy array format
+        if not existing_conversation:
+            existing_conversation = await db.conversations.find_one({
+                "participants": {"$all": [LEGACY_AIDA_ID, current_user_id]}
+            })
+            if existing_conversation:
+                logger.info(f"Found legacy AIDA conversation by participants array")
+        
+        if existing_conversation:
+            conv_id = str(existing_conversation["_id"])
+            logger.info(f"Found existing AIDA conversation: {conv_id}")
+            
+            # Determine which AIDA ID was used
+            aida_id = AIDA_BOT_ID if AIDA_BOT_ID in existing_conversation.get("participants", []) else LEGACY_AIDA_ID
+            actual_participants = [aida_id, current_user_id]
+            
+            # Enrich with participants data
+            enriched_participants, other_participant = await self._enrich_participants(
+                db, actual_participants, current_user_id
+            )
+            
+            conv_response = Conversation.format_response(existing_conversation)
+            conv_response["participants"] = enriched_participants
+            conv_response["other_participant"] = other_participant
+            
+            return {
+                "success": True,
+                "is_new": False,
+                "conversation": conv_response,
+            }
+        
+        # 2. Create new AIDA conversation
+        conversation_doc = Conversation.create_document(
+            listing_id="system",  # Generic ID for system chats
+            participants=participants,
+            listing_title="AIDA Assistant",
+            listing_image=None
+        )
+        
+        try:
+            result = await db.conversations.insert_one(conversation_doc)
+            conv_id = str(result.inserted_id)
+            conversation_doc["_id"] = result.inserted_id
+            logger.info(f"Created new AIDA conversation: {conv_id}")
+        except Exception as e:
+            error_str = str(e)
+            logger.error(f"Insert failed with error: {error_str}")
+            
+            if "duplicate key" in error_str.lower() or "E11000" in error_str:
+                 # Race condition fallback
+                 existing_conversation = await db.conversations.find_one({
+                     "participants": {"$all": [AIDA_BOT_ID, current_user_id]}
+                 })
+                 
+                 if existing_conversation:
+                     enriched_participants, other_participant = await self._enrich_participants(
+                         db, existing_conversation.get("participants", participants), current_user_id
+                     )
+                     conv_response = Conversation.format_response(existing_conversation)
+                     conv_response["participants"] = enriched_participants
+                     return {
+                        "success": True,
+                        "is_new": False,
+                        "conversation": conv_response,
+                     }
+                 else:
+                     raise
+            else:
+                raise
+        
+        # 3. Enrich with participants data
+        enriched_participants, other_participant = await self._enrich_participants(
+            db, participants, current_user_id
+        )
+        
+        conv_response = Conversation.format_response(conversation_doc)
+        conv_response["participants"] = enriched_participants
+        conv_response["other_participant"] = other_participant
+        
+        return {
+            "success": True,
+            "is_new": True,
+            "conversation": conv_response,
+        }
+
+    async def _enrich_participants(self, db, participant_ids: list[str], current_user_id: str = None) -> tuple[list[dict], dict]:
+        """
+        Enrich participant IDs with full user data and online status.
+        Returns (all_participants, other_participant).
+        """
+        from app.services.presence_service import presence_service
+        
+        AIDA_BOT_ID = "AIDA_BOT"
+        AIDA_PROFILE = {
+            "id": AIDA_BOT_ID,
+            "name": "AIDA",
+            "profile_picture": "https://imagedelivery.net/0utJlkqgAVuawL5OpMWxgw/3922956f-b69d-4cb3-97b9-3a185abec900/public",
+            "is_online": True,  # AIDA is always online
+            "last_seen": None,
+        }
+        
+        participants = []
+        other_participant = None
+        
+        # AIDA IDs to handle (both new and legacy)
+        AIDA_IDS = {AIDA_BOT_ID, "ai_assistant"}
+        
+        # Batch fetch all users (skip AIDA IDs)
+        users_map = {}
+        real_user_ids = [uid for uid in participant_ids if uid not in AIDA_IDS]
+        
+        for uid in real_user_ids:
+            if ObjectId.is_valid(uid):
+                user = await db.users.find_one({"_id": ObjectId(uid)})
+                if user:
+                    users_map[uid] = user
+        
+        # Get online statuses (only for real users)
+        online_statuses = await presence_service.get_bulk_status(real_user_ids) if real_user_ids else {}
+        
+        # Build enriched participants
+        for uid in participant_ids:
+            # Handle AIDA (both AIDA_BOT and ai_assistant) specially
+            if uid in AIDA_IDS:
+                participant = AIDA_PROFILE.copy()
+            else:
+                user = users_map.get(uid)
+                status_data = online_statuses.get(uid, {"is_online": False, "last_seen": None})
+                
+                participant = {
+                    "id": uid,
+                    "name": f"{user.get('firstName', '')} {user.get('lastName', '')}".strip() if user else "Unknown",
+                    "profile_picture": user.get("profilePicture") if user else None,
+                    "is_online": status_data.get("is_online", False),
+                    "last_seen": status_data.get("last_seen"),
+                }
+            
+            participants.append(participant)
+            
+            # Track the OTHER participant (not current user)
+            if current_user_id and uid != current_user_id:
+                other_participant = participant
+        
+        return participants, other_participant
+
+    async def get_user_conversations(
+        self,
+        user_id: str,
+    ) -> list[dict]:
+        """
+        Get all conversations for a user with ENRICHED participant data.
+        OPTIMIZED: Uses batch fetching to eliminate N+1 queries.
+        """
+        db = await get_db()
+        from app.services.presence_service import presence_service
+        
+        # 1. Fetch all conversation docs (single query)
+        conversations_cursor = db.conversations.find({
+            "participants": user_id
+        }).sort("updated_at", -1)
+        
+        conversations_docs = await conversations_cursor.to_list(100)
+        
+        if not conversations_docs:
+            return []
+            
+        # 2. Collect ALL distinct participant IDs
+        all_participant_ids = set()
+        for doc in conversations_docs:
+            all_participant_ids.update(doc.get("participants", []))
+            
+        # Filter out AIDA bots and invalid IDs
+        AIDA_IDS = {"AIDA_BOT", "ai_assistant"}
+        real_user_ids = {uid for uid in all_participant_ids if uid not in AIDA_IDS and ObjectId.is_valid(uid)}
+        
+        # 3. Batch fetch all User profiles (single query)
+        users = await db.users.find({
+            "_id": {"$in": [ObjectId(uid) for uid in real_user_ids]}
+        }).to_list(None)
+        
+        users_map = {str(u["_id"]): u for u in users}
+        
+        # 4. Batch fetch online statuses (single redis pipeline)
+        online_statuses = await presence_service.get_bulk_status(list(real_user_ids))
+        
+        # 5. Build response objects
+        AIDA_PROFILE = {
+            "id": "AIDA_BOT",
+            "name": "AIDA",
+            "profile_picture": "https://imagedelivery.net/0utJlkqgAVuawL5OpMWxgw/3922956f-b69d-4cb3-97b9-3a185abec900/public",
+            "is_online": True,
+            "last_seen": None,
+        }
+        
+        conversations = []
+        for doc in conversations_docs:
+            conv_id = str(doc["_id"])
+            participant_ids = doc.get("participants", [])
+            
+            enriched_participants = []
+            other_participant = None
+            
+            for uid in participant_ids:
+                # Handle AIDA
+                if uid in AIDA_IDS:
+                    participant = AIDA_PROFILE.copy()
+                else:
+                    # Handle Real Users
+                    user = users_map.get(uid)
+                    status_data = online_statuses.get(uid, {"is_online": False, "last_seen": None})
+                    
+                    participant = {
+                        "id": uid,
+                        "name": f"{user.get('firstName', '')} {user.get('lastName', '')}".strip() if user else "Unknown",
+                        "profile_picture": user.get("profilePicture") if user else None,
+                        "is_online": status_data.get("is_online", False),
+                        "last_seen": status_data.get("last_seen"),
+                    }
+                
+                enriched_participants.append(participant)
+                
+                if uid != user_id:
+                    other_participant = participant
+            
+            # Get unread count
+            unread_counts = doc.get("unread_count", {})
+            user_unread = unread_counts.get(user_id, 0) if isinstance(unread_counts, dict) else 0
+            
+            conversations.append({
+                "id": conv_id,
+                "listing_id": doc.get("listing_id", ""),
+                "listing_title": doc.get("listing_title", ""),
+                "listing_image": doc.get("listing_image"),
+                "participants": enriched_participants,
+                "other_participant": other_participant,
+                "last_message": doc.get("last_message", {}),
+                "unread_count": user_unread,
+                "status": doc.get("status", "active"),
+                "created_at": doc.get("created_at"),
+                "updated_at": doc.get("updated_at"),
+            })
+        
+        logger.info(f"OPTIMIZED: Found {len(conversations)} conversations for {user_id}")
+        return conversations
+
+    async def get_conversation_by_id(
+        self,
+        conversation_id: str,
+        current_user_id: str,
+        include_messages: bool = False,
+        mark_as_read: bool = False,
+        message_limit: int = 50,
+    ) -> dict:
+        """
+        Get a single conversation with optional messages.
+        """
+        db = await get_db()
+        
+        # Validate conversation ID
+        if not ObjectId.is_valid(conversation_id):
+            raise HTTPException(
+                status_code=status.HTTP_400_BAD_REQUEST,
+                detail="Invalid conversation ID format"
+            )
+        
+        # Fetch conversation
+        doc = await db.conversations.find_one({"_id": ObjectId(conversation_id)})
+        if not doc:
+            raise HTTPException(
+                status_code=status.HTTP_404_NOT_FOUND,
+                detail="Conversation not found"
+            )
+        
+        # Verify user is participant
+        if current_user_id not in doc.get("participants", []):
+            raise HTTPException(
+                status_code=status.HTTP_403_FORBIDDEN,
+                detail="You are not a participant in this conversation"
+            )
+        
+        # Enrich participants
+        participant_ids = doc.get("participants", [])
+        enriched_participants, other_participant = await self._enrich_participants(
+            db, participant_ids, current_user_id=current_user_id
+        )
+        
+        # Get unread count for current user
+        unread_counts = doc.get("unread_count", {})
+        user_unread = unread_counts.get(current_user_id, 0) if isinstance(unread_counts, dict) else 0
+        
+        # Build conversation response
+        conversation_data = {
+            "id": str(doc["_id"]),
+            "listing_id": doc.get("listing_id", ""),
+            "listing_title": doc.get("listing_title", ""),
+            "listing_image": doc.get("listing_image"),
+            "participants": enriched_participants,
+            "other_participant": other_participant,
+            "last_message": doc.get("last_message", {}),
+            "unread_count": user_unread,
+            "status": doc.get("status", "active"),
+            "created_at": doc.get("created_at"),
+            "updated_at": doc.get("updated_at"),
+        }
+        
+        result = {
+            "conversation": conversation_data,
+            "messages": [],
+            "has_more": False,
+            "total_messages": 0,
+        }
+        
+        # Include messages if requested
+        if include_messages:
+            # Requires MessageService mixin
+            messages = await self.get_conversation_messages(
+                conversation_id=conversation_id,
+                current_user_id=current_user_id,
+                limit=message_limit,
+            )
+            
+            # Count total messages
+            total_count = await db.messages.count_documents({
+                "conversation_id": conversation_id,
+                "deleted_for": {"$ne": current_user_id},
+            })
+            
+            result["messages"] = messages
+            result["has_more"] = total_count > len(messages)
+            result["total_messages"] = total_count
+        
+        # Mark as read if requested
+        if mark_as_read:
+            # Requires MessageService mixin
+            await self.mark_as_read(conversation_id, current_user_id)
+            result["conversation"]["unread_count"] = 0
+        
+        return result

app/services/conversation_parts/message_mixin.py

@@ -0,0 +1,334 @@
+import logging
+import asyncio
+from datetime import datetime
+from typing import Optional
+from bson import ObjectId
+from fastapi import HTTPException, status
+
+from app.database import get_db
+from app.models.message import Message
+from app.routes.websocket_chat import chat_manager
+
+logger = logging.getLogger(__name__)
+
+class ConversationMessageMixin:
+    """
+    Handling message operations: Send, Get, Mark Read
+    """
+
+    async def get_conversation_messages(
+        self,
+        conversation_id: str,
+        current_user_id: str,
+        limit: int = 50,
+        before_id: Optional[str] = None,
+    ) -> list[dict]:
+        """
+        Get messages for a conversation with pagination.
+        """
+        db = await get_db()
+        
+        # Validate conversation ID
+        if not ObjectId.is_valid(conversation_id):
+            raise HTTPException(
+                status_code=status.HTTP_400_BAD_REQUEST,
+                detail="Invalid conversation ID format"
+            )
+        
+        # Verify user is participant
+        conversation = await db.conversations.find_one({"_id": ObjectId(conversation_id)})
+        if not conversation:
+            raise HTTPException(
+                status_code=status.HTTP_404_NOT_FOUND,
+                detail="Conversation not found"
+            )
+        
+        if current_user_id not in conversation.get("participants", []):
+            raise HTTPException(
+                status_code=status.HTTP_403_FORBIDDEN,
+                detail="You are not a participant in this conversation"
+            )
+        
+        # Check if user has cleared the chat - get their cleared_at timestamp
+        user_cleared_at = conversation.get("cleared_at", {}).get(current_user_id)
+        
+        # Build query - exclude messages deleted specifically for this user
+        query = {
+            "conversation_id": conversation_id,
+            "deleted_for": {"$ne": current_user_id},  # Exclude messages deleted for this user
+        }
+        
+        # Filter out messages sent before user cleared the chat
+        # This ensures persistence across logout/login and new devices
+        if user_cleared_at:
+            query["created_at"] = {"$gt": user_cleared_at}
+        
+        if before_id and ObjectId.is_valid(before_id):
+            if "created_at" in query:
+                # Combine with existing created_at filter
+                query["$and"] = [
+                    {"created_at": query.pop("created_at")},
+                    {"_id": {"$lt": ObjectId(before_id)}}
+                ]
+            else:
+                query["_id"] = {"$lt": ObjectId(before_id)}
+        
+        # Get messages - sort ascending (oldest first, newest last)
+        # This gives chronological order: oldest at top, newest at bottom
+        cursor = db.messages.find(query).sort("created_at", 1).limit(limit)
+        
+        messages = []
+        async for doc in cursor:
+            # Pass user_id for proper filtering in format_response
+            formatted = Message.format_response(doc, for_user_id=current_user_id)
+            if formatted:  # Only add if not None (handles edge cases)
+                messages.append(formatted)
+        
+        logger.info(f"Found {len(messages)} messages for conversation {conversation_id}")
+        
+        return messages
+    
+    async def send_message(
+        self,
+        conversation_id: str,
+        current_user_id: str,
+        current_user_name: str,
+        current_user_avatar: Optional[str],
+        message_type: str,
+        content: Optional[str] = None,
+        media: Optional[dict] = None,
+        property_card: Optional[dict] = None,
+        replied_to_message_id: Optional[str] = None,
+        replied_to_content: Optional[str] = None,
+        replied_to_sender: Optional[str] = None,
+    ) -> dict:
+        """
+        Send a message in a conversation.
+        """
+        db = await get_db()
+        
+        # Validate conversation
+        if not ObjectId.is_valid(conversation_id):
+            raise HTTPException(
+                status_code=status.HTTP_400_BAD_REQUEST,
+                detail="Invalid conversation ID format"
+            )
+        
+        conversation = await db.conversations.find_one({"_id": ObjectId(conversation_id)})
+        if not conversation:
+            raise HTTPException(
+                status_code=status.HTTP_404_NOT_FOUND,
+                detail="Conversation not found"
+            )
+        
+        if current_user_id not in conversation.get("participants", []):
+            raise HTTPException(
+                status_code=status.HTTP_403_FORBIDDEN,
+                detail="You are not a participant in this conversation"
+            )
+        
+        # Auto-detect message type based on content
+        if property_card:
+            message_type = "property_inquiry"  # Override type when property card is present
+        
+        # Create message
+        message_doc = Message.create_document(
+            conversation_id=conversation_id,
+            sender_id=current_user_id,
+            sender_name=current_user_name,
+            sender_avatar=current_user_avatar,
+            message_type=message_type,
+            content=content,
+            media=media,
+            property_card=property_card,
+            replied_to_message_id=replied_to_message_id,
+            replied_to_content=replied_to_content,
+            replied_to_sender=replied_to_sender,
+        )
+        
+        result = await db.messages.insert_one(message_doc)
+        message_id = str(result.inserted_id)
+        
+        # Update conversation's last_message and unread count
+        other_user_id = next(
+            (p for p in conversation["participants"] if p != current_user_id),
+            None
+        )
+        
+        update_data = {
+            "last_message": {
+                "text": content or f"[{message_type}]",
+                "sender_id": current_user_id,
+                "timestamp": message_doc["created_at"],
+            },
+            "updated_at": datetime.utcnow(),
+        }
+        
+        # Increment unread count for other user
+        if other_user_id:
+            update_data[f"unread_count.{other_user_id}"] = conversation.get("unread_count", {}).get(other_user_id, 0) + 1
+        
+        await db.conversations.update_one(
+            {"_id": ObjectId(conversation_id)},
+            {"$set": update_data}
+        )
+        
+        logger.info(f"Message {message_id} sent in conversation {conversation_id}")
+        
+        # Get message with ID
+        message_doc["_id"] = result.inserted_id
+        
+        # Broadcast to all participants via WebSocket (for in-app notifications)
+        try:
+            formatted_message = Message.format_response(message_doc)
+            broadcast_message = {
+                "action": "new_message",
+                "conversation_id": conversation_id,
+                "message": formatted_message,
+            }
+            await chat_manager.broadcast_to_conversation(
+                conversation_id,
+                conversation["participants"],
+                broadcast_message
+            )
+            logger.info(f"WebSocket broadcast sent for message {message_id}")
+        except Exception as e:
+            # Don't fail the request if WebSocket broadcast fails
+            logger.warning(f"Failed to broadcast message via WebSocket: {e}")
+        
+        # NEW: Detect AIDA chat and route to AI brain
+        AIDA_IDS = {"AIDA_BOT", "ai_assistant"}
+        is_aida_chat = any(p in AIDA_IDS for p in conversation.get("participants", []))
+        
+        # Process if AIDA chat, user is not AIDA, and message is TEXT or VOICE
+        # (check content for text, or media for voice)
+        should_process = (
+            is_aida_chat and 
+            current_user_id not in AIDA_IDS and 
+            (
+                (message_type == "text" and content) or 
+                (message_type == "voice" and media and media.get("url"))
+            )
+        )
+        
+        if should_process:
+            try:
+                logger.info(f"AIDA DM detected (type={message_type}), routing to AI brain...")
+                
+                # Extract audio URL for voice messages
+                audio_url = media.get("url") if message_type == "voice" and media else None
+                
+                # Build reply context if replying to a message
+                reply_context = None
+                if replied_to_message_id:
+                    replied_msg = await db.messages.find_one({"_id": ObjectId(replied_to_message_id)})
+                    if replied_msg:
+                        reply_context = {
+                            "message_id": replied_to_message_id,
+                            "message_content": replied_msg.get("content"),
+                            "property_card": replied_msg.get("property_card"),
+                            "metadata": replied_msg.get("metadata"),
+                        }
+                
+                # Call AIDA brain (async, don't block response) (Assumes method exists on self)
+                asyncio.create_task(self._process_aida_response(
+                    db=db,
+                    conversation_id=conversation_id,
+                    conversation=conversation,
+                    user_id=current_user_id,
+                    user_message=content,
+                    user_message_id=message_id,
+                    reply_context=reply_context,
+                    audio_url=audio_url,
+                ))
+                
+            except Exception as e:
+                logger.error(f"Failed to route to AIDA brain: {e}")
+        
+        return {
+            "success": True,
+            "message": Message.format_response(message_doc),
+        }
+
+    async def mark_as_read(
+        self,
+        conversation_id: str,
+        current_user_id: str,
+    ) -> dict:
+        """
+        Mark all messages in a conversation as read for the current user.
+        """
+        db = await get_db()
+        
+        # Validate conversation
+        if not ObjectId.is_valid(conversation_id):
+            raise HTTPException(
+                status_code=status.HTTP_400_BAD_REQUEST,
+                detail="Invalid conversation ID format"
+            )
+        
+        conversation = await db.conversations.find_one({"_id": ObjectId(conversation_id)})
+        if not conversation:
+            raise HTTPException(
+                status_code=status.HTTP_404_NOT_FOUND,
+                detail="Conversation not found"
+            )
+        
+        if current_user_id not in conversation.get("participants", []):
+            raise HTTPException(
+                status_code=status.HTTP_403_FORBIDDEN,
+                detail="You are not a participant in this conversation"
+            )
+        
+        # Mark messages as read and add user to read_by array
+        now = datetime.utcnow()
+        result = await db.messages.update_many(
+            {
+                "conversation_id": conversation_id,
+                "sender_id": {"$ne": current_user_id},  # Not sent by this user
+                "read_by": {"$ne": current_user_id},  # Not already read by this user
+            },
+            {
+                "$set": {
+                    "is_read": True,  # Keep for backwards compatibility
+                    "read_at": now,
+                },
+                "$addToSet": {
+                    "read_by": current_user_id  # Add user to read_by array (no duplicates)
+                }
+            }
+        )
+        
+        logger.info(f"Marked {result.modified_count} messages as read, added {current_user_id} to read_by")
+        
+        # Reset unread count for current user
+        await db.conversations.update_one(
+            {"_id": ObjectId(conversation_id)},
+            {"$set": {f"unread_count.{current_user_id}": 0}}
+        )
+        
+        # Broadcast message_read event to other participants via WebSocket
+        try:
+            read_event = {
+                "action": "message_read",
+                "conversation_id": conversation_id,
+                "read_by": current_user_id,
+                "read_at": now.isoformat(),
+            }
+            
+            # Send to other participants
+            for participant_id in conversation.get("participants", []):
+                if participant_id != current_user_id:
+                    await chat_manager.send_to_user(participant_id, read_event)
+            
+            logger.info(f"Broadcasted message_read event for conversation {conversation_id}")
+        except Exception as e:
+            # Don't fail the request if WebSocket broadcast fails
+            logger.warning(f"Failed to broadcast message_read event: {e}")
+        
+        logger.info(f"Marked messages as read in conversation {conversation_id} for user {current_user_id}")
+        
+        return {
+            "success": True,
+            "message": "Messages marked as read",
+        }

app/services/conversation_service.py

@@ -3,1526 +3,32 @@
 # ============================================================
 
 import logging
-from datetime import datetime
-from typing import Optional
-from bson import ObjectId
-from fastapi import HTTPException, status
 
-from app.database import get_db
-from app.models.conversation import Conversation
-from app.models.message import Message
-from app.models.listing import Listing
-from app.routes.websocket_chat import chat_manager  # For WebSocket notifications
+from app.services.conversation_parts import (
+    ConversationCRUDMixin,
+    ConversationMessageMixin,
+    ConversationActionMixin,
+    ConversationAIMixin
+)
 
 logger = logging.getLogger(__name__)
 
-
-class ConversationService:
-    """Service for handling conversation operations"""
-    
-    async def start_or_get_conversation(
-        self,
-        listing_id: str,
-        current_user_id: str,
-        initial_message: Optional[str] = None,
-    ) -> dict:
-        """
-        Start a new conversation or get existing one between two users.
-        
-        KEY BEHAVIOR (Updated):
-        - One conversation per user pair (NOT per listing)
-        - If User A and User B already have a conversation, return it
-        - Always return property_card so frontend can ask to send inquiry
-        - Property inquiry cards can be sent for any listing in the same conversation
-        """
-        db = await get_db()
-        
-        # 1. Validate listing exists and get owner
-        if not ObjectId.is_valid(listing_id):
-            raise HTTPException(
-                status_code=status.HTTP_400_BAD_REQUEST,
-                detail="Invalid listing ID format"
-            )
-        
-        listing = await db.listings.find_one({"_id": ObjectId(listing_id)})
-        if not listing:
-            raise HTTPException(
-                status_code=status.HTTP_404_NOT_FOUND,
-                detail="Listing not found"
-            )
-        
-        owner_id = listing.get("user_id")
-        if not owner_id:
-            raise HTTPException(
-                status_code=status.HTTP_400_BAD_REQUEST,
-                detail="Listing has no owner"
-            )
-        
-        # 2. Check if user is trying to message themselves
-        if owner_id == current_user_id:
-            return {
-                "success": False,
-                "error": "self_chat",
-                "message": "Oops! You can't chat with yourself. This is your own listing! 😊",
-            }
-        
-        # 3. Prepare property card data (always needed for frontend prompt)
-        property_card = {
-            "listing_id": listing_id,
-            "title": listing.get("title", ""),
-            "price": listing.get("price", 0),
-            "currency": listing.get("currency", "NGN"),
-            "bedrooms": listing.get("bedrooms", 0),
-            "bathrooms": listing.get("bathrooms", 0),
-            "location": listing.get("location", ""),
-            "image_url": listing.get("images", [None])[0] if listing.get("images") else None,
-            "listing_type": listing.get("listing_type", ""),
-        }
-        
-        # 4. Check if conversation already exists BETWEEN THESE TWO USERS (regardless of listing)
-        participants = sorted([owner_id, current_user_id])  # Sort for consistent ordering
-        participants_key = "::".join(participants)
-        
-        existing_conversation = await db.conversations.find_one({
-            "participants_key": participants_key
-        })
-        
-        if existing_conversation:
-            logger.info(f"Found existing conversation between users: {existing_conversation['_id']}")
-            return {
-                "success": True,
-                "is_new": False,
-                "conversation": Conversation.format_response(existing_conversation),
-                "property_card": property_card,  # Always include so frontend can prompt to send
-            }
-        
-        # 5. Create new conversation (first time these two users chat)
-        conversation_doc = Conversation.create_document(
-            listing_id=listing_id,  # Store the first listing that started the conversation
-            participants=participants,
-            listing_title=listing.get("title", "Property"),
-            listing_image=listing.get("images", [None])[0] if listing.get("images") else None,
-        )
-        
-        result = await db.conversations.insert_one(conversation_doc)
-        conversation_id = str(result.inserted_id)
-        
-        logger.info(f"Created new conversation: {conversation_id}")
-        
-        # 6. Get conversation with ID
-        conversation_doc["_id"] = result.inserted_id
-        
-        return {
-            "success": True,
-            "is_new": True,
-            "conversation": Conversation.format_response(conversation_doc),
-            "property_card": property_card,  # Frontend will show confirmation dialog
-        }
-    
-    async def start_or_get_aida_conversation(
-        self,
-        current_user_id: str,
-    ) -> dict:
-        """
-        Start or get an AIDA DM conversation.
-        
-        This is used for:
-        - Opening the AIDA DM from the floating button
-        - Sending alert notifications to users
-        
-        Uses the same logic as alert_service.py for consistency.
-        """
-        db = await get_db()
-        
-        AIDA_BOT_ID = "AIDA_BOT"
-        LEGACY_AIDA_ID = "ai_assistant"  # Legacy ID that might exist in old conversations
-        
-        # 1. Find existing conversation between AIDA and User
-        # Try multiple formats for backwards compatibility
-        participants = sorted([AIDA_BOT_ID, current_user_id])
-        participants_key = "::".join(participants)
-        
-        # First, try newer format with participants_key
-        existing_conversation = await db.conversations.find_one({
-            "participants_key": participants_key
-        })
-        
-        # Fallback: Try legacy ai_assistant ID
-        if not existing_conversation:
-            legacy_participants = sorted([LEGACY_AIDA_ID, current_user_id])
-            legacy_key = "::".join(legacy_participants)
-            existing_conversation = await db.conversations.find_one({
-                "participants_key": legacy_key
-            })
-            if existing_conversation:
-                logger.info(f"Found legacy AIDA conversation with ai_assistant ID")
-        
-        # Fallback: Try querying by participants array
-        if not existing_conversation:
-            existing_conversation = await db.conversations.find_one({
-                "participants": {"$all": [AIDA_BOT_ID, current_user_id]}
-            })
-            if existing_conversation:
-                logger.info(f"Found AIDA conversation by participants array")
-        
-        # Fallback: Try legacy array format
-        if not existing_conversation:
-            existing_conversation = await db.conversations.find_one({
-                "participants": {"$all": [LEGACY_AIDA_ID, current_user_id]}
-            })
-            if existing_conversation:
-                logger.info(f"Found legacy AIDA conversation by participants array")
-        
-        if existing_conversation:
-            conv_id = str(existing_conversation["_id"])
-            logger.info(f"Found existing AIDA conversation: {conv_id}")
-            
-            # Determine which AIDA ID was used
-            aida_id = AIDA_BOT_ID if AIDA_BOT_ID in existing_conversation.get("participants", []) else LEGACY_AIDA_ID
-            actual_participants = [aida_id, current_user_id]
-            
-            # Enrich with participants data
-            enriched_participants, other_participant = await self._enrich_participants(
-                db, actual_participants, current_user_id
-            )
-            
-            # Format response
-            conv_response = Conversation.format_response(existing_conversation)
-            conv_response["participants"] = enriched_participants
-            conv_response["other_participant"] = other_participant
-            
-            return {
-                "success": True,
-                "is_new": False,
-                "conversation": conv_response,
-            }
-        
-        # 2. Create new AIDA conversation
-        conversation_doc = Conversation.create_document(
-            listing_id="system",  # Generic ID for system chats
-            participants=participants,
-            listing_title="AIDA Assistant",
-            listing_image=None
-        )
-        
-        try:
-            result = await db.conversations.insert_one(conversation_doc)
-            conv_id = str(result.inserted_id)
-            conversation_doc["_id"] = result.inserted_id
-            logger.info(f"Created new AIDA conversation: {conv_id}")
-        except Exception as e:
-            error_str = str(e)
-            logger.error(f"Insert failed with error: {error_str}")
-            
-            # Handle race condition - conversation was created between check and insert
-            if "duplicate key" in error_str.lower() or "E11000" in error_str:
-                logger.warning(f"Conversation already exists (race condition), trying all query methods...")
-                
-                # Try all possible query methods to find the existing conversation
-                existing_conversation = None
-                
-                # Method 1: by participants_key
-                existing_conversation = await db.conversations.find_one({
-                    "participants_key": participants_key
-                })
-                if existing_conversation:
-                    logger.info("Found by participants_key after race")
-                
-                # Method 2: by legacy participants_key
-                if not existing_conversation:
-                    legacy_key = "::".join(sorted([LEGACY_AIDA_ID, current_user_id]))
-                    existing_conversation = await db.conversations.find_one({
-                        "participants_key": legacy_key
-                    })
-                    if existing_conversation:
-                        logger.info("Found by legacy participants_key after race")
-                
-                # Method 3: by participants array (AIDA_BOT)
-                if not existing_conversation:
-                    existing_conversation = await db.conversations.find_one({
-                        "participants": {"$all": [AIDA_BOT_ID, current_user_id]}
-                    })
-                    if existing_conversation:
-                        logger.info("Found by participants array after race")
-                
-                # Method 4: by participants array (ai_assistant)
-                if not existing_conversation:
-                    existing_conversation = await db.conversations.find_one({
-                        "participants": {"$all": [LEGACY_AIDA_ID, current_user_id]}
-                    })
-                    if existing_conversation:
-                        logger.info("Found by legacy participants array after race")
-                
-                # Method 5: Any conversation with this user involving AIDA-like participants
-                if not existing_conversation:
-                    existing_conversation = await db.conversations.find_one({
-                        "$and": [
-                            {"participants": current_user_id},
-                            {"$or": [
-                                {"participants": AIDA_BOT_ID},
-                                {"participants": LEGACY_AIDA_ID},
-                                {"listing_title": "AIDA Assistant"},
-                                {"listing_id": "system"}
-                            ]}
-                        ]
-                    })
-                    if existing_conversation:
-                        logger.info(f"Found by broad AIDA query after race: {existing_conversation}")
-                
-                if existing_conversation:
-                    conv_id = str(existing_conversation["_id"])
-                    logger.info(f"Found existing AIDA conversation after race: {conv_id}")
-                    
-                    # Get actual participants from the found conversation
-                    actual_participants = existing_conversation.get("participants", participants)
-                    
-                    enriched_participants, other_participant = await self._enrich_participants(
-                        db, actual_participants, current_user_id
-                    )
-                    
-                    conv_response = Conversation.format_response(existing_conversation)
-                    conv_response["participants"] = enriched_participants
-                    conv_response["other_participant"] = other_participant
-                    
-                    return {
-                        "success": True,
-                        "is_new": False,
-                        "conversation": conv_response,
-                    }
-                else:
-                    # Log all indexes on the collection to help debug
-                    try:
-                        indexes = await db.conversations.index_information()
-                        logger.error(f"Collection indexes: {indexes}")
-                    except:
-                        pass
-                    logger.error(f"Could not find conversation after duplicate key error. user_id={current_user_id}, participants_key={participants_key}")
-                    raise HTTPException(
-                        status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
-                        detail=f"Failed to create AIDA conversation. Duplicate key but no match found."
-                    )
-            else:
-                logger.error(f"Unexpected error creating conversation: {e}")
-                raise
-        
-        # 3. Enrich with participants data
-        enriched_participants, other_participant = await self._enrich_participants(
-            db, participants, current_user_id
-        )
-        
-        conv_response = Conversation.format_response(conversation_doc)
-        conv_response["participants"] = enriched_participants
-        conv_response["other_participant"] = other_participant
-        
-        return {
-            "success": True,
-            "is_new": True,
-            "conversation": conv_response,
-        }
-    
-    async def _enrich_participants(self, db, participant_ids: list[str], current_user_id: str = None) -> tuple[list[dict], dict]:
-        """
-        Enrich participant IDs with full user data and online status.
-        Returns (all_participants, other_participant).
-        """
-        from app.services.presence_service import presence_service
-        
-        AIDA_BOT_ID = "AIDA_BOT"
-        AIDA_PROFILE = {
-            "id": AIDA_BOT_ID,
-            "name": "AIDA",
-            "profile_picture": "https://imagedelivery.net/0utJlkqgAVuawL5OpMWxgw/3922956f-b69d-4cb3-97b9-3a185abec900/public",
-            "is_online": True,  # AIDA is always online
-            "last_seen": None,
-        }
-        
-        participants = []
-        other_participant = None
-        
-        # AIDA IDs to handle (both new and legacy)
-        AIDA_IDS = {AIDA_BOT_ID, "ai_assistant"}
-        
-        # Batch fetch all users (skip AIDA IDs)
-        users_map = {}
-        real_user_ids = [uid for uid in participant_ids if uid not in AIDA_IDS]
-        
-        for uid in real_user_ids:
-            if ObjectId.is_valid(uid):
-                user = await db.users.find_one({"_id": ObjectId(uid)})
-                if user:
-                    users_map[uid] = user
-        
-        # Get online statuses (only for real users)
-        online_statuses = await presence_service.get_bulk_status(real_user_ids) if real_user_ids else {}
-        
-        # Build enriched participants
-        for uid in participant_ids:
-            # Handle AIDA (both AIDA_BOT and ai_assistant) specially
-            if uid in AIDA_IDS:
-                participant = AIDA_PROFILE.copy()
-            else:
-                user = users_map.get(uid)
-                status_data = online_statuses.get(uid, {"is_online": False, "last_seen": None})
-                
-                participant = {
-                    "id": uid,
-                    "name": f"{user.get('firstName', '')} {user.get('lastName', '')}".strip() if user else "Unknown",
-                    "profile_picture": user.get("profilePicture") if user else None,
-                    "is_online": status_data.get("is_online", False),
-                    "last_seen": status_data.get("last_seen"),
-                }
-            
-            participants.append(participant)
-            
-            # Track the OTHER participant (not current user)
-            if current_user_id and uid != current_user_id:
-                other_participant = participant
-        
-        return participants, other_participant
-    
-    async def get_user_conversations(
-        self,
-        user_id: str,
-    ) -> list[dict]:
-        """
-        Get all conversations for a user with ENRICHED participant data.
-        
-        Response includes for each conversation:
-        - participants: List with name, avatar, online status
-        - other_participant: Quick access to the other person
-        - unread_count: Just for current user (not dict)
-        """
-        db = await get_db()
-        
-        cursor = db.conversations.find({
-            "participants": user_id
-        }).sort("updated_at", -1)
-        
-        conversations = []
-        async for doc in cursor:
-            conv_id = str(doc["_id"])
-            participant_ids = doc.get("participants", [])
-            
-            # Enrich participants with user data + online status
-            enriched_participants, other_participant = await self._enrich_participants(
-                db, participant_ids, current_user_id=user_id
-            )
-            
-            # Get unread count for current user only
-            unread_counts = doc.get("unread_count", {})
-            user_unread = unread_counts.get(user_id, 0) if isinstance(unread_counts, dict) else 0
-            
-            # Build enriched conversation response
-            conversations.append({
-                "id": conv_id,
-                "listing_id": doc.get("listing_id", ""),
-                "listing_title": doc.get("listing_title", ""),
-                "listing_image": doc.get("listing_image"),
-                "participants": enriched_participants,
-                "other_participant": other_participant,
-                "last_message": doc.get("last_message", {}),
-                "unread_count": user_unread,
-                "status": doc.get("status", "active"),
-                "created_at": doc.get("created_at"),
-                "updated_at": doc.get("updated_at"),
-            })
-        
-        logger.info(f"Found {len(conversations)} enriched conversations for user {user_id}")
-        
-        return conversations
-    
-    async def get_conversation_by_id(
-        self,
-        conversation_id: str,
-        current_user_id: str,
-        include_messages: bool = False,
-        mark_as_read: bool = False,
-        message_limit: int = 50,
-    ) -> dict:
-        """
-        Get a single conversation with optional messages.
-        
-        This is the COMBINED endpoint that can:
-        - Return conversation with enriched participants
-        - Include messages (eliminating separate /messages call)
-        - Auto-mark as read (eliminating separate /read call)
-        """
-        db = await get_db()
-        
-        # Validate conversation ID
-        if not ObjectId.is_valid(conversation_id):
-            raise HTTPException(
-                status_code=status.HTTP_400_BAD_REQUEST,
-                detail="Invalid conversation ID format"
-            )
-        
-        # Fetch conversation
-        doc = await db.conversations.find_one({"_id": ObjectId(conversation_id)})
-        if not doc:
-            raise HTTPException(
-                status_code=status.HTTP_404_NOT_FOUND,
-                detail="Conversation not found"
-            )
-        
-        # Verify user is participant
-        if current_user_id not in doc.get("participants", []):
-            raise HTTPException(
-                status_code=status.HTTP_403_FORBIDDEN,
-                detail="You are not a participant in this conversation"
-            )
-        
-        # Enrich participants
-        participant_ids = doc.get("participants", [])
-        enriched_participants, other_participant = await self._enrich_participants(
-            db, participant_ids, current_user_id=current_user_id
-        )
-        
-        # Get unread count for current user
-        unread_counts = doc.get("unread_count", {})
-        user_unread = unread_counts.get(current_user_id, 0) if isinstance(unread_counts, dict) else 0
-        
-        # Build conversation response
-        conversation_data = {
-            "id": str(doc["_id"]),
-            "listing_id": doc.get("listing_id", ""),
-            "listing_title": doc.get("listing_title", ""),
-            "listing_image": doc.get("listing_image"),
-            "participants": enriched_participants,
-            "other_participant": other_participant,
-            "last_message": doc.get("last_message", {}),
-            "unread_count": user_unread,
-            "status": doc.get("status", "active"),
-            "created_at": doc.get("created_at"),
-            "updated_at": doc.get("updated_at"),
-        }
-        
-        result = {
-            "conversation": conversation_data,
-            "messages": [],
-            "has_more": False,
-            "total_messages": 0,
-        }
-        
-        # Include messages if requested
-        if include_messages:
-            messages = await self.get_conversation_messages(
-                conversation_id=conversation_id,
-                current_user_id=current_user_id,
-                limit=message_limit,
-            )
-            
-            # Count total messages
-            total_count = await db.messages.count_documents({
-                "conversation_id": conversation_id,
-                "deleted_for": {"$ne": current_user_id},
-            })
-            
-            result["messages"] = messages
-            result["has_more"] = total_count > len(messages)
-            result["total_messages"] = total_count
-        
-        # Mark as read if requested
-        if mark_as_read:
-            await self.mark_as_read(conversation_id, current_user_id)
-            result["conversation"]["unread_count"] = 0
-        
-        return result
-    
-    async def get_conversation_messages(
-        self,
-        conversation_id: str,
-        current_user_id: str,
-        limit: int = 50,
-        before_id: Optional[str] = None,
-    ) -> list[dict]:
-        """
-        Get messages for a conversation with pagination.
-        """
-        db = await get_db()
-        
-        # Validate conversation ID
-        if not ObjectId.is_valid(conversation_id):
-            raise HTTPException(
-                status_code=status.HTTP_400_BAD_REQUEST,
-                detail="Invalid conversation ID format"
-            )
-        
-        # Verify user is participant
-        conversation = await db.conversations.find_one({"_id": ObjectId(conversation_id)})
-        if not conversation:
-            raise HTTPException(
-                status_code=status.HTTP_404_NOT_FOUND,
-                detail="Conversation not found"
-            )
-        
-        if current_user_id not in conversation.get("participants", []):
-            raise HTTPException(
-                status_code=status.HTTP_403_FORBIDDEN,
-                detail="You are not a participant in this conversation"
-            )
-        
-        # Check if user has cleared the chat - get their cleared_at timestamp
-        user_cleared_at = conversation.get("cleared_at", {}).get(current_user_id)
-        
-        # Build query - exclude messages deleted specifically for this user
-        query = {
-            "conversation_id": conversation_id,
-            "deleted_for": {"$ne": current_user_id},  # Exclude messages deleted for this user
-        }
-        
-        # Filter out messages sent before user cleared the chat
-        # This ensures persistence across logout/login and new devices
-        if user_cleared_at:
-            query["created_at"] = {"$gt": user_cleared_at}
-        
-        if before_id and ObjectId.is_valid(before_id):
-            if "created_at" in query:
-                # Combine with existing created_at filter
-                query["$and"] = [
-                    {"created_at": query.pop("created_at")},
-                    {"_id": {"$lt": ObjectId(before_id)}}
-                ]
-            else:
-                query["_id"] = {"$lt": ObjectId(before_id)}
-        
-        # Get messages - sort ascending (oldest first, newest last)
-        # This gives chronological order: oldest at top, newest at bottom
-        cursor = db.messages.find(query).sort("created_at", 1).limit(limit)
-        
-        messages = []
-        async for doc in cursor:
-            # Pass user_id for proper filtering in format_response
-            formatted = Message.format_response(doc, for_user_id=current_user_id)
-            if formatted:  # Only add if not None (handles edge cases)
-                messages.append(formatted)
-        
-        logger.info(f"Found {len(messages)} messages for conversation {conversation_id}")
-        
-        return messages
-    
-    async def send_message(
-        self,
-        conversation_id: str,
-        current_user_id: str,
-        current_user_name: str,
-        current_user_avatar: Optional[str],
-        message_type: str,
-        content: Optional[str] = None,
-        media: Optional[dict] = None,
-        property_card: Optional[dict] = None,
-        replied_to_message_id: Optional[str] = None,
-        replied_to_content: Optional[str] = None,
-        replied_to_sender: Optional[str] = None,
-    ) -> dict:
-        """
-        Send a message in a conversation.
-        """
-        db = await get_db()
-        
-        # Validate conversation
-        if not ObjectId.is_valid(conversation_id):
-            raise HTTPException(
-                status_code=status.HTTP_400_BAD_REQUEST,
-                detail="Invalid conversation ID format"
-            )
-        
-        conversation = await db.conversations.find_one({"_id": ObjectId(conversation_id)})
-        if not conversation:
-            raise HTTPException(
-                status_code=status.HTTP_404_NOT_FOUND,
-                detail="Conversation not found"
-            )
-        
-        if current_user_id not in conversation.get("participants", []):
-            raise HTTPException(
-                status_code=status.HTTP_403_FORBIDDEN,
-                detail="You are not a participant in this conversation"
-            )
-        
-        # Auto-detect message type based on content
-        if property_card:
-            message_type = "property_inquiry"  # Override type when property card is present
-        
-        # Create message
-        message_doc = Message.create_document(
-            conversation_id=conversation_id,
-            sender_id=current_user_id,
-            sender_name=current_user_name,
-            sender_avatar=current_user_avatar,
-            message_type=message_type,
-            content=content,
-            media=media,
-            property_card=property_card,
-            replied_to_message_id=replied_to_message_id,
-            replied_to_content=replied_to_content,
-            replied_to_sender=replied_to_sender,
-        )
-        
-        result = await db.messages.insert_one(message_doc)
-        message_id = str(result.inserted_id)
-        
-        # Update conversation's last_message and unread count
-        other_user_id = next(
-            (p for p in conversation["participants"] if p != current_user_id),
-            None
-        )
-        
-        update_data = {
-            "last_message": {
-                "text": content or f"[{message_type}]",
-                "sender_id": current_user_id,
-                "timestamp": message_doc["created_at"],
-            },
-            "updated_at": datetime.utcnow(),
-        }
-        
-        # Increment unread count for other user
-        if other_user_id:
-            update_data[f"unread_count.{other_user_id}"] = conversation.get("unread_count", {}).get(other_user_id, 0) + 1
-        
-        await db.conversations.update_one(
-            {"_id": ObjectId(conversation_id)},
-            {"$set": update_data}
-        )
-        
-        logger.info(f"Message {message_id} sent in conversation {conversation_id}")
-        
-        # Get message with ID
-        message_doc["_id"] = result.inserted_id
-        
-        # Broadcast to all participants via WebSocket (for in-app notifications)
-        try:
-            formatted_message = Message.format_response(message_doc)
-            broadcast_message = {
-                "action": "new_message",
-                "conversation_id": conversation_id,
-                "message": formatted_message,
-            }
-            await chat_manager.broadcast_to_conversation(
-                conversation_id,
-                conversation["participants"],
-                broadcast_message
-            )
-            logger.info(f"WebSocket broadcast sent for message {message_id}")
-        except Exception as e:
-            # Don't fail the request if WebSocket broadcast fails
-            logger.warning(f"Failed to broadcast message via WebSocket: {e}")
-        
-        # NEW: Detect AIDA chat and route to AI brain
-        AIDA_IDS = {"AIDA_BOT", "ai_assistant"}
-        is_aida_chat = any(p in AIDA_IDS for p in conversation.get("participants", []))
-        
-        # Process if AIDA chat, user is not AIDA, and message is TEXT or VOICE
-        # (check content for text, or media for voice)
-        should_process = (
-            is_aida_chat and 
-            current_user_id not in AIDA_IDS and 
-            (
-                (message_type == "text" and content) or 
-                (message_type == "voice" and media and media.get("url"))
-            )
-        )
-        
-        if should_process:
-            try:
-                logger.info(f"AIDA DM detected (type={message_type}), routing to AI brain...")
-                
-                # Extract audio URL for voice messages
-                audio_url = media.get("url") if message_type == "voice" and media else None
-                
-                # Build reply context if replying to a message
-                reply_context = None
-                if replied_to_message_id:
-                    replied_msg = await db.messages.find_one({"_id": ObjectId(replied_to_message_id)})
-                    if replied_msg:
-                        reply_context = {
-                            "message_id": replied_to_message_id,
-                            "message_content": replied_msg.get("content"),
-                            "property_card": replied_msg.get("property_card"),
-                            "metadata": replied_msg.get("metadata"),
-                        }
-                
-                # Call AIDA brain (async, don't block response)
-                import asyncio
-                asyncio.create_task(self._process_aida_response(
-                    db=db,
-                    conversation_id=conversation_id,
-                    conversation=conversation,
-                    user_id=current_user_id,
-                    user_message=content,
-                    user_message_id=message_id,
-                    reply_context=reply_context,
-                    audio_url=audio_url,
-                ))
-                
-            except Exception as e:
-                logger.error(f"Failed to route to AIDA brain: {e}")
-        
-        return {
-            "success": True,
-            "message": Message.format_response(message_doc),
-        }
-    
-    async def _process_aida_response(
-        self,
-        db,
-        conversation_id: str,
-        conversation: dict,
-        user_id: str,
-        user_message: str,
-        user_message_id: str,
-        reply_context: Optional[dict] = None,
-        audio_url: Optional[str] = None,
-    ):
-        """
-        Process user message through AIDA AI brain and send response.
-        Runs asynchronously so user sees their message immediately.
-        """
-        try:
-            logger.info(f"Processing AIDA response for user {user_id}")
-            
-            # 1. Handle Voice Transcription if needed
-            if audio_url:
-                try:
-                    from app.services.voice_service import voice_service
-                    logger.info(f"Transcribing voice message: {audio_url}")
-                    transcript, lang = await voice_service.transcribe_audio(audio_url)
-                    
-                    if transcript:
-                        user_message = transcript
-                        logger.info(f"Voice transcribed: '{user_message}' ({lang})")
-                        
-                        # Optionally update the original message with transcript?
-                        # For now, we just let AIDA know the content
-                    else:
-                        logger.warning("Empty transcription result")
-                        user_message = "(Inaudible voice message)"
-                        
-                except Exception as e:
-                    logger.error(f"Transcription failed: {e}")
-                    # Don't crash, just let AIDA handle failure
-                    user_message = "(Voice message could not be transcribed)"
-            
-            # Import specialized DM brain
-            from app.ai.agent.dm_brain import DmBrain
-            
-            # Build context message with reply info
-            context_message = user_message
-            if reply_context:
-                property_card = reply_context.get("property_card")
-                replied_content = reply_context.get("message_content", "")
-                metadata = reply_context.get("metadata", {})
-                alert_title = metadata.get("alert_title") if metadata else None
-                
-                if property_card:
-                    listing_id = property_card.get("listing_id")
-                    listing_title = property_card.get("title")
-                    listing_location = property_card.get("location")
-                    
-                    context_message = f"""{user_message}
-
-[System Context: User is replying to property listing:
-- ID: {listing_id}
-- Title: {listing_title}
-- Location: {listing_location}
-
-When user says "this", "it", "this property", they mean this listing.
-If user says they found what they wanted, use delete_alert tool with location matching this listing.]"""
-                
-                elif alert_title:
-                    context_message = f"""{user_message}
-
-[System Context: User is replying to alert notification for: "{alert_title}"
-If user says they found what they wanted or wants to stop notifications, use delete_alert tool.]"""
-                
-                elif replied_content:
-                    context_message = f"""{user_message}
-
-[System Context: User is replying to AIDA's previous message: "{replied_content}"]"""
-            
-            # Call specialized DM brain (architecture separation)
-            brain = DmBrain()
-            result = await brain.process(
-                message=context_message,
-                user_id=user_id,
-                source="dm",
-            )
-            
-            # Extract response text and metadata
-            response_text = result.get("text", "I'm sorry, I couldn't process that. Please try again.")
-            response_metadata = result.get("metadata", {})
-            
-            # Check for property cards or alert results
-            property_card = None
-            if response_metadata.get("property_card"):
-                property_card = response_metadata.get("property_card")
-            
-            # ============================================================
-            # VOICE RESPONSE: If user sent a voice note, AIDA responds with voice too
-            # ============================================================
-            aida_audio_url = None
-            aida_audio_duration = None
-            message_type = "text"  # Default to text
-            
-            if audio_url:
-                # User sent voice, so AIDA should respond with voice
-                try:
-                    from app.services.voice_service import voice_service
-                    
-                    # Determine language from transcription or default to English
-                    response_language = "en"
-                    
-                    # Generate AIDA's voice response
-                    logger.info(f"Generating AIDA voice response for DM...")
-                    voice_result = await voice_service.generate_aida_voice_response(
-                        text=response_text,
-                        language=response_language
-                    )
-                    
-                    if voice_result:
-                        aida_audio_url = voice_result.get("audio_url")
-                        aida_audio_duration = voice_result.get("duration")
-                        message_type = "voice"  # Change message type to voice
-                        logger.info(f"✅ AIDA voice response generated: {aida_audio_url}")
-                        
-                except Exception as e:
-                    logger.error(f"Failed to generate AIDA voice response: {e}")
-                    # Fall back to text response
-                    message_type = "text"
-            
-            # Send AIDA's response as a message in the conversation
-            aida_message_doc = Message.create_document(
-                conversation_id=conversation_id,
-                sender_id="AIDA_BOT",
-                sender_name="AIDA",
-                sender_avatar="https://imagedelivery.net/0utJlkqgAVuawL5OpMWxgw/3922956f-b69d-4cb3-97b9-3a185abec900/public",
-                message_type=message_type,
-                content=response_text,
-                property_card=property_card,
-                replied_to_message_id=user_message_id if reply_context else None,
-                replied_to_content=user_message if reply_context else None,
-                replied_to_sender="User" if reply_context else None,
-            )
-            
-            # Add voice-specific fields if this is a voice response
-            if aida_audio_url:
-                aida_message_doc["audio_url"] = aida_audio_url
-                aida_message_doc["audio_duration"] = aida_audio_duration
-            
-            # Add metadata for rich content (alert results, etc.)
-            if response_metadata:
-                aida_message_doc["metadata"] = response_metadata
-            
-            # Insert AIDA's message
-            aida_result = await db.messages.insert_one(aida_message_doc)
-            aida_message_id = str(aida_result.inserted_id)
-            aida_message_doc["_id"] = aida_result.inserted_id
-            
-            logger.info(f"AIDA response {aida_message_id} sent in conversation {conversation_id}")
-            
-            # Update conversation's last_message (AIDA's response)
-            update_data = {
-                "last_message": {
-                    "text": response_text[:100] if response_text else "[AI Response]",
-                    "sender_id": "AIDA_BOT",
-                    "timestamp": aida_message_doc["created_at"],
-                },
-                "updated_at": datetime.utcnow(),
-            }
-            
-            # Increment unread count for user (AIDA just sent a message)
-            update_data[f"unread_count.{user_id}"] = conversation.get("unread_count", {}).get(user_id, 0) + 1
-            
-            await db.conversations.update_one(
-                {"_id": ObjectId(conversation_id)},
-                {"$set": update_data}
-            )
-            
-            # Broadcast AIDA's response via WebSocket
-            try:
-                formatted_message = Message.format_response(aida_message_doc)
-                broadcast_message = {
-                    "action": "new_message",
-                    "conversation_id": conversation_id,
-                    "message": formatted_message,
-                }
-                await chat_manager.broadcast_to_conversation(
-                    conversation_id,
-                    conversation["participants"],
-                    broadcast_message
-                )
-                logger.info(f"AIDA response broadcast via WebSocket")
-            except Exception as e:
-                logger.warning(f"Failed to broadcast AIDA response: {e}")
-                
-        except Exception as e:
-            logger.error(f"Error processing AIDA response: {e}")
-            # Optionally send error message to user
-            try:
-                error_message_doc = Message.create_document(
-                    conversation_id=conversation_id,
-                    sender_id="AIDA_BOT",
-                    sender_name="AIDA",
-                    sender_avatar="https://imagedelivery.net/0utJlkqgAVuawL5OpMWxgw/3922956f-b69d-4cb3-97b9-3a185abec900/public",
-                    message_type="text",
-                    content="I'm having trouble processing that right now. Please try again in a moment. 😅",
-                )
-                await db.messages.insert_one(error_message_doc)
-            except:
-                pass
-    
-    async def mark_as_read(
-        self,
-        conversation_id: str,
-        current_user_id: str,
-    ) -> dict:
-        """
-        Mark all messages in a conversation as read for the current user.
-        """
-        db = await get_db()
-        
-        # Validate conversation
-        if not ObjectId.is_valid(conversation_id):
-            raise HTTPException(
-                status_code=status.HTTP_400_BAD_REQUEST,
-                detail="Invalid conversation ID format"
-            )
-        
-        conversation = await db.conversations.find_one({"_id": ObjectId(conversation_id)})
-        if not conversation:
-            raise HTTPException(
-                status_code=status.HTTP_404_NOT_FOUND,
-                detail="Conversation not found"
-            )
-        
-        if current_user_id not in conversation.get("participants", []):
-            raise HTTPException(
-                status_code=status.HTTP_403_FORBIDDEN,
-                detail="You are not a participant in this conversation"
-            )
-        
-        # Mark messages as read and add user to read_by array
-        now = datetime.utcnow()
-        result = await db.messages.update_many(
-            {
-                "conversation_id": conversation_id,
-                "sender_id": {"$ne": current_user_id},  # Not sent by this user
-                "read_by": {"$ne": current_user_id},  # Not already read by this user
-            },
-            {
-                "$set": {
-                    "is_read": True,  # Keep for backwards compatibility
-                    "read_at": now,
-                },
-                "$addToSet": {
-                    "read_by": current_user_id  # Add user to read_by array (no duplicates)
-                }
-            }
-        )
-        
-        logger.info(f"Marked {result.modified_count} messages as read, added {current_user_id} to read_by")
-        
-        # Reset unread count for current user
-        await db.conversations.update_one(
-            {"_id": ObjectId(conversation_id)},
-            {"$set": {f"unread_count.{current_user_id}": 0}}
-        )
-        
-        # Broadcast message_read event to other participants via WebSocket
-        try:
-            from app.routes.websocket_chat import chat_manager
-            
-            read_event = {
-                "action": "message_read",
-                "conversation_id": conversation_id,
-                "read_by": current_user_id,
-                "read_at": now.isoformat(),
-            }
-            
-            # Send to other participants
-            for participant_id in conversation.get("participants", []):
-                if participant_id != current_user_id:
-                    await chat_manager.send_to_user(participant_id, read_event)
-            
-            logger.info(f"Broadcasted message_read event for conversation {conversation_id}")
-        except Exception as e:
-            # Don't fail the request if WebSocket broadcast fails
-            logger.warning(f"Failed to broadcast message_read event: {e}")
-        
-        logger.info(f"Marked messages as read in conversation {conversation_id} for user {current_user_id}")
-        
-        return {
-            "success": True,
-            "message": "Messages marked as read",
-        }
-    
-    # ============================================================
-    # NEW METHODS: Edit, Delete, Reactions, Clear Chat
-    # ============================================================
-    
-    async def edit_message(
-        self,
-        conversation_id: str,
-        message_id: str,
-        user_id: str,
-        new_content: str,
-    ) -> dict:
-        """
-        Edit a message content.
-        Only the sender can edit, within 24 hours, text messages only.
-        """
-        db = await get_db()
-        
-        # Validate IDs
-        if not ObjectId.is_valid(conversation_id) or not ObjectId.is_valid(message_id):
-            raise HTTPException(
-                status_code=status.HTTP_400_BAD_REQUEST,
-                detail="Invalid ID format"
-            )
-        
-        # Get the message
-        message_doc = await db.messages.find_one({"_id": ObjectId(message_id)})
-        if not message_doc:
-            raise HTTPException(
-                status_code=status.HTTP_404_NOT_FOUND,
-                detail="Message not found"
-            )
-        
-        # Verify sender
-        if message_doc.get("sender_id") != user_id:
-            raise HTTPException(
-                status_code=status.HTTP_403_FORBIDDEN,
-                detail="You can only edit your own messages"
-            )
-        
-        # Verify message is in the correct conversation
-        if message_doc.get("conversation_id") != conversation_id:
-            raise HTTPException(
-                status_code=status.HTTP_400_BAD_REQUEST,
-                detail="Message does not belong to this conversation"
-            )
-        
-        # Only allow editing text messages
-        if message_doc.get("message_type") != "text":
-            raise HTTPException(
-                status_code=status.HTTP_400_BAD_REQUEST,
-                detail="Only text messages can be edited"
-            )
-        
-        # Check 15-minute edit window
-        created_at = message_doc.get("created_at")
-        if created_at:
-            minutes_since = (datetime.utcnow() - created_at).total_seconds() / 60
-            if minutes_since > 15:
-                raise HTTPException(
-                    status_code=status.HTTP_400_BAD_REQUEST,
-                    detail="Edit window expired (15 minutes)"
-                )
-        
-        # Cannot edit deleted messages
-        if message_doc.get("is_deleted"):
-            raise HTTPException(
-                status_code=status.HTTP_400_BAD_REQUEST,
-                detail="Cannot edit a deleted message"
-            )
-        
-        # Update the message
-        now = datetime.utcnow()
-        await db.messages.update_one(
-            {"_id": ObjectId(message_id)},
-            {
-                "$set": {
-                    "content": new_content.strip(),
-                    "is_edited": True,
-                    "edited_at": now,
-                }
-            }
-        )
-        
-        # Broadcast message_edited event to all participants via WebSocket
-        try:
-            from app.routes.websocket_chat import chat_manager
-            
-            conversation = await db.conversations.find_one({"_id": ObjectId(conversation_id)})
-            if conversation:
-                edit_event = {
-                    "action": "message_edited",
-                    "conversation_id": conversation_id,
-                    "message_id": message_id,
-                    "new_content": new_content.strip(),
-                    "edited_at": now.isoformat(),
-                    "edited_by": user_id,
-                }
-                
-                await chat_manager.broadcast_to_conversation(
-                    conversation_id,
-                    conversation.get("participants", []),
-                    edit_event
-                )
-                
-                logger.info(f"Broadcasted message_edited event for message {message_id}")
-        except Exception as e:
-            logger.warning(f"Failed to broadcast message_edited event: {e}")
-        
-        logger.info(f"Message {message_id} edited by {user_id}")
-        
-        return {
-            "success": True,
-            "message_id": message_id,
-            "new_content": new_content.strip(),
-            "edited_at": now.isoformat(),
-        }
-    
-    async def delete_message(
-        self,
-        conversation_id: str,
-        message_id: str,
-        user_id: str,
-        delete_for: str = "me",  # "everyone" or "me"
-    ) -> dict:
-        """
-        Delete a message.
-        - "everyone": Only sender, within 1 hour
-        - "me": Any participant
-        """
-        db = await get_db()
-        
-        # Validate IDs
-        if not ObjectId.is_valid(conversation_id) or not ObjectId.is_valid(message_id):
-            raise HTTPException(
-                status_code=status.HTTP_400_BAD_REQUEST,
-                detail="Invalid ID format"
-            )
-        
-        # Get the message
-        message_doc = await db.messages.find_one({"_id": ObjectId(message_id)})
-        if not message_doc:
-            raise HTTPException(
-                status_code=status.HTTP_404_NOT_FOUND,
-                detail="Message not found"
-            )
-        
-        # Verify message is in the correct conversation
-        if message_doc.get("conversation_id") != conversation_id:
-            raise HTTPException(
-                status_code=status.HTTP_400_BAD_REQUEST,
-                detail="Message does not belong to this conversation"
-            )
-        
-        # Verify user is a participant
-        conversation = await db.conversations.find_one({"_id": ObjectId(conversation_id)})
-        if not conversation or user_id not in conversation.get("participants", []):
-            raise HTTPException(
-                status_code=status.HTTP_403_FORBIDDEN,
-                detail="You are not a participant in this conversation"
-            )
-        
-        now = datetime.utcnow()
-        
-        if delete_for == "everyone":
-            # Only sender can delete for everyone
-            if message_doc.get("sender_id") != user_id:
-                raise HTTPException(
-                    status_code=status.HTTP_403_FORBIDDEN,
-                    detail="Only the sender can delete for everyone"
-                )
-            
-            # Check 1-hour delete window
-            created_at = message_doc.get("created_at")
-            if created_at:
-                hours_since = (datetime.utcnow() - created_at).total_seconds() / 3600
-                if hours_since > 1:
-                    raise HTTPException(
-                        status_code=status.HTTP_400_BAD_REQUEST,
-                        detail="Delete-for-everyone window expired (1 hour)"
-                    )
-            
-            # Mark as deleted for everyone
-            await db.messages.update_one(
-                {"_id": ObjectId(message_id)},
-                {
-                    "$set": {
-                        "is_deleted": True,
-                        "deleted_at": now,
-                    }
-                }
-            )
-            
-            # Broadcast message_deleted event to all participants via WebSocket
-            try:
-                from app.routes.websocket_chat import chat_manager
-                
-                delete_event = {
-                    "action": "message_deleted",
-                    "conversation_id": conversation_id,
-                    "message_id": message_id,
-                    "deleted_for": "everyone",
-                    "deleted_at": now.isoformat(),
-                    "deleted_by": user_id,
-                }
-                
-                await chat_manager.broadcast_to_conversation(
-                    conversation_id,
-                    conversation.get("participants", []),
-                    delete_event
-                )
-                
-                logger.info(f"Broadcasted message_deleted event for message {message_id}")
-            except Exception as e:
-                logger.warning(f"Failed to broadcast message_deleted event: {e}")
-            
-            logger.info(f"Message {message_id} deleted for everyone by {user_id}")
-            
-            return {
-                "success": True,
-                "message_id": message_id,
-                "deleted_for": "everyone",
-                "deleted_at": now.isoformat(),
-            }
-        
-        else:  # delete_for == "me"
-            # Add user to deleted_for list
-            await db.messages.update_one(
-                {"_id": ObjectId(message_id)},
-                {"$addToSet": {"deleted_for": user_id}}
-            )
-            
-            logger.info(f"Message {message_id} deleted for {user_id} only")
-            
-            return {
-                "success": True,
-                "message_id": message_id,
-                "deleted_for": "me",
-                "deleted_at": now.isoformat(),
-            }
-    
-    async def add_reaction(
-        self,
-        conversation_id: str,
-        message_id: str,
-        user_id: str,
-        emoji: str,
-    ) -> dict:
-        """Add an emoji reaction to a message."""
-        db = await get_db()
-        
-        # Validate IDs
-        if not ObjectId.is_valid(conversation_id) or not ObjectId.is_valid(message_id):
-            raise HTTPException(
-                status_code=status.HTTP_400_BAD_REQUEST,
-                detail="Invalid ID format"
-            )
-        
-        # Get the message
-        message_doc = await db.messages.find_one({"_id": ObjectId(message_id)})
-        if not message_doc:
-            raise HTTPException(
-                status_code=status.HTTP_404_NOT_FOUND,
-                detail="Message not found"
-            )
-        
-        # Verify message is in the correct conversation
-        if message_doc.get("conversation_id") != conversation_id:
-            raise HTTPException(
-                status_code=status.HTTP_400_BAD_REQUEST,
-                detail="Message does not belong to this conversation"
-            )
-        
-        # Verify user is a participant
-        conversation = await db.conversations.find_one({"_id": ObjectId(conversation_id)})
-        if not conversation or user_id not in conversation.get("participants", []):
-            raise HTTPException(
-                status_code=status.HTTP_403_FORBIDDEN,
-                detail="You are not a participant in this conversation"
-            )
-        
-        # Cannot react to deleted messages
-        if message_doc.get("is_deleted"):
-            raise HTTPException(
-                status_code=status.HTTP_400_BAD_REQUEST,
-                detail="Cannot react to a deleted message"
-            )
-        
-        # Add reaction
-        await db.messages.update_one(
-            {"_id": ObjectId(message_id)},
-            {"$addToSet": {f"reactions.{emoji}": user_id}}
-        )
-        
-        # Broadcast reaction_added event to all participants via WebSocket
-        try:
-            from app.routes.websocket_chat import chat_manager
-            
-            reaction_event = {
-                "action": "reaction_added",
-                "conversation_id": conversation_id,
-                "message_id": message_id,
-                "emoji": emoji,
-                "user_id": user_id,
-            }
-            
-            await chat_manager.broadcast_to_conversation(
-                conversation_id,
-                conversation.get("participants", []),
-                reaction_event
-            )
-            
-            logger.info(f"Broadcasted reaction_added event for message {message_id}")
-        except Exception as e:
-            logger.warning(f"Failed to broadcast reaction_added event: {e}")
-        
-        logger.info(f"Reaction {emoji} added to message {message_id} by {user_id}")
-        
-        return {
-            "success": True,
-            "message_id": message_id,
-            "emoji": emoji,
-            "user_id": user_id,
-        }
-    
-    async def remove_reaction(
-        self,
-        conversation_id: str,
-        message_id: str,
-        user_id: str,
-        emoji: str,
-    ) -> dict:
-        """Remove an emoji reaction from a message."""
-        db = await get_db()
-        
-        # Validate IDs
-        if not ObjectId.is_valid(conversation_id) or not ObjectId.is_valid(message_id):
-            raise HTTPException(
-                status_code=status.HTTP_400_BAD_REQUEST,
-                detail="Invalid ID format"
-            )
-        
-        # Get the message
-        message_doc = await db.messages.find_one({"_id": ObjectId(message_id)})
-        if not message_doc:
-            raise HTTPException(
-                status_code=status.HTTP_404_NOT_FOUND,
-                detail="Message not found"
-            )
-        
-        # Verify message is in the correct conversation
-        if message_doc.get("conversation_id") != conversation_id:
-            raise HTTPException(
-                status_code=status.HTTP_400_BAD_REQUEST,
-                detail="Message does not belong to this conversation"
-            )
-        
-        # Verify user is a participant
-        conversation = await db.conversations.find_one({"_id": ObjectId(conversation_id)})
-        if not conversation or user_id not in conversation.get("participants", []):
-            raise HTTPException(
-                status_code=status.HTTP_403_FORBIDDEN,
-                detail="You are not a participant in this conversation"
-            )
-        
-        # Remove reaction
-        await db.messages.update_one(
-            {"_id": ObjectId(message_id)},
-            {"$pull": {f"reactions.{emoji}": user_id}}
-        )
-        
-        # Clean up empty reaction arrays
-        updated_msg = await db.messages.find_one({"_id": ObjectId(message_id)})
-        if updated_msg:
-            reactions = updated_msg.get("reactions", {})
-            if emoji in reactions and len(reactions[emoji]) == 0:
-                await db.messages.update_one(
-                    {"_id": ObjectId(message_id)},
-                    {"$unset": {f"reactions.{emoji}": ""}}
-                )
-        
-        # Broadcast reaction_removed event to all participants via WebSocket
-        try:
-            from app.routes.websocket_chat import chat_manager
-            
-            reaction_event = {
-                "action": "reaction_removed",
-                "conversation_id": conversation_id,
-                "message_id": message_id,
-                "emoji": emoji,
-                "user_id": user_id,
-            }
-            
-            await chat_manager.broadcast_to_conversation(
-                conversation_id,
-                conversation.get("participants", []),
-                reaction_event
-            )
-            
-            logger.info(f"Broadcasted reaction_removed event for message {message_id}")
-        except Exception as e:
-            logger.warning(f"Failed to broadcast reaction_removed event: {e}")
-        
-        logger.info(f"Reaction {emoji} removed from message {message_id} by {user_id}")
-        
-        return {
-            "success": True,
-            "message_id": message_id,
-            "emoji": emoji,
-            "user_id": user_id,
-        }
-    
-    async def clear_chat(
-        self,
-        conversation_id: str,
-        user_id: str,
-    ) -> dict:
-        """
-        Clear all messages in a conversation for the current user only.
-        
-        Stores a cleared_at timestamp on the conversation document.
-        Messages with created_at <= cleared_at won't be shown to this user,
-        even after logout/login or on a new device.
-        Other participants still see all messages.
-        """
-        db = await get_db()
-        
-        # Validate ID
-        if not ObjectId.is_valid(conversation_id):
-            raise HTTPException(
-                status_code=status.HTTP_400_BAD_REQUEST,
-                detail="Invalid conversation ID format"
-            )
-        
-        # Verify user is a participant
-        conversation = await db.conversations.find_one({"_id": ObjectId(conversation_id)})
-        if not conversation or user_id not in conversation.get("participants", []):
-            raise HTTPException(
-                status_code=status.HTTP_403_FORBIDDEN,
-                detail="You are not a participant in this conversation"
-            )
-        
-        now = datetime.utcnow()
-        
-        # Store cleared_at timestamp on conversation for persistent filtering
-        # This is the key change - ensures clear persists across sessions/devices
-        await db.conversations.update_one(
-            {"_id": ObjectId(conversation_id)},
-            {"$set": {f"cleared_at.{user_id}": now}}
-        )
-        
-        # Also mark existing messages with deleted_for (backwards compatibility)
-        # This provides immediate UI response and works with existing code
-        result = await db.messages.update_many(
-            {"conversation_id": conversation_id},
-            {"$addToSet": {"deleted_for": user_id}}
-        )
-        
-        logger.info(f"Chat {conversation_id} cleared for {user_id} at {now.isoformat()} ({result.modified_count} messages)")
-        
-        return {
-            "success": True,
-            "conversation_id": conversation_id,
-            "cleared_count": result.modified_count,
-            "cleared_at": now.isoformat(),
-        }
-
+class ConversationService(
+    ConversationCRUDMixin,
+    ConversationMessageMixin,
+    ConversationActionMixin,
+    ConversationAIMixin
+):
+    """
+    Main Conversation Service.
+    
+    Refactored into Mixins for maintainability:
+    - CRUD: Basic init, get lists, enrich participants
+    - Messages: Send, get, mark read
+    - Actions: Edit, delete, react, clear
+    - AI: AIDA integration
+    """
+    pass
 
 # Singleton instance
 conversation_service = ConversationService()
-

app/services/redis_pubsub.py

@@ -0,0 +1,102 @@
+# ============================================================
+# app/services/redis_pubsub.py - Redis Pub/Sub for WebSockets
+# ============================================================
+#
+# Enables horizontal scaling effectively by broadcasting WebSocket
+# events across multiple server instances via Redis.
+#
+# Pattern: "Broadcast"
+# 1. Any server can publish a message to the global channel
+# 2. All servers subscribe to the channel
+# 3. Each server checks if the target user(s) are connected locally
+# 4. If connected, the server delivers the message via WebSocket
+# ============================================================
+
+import json
+import logging
+import asyncio
+from typing import Callable, Any, Optional
+from app.ai.config import redis_client
+
+logger = logging.getLogger(__name__)
+
+# Global channel name
+CHAT_CHANNEL = "lojiz_chat_global"
+
+class RedisPubSubService:
+    """Service for handling Redis Pub/Sub operations"""
+    
+    def __init__(self):
+        self.is_listening = False
+        self.message_handler: Optional[Callable[[dict], Any]] = None
+    
+    async def publish(self, event_type: str, data: dict):
+        """
+        Publish an event to the global Redis channel.
+        
+        Args:
+            event_type: "broadcast" (multiple users) or "direct" (single user)
+            data: The payload containing target_users and message
+        """
+        if not redis_client:
+            logger.warning("Redis not available, pub/sub disabled")
+            return
+            
+        payload = {
+            "type": event_type,
+            "data": data,
+            "source_server": id(self),  # Simple way to identify sender (not robust across processes but helpful)
+        }
+        
+        try:
+            await redis_client.publish(CHAT_CHANNEL, json.dumps(payload))
+        except Exception as e:
+            logger.error(f"Redis publish failed: {e}")
+
+    async def start_listening(self, handler: Callable[[dict], Any]):
+        """
+        Start listening to the global channel in a background task.
+        
+        Args:
+            handler: Async function to process received messages
+        """
+        if not redis_client:
+            return
+
+        self.message_handler = handler
+        self.is_listening = True
+        
+        # Start background listener
+        asyncio.create_task(self._listener_loop())
+        logger.info(f"✅ Redis Pub/Sub listener started on channel: {CHAT_CHANNEL}")
+
+    async def _listener_loop(self):
+        """Background loop to listen for Redis messages"""
+        pubsub = redis_client.pubsub()
+        await pubsub.subscribe(CHAT_CHANNEL)
+        
+        try:
+            async for message in pubsub.listen():
+                if not self.is_listening:
+                    break
+                    
+                if message["type"] == "message":
+                    try:
+                        payload = json.loads(message["data"])
+                        # Call the handler with the payload
+                        if self.message_handler:
+                            await self.message_handler(payload)
+                    except json.JSONDecodeError:
+                        logger.warning("Received invalid JSON in Redis message")
+                    except Exception as e:
+                        logger.error(f"Error processing Redis message: {e}")
+                        
+        except Exception as e:
+            logger.error(f"Redis listener loop crashed: {e}")
+            # Retry logic could go here
+        finally:
+            await pubsub.unsubscribe(CHAT_CHANNEL)
+            await pubsub.close()
+
+# Singleton instance
+redis_pubsub = RedisPubSubService()

main.py

@@ -5,6 +5,10 @@ from fastapi import FastAPI, Request
 from fastapi.middleware.cors import CORSMiddleware
 from fastapi.responses import JSONResponse
 from fastapi.exceptions import RequestValidationError
+from slowapi import _rate_limit_exceeded_handler
+from slowapi.errors import RateLimitExceeded
+from slowapi.middleware import SlowAPIMiddleware
+from app.middleware import limiter, rate_limit_exceeded_handler
 from contextlib import asynccontextmanager
 import logging
 import os
@@ -120,6 +124,14 @@ async def lifespan(app: FastAPI):
     except Exception as e:
         logger.warning(f"⚠️ Memory Manager initialization warning: {e}")
 
+    # Initialize Distributed Chat (Redis Pub/Sub)
+    try:
+        from app.routes.websocket_chat import chat_manager
+        await chat_manager.initialize()
+        logger.info("✅ Redis Pub/Sub for Chat initialized")
+    except Exception as e:
+        logger.warning(f"⚠️ Chat manager init failed: {e}")
+
     logger.info("=" * 70)
     logger.info("✅ APPLICATION READY - Graph-Based Architecture Active!")
     logger.info("=" * 70)
@@ -160,6 +172,11 @@ app = FastAPI(
     lifespan=lifespan,
 )
 
+# RATE LIMITING
+app.state.limiter = limiter
+app.add_exception_handler(RateLimitExceeded, rate_limit_exceeded_handler)
+app.add_middleware(SlowAPIMiddleware)
+
 # CORS
 cors_origins = [
     "https://lojiz.onrender.com",

requirements.txt

@@ -12,6 +12,7 @@ charset-normalizer>=3.2.0
 
 # --- FastAPI & Web Framework ---
 fastapi==0.104.1
+slowapi>=0.1.9  # Rate Limiting
 uvicorn[standard]==0.24.0
 python-multipart==0.0.6