api/modules/notifications/services.py - master - systemsstrategyy/Treker

api/modules/notifications/services.py
181 строка7 KB
HaGaSRus
Волна 4 (аудит «как работает»): 24 фикса корректности/надёжности/эксплуатации
07 июл 2026, 14:15
07 июл 2026, 14:150dcebaa
"""NotificationService — публичный интерфейс notifications-модуля.

Все остальные модули (Kanban, Billing, Quality) вызывают ТОЛЬКО
методы этого класса через `from api.modules.notifications.api import NotificationService`.

Прямой импорт ORM-моделей `Notification` / `NotificationSetting` из других
модулей запрещён (конвенций разработки правило #4)."""

from __future__ import annotations

from uuid import UUID

import structlog
from sqlalchemy import select
from sqlalchemy import text as sa_text
from sqlalchemy.dialects.postgresql import insert as pg_insert
from sqlalchemy.ext.asyncio import AsyncSession

from api.core.settings import settings
from api.modules.auth.api import UserDirectory
from api.modules.notifications.models import Notification, NotificationSetting
from api.modules.notifications.tokens import make_unsubscribe_token
from api.shared.db.session import defer_task
from api.shared.schemas.notification import NotificationKind
from workers.tasks.notifications import send_email_task, send_telegram_task

log = structlog.get_logger(__name__)

# telegram по умолчанию off — юзер включает после привязки chat_id.
DEFAULT_CHANNELS: dict[str, bool] = {"in_app": True, "email": True, "telegram": False}


class NotificationService:
    """Stateless-сервис над уведомлениями. Состояние — в БД и Procrastinate."""

    @staticmethod
    async def get_preferences(
        db: AsyncSession,
        *,
        user_uid: UUID,
        kind: NotificationKind,
    ) -> dict[str, bool]:
        """Возвращает channels для (user, kind). Дефолт если row не существует."""
        setting = await db.scalar(
            select(NotificationSetting).where(
                NotificationSetting.user_id == user_uid,
                NotificationSetting.kind == kind.value,
            )
        )
        if setting is None:
            return dict(DEFAULT_CHANNELS)
        return dict(setting.channels)

    @staticmethod
    async def set_preference(
        db: AsyncSession,
        *,
        user_uid: UUID,
        kind: NotificationKind,
        in_app: bool,
        email: bool,
        telegram: bool = False,
    ) -> None:
        """UPSERT настроек: создаёт row если нет, иначе обновляет channels."""
        channels = {"in_app": in_app, "email": email, "telegram": telegram}
        stmt = (
            pg_insert(NotificationSetting)
            .values(user_id=user_uid, kind=kind.value, channels=channels)
            .on_conflict_do_update(
                index_elements=["user_id", "kind"],
                set_={"channels": channels},
            )
        )
        await db.execute(stmt)
        await db.flush()

    @staticmethod
    async def notify(
        db: AsyncSession,
        *,
        user_uid: UUID,
        kind: NotificationKind,
        title: str,
        body: str,
        link: str | None = None,
    ) -> None:
        """Главный публичный метод. Fire-and-forget.

        Шаги:
        1. Читает per-kind настройки юзера (или дефолты).
        2. Если in_app=True — добавляет row в notifications.notifications
           (триггер pg_notify публикует в SSE-канал автоматически).
        3. Если email=True — defer'ит send_email_task в Procrastinate.

        Деактивированные (is_active=False) и несуществующие получатели
        отфильтровываются целиком — ни in-app, ни email, ни telegram
        (аудит w4, ранг 13: уволенный сотрудник получал дедлайн-письма)."""
        # cross-module (правило #4): не импортируем auth.User, читаем raw SQL.
        recipient_active = await db.scalar(
            sa_text("SELECT is_active FROM auth.users WHERE uid = :uid"),
            {"uid": str(user_uid)},
        )
        if not recipient_active:
            log.info(
                "notification.skipped_inactive_user",
                user_uid=str(user_uid),
                kind=kind.value,
            )
            return

        prefs = await NotificationService.get_preferences(db, user_uid=user_uid, kind=kind)

        if prefs["in_app"]:
            notification = Notification(
                user_id=user_uid,
                kind=kind.value,
                title=title,
                body=body,
                link=link,
            )
            db.add(notification)
            await db.flush()
            log.info("notification.created", user_uid=str(user_uid), kind=kind.value)

        if prefs["email"]:
            user_public = await UserDirectory.get_user_public(db, user_uid)
            if user_public is None:
                log.warning(
                    "notification.email_skipped_unknown_user",
                    user_uid=str(user_uid),
                )
                return
            secret = settings.secret_key.get_secret_value().encode("utf-8")
            unsubscribe_token = make_unsubscribe_token(user_uid, kind, secret)
            unsubscribe_link = (
                f"{settings.base_url}/api/v1/notifications/unsubscribe?token={unsubscribe_token}"
            )
            # defer_task: в сессиях с after-commit режимом (get_db, periodic
            # check_due_dates) заявка уйдёт в Procrastinate только после
            # успешного commit бизнес-транзакции (аудит w4, ранг 2).
            await defer_task(
                db,
                send_email_task,
                to=user_public.email,
                subject=title,
                template=kind.value,
                context={
                    "user": {
                        "full_name": user_public.full_name,
                        "email": user_public.email,
                    },
                    "title": title,
                    "body": body,
                    "link": link,
                    "base_url": settings.base_url,
                    "unsubscribe_link": unsubscribe_link,
                },
                user_uid=str(user_uid),
            )
            log.info(
                "notification.email_enqueued",
                user_uid=str(user_uid),
                kind=kind.value,
            )

        if prefs.get("telegram"):
            # cross-zone (правило #4): не импортируем auth.User, читаем raw SQL.
            chat_id = await db.scalar(
                sa_text("SELECT telegram_chat_id FROM auth.users WHERE uid = :uid"),
                {"uid": str(user_uid)},
            )
            if chat_id:
                tg_text = f"{title}\n\n{body}"
                if link:
                    tg_text += f"\n\n{settings.base_url}{link}"
                await defer_task(db, send_telegram_task, chat_id=chat_id, text=tg_text)
                log.info(
                    "notification.telegram_enqueued",
                    user_uid=str(user_uid),
                    kind=kind.value,
                )