Merge pull request #2 from santanu2310/feature

remove mongodb watcher and replaced with rabbitMQ pub/sub

Author: santanu2310

server/app/api/friends/router.py

@@ -13,6 +13,8 @@
     Friends_Status,
     FriendRequestOut,
     AddFriendMessage,
+    FriendRequestMessage,
+    SyncMessageType,
 )
 
 from app.core.db import AsyncDatabase, get_async_database
@@ -98,10 +100,28 @@ async def make_friend_request(
     )
 
     # Inserting the request into the collection
-    await db.friend_request.insert_one(
+    result = await db.friend_request.insert_one(
         request.model_dump(by_alias=True, exclude={"id"})
     )
 
+    full_user = await get_full_user(db=db, user_id=request.sender_id)
+    user_brief = UserBrief.model_validate(full_user.model_dump())
+
+    message = FriendRequestMessage(
+        type=SyncMessageType.friend_request,
+        id=str(result.inserted_id),
+        message=request.message,
+        user=user_brief,
+        status=Friends_Status.pending,
+        created_time=request.created_at,
+    )
+
+    # Send the status update to the message sender
+    await send_message(
+        user_ids=[request.receiver_id],
+        message_data=message,
+    )
+
     return
 
 

server/app/api/sync_socket/router.py

@@ -38,6 +38,7 @@
     get_call_record,
     list_call_record,
 )
+from .schemas import OnlineStatus
 
 router = APIRouter()
 logger = logging.getLogger(__name__)
@@ -92,16 +93,21 @@ async def websocket_endpoint(
                     user_id=user.id, message=SyncPacket(type=PacketType.pong)
                 )
             elif packet.type == PacketType.message and packet.data:
-                await handle_recieved_message(db, user_id=user.id, message=packet.data)
+                await handle_recieved_message(
+                    db, user_id=user.id, message=packet.data, queue=queue_connection
+                )
 
     except WebSocketDisconnect as e:
-        logger.error(e)
+        logger.error(f"User disconnected: {e}")
         await connections.disconnect(user.id, queue_connection)
     return
 
 
 async def handle_recieved_message(
-    db: AsyncDatabase, user_id: ObjectId, message: SyncSocketMessage
+    db: AsyncDatabase,
+    user_id: ObjectId,
+    message: SyncSocketMessage,
+    queue: AbstractRobustConnection,
 ):
     if message.type == SyncMessageType.message_status:
         try:
@@ -137,6 +143,14 @@ async def handle_recieved_message(
 
             if updates:
                 await db.message.bulk_write(updates)
+
+                # Publish the updated messages to the queue for further processing
+                await publish_message(
+                    connection=queue,
+                    exchange_name=settings.EXCHANGES.sync_message.value,
+                    topic=settings.TOPICS.message_status_update.value,
+                    data=msg,
+                )
         except Exception as e:
             logger.critical(e)
 
@@ -225,7 +239,7 @@ async def notify_online_status(
     user_id: ObjectId,
     is_online: Literal["online", "offline"],
 ):
-    data = {"user_id": str(user_id), "status": is_online}
+    data = OnlineStatus(user_id=str(user_id), status=is_online)
     await publish_message(
         connection=connection,
         exchange_name=settings.EXCHANGES.sync_message.value,

server/app/api/sync_socket/schemas.py

@@ -0,0 +1,6 @@
+from pydantic import BaseModel
+
+
+class OnlineStatus(BaseModel):
+    user_id: str
+    status: str

server/app/background_tasks/async_ops/tasks.py

@@ -1,5 +1,7 @@
+from bson import ObjectId
 import logging
 from aio_pika.abc import AbstractIncomingMessage
+from pymongo.errors import PyMongoError
 from app.core.config import settings
 from app.api.sync_socket.router import send_message as send_sync_message
 from app.core.schemas import (
@@ -11,6 +13,8 @@
     UserBrief,
     FriendRequestMessage,
     SyncMessageType,
+    BrodcastMessage,
+    Message_Status,
 )
 from app.core.db import (
     AsyncDatabase,
@@ -27,36 +31,36 @@
 logger = logging.getLogger(__name__)
 
 
-async def watch_user_updates():
-    client = create_async_client()
-    db = AsyncDatabase(client, settings.DATABASE_NAME)
+# async def watch_user_updates():
+#     client = create_async_client()
+#     db = AsyncDatabase(client, settings.DATABASE_NAME)
 
-    async with db.user_profile.watch(
-        pipeline=[{"$match": {"operationType": "update"}}], full_document="updateLookup"
-    ) as stream:
-        async for change in stream:
-            try:
-                # user_id = await db.user_profile.find_one({"_id": change['documentKey']['auth_id']})
-                cursor = db.friends.find({"user_id": change["fullDocument"]["auth_id"]})
+#     async with db.user_profile.watch(
+#         pipeline=[{"$match": {"operationType": "update"}}], full_document="updateLookup"
+#     ) as stream:
+#         async for change in stream:
+#             try:
+#                 # user_id = await db.user_profile.find_one({"_id": change['documentKey']['auth_id']})
+#                 cursor = db.friends.find({"user_id": change["fullDocument"]["auth_id"]})
 
-                # Extracting the `friends_id` values from each result document
-                friend_ids = [doc["friend_id"] async for doc in cursor]
+#                 # Extracting the `friends_id` values from each result document
+#                 friend_ids = [doc["friend_id"] async for doc in cursor]
 
-                # Sending the data to the online frinds
-                data: dict[str:any] = change["updateDescription"]["updatedFields"]
-                data["id"] = change["fullDocument"]["auth_id"]
+#                 # Sending the data to the online frinds
+#                 data: dict[str:any] = change["updateDescription"]["updatedFields"]
+#                 data["id"] = change["fullDocument"]["auth_id"]
 
-                friend_update = FriendUpdateMessage.model_validate(data)
-                await send_sync_message(friend_ids, friend_update)
+#                 friend_update = FriendUpdateMessage.model_validate(data)
+#                 await send_sync_message(friend_ids, friend_update)
 
-                # updating the lastupdate fo friends data
-                await db.friends.update_many(
-                    {"friend_id": change["fullDocument"]["auth_id"]},
-                    {"$set": {"update_at": change["wallTime"]}},
-                )
+#                 # updating the lastupdate fo friends data
+#                 await db.friends.update_many(
+#                     {"friend_id": change["fullDocument"]["auth_id"]},
+#                     {"$set": {"update_at": change["wallTime"]}},
+#                 )
 
-            except Exception as e:
-                logger.error(f"Error processing user update : {e}")
+#             except Exception as e:
+#                 logger.error(f"Error processing user update : {e}")
 
 
 @rabbit_consumer(
@@ -69,52 +73,100 @@ async def handle_online_status_update(
     await distribute_online_status_update(message=message, db=db)
 
 
-async def watch_message_updates():
-    """
-    Watches for updates in the MessageCollection and sends message status updates
-    to the sender when the message status changes.
+@rabbit_consumer(
+    topic_name=settings.TOPICS.message_status_update.value,
+    exchange_name=settings.EXCHANGES.sync_message.value,
+)
+async def process_message_status_updates(
+    message: AbstractIncomingMessage, db: AsyncDatabase
+):
     """
+    Processes message status updates and forwards them to relevant senders.
 
-    client = create_async_client()
-    db = AsyncDatabase(client, settings.DATABASE_NAME)
+    This function:
+    1. Decodes incoming message status updates
+    2. Queries affected messages from database
+    3. Groups messages by sender
+    4. Forwards status updates to each sender
 
-    async with db.message.watch(
-        pipeline=[{"$match": {"operationType": "update"}}],
-        full_document="updateLookup",
-    ) as stream:
-        async for change in stream:
-            try:
-                if not change["updateDescription"]["updatedFields"]["status"]:
+    Args:
+        message: Incoming RabbitMQ message containing status update payload
+        db: Async database connection
+    """
+    try:
+        try:
+            decoded_data = message.body.decode("utf-8")
+            payload: MessageStatusUpdate = MessageStatusUpdate.model_validate_json(
+                decoded_data
+            )
+
+            message_ids = [ObjectId(data.message_id) for data in payload.data]
+
+        except (UnicodeDecodeError, ValueError) as e:
+            logger.error(f"Failed to decode or validate message payload: {e}")
+            return
+        state = (
+            "received_time"
+            if payload.status == Message_Status.recieved.value
+            else "seen_time"
+        )
+
+        try:
+            cursor = db.message.find(
+                {"_id": {"$in": message_ids}},
+                projection={"sender_id": 1, state: 1},
+            )
+        except PyMongoError as e:
+            logger.error(f"Database query failed: {e}")
+            return
+
+        sender_data: dict[str, list[dict[str, any]]] = {}
+        try:
+            async for db_message in cursor:
+                sender_id = db_message["sender_id"]
+
+                # Initialize sender list if not exists
+                if sender_id not in sender_data:
+                    sender_data[sender_id] = []
+
+                # Skip messages without timestamp
+                if db_message[state] is None:
+                    logger.warning(f"Message {db_message['_id']} has null {state}")
                     continue
 
-                message_id = change["documentKey"]["_id"]
-                updated_field = change["updateDescription"]["updatedFields"]
-                status = change["updateDescription"]["updatedFields"]["status"]
-
-                # Get the timestamp of the update
-                timestamp = next(iter(updated_field.values()))
-
-                # Fetch the updated message from the database
-                message = Message.model_validate(change["fullDocument"])
-
-                # Construct a MessageEvent Object
-                message_event = MessageEvent(
-                    message_id=str(message_id), timestamp=timestamp
+                # Add message data to sender's list
+                sender_data[sender_id].append(
+                    {"timestamp": db_message[state], "message_id": db_message["_id"]}
                 )
 
-                # Create a MessageStatusUpdate object to send to the user
-                sync_message = MessageStatusUpdate(
-                    data=[message_event],
-                    status=status,
-                )
+        except PyMongoError as e:
+            logger.error(f"Error iterating database cursor: {e}")
+            return
 
-                # Send the status update to the message sender
+        # send the data to the sender
+        for sender_id, data in sender_data.items():
+            try:
+                data = [
+                    MessageEvent(
+                        message_id=str(msg["message_id"]), timestamp=msg["timestamp"]
+                    )
+                    for msg in data
+                ]
                 await send_sync_message(
-                    user_ids=[message.sender_id], message_data=sync_message
+                    user_ids=[sender_id],
+                    message_data=MessageStatusUpdate(data=data, status=payload.status),
                 )
 
             except Exception as e:
-                print(f"Error processing user update(line:147) : {e}")
+                logger.error(f"Failed to send status update to sender {sender_id}: {e}")
+
+    except Exception as e:
+        logger.error(
+            f"Unexpected error in process_message_status_updates: {e}", exc_info=True
+        )
+    finally:
+        # retry won't happen
+        message.ack()
 
 
 @rabbit_consumer(
@@ -181,3 +233,17 @@ async def profile_media_update_confirmation(
     message: AbstractIncomingMessage,
 ):
     await send_profilemedia_update_confirmation(data=message)
+
+
+@rabbit_consumer(
+    topic_name=settings.TOPICS.chat_broadcast_selected.value,
+    exchange_name=settings.EXCHANGES.sync_message.value,
+)
+async def send_message_to_users(
+    message: AbstractIncomingMessage,
+):
+    decoded_data = message.body.decode("utf-8")
+    logger.info(f"{decoded_data=}")
+    payload: BrodcastMessage = BrodcastMessage.model_validate_json(decoded_data)
+
+    await send_sync_message(user_ids=payload.ids, message_data=payload.data)

server/app/background_tasks/celery/tasks.py

@@ -7,10 +7,18 @@
 from app.background_tasks.celery.dependency import get_dependency_manager, Dependency
 from app.core.config import create_celery_client, settings
 from app.core.message_broker import publish_bloking_message
-from app.core.schemas import MediaType, Message, ProfileMediaUpdate, UserProfile
+from app.core.schemas import (
+    MediaType,
+    Message,
+    ProfileMediaUpdate,
+    UserProfile,
+    FriendUpdateMessage,
+    BrodcastMessage,
+)
 from app.utils import get_file_extension
 from botocore.exceptions import ClientError, NoCredentialsError  # type: ignore
 
+from .utils import list_friends_id
 from .services import process_image_to_aspect, send_otp_email
 
 celery_app = create_celery_client()
@@ -169,6 +177,8 @@ def process_profile_media(file_id: str, user_id: str, media_type: str):
                 return_document=ReturnDocument.BEFORE,
             )
 
+            friends_list = list_friends_id(user_id=ObjectId(user_id), db=db)
+
         user_profile: UserProfile = UserProfile.model_validate(user_profile_response)
 
         message = ProfileMediaUpdate(
@@ -189,6 +199,19 @@ def process_profile_media(file_id: str, user_id: str, media_type: str):
                 topic=settings.TOPICS.media_update,
                 data=data,
             )
+
+            # Send the updated user data to all friends
+            data = FriendUpdateMessage(**{"id": ObjectId(user_id), media_type: new_key})
+            payload = BrodcastMessage(ids=friends_list, data=data)
+
+            publish_bloking_message(
+                connection=queue,
+                exchange_name=settings.EXCHANGES.sync_message.value,
+                topic=settings.TOPICS.chat_broadcast_selected.value,
+                data=payload.model_dump_json(),
+            )
+
+            logger.error(f"{payload=}")
     except Exception as e:
         logger.exception(
             f"Infrastructure error processing media for user {user_id}: {e}"

server/app/background_tasks/celery/utils.py

@@ -0,0 +1,11 @@
+from bson import ObjectId
+from app.core.db import SyncDatabase
+
+
+def list_friends_id(user_id: ObjectId, db: SyncDatabase) -> list[ObjectId]:
+    friends = db.friends.find(
+        {"user_id": user_id},
+        projection={"friend_id": 1, "_id": 0},
+    )
+    friend_ids = [friend["friend_id"] for friend in friends]
+    return friend_ids