# communications/consumers.py """ WebSocket Consumer Mixin для чатов. Этот модуль содержит ChatConsumerMixin - миксин с chat-функциональностью для WebSocket consumer'ов. Используется в realtime.UserConsumer для обработки всех операций с чатами и сообщениями. История: - Создан: 11 марта 2026 (извлечен из realtime/consumers.py) - Цель: Сделать приложение communications автономным и переиспользуемым """ import logging import rules from channels.db import database_sync_to_async from django.db.models import Prefetch, Q from django.utils import timezone from realtime.protocol import parse_positive_int, parse_text from .mentions import ( build_message_mentions, mentioned_user_ids, parse_mentioned_user_ids, update_unread_mention_counters, ) from .models import ( Chat, ChatReadState, Message, MessageReaction, Poll, ) from .polls import PollActionError, cast_poll_vote from .presence import clear_active_chat, set_active_chat, touch_active_chat from .serialization import serialize_message, serialize_poll from .services import soft_delete_message from .utils import user_can_access_chat logger = logging.getLogger(__name__) class ChatConsumerMixin: """ Mixin для WebSocket consumer с chat-функциональностью. Предоставляет методы для обработки: - Сообщений (отправка, редактирование, удаление) - Реакций (добавление, удаление) - Индикатора "печатает..." - Отметки прочитанного - Голосований Использование: from communications.consumers import ChatConsumerMixin from channels.generic.websocket import AsyncJsonWebsocketConsumer class MyConsumer(ChatConsumerMixin, AsyncJsonWebsocketConsumer): # Mixin автоматически предоставляет все chat_* методы pass Требования: - Consumer должен иметь атрибуты: active_chat_id, user - Consumer должен иметь методы: send_json(), channel_layer События от клиента (action): - open_chat, close_chat - управление активным чатом - send_message, edit_message, delete_message - работа с сообщениями - add_reaction, remove_reaction - реакции - typing, stop_typing - индикатор печати - mark_read - отметка прочитанного - vote_poll - голосование События в channel layer (type): - chat_message - новое сообщение - chat_message_edited - сообщение отредактировано - chat_message_deleted - сообщение удалено - chat_reaction_added/removed - реакция добавлена/удалена - chat_user_typing/stopped_typing - индикатор печати - chat_poll_update - обновление голосования - chat_marked_read - синхронизация отметки прочитанного """ # ==================== Управление активным чатом ==================== @database_sync_to_async def _set_active_chat_presence(self, chat_id): set_active_chat( user_id=self.user.id, chat_id=chat_id, connection_id=self.channel_name, ) @database_sync_to_async def _touch_active_chat_presence(self): touch_active_chat(connection_id=self.channel_name) @database_sync_to_async def _clear_active_chat_presence(self): clear_active_chat(connection_id=self.channel_name) async def _handle_open_chat(self, content): """Открытие чата. Устанавливает активный чат и отправляет начальную историю. """ chat_id = parse_positive_int(content.get("chat_id"), field="chat_id") # Проверяем доступ chat = await self._get_chat(chat_id) if not chat or not await self._user_can_access(chat, self.user): await self.send_json( { "type": "error", "error": "Chat not found or access denied", "chat_id": chat_id, } ) return # Устанавливаем активный чат self.active_chat_id = chat_id await self._set_active_chat_presence(chat_id) # Отправляем начальную историю (если запрошена) if content.get("load_history", False): await self._send_initial_messages(chat_id) # НЕ отмечаем как прочитанное при открытии! # Автоотметка происходит автоматически при загрузке сообщений через: # - GET /messages-around/ → отмечает последнее ЗАГРУЖЕННОЕ # - GET /messages/?after_id= → отмечает последнее ЗАГРУЖЕННОЕ # - WebSocket new_message → клиент вызывает HTTP mark-read, если он внизу # Старый вызов chat.mark_read() отмечал последнее в чате, # что # неправильно! await self.send_json({"type": "chat_opened", "chat_id": chat_id}) logger.info( "WebSocket chat opened", extra={ "event": "websocket.chat.opened", "chat_id": chat_id, "user_id": self.user.id, }, ) async def _handle_close_chat(self, content): """Закрытие чата - сброс активного чата""" chat_id = parse_positive_int(content.get("chat_id"), field="chat_id") if chat_id == self.active_chat_id: # Сбрасываем индикатор печати if self.active_chat_id: chat = await self._get_chat(self.active_chat_id) if chat: await self._set_typing_status(chat, self.user, False) self.active_chat_id = None await self._clear_active_chat_presence() await self.send_json({"type": "chat_closed", "chat_id": chat_id}) logger.info( "WebSocket chat closed", extra={ "event": "websocket.chat.closed", "chat_id": chat_id, "user_id": self.user.id, }, ) # ==================== Обработчики событий из channel layer ============== async def chat_message(self, event): """ Новое сообщение в чате. Отправляем полные данные для активного чата, компактные - для обновления списка. """ chat_id = event.get("chat_id") payload = event.get("payload", {}) # Для активного чата - полное сообщение if chat_id == self.active_chat_id: await self.send_json( {"type": "new_message", "chat_id": chat_id, "message": payload} ) # Для списка чатов - компактное обновление await self.send_json( {"type": "list_update", "chat_id": chat_id, "message": payload} ) async def chat_message_edited(self, event): """Сообщение отредактировано""" chat_id = event.get("chat_id") payload = event.get("payload", {}) # Для активного чата - обновляем сообщение if chat_id == self.active_chat_id: await self.send_json( { "type": "message_updated", "chat_id": chat_id, "message": payload, } ) # Для списка - обновляем если это последнее сообщение await self.send_json( {"type": "message_edited", "chat_id": chat_id, "message": payload} ) async def chat_message_deleted(self, event): """Сообщение удалено""" chat_id = event.get("chat_id") message_id = event.get("message_id") author_id = event.get("author_id") if chat_id == self.active_chat_id: await self.send_json( { "type": "message_deleted", "chat_id": chat_id, "message_id": message_id, "author_id": author_id, } ) logger.debug( "WebSocket message deletion delivered", extra={ "event": "websocket.message_deleted.delivered", "chat_id": chat_id, "message_id": message_id, }, ) await self.send_json( { "type": "list_update", "event": "message_deleted", "chat_id": chat_id, "message_id": message_id, "author_id": author_id, } ) async def chat_reaction_added(self, event): """Реакция добавлена""" chat_id = event.get("chat_id") if chat_id == self.active_chat_id: await self.send_json( { "type": "reaction_added", "chat_id": chat_id, "message_id": event.get("message_id"), "emoji": event.get("emoji"), "user_id": event.get("user_id"), "user_name": event.get("user_name"), "reactions_summary": event.get("reactions_summary"), } ) logger.debug( "WebSocket reaction delivered", extra={ "event": "websocket.reaction.delivered", "reaction_action": "added", "chat_id": chat_id, "message_id": event.get("message_id"), }, ) async def chat_reaction_removed(self, event): """Реакция удалена""" chat_id = event.get("chat_id") if chat_id == self.active_chat_id: await self.send_json( { "type": "reaction_removed", "chat_id": chat_id, "message_id": event.get("message_id"), "emoji": event.get("emoji"), "user_id": event.get("user_id"), "reactions_summary": event.get("reactions_summary"), } ) logger.debug( "WebSocket reaction delivered", extra={ "event": "websocket.reaction.delivered", "reaction_action": "removed", "chat_id": chat_id, "message_id": event.get("message_id"), }, ) async def chat_user_typing(self, event): """Пользователь печатает""" chat_id = event.get("chat_id") user_id = event.get("user_id") # Показываем только в активном чате и не показываем свой статус if chat_id == self.active_chat_id and user_id != self.user.id: await self.send_json( { "type": "typing_start", "chat_id": chat_id, "user_id": user_id, "user_name": event.get("user_name"), } ) async def chat_user_stopped_typing(self, event): """Пользователь перестал печатать""" chat_id = event.get("chat_id") user_id = event.get("user_id") if chat_id == self.active_chat_id and user_id != self.user.id: await self.send_json( {"type": "typing_stop", "chat_id": chat_id, "user_id": user_id} ) async def chat_poll_update(self, event): """Обновление голосования""" chat_id = event.get("chat_id") payload = event.get("payload", {}) if chat_id == self.active_chat_id: await self.send_json( { "type": "poll_update", "chat_id": chat_id, "poll_id": payload.get("poll_id"), "message_id": payload.get("message_id"), "results": payload.get("results"), "voter_user_id": payload.get("voter_user_id"), "selected_option_ids": payload.get( "selected_option_ids", [], ), } ) async def _send_marked_read_to_client(self, event, *, allow_self: bool): chat_id = event.get("chat_id") last_read_message_id = event.get("last_read_message_id") reader_user_id = event.get("reader_user_id") reader = event.get("reader") if chat_id != self.active_chat_id: return if not allow_self and reader_user_id == getattr(self.user, "id", None): return logger.info( "Chat read marker sent to WebSocket client", extra={ "event": "websocket.chat.read_marker_sent", "chat_id": chat_id, "message_id": last_read_message_id, "reader_user_id": reader_user_id, }, ) await self.send_json( { "type": "marked_read", "chat_id": chat_id, "last_read_message_id": last_read_message_id, "reader_user_id": reader_user_id, "reader": reader, } ) async def chat_marked_read(self, event): """Read receipt для других участников активного чата.""" await self._send_marked_read_to_client(event, allow_self=False) async def chat_marked_read_sync(self, event): """Синхронизация отметки прочитанного между вкладками читателя.""" await self._send_marked_read_to_client(event, allow_self=True) # ==================== Обработка действий пользователя ==================== async def _handle_send_message(self, content): """Отправка нового сообщения""" if not self.active_chat_id: await self.send_json({"type": "error", "error": "No active chat"}) return text = parse_text( content.get("content") or content.get("text"), field="content", max_length=20_000, ) mention_user_ids = ( parse_mentioned_user_ids(content.get("mentioned_user_ids")) if "mentioned_user_ids" in content else None ) chat = await self._get_chat(self.active_chat_id) if not chat or not await self._user_can_access(chat, self.user): await self._deny_chat_action("send_message", self.active_chat_id) return if not await self._user_can_send(chat, self.user): await self._deny_chat_action("send_message", self.active_chat_id) return # Проверка прав на отправку if chat.type == "announcement" and chat.created_by_id != self.user.id: await self.send_json( { "type": "error", "error": "Только автор может публиковать в это объявление", } ) return # Создаем сообщение msg = await self._create_message(chat, self.user, text, mention_user_ids) if msg: # Сериализуем и отправляем всем в группе msg_data = await database_sync_to_async(serialize_message)(msg) await self.channel_layer.group_send( f"chat_{chat.id}", { "type": "chat_message", "chat_id": chat.id, "payload": msg_data, }, ) async def _handle_edit_message(self, content): """Редактирование сообщения""" if not self.active_chat_id: return message_id = parse_positive_int(content.get("message_id"), field="message_id") new_content = parse_text( content.get("content"), field="content", max_length=20_000, ) mention_user_ids = ( parse_mentioned_user_ids(content.get("mentioned_user_ids")) if "mentioned_user_ids" in content else None ) msg = await self._get_message(message_id) if ( not msg or msg.chat_id != self.active_chat_id or not await self._user_can_access(msg.chat, self.user) or not await self._user_can_change_message(msg, self.user) ): await self._deny_chat_action("edit_message", self.active_chat_id) return # Обновляем сообщение await self._update_message(msg, new_content, mention_user_ids) # Сериализуем и отправляем msg_data = await database_sync_to_async(serialize_message)(msg) await self.channel_layer.group_send( f"chat_{msg.chat_id}", { "type": "chat_message_edited", "chat_id": msg.chat_id, "payload": msg_data, }, ) async def _handle_delete_message(self, content): """Удаление сообщения""" if not self.active_chat_id: return message_id = parse_positive_int(content.get("message_id"), field="message_id") msg = await self._get_message(message_id) if ( not msg or msg.chat_id != self.active_chat_id or not await self._user_can_access(msg.chat, self.user) or not await self._user_can_delete_message(msg, self.user) ): await self._deny_chat_action("delete_message", self.active_chat_id) return chat_id = msg.chat_id # Помечаем как удаленное await self._soft_delete_message(msg) await self.channel_layer.group_send( f"chat_{chat_id}", { "type": "chat_message_deleted", "chat_id": chat_id, "message_id": message_id, "author_id": msg.author_id, }, ) async def _handle_add_reaction(self, content): """Добавление реакции""" message_id = parse_positive_int(content.get("message_id"), field="message_id") emoji = parse_text(content.get("emoji"), field="emoji", max_length=32) msg = await self._get_message(message_id) if ( not msg or msg.chat_id != self.active_chat_id or not await self._user_can_access(msg.chat, self.user) ): await self._deny_chat_action("add_reaction", getattr(msg, "chat_id", None)) return # Создаем реакцию created = await self._add_reaction(msg, self.user, emoji) if created: user_name = await database_sync_to_async( lambda: self.user.get_full_name() or self.user.username )() await self.channel_layer.group_send( f"chat_{msg.chat_id}", { "type": "chat_reaction_added", "chat_id": msg.chat_id, "message_id": message_id, "emoji": emoji, "user_id": self.user.id, "user_name": user_name, }, ) async def _handle_remove_reaction(self, content): """Удаление реакции""" message_id = parse_positive_int(content.get("message_id"), field="message_id") emoji = parse_text(content.get("emoji"), field="emoji", max_length=32) msg = await self._get_message(message_id) if ( not msg or msg.chat_id != self.active_chat_id or not await self._user_can_access(msg.chat, self.user) ): await self._deny_chat_action( "remove_reaction", getattr(msg, "chat_id", None), ) return # Удаляем реакцию deleted = await self._remove_reaction(msg, self.user, emoji) if deleted: await self.channel_layer.group_send( f"chat_{msg.chat_id}", { "type": "chat_reaction_removed", "chat_id": msg.chat_id, "message_id": message_id, "emoji": emoji, "user_id": self.user.id, }, ) async def _handle_typing(self, content): """Индикатор печати""" if not self.active_chat_id: return chat = await self._get_chat(self.active_chat_id) if ( not chat or not await self._user_can_access(chat, self.user) or not await self._user_can_send(chat, self.user) ): return await self._set_typing_status(chat, self.user, True) user_name = await database_sync_to_async( lambda: self.user.get_full_name() or self.user.username )() await self.channel_layer.group_send( f"chat_{chat.id}", { "type": "chat_user_typing", "chat_id": chat.id, "user_id": self.user.id, "user_name": user_name, }, ) async def _handle_stop_typing(self, content=None): """Остановка индикатора печати""" if not self.active_chat_id: return chat = await self._get_chat(self.active_chat_id) if not chat or not await self._user_can_access(chat, self.user): return 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, }, ) async def _handle_mark_read(self, content): """ [DEPRECATED] Отметка сообщений как прочитанных через WebSocket. Используйте POST /api/v1/communications/chats/{id}/mark-read/ вместо этого. Автоотметка происходит автоматически при загрузке сообщений. """ chat_id = parse_positive_int(content.get("chat_id"), field="chat_id") logger.warning( "Deprecated WebSocket read marker call ignored", extra={ "event": "websocket.chat.deprecated_read_marker_ignored", "user_id": self.user.id, "chat_id": chat_id, }, ) # НЕ вызываем _mark_read! Автоотметка происходит автоматически. # Если нужна точная отметка, # frontend должен вызвать POST API endpoint. async def _handle_vote_poll(self, content): """Голосование в опросе""" poll_id = parse_positive_int(content.get("poll_id"), field="poll_id") option_ids = content.get("option_ids") if ( not self.active_chat_id or not isinstance(option_ids, list) or len(option_ids) > 50 ): await self.send_json( { "type": "error", "code": "poll_vote_failed", "error": "Некорректные данные голосования", } ) return option_ids = [ parse_positive_int(option_id, field="option_ids") for option_id in option_ids ] chat = await self._get_chat(self.active_chat_id) if not chat or not await self._user_can_access(chat, self.user): await self._deny_chat_action("vote_poll", self.active_chat_id) return result = await self._cast_poll_vote( poll_id, option_ids, self.active_chat_id, self.user, ) if not result["ok"]: await self.send_json( { "type": "error", "code": "poll_vote_failed", "error": result["error"], } ) return await self.channel_layer.group_send( f"chat_{self.active_chat_id}", { "type": "chat_poll_update", "chat_id": self.active_chat_id, "payload": { "poll_id": result["poll_id"], "message_id": result["message_id"], "results": result["results"], }, }, ) await self.channel_layer.group_send( f"user_{self.user.id}", { "type": "poll_update_sync", "chat_id": self.active_chat_id, "payload": { "poll_id": result["poll_id"], "message_id": result["message_id"], "results": result["results"], "voter_user_id": self.user.id, "selected_option_ids": result["selected_option_ids"], }, }, ) # ==================== Вспомогательные методы ==================== async def _send_initial_messages(self, chat_id, limit=50): """Отправка начальной истории сообщений""" messages = await self._get_recent_messages(chat_id, limit) await self.send_json( { "type": "initial_messages", "chat_id": chat_id, "messages": messages, } ) @database_sync_to_async def _get_available_chat_ids(self, user): """Получить ID всех доступных чатов пользователя""" candidates = ( Chat.objects.filter( Q(type="global") | Q(memberships__user=user, memberships__is_active=True) | Q(participants=user) | Q(created_by=user) | Q(include_all_users=True) | Q(type="comments", flags__show_in_messages=True) ) .select_related("created_by", "context_content_type") .distinct() ) return [chat.id for chat in candidates if user_can_access_chat(chat, user)] @database_sync_to_async def _get_chat(self, chat_id): """Получить чат по ID""" try: return Chat.objects.get(pk=chat_id) except Chat.DoesNotExist: return None @database_sync_to_async def _get_message(self, message_id): """Получить сообщение по ID""" try: return Message.objects.select_related("author", "chat").get(pk=message_id) except Message.DoesNotExist: return None @database_sync_to_async def _user_can_access(self, chat, user): """Проверка доступа пользователя к чату""" return user_can_access_chat(chat, user) @database_sync_to_async def _user_can_send(self, chat, user): return rules.test_rule("communications.send_message", user, chat) @database_sync_to_async def _user_can_change_message(self, message, user): return rules.test_rule("communications.change_message", user, message) @database_sync_to_async def _user_can_delete_message(self, message, user): return rules.test_rule("communications.delete_message", user, message) async def _deny_chat_action(self, action, chat_id): await self.send_json( { "type": "error", "code": "access_denied", "action": action, "chat_id": chat_id, "error": "Чат не найден или доступ запрещён.", } ) @database_sync_to_async def _create_message(self, chat, user, content, mention_user_ids=None): """Создать новое сообщение""" try: msg = Message.objects.create( chat=chat, author=user, content=content, mentions=build_message_mentions( chat, user, content, mention_user_ids or [], ), ) return msg except Exception as e: logger.exception( "Chat message creation failed", extra={ "event": "chat.message.creation_failed", "chat_id": chat.id, "user_id": user.id, "error_type": type(e).__name__, }, ) return None @database_sync_to_async def _update_message(self, msg, new_content, mention_user_ids=None): """Обновить содержимое сообщения""" previous_mention_ids = mentioned_user_ids(msg.mentions) requested_mention_ids = ( mention_user_ids if mention_user_ids is not None else list(previous_mention_ids) ) next_mentions = build_message_mentions( msg.chat, msg.author, new_content, requested_mention_ids, ) added_mention_ids = mentioned_user_ids(next_mentions) - previous_mention_ids removed_mention_ids = previous_mention_ids - mentioned_user_ids(next_mentions) msg.content = new_content msg.mentions = next_mentions msg.is_edited = True msg.edited_at = timezone.now() msg.save(update_fields=["content", "mentions", "is_edited", "edited_at"]) update_unread_mention_counters( msg, added_user_ids=added_mention_ids, removed_user_ids=removed_mention_ids, ) if added_mention_ids: from communications.notifications.handlers import notify_message_mentions notify_message_mentions(msg, recipient_ids=added_mention_ids) from notifications.push_lifecycle import replace_message_notifications replace_message_notifications(msg) @database_sync_to_async def _soft_delete_message(self, msg): """Мягкое удаление сообщения""" return soft_delete_message( msg, deleted_by=self.user, replacement_content="[Сообщение удалено]", ) @database_sync_to_async def _add_reaction(self, msg, user, emoji): """Добавить реакцию""" reaction, created = MessageReaction.objects.get_or_create( message=msg, user=user, emoji=emoji ) return created @database_sync_to_async def _remove_reaction(self, msg, user, emoji): """Удалить реакцию""" deleted_count, _ = MessageReaction.objects.filter( message=msg, user=user, emoji=emoji ).delete() return deleted_count > 0 @database_sync_to_async def _cast_poll_vote(self, poll_id, option_ids, chat_id, user): try: poll = Poll.objects.select_related( "message", "message__chat", ).get(pk=poll_id, message__chat_id=chat_id) except Poll.DoesNotExist: return {"ok": False, "error": "Голосование не найдено"} try: poll, selected_option_ids = cast_poll_vote( poll, user, option_ids, ) except PollActionError as exc: return {"ok": False, "error": exc.message} return { "ok": True, "poll_id": poll.id, "message_id": poll.message_id, "selected_option_ids": selected_option_ids, "results": serialize_poll(poll), } @database_sync_to_async def _mark_read(self, chat, user): """ [DEPRECATED] НЕ ИСПОЛЬЗУЕТСЯ! Автоотметка происходит автоматически через ChatViewSet._auto_mark_read() при загрузке сообщений через GET запросы. Этот метод отмечал ПОСЛЕДНЕЕ СООБЩЕНИЕ В ЧАТЕ, что неправильно! Правильная логика: отмечать последнее ЗАГРУЖЕННОЕ сообщение. """ logger.warning( "Deprecated internal chat read marker call ignored", extra={ "event": "chat.deprecated_read_marker_ignored", "chat_id": chat.id, "user_id": user.id, }, ) # НЕ вызываем chat.mark_read(user)! # Это отмечало последнее сообщение в чате вместо последнего # загруженного. @database_sync_to_async def _set_typing_status(self, chat, user, is_typing): """Установить статус печати (можно кэшировать в Redis)""" # TODO: Реализовать через Redis для производительности pass @database_sync_to_async def _get_recent_messages(self, chat_id, limit=50): """Получить последние сообщения чата""" messages = ( Message.objects.filter(chat_id=chat_id) .select_related("author", "chat") .prefetch_related( "attachments", "reactions", Prefetch( "chat__read_states", queryset=ChatReadState.objects.select_related("user"), ), ) .order_by("-created_at")[:limit] ) # Реверсируем порядок (старые -> новые) messages = list(reversed(messages)) return [serialize_message(msg) for msg in messages] __all__ = ["ChatConsumerMixin"]