214 lines
7.7 KiB
Python
214 lines
7.7 KiB
Python
"""
|
||
Обработчики каналов доставки уведомлений
|
||
|
||
Слушают 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)
|
||
"""
|
||
|
||
from django.db import transaction
|
||
from django.db.models.signals import post_save
|
||
from django.dispatch import receiver
|
||
import logging
|
||
|
||
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_websocket_notification,
|
||
send_email_notification,
|
||
send_push_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__,
|
||
},
|
||
)
|