647 lines
24 KiB
Python
647 lines
24 KiB
Python
# 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)
|