Implemented mongodb schema for messaging #48

docker-compose.yml

@@ -37,7 +37,15 @@ services:
       test: ["CMD-SHELL", "pg_isready -U $$POSTGRES_USER -d $$POSTGRES_DB"]
       interval: 10s
       retries: 5
+  mongo:
+    image: mongo:6.0
+    restart: unless-stopped
+    ports: 
+      - "27017:27017"
+    volumes:
+      - mongo_data:/data/db
 
 volumes:
   postgres_data:
-  pgadmin_data:
+  pgadmin_data:
+  mongo_data:

forum/communications/consumers.py

@@ -0,0 +1,84 @@
+import json
+import logging, traceback
+from django.utils import timezone
+from channels.generic.websocket import AsyncWebsocketConsumer
+from .models_mongo import Room, Message
+
+logger = logging.getLogger("communications")
+
+class TestConsumer(AsyncWebsocketConsumer):
+    async def connect(self):
+        user = self.scope["user"]
+        if user.is_authenticated:
+            await self.accept()
+            await self.send(text_data="WebSocket connected.")
+        else:
+            await self.close()
+
+    async def receive(self, text_data):
+        await self.send(text_data=f"You sent: {text_data}")
+
+    async def disconnect(self, code):
+        print(f"WebSocket disconnected with code: {code}")
+
+class ChatConsumer(AsyncWebsocketConsumer):
+    async def connect(self):
+        user = self.scope["user"]
+        if not user.is_authenticated:
+            return await self.close()
+
+        self.room_id = self.scope["url_route"]["kwargs"]["room_id"]
+        self.room_name = f"room_{self.room_id}"
+
+        await self.channel_layer.group_add(self.room_name, self.channel_name)
+        await self.accept()
+
+    async def disconnect(self, code):
+        await self.channel_layer.group_discard(self.room_name, self.channel_name)
+        print(f"WebSocket disconnected with code: {code}")
+
+    async def receive(self, text_data, bytes_data=None):
+        try:
+            user = self.scope["user"]
+            payload = json.loads(text_data or "{}")
+            text = payload.get("text", "").strip()
+            if not text:
+                return
+
+            user_pk = user.pk
+
+            room = Room.objects(room_id=self.room_id).first()
+
+            if room is None:
+                room = Room(room_id=self.room_id, participants=[user_pk])
+            else:
+                if user_pk not in room.participants:
+                    room.participants.append(user_pk)
+                    room.updated_at = timezone.now()
+
+            room.save() 
+
+            Message(
+                room=room,
+                sender_id=user_pk,
+                text=text,
+                timestamp=timezone.now()
+            ).save()
+
+            await self.channel_layer.group_send(
+                self.room_name,
+                {
+                    "type": "chat.message", 
+                    "user": user.user_name, 
+                    "text": text
+                }
+            )
+        except Exception as e:
+            logging.error(f"Error in ChatConsumer.receive: {e} \n{traceback.format_exc()}")
+            await self.close()
+
+    async def chat_message(self, event):
+        await self.send(text_data=json.dumps({
+            "user": event["user"],
+            "text": event["text"]
+        }))

forum/communications/middleware.py

@@ -0,0 +1,68 @@
+"""
+Use UntypedToken to validate the signature and expiry
+=============================
+jwt.decode() with SECRET_KEY and ALGORITHM from our SIMPLE_JWT settings to extract user
+=============================
+get_user() fetches the real Django user
+"""
+
+
+import logging, traceback
+from urllib.parse import parse_qs
+import jwt
+from django.conf import settings
+from django.contrib.auth import get_user_model
+from django.contrib.auth.models import AnonymousUser
+from channels.db import database_sync_to_async
+from rest_framework_simplejwt.tokens import UntypedToken
+from rest_framework_simplejwt.exceptions import InvalidToken, TokenError
+
+logger = logging.getLogger("communications")
+
+User = get_user_model()
+
+@database_sync_to_async
+def get_user(user_id):
+    pk_name = User._meta.pk.name
+    try:
+        return User.objects.get(**{pk_name: user_id})
+    except User.DoesNotExist:
+        return AnonymousUser()
+
+class JwtAuthMiddleware:
+    """
+    Extracts a JWT access token from ?token=<token> and
+    populates scope['user'] if valid, or AnonymousUser
+    """
+
+    def __init__(self, app):
+        self.app = app
+
+    async def __call__(self, scope, receive, send):
+        try:
+            qs = parse_qs(scope.get("query_string", b"").decode())
+            token = qs.get("token", [None])[0]
+
+            user = AnonymousUser()
+
+            if token:
+                UntypedToken(token)
+                payload = jwt.decode(
+                    token,
+                    settings.SECRET_KEY,
+                    algorithms=[settings.SIMPLE_JWT["ALGORITHM"]],
+                    options={"verify_exp": True},
+                )
+                user = await get_user(payload[settings.SIMPLE_JWT["USER_ID_CLAIM"]])
+
+            scope["user"] = user
+            return await self.app(scope, receive, send)
+
+        except (InvalidToken, TokenError, jwt.PyJWTError) as e:
+            logger.warning(f"JWT authorization failed: {e}")
+            scope["user"] = AnonymousUser() 
+            return await self.app(scope, receive, send)
+
+        except Exception as e:
+            logger.error(f"Error in JwtAuthMiddleWare: {e}\n {traceback.format_exc()}")
+            raise

forum/communications/models_mongo.py

@@ -0,0 +1,30 @@
+from django.utils import timezone
+import mongoengine as me
+from django.contrib.auth import get_user_model
+
+User = get_user_model()
+
+class Room(me.Document):
+    meta = {
+        "db_alias": "chat",
+        "collection": "rooms",
+        "indexes": ["participants", "updated_at"]
+    }
+    room_id = me.StringField(primary_key=True, required=True)
+    participants = me.ListField(me.IntField(), default=list, required=True)
+    created_at = me.DateTimeField(required=True, default=timezone.now)
+    updated_at = me.DateTimeField(required=True, default=timezone.now)
+
+class Message(me.Document):
+    meta = {
+        "db_alias": "chat",
+        "collection": "messages",
+        "indexes": [
+            {"fields": ["room", "timestamp"]},
+        ]
+    }
+    room = me.ReferenceField(Room, required=True, reverse_delete_rule=me.CASCADE)
+    sender_id = me.IntField(required=True)
+    text = me.StringField(required=True)
+    timestamp = me.DateTimeField(required=True, default=timezone.now)
+

forum/forum/asgi.py

@@ -37,18 +37,26 @@
 """
 
 import os
+os.environ.setdefault("DJANGO_SETTINGS_MODULE", "forum.settings")
 
 from django.core.asgi import get_asgi_application
 from channels.routing import ProtocolTypeRouter, URLRouter
 from channels.auth import AuthMiddlewareStack
+from communications.middleware import JwtAuthMiddleware
 try:
     from forum.routing import websocket_urlpatterns
 except ImportError:
     websocket_urlpatterns = []
 
-os.environ.setdefault("DJANGO_SETTINGS_MODULE", "forum.settings")
 
 application = ProtocolTypeRouter({
+
     "http": get_asgi_application(),
-    "websocket": AuthMiddlewareStack(URLRouter(websocket_urlpatterns))
+
+    "websocket": JwtAuthMiddleware(
+        AuthMiddlewareStack(
+            URLRouter(websocket_urlpatterns)
+        )
+    ),
+
 })

forum/forum/routing.py

@@ -1,24 +1,9 @@
-from django.urls import path
-from channels.generic.websocket import AsyncWebsocketConsumer
-
-
-class Consumer(AsyncWebsocketConsumer):
-    async def connect(self):
-        user = self.scope["user"]
-        if user.is_authenticated:
-            await self.accept()
-            await self.send(text_data="WebSocket connected.")
-        else:
-            await self.close()
-
-    async def receive(self, text_data):
-        await self.send(text_data=f"You sent: {text_data}")
-
-    async def disconnect(self, code):
-        print(f"WebSocket disconnected with code: {code}")
+from django.urls import path, re_path
+from communications.consumers import ChatConsumer, TestConsumer
 
 websocket_urlpatterns =[
-    path("ws/test/", Consumer.as_asgi()),
+    path("ws/test/", TestConsumer.as_asgi(), name="ws-test"),
+    path("ws/chat/<str:room_id>/", ChatConsumer.as_asgi(), name="ws-chat"),
 ]
 
 

forum/forum/settings.py

@@ -20,6 +20,7 @@
 
 import os
 from datetime import timedelta
+from mongoengine import connect
 
 SECRET_KEY = os.getenv("DJANGO_SECRET_KEY", "fallback-key-for-dev-only")
 
@@ -95,6 +96,22 @@
     }
 }
 
+MONGO_DB_NAME = os.getenv("MONGO_DB", "forum_chat")
+MONGO_HOST = os.getenv("MONGO_HOST", "mongo")
+MONGO_PORT = int(os.getenv("MONGO_PORT", 27017))
+connect(
+    db=MONGO_DB_NAME,
+    host=MONGO_HOST,
+    port=MONGO_PORT,
+    alias="chat"
+)
+
+CHANNEL_LAYERS = {
+    "default": {
+        "BACKEND": "channels.layers.InMemoryChannelLayer",
+    },
+}
+
 # Password validation
 # https://docs.djangoproject.com/en/4.2/ref/settings/#auth-password-validators
 

forum/requirements.txt

@@ -10,4 +10,6 @@ django-phonenumber-field==8.0.0
 phonenumberslite==9.0.0
 djangorestframework-simplejwt==5.3.1
 channels==4.0.0
-daphne==4.0.0
+daphne==4.0.0
+mongoengine>=0.24.0
+dnspython