fyp

app/database.py

@@ -79,7 +79,13 @@ async def ensure_indexes():
         await users_col.create_index("role")
         await users_col.create_index("isActive")
         
-        # OTP indexes
+        # OTP indexes - handle TTL index conflict by dropping if options differ
+        try:
+            # Try to drop existing TTL index with different options
+            await otps_col.drop_index("createdAt_1")
+        except Exception:
+            pass  # Index doesn't exist, that's fine
+        
         await otps_col.create_index("identifier")
         await otps_col.create_index("purpose")
         await otps_col.create_index([("createdAt", 1)], expireAfterSeconds=900)  # 15 min TTL
@@ -123,6 +129,12 @@ async def ensure_chat_indexes():
         # Conversations indexes
         # Unique index on participants: ONE conversation per user pair (not per listing)
         # This enables DM-style conversations where User A & B have a single thread
+        # Drop existing non-unique index if it exists to recreate with unique constraint
+        try:
+            await conversations_col.drop_index("participants_1")
+        except Exception:
+            pass  # Index doesn't exist, that's fine
+        
         await conversations_col.create_index(
             "participants",
             unique=True

app/models/conversation.py

@@ -7,6 +7,15 @@ from typing import Optional
 class Conversation:
     """Conversation model for MongoDB"""
     
+    @staticmethod
+    def _serialize_datetime(dt) -> Optional[str]:
+        """Convert datetime to ISO string, handling None values"""
+        if dt is None:
+            return None
+        if isinstance(dt, datetime):
+            return dt.isoformat()
+        return dt  # Already a string or other serializable type
+    
     @staticmethod
     def create_document(
         listing_id: str,
@@ -41,15 +50,25 @@ class Conversation:
         if not conversation_doc:
             return None
         
+        # Serialize last_message with proper datetime handling
+        last_message = conversation_doc.get("last_message")
+        if last_message:
+            last_message = {
+                "text": last_message.get("text"),
+                "sender_id": last_message.get("sender_id"),
+                "timestamp": Conversation._serialize_datetime(last_message.get("timestamp")),
+            }
+        
         return {
             "id": str(conversation_doc.get("_id", "")),
             "listing_id": conversation_doc.get("listing_id"),
             "participants": conversation_doc.get("participants", []),
             "listing_title": conversation_doc.get("listing_title"),
             "listing_image": conversation_doc.get("listing_image"),
-            "last_message": conversation_doc.get("last_message"),
+            "last_message": last_message,
             "unread_count": conversation_doc.get("unread_count", {}),
             "status": conversation_doc.get("status", "active"),
-            "created_at": conversation_doc.get("created_at"),
-            "updated_at": conversation_doc.get("updated_at"),
+            "created_at": Conversation._serialize_datetime(conversation_doc.get("created_at")),
+            "updated_at": Conversation._serialize_datetime(conversation_doc.get("updated_at")),
         }
+

app/models/message.py

@@ -2,7 +2,12 @@
 # app/models/message.py - Message Model
 # ============================================================
 from datetime import datetime
-from typing import Optional
+from typing import Optional, List, Dict
+
+# Time windows for message actions (in minutes)
+EDIT_WINDOW_MINUTES = 15           # Can edit within 15 minutes
+DELETE_FOR_EVERYONE_WINDOW_MINUTES = 60  # Can delete for everyone within 1 hour
+
 
 class Message:
     """Message model for MongoDB"""
@@ -31,26 +36,125 @@ class Message:
             "property_card": property_card,  # {listing_id, title, price, currency, bedrooms, bathrooms, location, image_url, listing_type}
             "is_read": False,
             "read_at": None,
+            # Edit tracking
+            "is_edited": False,
+            "edited_at": None,
+            # Deletion tracking
+            "is_deleted": False,           # Deleted for everyone
+            "deleted_at": None,
+            "deleted_for": [],             # List of user_ids who deleted this message for themselves
+            # Reactions: {emoji: [user_id1, user_id2, ...]}
+            "reactions": {},
             "created_at": now,
         }
     
     @staticmethod
-    def format_response(message_doc: dict) -> dict:
-        """Format message document for API response"""
-        if not message_doc:
+    def _serialize_datetime(dt) -> Optional[str]:
+        """Convert datetime to ISO string, handling None values"""
+        if dt is None:
             return None
+        if isinstance(dt, datetime):
+            return dt.isoformat()
+        return dt  # Already a string or other serializable type
+    
+    @staticmethod
+    def _calculate_action_availability(
+        message_doc: dict,
+        for_user_id: Optional[str] = None
+    ) -> dict:
+        """
+        Calculate whether edit and delete-for-everyone actions are available.
+        
+        Returns:
+            dict with can_edit and can_delete_for_everyone flags
+        """
+        is_deleted = message_doc.get("is_deleted", False)
+        sender_id = message_doc.get("sender_id")
+        message_type = message_doc.get("message_type")
+        created_at = message_doc.get("created_at")
+        
+        # Default: no actions available
+        can_edit = False
+        can_delete_for_everyone = False
+        
+        # If message is deleted, no actions available
+        if is_deleted:
+            return {"can_edit": False, "can_delete_for_everyone": False}
+        
+        # Only sender can edit/delete for everyone
+        is_sender = for_user_id and sender_id == for_user_id
+        
+        if is_sender and created_at:
+            now = datetime.utcnow()
+            minutes_since = (now - created_at).total_seconds() / 60
+            
+            # Edit: only text messages, within 15 minutes
+            if message_type == "text" and minutes_since <= EDIT_WINDOW_MINUTES:
+                can_edit = True
+            
+            # Delete for everyone: within 1 hour
+            if minutes_since <= DELETE_FOR_EVERYONE_WINDOW_MINUTES:
+                can_delete_for_everyone = True
         
         return {
+            "can_edit": can_edit,
+            "can_delete_for_everyone": can_delete_for_everyone,
+        }
+    
+    @staticmethod
+    def format_response(message_doc: dict, for_user_id: Optional[str] = None) -> Optional[dict]:
+        """
+        Format message document for API response.
+        
+        Args:
+            message_doc: The message document from MongoDB
+            for_user_id: If provided, checks if message is deleted for this user
+                         and calculates action availability
+            
+        Returns:
+            Formatted message dict, or None if deleted for the requesting user
+        """
+        if not message_doc:
+            return None
+        
+        # If message is deleted for everyone, show placeholder
+        is_deleted = message_doc.get("is_deleted", False)
+        
+        # If deleted specifically for this user, don't return the message
+        deleted_for = message_doc.get("deleted_for", [])
+        if for_user_id and for_user_id in deleted_for:
+            return None  # Message was "deleted for me" by this user
+        
+        # Calculate action availability for frontend
+        action_availability = Message._calculate_action_availability(message_doc, for_user_id)
+        
+        # Build response
+        response = {
             "id": str(message_doc.get("_id", "")),
             "conversation_id": message_doc.get("conversation_id"),
             "sender_id": message_doc.get("sender_id"),
             "sender_name": message_doc.get("sender_name"),
             "sender_avatar": message_doc.get("sender_avatar"),
             "message_type": message_doc.get("message_type"),
-            "content": message_doc.get("content"),
-            "media": message_doc.get("media"),
-            "property_card": message_doc.get("property_card"),
+            "content": message_doc.get("content") if not is_deleted else None,
+            "media": message_doc.get("media") if not is_deleted else None,
+            "property_card": message_doc.get("property_card") if not is_deleted else None,
             "is_read": message_doc.get("is_read", False),
-            "read_at": message_doc.get("read_at"),
-            "created_at": message_doc.get("created_at"),
+            "read_at": Message._serialize_datetime(message_doc.get("read_at")),
+            # Edit fields
+            "is_edited": message_doc.get("is_edited", False),
+            "edited_at": Message._serialize_datetime(message_doc.get("edited_at")),
+            # Deletion status
+            "is_deleted": is_deleted,
+            "deleted_at": Message._serialize_datetime(message_doc.get("deleted_at")) if is_deleted else None,
+            # Reactions
+            "reactions": message_doc.get("reactions", {}),
+            "created_at": Message._serialize_datetime(message_doc.get("created_at")),
+            # Action availability flags for frontend UI
+            "can_edit": action_availability["can_edit"],
+            "can_delete_for_everyone": action_availability["can_delete_for_everyone"],
         }
+        
+        return response
+
+

app/routes/conversations.py

@@ -275,3 +275,180 @@ async def get_bulk_online_status(
         "success": True,
         "data": statuses,
     }
+
+
+# ============================================================
+# MESSAGE ACTIONS: Edit, Delete, Reactions, Clear Chat
+# ============================================================
+
+class EditMessageDto(BaseModel):
+    content: str
+
+
+class ReactionDto(BaseModel):
+    emoji: str
+
+
+@router.put("/{conversation_id}/messages/{message_id}", status_code=status.HTTP_200_OK)
+async def edit_message(
+    conversation_id: str,
+    message_id: str,
+    dto: EditMessageDto,
+    current_user: dict = Depends(get_current_user),
+):
+    """
+    Edit a message (REST fallback).
+    
+    - Only the sender can edit their own message
+    - Edit window: 15 minutes after sending
+    - Only text messages can be edited
+    
+    The edit is broadcast to all participants via WebSocket if they're connected.
+    """
+    user_id = current_user.get("user_id") or current_user.get("sub")
+    
+    if not user_id:
+        raise HTTPException(
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            detail="Could not extract user ID from token"
+        )
+    
+    logger.info(f"Edit message: conversation={conversation_id}, message={message_id}, user={user_id}")
+    
+    result = await conversation_service.edit_message(
+        conversation_id=conversation_id,
+        message_id=message_id,
+        user_id=user_id,
+        new_content=dto.content,
+    )
+    
+    return result
+
+
+@router.delete("/{conversation_id}/messages/{message_id}", status_code=status.HTTP_200_OK)
+async def delete_message(
+    conversation_id: str,
+    message_id: str,
+    delete_for: str = Query("me", regex="^(everyone|me)$"),
+    current_user: dict = Depends(get_current_user),
+):
+    """
+    Delete a message (REST fallback).
+    
+    - **delete_for=everyone**: Delete for all participants (sender only, within 1 hour)
+    - **delete_for=me**: Delete only for yourself
+    
+    The deletion is broadcast via WebSocket if deleting for everyone.
+    """
+    user_id = current_user.get("user_id") or current_user.get("sub")
+    
+    if not user_id:
+        raise HTTPException(
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            detail="Could not extract user ID from token"
+        )
+    
+    logger.info(f"Delete message: conversation={conversation_id}, message={message_id}, for={delete_for}, user={user_id}")
+    
+    result = await conversation_service.delete_message(
+        conversation_id=conversation_id,
+        message_id=message_id,
+        user_id=user_id,
+        delete_for=delete_for,
+    )
+    
+    return result
+
+
+@router.post("/{conversation_id}/messages/{message_id}/reactions", status_code=status.HTTP_200_OK)
+async def add_reaction(
+    conversation_id: str,
+    message_id: str,
+    dto: ReactionDto,
+    current_user: dict = Depends(get_current_user),
+):
+    """
+    Add an emoji reaction to a message (REST fallback).
+    
+    The reaction is broadcast to all participants via WebSocket.
+    """
+    user_id = current_user.get("user_id") or current_user.get("sub")
+    
+    if not user_id:
+        raise HTTPException(
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            detail="Could not extract user ID from token"
+        )
+    
+    logger.info(f"Add reaction: conversation={conversation_id}, message={message_id}, emoji={dto.emoji}, user={user_id}")
+    
+    result = await conversation_service.add_reaction(
+        conversation_id=conversation_id,
+        message_id=message_id,
+        user_id=user_id,
+        emoji=dto.emoji,
+    )
+    
+    return result
+
+
+@router.delete("/{conversation_id}/messages/{message_id}/reactions/{emoji}", status_code=status.HTTP_200_OK)
+async def remove_reaction(
+    conversation_id: str,
+    message_id: str,
+    emoji: str,
+    current_user: dict = Depends(get_current_user),
+):
+    """
+    Remove an emoji reaction from a message (REST fallback).
+    
+    The removal is broadcast to all participants via WebSocket.
+    """
+    user_id = current_user.get("user_id") or current_user.get("sub")
+    
+    if not user_id:
+        raise HTTPException(
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            detail="Could not extract user ID from token"
+        )
+    
+    logger.info(f"Remove reaction: conversation={conversation_id}, message={message_id}, emoji={emoji}, user={user_id}")
+    
+    result = await conversation_service.remove_reaction(
+        conversation_id=conversation_id,
+        message_id=message_id,
+        user_id=user_id,
+        emoji=emoji,
+    )
+    
+    return result
+
+
+@router.delete("/{conversation_id}/clear", status_code=status.HTTP_200_OK)
+async def clear_chat(
+    conversation_id: str,
+    current_user: dict = Depends(get_current_user),
+):
+    """
+    Clear all messages in a conversation for the current user only.
+    
+    Other participants will still see the messages.
+    This operation is NOT broadcast to other users.
+    """
+    user_id = current_user.get("user_id") or current_user.get("sub")
+    
+    if not user_id:
+        raise HTTPException(
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            detail="Could not extract user ID from token"
+        )
+    
+    logger.info(f"Clear chat: conversation={conversation_id}, user={user_id}")
+    
+    result = await conversation_service.clear_chat(
+        conversation_id=conversation_id,
+        user_id=user_id,
+    )
+    
+    return result
+

app/routes/websocket_chat.py

@@ -106,6 +106,11 @@ async def websocket_chat_endpoint(websocket: WebSocket, token: str = Query(...))
     - typing: User is typing
     - mark_read: Mark messages as read
     - heartbeat: Keep connection alive
+    - edit_message: Edit a message (sender only)
+    - delete_message: Delete a message (for everyone or for me)
+    - add_reaction: Add emoji reaction to a message
+    - remove_reaction: Remove emoji reaction from a message
+    - clear_chat: Clear all messages for current user
     
     Message Types (Server -> Client):
     - new_message: New message received
@@ -113,6 +118,11 @@ async def websocket_chat_endpoint(websocket: WebSocket, token: str = Query(...))
     - message_read: Messages were read
     - user_status_changed: User came online/offline
     - heartbeat_ack: Response to heartbeat
+    - message_edited: A message was edited
+    - message_deleted: A message was deleted for everyone
+    - reaction_added: Reaction added to a message
+    - reaction_removed: Reaction removed from a message
+    - chat_cleared: Chat was cleared (confirmation)
     """
     
     # Verify JWT token
@@ -188,6 +198,23 @@ async def websocket_chat_endpoint(websocket: WebSocket, token: str = Query(...))
                     "action": "online_status",
                     "statuses": statuses
                 })
+            
+            # === NEW ACTIONS ===
+            
+            elif action == "edit_message":
+                await handle_edit_message(user_id, message)
+            
+            elif action == "delete_message":
+                await handle_delete_message(user_id, message)
+            
+            elif action == "add_reaction":
+                await handle_add_reaction(user_id, message)
+            
+            elif action == "remove_reaction":
+                await handle_remove_reaction(user_id, message)
+            
+            elif action == "clear_chat":
+                await handle_clear_chat(user_id, websocket, message)
     
     except WebSocketDisconnect:
         chat_manager.disconnect(websocket)
@@ -397,6 +424,388 @@ async def broadcast_user_status(user_id: str, is_online: bool):
                 await chat_manager.send_to_user(participant_id, status_message)
 
 
+# ============================================================
+# NEW HANDLERS: Edit, Delete, Reactions, Clear Chat
+# ============================================================
+
+async def handle_edit_message(user_id: str, data: dict):
+    """
+    Handle editing a message via WebSocket.
+    Only the sender can edit, within 24 hours, text messages only.
+    Broadcasts the edit to all participants.
+    """
+    conversation_id = data.get("conversation_id")
+    message_id = data.get("message_id")
+    new_content = data.get("new_content", "").strip()
+    
+    if not conversation_id or not message_id or not new_content:
+        return
+    
+    db = await get_db()
+    
+    # Validate IDs
+    if not ObjectId.is_valid(conversation_id) or not ObjectId.is_valid(message_id):
+        return
+    
+    # Get the message
+    message_doc = await db.messages.find_one({"_id": ObjectId(message_id)})
+    if not message_doc:
+        return
+    
+    # Verify sender is the one editing
+    if message_doc.get("sender_id") != user_id:
+        logger.warning(f"[Chat WS] User {user_id} tried to edit message they didn't send")
+        return
+    
+    # Verify message is in the correct conversation
+    if message_doc.get("conversation_id") != conversation_id:
+        return
+    
+    # Only allow editing text messages
+    if message_doc.get("message_type") != "text":
+        logger.warning(f"[Chat WS] Cannot edit non-text message")
+        return
+    
+    # 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:
+            logger.warning(f"[Chat WS] Edit window expired for message {message_id}")
+            return
+    
+    # Cannot edit deleted messages
+    if message_doc.get("is_deleted"):
+        return
+    
+    # Update the message
+    now = datetime.utcnow()
+    await db.messages.update_one(
+        {"_id": ObjectId(message_id)},
+        {
+            "$set": {
+                "content": new_content,
+                "is_edited": True,
+                "edited_at": now,
+            }
+        }
+    )
+    
+    # Get conversation for participants
+    conversation = await db.conversations.find_one({"_id": ObjectId(conversation_id)})
+    if not conversation:
+        return
+    
+    # Broadcast to all participants
+    edit_broadcast = {
+        "action": "message_edited",
+        "conversation_id": conversation_id,
+        "message_id": message_id,
+        "new_content": new_content,
+        "edited_at": now.isoformat(),
+        "edited_by": user_id,
+    }
+    
+    await chat_manager.broadcast_to_conversation(
+        conversation_id,
+        conversation["participants"],
+        edit_broadcast
+    )
+    
+    logger.info(f"[Chat WS] Message {message_id} edited by {user_id}")
+
+
+async def handle_delete_message(user_id: str, data: dict):
+    """
+    Handle deleting a message via WebSocket.
+    
+    delete_for options:
+    - "everyone": Marks message as deleted for all (sender only, within 1 hour)
+    - "me": Hides message only for the requesting user
+    """
+    conversation_id = data.get("conversation_id")
+    message_id = data.get("message_id")
+    delete_for = data.get("delete_for", "me")  # "everyone" or "me"
+    
+    if not conversation_id or not message_id:
+        return
+    
+    db = await get_db()
+    
+    # Validate IDs
+    if not ObjectId.is_valid(conversation_id) or not ObjectId.is_valid(message_id):
+        return
+    
+    # Get the message
+    message_doc = await db.messages.find_one({"_id": ObjectId(message_id)})
+    if not message_doc:
+        return
+    
+    # Verify message is in the correct conversation
+    if message_doc.get("conversation_id") != conversation_id:
+        return
+    
+    # Get conversation for validation and broadcast
+    conversation = await db.conversations.find_one({"_id": ObjectId(conversation_id)})
+    if not conversation:
+        return
+    
+    # Verify user is a participant
+    if user_id not in conversation.get("participants", []):
+        return
+    
+    now = datetime.utcnow()
+    
+    if delete_for == "everyone":
+        # Only sender can delete for everyone
+        if message_doc.get("sender_id") != user_id:
+            logger.warning(f"[Chat WS] User {user_id} tried to delete message for everyone but isn't sender")
+            return
+        
+        # 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:
+                logger.warning(f"[Chat WS] Delete-for-everyone window expired for message {message_id}")
+                return
+        
+        # Mark as deleted for everyone
+        await db.messages.update_one(
+            {"_id": ObjectId(message_id)},
+            {
+                "$set": {
+                    "is_deleted": True,
+                    "deleted_at": now,
+                }
+            }
+        )
+        
+        # Broadcast to all participants
+        delete_broadcast = {
+            "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["participants"],
+            delete_broadcast
+        )
+        
+        logger.info(f"[Chat WS] Message {message_id} deleted for everyone by {user_id}")
+    
+    else:  # delete_for == "me"
+        # Add user to deleted_for list
+        await db.messages.update_one(
+            {"_id": ObjectId(message_id)},
+            {
+                "$addToSet": {"deleted_for": user_id}
+            }
+        )
+        
+        # Only send confirmation to the requesting user (not broadcast)
+        delete_confirm = {
+            "action": "message_deleted",
+            "conversation_id": conversation_id,
+            "message_id": message_id,
+            "deleted_for": "me",
+            "deleted_at": now.isoformat(),
+        }
+        
+        await chat_manager.send_to_user(user_id, delete_confirm)
+        
+        logger.info(f"[Chat WS] Message {message_id} deleted for {user_id} only")
+
+
+async def handle_add_reaction(user_id: str, data: dict):
+    """
+    Handle adding an emoji reaction to a message.
+    Broadcasts the reaction to all participants.
+    """
+    conversation_id = data.get("conversation_id")
+    message_id = data.get("message_id")
+    emoji = data.get("emoji", "").strip()
+    
+    if not conversation_id or not message_id or not emoji:
+        return
+    
+    db = await get_db()
+    
+    # Validate IDs
+    if not ObjectId.is_valid(conversation_id) or not ObjectId.is_valid(message_id):
+        return
+    
+    # Get the message
+    message_doc = await db.messages.find_one({"_id": ObjectId(message_id)})
+    if not message_doc:
+        return
+    
+    # Verify message is in the correct conversation
+    if message_doc.get("conversation_id") != conversation_id:
+        return
+    
+    # Get conversation for validation and broadcast
+    conversation = await db.conversations.find_one({"_id": ObjectId(conversation_id)})
+    if not conversation:
+        return
+    
+    # Verify user is a participant
+    if user_id not in conversation.get("participants", []):
+        return
+    
+    # Cannot react to deleted messages
+    if message_doc.get("is_deleted"):
+        return
+    
+    # Add reaction: reactions is a dict like {"👍": ["user1", "user2"]}
+    await db.messages.update_one(
+        {"_id": ObjectId(message_id)},
+        {
+            "$addToSet": {f"reactions.{emoji}": user_id}
+        }
+    )
+    
+    # Broadcast to all participants
+    reaction_broadcast = {
+        "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["participants"],
+        reaction_broadcast
+    )
+    
+    logger.info(f"[Chat WS] Reaction {emoji} added to message {message_id} by {user_id}")
+
+
+async def handle_remove_reaction(user_id: str, data: dict):
+    """
+    Handle removing an emoji reaction from a message.
+    Broadcasts the removal to all participants.
+    """
+    conversation_id = data.get("conversation_id")
+    message_id = data.get("message_id")
+    emoji = data.get("emoji", "").strip()
+    
+    if not conversation_id or not message_id or not emoji:
+        return
+    
+    db = await get_db()
+    
+    # Validate IDs
+    if not ObjectId.is_valid(conversation_id) or not ObjectId.is_valid(message_id):
+        return
+    
+    # Get the message
+    message_doc = await db.messages.find_one({"_id": ObjectId(message_id)})
+    if not message_doc:
+        return
+    
+    # Verify message is in the correct conversation
+    if message_doc.get("conversation_id") != conversation_id:
+        return
+    
+    # Get conversation for validation and broadcast
+    conversation = await db.conversations.find_one({"_id": ObjectId(conversation_id)})
+    if not conversation:
+        return
+    
+    # Verify user is a participant
+    if user_id not in conversation.get("participants", []):
+        return
+    
+    # Remove user from the emoji's reactions list
+    await db.messages.update_one(
+        {"_id": ObjectId(message_id)},
+        {
+            "$pull": {f"reactions.{emoji}": user_id}
+        }
+    )
+    
+    # Clean up empty reaction arrays (optional optimization)
+    # If an emoji has no more users, remove the key
+    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 to all participants
+    reaction_broadcast = {
+        "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["participants"],
+        reaction_broadcast
+    )
+    
+    logger.info(f"[Chat WS] Reaction {emoji} removed from message {message_id} by {user_id}")
+
+
+async def handle_clear_chat(user_id: str, websocket: WebSocket, data: dict):
+    """
+    Handle clearing all messages in a conversation for the current user only.
+    Other participants still see the messages.
+    """
+    conversation_id = data.get("conversation_id")
+    
+    if not conversation_id:
+        return
+    
+    db = await get_db()
+    
+    # Validate ID
+    if not ObjectId.is_valid(conversation_id):
+        return
+    
+    # Get conversation for validation
+    conversation = await db.conversations.find_one({"_id": ObjectId(conversation_id)})
+    if not conversation:
+        return
+    
+    # Verify user is a participant
+    if user_id not in conversation.get("participants", []):
+        return
+    
+    # Add user to deleted_for for ALL messages in this conversation
+    result = await db.messages.update_many(
+        {"conversation_id": conversation_id},
+        {"$addToSet": {"deleted_for": user_id}}
+    )
+    
+    # Send confirmation to the requesting user only
+    clear_confirm = {
+        "action": "chat_cleared",
+        "conversation_id": conversation_id,
+        "cleared_count": result.modified_count,
+        "cleared_at": datetime.utcnow().isoformat(),
+    }
+    
+    await websocket.send_json(clear_confirm)
+    
+    logger.info(f"[Chat WS] Chat {conversation_id} cleared for {user_id} ({result.modified_count} messages)")
+
+
 # Export manager for use in other modules
 def get_chat_manager():
     return chat_manager
+

app/services/conversation_service.py

@@ -171,8 +171,11 @@ class ConversationService:
                 detail="You are not a participant in this conversation"
             )
         
-        # Build query
-        query = {"conversation_id": conversation_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
+        }
         
         if before_id and ObjectId.is_valid(before_id):
             query["_id"] = {"$lt": ObjectId(before_id)}
@@ -183,7 +186,10 @@ class ConversationService:
         
         messages = []
         async for doc in cursor:
-            messages.append(Message.format_response(doc))
+            # 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}")
         
@@ -331,7 +337,365 @@ class ConversationService:
             "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,
+                }
+            }
+        )
+        
+        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,
+                    }
+                }
+            )
+            
+            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}}
+        )
+        
+        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}": ""}}
+                )
+        
+        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."""
+        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"
+            )
+        
+        # Add user to deleted_for for ALL messages in this conversation
+        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} ({result.modified_count} messages)")
+        
+        return {
+            "success": True,
+            "conversation_id": conversation_id,
+            "cleared_count": result.modified_count,
+            "cleared_at": datetime.utcnow().isoformat(),
+        }
 
 
 # Singleton instance
 conversation_service = ConversationService()
+