From 59ba1757877ea66a62cec052bc5a86caf6725bf6 Mon Sep 17 00:00:00 2001 From: Hlib Sokolovskyi Date: Mon, 16 Jun 2025 14:04:52 +0200 Subject: [PATCH] implemented mongodb schema for messaging --- docker-compose.yml | 10 +++- forum/communications/consumers.py | 84 ++++++++++++++++++++++++++++ forum/communications/middleware.py | 68 ++++++++++++++++++++++ forum/communications/models_mongo.py | 30 ++++++++++ forum/forum/asgi.py | 12 +++- forum/forum/routing.py | 23 ++------ forum/forum/settings.py | 17 ++++++ forum/requirements.txt | 4 +- 8 files changed, 225 insertions(+), 23 deletions(-) create mode 100644 forum/communications/consumers.py create mode 100644 forum/communications/middleware.py create mode 100644 forum/communications/models_mongo.py diff --git a/docker-compose.yml b/docker-compose.yml index 58849ec..dbbb6fe 100644 --- a/docker-compose.yml +++ b/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: \ No newline at end of file + pgadmin_data: + mongo_data: \ No newline at end of file diff --git a/forum/communications/consumers.py b/forum/communications/consumers.py new file mode 100644 index 0000000..941c9dd --- /dev/null +++ b/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"] + })) \ No newline at end of file diff --git a/forum/communications/middleware.py b/forum/communications/middleware.py new file mode 100644 index 0000000..b3f9192 --- /dev/null +++ b/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= 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 \ No newline at end of file diff --git a/forum/communications/models_mongo.py b/forum/communications/models_mongo.py new file mode 100644 index 0000000..5eff5ff --- /dev/null +++ b/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) + \ No newline at end of file diff --git a/forum/forum/asgi.py b/forum/forum/asgi.py index 105fe20..8d96e9a 100644 --- a/forum/forum/asgi.py +++ b/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) + ) + ), + }) diff --git a/forum/forum/routing.py b/forum/forum/routing.py index 9ccbbd8..089db90 100644 --- a/forum/forum/routing.py +++ b/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//", ChatConsumer.as_asgi(), name="ws-chat"), ] diff --git a/forum/forum/settings.py b/forum/forum/settings.py index 0fe0952..83b3ee1 100644 --- a/forum/forum/settings.py +++ b/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 diff --git a/forum/requirements.txt b/forum/requirements.txt index 5039be2..7ddc90d 100644 --- a/forum/requirements.txt +++ b/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 \ No newline at end of file +daphne==4.0.0 +mongoengine>=0.24.0 +dnspython \ No newline at end of file