""" WebSocket consumer for real-time chat and audio calls. """ import json import asyncio from datetime import datetime from channels.generic.websocket import AsyncJsonWebsocketConsumer from channels.db import database_sync_to_async from rest_framework.authtoken.models import Token from urllib.parse import parse_qs from chat.online import online_manager from .services import can_message_user, get_or_create_conversation # In-memory call state tracking # Key: canonical call ID (sorted user IDs), Value: call state dict active_calls = {} # Key: conversation_id, Value: set of user_ids active_group_calls = {} class ChatConsumer(AsyncJsonWebsocketConsumer): """ WebSocket consumer for handling real-time chat messages. Authentication: Token passed via query string (?token=xxx) Events handled: - new_message: Send a new message - new_message: Send a new message - typing_status: User typing status update - mark_read: Mark messages as read - join_conversation: Join a specific conversation room """ def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) self.user = None self.conversation_ids = set() async def connect(self): """ Handle WebSocket connection. Authenticate user via token in query string. """ # Get token from query string query_string = self.scope.get('query_string', b'').decode() params = parse_qs(query_string) token_key = params.get('token', [None])[0] if not token_key: await self.close(code=4001) return # Authenticate user self.user = await self.get_user_from_token(token_key) if not self.user: await self.close(code=4001) return # Accept connection await self.accept() # Mark user as online await online_manager.set_online(self.user.id, self.channel_name) # Join user's personal group for direct notifications await self.channel_layer.group_add( f"user_{self.user.id}", self.channel_name ) # Auto-join all user's conversation groups conversation_ids = await self.get_user_conversation_ids() for conv_id in conversation_ids: await online_manager.join_conversation_group(conv_id, self.channel_name) self.conversation_ids.add(conv_id) # Send connected confirmation await self.send_json({ "type": "connected", "user_id": self.user.id, "conversations": [str(cid) for cid in self.conversation_ids] }) async def disconnect(self, close_code): """ Handle WebSocket disconnection. """ if self.user: # Mark user as offline await online_manager.set_offline(self.user.id, self.channel_name) # Leave personal group await self.channel_layer.group_discard( f"user_{self.user.id}", self.channel_name ) # Leave all conversation groups for conv_id in self.conversation_ids: # Cleanup group call participation if conv_id in active_group_calls and self.user.id in active_group_calls[conv_id]: active_group_calls[conv_id].remove(self.user.id) # Notify others normalized_id = str(conv_id).replace('-', '') await self.channel_layer.group_send( f"chat_{normalized_id}", { "type": "group_call_participant_left", "conversation_id": conv_id, "user_id": self.user.id } ) if not active_group_calls[conv_id]: del active_group_calls[conv_id] await online_manager.leave_conversation_group(conv_id, self.channel_name) async def receive_json(self, content): """ Handle incoming WebSocket messages. """ event_type = content.get('type') handlers = { 'new_message': self.handle_new_message, 'typing_status': self.handle_typing_status, 'mark_read': self.handle_mark_read, 'join_conversation': self.handle_join_conversation, # Call signaling events 'call:start': self.handle_call_start, 'call:accept': self.handle_call_accept, 'call:reject': self.handle_call_reject, 'call:end': self.handle_call_end, 'webrtc:offer': self.handle_webrtc_offer, 'webrtc:answer': self.handle_webrtc_answer, 'webrtc:ice-candidate': self.handle_webrtc_ice_candidate, # Group call events 'group_call:join': self.handle_group_call_join, 'group_call:leave': self.handle_group_call_leave, 'group_webrtc:offer': self.handle_group_webrtc_offer, 'group_webrtc:answer': self.handle_group_webrtc_answer, 'group_webrtc:ice-candidate': self.handle_group_webrtc_ice_candidate, 'group_call:update_status': self.handle_group_call_status_update, } handler = handlers.get(event_type) if handler: await handler(content) else: await self.send_json({ "type": "error", "message": f"Unknown event type: {event_type}" }) async def handle_new_message(self, content): """ Handle new message event. Save to DB and broadcast to conversation. """ conversation_id = content.get('conversation_id') message_content = content.get('content', '').strip() reply_to_id = content.get('reply_to') if not conversation_id or not message_content: await self.send_json({ "type": "error", "message": "conversation_id and content are required" }) return # Verify user is participant if not await self.is_conversation_participant(conversation_id): await self.send_json({ "type": "error", "message": "Not a participant of this conversation" }) return # Check permissions (blocking, following, etc.) recipient = await self.get_conversation_recipient(conversation_id) if recipient: from chat.services import can_message_user allowed, error_message = await database_sync_to_async(can_message_user)(self.user, recipient) if not allowed: await self.send_json({ "type": "error", "message": error_message }) return # Save message to database message = await self.create_message( conversation_id, message_content, reply_to_id ) if not message: await self.send_json({ "type": "error", "message": "Failed to create message" }) return # Prepare message data for broadcast message_data = await self.serialize_message(message) # 1. Broadcast to conversation group normalized_id = str(conversation_id).replace('-', '') await self.channel_layer.group_send( f"chat_{normalized_id}", { "type": "chat_message", "message_id": message.id, "message": message_data } ) # 2. ALSO broadcast to each participant's personal room # This is CRITICAL for new conversations where participants haven't joined the chat room yet. from chat.models import Conversation conversation = await database_sync_to_async(Conversation.objects.get)(id=conversation_id) participants = await database_sync_to_async(list)(conversation.participants.all()) for participant in participants: if participant == self.user: continue # Check visibility and mute status for each participant visibility = await database_sync_to_async( lambda: conversation.visibilities.filter(user=participant).first() )() # If hidden, do not send to their personal room (no notification, doesn't appear in list) if visibility and not visibility.is_visible: continue await self.channel_layer.group_send( f"user_{participant.id}", { "type": "chat_message", "message_id": message.id, "message": message_data, "is_muted": visibility.is_muted if visibility else False } ) async def handle_typing_status(self, content): """ Handle typing status update. Payload: { conversation_id, is_typing: bool } """ conversation_id = content.get('conversation_id') is_typing = content.get('is_typing', False) if not conversation_id or not await self.is_conversation_participant(conversation_id): return normalized_id = str(conversation_id).replace('-', '') await self.channel_layer.group_send( f"chat_{normalized_id}", { "type": "typing_indicator", "conversation_id": normalized_id, "user_id": self.user.id, "username": self.user.username, "is_typing": is_typing } ) async def handle_mark_read(self, content): """ Handle mark as read event. """ conversation_id = content.get('conversation_id') if not conversation_id or not await self.is_conversation_participant(conversation_id): return # Mark messages as read await self.mark_messages_read(conversation_id) # 1. Notify conversation room normalized_id = str(conversation_id).replace('-', '') await self.channel_layer.group_send( f"chat_{normalized_id}", { "type": "messages_read", "conversation_id": normalized_id, "user_id": self.user.id } ) # 2. Notify participants' personal rooms from chat.models import Conversation conversation = await database_sync_to_async(Conversation.objects.get)(id=conversation_id) participants = await database_sync_to_async(list)(conversation.participants.all()) for participant in participants: await self.channel_layer.group_send( f"user_{participant.id}", { "type": "messages_read", "conversation_id": normalized_id, "user_id": self.user.id } ) async def handle_join_conversation(self, content): """ Handle joining a new conversation room. """ conversation_id = content.get('conversation_id') if not conversation_id: return if not await self.is_conversation_participant(conversation_id): await self.send_json({ "type": "error", "message": "Not a participant of this conversation" }) return if conversation_id not in self.conversation_ids: await online_manager.join_conversation_group(conversation_id, self.channel_name) self.conversation_ids.add(conversation_id) # Check for active group call active_participants = [] if conversation_id in active_group_calls: user_ids = list(active_group_calls[conversation_id]) if user_ids: active_participants = await self._get_users_details(user_ids) await self.send_json({ "type": "joined_conversation", "conversation_id": conversation_id, "active_call_participants": active_participants }) # ==================== Channel layer event handlers ==================== async def chat_message(self, event): """ Handle chat message event from channel layer. """ message_id = event.get('message_id') message_data = event.get('message') # If message_id is provided, re-serialize for this specific user # to ensure privacy/visibility rules are applied for them. if message_id: from chat.models import Message try: message = await database_sync_to_async(Message.objects.get)(id=message_id) message_data = await self.serialize_message(message) except Exception: # Fallback to shared message if something goes wrong, # though serialize_message should handle restricted content. pass await self.send_json({ "type": "chat_message", "event_type": event.get("event_type", "chat_message"), "message": message_data }) async def typing_indicator(self, event): """ Handle typing indicator event from channel layer. """ # Don't send typing indicator to the user who is typing if event["user_id"] == self.user.id: return await self.send_json({ "type": "typing", "conversation_id": event.get("conversation_id"), "user_id": event["user_id"], "username": event["username"], "is_typing": event["is_typing"] }) async def messages_read(self, event): """ Handle messages read event from channel layer. """ await self.send_json({ "type": "messages_read", "conversation_id": event["conversation_id"], "user_id": event["user_id"] }) async def user_status(self, event): """ Handle user online/offline status event. """ await self.send_json({ "type": "user_status", "user_id": event["user_id"], "is_online": event["is_online"] }) async def new_notification(self, event): """ Handle new notification event from channel layer. """ await self.send_json({ "type": "new_notification", "notification": event["notification"] }) if "appearance" in event: await self.send_json({ "type": "appearance_update", "conversation_id": event.get("conversation_id"), "appearance": event["appearance"] }) # ==================== Group Chat Event Handlers ==================== async def group_created(self, event): """Handle group created event from channel layer.""" await self.send_json({ "type": "group_created", "conversation": event.get("conversation") }) async def group_updated(self, event): """Handle group updated event from channel layer.""" await self.send_json({ "type": "group_updated", "conversation": event.get("conversation") }) async def group_user_added(self, event): """Handle user added to group event from channel layer.""" await self.send_json({ "type": "group_user_added", "conversation_id": event.get("conversation_id"), "user": event.get("user") }) async def group_user_removed(self, event): """Handle user removed from group event from channel layer.""" await self.send_json({ "type": "group_user_removed", "conversation_id": event.get("conversation_id"), "user_id": event.get("user_id"), "removed_by": event.get("removed_by"), "left_voluntarily": event.get("left_voluntarily", False) }) async def group_join_request(self, event): """Handle group join request event from channel layer (admin only).""" await self.send_json({ "type": "group_join_request", "conversation_id": event.get("conversation_id"), "request_id": event.get("request_id"), "user": event.get("user"), "requested_by": event.get("requested_by") }) async def group_request_accepted(self, event): """Handle group request accepted event from channel layer.""" await self.send_json({ "type": "group_request_accepted", "conversation_id": event.get("conversation_id"), "user": event.get("user") }) # ==================== System Message Helpers ==================== @database_sync_to_async def _create_call_log_message(self, conversation_id, message_type, sender_id=None): """Create a system message for call events.""" from chat.models import Message return Message.objects.create( conversation_id=conversation_id, sender_id=sender_id, # Associate with user for directionality message_type=message_type, content=message_type.replace('_', ' ').title(), is_read=True ) @database_sync_to_async def _get_or_create_conversation(self, user_id_a, user_id_b): """Get or create conversation between two users.""" from chat.models import Conversation from django.contrib.auth import get_user_model User = get_user_model() # Try to find existing conversation conversations = Conversation.objects.filter(participants=user_id_a).filter(participants=user_id_b) if conversations.exists(): return conversations.first() # Create new if doesn't exist conversation = Conversation.objects.create() conversation.participants.add(user_id_a, user_id_b) return conversation # ==================== Call signaling event handlers ==================== def _get_call_id(self, user_id_a, user_id_b): """Generate canonical call ID from two user IDs.""" ids = sorted([str(user_id_a), str(user_id_b)]) return f"{ids[0]}_{ids[1]}" @database_sync_to_async def _check_can_call(self, sender_id, recipient_id): """Check if sender can call recipient.""" from django.contrib.auth import get_user_model User = get_user_model() try: sender = User.objects.get(id=sender_id) recipient = User.objects.get(id=recipient_id) return can_message_user(sender, recipient) except User.DoesNotExist: return False, "User not found" async def handle_call_start(self, content): """ Handle call initiation from caller. Notifies the callee of an incoming call. """ callee_id = content.get('callee_id') if not callee_id: await self.send_json({"type": "error", "message": "callee_id required"}) return # Security Check: Can calling user actually message/call the callee? can_call, error_msg = await self._check_can_call(self.user.id, callee_id) if not can_call: await self.send_json({"type": "error", "message": error_msg}) # Also notify client to end the call state await self.send_json({"type": "call:rejected"}) return call_id = self._get_call_id(self.user.id, callee_id) # Check if there's already an active call between these users if call_id in active_calls: await self.send_json({"type": "call:busy", "message": "User is busy"}) return call_type = content.get('call_type', 'audio') # Create call state active_calls[call_id] = { "caller_id": self.user.id, "callee_id": callee_id, "status": "ringing", "type": call_type, "started_at": datetime.now().isoformat() } # Notify callee of incoming call await self.channel_layer.group_send( f"user_{callee_id}", { "type": "call_incoming", "caller_id": self.user.id, "caller_name": f"{self.user.first_name} {self.user.last_name}".strip() or self.user.username, "caller_avatar": self.user.avatar.url if self.user.avatar else None, "call_type": call_type } ) # Confirm to caller that call is ringing await self.send_json({ "type": "call:ringing", "callee_id": callee_id }) # Start timeout task (30 seconds) asyncio.create_task(self._call_timeout(call_id, callee_id)) async def _call_timeout(self, call_id, callee_id): """Cancel call after 60 seconds if not answered.""" await asyncio.sleep(60) if call_id in active_calls and active_calls[call_id]["status"] == "ringing": # Log missed call caller_id = active_calls[call_id]["caller_id"] call_type = active_calls[call_id].get("type", "audio") log_type = "video_call_missed" if call_type == "video" else "call_missed" conversation = await self._get_or_create_conversation(caller_id, callee_id) if conversation: # Pass caller_id as sender msg = await self._create_call_log_message(conversation.id, log_type, sender_id=caller_id) # Broadcast the system message serialized_msg = await self.serialize_message(msg) normalized_id = str(conversation.id).replace('-', '') await self.channel_layer.group_send( f"chat_{normalized_id}", {"type": "chat_message", "message": serialized_msg} ) del active_calls[call_id] # Notify caller await self.send_json({ "type": "call:timeout", "message": "No answer" }) # Notify callee await self.channel_layer.group_send( f"user_{callee_id}", {"type": "call_ended", "reason": "timeout"} ) async def handle_call_accept(self, content): """ Handle call acceptance from callee. """ caller_id = content.get('caller_id') if not caller_id: await self.send_json({"type": "error", "message": "caller_id required"}) return call_id = self._get_call_id(self.user.id, caller_id) if call_id not in active_calls: await self.send_json({"type": "error", "message": "No active call"}) return if active_calls[call_id]["status"] == "in_call": await self.send_json({"type": "error", "message": "Call already active on another device"}) return active_calls[call_id]["status"] = "in_call" # Notify OTHER sessions of the callee to stop ringing await self.channel_layer.group_send( f"user_{self.user.id}", { "type": "call_answered_elsewhere", "answering_channel": self.channel_name } ) # Log call started conversation = await self._get_or_create_conversation(active_calls[call_id]["caller_id"], self.user.id) if conversation: call_type = active_calls[call_id].get("type", "audio") log_type = "video_call_started" if call_type == "video" else "call_started" # Caller is the sender of "Call Started" msg = await self._create_call_log_message(conversation.id, log_type, sender_id=active_calls[call_id]["caller_id"]) serialized_msg = await self.serialize_message(msg) normalized_id = str(conversation.id).replace('-', '') await self.channel_layer.group_send( f"chat_{normalized_id}", {"type": "chat_message", "message": serialized_msg} ) # Notify caller that call was accepted await self.channel_layer.group_send( f"user_{caller_id}", { "type": "call_accepted", "callee_id": self.user.id } ) await self.send_json({"type": "call:connected"}) async def handle_call_reject(self, content): """ Handle call rejection from callee. """ caller_id = content.get('caller_id') if not caller_id: return call_id = self._get_call_id(self.user.id, caller_id) if call_id in active_calls: # Log missed call if rejected while ringing if active_calls[call_id]["status"] == "ringing": caller_id = active_calls[call_id]["caller_id"] call_type = active_calls[call_id].get("type", "audio") log_type = "video_call_missed" if call_type == "video" else "call_missed" # Create conversation if needed conversation = await self._get_or_create_conversation(caller_id, self.user.id) if conversation: # Caller is sender of missed call msg = await self._create_call_log_message(conversation.id, log_type, sender_id=caller_id) serialized_msg = await self.serialize_message(msg) normalized_id = str(conversation.id).replace('-', '') await self.channel_layer.group_send( f"chat_{normalized_id}", {"type": "chat_message", "message": serialized_msg} ) del active_calls[call_id] # Notify caller await self.channel_layer.group_send( f"user_{caller_id}", {"type": "call_rejected"} ) async def handle_call_end(self, content): """ Handle call termination from either party. """ peer_id = content.get('peer_id') if not peer_id: return call_id = self._get_call_id(self.user.id, peer_id) if call_id in active_calls: # Log call ended or missed (cancelled) status = active_calls[call_id]["status"] caller_id = active_calls[call_id]["caller_id"] callee_id = active_calls[call_id]["callee_id"] message_type = None call_type = active_calls[call_id].get("type", "audio") if status == "in_call": message_type = 'video_call_ended' if call_type == 'video' else 'call_ended' elif status == "ringing": # If caller cancels, it's a missed call for the callee message_type = 'video_call_missed' if call_type == 'video' else 'call_missed' if message_type: conversation = await self._get_or_create_conversation(caller_id, callee_id) if conversation: # For ended, maybe sender is less relevant, but let's attribute to original caller for consistency # Or we could leave it neutral. Let's attribute to caller. sender_id_for_log = caller_id if 'ended' not in message_type else None # Actually for call_ended let's keep it neutral (None) or maybe the one who hung up? # Let's keep call_ended neutral (None) for now to symbolize shared event. # But missed MUST be caller. log_sender = caller_id if 'missed' in message_type else None msg = await self._create_call_log_message(conversation.id, message_type, sender_id=log_sender) serialized_msg = await self.serialize_message(msg) normalized_id = str(conversation.id).replace('-', '') await self.channel_layer.group_send( f"chat_{normalized_id}", {"type": "chat_message", "message": serialized_msg} ) del active_calls[call_id] # Notify peer await self.channel_layer.group_send( f"user_{peer_id}", {"type": "call_ended", "reason": "ended"} ) async def handle_webrtc_offer(self, content): """Forward SDP offer from caller to callee.""" callee_id = content.get('callee_id') sdp = content.get('sdp') if not callee_id or not sdp: return # Security: Verify call is active call_id = self._get_call_id(self.user.id, callee_id) if call_id not in active_calls: # Optionally send error or just ignore return await self.channel_layer.group_send( f"user_{callee_id}", { "type": "webrtc_offer", "sdp": sdp, "caller_id": self.user.id } ) async def handle_webrtc_answer(self, content): """Forward SDP answer from callee to caller.""" caller_id = content.get('caller_id') sdp = content.get('sdp') if not caller_id or not sdp: return # Security: Verify call is active call_id = self._get_call_id(self.user.id, caller_id) if call_id not in active_calls: return await self.channel_layer.group_send( f"user_{caller_id}", { "type": "webrtc_answer", "sdp": sdp, "callee_id": self.user.id } ) async def handle_webrtc_ice_candidate(self, content): """Forward ICE candidate to peer.""" peer_id = content.get('peer_id') candidate = content.get('candidate') if not peer_id or not candidate: return # Security: Verify call is active call_id = self._get_call_id(self.user.id, peer_id) if call_id not in active_calls: return await self.channel_layer.group_send( f"user_{peer_id}", { "type": "webrtc_ice", "candidate": candidate, "from_user_id": self.user.id } ) # ==================== Call channel layer handlers ==================== async def call_incoming(self, event): """Handle incoming call notification.""" await self.send_json({ "type": "call:incoming", "caller_id": event["caller_id"], "caller_name": event["caller_name"], "caller_avatar": event.get("caller_avatar"), "call_type": event.get("call_type", "audio") }) async def call_accepted(self, event): """Handle call accepted notification.""" await self.send_json({ "type": "call:accepted", "callee_id": event["callee_id"] }) async def call_rejected(self, event): """Handle call rejected notification.""" await self.send_json({"type": "call:rejected"}) async def call_ended(self, event): """Handle call ended notification.""" await self.send_json({ "type": "call:ended", "reason": event.get("reason", "ended") }) async def call_answered_elsewhere(self, event): """Handle call answered on another device.""" # Check if this is the device that answered (don't dismiss the active call) if hasattr(self, 'channel_name') and self.channel_name == event.get('answering_channel'): return await self.send_json({ "type": "call:answered_elsewhere" }) async def webrtc_offer(self, event): """Handle WebRTC offer.""" await self.send_json({ "type": "webrtc:offer", "sdp": event["sdp"], "caller_id": event["caller_id"] }) async def webrtc_answer(self, event): """Handle WebRTC answer.""" await self.send_json({ "type": "webrtc:answer", "sdp": event["sdp"], "callee_id": event["callee_id"] }) async def webrtc_ice(self, event): """Handle ICE candidate.""" await self.send_json({ "type": "webrtc:ice-candidate", "candidate": event["candidate"], "from_user_id": event["from_user_id"] }) # ==================== Database operations ==================== @database_sync_to_async def get_user_from_token(self, token_key): """ Get user from auth token. """ try: token = Token.objects.select_related('user').get(key=token_key) return token.user except Token.DoesNotExist: return None @database_sync_to_async def _get_users_details(self, user_ids): """Fetch basic details for a list of user IDs.""" from django.contrib.auth import get_user_model User = get_user_model() users = User.objects.filter(id__in=user_ids) return [ { "id": u.id, "username": u.username, "first_name": u.first_name, "last_name": u.last_name, "avatar": u.avatar.url if u.avatar else None } for u in users ] @database_sync_to_async def get_user_conversation_ids(self): """ Get all conversation IDs for the current user. """ from chat.models import Conversation, ConversationVisibility return list( Conversation.objects.filter( participants=self.user, visibilities__user=self.user, visibilities__is_visible=True ).values_list('id', flat=True) ) @database_sync_to_async def is_conversation_participant(self, conversation_id): """ Check if user is a participant of the conversation. """ from chat.models import Conversation return Conversation.objects.filter( id=conversation_id, participants=self.user ).exists() @database_sync_to_async def _create_system_message(self, conversation_id, message_type, sender=None): """Helper to create a system message.""" from chat.models import Message, Conversation try: conversation = Conversation.objects.get(id=conversation_id) msg = Message.objects.create( conversation=conversation, sender=sender, message_type=message_type, content="" # System messages use type for display ) return msg except Exception as e: print(f"Error creating system message: {e}") return None @database_sync_to_async def create_message(self, conversation_id, content, reply_to_id=None): """ Create a new message in the database. """ from chat.models import Message, Conversation try: conversation = Conversation.objects.get(id=conversation_id) message = Message.objects.create( conversation=conversation, sender=self.user, sender_name_snapshot=f"{self.user.first_name} {self.user.last_name}".strip() or self.user.username, sender_avatar_snapshot=self.user.avatar.url if self.user.avatar else None, content=content, reply_to_id=reply_to_id ) # Update conversation timestamp conversation.save() return message except Conversation.DoesNotExist: return None @database_sync_to_async def serialize_message(self, message): """ Serialize message for sending via WebSocket. """ from chat.serializers import MessageSerializer return MessageSerializer(message, context={'user': self.user}).data @database_sync_to_async def mark_messages_read(self, conversation_id): """ Mark all unread messages in conversation as read. """ from chat.models import Message Message.objects.filter( conversation_id=conversation_id, is_read=False ).exclude( sender=self.user ).update(is_read=True) @database_sync_to_async def get_conversation_recipient(self, conversation_id): """ Get the other participant in the conversation. """ from chat.models import Conversation try: conversation = Conversation.objects.get(id=conversation_id) return conversation.participants.exclude(id=self.user.id).first() except Conversation.DoesNotExist: return None # ==================== Group Audio Call Handlers ==================== async def handle_group_call_join(self, content): """ Handle user joining a group call. """ conversation_id = content.get('conversation_id') if not conversation_id or not await self.is_conversation_participant(conversation_id): return # Initialize group call active list if needed is_new_call = False if conversation_id not in active_group_calls or not active_group_calls[conversation_id]: active_group_calls[conversation_id] = set() is_new_call = True # Add user to active participants if self.user.id in active_group_calls[conversation_id]: await self.send_json({"type": "error", "message": "You have already joined this call from another device."}) return active_group_calls[conversation_id].add(self.user.id) # Broadcast to group that user joined (so others can connect) normalized_id = str(conversation_id).replace('-', '') await self.channel_layer.group_send( f"chat_{normalized_id}", { "type": "group_call_participant_joined", "conversation_id": conversation_id, "user": { "id": self.user.id, "username": self.user.username, "first_name": self.user.first_name, "last_name": self.user.last_name, "avatar": self.user.avatar.url if self.user.avatar else None } } ) # Create system message for chat history msg_type = 'group_call_started' if is_new_call else 'group_call_joined' msg = await self._create_system_message(conversation_id, msg_type, sender=self.user) if msg: serialized = await self.serialize_message(msg) await self.channel_layer.group_send( f"chat_{normalized_id}", {"type": "chat_message", "message": serialized} ) # Send list of current participants to the joining user (so they can connect) other_participants = [uid for uid in active_group_calls[conversation_id] if uid != self.user.id] await self.send_json({ "type": "group_call:joined", "conversation_id": conversation_id, "participants": other_participants }) async def handle_group_call_leave(self, content): """ Handle user leaving a group call. """ conversation_id = content.get('conversation_id') if not conversation_id: return if conversation_id in active_group_calls and self.user.id in active_group_calls[conversation_id]: active_group_calls[conversation_id].remove(self.user.id) normalized_id = str(conversation_id).replace('-', '') # Broadcast leave event await self.channel_layer.group_send( f"chat_{normalized_id}", { "type": "group_call_participant_left", "conversation_id": conversation_id, "user_id": self.user.id } ) if not active_group_calls[conversation_id]: del active_group_calls[conversation_id] # Call ended msg = await self._create_system_message(conversation_id, 'group_call_ended') if msg: serialized = await self.serialize_message(msg) await self.channel_layer.group_send( f"chat_{normalized_id}", {"type": "chat_message", "message": serialized} ) async def handle_group_webrtc_offer(self, content): """Relay offer to a specific peer in the group call.""" conversation_id = content.get('conversation_id') to_user_id = content.get('to_user_id') sdp = content.get('sdp') if not conversation_id or not to_user_id or not sdp: return # Verify both are in the group call (basic security) if conversation_id not in active_group_calls: return if self.user.id not in active_group_calls[conversation_id]: return # We don't necessarily enforce target is in list (race condition), but good practice. # But if target left, relaying is useless. await self.channel_layer.group_send( f"user_{to_user_id}", { "type": "group_webrtc_offer", "conversation_id": conversation_id, "from_user_id": self.user.id, "sdp": sdp } ) async def handle_group_webrtc_answer(self, content): """Relay answer to a specific peer.""" conversation_id = content.get('conversation_id') to_user_id = content.get('to_user_id') sdp = content.get('sdp') if not conversation_id or not to_user_id or not sdp: return await self.channel_layer.group_send( f"user_{to_user_id}", { "type": "group_webrtc_answer", "conversation_id": conversation_id, "from_user_id": self.user.id, "sdp": sdp } ) async def handle_group_webrtc_ice_candidate(self, content): """Relay ICE candidate to a specific peer.""" conversation_id = content.get('conversation_id') to_user_id = content.get('to_user_id') candidate = content.get('candidate') if not conversation_id or not to_user_id or not candidate: return await self.channel_layer.group_send( f"user_{to_user_id}", { "type": "group_webrtc_ice", "conversation_id": conversation_id, "from_user_id": self.user.id, "candidate": candidate } ) # ==================== Group Channel Layer Handlers ==================== async def group_call_participant_joined(self, event): await self.send_json({ "type": "group_call:participant_joined", "conversation_id": event["conversation_id"], "user": event["user"] }) async def group_call_participant_left(self, event): await self.send_json({ "type": "group_call:participant_left", "conversation_id": event["conversation_id"], "user_id": event["user_id"] }) async def group_webrtc_offer(self, event): await self.send_json({ "type": "group_webrtc:offer", "conversation_id": event["conversation_id"], "from_user_id": event["from_user_id"], "sdp": event["sdp"] }) async def group_webrtc_answer(self, event): await self.send_json({ "type": "group_webrtc:answer", "conversation_id": event["conversation_id"], "from_user_id": event["from_user_id"], "sdp": event["sdp"] }) async def group_webrtc_ice(self, event): await self.send_json({ "type": "group_webrtc:ice-candidate", "conversation_id": event["conversation_id"], "from_user_id": event["from_user_id"], "candidate": event["candidate"] }) async def handle_group_call_status_update(self, content): """Relay group call status update (mute/camera).""" conversation_id = content.get('conversation_id') status = content.get('status') normalized_id = str(conversation_id).replace('-', '') await self.channel_layer.group_send( f"chat_{normalized_id}", { "type": "group_call_status_update", "conversation_id": conversation_id, "sender_id": self.user.id, "status": status } ) async def group_call_status_update(self, event): await self.send_json({ "type": "group_call:status_update", "conversation_id": event['conversation_id'], "sender_id": event['sender_id'], "status": event['status'] })