diff --git a/backend/bot/services/email_auth_service.py b/backend/bot/services/email_auth_service.py index 8d9b90b..9391932 100644 --- a/backend/bot/services/email_auth_service.py +++ b/backend/bot/services/email_auth_service.py @@ -17,8 +17,9 @@ from sqlalchemy.ext.asyncio import AsyncSession from bot.middlewares.i18n import JsonI18n from bot.services.email_templates import EmailContent, render_login_code +from bot.services.message_audit import log_user_message_delivery from config.settings import Settings -from db.dal import security_dal +from db.dal import security_dal, user_dal from db.models import EmailVerificationCode logger = logging.getLogger(__name__) @@ -222,6 +223,28 @@ class EmailAuthService: magic_link=magic_link, purpose=purpose, ) + resolved_target_user_id = target_user_id + if resolved_target_user_id is None: + try: + existing_user = await user_dal.get_user_by_email(session, normalized_email) + resolved_target_user_id = ( + int(existing_user.user_id) if existing_user is not None else None + ) + except Exception: + logger.exception( + "Failed to resolve email auth target user for audit log: %s", + normalized_email, + ) + await log_user_message_delivery( + session, + target_user_id=resolved_target_user_id, + event_type="email_login_code_sent" + if purpose == "login" + else "email_verification_code_sent", + channel="email", + recipient=normalized_email, + content=f"purpose={purpose} magic_link={bool(magic_link)}", + ) return EmailCodeRequestResult(ok=True) async def verify_code( diff --git a/backend/bot/services/message_audit.py b/backend/bot/services/message_audit.py new file mode 100644 index 0000000..0c3594c --- /dev/null +++ b/backend/bot/services/message_audit.py @@ -0,0 +1,57 @@ +import logging +from datetime import datetime, timezone +from typing import Optional + +from sqlalchemy.ext.asyncio import AsyncSession + +from db.dal import message_log_dal + +logger = logging.getLogger(__name__) + + +def _clean_piece(value: Optional[object]) -> str: + return str(value or "").strip() + + +async def log_user_message_delivery( + session: AsyncSession, + *, + target_user_id: Optional[int], + event_type: str, + channel: str, + content: str, + recipient: Optional[str] = None, + timestamp: Optional[datetime] = None, +) -> None: + """Add a best-effort user log entry for important outbound messages.""" + clean_event = _clean_piece(event_type) + clean_channel = _clean_piece(channel) + if not clean_event or not clean_channel: + return + + parts = [f"channel={clean_channel}"] + clean_recipient = _clean_piece(recipient) + if clean_recipient: + parts.append(f"recipient={clean_recipient}") + clean_content = _clean_piece(content) + if clean_content: + parts.append(clean_content) + + try: + await message_log_dal.create_message_log_no_commit( + session, + { + "user_id": None, + "event_type": clean_event, + "content": " | ".join(parts)[:4000], + "is_admin_event": False, + "target_user_id": int(target_user_id) if target_user_id is not None else None, + "timestamp": timestamp or datetime.now(timezone.utc), + }, + ) + except Exception: + logger.exception( + "Failed to add outbound message audit log for user %s event %s", + target_user_id, + clean_event, + ) diff --git a/backend/bot/services/subscription_lifecycle_notifications.py b/backend/bot/services/subscription_lifecycle_notifications.py index 8fedd60..7cb2510 100644 --- a/backend/bot/services/subscription_lifecycle_notifications.py +++ b/backend/bot/services/subscription_lifecycle_notifications.py @@ -13,6 +13,7 @@ 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 bot.services.message_audit import log_user_message_delivery from bot.services.telegram_notifications import ( TELEGRAM_NOTIFICATIONS_BLOCKED, TELEGRAM_NOTIFICATIONS_ENABLED, @@ -188,6 +189,18 @@ class SubscriptionLifecycleNotificationService: self._channel_key(stage.key, "telegram"), sent_at=sent_at, ) + await log_user_message_delivery( + session, + target_user_id=getattr(sub, "user_id", None), + event_type="telegram_subscription_notification_sent", + channel="telegram", + recipient=str(chat_id), + content=( + f"stage={stage.key} message_key={stage.message_key} " + f"subscription_id={getattr(sub, 'subscription_id', '')}" + ), + timestamp=sent_at, + ) if user: status = normalize_telegram_notification_status( getattr(user, "telegram_notifications_status", None) @@ -253,6 +266,18 @@ class SubscriptionLifecycleNotificationService: self._channel_key(stage.key, "email"), sent_at=sent_at, ) + await log_user_message_delivery( + session, + target_user_id=getattr(sub, "user_id", None), + event_type="email_subscription_notification_sent", + channel="email", + recipient=recipient, + content=( + f"stage={stage.key} message_key={stage.message_key} " + f"subscription_id={getattr(sub, 'subscription_id', '')}" + ), + timestamp=sent_at, + ) return True async def _already_sent( diff --git a/backend/bot/services/subscription_notification_worker.py b/backend/bot/services/subscription_notification_worker.py index 5cde438..674a5d9 100644 --- a/backend/bot/services/subscription_notification_worker.py +++ b/backend/bot/services/subscription_notification_worker.py @@ -13,6 +13,7 @@ from sqlalchemy.orm import selectinload, sessionmaker 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.message_audit import log_user_message_delivery from bot.services.panel_api_service import PanelApiService from bot.services.subscription_lifecycle_notifications import ( SubscriptionLifecycleNotificationService, @@ -331,6 +332,17 @@ class SubscriptionNotificationWorker: parse_mode="HTML", ) telegram_sent = True + await log_user_message_delivery( + session, + target_user_id=user_id, + event_type="telegram_traffic_warning_sent", + channel="telegram", + recipient=str(telegram_chat_id), + content=( + "kind=trial warning_key=trial_traffic_depleted " + f"used_bytes={used} limit_bytes={limit}" + ), + ) except Exception as exc: status = telegram_notification_status_from_error(exc) if status and user and user_id: @@ -355,6 +367,12 @@ class SubscriptionNotificationWorker: subject_key="email_trial_traffic_depleted_subject", message_text=message_text, dashboard_url=(getattr(self.settings, "SUBSCRIPTION_MINI_APP_URL", "") or None), + session=session, + audit_event_type="email_traffic_warning_sent", + audit_content=( + "kind=trial warning_key=trial_traffic_depleted " + f"used_bytes={used} limit_bytes={limit}" + ), ) return {"telegram": telegram_sent, "email": email_sent} diff --git a/backend/bot/services/tariff_worker.py b/backend/bot/services/tariff_worker.py index 49c2dad..3e19047 100644 --- a/backend/bot/services/tariff_worker.py +++ b/backend/bot/services/tariff_worker.py @@ -13,6 +13,7 @@ from sqlalchemy.orm import sessionmaker from bot.infra.redis import redis_lock from bot.middlewares.i18n import JsonI18n +from bot.services.message_audit import log_user_message_delivery from bot.services.panel_api_service import PanelApiService from bot.services.subscription_service import SubscriptionService from bot.services.user_email_notifications import send_user_notification_email @@ -111,6 +112,8 @@ class TariffTrafficWorker: subject_key: str, message_text: str, kind: str, + warning_key: str, + audit_content: str, ) -> None: try: user = await user_dal.get_user_by_id(session, user_id) @@ -131,6 +134,9 @@ class TariffTrafficWorker: if kind == "premium" else "email_traffic_warning_regular_cta" ), + session=session, + audit_event_type="email_traffic_warning_sent", + audit_content=f"{audit_content} subject_key={subject_key} warning_key={warning_key}", ) async def run(self) -> None: @@ -602,6 +608,15 @@ class TariffTrafficWorker: **usage, ) subject_key = "email_traffic_warning_regular_depleted_subject" + warning_key = ( + "traffic_warning_regular_almost" + if level < 100 + else "traffic_warning_regular_depleted" + ) + audit_content = ( + f"kind=regular warning_key={warning_key} level={level} " + f"used_bytes={used_val} limit_bytes={limit_val}" + ) if self.bot: try: markup = self._traffic_topup_markup(user_lang, "regular") @@ -611,6 +626,14 @@ class TariffTrafficWorker: reply_markup=markup, parse_mode="HTML", ) + await log_user_message_delivery( + session, + target_user_id=sub.user_id, + event_type="telegram_traffic_warning_sent", + channel="telegram", + recipient=str(sub.user_id), + content=audit_content, + ) except Exception: logging.exception("Failed to send traffic warning to user %s", sub.user_id) await self._send_traffic_warning_email( @@ -619,6 +642,8 @@ class TariffTrafficWorker: subject_key=subject_key, message_text=text, kind="regular", + warning_key=warning_key, + audit_content=audit_content, ) if ratio >= 1.0 and not sub.is_throttled: logging.info( @@ -1066,6 +1091,11 @@ class TariffTrafficWorker: servers=servers, **usage, ) + warning_key = "traffic_warning_premium_depleted" + audit_content = ( + f"kind=premium warning_key={warning_key} " + f"used_bytes={used_val} limit_bytes={limit_val}" + ) if self.bot: try: markup = self._traffic_topup_markup(user_lang, "premium") @@ -1075,6 +1105,14 @@ class TariffTrafficWorker: reply_markup=markup, parse_mode="HTML", ) + await log_user_message_delivery( + session, + target_user_id=sub.user_id, + event_type="telegram_traffic_warning_sent", + channel="telegram", + recipient=str(sub.user_id), + content=audit_content, + ) except Exception: logging.exception( "Failed to send premium traffic depleted warning to user %s", sub.user_id @@ -1085,6 +1123,8 @@ class TariffTrafficWorker: subject_key="email_traffic_warning_premium_depleted_subject", message_text=text, kind="premium", + warning_key=warning_key, + audit_content=audit_content, ) return @@ -1134,6 +1174,11 @@ class TariffTrafficWorker: servers=servers, **usage, ) + warning_key = "traffic_warning_premium_almost" + audit_content = ( + f"kind=premium warning_key={warning_key} level={int(level)} " + f"used_bytes={used_val} limit_bytes={limit_val}" + ) if self.bot: try: markup = self._traffic_topup_markup(user_lang, "premium") @@ -1143,6 +1188,14 @@ class TariffTrafficWorker: reply_markup=markup, parse_mode="HTML", ) + await log_user_message_delivery( + session, + target_user_id=sub.user_id, + event_type="telegram_traffic_warning_sent", + channel="telegram", + recipient=str(sub.user_id), + content=audit_content, + ) except Exception: logging.exception( "Failed to send premium traffic warning to user %s", sub.user_id @@ -1153,6 +1206,8 @@ class TariffTrafficWorker: subject_key="email_traffic_warning_premium_almost_subject", message_text=text, kind="premium", + warning_key=warning_key, + audit_content=audit_content, ) async def _premium_node_uuids_for_tariff(self, tariff) -> list[str]: diff --git a/backend/bot/services/user_email_notifications.py b/backend/bot/services/user_email_notifications.py index 6dc86be..e889a4e 100644 --- a/backend/bot/services/user_email_notifications.py +++ b/backend/bot/services/user_email_notifications.py @@ -1,9 +1,12 @@ import logging from typing import Any, Optional +from sqlalchemy.ext.asyncio import AsyncSession + from bot.middlewares.i18n import JsonI18n from bot.services.email_auth_service import EmailAuthService from bot.services.email_templates import render_user_notification +from bot.services.message_audit import log_user_message_delivery from config.settings import Settings @@ -34,6 +37,9 @@ async def send_user_notification_email( subject_kwargs: Optional[dict[str, Any]] = None, heading_key: Optional[str] = None, intro_key: Optional[str] = None, + session: Optional[AsyncSession] = None, + audit_event_type: Optional[str] = None, + audit_content: Optional[str] = None, ) -> bool: if not getattr(settings, "email_auth_configured", False): return False @@ -78,6 +84,15 @@ async def send_user_notification_email( email=recipient, content=content, ) + if session is not None and audit_event_type: + await log_user_message_delivery( + session, + target_user_id=getattr(user, "user_id", None), + event_type=audit_event_type, + channel="email", + recipient=recipient, + content=audit_content or f"subject_key={subject_key}", + ) return True except Exception: logging.exception("Failed to send user notification email to %s.", recipient) diff --git a/tests/test_message_audit.py b/tests/test_message_audit.py new file mode 100644 index 0000000..95dd9fb --- /dev/null +++ b/tests/test_message_audit.py @@ -0,0 +1,47 @@ +import unittest +from datetime import datetime, timezone + +from bot.services import message_audit + + +class MessageAuditTests(unittest.IsolatedAsyncioTestCase): + async def test_log_user_message_delivery_adds_targeted_log(self): + calls = [] + sent_at = datetime(2026, 5, 31, tzinfo=timezone.utc) + + async def fake_create(_session, payload): + calls.append(payload) + + original = message_audit.message_log_dal.create_message_log_no_commit + message_audit.message_log_dal.create_message_log_no_commit = fake_create + self.addCleanup( + self._restore_create_message_log, + original, + ) + + await message_audit.log_user_message_delivery( + object(), + target_user_id=42, + event_type="telegram_traffic_warning_sent", + channel="telegram", + recipient="100500", + content="kind=regular level=90", + timestamp=sent_at, + ) + + self.assertEqual( + calls, + [ + { + "user_id": None, + "event_type": "telegram_traffic_warning_sent", + "content": "channel=telegram | recipient=100500 | kind=regular level=90", + "is_admin_event": False, + "target_user_id": 42, + "timestamp": sent_at, + } + ], + ) + + def _restore_create_message_log(self, original): + message_audit.message_log_dal.create_message_log_no_commit = original