Merge pull request #178 from AbdulmujibOladayo/main

Typing event broadcast to all participants, not persisted

Author: BigBen-7

src/chat.py

@@ -2,7 +2,7 @@
 import asyncio
 import json
 from datetime import datetime
-from typing import Any, Dict, List, Optional, Set
+from typing import Any, Dict, List, Literal, Optional, Set, Union
 
 from fastapi import WebSocket
 from pydantic import BaseModel
@@ -27,6 +27,24 @@ class ChatMessage(BaseModel):
     metadata: Optional[Dict[str, Any]] = None
 
 
+class TypingEvent(BaseModel):
+    """Ephemeral event indicating a participant is typing."""
+
+    type: Literal["typing"] = "typing"
+    sender_id: str
+    conversation_id: str
+    is_typing: bool
+
+
+class ReadReceiptEvent(BaseModel):
+    """Ephemeral event indicating messages have been read."""
+
+    type: Literal["read_receipt"] = "read_receipt"
+    sender_id: str
+    conversation_id: str
+    last_read_message_id: str
+
+
 class EscalationEvent(BaseModel):
     """Represents an escalation event."""
 
@@ -119,6 +137,33 @@ async def send_message(self, message: ChatMessage) -> bool:
 
         return True
 
+    async def broadcast_event(
+        self, event: Union[TypingEvent, ReadReceiptEvent]
+    ) -> bool:
+        """Broadcast an ephemeral event to all participants without persisting it."""
+        conversation_id = event.conversation_id
+        if conversation_id not in self.active_connections:
+            return True
+
+        disconnected: List[WebSocket] = []
+        for websocket in self.active_connections[conversation_id]:
+            try:
+                await websocket.send_text(event.model_dump_json())
+            except Exception as exc:
+                log_warning(
+                    "Failed to send event to websocket",
+                    {"conversation_id": conversation_id, "error": str(exc)},
+                )
+                disconnected.append(websocket)
+
+        if disconnected:
+            async with self._lock:
+                for ws in disconnected:
+                    if ws in self.active_connections.get(conversation_id, []):
+                        self.active_connections[conversation_id].remove(ws)
+
+        return True
+
     def get_message_history(
         self, conversation_id: str, limit: int = 50
     ) -> List[ChatMessage]:

src/main.py

@@ -17,7 +17,7 @@
 from src.auth.dependencies import require_admin_key, require_service_key
 
 from src.analytics.service import analytics_service
-from src.chat import ChatMessage, EscalationEvent, chat_manager
+from src.chat import ChatMessage, EscalationEvent, ReadReceiptEvent, TypingEvent, chat_manager
 from src.config import get_settings
 from src.core.ratelimit import limiter
 from src.etl import diff_etl_output, extract_events_and_sales, run_etl_once, transform_summary
@@ -79,6 +79,8 @@
     ChatMessageHistoryResponse,
     ChatMessageSendRequest,
     ChatMessageSendResponse,
+    ChatTypingRequest,
+    ChatTypingResponse,
     ChatUserConversationsResponse,
     DailyReportRequest,
     DailyReportResponse,
@@ -747,16 +749,32 @@ async def websocket_chat(
             data = await websocket.receive_text()
             try:
                 message_data: Dict[str, Any] = json.loads(data)
-                message = ChatMessage(
-                    id=str(uuid.uuid4()),
-                    sender_id=user_id,
-                    sender_type=message_data.get("sender_type", "user"),
-                    content=message_data["content"],
-                    timestamp=datetime.utcnow(),
-                    conversation_id=conversation_id,
-                    metadata=message_data.get("metadata", {}),
-                )
-                await chat_manager.send_message(message)
+                msg_type = message_data.get("type")
+                if msg_type == "typing":
+                    event = TypingEvent(
+                        sender_id=user_id,
+                        conversation_id=conversation_id,
+                        is_typing=message_data.get("is_typing", False),
+                    )
+                    await chat_manager.broadcast_event(event)
+                elif msg_type == "read_receipt":
+                    event = ReadReceiptEvent(
+                        sender_id=user_id,
+                        conversation_id=conversation_id,
+                        last_read_message_id=message_data["last_read_message_id"],
+                    )
+                    await chat_manager.broadcast_event(event)
+                else:
+                    message = ChatMessage(
+                        id=str(uuid.uuid4()),
+                        sender_id=user_id,
+                        sender_type=message_data.get("sender_type", "user"),
+                        content=message_data["content"],
+                        timestamp=datetime.utcnow(),
+                        conversation_id=conversation_id,
+                        metadata=message_data.get("metadata", {}),
+                    )
+                    await chat_manager.send_message(message)
             except json.JSONDecodeError:
                 logger.warning("Invalid JSON received from client")
             except KeyError as exc:
@@ -801,6 +819,24 @@ async def send_message(
         raise HTTPException(status_code=500, detail="Failed to send message")
 
 
+@app.post("/chat/{conversation_id}/typing", response_model=ChatTypingResponse)
+async def send_typing_indicator(
+    conversation_id: str, body: ChatTypingRequest
+) -> ChatTypingResponse:
+    """Broadcast a typing indicator to all participants in a conversation (not persisted)."""
+    try:
+        event = TypingEvent(
+            sender_id=body.sender_id,
+            conversation_id=conversation_id,
+            is_typing=body.is_typing,
+        )
+        await chat_manager.broadcast_event(event)
+        return ChatTypingResponse(status="success")
+    except Exception as exc:
+        logger.error("Error sending typing indicator: %s", exc)
+        raise HTTPException(status_code=500, detail="Failed to send typing indicator")
+
+
 @app.get("/chat/{conversation_id}/history", response_model=ChatMessageHistoryResponse)
 async def get_message_history(
     conversation_id: str,

src/types_custom.py

@@ -216,6 +216,17 @@ class ChatUserConversationsResponse(BaseModel):
     count: int
 
 
+class ChatTypingRequest(BaseModel):
+    model_config = ConfigDict(extra="forbid")
+    sender_id: str = Field(..., min_length=1)
+    is_typing: bool
+
+
+class ChatTypingResponse(BaseModel):
+    model_config = ConfigDict(extra="forbid")
+    status: Literal["success"]
+
+
 class AnalyticsStatsQuery(BaseModel):
     model_config = ConfigDict(extra="forbid")
     event_id: Optional[str] = None

tests/test_chat.py

@@ -6,7 +6,7 @@
 from unittest.mock import AsyncMock, Mock, patch
 from fastapi.testclient import TestClient
 from src.main import app
-from src.chat import ChatManager, ChatMessage, EscalationEvent
+from src.chat import ChatManager, ChatMessage, EscalationEvent, ReadReceiptEvent, TypingEvent
 
 
 @pytest.fixture
@@ -344,4 +344,192 @@ async def test_cleanup_on_websocket_disconnect(chat_manager):
     await chat_manager.disconnect(mock_websocket, conversation_id, user_id)
 
     # Check cleanup
-    assert len(chat_manager.active_connections.get(conversation_id, [])) == 0
+    assert len(chat_manager.active_connections.get(conversation_id, [])) == 0
+
+
+# ---------------------------------------------------------------------------
+# Typing indicators and read receipts
+# ---------------------------------------------------------------------------
+
+@pytest.mark.asyncio
+async def test_typing_event_broadcast_not_persisted(chat_manager):
+    """Typing events are broadcast to all participants but not stored in message history."""
+    conversation_id = "typing_test_conv"
+
+    websockets = [AsyncMock() for _ in range(2)]
+    for i, ws in enumerate(websockets):
+        await chat_manager.connect(ws, conversation_id, f"user_{i}")
+
+    event = TypingEvent(
+        sender_id="user_0",
+        conversation_id=conversation_id,
+        is_typing=True,
+    )
+    await chat_manager.broadcast_event(event)
+
+    # All participants should receive the event
+    for ws in websockets:
+        ws.send_text.assert_called()
+
+    # Nothing should be stored in message history
+    assert chat_manager.message_history.get(conversation_id, []) == []
+
+
+@pytest.mark.asyncio
+async def test_typing_event_stop_broadcast_not_persisted(chat_manager):
+    """is_typing=False is broadcast and not persisted."""
+    conversation_id = "typing_stop_conv"
+
+    ws = AsyncMock()
+    await chat_manager.connect(ws, conversation_id, "user_1")
+
+    event = TypingEvent(
+        sender_id="user_1",
+        conversation_id=conversation_id,
+        is_typing=False,
+    )
+    await chat_manager.broadcast_event(event)
+
+    ws.send_text.assert_called_once()
+    assert chat_manager.message_history.get(conversation_id, []) == []
+
+
+@pytest.mark.asyncio
+async def test_read_receipt_broadcast_not_persisted(chat_manager):
+    """Read receipt events are broadcast to all participants but not stored in message history."""
+    conversation_id = "receipt_test_conv"
+
+    websockets = [AsyncMock() for _ in range(3)]
+    for i, ws in enumerate(websockets):
+        await chat_manager.connect(ws, conversation_id, f"user_{i}")
+
+    event = ReadReceiptEvent(
+        sender_id="user_0",
+        conversation_id=conversation_id,
+        last_read_message_id="msg_42",
+    )
+    await chat_manager.broadcast_event(event)
+
+    for ws in websockets:
+        ws.send_text.assert_called()
+
+    assert chat_manager.message_history.get(conversation_id, []) == []
+
+
+@pytest.mark.asyncio
+async def test_broadcast_event_no_active_connections(chat_manager):
+    """broadcast_event returns True when there are no active connections."""
+    event = TypingEvent(
+        sender_id="user_x",
+        conversation_id="empty_conv",
+        is_typing=True,
+    )
+    result = await chat_manager.broadcast_event(event)
+    assert result is True
+
+
+@pytest.mark.asyncio
+async def test_typing_event_payload_shape(chat_manager):
+    """The JSON sent for a typing event contains the correct fields."""
+    conversation_id = "shape_test_conv"
+    ws = AsyncMock()
+    await chat_manager.connect(ws, conversation_id, "user_1")
+
+    event = TypingEvent(
+        sender_id="user_1",
+        conversation_id=conversation_id,
+        is_typing=True,
+    )
+    await chat_manager.broadcast_event(event)
+
+    ws.send_text.assert_called_once()
+    payload = json.loads(ws.send_text.call_args[0][0])
+    assert payload["type"] == "typing"
+    assert payload["sender_id"] == "user_1"
+    assert payload["conversation_id"] == conversation_id
+    assert payload["is_typing"] is True
+
+
+@pytest.mark.asyncio
+async def test_read_receipt_payload_shape(chat_manager):
+    """The JSON sent for a read receipt event contains the correct fields."""
+    conversation_id = "receipt_shape_conv"
+    ws = AsyncMock()
+    await chat_manager.connect(ws, conversation_id, "user_2")
+
+    event = ReadReceiptEvent(
+        sender_id="user_2",
+        conversation_id=conversation_id,
+        last_read_message_id="msg_99",
+    )
+    await chat_manager.broadcast_event(event)
+
+    ws.send_text.assert_called_once()
+    payload = json.loads(ws.send_text.call_args[0][0])
+    assert payload["type"] == "read_receipt"
+    assert payload["sender_id"] == "user_2"
+    assert payload["last_read_message_id"] == "msg_99"
+
+
+@pytest.mark.asyncio
+async def test_regular_message_still_persisted_after_typing(chat_manager):
+    """Regular chat messages are still persisted even after typing events are broadcast."""
+    conversation_id = "mixed_conv"
+    ws = AsyncMock()
+    await chat_manager.connect(ws, conversation_id, "user_1")
+
+    # Broadcast a typing event (should not persist)
+    typing_event = TypingEvent(
+        sender_id="user_1",
+        conversation_id=conversation_id,
+        is_typing=True,
+    )
+    await chat_manager.broadcast_event(typing_event)
+
+    # Send a real message (should persist)
+    message = ChatMessage(
+        id="msg_real",
+        sender_id="user_1",
+        sender_type="user",
+        content="Here is my message",
+        timestamp=datetime.utcnow(),
+        conversation_id=conversation_id,
+    )
+    await chat_manager.send_message(message)
+
+    history = chat_manager.get_message_history(conversation_id)
+    assert len(history) == 1
+    assert history[0].id == "msg_real"
+
+
+# ---------------------------------------------------------------------------
+# HTTP typing endpoint
+# ---------------------------------------------------------------------------
+
+def test_typing_endpoint_broadcasts_and_returns_success(client):
+    """POST /chat/{conversation_id}/typing returns 200 with status=success."""
+    response = client.post(
+        "/chat/conv_typing_http/typing",
+        json={"sender_id": "user_1", "is_typing": True},
+    )
+    assert response.status_code == 200
+    assert response.json()["status"] == "success"
+
+
+def test_typing_endpoint_is_typing_false(client):
+    """POST /chat/{conversation_id}/typing with is_typing=False returns success."""
+    response = client.post(
+        "/chat/conv_typing_stop/typing",
+        json={"sender_id": "user_2", "is_typing": False},
+    )
+    assert response.status_code == 200
+    assert response.json()["status"] == "success"
+
+
+def test_typing_endpoint_missing_sender_id(client):
+    """POST /chat/{conversation_id}/typing with missing sender_id returns 422."""
+    response = client.post(
+        "/chat/conv_typing_bad/typing",
+        json={"is_typing": True},
+    )
+    assert response.status_code == 422