diff --git a/chats/consumers/chat.py b/chats/consumers/chat.py index a93d615f..305d7f22 100644 --- a/chats/consumers/chat.py +++ b/chats/consumers/chat.py @@ -3,7 +3,6 @@ from json import JSONDecodeError from typing import Optional -from asgiref.sync import sync_to_async from channels.db import database_sync_to_async from channels.generic.websocket import AsyncJsonWebsocketConsumer from django.conf import settings @@ -208,22 +207,11 @@ async def __process_typing_event(self, event: Event, room_name: str): event_data["_project_chat_id"] = event.content["chat_id"] if event.content["chat_type"] == ChatType.DIRECT: - # fixme: need to move this to func - chat_id, other_user = await get_chat_and_user_ids_from_content( + _, other_user = await get_chat_and_user_ids_from_content( event.content, self.user ) - # if chat_id == 17_7, then chat_id will be == 7_17 - chat_id = DirectChat.get_chat_id_from_users(self.user, other_user) - - # check if chat exists - try: - await sync_to_async(DirectChat.objects.get)(pk=chat_id) - except DirectChat.DoesNotExist: - # if not, create such chat - await sync_to_async(DirectChat.create_from_two_users)( - self.user, other_user - ) + await database_sync_to_async(DirectChat.get_chat)(self.user, other_user) # send message to user's channel other_user_channel = cache.get(get_user_channel_cache_key(other_user), None) diff --git a/chats/consumers/event_types/DirectEvent.py b/chats/consumers/event_types/DirectEvent.py index 37d95053..3fa43b9a 100644 --- a/chats/consumers/event_types/DirectEvent.py +++ b/chats/consumers/event_types/DirectEvent.py @@ -1,3 +1,5 @@ +from channels.db import database_sync_to_async + from chats.models import DirectChatMessage, DirectChat from chats.websockets_settings import Event, EventType from chats.exceptions import ( @@ -8,16 +10,12 @@ from django.core.cache import cache from chats.utils import ( get_user_channel_cache_key, - create_message, get_chat_and_user_ids_from_content, - match_files_and_messages, - orm_create, - orm_exists, orm_get, orm_save, - orm_set, ) from chats.serializers import DirectChatMessageListSerializer +from chats.services import create_direct_message class DirectEvent: @@ -27,53 +25,22 @@ def __init__(self, user, channel_layer, channel_name): self.channel_name = channel_name async def process_new_message_event(self, event: Event, room_name: str): - chat_id, other_user = await get_chat_and_user_ids_from_content( - event.content, self.user - ) - - # if chat_id == 17_7, then chat_id will be == 7_17 + _, other_user = await get_chat_and_user_ids_from_content(event.content, self.user) chat_id = DirectChat.get_chat_id_from_users(self.user, other_user) - # check if chat exists - if not await orm_exists(DirectChat.objects.filter(pk=chat_id)): - # if not, create such chat - chat = await orm_create(DirectChat.objects, pk=chat_id) - await orm_set(chat.users, [self.user, other_user]) - - reply_to_message = None - if event.content["reply_to"] is not None: - try: - reply_to_message = await orm_get( - DirectChatMessage.objects, pk=event.content["reply_to"] - ) - except DirectChatMessage.DoesNotExist: - reply_to_message = None - - msg = await create_message( - chat_id=chat_id, - chat_model=DirectChatMessage, + message_data, created = await database_sync_to_async(create_direct_message)( author=self.user, - text=event.content["text"], - reply_to=reply_to_message, - ) - - messages = { - "direct_message": msg, - "project_message": None, - } - await match_files_and_messages(event.content["file_urls"], messages) - - serialized_message = await orm_get( - DirectChatMessage.objects.select_related("author", "reply_to__author").prefetch_related( - "file_to_message__file" - ), - pk=msg.pk, + other_user=other_user, + text=event.content.get("text", ""), + reply_to_id=event.content.get("reply_to"), + file_urls=event.content.get("file_urls", []), + client_message_id=event.content.get("client_message_id"), ) - message_data = DirectChatMessageListSerializer(serialized_message).data content = { "chat_id": chat_id, "message": message_data, + "is_replay": not created, } # send message to user's channel @@ -83,7 +50,7 @@ async def process_new_message_event(self, event: Event, room_name: str): await self.channel_layer.send(self.channel_name, event_data) - if other_user_channel: + if created and other_user_channel and other_user_channel != self.channel_name: await self.channel_layer.send(other_user_channel, event_data) async def process_read_message_event(self, event: Event, room_name: str): @@ -164,9 +131,9 @@ async def process_edit_message_event(self, event, room_name): await orm_save(msg, update_fields=["text", "is_edited"]) serialized_message = await orm_get( - DirectChatMessage.objects.select_related("author", "reply_to__author").prefetch_related( - "file_to_message__file" - ), + DirectChatMessage.objects.select_related( + "author", "reply_to__author" + ).prefetch_related("file_to_message__file"), pk=msg.pk, ) message_data = DirectChatMessageListSerializer(serialized_message).data diff --git a/chats/consumers/event_types/ProjectEvent.py b/chats/consumers/event_types/ProjectEvent.py index 96461184..d74b3368 100644 --- a/chats/consumers/event_types/ProjectEvent.py +++ b/chats/consumers/event_types/ProjectEvent.py @@ -1,11 +1,12 @@ +from channels.db import database_sync_to_async + from chats.models import ProjectChat, ProjectChatMessage from chats.utils import ( - create_message, - match_files_and_messages, orm_exists, orm_get, orm_save, ) +from chats.services import create_project_message from chats.websockets_settings import Event, EventType from chats.exceptions import ( WrongChatIdException, @@ -28,53 +29,18 @@ def __init__(self, user, channel_layer, channel_name): async def process_new_message_event(self, event: Event, room_name: str): chat_id = event.content["chat_id"] - chat = await orm_get( - ProjectChat.objects.select_related("project__leader"), - pk=chat_id, - ) - - # check that user is in this chat - is_member = chat.project.leader_id == self.user.id or await orm_exists( - Collaborator.objects.filter(project_id=chat.project_id, user_id=self.user.id) - ) - if not is_member: - raise UserNotInChatException( - f"User {self.user.id} is not in project chat {chat_id}" - ) - - reply_to_message = None - if event.content["reply_to"] is not None: - try: - reply_to_message = await orm_get( - ProjectChatMessage.objects, pk=event.content["reply_to"] - ) - except ProjectChatMessage.DoesNotExist: - reply_to_message = None - - msg = await create_message( + message_data, created = await database_sync_to_async(create_project_message)( chat_id=chat_id, - chat_model=ProjectChatMessage, author=self.user, - text=event.content["text"], - reply_to=reply_to_message, + text=event.content.get("text", ""), + reply_to_id=event.content.get("reply_to"), + file_urls=event.content.get("file_urls", []), + client_message_id=event.content.get("client_message_id"), ) - - messages = { - "direct_message": None, - "project_message": msg, - } - await match_files_and_messages(event.content["file_urls"], messages) - - serialized_message = await orm_get( - ProjectChatMessage.objects.select_related( - "author", "reply_to__author" - ).prefetch_related("file_to_message__file"), - pk=msg.pk, - ) - message_data = ProjectChatMessageListSerializer(serialized_message).data content = { "chat_id": chat_id, "message": message_data, + "is_replay": not created, } event_data = { @@ -83,7 +49,10 @@ async def process_new_message_event(self, event: Event, room_name: str): "_project_chat_id": chat_id, } - await self.channel_layer.group_send(room_name, event_data) + if created: + await self.channel_layer.group_send(room_name, event_data) + else: + await self.channel_layer.send(self.channel_name, event_data) async def process_read_message_event(self, event: Event, room_name: str): msg = await orm_get(ProjectChatMessage.objects, pk=event.content["message_id"]) diff --git a/chats/exceptions.py b/chats/exceptions.py index fd9a2a80..ee09075d 100644 --- a/chats/exceptions.py +++ b/chats/exceptions.py @@ -23,5 +23,5 @@ class UserNotMessageAuthorException(ChatException): pass -class NonMatchingReplyChatIdException(ChatException): +class MessageValidationException(ChatException): pass diff --git a/chats/migrations/0012_directchatmessage_client_message_id_and_more.py b/chats/migrations/0012_directchatmessage_client_message_id_and_more.py new file mode 100644 index 00000000..9303db59 --- /dev/null +++ b/chats/migrations/0012_directchatmessage_client_message_id_and_more.py @@ -0,0 +1,41 @@ +# Generated by Django 5.2.18 on 2026-11-13 16:33 + +from django.conf import settings +from django.db import migrations, models + + +class Migration(migrations.Migration): + + dependencies = [ + ("chats", "0011_alter_filetomessage_direct_message_and_more"), + migrations.swappable_dependency(settings.AUTH_USER_MODEL), + ] + + operations = [ + migrations.AddField( + model_name="directchatmessage", + name="client_message_id", + field=models.UUIDField(blank=True, editable=False, null=True), + ), + migrations.AddField( + model_name="projectchatmessage", + name="client_message_id", + field=models.UUIDField(blank=True, editable=False, null=True), + ), + migrations.AddConstraint( + model_name="directchatmessage", + constraint=models.UniqueConstraint( + condition=models.Q(("client_message_id__isnull", False)), + fields=("chat", "author", "client_message_id"), + name="unique_direct_chat_client_message", + ), + ), + migrations.AddConstraint( + model_name="projectchatmessage", + constraint=models.UniqueConstraint( + condition=models.Q(("client_message_id__isnull", False)), + fields=("chat", "author", "client_message_id"), + name="unique_project_chat_client_message", + ), + ), + ] diff --git a/chats/models.py b/chats/models.py index 3a84ddbd..90e1ca27 100644 --- a/chats/models.py +++ b/chats/models.py @@ -3,7 +3,7 @@ from django.contrib.auth import get_user_model from django.core.exceptions import ValidationError -from django.db import models +from django.db import models, transaction from files.models import UserFile from projects.models import Project @@ -137,6 +137,7 @@ def get_avatar(self, user): return other_user.avatar @classmethod + @transaction.atomic def get_chat(cls, user1, user2) -> "DirectChat": """ Returns chat between two users. @@ -150,12 +151,10 @@ def get_chat(cls, user1, user2) -> "DirectChat": """ pk = cls.get_chat_id_from_users(user1, user2) - try: - return cls.objects.get(pk=pk) - except cls.DoesNotExist: - chat = cls.objects.create(pk=pk) + chat, created = cls.objects.get_or_create(pk=pk) + if created: chat.users.set([user1, user2]) - return chat + return chat def get_last_messages(self, message_count): return self.messages.order_by("-created_at")[:message_count] @@ -163,12 +162,6 @@ def get_last_messages(self, message_count): def get_other_user(self, user) -> User: return self.users.exclude(pk=user.pk).first() - @classmethod - def create_from_two_users(cls, user1, user2): - chat = cls.objects.create(pk=cls.get_chat_id_from_users(user1, user2)) - chat.users.set([user1, user2]) - return chat - @classmethod def get_chat_id_from_users(cls, user1, user2): first_user = user1 if user1.pk < user2.pk else user2 @@ -198,6 +191,7 @@ class BaseMessage(models.Model): is_read = models.BooleanField(default=False) is_deleted = models.BooleanField(default=False) is_edited = models.BooleanField(default=False) + client_message_id = models.UUIDField(null=True, blank=True, editable=False) created_at = models.DateTimeField(auto_now_add=True) def __str__(self): @@ -246,6 +240,13 @@ def __str__(self): class Meta: verbose_name = "Сообщение в чате проекта" verbose_name_plural = "Сообщения в чатах проектов" + constraints = [ + models.UniqueConstraint( + fields=["chat", "author", "client_message_id"], + condition=models.Q(client_message_id__isnull=False), + name="unique_project_chat_client_message", + ) + ] class DirectChatMessage(BaseMessage): @@ -284,6 +285,13 @@ def __str__(self): class Meta: verbose_name = "Сообщение в личном чате" verbose_name_plural = "Сообщения в личных чатах" + constraints = [ + models.UniqueConstraint( + fields=["chat", "author", "client_message_id"], + condition=models.Q(client_message_id__isnull=False), + name="unique_direct_chat_client_message", + ) + ] class FileToMessage(models.Model): diff --git a/chats/serializers.py b/chats/serializers.py index 27c8cc70..44110f6e 100644 --- a/chats/serializers.py +++ b/chats/serializers.py @@ -147,6 +147,7 @@ class Meta: "id", "author", "text", + "client_message_id", "reply_to", "files", "is_edited", @@ -196,6 +197,7 @@ class Meta: "id", "author", "text", + "client_message_id", "files", "reply_to", "is_edited", diff --git a/chats/services/__init__.py b/chats/services/__init__.py new file mode 100644 index 00000000..c3ec9741 --- /dev/null +++ b/chats/services/__init__.py @@ -0,0 +1,6 @@ +from chats.services.messages import ( + create_direct_message, + create_project_message, +) + +__all__ = ["create_direct_message", "create_project_message"] diff --git a/chats/services/messages.py b/chats/services/messages.py new file mode 100644 index 00000000..4fa74daa --- /dev/null +++ b/chats/services/messages.py @@ -0,0 +1,199 @@ +from uuid import UUID + +from django.db import transaction + +from chats.exceptions import MessageValidationException +from chats.models import ( + DirectChat, + DirectChatMessage, + FileToMessage, + ProjectChat, + ProjectChatMessage, +) +from chats.permissions import is_project_member +from chats.serializers import ( + DirectChatMessageListSerializer, + ProjectChatMessageListSerializer, +) +from files.models import UserFile + + +def _normalize_client_message_id(value): + if value in (None, ""): + return None + try: + return UUID(str(value)) + except (TypeError, ValueError, AttributeError) as error: + raise MessageValidationException( + "Некорректный идентификатор отправляемого сообщения." + ) from error + + +def _normalize_message(text, file_urls): + if not isinstance(text, str) or "\x00" in text: + raise MessageValidationException("Некорректный текст сообщения.") + if file_urls is None: + file_urls = [] + if not isinstance(file_urls, list) or any( + not isinstance(url, str) or not url or "\x00" in url for url in file_urls + ): + raise MessageValidationException("Некорректный список вложений.") + if not text.strip() and not file_urls: + raise MessageValidationException("Сообщение не может быть пустым.") + if len(text) > DirectChatMessage._meta.get_field("text").max_length: + raise MessageValidationException( + "Текст сообщения не должен превышать 8192 символа." + ) + return list(dict.fromkeys(file_urls)) + + +def _get_reply(message_model, chat_id, reply_to_id): + if reply_to_id is None: + return None + try: + return message_model.objects.get( + pk=reply_to_id, + chat_id=chat_id, + is_deleted=False, + ) + except (message_model.DoesNotExist, TypeError, ValueError) as error: + raise MessageValidationException( + "Нельзя ответить на сообщение из другого чата или удаленное сообщение." + ) from error + + +def _get_files(file_urls, author): + if not file_urls: + return [] + files = { + user_file.pk: user_file + for user_file in UserFile.objects.select_for_update() + .filter( + pk__in=file_urls, + user_id=author.pk, + ) + .order_by("pk") + } + if len(files) != len(file_urls): + raise MessageValidationException( + "Один или несколько файлов недоступны для отправки." + ) + if FileToMessage.objects.filter(file_id__in=file_urls).exists(): + raise MessageValidationException( + "Один или несколько файлов уже прикреплены к сообщению." + ) + return [files[url] for url in file_urls] + + +def _serialize_message(message_model, serializer_class, message_id): + message = ( + message_model.objects.select_related("author", "reply_to__author") + .prefetch_related("file_to_message__file") + .get(pk=message_id) + ) + return serializer_class(message).data + + +def _create_message( + *, + message_model, + serializer_class, + message_link_field, + chat_id, + author, + text, + reply_to_id, + file_urls, + client_message_id, +): + client_message_id = _normalize_client_message_id(client_message_id) + if client_message_id is not None: + existing_message = message_model.objects.filter( + chat_id=chat_id, + author=author, + client_message_id=client_message_id, + ).first() + if existing_message is not None: + if existing_message.is_deleted: + raise MessageValidationException("Сообщение уже удалено.") + return ( + _serialize_message( + message_model, + serializer_class, + existing_message.pk, + ), + False, + ) + + file_urls = _normalize_message(text, file_urls) + reply_to = _get_reply(message_model, chat_id, reply_to_id) + files = _get_files(file_urls, author) + message = message_model.objects.create( + chat_id=chat_id, + author=author, + text=text, + reply_to=reply_to, + client_message_id=client_message_id, + ) + for user_file in files: + FileToMessage.objects.create(file=user_file, **{message_link_field: message}) + return _serialize_message(message_model, serializer_class, message.pk), True + + +@transaction.atomic +def create_direct_message( + *, + author, + other_user, + text, + reply_to_id=None, + file_urls=None, + client_message_id=None, +): + chat = DirectChat.get_chat(author, other_user) + # Serialize sends in this chat before checking for a repeat or locking files. + DirectChat.objects.select_for_update().get(pk=chat.pk) + return _create_message( + message_model=DirectChatMessage, + serializer_class=DirectChatMessageListSerializer, + message_link_field="direct_message", + chat_id=chat.pk, + author=author, + text=text, + reply_to_id=reply_to_id, + file_urls=file_urls, + client_message_id=client_message_id, + ) + + +@transaction.atomic +def create_project_message( + *, + chat_id, + author, + text, + reply_to_id=None, + file_urls=None, + client_message_id=None, +): + try: + chat = ( + ProjectChat.objects.select_related("project") + .select_for_update(of=("self",)) + .get(pk=chat_id) + ) + except (ProjectChat.DoesNotExist, TypeError, ValueError) as error: + raise MessageValidationException("У вас нет доступа к чату проекта.") from error + if not is_project_member(author, chat.project): + raise MessageValidationException("У вас нет доступа к чату проекта.") + return _create_message( + message_model=ProjectChatMessage, + serializer_class=ProjectChatMessageListSerializer, + message_link_field="project_message", + chat_id=chat.pk, + author=author, + text=text, + reply_to_id=reply_to_id, + file_urls=file_urls, + client_message_id=client_message_id, + ) diff --git a/chats/tests/test_message_delivery.py b/chats/tests/test_message_delivery.py new file mode 100644 index 00000000..a3b68b67 --- /dev/null +++ b/chats/tests/test_message_delivery.py @@ -0,0 +1,247 @@ +from concurrent.futures import ThreadPoolExecutor +from threading import Barrier +from unittest.mock import patch +from uuid import uuid4 + +from channels.testing import WebsocketCommunicator +from django.contrib.auth import get_user_model +from django.db import IntegrityError, connections +from django.test import TransactionTestCase, override_settings, skipUnlessDBFeature + +from chats.consumers import ChatConsumer +from chats.models import DirectChat, DirectChatMessage, FileToMessage, ProjectChat +from chats.models import ProjectChatMessage +from chats.services import create_direct_message, create_project_message +from chats.tests.constants import TEST_USER1, TEST_USER2, TEST_USER3 +from files.models import UserFile +from projects.models import Collaborator, Project + + +class MessageDeliveryTests(TransactionTestCase): + reset_sequences = True + + def setUp(self): + super().setUp() + self.user = get_user_model().objects.create(**TEST_USER1) + + async def connect(self, user=None): + communicator = WebsocketCommunicator(ChatConsumer.as_asgi(), "/ws/chat/") + communicator.scope["user"] = user or self.user + connected, _ = await communicator.connect(timeout=5) + self.assertTrue(connected) + return communicator + + @staticmethod + def direct_message(chat_id, **overrides): + content = { + "chat_type": "direct", + "chat_id": chat_id, + "text": "Тестовое сообщение", + "reply_to": None, + "file_urls": [], + "client_message_id": str(uuid4()), + } + content.update(overrides) + return {"type": "new_message", "content": content} + + async def test_direct_retry_returns_original_message_without_duplicate(self): + communicator = await self.connect() + data = self.direct_message("1_1") + + await communicator.send_json_to(data) + first_response = await communicator.receive_json_from() + await communicator.send_json_to(data) + second_response = await communicator.receive_json_from() + + self.assertEqual(DirectChatMessage.objects.count(), 1) + self.assertEqual( + first_response["content"]["message"]["id"], + second_response["content"]["message"]["id"], + ) + self.assertFalse(first_response["content"]["is_replay"]) + self.assertTrue(second_response["content"]["is_replay"]) + self.assertEqual( + second_response["content"]["message"]["client_message_id"], + data["content"]["client_message_id"], + ) + self.assertTrue(await communicator.receive_nothing()) + await communicator.disconnect() + + async def test_file_only_message_can_be_retried(self): + user_file = UserFile.objects.create( + link="https://cdn.example.test/chat-file.pdf", + user=self.user, + name="chat-file", + extension="pdf", + mime_type="application/pdf", + detected_mime_type="application/pdf", + size=128, + ) + communicator = await self.connect() + data = self.direct_message( + "1_1", + text="", + file_urls=[user_file.pk], + ) + + await communicator.send_json_to(data) + response = await communicator.receive_json_from() + await communicator.send_json_to(data) + replay = await communicator.receive_json_from() + + self.assertNotIn("error", response) + self.assertEqual(DirectChatMessage.objects.count(), 1) + self.assertEqual(FileToMessage.objects.count(), 1) + self.assertEqual(len(response["content"]["message"]["files"]), 1) + self.assertTrue(replay["content"]["is_replay"]) + self.assertEqual(replay["content"]["message"], response["content"]["message"]) + await communicator.disconnect() + + async def test_file_from_another_account_is_rejected(self): + other_user = get_user_model().objects.create(**TEST_USER2) + user_file = UserFile.objects.create( + link="https://cdn.example.test/foreign-file.pdf", + user=other_user, + name="foreign-file", + extension="pdf", + mime_type="application/pdf", + detected_mime_type="application/pdf", + size=128, + ) + communicator = await self.connect() + data = self.direct_message("1_1", file_urls=[user_file.pk]) + + await communicator.send_json_to(data) + response = await communicator.receive_json_from() + + self.assertIn("error", response) + self.assertEqual(DirectChatMessage.objects.count(), 0) + self.assertEqual(DirectChat.objects.count(), 0) + await communicator.disconnect() + + async def test_reply_to_another_chat_is_rejected(self): + get_user_model().objects.create(**TEST_USER2) + get_user_model().objects.create(**TEST_USER3) + communicator = await self.connect() + + await communicator.send_json_to(self.direct_message("1_2")) + first_response = await communicator.receive_json_from() + reply_id = first_response["content"]["message"]["id"] + await communicator.send_json_to(self.direct_message("1_3", reply_to=reply_id)) + response = await communicator.receive_json_from() + + self.assertIn("error", response) + self.assertEqual(DirectChatMessage.objects.count(), 1) + await communicator.disconnect() + + +class ProjectMessageDeliveryTests(TransactionTestCase): + reset_sequences = True + + def setUp(self): + super().setUp() + self.leader = get_user_model().objects.create(**TEST_USER1) + self.member = get_user_model().objects.create(**TEST_USER2) + self.project = Project.objects.create(leader=self.leader) + self.chat, _ = ProjectChat.objects.get_or_create( + id=self.project.pk, + defaults={"project": self.project}, + ) + Collaborator.objects.create( + user=self.member, + project=self.project, + role="User", + ) + + async def connect(self, user): + communicator = WebsocketCommunicator(ChatConsumer.as_asgi(), "/ws/chat/") + communicator.scope["user"] = user + connected, _ = await communicator.connect(timeout=5) + self.assertTrue(connected) + return communicator + + @override_settings(RUNNING_TESTS=False) + async def test_project_retry_is_not_broadcast_as_a_new_message(self): + client_message_id = str(uuid4()) + data = { + "type": "new_message", + "content": { + "chat_type": "project", + "chat_id": str(self.chat.pk), + "text": "Сообщение проекта", + "reply_to": None, + "file_urls": [], + "client_message_id": client_message_id, + }, + } + communicator = await self.connect(self.member) + recipient = await self.connect(self.leader) + self.assertEqual((await communicator.receive_json_from())["type"], "set_online") + + await communicator.send_json_to(data) + first_response = await communicator.receive_json_from() + self.assertEqual(await recipient.receive_json_from(), first_response) + await communicator.send_json_to(data) + second_response = await communicator.receive_json_from() + + self.assertEqual(ProjectChatMessage.objects.count(), 1) + self.assertEqual( + first_response["content"]["message"]["id"], + second_response["content"]["message"]["id"], + ) + self.assertFalse(first_response["content"]["is_replay"]) + self.assertTrue(second_response["content"]["is_replay"]) + self.assertTrue(await recipient.receive_nothing()) + await communicator.disconnect() + await recipient.disconnect() + + @skipUnlessDBFeature("has_select_for_update") + def test_simultaneous_project_retries_keep_one_message_and_one_attachment(self): + user_file = UserFile.objects.create( + link="https://cdn.example.test/concurrent.pdf", user=self.member + ) + client_message_id = uuid4() + ready = Barrier(2) + + def send(): + try: + ready.wait(timeout=10) + return create_project_message( + chat_id=self.chat.pk, + author=self.member, + text="", + file_urls=[user_file.pk], + client_message_id=client_message_id, + ) + finally: + connections.close_all() + + with ThreadPoolExecutor(max_workers=2) as pool: + results = [pool.submit(send) for _ in range(2)] + (first, first_created), (second, second_created) = [ + result.result(timeout=10) for result in results + ] + + self.assertCountEqual([first_created, second_created], [True, False]) + self.assertEqual(first["id"], second["id"]) + self.assertEqual(ProjectChatMessage.objects.count(), 1) + self.assertEqual(FileToMessage.objects.count(), 1) + + def test_failed_attachment_write_rolls_back_new_direct_chat_and_message(self): + user_file = UserFile.objects.create( + link="https://cdn.example.test/rollback.pdf", user=self.member + ) + with ( + patch.object(FileToMessage.objects, "create", side_effect=IntegrityError), + self.assertRaises(IntegrityError), + ): + create_direct_message( + author=self.member, + other_user=self.leader, + text="", + file_urls=[user_file.pk], + client_message_id=uuid4(), + ) + self.assertFalse(DirectChat.objects.exists()) + self.assertFalse(DirectChatMessage.objects.exists()) + self.assertFalse(FileToMessage.objects.exists()) diff --git a/chats/utils.py b/chats/utils.py index e92ffc05..8d85a7a0 100644 --- a/chats/utils.py +++ b/chats/utils.py @@ -1,16 +1,10 @@ -from typing import Union, Type - from django.conf import settings from django.contrib.auth import get_user_model -from django.core.exceptions import ValidationError from chats.exceptions import ( - NonMatchingReplyChatIdException, WrongChatIdException, NonMatchingDirectChatIdException, ) -from chats.models import DirectChatMessage, ProjectChatMessage, FileToMessage -from files.models import UserFile User = get_user_model() @@ -21,12 +15,6 @@ async def orm_get(queryset, **kwargs): return await queryset.aget(**kwargs) -async def orm_create(manager, **kwargs): - if settings.RUNNING_TESTS: - return manager.create(**kwargs) - return await manager.acreate(**kwargs) - - async def orm_exists(queryset) -> bool: if settings.RUNNING_TESTS: return queryset.exists() @@ -40,78 +28,17 @@ async def orm_save(instance, update_fields=None): await instance.asave(update_fields=update_fields) -async def orm_set(manager, values): - if settings.RUNNING_TESTS: - manager.set(values) - return - await manager.aset(values) - - -def clean_message_text(text: str) -> str: - """ - Cleans message text. - - """ - - return text.strip() - - -def validate_message_text(text: str) -> bool: - """ - Validates message text. - - """ - # TODO: add bad word filter - return 0 < len(text) <= 8192 - - def get_user_channel_cache_key(user: User) -> str: return f"user_channel_{user.pk}" -async def create_message( - chat_id: Union[str, int], - chat_model: Union[Type[DirectChatMessage], Type[ProjectChatMessage]], - author: User, - text: str, - reply_to: int = None, -) -> Union[DirectChatMessage, ProjectChatMessage]: - """ - Creates message. - - Args: - chat_id: An integer representing chat id. - chat_model: A chat model. - author: A user instance. - text: A string representing message text. - reply_to: An integer representing message id to reply to. - - Returns: - A message instance. - - Raises: - NonMatchingReplyChatIdException: If reply_to message is not in chat. - """ - - try: - return await orm_create( - chat_model.objects, - chat_id=chat_id, - author=author, - text=text, - reply_to=reply_to, - ) - except ValidationError: - raise NonMatchingReplyChatIdException( - f"Message {reply_to} is not in chat {chat_id}" - ) - - async def get_chat_and_user_ids_from_content(content, current_user) -> tuple[str, User]: chat_id = content["chat_id"] # check if chat_id is in the format of _ try: user1_id, user2_id = map(int, chat_id.split("_")) - except ValueError: + except (AttributeError, ValueError): raise WrongChatIdException( f'Chat id "{chat_id}" is not in the format of' f" _, where user1_id < user2_id" @@ -119,10 +46,12 @@ async def get_chat_and_user_ids_from_content(content, current_user) -> tuple[str # check if user is a member of this chat and get other user if user1_id == current_user.id or user2_id == current_user.id: - other_user = await orm_get( - User.objects, - id=user1_id if user1_id != current_user.id else user2_id - ) + try: + other_user = await orm_get( + User.objects, id=user1_id if user1_id != current_user.id else user2_id + ) + except User.DoesNotExist as error: + raise WrongChatIdException("Собеседник не найден.") from error else: raise NonMatchingDirectChatIdException( f"User {current_user.id} is not a member of chat {chat_id}" @@ -130,28 +59,6 @@ async def get_chat_and_user_ids_from_content(content, current_user) -> tuple[str return chat_id, other_user -async def create_file_to_message( - direct_message: Union[str, None, DirectChatMessage], - project_message: Union[str, None, ProjectChatMessage], - file: str, -) -> FileToMessage: - return await orm_create( - FileToMessage.objects, - direct_message=direct_message, project_message=project_message, file=file - ) - - -async def match_files_and_messages(file_urls, messages): - for url in file_urls: - file = await orm_get(UserFile.objects, pk=url) - # implicitly matches a file and a message - await create_file_to_message( - direct_message=messages["direct_message"], - project_message=messages["project_message"], - file=file, - ) - - def get_all_files(messages): # looks like something bad - files = []