feat: add message reliability foundation

- implement idempotent message creation via client_message_id

- add persistent delivered/read receipts

- expose /messages/status and wire websocket receipt events

- update web client to send client ids and auto-ack delivered/read

app/messages/service.py

@@ -1,29 +1,60 @@
 from fastapi import HTTPException, status
+from sqlalchemy.exc import IntegrityError
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from app.chats.service import ensure_chat_membership
 from app.messages import repository
 from app.messages.models import Message
 from app.messages.spam_guard import enforce_message_spam_policy
-from app.messages.schemas import MessageCreateRequest, MessageUpdateRequest
+from app.messages.schemas import MessageCreateRequest, MessageStatusUpdateRequest, MessageUpdateRequest
 from app.notifications.service import dispatch_message_notifications
 
 
 async def create_chat_message(db: AsyncSession, *, sender_id: int, payload: MessageCreateRequest) -> Message:
     await ensure_chat_membership(db, chat_id=payload.chat_id, user_id=sender_id)
+    if payload.client_message_id:
+        existing = await repository.get_message_by_client_message_id(
+            db,
+            chat_id=payload.chat_id,
+            sender_id=sender_id,
+            client_message_id=payload.client_message_id,
+        )
+        if existing:
+            return existing
     if payload.type.value == "text" and not (payload.text and payload.text.strip()):
         raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail="Text message cannot be empty")
     await enforce_message_spam_policy(user_id=sender_id, chat_id=payload.chat_id, text=payload.text)
 
-    message = await repository.create_message(
-        db,
-        chat_id=payload.chat_id,
-        sender_id=sender_id,
-        message_type=payload.type,
-        text=payload.text,
-    )
-    await db.commit()
-    await db.refresh(message)
+    try:
+        message = await repository.create_message(
+            db,
+            chat_id=payload.chat_id,
+            sender_id=sender_id,
+            message_type=payload.type,
+            text=payload.text,
+        )
+        if payload.client_message_id:
+            await repository.create_message_idempotency_key(
+                db,
+                chat_id=payload.chat_id,
+                sender_id=sender_id,
+                client_message_id=payload.client_message_id,
+                message_id=message.id,
+            )
+        await db.commit()
+        await db.refresh(message)
+    except IntegrityError:
+        await db.rollback()
+        if payload.client_message_id:
+            existing = await repository.get_message_by_client_message_id(
+                db,
+                chat_id=payload.chat_id,
+                sender_id=sender_id,
+                client_message_id=payload.client_message_id,
+            )
+            if existing:
+                return existing
+        raise
     try:
         await dispatch_message_notifications(db, message)
     except Exception:
@@ -73,3 +104,43 @@ async def delete_message(db: AsyncSession, *, message_id: int, user_id: int) ->
         raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="You can delete only your own messages")
     await repository.delete_message(db, message)
     await db.commit()
+
+
+async def mark_message_status(
+    db: AsyncSession,
+    *,
+    user_id: int,
+    payload: MessageStatusUpdateRequest,
+) -> dict[str, int]:
+    await ensure_chat_membership(db, chat_id=payload.chat_id, user_id=user_id)
+    message = await repository.get_message_by_id(db, payload.message_id)
+    if not message or message.chat_id != payload.chat_id:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Message not found")
+
+    receipt = await repository.get_message_receipt(db, chat_id=payload.chat_id, user_id=user_id)
+    if not receipt:
+        last_delivered = payload.message_id if payload.status in {"message_delivered", "message_read"} else None
+        last_read = payload.message_id if payload.status == "message_read" else None
+        receipt = await repository.create_message_receipt(
+            db,
+            chat_id=payload.chat_id,
+            user_id=user_id,
+            last_delivered_message_id=last_delivered,
+            last_read_message_id=last_read,
+        )
+    else:
+        if payload.status in {"message_delivered", "message_read"}:
+            current = receipt.last_delivered_message_id or 0
+            receipt.last_delivered_message_id = max(current, payload.message_id)
+        if payload.status == "message_read":
+            current_read = receipt.last_read_message_id or 0
+            receipt.last_read_message_id = max(current_read, payload.message_id)
+
+    await db.commit()
+    await db.refresh(receipt)
+    return {
+        "chat_id": payload.chat_id,
+        "message_id": payload.message_id,
+        "last_delivered_message_id": receipt.last_delivered_message_id or 0,
+        "last_read_message_id": receipt.last_read_message_id or 0,
+    }