feat: mirror subscription lifecycle notifications

2026-05-29 18:08:28 +03:00
parent 6fbb8eebec
commit 19f0f27a3b
13 changed files with 864 additions and 182 deletions
@@ -396,6 +396,13 @@ SETTINGS_MANIFEST: List[SettingField] = [
        "notifications",
        "Включены уведомления о подписке",
    ),
+    SettingField(
+        "SUBSCRIPTION_EMAIL_NOTIFICATIONS_ENABLED",
+        "bool",
+        "notifications",
+        "Дублировать уведомления о подписке на email",
+        "Письма отправляются только пользователям с привязанным email и рабочим SMTP.",
+    ),
    SettingField(
        "SUBSCRIPTION_NOTIFY_ON_EXPIRE", "bool", "notifications", "Уведомлять об истечении"
    ),
@@ -497,6 +497,101 @@ def render_subscription_expiring(
    return EmailContent(subject=subject, text="\n".join(text_lines), html=rendered)


+def _subscription_lifecycle_title(
+    i18n: JsonI18n,
+    lang: str,
+    notification_key: str,
+    *,
+    days_left: Optional[int],
+    hours_before: Optional[int],
+) -> str:
+    if notification_key == "before_2d_autorenew":
+        return _t_text(i18n, lang, "email_subscription_lifecycle_subject_autorenew")
+    if notification_key == "expired":
+        return _t_text(i18n, lang, "email_subscription_lifecycle_subject_expired")
+    if notification_key == "expired_24h_after":
+        return _t_text(i18n, lang, "email_subscription_lifecycle_subject_expired_after")
+    if hours_before is not None:
+        return _t_text(
+            i18n,
+            lang,
+            "email_subscription_lifecycle_subject_before_hours",
+            hours=hours_before,
+        )
+    return _t_text(
+        i18n,
+        lang,
+        "email_subscription_lifecycle_subject_before_days",
+        days=max(0, int(days_left or 0)),
+    )
+
+
+def render_subscription_lifecycle_notification(
+    settings: Settings,
+    *,
+    language_code: Optional[str],
+    notification_key: str,
+    message_text: str,
+    end_date_text: str,
+    dashboard_url: Optional[str],
+    days_left: Optional[int] = None,
+    hours_before: Optional[int] = None,
+    i18n: Optional[JsonI18n] = None,
+) -> EmailContent:
+    i18n = _resolve_i18n(i18n)
+    lang = _normalize_lang(language_code, settings)
+    accent = _safe_color(settings.WEBAPP_PRIMARY_COLOR)
+    brand = _brand_title(settings)
+    safe_dashboard_url = (dashboard_url or "").strip()
+    end_date = end_date_text or "—"
+    subject = _subscription_lifecycle_title(
+        i18n,
+        lang,
+        notification_key,
+        days_left=days_left,
+        hours_before=hours_before,
+    )
+    intro = _t_text(i18n, lang, "email_subscription_lifecycle_intro")
+    footer = _t_html(i18n, lang, "email_footer_auto", brand=brand)
+    cta_label = _t_text(i18n, lang, "email_subscription_lifecycle_cta")
+
+    rows = [
+        (_t_text(i18n, lang, "email_subscription_lifecycle_row_end_date"), end_date),
+    ]
+    message_html = (
+        f'<div style="margin:0 0 16px 0;background:{_BG};border:1px solid {_BORDER};'
+        f"border-radius:14px;padding:14px 16px;font-size:14px;line-height:1.55;color:{_TEXT};"
+        f'white-space:pre-wrap;">{html.escape(message_text or "")}</div>'
+    )
+    body_parts = [_info_rows_html(rows), message_html]
+    if safe_dashboard_url:
+        body_parts.append(_cta_button_html(label=cta_label, url=safe_dashboard_url, accent=accent))
+
+    rendered = _layout(
+        settings=settings,
+        preheader=subject,
+        heading=subject,
+        intro_html=html.escape(intro),
+        body_html="".join(body_parts),
+        footer_html=footer,
+    )
+
+    text_lines = [subject, "", message_text]
+    if safe_dashboard_url:
+        text_lines.extend(
+            [
+                "",
+                _t_text(
+                    i18n,
+                    lang,
+                    "email_subscription_lifecycle_text_renew",
+                    url=safe_dashboard_url,
+                ),
+            ]
+        )
+    return EmailContent(subject=subject, text="\n".join(text_lines), html=rendered)
+
+
 def _support_email(
    settings: Settings,
    i18n: Optional[JsonI18n],
@@ -3,12 +3,15 @@ import hashlib
 import hmac
 import json
 import logging
+from datetime import datetime, timedelta, timezone
 from typing import Optional

 from aiogram import Bot
 from aiogram.types import InlineKeyboardMarkup
 from aiohttp import web
-from sqlalchemy.orm import sessionmaker
+from sqlalchemy import or_, select
+from sqlalchemy.ext.asyncio import AsyncSession
+from sqlalchemy.orm import selectinload, sessionmaker

 from bot.infra.webhook_queue import enqueue_webhook_event
 from bot.keyboards.inline.user_keyboards import (
@@ -16,17 +19,32 @@ from bot.keyboards.inline.user_keyboards import (
    get_subscribe_only_markup,
 )
 from bot.middlewares.i18n import JsonI18n
+from bot.services.subscription_lifecycle_notifications import (
+    SubscriptionLifecycleNotificationService,
+    SubscriptionNotificationStage,
+)
 from config.settings import Settings
-from db.dal import tariff_dal, user_dal
+from db.dal import subscription_dal, tariff_dal, user_dal
+from db.models import Subscription, User

-from .email_auth_service import EmailAuthService
-from .email_templates import render_subscription_expiring
 from .panel_api_service import PanelApiService

 EVENT_MAP = {
-    "user.expires_in_72_hours": (3, "subscription_72h_notification"),
-    "user.expires_in_48_hours": (2, "subscription_48h_notification"),
-    "user.expires_in_24_hours": (1, "subscription_24h_notification"),
+    "user.expires_in_72_hours": SubscriptionNotificationStage(
+        key="before_3d",
+        message_key="subscription_72h_notification",
+        days_left=3,
+    ),
+    "user.expires_in_48_hours": SubscriptionNotificationStage(
+        key="before_2d",
+        message_key="subscription_48h_notification",
+        days_left=2,
+    ),
+    "user.expires_in_24_hours": SubscriptionNotificationStage(
+        key="before_1d",
+        message_key="subscription_24h_notification",
+        days_left=1,
+    ),
 }


@@ -48,6 +66,11 @@ class PanelWebhookService:
        self.i18n = i18n
        self.async_session_factory = async_session_factory
        self.panel_service = panel_service
+        self.lifecycle_notifications = SubscriptionLifecycleNotificationService(
+            settings,
+            bot,
+            i18n,
+        )
        self._event_semaphore = asyncio.Semaphore(self._MAX_CONCURRENT_EVENTS)
        if not self.settings.PANEL_WEBHOOK_SECRET:
            logging.error(
@@ -102,113 +125,182 @@ class PanelWebhookService:
        )

    async def handle_event(self, event_name: str, user_payload: dict):
-        telegram_id = user_payload.get("telegramId")
-        if not telegram_id:
-            logging.warning("Panel webhook without telegramId received")
-            return
-        user_id = int(telegram_id)
-
        if not self.settings.SUBSCRIPTION_NOTIFICATIONS_ENABLED:
            return

        async with self.async_session_factory() as session:
-            db_user = await user_dal.get_user_by_telegram_id(session, user_id)
-            if not db_user:
-                db_user = await user_dal.get_user_by_id(session, user_id)
-            internal_user_id = db_user.user_id if db_user else user_id
+            db_user = await self._user_for_payload(session, user_payload)
+            sub = await self._subscription_for_payload(session, user_payload, db_user)
+            telegram_id = self._payload_telegram_id(user_payload)
+            internal_user_id = (
+                int(db_user.user_id)
+                if db_user
+                else int(getattr(sub, "user_id", 0) or telegram_id or 0)
+            )
            lang = (
                db_user.language_code
                if db_user and db_user.language_code
                else self.settings.DEFAULT_LANGUAGE
            )
-            first_name = db_user.first_name or f"User {user_id}" if db_user else f"User {user_id}"
-            user_email = (db_user.email or "").strip() if db_user else ""
+            if not sub:
+                if not telegram_id:
+                    logging.warning("Panel webhook event %s has no local subscription", event_name)
+                    return
+                await self._send_legacy_without_dedupe(
+                    event_name,
+                    user_payload,
+                    int(telegram_id),
+                    lang,
+                    db_user,
+                )
+                return

-        markup = get_subscribe_only_markup(lang, self.i18n)
+            markup = get_subscribe_only_markup(lang, self.i18n)
+            end_date_text = self._payload_expire_date(user_payload)

-        if event_name in EVENT_MAP:
-            days_left, msg_key = EVENT_MAP[event_name]
-            hwid_renewal_note = await self._hwid_renewal_note(internal_user_id, lang)
-            if days_left == 1:
-                # Trigger auto-renew via SubscriptionService (wired in at factory)
-                try:
-                    subscription_service = getattr(self, "subscription_service", None)
-                    if subscription_service:
-                        async with self.async_session_factory() as session:
-                            from db.dal import subscription_dal
-
-                            sub = await subscription_dal.get_active_subscription_by_user_id(
-                                session, internal_user_id
-                            )
-                            if sub and sub.auto_renew_enabled and sub.provider == "yookassa":
-                                try:
-                                    ok = await subscription_service.charge_subscription_renewal(
-                                        session, sub
+            if event_name in EVENT_MAP:
+                stage = EVENT_MAP[event_name]
+                days_left = int(stage.days_left or 0)
+                hwid_renewal_note = await self._hwid_renewal_note(internal_user_id, lang)
+                if days_left == 1:
+                    # Trigger auto-renew via SubscriptionService (wired in at factory)
+                    try:
+                        subscription_service = getattr(self, "subscription_service", None)
+                        if subscription_service:
+                            async with self.async_session_factory() as renewal_session:
+                                active_sub = (
+                                    await subscription_dal.get_active_subscription_by_user_id(
+                                        renewal_session,
+                                        internal_user_id,
                                    )
-                                    # If initiation succeeded, suppress the 24h reminder by returning early  # noqa: E501
-                                    if ok:
-                                        await session.commit()
-                                        return
-                                    else:
-                                        await session.rollback()
-                                except Exception:
-                                    await session.rollback()
-                                    logging.exception("Auto-renew attempt (24h) failed")
-                except Exception:
-                    logging.exception("Auto-renew trigger (24h) failed pre-check")
-            if days_left <= self.settings.SUBSCRIPTION_NOTIFY_DAYS_BEFORE:
-                # For 48h event, if auto-renew is enabled, show special notice with cancel button
-                if days_left == 2:
-                    async with self.async_session_factory() as session:
-                        from db.dal import subscription_dal
-
-                        sub = await subscription_dal.get_active_subscription_by_user_id(
-                            session, internal_user_id
+                                )
+                                if (
+                                    active_sub
+                                    and active_sub.auto_renew_enabled
+                                    and active_sub.provider == "yookassa"
+                                ):
+                                    try:
+                                        ok = await subscription_service.charge_subscription_renewal(
+                                            renewal_session,
+                                            active_sub,
+                                        )
+                                        # If initiation succeeded, suppress the 24h reminder by returning early  # noqa: E501
+                                        if ok:
+                                            await renewal_session.commit()
+                                            return
+                                        await renewal_session.rollback()
+                                    except Exception:
+                                        await renewal_session.rollback()
+                                        logging.exception("Auto-renew attempt (24h) failed")
+                    except Exception:
+                        logging.exception("Auto-renew trigger (24h) failed pre-check")
+                if days_left <= self.settings.SUBSCRIPTION_NOTIFY_DAYS_BEFORE:
+                    # For 48h, auto-renew users get a cancel button instead.
+                    if days_left == 2:
+                        active_sub = await subscription_dal.get_active_subscription_by_user_id(
+                            session,
+                            internal_user_id,
                        )
                        logging.info(
                            "48h webhook check: user_id=%s sub_found=%s auto_renew=%s provider=%s",
-                            user_id,
-                            bool(sub),
-                            getattr(sub, "auto_renew_enabled", None) if sub else None,
-                            getattr(sub, "provider", None) if sub else None,
+                            internal_user_id,
+                            bool(active_sub),
+                            getattr(active_sub, "auto_renew_enabled", None) if active_sub else None,
+                            getattr(active_sub, "provider", None) if active_sub else None,
                        )
-                        if sub and sub.auto_renew_enabled and sub.provider == "yookassa":
+                        if (
+                            active_sub
+                            and active_sub.auto_renew_enabled
+                            and active_sub.provider == "yookassa"
+                        ):
                            cancel_kb = get_autorenew_cancel_keyboard(lang, self.i18n)
-                            await self._send_message(
-                                user_id,
-                                lang,
-                                "autorenew_48h_charge_tomorrow_notice",
-                                reply_markup=cancel_kb,
-                                user_name=first_name,
+                            await self.lifecycle_notifications.send_stage(
+                                session,
+                                sub,
+                                SubscriptionNotificationStage(
+                                    key="before_2d_autorenew",
+                                    message_key="autorenew_48h_charge_tomorrow_notice",
+                                    days_left=2,
+                                ),
+                                user=db_user,
+                                telegram_markup=cancel_kb,
                                extra_text=hwid_renewal_note,
+                                end_date_text=end_date_text,
                            )
+                            await session.commit()
                            return
-                await self._send_message(
-                    user_id,
-                    lang,
-                    msg_key,
-                    reply_markup=markup,
-                    user_name=first_name,
-                    end_date=user_payload.get("expireAt", "")[:10],
-                    extra_text=hwid_renewal_note,
-                )
-                if days_left == 3 and user_email:
-                    await self._send_subscription_expiring_email(
-                        recipient=user_email,
-                        lang=lang,
-                        days_left=days_left,
-                        end_date_text=user_payload.get("expireAt", "")[:10],
+                    await self.lifecycle_notifications.send_stage(
+                        session,
+                        sub,
+                        stage,
+                        user=db_user,
+                        telegram_markup=markup,
+                        extra_text=hwid_renewal_note,
+                        end_date_text=end_date_text,
                    )
-        elif event_name == "user.expired":
-            if self.settings.SUBSCRIPTION_NOTIFY_ON_EXPIRE:
-                await self._send_message(
-                    user_id,
-                    lang,
-                    "subscription_expired_notification",
-                    reply_markup=markup,
-                    user_name=first_name,
-                    end_date=user_payload.get("expireAt", "")[:10],
+                    await session.commit()
+            elif event_name == "user.expired":
+                if self.settings.SUBSCRIPTION_NOTIFY_ON_EXPIRE:
+                    await self.lifecycle_notifications.send_stage(
+                        session,
+                        sub,
+                        SubscriptionNotificationStage(
+                            key="expired",
+                            message_key="subscription_expired_notification",
+                            days_left=0,
+                        ),
+                        user=db_user,
+                        telegram_markup=markup,
+                        end_date_text=end_date_text,
+                    )
+                    await session.commit()
+            elif (
+                event_name == "user.expired_24_hours_ago"
+                and self.settings.SUBSCRIPTION_NOTIFY_AFTER_EXPIRE
+            ):
+                await self.lifecycle_notifications.send_stage(
+                    session,
+                    sub,
+                    SubscriptionNotificationStage(
+                        key="expired_24h_after",
+                        message_key="subscription_expired_yesterday_notification",
+                        days_left=0,
+                    ),
+                    user=db_user,
+                    telegram_markup=markup,
+                    end_date_text=end_date_text,
                )
+                await session.commit()
+
+    async def _send_legacy_without_dedupe(
+        self,
+        event_name: str,
+        user_payload: dict,
+        user_id: int,
+        lang: str,
+        db_user: Optional[User],
+    ) -> None:
+        first_name = getattr(db_user, "first_name", None) or f"User {user_id}"
+        markup = get_subscribe_only_markup(lang, self.i18n)
+        if event_name in EVENT_MAP:
+            stage = EVENT_MAP[event_name]
+            await self._send_message(
+                user_id,
+                lang,
+                stage.message_key,
+                reply_markup=markup,
+                user_name=first_name,
+                end_date=self._payload_expire_date(user_payload),
+            )
+        elif event_name == "user.expired" and self.settings.SUBSCRIPTION_NOTIFY_ON_EXPIRE:
+            await self._send_message(
+                user_id,
+                lang,
+                "subscription_expired_notification",
+                reply_markup=markup,
+                user_name=first_name,
+                end_date=self._payload_expire_date(user_payload),
+            )
        elif (
            event_name == "user.expired_24_hours_ago"
            and self.settings.SUBSCRIPTION_NOTIFY_AFTER_EXPIRE
@@ -219,33 +311,113 @@ class PanelWebhookService:
                "subscription_expired_yesterday_notification",
                reply_markup=markup,
                user_name=first_name,
-                end_date=user_payload.get("expireAt", "")[:10],
+                end_date=self._payload_expire_date(user_payload),
            )

-    async def _send_subscription_expiring_email(
+    async def _user_for_payload(
        self,
-        *,
-        recipient: str,
-        lang: str,
-        days_left: int,
-        end_date_text: str,
-    ) -> None:
-        """Best-effort branded reminder; silently no-ops without SMTP config."""
-        if not self.settings.email_auth_configured:
-            return
-        try:
-            content = render_subscription_expiring(
-                self.settings,
-                language_code=lang,
-                days_left=days_left,
-                end_date_text=end_date_text,
-                dashboard_url=(self.settings.SUBSCRIPTION_MINI_APP_URL or "").strip() or None,
-                i18n=self.i18n,
+        session: AsyncSession,
+        user_payload: dict,
+    ) -> Optional[User]:
+        telegram_id = self._payload_telegram_id(user_payload)
+        if telegram_id:
+            user = await user_dal.get_user_by_telegram_id(session, telegram_id)
+            if user:
+                return user
+            user = await user_dal.get_user_by_id(session, telegram_id)
+            if user:
+                return user
+
+        panel_uuid = self._payload_panel_uuid(user_payload)
+        if panel_uuid:
+            user = await user_dal.get_user_by_panel_uuid(session, panel_uuid)
+            if user:
+                return user
+
+        email = str(user_payload.get("email") or "").strip()
+        if email:
+            return await user_dal.get_user_by_email(session, email)
+        return None
+
+    async def _subscription_for_payload(
+        self,
+        session: AsyncSession,
+        user_payload: dict,
+        db_user: Optional[User],
+    ) -> Optional[Subscription]:
+        conditions = []
+        if db_user:
+            conditions.append(Subscription.user_id == db_user.user_id)
+        panel_uuid = self._payload_panel_uuid(user_payload)
+        if panel_uuid:
+            conditions.append(Subscription.panel_user_uuid == panel_uuid)
+        if not conditions:
+            return None
+        base_stmt = (
+            select(Subscription)
+            .where(
+                Subscription.skip_notifications == False,
+                or_(*conditions),
            )
-            email_service = EmailAuthService(self.settings, self.i18n)
-            await email_service.send_rendered_email(email=recipient, content=content)
-        except Exception:
-            logging.exception("Failed to send subscription-expiring email to %s", recipient)
+            .options(selectinload(Subscription.user))
+        )
+
+        expire_at = self._payload_expire_datetime(user_payload)
+        if expire_at is not None:
+            window_stmt = (
+                base_stmt.where(
+                    Subscription.end_date >= expire_at - timedelta(days=1),
+                    Subscription.end_date <= expire_at + timedelta(days=1),
+                )
+                .order_by(Subscription.end_date.desc())
+                .limit(1)
+            )
+            result = await session.execute(window_stmt)
+            found = result.scalars().first()
+            if found:
+                return found
+
+        stmt = base_stmt.order_by(Subscription.end_date.desc()).limit(1)
+        result = await session.execute(stmt)
+        return result.scalars().first()
+
+    @staticmethod
+    def _payload_telegram_id(user_payload: dict) -> Optional[int]:
+        raw = user_payload.get("telegramId")
+        try:
+            value = int(raw or 0)
+        except (TypeError, ValueError):
+            return None
+        return value if value > 0 else None
+
+    @staticmethod
+    def _payload_panel_uuid(user_payload: dict) -> str:
+        return str(
+            user_payload.get("uuid")
+            or user_payload.get("userUuid")
+            or user_payload.get("shortUuid")
+            or ""
+        ).strip()
+
+    @staticmethod
+    def _payload_expire_date(user_payload: dict) -> str:
+        return str(user_payload.get("expireAt") or "")[:10]
+
+    @staticmethod
+    def _payload_expire_datetime(user_payload: dict) -> Optional[datetime]:
+        raw = str(user_payload.get("expireAt") or "").strip()
+        if not raw:
+            return None
+        try:
+            value = datetime.fromisoformat(raw.replace("Z", "+00:00"))
+        except ValueError:
+            try:
+                value = datetime.fromisoformat(raw[:10])
+            except ValueError:
+                return None
+        if value.tzinfo is None:
+            return value.replace(tzinfo=timezone.utc)
+        return value.astimezone(timezone.utc)

    async def handle_webhook(
        self, raw_body: bytes, signature_header: Optional[str]
@@ -0,0 +1,239 @@
+import logging
+from dataclasses import dataclass
+from datetime import datetime, timezone
+from typing import Optional
+
+from aiogram import Bot
+from aiogram.types import InlineKeyboardMarkup
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from bot.keyboards.inline.user_keyboards import get_subscribe_only_markup
+from bot.middlewares.i18n import JsonI18n
+from bot.services.email_auth_service import EmailAuthService
+from bot.services.email_templates import render_subscription_lifecycle_notification
+from config.settings import Settings
+from db.dal import subscription_dal
+from db.models import Subscription, User
+
+
+@dataclass(frozen=True)
+class SubscriptionNotificationStage:
+    key: str
+    message_key: str
+    days_left: Optional[int] = None
+    hours_before: Optional[int] = None
+
+
+@dataclass(frozen=True)
+class SubscriptionNotificationDelivery:
+    telegram_sent: bool = False
+    email_sent: bool = False
+
+    @property
+    def any_sent(self) -> bool:
+        return self.telegram_sent or self.email_sent
+
+
+class SubscriptionLifecycleNotificationService:
+    def __init__(
+        self,
+        settings: Settings,
+        bot: Bot,
+        i18n: JsonI18n,
+        *,
+        email_service: Optional[EmailAuthService] = None,
+    ) -> None:
+        self.settings = settings
+        self.bot = bot
+        self.i18n = i18n
+        self.email_service = email_service
+
+    async def send_stage(
+        self,
+        session: AsyncSession,
+        sub: Subscription,
+        stage: SubscriptionNotificationStage,
+        *,
+        user: Optional[User] = None,
+        telegram_markup: Optional[InlineKeyboardMarkup] = None,
+        extra_text: str = "",
+        end_date_text: Optional[str] = None,
+        sent_at: Optional[datetime] = None,
+    ) -> SubscriptionNotificationDelivery:
+        if sent_at is None:
+            sent_at = datetime.now(timezone.utc)
+
+        resolved_user = user or getattr(sub, "user", None)
+        lang = getattr(resolved_user, "language_code", None) or self.settings.DEFAULT_LANGUAGE
+        user_id = int(getattr(sub, "user_id", 0) or 0)
+        user_name = getattr(resolved_user, "first_name", None) or f"User {user_id}"
+        final_end_date_text = end_date_text
+        if final_end_date_text is None:
+            end_date = self._as_utc(getattr(sub, "end_date", None))
+            final_end_date_text = end_date.strftime("%Y-%m-%d") if end_date else ""
+
+        kwargs = {"user_name": user_name, "end_date": final_end_date_text}
+        if stage.hours_before is not None:
+            kwargs["hours"] = stage.hours_before
+
+        message_text = self.i18n.gettext(lang, stage.message_key, **kwargs)
+        final_extra_text = str(extra_text or "").strip()
+        if final_extra_text:
+            message_text = f"{message_text}\n\n{final_extra_text}"
+
+        telegram_sent = await self._send_telegram(
+            session,
+            sub,
+            stage,
+            resolved_user,
+            lang=lang,
+            message_text=message_text,
+            markup=telegram_markup or get_subscribe_only_markup(lang, self.i18n),
+            sent_at=sent_at,
+        )
+        email_sent = await self._send_email(
+            session,
+            sub,
+            stage,
+            resolved_user,
+            lang=lang,
+            message_text=message_text,
+            end_date_text=final_end_date_text,
+            sent_at=sent_at,
+        )
+        return SubscriptionNotificationDelivery(
+            telegram_sent=telegram_sent,
+            email_sent=email_sent,
+        )
+
+    async def _send_telegram(
+        self,
+        session: AsyncSession,
+        sub: Subscription,
+        stage: SubscriptionNotificationStage,
+        user: Optional[User],
+        *,
+        lang: str,
+        message_text: str,
+        markup: Optional[InlineKeyboardMarkup],
+        sent_at: datetime,
+    ) -> bool:
+        chat_id = self._telegram_chat_id(user, getattr(sub, "user_id", None))
+        if chat_id is None:
+            return False
+        if await self._already_sent(session, sub.subscription_id, stage.key, "telegram"):
+            return False
+        try:
+            await self.bot.send_message(chat_id, message_text, reply_markup=markup)
+        except Exception:
+            logging.exception(
+                "Failed to send subscription notification %s to Telegram user %s",
+                stage.key,
+                chat_id,
+            )
+            return False
+        await subscription_dal.record_subscription_notification(
+            session,
+            sub.subscription_id,
+            self._channel_key(stage.key, "telegram"),
+            sent_at=sent_at,
+        )
+        return True
+
+    async def _send_email(
+        self,
+        session: AsyncSession,
+        sub: Subscription,
+        stage: SubscriptionNotificationStage,
+        user: Optional[User],
+        *,
+        lang: str,
+        message_text: str,
+        end_date_text: str,
+        sent_at: datetime,
+    ) -> bool:
+        if not getattr(self.settings, "SUBSCRIPTION_EMAIL_NOTIFICATIONS_ENABLED", True):
+            return False
+        if not getattr(self.settings, "email_auth_configured", False):
+            return False
+        recipient = str(getattr(user, "email", "") or "").strip() if user else ""
+        if not recipient:
+            return False
+        if await self._already_sent(session, sub.subscription_id, stage.key, "email"):
+            return False
+
+        try:
+            content = render_subscription_lifecycle_notification(
+                self.settings,
+                language_code=lang,
+                notification_key=stage.key,
+                message_text=message_text,
+                end_date_text=end_date_text,
+                dashboard_url=(self.settings.SUBSCRIPTION_MINI_APP_URL or "").strip() or None,
+                days_left=stage.days_left,
+                hours_before=stage.hours_before,
+                i18n=self.i18n,
+            )
+            email_service = self.email_service or EmailAuthService(self.settings, self.i18n)
+            await email_service.send_rendered_email(email=recipient, content=content)
+        except Exception:
+            logging.exception(
+                "Failed to send subscription notification %s to email %s",
+                stage.key,
+                recipient,
+            )
+            return False
+        await subscription_dal.record_subscription_notification(
+            session,
+            sub.subscription_id,
+            self._channel_key(stage.key, "email"),
+            sent_at=sent_at,
+        )
+        return True
+
+    async def _already_sent(
+        self,
+        session: AsyncSession,
+        subscription_id: int,
+        stage_key: str,
+        channel: str,
+    ) -> bool:
+        channel_key = self._channel_key(stage_key, channel)
+        if await subscription_dal.has_subscription_notification(
+            session,
+            subscription_id,
+            channel_key,
+        ):
+            return True
+
+        # Legacy rows were stored without a channel. Treat them as Telegram-only
+        # history so existing installs do not re-send old bot messages, while
+        # still allowing the newly introduced email channel to catch up.
+        return channel == "telegram" and await subscription_dal.has_subscription_notification(
+            session,
+            subscription_id,
+            stage_key,
+        )
+
+    @staticmethod
+    def _channel_key(stage_key: str, channel: str) -> str:
+        return f"{stage_key}:{channel}"
+
+    @staticmethod
+    def _telegram_chat_id(user: Optional[User], fallback_user_id: Optional[int]) -> Optional[int]:
+        for candidate in (getattr(user, "telegram_id", None), fallback_user_id):
+            try:
+                chat_id = int(candidate or 0)
+            except (TypeError, ValueError):
+                continue
+            if chat_id > 0:
+                return chat_id
+        return None
+
+    @staticmethod
+    def _as_utc(value: Optional[datetime]) -> Optional[datetime]:
+        if value is None:
+            return None
+        if value.tzinfo is None:
+            return value.replace(tzinfo=timezone.utc)
+        return value.astimezone(timezone.utc)
@@ -1,7 +1,6 @@
 import asyncio
 import logging
 import time
-from dataclasses import dataclass
 from datetime import datetime, timedelta, timezone
 from typing import Optional

@@ -15,6 +14,10 @@ from bot.infra.redis import redis_lock
 from bot.keyboards.inline.user_keyboards import get_subscribe_only_markup
 from bot.middlewares.i18n import JsonI18n
 from bot.services.panel_api_service import PanelApiService
+from bot.services.subscription_lifecycle_notifications import (
+    SubscriptionLifecycleNotificationService,
+    SubscriptionNotificationStage,
+)
 from bot.services.subscription_service import SubscriptionService
 from config.settings import Settings
 from db.dal import subscription_dal
@@ -26,13 +29,6 @@ EXPIRED_NOTIFICATION_WINDOW = timedelta(hours=24)
 EXPIRED_AFTER_NOTIFICATION_WINDOW = timedelta(hours=48)


-@dataclass(frozen=True)
-class SubscriptionNotificationStage:
-    key: str
-    message_key: str
-    hours_before: Optional[int] = None
-
-
 class SubscriptionNotificationWorker:
    def __init__(
        self,
@@ -49,6 +45,11 @@ class SubscriptionNotificationWorker:
        self.i18n = i18n
        self.panel_service = panel_service
        self.subscription_service = subscription_service
+        self.lifecycle_notifications = SubscriptionLifecycleNotificationService(
+            settings,
+            bot,
+            i18n,
+        )
        self._stopped = asyncio.Event()

    async def run(self) -> None:
@@ -114,18 +115,10 @@ class SubscriptionNotificationWorker:
            stage = self.stage_for_subscription(sub, now)
            if stage is None:
                continue
-            if await subscription_dal.has_subscription_notification(
+            await self.lifecycle_notifications.send_stage(
                session,
-                sub.subscription_id,
-                stage.key,
-            ):
-                continue
-            if not await self._send_expiry_notification(sub, stage):
-                continue
-            await subscription_dal.record_subscription_notification(
-                session,
-                sub.subscription_id,
-                stage.key,
+                sub,
+                stage,
                sent_at=now,
            )

@@ -164,6 +157,7 @@ class SubscriptionNotificationWorker:
                    return SubscriptionNotificationStage(
                        key=f"before_{days_before}d",
                        message_key=message_key,
+                        days_left=days_before,
                    )
            return None

@@ -175,6 +169,7 @@ class SubscriptionNotificationWorker:
            return SubscriptionNotificationStage(
                key="expired",
                message_key="subscription_expired_notification",
+                days_left=0,
            )
        if (
            getattr(self.settings, "SUBSCRIPTION_NOTIFY_AFTER_EXPIRE", True)
@@ -183,6 +178,7 @@ class SubscriptionNotificationWorker:
            return SubscriptionNotificationStage(
                key="expired_24h_after",
                message_key="subscription_expired_yesterday_notification",
+                days_left=0,
            )
        return None

@@ -257,38 +253,6 @@ class SubscriptionNotificationWorker:
            return None
        return data if isinstance(data, dict) else None

-    async def _send_expiry_notification(
-        self,
-        sub: Subscription,
-        stage: SubscriptionNotificationStage,
-    ) -> bool:
-        user_id = int(getattr(sub, "user_id", 0) or 0)
-        if user_id <= 0:
-            return False
-        user = getattr(sub, "user", None)
-        lang = getattr(user, "language_code", None) or self.settings.DEFAULT_LANGUAGE
-        user_name = getattr(user, "first_name", None) or f"User {user_id}"
-        end_date = self._as_utc(getattr(sub, "end_date", None))
-        end_date_text = end_date.strftime("%Y-%m-%d") if end_date else ""
-        translate = lambda k, **kw: self.i18n.gettext(lang, k, **kw)
-        kwargs = {"user_name": user_name, "end_date": end_date_text}
-        if stage.hours_before is not None:
-            kwargs["hours"] = stage.hours_before
-        try:
-            await self.bot.send_message(
-                user_id,
-                translate(stage.message_key, **kwargs),
-                reply_markup=get_subscribe_only_markup(lang, self.i18n),
-            )
-            return True
-        except Exception:
-            logging.exception(
-                "Failed to send subscription notification %s to user %s",
-                stage.key,
-                user_id,
-            )
-            return False
-
    async def _send_trial_traffic_depleted(
        self,
        sub: Subscription,