diff --git a/.env.example b/.env.example index 9a981e4d..619f054e 100644 --- a/.env.example +++ b/.env.example @@ -97,3 +97,7 @@ PAYROLL_SIMPLE_ADMIN_ACCESS=true # Optional Fernet key for encrypted commerce integration credentials. # Generate: python -c "from cryptography.fernet import Fernet; print(Fernet.generate_key().decode())" COMMERCE_CREDENTIAL_KEY= +# A queued import should normally be picked up within seconds. Celery Beat marks +# lost queue messages as failed after this timeout; running imports use a longer limit. +COMMERCE_QUEUED_STALE_SECONDS=120 +COMMERCE_SYNCING_STALE_MINUTES=45 diff --git a/backend/commerce/tasks/__init__.py b/backend/commerce/tasks/__init__.py index bafd9956..be7d0280 100644 --- a/backend/commerce/tasks/__init__.py +++ b/backend/commerce/tasks/__init__.py @@ -2,6 +2,8 @@ import logging from datetime import timedelta from celery import shared_task +from celery.utils import uuid as celery_uuid +from django.conf import settings from django.db import transaction from django.db.models import Q from django.utils import timezone @@ -16,6 +18,11 @@ from commerce.services.credentials import CredentialError logger = logging.getLogger(__name__) +QUEUE_PUBLISH_ERROR = "Не удалось поставить синхронизацию в очередь." +MISSING_RUN_ERROR = "Запуск синхронизации не найден. Повторите импорт." +STALE_QUEUED_ERROR = "Задача не была получена обработчиком и снята с очереди." +STALE_SYNCING_ERROR = "Предыдущая синхронизация была прервана." + def _next_sync_at(connection, *, now=None): if not connection.auto_sync: @@ -23,16 +30,96 @@ def _next_sync_at(connection, *, now=None): return (now or timezone.now()) + timedelta(minutes=connection.sync_interval_minutes) +def _safe_send_connection_update(connection, *, run=None): + """Realtime is best-effort and must never break the synchronization itself.""" + + try: + send_connection_update(connection, run=run) + except Exception: + logger.exception( + "Commerce realtime update failed", + extra={ + "event": "commerce.connection.realtime_update_failed", + "connection_id": str(connection.pk), + "run_id": str(run.pk) if run is not None else None, + }, + ) + + +def _mark_enqueue_failed(*, connection_id, run_id, task_id): + now = timezone.now() + with transaction.atomic(): + run = ( + IntegrationSyncRun.objects.select_for_update() + .filter(pk=run_id, task_id=task_id) + .first() + ) + connection = ( + IntegrationConnection.objects.select_for_update() + .filter(pk=connection_id, task_id=task_id) + .first() + ) + if run is not None and run.status == "queued": + run.status = "error" + run.error = QUEUE_PUBLISH_ERROR + run.finished_at = now + run.save(update_fields=["status", "error", "finished_at"]) + if connection is not None and connection.status == IntegrationStatus.QUEUED: + connection.status = IntegrationStatus.ERROR + connection.last_error = QUEUE_PUBLISH_ERROR + connection.next_sync_at = ( + now + timedelta(minutes=5) if connection.auto_sync else None + ) + connection.save( + update_fields=[ + "status", + "last_error", + "next_sync_at", + "updated_at", + ] + ) + if connection is not None: + _safe_send_connection_update(connection, run=run) + + +def _publish_connection_sync(*, connection_id, run_id, task_id): + connection = IntegrationConnection.objects.get(pk=connection_id) + run = IntegrationSyncRun.objects.get(pk=run_id) + _safe_send_connection_update(connection, run=run) + try: + sync_connection_run.apply_async( + args=[str(run_id), str(connection_id)], + task_id=task_id, + ) + except Exception: + _mark_enqueue_failed( + connection_id=connection_id, + run_id=run_id, + task_id=task_id, + ) + logger.exception( + "Commerce synchronization could not be published", + extra={ + "event": "commerce.connection.enqueue_failed", + "connection_id": str(connection_id), + "run_id": str(run_id), + "task_id": task_id, + }, + ) + raise + + def enqueue_connection_sync(connection): """Create and publish one run while preventing overlapping scans.""" + task_id = celery_uuid() with transaction.atomic(): locked = IntegrationConnection.objects.select_for_update().get(pk=connection.pk) if locked.status in {IntegrationStatus.QUEUED, IntegrationStatus.SYNCING}: return None - run = IntegrationSyncRun.objects.create(connection=locked) + run = IntegrationSyncRun.objects.create(connection=locked, task_id=task_id) locked.status = IntegrationStatus.QUEUED - locked.task_id = "" + locked.task_id = task_id locked.last_error = "" locked.next_sync_at = _next_sync_at(locked) locked.save( @@ -44,29 +131,13 @@ def enqueue_connection_sync(connection): "updated_at", ] ) - send_connection_update(locked, run=run) - try: - async_result = sync_moysklad_run.delay(str(run.pk)) - except Exception: - retry_at = timezone.now() + timedelta(minutes=5) - IntegrationSyncRun.objects.filter(pk=run.pk).update( - status="error", - error="Не удалось поставить синхронизацию в очередь.", - finished_at=timezone.now(), + transaction.on_commit( + lambda: _publish_connection_sync( + connection_id=locked.pk, + run_id=run.pk, + task_id=task_id, + ) ) - IntegrationConnection.objects.filter(pk=connection.pk).update( - status=IntegrationStatus.ERROR, - last_error="Не удалось поставить синхронизацию в очередь.", - next_sync_at=retry_at, - ) - connection.refresh_from_db() - run.refresh_from_db() - send_connection_update(connection, run=run) - raise - task_id = async_result.id or "" - IntegrationSyncRun.objects.filter(pk=run.pk).update(task_id=task_id) - IntegrationConnection.objects.filter(pk=connection.pk).update(task_id=task_id) - run.task_id = task_id return run @@ -81,8 +152,43 @@ def sync_source(source_code: str, entity_type: str): @shared_task(bind=True, name="commerce.sync_connection_run") -def sync_connection_run(self, run_id: str): - run = IntegrationSyncRun.objects.select_related("connection__source").get(pk=run_id) +def sync_connection_run(self, run_id: str, connection_id: str | None = None): + try: + run = IntegrationSyncRun.objects.select_related("connection__source").get( + pk=run_id + ) + except IntegrationSyncRun.DoesNotExist: + task_id = self.request.id or "" + connection = None + if connection_id is not None: + connection = IntegrationConnection.objects.filter(pk=connection_id).first() + if connection is not None and (not task_id or connection.task_id == task_id): + finished_at = timezone.now() + connection.status = IntegrationStatus.ERROR + connection.last_error = MISSING_RUN_ERROR + connection.next_sync_at = ( + finished_at + timedelta(minutes=5) if connection.auto_sync else None + ) + connection.save( + update_fields=[ + "status", + "last_error", + "next_sync_at", + "updated_at", + ] + ) + _safe_send_connection_update(connection) + logger.error( + "Commerce synchronization run does not exist", + extra={ + "event": "commerce.connection.run_missing", + "connection_id": connection_id, + "run_id": run_id, + "task_id": task_id, + }, + ) + return {"status": "missing", "run_id": run_id} + connection = run.connection started_at = timezone.now() run.status = "running" @@ -96,7 +202,7 @@ def sync_connection_run(self, run_id: str): connection.save( update_fields=["status", "last_sync_started_at", "last_error", "updated_at"] ) - send_connection_update(connection, run=run) + _safe_send_connection_update(connection, run=run) def report_progress(current_stats): now = timezone.now() @@ -113,7 +219,7 @@ def sync_connection_run(self, run_id: str): IntegrationConnection.objects.filter(pk=connection.pk).update(updated_at=now) run.refresh_from_db() connection.refresh_from_db() - send_connection_update(connection, run=run) + _safe_send_connection_update(connection, run=run) try: if connection.provider == "wildberries": @@ -156,7 +262,7 @@ def sync_connection_run(self, run_id: str): "updated_at", ] ) - send_connection_update(connection, run=run) + _safe_send_connection_update(connection, run=run) raise finished_at = timezone.now() @@ -199,7 +305,7 @@ def sync_connection_run(self, run_id: str): "updated_at", ] ) - send_connection_update(connection, run=run) + _safe_send_connection_update(connection, run=run) return stats @@ -208,20 +314,76 @@ def sync_connection_run(self, run_id: str): sync_moysklad_run = sync_connection_run +def _recover_stale_connections(*, now): + queued_cutoff = now - timedelta( + seconds=getattr(settings, "COMMERCE_QUEUED_STALE_SECONDS", 120) + ) + syncing_cutoff = now - timedelta( + minutes=getattr(settings, "COMMERCE_SYNCING_STALE_MINUTES", 45) + ) + recovered = 0 + recovery_rules = ( + (IntegrationStatus.QUEUED, queued_cutoff, ("queued",), STALE_QUEUED_ERROR), + ( + IntegrationStatus.SYNCING, + syncing_cutoff, + ("queued", "running"), + STALE_SYNCING_ERROR, + ), + ) + for connection_status, cutoff, run_statuses, message in recovery_rules: + candidate_ids = list( + IntegrationConnection.objects.filter( + status=connection_status, + updated_at__lte=cutoff, + ).values_list("pk", flat=True) + ) + for connection_id in candidate_ids: + with transaction.atomic(): + connection = IntegrationConnection.objects.select_for_update().get( + pk=connection_id + ) + if connection.status != connection_status or connection.updated_at > cutoff: + continue + run = ( + connection.sync_runs.select_for_update() + .filter(status__in=run_statuses) + .order_by("-created_at") + .first() + ) + if run is not None: + run.status = "error" + run.error = message + run.finished_at = now + run.save(update_fields=["status", "error", "finished_at"]) + connection.status = IntegrationStatus.ERROR + connection.last_error = message + connection.next_sync_at = now if connection.auto_sync else None + connection.save( + update_fields=[ + "status", + "last_error", + "next_sync_at", + "updated_at", + ] + ) + _safe_send_connection_update(connection, run=run) + recovered += 1 + if recovered: + logger.warning( + "Recovered stale commerce synchronizations", + extra={ + "event": "commerce.connection.stale_runs_recovered", + "recovered": recovered, + }, + ) + return recovered + + @shared_task(name="commerce.dispatch_due_connection_syncs") def dispatch_due_connection_syncs(): now = timezone.now() - stale_cutoff = now - timedelta(minutes=45) - IntegrationConnection.objects.filter( - auto_sync=True, - status__in=[IntegrationStatus.QUEUED, IntegrationStatus.SYNCING], - updated_at__lte=stale_cutoff, - ).update( - status=IntegrationStatus.ERROR, - last_error="Предыдущая синхронизация была прервана и будет перезапущена.", - next_sync_at=now, - updated_at=now, - ) + _recover_stale_connections(now=now) due_ids = list( IntegrationConnection.objects.filter( auto_sync=True, diff --git a/backend/communications/consumers.py b/backend/communications/consumers.py index facb39cc..34a5c79a 100644 --- a/backend/communications/consumers.py +++ b/backend/communications/consumers.py @@ -34,6 +34,7 @@ from .mentions import ( update_unread_mention_counters, ) 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 @@ -84,6 +85,22 @@ class ChatConsumerMixin: # ==================== Управление активным чатом ==================== + @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): """Открытие чата. @@ -105,6 +122,7 @@ class ChatConsumerMixin: # Устанавливаем активный чат self.active_chat_id = chat_id + await self._set_active_chat_presence(chat_id) # Отправляем начальную историю (если запрошена) if content.get("load_history", False): @@ -150,6 +168,7 @@ class ChatConsumerMixin: 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}) diff --git a/backend/communications/migrations/0043_activechatsession.py b/backend/communications/migrations/0043_activechatsession.py new file mode 100644 index 00000000..1f3b27eb --- /dev/null +++ b/backend/communications/migrations/0043_activechatsession.py @@ -0,0 +1,53 @@ +from django.conf import settings +from django.db import migrations, models +import django.db.models.deletion + + +class Migration(migrations.Migration): + dependencies = [ + ("communications", "0042_chatusersettings_notification_preferences"), + migrations.swappable_dependency(settings.AUTH_USER_MODEL), + ] + + operations = [ + migrations.CreateModel( + name="ActiveChatSession", + fields=[ + ( + "id", + models.BigAutoField( + auto_created=True, + primary_key=True, + serialize=False, + verbose_name="ID", + ), + ), + ("connection_id", models.CharField(max_length=255, unique=True)), + ("last_seen_at", models.DateTimeField(auto_now=True)), + ("expires_at", models.DateTimeField(db_index=True)), + ( + "chat", + models.ForeignKey( + on_delete=django.db.models.deletion.CASCADE, + related_name="active_sessions", + to="communications.chat", + ), + ), + ( + "user", + models.ForeignKey( + on_delete=django.db.models.deletion.CASCADE, + related_name="active_chat_sessions", + to=settings.AUTH_USER_MODEL, + ), + ), + ], + ), + migrations.AddIndex( + model_name="activechatsession", + index=models.Index( + fields=["user", "chat", "expires_at"], + name="communicati_user_id_412dae_idx", + ), + ), + ] diff --git a/backend/communications/models.py b/backend/communications/models.py index eb1f65ba..db8d8795 100644 --- a/backend/communications/models.py +++ b/backend/communications/models.py @@ -89,6 +89,30 @@ class ChatReadState(models.Model): return self.updated_at +class ActiveChatSession(models.Model): + """Cross-worker presence for a user currently viewing a chat.""" + + connection_id = models.CharField(max_length=255, unique=True) + user = models.ForeignKey( + settings.AUTH_USER_MODEL, + on_delete=models.CASCADE, + related_name="active_chat_sessions", + ) + chat = models.ForeignKey( + "Chat", + on_delete=models.CASCADE, + related_name="active_sessions", + ) + last_seen_at = models.DateTimeField(auto_now=True) + expires_at = models.DateTimeField(db_index=True) + + class Meta: + indexes = [models.Index(fields=["user", "chat", "expires_at"])] + + def __str__(self): + return f"{self.user_id} @ {self.chat_id} ({self.connection_id})" + + class Chat(models.Model): CHAT_TYPE_CHOICES = [ ("private", "Личный"), diff --git a/backend/communications/presence.py b/backend/communications/presence.py new file mode 100644 index 00000000..de574edb --- /dev/null +++ b/backend/communications/presence.py @@ -0,0 +1,42 @@ +from datetime import timedelta + +from django.utils import timezone + +from communications.models import ActiveChatSession + + +ACTIVE_CHAT_SESSION_TTL = timedelta(seconds=75) + + +def set_active_chat(*, user_id: int, chat_id: int, connection_id: str) -> None: + now = timezone.now() + ActiveChatSession.objects.update_or_create( + connection_id=connection_id, + defaults={ + "user_id": user_id, + "chat_id": chat_id, + "expires_at": now + ACTIVE_CHAT_SESSION_TTL, + }, + ) + + +def touch_active_chat(*, connection_id: str) -> None: + now = timezone.now() + ActiveChatSession.objects.filter(connection_id=connection_id).update( + last_seen_at=now, + expires_at=now + ACTIVE_CHAT_SESSION_TTL, + ) + + +def clear_active_chat(*, connection_id: str) -> None: + ActiveChatSession.objects.filter(connection_id=connection_id).delete() + + +def user_is_viewing_chat(*, user_id: int, chat_id: int) -> bool: + now = timezone.now() + queryset = ActiveChatSession.objects.filter( + user_id=user_id, + chat_id=chat_id, + ) + queryset.filter(expires_at__lte=now).delete() + return queryset.filter(expires_at__gt=now).exists() diff --git a/backend/communications/services/chat_notifications.py b/backend/communications/services/chat_notifications.py index e46273d9..0c0d542d 100644 --- a/backend/communications/services/chat_notifications.py +++ b/backend/communications/services/chat_notifications.py @@ -48,7 +48,11 @@ class ChatNotificationReadService: def notify_after_commit(): invalidate_unread_summary(user.id) send_notifications_read_all_event(user.id, notification_ids) - dismiss_chat_push(user_id=user.id, chat_id=chat_id) + dismiss_chat_push( + user_id=user.id, + chat_id=chat_id, + notification_ids=notification_ids, + ) transaction.on_commit(notify_after_commit) diff --git a/backend/eusrr_backend/settings.py b/backend/eusrr_backend/settings.py index 82b4be81..40b4d01d 100644 --- a/backend/eusrr_backend/settings.py +++ b/backend/eusrr_backend/settings.py @@ -754,6 +754,12 @@ CELERY_TASK_SOFT_TIME_LIMIT = 25 * 60 # Мягкий лимит - 25 минут # Integration credentials are encrypted at rest. Set an explicit Fernet key in # production; when omitted, a stable key is derived from DJANGO_SECRET_KEY. COMMERCE_CREDENTIAL_KEY = os.getenv("COMMERCE_CREDENTIAL_KEY", "") +COMMERCE_QUEUED_STALE_SECONDS = int( + os.getenv("COMMERCE_QUEUED_STALE_SECONDS", "120") +) +COMMERCE_SYNCING_STALE_MINUTES = int( + os.getenv("COMMERCE_SYNCING_STALE_MINUTES", "45") +) MOYSKLAD_API_BASE_URL = os.getenv( "MOYSKLAD_API_BASE_URL", "https://api.moysklad.ru/api/remap/1.2" ) diff --git a/backend/notifications/chat_suppression.py b/backend/notifications/chat_suppression.py new file mode 100644 index 00000000..fc8fdba8 --- /dev/null +++ b/backend/notifications/chat_suppression.py @@ -0,0 +1,37 @@ +"""Suppress chat alerts when the recipient is already viewing that chat.""" + +from communications.presence import user_is_viewing_chat +from communications.services import ChatNotificationReadService + + +CHAT_NOTIFICATION_VERBS = { + "chat_new_message", + "chat_mention", + "chat_reply", + "announcement_new_message", +} + + +def notification_chat_id(notification) -> int | None: + if notification.verb not in CHAT_NOTIFICATION_VERBS: + return None + data = notification.data if isinstance(notification.data, dict) else {} + try: + return int(data.get("chat_id")) + except (TypeError, ValueError): + return None + + +def suppress_for_active_chat(notification) -> bool: + chat_id = notification_chat_id(notification) + if chat_id is None or not user_is_viewing_chat( + user_id=notification.recipient_id, + chat_id=chat_id, + ): + return False + + ChatNotificationReadService().mark_chat_notifications_as_read( + user=notification.recipient, + chat_id=chat_id, + ) + return True diff --git a/backend/notifications/push_lifecycle.py b/backend/notifications/push_lifecycle.py index 833c322d..a460d618 100644 --- a/backend/notifications/push_lifecycle.py +++ b/backend/notifications/push_lifecycle.py @@ -10,6 +10,9 @@ from notifications.models import Notification from notifications.realtime import send_notifications_deleted_event +PUSH_CONTROL_ID_BATCH_SIZE = 150 + + def queue_push_control( *, user_id: int, @@ -30,23 +33,36 @@ def queue_push_control( transaction.on_commit(enqueue) -def dismiss_chat_push(*, user_id: int, chat_id: int) -> None: +def dismiss_chat_push( + *, + user_id: int, + chat_id: int, + notification_ids=None, +) -> None: + scope = {"chat_id": chat_id} + if notification_ids: + scope["notification_ids"] = [ + int(item) for item in list(notification_ids)[:PUSH_CONTROL_ID_BATCH_SIZE] + ] queue_push_control( user_id=user_id, action="dismiss_chat", - scope={"chat_id": chat_id}, + scope=scope, ) def dismiss_notifications_push(*, user_id: int, notification_ids) -> None: normalized_ids = [int(notification_id) for notification_id in notification_ids] - if not normalized_ids: - return - queue_push_control( - user_id=user_id, - action="dismiss_notifications", - scope={"notification_ids": normalized_ids}, - ) + for start in range(0, len(normalized_ids), PUSH_CONTROL_ID_BATCH_SIZE): + queue_push_control( + user_id=user_id, + action="dismiss_notifications", + scope={ + "notification_ids": normalized_ids[ + start : start + PUSH_CONTROL_ID_BATCH_SIZE + ] + }, + ) def delete_message_notifications(*, chat_id: int, message_id: int) -> None: diff --git a/backend/notifications/senders/push.py b/backend/notifications/senders/push.py index 22922ed7..898c943a 100644 --- a/backend/notifications/senders/push.py +++ b/backend/notifications/senders/push.py @@ -13,6 +13,7 @@ from .base import BaseNotificationSender from ..config import get from ..models import NotificationDeliveryRouting, NotificationInstallation from ..routing import APP_MAIN, APP_MESSENGER, normalize_routes, notification_category +from notifications.chat_suppression import suppress_for_active_chat class PushNotificationSender(BaseNotificationSender): @@ -169,10 +170,9 @@ class PushNotificationSender(BaseNotificationSender): return any(marker in error_text for marker in permanent_markers) def _deactivate_device(self, device, reason: str) -> None: - if not device.active: - return - device.active = False - device.save(update_fields=["active"]) + if device.active: + device.active = False + device.save(update_fields=["active"]) NotificationInstallation.objects.filter( push_device_id=device.id, ).update(endpoint=None, push_device_id=None) @@ -185,6 +185,12 @@ class PushNotificationSender(BaseNotificationSender): }, ) + @staticmethod + def _push_result_succeeded(result) -> bool: + if not isinstance(result, dict): + return True + return bool(result.get("success", 0)) and not result.get("failure", 0) + def _delivery_devices(self, notification): user = notification.recipient active_devices = list(WebPushDevice.objects.filter(user=user, active=True)) @@ -271,6 +277,9 @@ class PushNotificationSender(BaseNotificationSender): if notification.deleted or not notification.unread: self.log_skip(notification, "notification no longer active") return False + if suppress_for_active_chat(notification): + self.log_skip(notification, "recipient is viewing chat") + return False chat_preferences = self._chat_preferences(notification) if ( chat_preferences is not None @@ -298,7 +307,21 @@ class PushNotificationSender(BaseNotificationSender): def send_to_device(device): nonlocal sent_count, failed_count try: - device.send_message(message) + result = device.send_message(message) + if not self._push_result_succeeded(result): + failed_count += 1 + device.refresh_from_db(fields=["active"]) + if not device.active: + self._deactivate_device(device, "provider rejected subscription") + self.logger.warning( + "Push provider rejected delivery", + extra={ + "event": "notification.push.provider_rejected", + "device_id": device.id, + "notification_id": notification.id, + }, + ) + return False sent_count += 1 sent_device_ids.add(device.id) self.logger.debug( diff --git a/backend/notifications/senders/push_control.py b/backend/notifications/senders/push_control.py index abc386d7..04f9c3d8 100644 --- a/backend/notifications/senders/push_control.py +++ b/backend/notifications/senders/push_control.py @@ -44,7 +44,24 @@ class PushControlSender: push_sender = PushNotificationSender() for device in WebPushDevice.objects.filter(user_id=user_id, active=True): try: - device.send_message(message) + result = device.send_message(message) + if not push_sender._push_result_succeeded(result): + device.refresh_from_db(fields=["active"]) + if not device.active: + push_sender._deactivate_device( + device, + "provider rejected control command", + ) + logger.warning( + "Push provider rejected control command", + extra={ + "event": "notification.push.control_rejected", + "user_id": user_id, + "device_id": device.id, + "action": action, + }, + ) + continue delivered = True except Exception as error: # noqa: BLE001 - push clients raise vendor exceptions logger.warning( diff --git a/backend/notifications/senders/websocket.py b/backend/notifications/senders/websocket.py index b35dfba7..4a02b4ed 100644 --- a/backend/notifications/senders/websocket.py +++ b/backend/notifications/senders/websocket.py @@ -5,6 +5,7 @@ from channels.layers import get_channel_layer from asgiref.sync import async_to_sync from .base import BaseNotificationSender +from notifications.chat_suppression import suppress_for_active_chat class WebSocketNotificationSender(BaseNotificationSender): @@ -33,6 +34,9 @@ class WebSocketNotificationSender(BaseNotificationSender): True если отправлено успешно, False иначе """ try: + if suppress_for_active_chat(notification): + self.log_skip(notification, "recipient is viewing chat") + return False channel_layer = get_channel_layer() if not channel_layer: self.log_skip(notification, "channel_layer not configured") diff --git a/backend/realtime/consumers.py b/backend/realtime/consumers.py index 11b1eed0..40a0b0eb 100644 --- a/backend/realtime/consumers.py +++ b/backend/realtime/consumers.py @@ -198,6 +198,8 @@ class UserConsumer(ChatConsumerMixin, AsyncJsonWebsocketConsumer): chat = await self._get_chat(self.active_chat_id) if chat: await self._set_typing_status(chat, self.user, False) + if self.user: + await self._clear_active_chat_presence() # Отписываемся от всех групп чатов for chat_id in self.subscribed_chats: @@ -351,6 +353,8 @@ class UserConsumer(ChatConsumerMixin, AsyncJsonWebsocketConsumer): 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): diff --git a/backend/tests/api/v1/notifications/test_push_lifecycle.py b/backend/tests/api/v1/notifications/test_push_lifecycle.py index 6d803565..20f60198 100644 --- a/backend/tests/api/v1/notifications/test_push_lifecycle.py +++ b/backend/tests/api/v1/notifications/test_push_lifecycle.py @@ -6,12 +6,16 @@ import pytest from django.utils import timezone from communications.models import Chat, ChatUserSettings +from communications.presence import set_active_chat, user_is_viewing_chat from communications.services.chat_notifications import ChatNotificationReadService from notifications.models import Notification, UserChannelPreferences from notifications.push_lifecycle import ( delete_message_notifications, + dismiss_notifications_push, replace_message_notifications, ) +from notifications.senders.push import PushNotificationSender +from notifications.senders.websocket import WebSocketNotificationSender pytestmark = pytest.mark.django_db(transaction=True) @@ -93,7 +97,11 @@ def test_reading_chat_schedules_native_notification_cleanup(user_factory): notification.refresh_from_db() assert result.notification_ids == [notification.id] assert notification.unread is False - dismiss.assert_called_once_with(user_id=recipient.id, chat_id=chat.id) + dismiss.assert_called_once_with( + user_id=recipient.id, + chat_id=chat.id, + notification_ids=[notification.id], + ) def test_chat_policy_supports_mentions_full_mute_and_temporary_mute(user_factory): @@ -119,3 +127,66 @@ def test_chat_policy_supports_mentions_full_mute_and_temporary_mute(user_factory settings.notifications_enabled = False settings.muted_until = timezone.now() - timedelta(seconds=1) assert settings.allows_notification("chat_new_message") is True + + +def test_active_chat_on_another_connection_suppresses_all_realtime_alerts(user_factory): + recipient = user_factory(email="presence-recipient@example.com") + chat = Chat.objects.create(type="private", created_by=recipient) + set_active_chat( + user_id=recipient.id, + chat_id=chat.id, + connection_id="specific.channel!device-one", + ) + notification = _notification( + recipient, + chat_id=chat.id, + message_id=44, + ) + + with ( + patch("push_notifications.models.WebPushDevice.send_message") as push_send, + patch("notifications.senders.websocket.async_to_sync") as websocket_send, + patch("notifications.tasks.send_push_control.delay"), + ): + assert PushNotificationSender().send(notification) is False + assert WebSocketNotificationSender().send(notification) is False + + notification.refresh_from_db() + assert notification.unread is False + push_send.assert_not_called() + websocket_send.assert_not_called() + + +def test_expired_active_chat_presence_is_not_considered_online(user_factory): + recipient = user_factory(email="expired-presence@example.com") + chat = Chat.objects.create(type="private", created_by=recipient) + set_active_chat( + user_id=recipient.id, + chat_id=chat.id, + connection_id="specific.channel!expired", + ) + from communications.models import ActiveChatSession + + ActiveChatSession.objects.filter(connection_id="specific.channel!expired").update( + expires_at=timezone.now() - timedelta(seconds=1), + ) + + assert user_is_viewing_chat(user_id=recipient.id, chat_id=chat.id) is False + assert ActiveChatSession.objects.filter( + connection_id="specific.channel!expired" + ).exists() is False + + +def test_large_cleanup_is_split_into_web_push_sized_batches(): + with patch("notifications.tasks.send_push_control.delay") as delay: + dismiss_notifications_push( + user_id=91, + notification_ids=range(1, 352), + ) + + assert delay.call_count == 3 + assert [len(call.args[2]["notification_ids"]) for call in delay.call_args_list] == [ + 150, + 150, + 51, + ] diff --git a/backend/tests/commerce/test_connections.py b/backend/tests/commerce/test_connections.py index 62f42581..2987318a 100644 --- a/backend/tests/commerce/test_connections.py +++ b/backend/tests/commerce/test_connections.py @@ -1,11 +1,13 @@ from datetime import timedelta from decimal import Decimal import io +import uuid from unittest.mock import Mock, patch import httpx import pytest from PIL import Image +from django.db import transaction from django.urls import reverse from django.utils import timezone from rest_framework import status @@ -13,6 +15,7 @@ from rest_framework import status from commerce.models import ( IntegrationConnection, IntegrationStatus, + IntegrationSyncRun, Product, ProductBinding, ProductImage, @@ -40,7 +43,12 @@ from commerce.services.moysklad_sync import ( normalize_moysklad_stock_row, reconcile_moysklad_variant_parents, ) -from commerce.tasks import dispatch_due_connection_syncs, sync_moysklad_run +from commerce.tasks import ( + STALE_QUEUED_ERROR, + dispatch_due_connection_syncs, + enqueue_connection_sync, + sync_moysklad_run, +) pytestmark = pytest.mark.django_db @@ -385,7 +393,11 @@ def test_connection_test_action_updates_account(api_client, user_factory): assert response.data["account_name"] == "Иван Иванов" -def test_connection_sync_action_queues_background_run(api_client, user_factory): +def test_connection_sync_action_queues_background_run( + api_client, + user_factory, + django_capture_on_commit_callbacks, +): user = user_factory(staff=True) api_client.force_authenticate(user=user) create_response = api_client.post( @@ -393,9 +405,11 @@ def test_connection_sync_action_queues_background_run(api_client, user_factory): {"provider": "moysklad", "auth_type": "token", "token": "secret-token"}, format="json", ) - async_result = Mock(id="celery-task-1") - - with patch("commerce.tasks.sync_moysklad_run.delay", return_value=async_result): + with ( + patch("commerce.tasks.celery_uuid", return_value="celery-task-1"), + patch("commerce.tasks.sync_connection_run.apply_async") as apply_async, + django_capture_on_commit_callbacks(execute=True), + ): response = api_client.post( reverse( "api:v1:commerce:connection-sync", @@ -406,8 +420,113 @@ def test_connection_sync_action_queues_background_run(api_client, user_factory): assert response.status_code == status.HTTP_202_ACCEPTED assert response.data["status"] == "queued" connection = IntegrationConnection.objects.get() + run = IntegrationSyncRun.objects.get() assert connection.status == IntegrationStatus.QUEUED assert connection.task_id == "celery-task-1" + assert run.task_id == "celery-task-1" + apply_async.assert_called_once_with( + args=[str(run.pk), str(connection.pk)], + task_id="celery-task-1", + ) + + +@pytest.mark.django_db(transaction=True) +def test_connection_sync_is_published_only_after_outer_transaction_commit(): + source = SourceSystem.objects.create(code="moysklad", name="МойСклад") + connection = IntegrationConnection.objects.create( + source=source, + provider="moysklad", + encrypted_credentials="unused", + status=IntegrationStatus.READY, + ) + publication_checks = [] + + def verify_committed(*, args, task_id): + publication_checks.append( + { + "in_atomic_block": transaction.get_connection().in_atomic_block, + "run_exists": IntegrationSyncRun.objects.filter( + pk=args[0], task_id=task_id + ).exists(), + "connection_has_task": IntegrationConnection.objects.filter( + pk=args[1], task_id=task_id + ).exists(), + } + ) + + with ( + patch("commerce.tasks.celery_uuid", return_value="commit-safe-task"), + patch( + "commerce.tasks.sync_connection_run.apply_async", + side_effect=verify_committed, + ) as apply_async, + ): + with transaction.atomic(): + run = enqueue_connection_sync(connection) + assert run.task_id == "commit-safe-task" + assert apply_async.call_count == 0 + + assert apply_async.call_count == 1 + + assert publication_checks == [ + { + "in_atomic_block": False, + "run_exists": True, + "connection_has_task": True, + } + ] + + +def test_enqueue_failure_transitions_run_and_connection_to_error( + django_capture_on_commit_callbacks, +): + source = SourceSystem.objects.create(code="moysklad", name="МойСклад") + connection = IntegrationConnection.objects.create( + source=source, + provider="moysklad", + encrypted_credentials="unused", + status=IntegrationStatus.READY, + ) + + with ( + patch("commerce.tasks.celery_uuid", return_value="failed-task"), + patch( + "commerce.tasks.sync_connection_run.apply_async", + side_effect=RuntimeError("broker unavailable"), + ), + patch("commerce.tasks.send_connection_update"), + pytest.raises(RuntimeError, match="broker unavailable"), + django_capture_on_commit_callbacks(execute=True), + ): + enqueue_connection_sync(connection) + + connection.refresh_from_db() + run = IntegrationSyncRun.objects.get() + assert connection.status == IntegrationStatus.ERROR + assert run.status == "error" + assert run.finished_at is not None + + +def test_missing_sync_run_does_not_leave_connection_queued(): + source = SourceSystem.objects.create(code="moysklad", name="МойСклад") + connection = IntegrationConnection.objects.create( + source=source, + provider="moysklad", + encrypted_credentials="unused", + status=IntegrationStatus.QUEUED, + task_id="missing-run-task", + ) + + with patch("commerce.tasks.send_connection_update"): + result = sync_moysklad_run.apply( + args=[str(uuid.uuid4()), str(connection.pk)], + task_id="missing-run-task", + ).get() + + connection.refresh_from_db() + assert result["status"] == "missing" + assert connection.status == IntegrationStatus.ERROR + assert connection.last_error def test_sync_run_persists_product_image_progress(): @@ -447,7 +566,7 @@ def test_sync_run_persists_product_image_progress(): assert connection.last_stats["images"] == 12 -def test_beat_dispatcher_queues_due_connection(): +def test_beat_dispatcher_queues_due_connection(django_capture_on_commit_callbacks): source = SourceSystem.objects.create(code="moysklad", name="МойСклад") connection = IntegrationConnection.objects.create( source=source, @@ -458,9 +577,11 @@ def test_beat_dispatcher_queues_due_connection(): sync_interval_minutes=15, next_sync_at=timezone.now() - timedelta(minutes=1), ) - async_result = Mock(id="scheduled-task-1") - - with patch("commerce.tasks.sync_moysklad_run.delay", return_value=async_result): + with ( + patch("commerce.tasks.celery_uuid", return_value="scheduled-task-1"), + patch("commerce.tasks.sync_connection_run.apply_async") as apply_async, + django_capture_on_commit_callbacks(execute=True), + ): result = dispatch_due_connection_syncs() connection.refresh_from_db() @@ -468,6 +589,35 @@ def test_beat_dispatcher_queues_due_connection(): assert connection.status == IntegrationStatus.QUEUED assert connection.task_id == "scheduled-task-1" assert connection.next_sync_at > timezone.now() + apply_async.assert_called_once() + + +def test_dispatcher_recovers_old_queued_run_before_syncing_timeout(settings): + settings.COMMERCE_QUEUED_STALE_SECONDS = 60 + source = SourceSystem.objects.create(code="moysklad", name="МойСклад") + connection = IntegrationConnection.objects.create( + source=source, + provider="moysklad", + encrypted_credentials="unused", + status=IntegrationStatus.QUEUED, + task_id="lost-task", + auto_sync=False, + ) + run = connection.sync_runs.create(status="queued", task_id="lost-task") + stale_at = timezone.now() - timedelta(minutes=2) + IntegrationConnection.objects.filter(pk=connection.pk).update(updated_at=stale_at) + + with patch("commerce.tasks.send_connection_update"): + result = dispatch_due_connection_syncs() + + connection.refresh_from_db() + run.refresh_from_db() + assert result == {"queued": 0} + assert connection.status == IntegrationStatus.ERROR + assert connection.last_error == STALE_QUEUED_ERROR + assert run.status == "error" + assert run.error == STALE_QUEUED_ERROR + assert run.finished_at is not None def test_schedule_update_does_not_disconnect_connection(api_client, user_factory): diff --git a/backend/tests/realtime/test_user_consumer.py b/backend/tests/realtime/test_user_consumer.py index b9b289f4..e84b51ba 100644 --- a/backend/tests/realtime/test_user_consumer.py +++ b/backend/tests/realtime/test_user_consumer.py @@ -14,7 +14,7 @@ from channels.layers import get_channel_layer from channels.db import database_sync_to_async from django.contrib.auth.models import AnonymousUser -from communications.models import Chat, ChatMembership +from communications.models import ActiveChatSession, Chat, ChatMembership from commerce.realtime import COMMERCE_CONNECTIONS_GROUP from notifications.models import Notification @@ -160,8 +160,22 @@ class TestUserConsumerChatManagement: assert response["type"] == "chat_opened" assert response["chat_id"] == test_chat.id + presence_exists = await database_sync_to_async( + ActiveChatSession.objects.filter( + user=user, + chat=test_chat, + ).exists + )() + assert presence_exists is True await communicator.disconnect() + presence_exists = await database_sync_to_async( + ActiveChatSession.objects.filter( + user=user, + chat=test_chat, + ).exists + )() + assert presence_exists is False async def test_open_chat_marks_related_notifications_as_read(self, ws_communicator, user, test_chat): """Открытие конкретного чата читает только связанные с ним уведомления.""" @@ -336,6 +350,13 @@ class TestUserConsumerChatManagement: assert response["type"] == "chat_closed" assert response["chat_id"] == test_chat.id + presence_exists = await database_sync_to_async( + ActiveChatSession.objects.filter( + user=user, + chat=test_chat, + ).exists + )() + assert presence_exists is False await communicator.disconnect() diff --git a/docs/guides/NOTIFICATIONS.md b/docs/guides/NOTIFICATIONS.md index f9fd8267..47b647a9 100644 --- a/docs/guides/NOTIFICATIONS.md +++ b/docs/guides/NOTIFICATIONS.md @@ -129,6 +129,8 @@ Push работает только в secure context, кроме localhost. Perm Уведомления сообщений используют стабильный `tag=chat-{chat_id}` и несут `chat_id`, `message_id`, `notification_id`. Backend отправляет невидимые control push-команды, а `frontend/public/sw.js` применяет их через `getNotifications()`: прочтение закрывает уведомление на остальных устройствах, удаление сообщения убирает относящийся к нему push, редактирование заменяет только уже показанный push. Если исходная Celery-задача задержалась, sender повторно проверяет `unread/deleted` и не показывает устаревшее уведомление. +Активный чат хранится как `ActiveChatSession` с TTL 75 секунд и продлевается WebSocket heartbeat. Поэтому presence видна всем backend/Celery workers: если чат открыт хотя бы на одном устройстве пользователя, WebSocket-уведомление и Web Push для этого чата подавляются до доставки на остальные устройства, а запись сразу отмечается прочитанной. После добавления или изменения Celery tasks worker требуется перезапустить — Django autoreload задачи Celery не обновляет. + ## Email `instant` ставит отдельную Celery task при создании уведомления. `daily` и `weekly` обрабатываются Celery Beat в 09:00 по timezone проекта. `never` отключает email. diff --git a/docs/guides/WEBSOCKET.md b/docs/guides/WEBSOCKET.md index 2dd8cd51..4792f9bc 100644 --- a/docs/guides/WEBSOCKET.md +++ b/docs/guides/WEBSOCKET.md @@ -62,7 +62,7 @@ ASGI проверяет `Origin` через `AllowedHostsOriginValidator`. Ес Изменение `ChatMembership` синхронизируется с уже открытыми соединениями: при добавлении consumer входит в группу, при отзыве доступа немедленно выходит из неё и закрывает активный чат. -`active_chat_id` определяет, получает соединение полное chat event или только обновление списка. При открытии чата связанные message notifications отмечаются прочитанными; уведомление об уже открытом чате подавляется. +`active_chat_id` определяет, получает соединение полное chat event или только обновление списка. При открытии чата связанные message notifications отмечаются прочитанными. Состояние дополнительно публикуется в `ActiveChatSession` и продлевается heartbeat, поэтому WebSocket/Push-уведомление об уже открытом чате подавляется для всех устройств пользователя, а не только для текущего соединения. ## Команды клиента diff --git a/frontend/public/sw.js b/frontend/public/sw.js index 5b3a5d91..f1630d6b 100644 --- a/frontend/public/sw.js +++ b/frontend/public/sw.js @@ -6,7 +6,7 @@ * нативные уведомления даже когда страница закрыта. */ -const SW_VERSION = '2.2.0'; +const SW_VERSION = '2.2.1'; const SW_APP_KIND = self.CORP_SW_APP_KIND || 'main'; const STATIC_CACHE = `eusrr-static-${SW_VERSION}`; const PRECACHE_URLS = [ @@ -61,9 +61,10 @@ function sameIdentity(actual, expected) { function matchesNotificationScope(notification, scope = {}) { const data = notification.data || {}; if (Array.isArray(scope.notification_ids)) { - return scope.notification_ids.some( + const matchesId = scope.notification_ids.some( (id) => sameIdentity(data.notification_id, id), ); + if (matchesId) return true; } if ( scope.chat_id === undefined &&