Ross
316 lines
13 KiB
Python
316 lines
13 KiB
Python
from loguru import logger
|
|
from channels.generic.websocket import AsyncJsonWebsocketConsumer
|
|
from django.core.cache import cache
|
|
from asgiref.sync import sync_to_async
|
|
from channels.db import database_sync_to_async
|
|
from django.core.exceptions import ValidationError
|
|
from atlas.models import MultiuserSession
|
|
import urllib.parse
|
|
|
|
class MultiuserConsumer(AsyncJsonWebsocketConsumer):
|
|
@database_sync_to_async
|
|
def get_session(self, session_id):
|
|
try:
|
|
return MultiuserSession.objects.select_related('creator').get(id=session_id)
|
|
except (MultiuserSession.DoesNotExist, ValidationError):
|
|
return None
|
|
|
|
async def connect(self):
|
|
self.room_name = self.scope['url_route']['kwargs']['room_name']
|
|
self.room_group_name = f'multiuser_{self.room_name}'
|
|
self.user = self.scope.get('user')
|
|
|
|
# Get session from DB using self.room_name
|
|
session = await self.get_session(self.room_name)
|
|
if not session:
|
|
logger.warning("[Multiuser WS] Rejecting connection: session {} not found", self.room_name)
|
|
await self.close()
|
|
return
|
|
|
|
# Determine username and staff privilege
|
|
if self.user and self.user.is_authenticated:
|
|
self.username = self.user.username
|
|
# Manager is the creator or a site admin/staff
|
|
self.is_staff = (self.user == session.creator) or self.user.is_staff
|
|
else:
|
|
self.username = f"Guest_{self.channel_name[-4:]}"
|
|
self.is_staff = False
|
|
|
|
# Parse query string for client_type (default to 'viewer')
|
|
query_string = self.scope.get('query_string', b'').decode('utf-8')
|
|
params = urllib.parse.parse_qs(query_string)
|
|
self.client_type = params.get('client_type', ['viewer'])[0]
|
|
|
|
logger.info(
|
|
"[Multiuser WS] Connect request: room={}, user={}, type={}, channel={}",
|
|
self.room_name, self.username, self.client_type, self.channel_name
|
|
)
|
|
|
|
# Add to channels group
|
|
await self.channel_layer.group_add(
|
|
self.room_group_name,
|
|
self.channel_name
|
|
)
|
|
await self.accept()
|
|
await self.send_json({
|
|
'type': 'welcome',
|
|
'channel_name': self.channel_name
|
|
})
|
|
|
|
if self.client_type == 'viewer':
|
|
# Update room state in cache & notify others
|
|
await self.add_user_to_room()
|
|
await self.broadcast_room_status()
|
|
else:
|
|
# Passive session listener: just broadcast the current room status to this new client
|
|
logger.info("[Multiuser WS] Passive listener connected: {}", self.channel_name)
|
|
await self.broadcast_room_status()
|
|
|
|
async def disconnect(self, close_code):
|
|
logger.info(
|
|
"[Multiuser WS] Disconnect: room={}, user={}, type={}, channel={}, code={}",
|
|
self.room_name, self.username, self.client_type, self.channel_name, close_code
|
|
)
|
|
|
|
# Remove from channels group
|
|
await self.channel_layer.group_discard(
|
|
self.room_group_name,
|
|
self.channel_name
|
|
)
|
|
|
|
# Update room state in cache & notify others
|
|
if self.client_type == 'viewer':
|
|
await self.remove_user_from_room()
|
|
await self.broadcast_room_status()
|
|
|
|
async def receive_json(self, content):
|
|
msg_type = content.get('type')
|
|
logger.info(
|
|
"[Multiuser WS] Received message from user={} (type={}): {}",
|
|
self.username, msg_type, content
|
|
)
|
|
|
|
if self.client_type != 'viewer':
|
|
logger.warning(
|
|
"[Multiuser WS] Rejected message from passive listener user={} (type={})",
|
|
self.username, msg_type
|
|
)
|
|
return
|
|
|
|
if msg_type == 'viewer_state':
|
|
# Replicate view state changes from the active controller to others
|
|
if await self.has_control():
|
|
logger.debug("[Multiuser WS] Broadcasting viewer_state from user={}", self.username)
|
|
await self.channel_layer.group_send(
|
|
self.room_group_name,
|
|
{
|
|
'type': 'viewer_state_update',
|
|
'sender': self.channel_name,
|
|
'data': content.get('data')
|
|
}
|
|
)
|
|
else:
|
|
logger.warning(
|
|
"[Multiuser WS] viewer_state rejected: user={} does not have control",
|
|
self.username
|
|
)
|
|
elif msg_type == 'annotations':
|
|
# Replicate annotation changes from the active controller to others
|
|
if await self.has_control():
|
|
logger.debug("[Multiuser WS] Broadcasting annotations from user={}", self.username)
|
|
await self.channel_layer.group_send(
|
|
self.room_group_name,
|
|
{
|
|
'type': 'annotations_update',
|
|
'sender': self.channel_name,
|
|
'data': content.get('data')
|
|
}
|
|
)
|
|
else:
|
|
logger.warning(
|
|
"[Multiuser WS] annotations rejected: user={} does not have control",
|
|
self.username
|
|
)
|
|
elif msg_type == 'request_control':
|
|
# Broadcast a control request to all participants (so the admin/controller can grant it)
|
|
logger.info("[Multiuser WS] Control request from user={}", self.username)
|
|
await self.channel_layer.group_send(
|
|
self.room_group_name,
|
|
{
|
|
'type': 'control_request',
|
|
'sender_name': self.username,
|
|
'sender_channel': self.channel_name
|
|
}
|
|
)
|
|
elif msg_type == 'take_control':
|
|
# Supervisors/admins can immediately assume control
|
|
if self.is_staff:
|
|
logger.info("[Multiuser WS] Supervisor user={} taking control", self.username)
|
|
await self.set_controller(self.channel_name, self.username)
|
|
await self.broadcast_room_status()
|
|
else:
|
|
logger.warning(
|
|
"[Multiuser WS] take_control rejected: user={} is not supervisor/staff",
|
|
self.username
|
|
)
|
|
elif msg_type == 'grant_control':
|
|
# The active controller or a supervisor can hand over control to someone else
|
|
if self.is_staff or await self.has_control():
|
|
target_channel = content.get('target_channel')
|
|
target_username = content.get('target_username')
|
|
if target_channel:
|
|
logger.info(
|
|
"[Multiuser WS] Control granted by user={} to target_user={} ({})",
|
|
self.username, target_username, target_channel
|
|
)
|
|
await self.set_controller(target_channel, target_username)
|
|
await self.broadcast_room_status()
|
|
else:
|
|
logger.warning(
|
|
"[Multiuser WS] grant_control rejected: user={} is not controller/staff",
|
|
self.username
|
|
)
|
|
elif msg_type == 'release_control':
|
|
# Relinquish control
|
|
if await self.has_control():
|
|
logger.info("[Multiuser WS] Control released by user={}", self.username)
|
|
await self.set_controller(None, None)
|
|
await self.broadcast_room_status()
|
|
else:
|
|
logger.warning(
|
|
"[Multiuser WS] release_control rejected: user={} is not active controller",
|
|
self.username
|
|
)
|
|
|
|
# Channels group event handlers
|
|
async def viewer_state_update(self, event):
|
|
if event['sender'] != self.channel_name:
|
|
logger.debug("[Multiuser WS] Forwarding viewer_state_update to channel={}", self.channel_name)
|
|
await self.send_json({
|
|
'type': 'viewer_state',
|
|
'data': event['data']
|
|
})
|
|
|
|
async def annotations_update(self, event):
|
|
if event['sender'] != self.channel_name:
|
|
logger.debug("[Multiuser WS] Forwarding annotations_update to channel={}", self.channel_name)
|
|
await self.send_json({
|
|
'type': 'annotations',
|
|
'data': event['data']
|
|
})
|
|
|
|
async def control_request(self, event):
|
|
logger.debug("[Multiuser WS] Forwarding control_request to channel={}", self.channel_name)
|
|
await self.send_json({
|
|
'type': 'control_request',
|
|
'sender_name': event['sender_name'],
|
|
'sender_channel': event['sender_channel']
|
|
})
|
|
|
|
async def room_status_update(self, event):
|
|
logger.debug("[Multiuser WS] Forwarding room_status_update to channel={}", self.channel_name)
|
|
await self.send_json({
|
|
'type': 'room_status',
|
|
'users': event['users'],
|
|
'controller_channel': event['controller_channel'],
|
|
'controller_username': event['controller_username']
|
|
})
|
|
|
|
async def change_case(self, event):
|
|
logger.info("[Multiuser WS] Forwarding change_case (case_id={}) to channel={}", event['case_id'], self.channel_name)
|
|
await self.send_json({
|
|
'type': 'change_case',
|
|
'case_id': event['case_id']
|
|
})
|
|
|
|
# Cache state helpers (using sync_to_async for thread-safe Memcached/Redis interactions)
|
|
def get_cache_key(self):
|
|
return f"multiuser_room_{self.room_name}"
|
|
|
|
@sync_to_async
|
|
def _get_state(self):
|
|
return cache.get(self.get_cache_key())
|
|
|
|
@sync_to_async
|
|
def _set_state(self, state):
|
|
cache.set(self.get_cache_key(), state, timeout=86400) # Expire after 1 day
|
|
|
|
async def add_user_to_room(self):
|
|
state = await self._get_state()
|
|
if not state:
|
|
state = {
|
|
'users': {},
|
|
'controller_channel': None,
|
|
'controller_username': None
|
|
}
|
|
|
|
# Remove any existing channel names for the same username to prevent duplicate presence counts on refresh
|
|
for old_channel in list(state['users'].keys()):
|
|
if state['users'][old_channel]['username'] == self.username:
|
|
logger.info(
|
|
"[Multiuser WS] Pruning stale channel name {} for user={}",
|
|
old_channel, self.username
|
|
)
|
|
state['users'].pop(old_channel, None)
|
|
|
|
state['users'][self.channel_name] = {
|
|
'username': self.username,
|
|
'is_staff': self.is_staff
|
|
}
|
|
|
|
# Auto-assign control if no active controller exists or the previous controller was pruned
|
|
if not state['controller_channel'] or state['controller_channel'] not in state['users']:
|
|
state['controller_channel'] = self.channel_name
|
|
state['controller_username'] = self.username
|
|
logger.info(
|
|
"[Multiuser WS] Control assigned to user={} (channel={})",
|
|
self.username, self.channel_name
|
|
)
|
|
|
|
await self._set_state(state)
|
|
|
|
async def remove_user_from_room(self):
|
|
state = await self._get_state()
|
|
if state:
|
|
state['users'].pop(self.channel_name, None)
|
|
# If the leaving user was in control, elect a replacement
|
|
if state['controller_channel'] == self.channel_name:
|
|
if state['users']:
|
|
next_channel = list(state['users'].keys())[0]
|
|
state['controller_channel'] = next_channel
|
|
state['controller_username'] = state['users'][next_channel]['username']
|
|
logger.info(
|
|
"[Multiuser WS] Control passed to user={} (channel={})",
|
|
state['controller_username'], next_channel
|
|
)
|
|
else:
|
|
state['controller_channel'] = None
|
|
state['controller_username'] = None
|
|
logger.info("[Multiuser WS] Room is empty, control cleared")
|
|
|
|
await self._set_state(state)
|
|
|
|
async def has_control(self):
|
|
state = await self._get_state()
|
|
return state and state['controller_channel'] == self.channel_name
|
|
|
|
async def set_controller(self, channel_name, username):
|
|
state = await self._get_state()
|
|
if state:
|
|
state['controller_channel'] = channel_name
|
|
state['controller_username'] = username
|
|
await self._set_state(state)
|
|
|
|
async def broadcast_room_status(self):
|
|
state = await self._get_state()
|
|
if state:
|
|
await self.channel_layer.group_send(
|
|
self.room_group_name,
|
|
{
|
|
'type': 'room_status_update',
|
|
'users': state['users'],
|
|
'controller_channel': state['controller_channel'],
|
|
'controller_username': state['controller_username']
|
|
}
|
|
)
|