""" Обработчики каналов доставки уведомлений Слушают post_save сигнал от Notification и асинхронно отправляют через Celery: - WebSocket (realtime) - мгновенно - Email - с retry и rate limiting - Web Push - с автоматическим удалением неактивных устройств TODO: Добавить integration тесты: - Проверка вызова Celery tasks при создании Notification - Режим DND (только WebSocket silent) - Отключенные каналы (email_enabled=False) - Отключенные verb'ы (is_verb_enabled) """ import logging from django.db import transaction from django.db.models.signals import post_save from django.dispatch import receiver logger = logging.getLogger(__name__) def _enqueue_notification_task(task, *args, **kwargs) -> bool: try: task.delay(*args, **kwargs) return True except Exception as exc: logger.exception( "Notification task enqueue failed", extra={ "event": "notification.task.enqueue_failed", "task_name": getattr(task, "name", type(task).__name__), "error_type": type(exc).__name__, }, ) return False @receiver(post_save, sender="notifications.Notification") def route_notification_to_channels(sender, instance, created, **kwargs): """ Главный роутер уведомлений по каналам. Вызывается при создании нового уведомления. Асинхронно отправляет в Celery очередь для каждого канала. ВАЖНО: Использует transaction.on_commit() чтобы гарантировать, что уведомление сохранено в БД перед отправкой в Celery. """ if not created: return # Обрабатываем только новые уведомления notification = instance user = notification.recipient # Проверяем что получатель указан if not user: logger.warning( "Notification routing skipped because recipient is missing", extra={ "event": "notification.routing.skipped", "notification_id": notification.id, "reason": "recipient_missing", }, ) return # Получаем настройки пользователя try: prefs = user.channel_preferences except Exception: # Если настроек нет - создаем с дефолтными значениями from .models import UserChannelPreferences prefs, _created = UserChannelPreferences.objects.get_or_create(user=user) # Проверяем, не отключен ли этот тип уведомлений if not prefs.is_verb_enabled(notification.verb): logger.debug( "Notification routing skipped because event type is disabled", extra={ "event": "notification.routing.skipped", "notification_id": notification.id, "user_id": user.id, "notification_verb": notification.verb, "reason": "verb_disabled", }, ) return # Импортируем задачи from .tasks import ( send_email_notification, send_push_notification, send_websocket_notification, ) # Проверяем режим "Не беспокоить" is_dnd = prefs.is_in_dnd_period() # КРИТИЧНО: Отправляем задачи ПОСЛЕ коммита транзакции # Без этого Celery не найдет уведомление в БД (race condition) # Django документация: # https://docs.djangoproject.com/en/5.0/topics/db/transactions/#performing-actions-after-commit def send_to_channels(): """Отправка по каналам после успешного коммита транзакции""" if is_dnd: logger.debug( "Notification routing restricted by DND", extra={ "event": "notification.routing.dnd_restricted", "notification_id": notification.id, "user_id": user.id, "web_enabled": prefs.web_enabled, }, ) if prefs.web_enabled: _enqueue_notification_task( send_websocket_notification, notification.id, silent=True, ) return # Отправляем по каналам асинхронно через Celery if prefs.web_enabled: _enqueue_notification_task( send_websocket_notification, notification.id, silent=False, ) if prefs.email_enabled and prefs.email_frequency == "instant": _enqueue_notification_task(send_email_notification, notification.id) if prefs.push_enabled: _enqueue_notification_task(send_push_notification, notification.id) transaction.on_commit(send_to_channels) # === Digest Email (для email_frequency = daily/weekly) === def send_email_digest(user, frequency="daily"): """ Отправка дайджеста уведомлений Args: user: User объект frequency: 'daily' или 'weekly' """ try: from datetime import timedelta from django.utils import timezone from .models import Notification # Определяем период if frequency == "daily": since = timezone.now() - timedelta(days=1) elif frequency == "weekly": since = timezone.now() - timedelta(weeks=1) else: raise ValueError(f"Invalid frequency: {frequency}") # Получаем неотправленные уведомления notifications = Notification.objects.filter( recipient=user, timestamp__gte=since, emailed=False, deleted=False, ).order_by("-timestamp") if not notifications.exists(): logger.debug( "Notification digest skipped because no items are pending", extra={ "event": "notification.email_digest.skipped", "user_id": user.id, "frequency": frequency, "reason": "no_pending_notifications", }, ) return # Используем EmailNotificationSender для отправки дайджеста from .senders import EmailNotificationSender sender = EmailNotificationSender() success = sender.send_digest(user, notifications, frequency) if success: logger.info( "Notification email digest sent", extra={ "event": "notification.email_digest.sent", "user_id": user.id, "frequency": frequency, "notification_count": notifications.count(), }, ) except Exception as error: logger.exception( "Notification email digest failed", extra={ "event": "notification.email_digest.failed", "user_id": user.id, "frequency": frequency, "error_type": type(error).__name__, }, )