# realtime/consumers.py """ WebSocket Consumer для пользователя. Универсальное WebSocket соединение для всех real-time операций: - Чаты и сообщения (через ChatConsumerMixin) - Уведомления - Обновления бейджей - Онлайн-статус - Закупки и другие real-time события История: - 11 марта 2026: Извлечен ChatConsumerMixin в communications.consumers """ import asyncio import json import logging from time import monotonic from channels.db import database_sync_to_async from channels.generic.websocket import AsyncJsonWebsocketConsumer from django.contrib.auth.models import AnonymousUser from django.utils import timezone from commerce.realtime import COMMERCE_CONNECTIONS_GROUP from communications.consumers import ChatConsumerMixin from scheduling.realtime import ( MAX_CALENDAR_SUBSCRIPTIONS, CalendarSubscriptionError, build_calendar_snapshot, parse_calendar_subscription, ) from .protocol import ConnectionRateLimiter, RealtimeProtocolError, parse_action logger = logging.getLogger(__name__) class UserConsumer(ChatConsumerMixin, AsyncJsonWebsocketConsumer): """ WebSocket Consumer для пользователя. Управляет всеми real-time обновлениями: - Чаты: обновления списка (бейдж, карточки), сообщения в активном чате - Реакции, редактирование, удаление сообщений - Индикатор "печатает..." - Уведомления - Бейджи в sidebar - Онлайн-статус """ ACTION_HANDLERS = { "open_chat": "_handle_open_chat", "close_chat": "_handle_close_chat", "send_message": "_handle_send_message", "edit_message": "_handle_edit_message", "delete_message": "_handle_delete_message", "add_reaction": "_handle_add_reaction", "remove_reaction": "_handle_remove_reaction", "typing": "_handle_typing", "stop_typing": "_handle_stop_typing", "mark_read": "_handle_mark_read", "vote_poll": "_handle_vote_poll", "calendar_subscribe": "_handle_calendar_subscribe", "calendar_unsubscribe": "_handle_calendar_unsubscribe", "pong": "_handle_pong", } def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) self.ping_task = None self.ping_interval = 20 # Ping каждые 20 секунд self.active_chat_id = None # ID открытого чата self.subscribed_chats = set() # Чаты, на которые подписан self.user = None self.calendar_subscriptions = {} self.calendar_refresh_task = None self.pending_calendar_refresh_ids = set() self.last_pong_at = monotonic() self.rate_limiter = ConnectionRateLimiter() async def receive(self, text_data=None, bytes_data=None, **kwargs): """Reject malformed, binary and oversized commands without crashing.""" if bytes_data is not None: await self.send_json( { "type": "error", "code": "binary_not_supported", "error": "Поддерживаются только JSON-команды.", } ) return if text_data is None: return if len(text_data.encode("utf-8")) > 64 * 1024: await self.send_json( { "type": "error", "code": "payload_too_large", "error": "WebSocket-команда превышает лимит 64 КиБ.", } ) return try: content = await self.decode_json(text_data) except (json.JSONDecodeError, TypeError, ValueError): await self.send_json( { "type": "error", "code": "invalid_json", "error": "Некорректный JSON.", } ) return await self.receive_json(content, **kwargs) async def connect(self): """Подключение пользователя - подписка на все его чаты и каналы""" self.user = self.scope.get("user") if ( not self.user or isinstance(self.user, AnonymousUser) or not self.user.is_authenticated ): await self.close(code=4401) return # Получаем все чаты пользователя chat_ids = await self._get_available_chat_ids(self.user) # Подписываемся на группы всех чатов for chat_id in chat_ids: group_name = f"chat_{chat_id}" await self.channel_layer.group_add(group_name, self.channel_name) self.subscribed_chats.add(chat_id) # Подписываемся на личный канал пользователя (для уведомлений) user_channel = f"user_{self.user.id}" await self.channel_layer.group_add(user_channel, self.channel_name) # Подписываемся на канал уведомлений notifications_channel = f"notifications_{self.user.id}" await self.channel_layer.group_add(notifications_channel, self.channel_name) # Статусы интеграций не требуют отдельного сокета: все сотрудники уже # имеют read-доступ к commerce API, поэтому используем общий канал. await self.channel_layer.group_add( COMMERCE_CONNECTIONS_GROUP, self.channel_name, ) await self.accept() # Запускаем ping цикл для keepalive self.last_pong_at = monotonic() self.ping_task = asyncio.create_task(self._ping_loop()) logger.info( "Realtime consumer initialized", extra={ "event": "websocket.consumer.ready", "user_id": self.user.id, "subscribed_chat_count": len(self.subscribed_chats), }, ) async def disconnect(self, code): """Отключение - отписка от всех групп""" # Останавливаем ping цикл if self.ping_task: self.ping_task.cancel() try: await self.ping_task except asyncio.CancelledError: pass if self.calendar_refresh_task: self.calendar_refresh_task.cancel() try: await self.calendar_refresh_task except asyncio.CancelledError: pass self.calendar_refresh_task = None self.calendar_subscriptions.clear() self.pending_calendar_refresh_ids.clear() # Сбрасываем статус "набирает текст" если был активный чат if self.active_chat_id and self.user: chat = await self._get_chat(self.active_chat_id) if chat: await self._set_typing_status(chat, self.user, False) await self.channel_layer.group_send( f"chat_{chat.id}", { "type": "chat_user_stopped_typing", "chat_id": chat.id, "user_id": self.user.id, }, ) if self.user: await self._clear_active_chat_presence() # Отписываемся от всех групп чатов for chat_id in self.subscribed_chats: group_name = f"chat_{chat_id}" await self.channel_layer.group_discard(group_name, self.channel_name) # Отписываемся от личных каналов if self.user: user_channel = f"user_{self.user.id}" await self.channel_layer.group_discard(user_channel, self.channel_name) notifications_channel = f"notifications_{self.user.id}" await self.channel_layer.group_discard( notifications_channel, self.channel_name ) await self.channel_layer.group_discard( COMMERCE_CONNECTIONS_GROUP, self.channel_name, ) logger.debug( "Realtime consumer disconnected", extra={ "event": "websocket.consumer.disconnected", "user_id": getattr(self.user, "id", None), }, ) async def _ping_loop(self): """Отправка ping каждые N секунд для keepalive""" try: while True: await asyncio.sleep(self.ping_interval) if monotonic() - self.last_pong_at > self.ping_interval * 3: logger.warning( "WebSocket client heartbeat timed out", extra={ "event": "websocket.heartbeat.timeout", "user_id": self.user.id, }, ) await self.close(code=4408) return await self.send_json( {"type": "ping", "timestamp": timezone.now().isoformat()} ) except asyncio.CancelledError: pass except Exception as error: logger.exception( "WebSocket ping loop failed", extra={ "event": "websocket.ping.failed", "connection_id": self.channel_name, "error_type": type(error).__name__, }, ) async def receive_json(self, content, **kwargs): """Обработка входящих сообщений от клиента""" try: action = parse_action(content) except RealtimeProtocolError as error: await self._send_protocol_error(error) return if not self.rate_limiter.allow(action): await self.send_json( { "type": "error", "code": "rate_limited", "action": action, "error": "Слишком много WebSocket-команд. Повторите позже.", } ) return if action != "pong": logger.debug( "WebSocket action received", extra={ "event": "websocket.action.received", "action": action, "active_chat_id": self.active_chat_id, }, ) handler_name = self.ACTION_HANDLERS.get(action) if handler_name is None: await self.send_json( { "type": "error", "code": "unknown_action", "action": action, "error": "Неизвестная WebSocket-команда.", } ) return handler = getattr(self, handler_name) try: await handler(content) except RealtimeProtocolError as error: await self._send_protocol_error(error, action=action) except (TypeError, ValueError) as error: logger.info( "Invalid WebSocket command payload rejected", extra={ "event": "websocket.command.invalid", "action": action, "user_id": self.user.id, "error_type": type(error).__name__, }, ) await self.send_json( { "type": "error", "code": "invalid_payload", "action": action, "error": "Некорректные данные WebSocket-команды.", } ) except Exception as error: # noqa: BLE001 - protocol boundary must survive logger.exception( "WebSocket command failed", extra={ "event": "websocket.command.failed", "action": action, "user_id": self.user.id, "error_type": type(error).__name__, }, ) await self.send_json( { "type": "error", "code": "internal_error", "action": action, "error": "Не удалось выполнить WebSocket-команду.", } ) async def _send_protocol_error(self, error, *, action=None): await self.send_json( { "type": "error", "code": error.code, "action": action, "error": error.detail, } ) async def _handle_pong(self, content): self.last_pong_at = monotonic() if self.active_chat_id is not None: await self._touch_active_chat_presence() @database_sync_to_async def _build_calendar_snapshot(self, subscription): return build_calendar_snapshot(user=self.user, subscription=subscription) async def _send_calendar_snapshot(self, subscription_id, *, reason): subscription = self.calendar_subscriptions.get(subscription_id) if subscription is None: return try: snapshot = await self._build_calendar_snapshot(subscription) except Exception as error: logger.exception( "Calendar WebSocket snapshot failed", extra={ "event": "calendar.websocket_snapshot_failed", "user_id": self.user.id, "subscription_id": subscription_id, "error_type": type(error).__name__, }, ) await self.send_json( { "type": "calendar_error", "subscription_id": subscription_id, "code": "snapshot_failed", "detail": "Не удалось загрузить события календаря.", } ) return # Подписка могла измениться, пока выполнялся запрос к БД. if self.calendar_subscriptions.get(subscription_id) != subscription: return await self.send_json( { "type": "calendar_snapshot", "reason": reason, "data": snapshot, } ) async def _handle_calendar_subscribe(self, content): try: subscription = parse_calendar_subscription(content, user=self.user) except CalendarSubscriptionError as error: await self.send_json( { "type": "calendar_error", "subscription_id": content.get("subscription_id"), "code": "invalid_subscription", "detail": str(error), } ) return if ( subscription.subscription_id not in self.calendar_subscriptions and len(self.calendar_subscriptions) >= MAX_CALENDAR_SUBSCRIPTIONS ): await self.send_json( { "type": "calendar_error", "subscription_id": subscription.subscription_id, "code": "subscription_limit", "detail": "Превышен лимит подписок календаря.", } ) return self.calendar_subscriptions[subscription.subscription_id] = subscription await self._send_calendar_snapshot( subscription.subscription_id, reason="subscribed", ) async def _handle_calendar_unsubscribe(self, content): subscription_id = str(content.get("subscription_id") or "").strip() if subscription_id: self.calendar_subscriptions.pop(subscription_id, None) async def _refresh_calendar_subscriptions(self): try: while self.pending_calendar_refresh_ids: # Несколько сигналов одной транзакции объединяются в один snapshot. await asyncio.sleep(0.05) calendar_ids = set(self.pending_calendar_refresh_ids) self.pending_calendar_refresh_ids.clear() subscriptions = list(self.calendar_subscriptions.values()) for subscription in subscriptions: if ( subscription.calendar_id is not None and subscription.calendar_id not in calendar_ids ): continue await self._send_calendar_snapshot( subscription.subscription_id, reason="updated", ) except asyncio.CancelledError: pass finally: self.calendar_refresh_task = None # ==================== Обработчики уведомлений ==================== async def notification_message(self, event): """ Новое уведомление от WebSocketNotificationSender. Вызывается из notifications/senders/websocket.py """ notification = event.get("message", {}) await self.send_json({"type": "notification", "notification": notification}) async def notification_new(self, event): """Новое уведомление (вызывается из notifications/services.py)""" notification = event.get("notification", {}) await self.send_json( { "type": "notification", "notification": notification, } ) async def notification_count_update(self, event): """ Обновление счетчика уведомлений. Вызывается из notifications/services.py """ await self.send_json({"type": "unread_count", "count": event.get("count", 0)}) async def notification_read(self, event): """Синхронизация прочтения одного уведомления между сессиями.""" await self.send_json( { "type": "notification_read", "notification_id": event.get("notification_id"), "unread_count": event.get("unread_count"), } ) async def notifications_read_all(self, event): """Синхронизация массового прочтения уведомлений между сессиями.""" await self.send_json( { "type": "notifications_read_all", "notification_ids": event.get("notification_ids", []), "category": event.get("category"), "unread_count": event.get("unread_count"), } ) async def notifications_deleted(self, event): """Remove deleted notifications from every active user session.""" await self.send_json( { "type": "notifications_deleted", "notification_ids": event.get("notification_ids", []), "unread_count": event.get("unread_count"), } ) async def commerce_connection_updated(self, event): """Передаёт изменение статуса интеграции через общий WebSocket.""" await self.send_json( { "type": "commerce_connection_updated", "connection": event.get("connection", {}), } ) # ==================== Обработка событий закупок ==================== async def poll_update(self, event): """Обработчик для события poll.update (алиас для chat_poll_update)""" await self.chat_poll_update(event) async def poll_update_sync(self, event): """Персональная синхронизация выбора в голосовании между сессиями.""" await self.chat_poll_update(event) async def procurement_update(self, event): """ Обновление заявки на закупку. Вызывается из procurement/signals.py """ await self.send_json( { "type": "procurement_update", "event": event.get("event", "request_updated"), "data": event.get("data", {}), } ) async def task_board_update(self, event): """ Обновление доски задач. Вызывается из tasks/realtime.py """ await self.send_json( { "type": "task_board_update", "event": event.get("event", "updated"), "data": event.get("data", {}), } ) async def calendar_update(self, event): """Передать метаданные изменения и обновить активные snapshots.""" data = event.get("data", {}) await self.send_json( { "type": "calendar_update", "event": event.get("event", "event_updated"), "data": data, } ) if not self.calendar_subscriptions: return try: calendar_id = int(data.get("calendar_id")) except (TypeError, ValueError): return self.pending_calendar_refresh_ids.add(calendar_id) if not self.calendar_refresh_task: self.calendar_refresh_task = asyncio.create_task( self._refresh_calendar_subscriptions() ) async def chat_access_changed(self, event): """Apply membership changes to a live connection without reconnecting.""" try: chat_id = int(event.get("chat_id")) except (TypeError, ValueError): return chat = await self._get_chat(chat_id) can_access = bool(chat and await self._user_can_access(chat, self.user)) group_name = f"chat_{chat_id}" if can_access: if chat_id not in self.subscribed_chats: await self.channel_layer.group_add(group_name, self.channel_name) self.subscribed_chats.add(chat_id) elif chat_id in self.subscribed_chats: await self.channel_layer.group_discard(group_name, self.channel_name) self.subscribed_chats.discard(chat_id) if self.active_chat_id == chat_id: self.active_chat_id = None await self.send_json( { "type": "chat_access_changed", "chat_id": chat_id, "can_access": can_access, } ) logger.info( "Live chat access synchronized", extra={ "event": "websocket.chat.access_synchronized", "chat_id": chat_id, "user_id": self.user.id, "allowed": can_access, }, ) async def auth_session_revoked(self, event): """Immediately terminate the socket that belongs to a revoked session.""" connection_session_id = self.scope.get("auth_session_id") revoked_session_id = event.get("session_id") if not connection_session_id or str(connection_session_id) != str( revoked_session_id ): return logger.info( "WebSocket closed after authentication session revocation", extra={ "event": "websocket.authentication.session_revoked", "user_id": self.user.id, "auth_session_id": str(connection_session_id), }, ) await self.close(code=4401)