diff --git a/app/cabinet/routes/gift.py b/app/cabinet/routes/gift.py index 44be36b5..685759cc 100644 --- a/app/cabinet/routes/gift.py +++ b/app/cabinet/routes/gift.py @@ -1,5 +1,6 @@ """Gift subscription routes for cabinet.""" +import asyncio import re from datetime import UTC, datetime @@ -12,15 +13,16 @@ from app.config import settings from app.database.crud.landing import get_purchase_by_token from app.database.crud.system_setting import get_setting_value from app.database.crud.tariff import get_tariff_by_id -from app.database.crud.transaction import create_transaction +from app.database.crud.transaction import create_transaction, emit_transaction_side_effects from app.database.crud.user import subtract_user_balance -from app.database.models import GuestPurchaseStatus, PaymentMethod, Tariff, TransactionType, User +from app.database.models import GuestPurchase, GuestPurchaseStatus, PaymentMethod, Tariff, TransactionType, User from app.services.guest_purchase_service import ( GuestPurchaseError, create_purchase, fulfill_purchase, ) from app.services.payment_method_config_service import get_enabled_methods_for_user +from app.utils.cache import RateLimitCache from ..dependencies import get_cabinet_db, get_current_cabinet_user from ..schemas.gift import ( @@ -32,6 +34,7 @@ from ..schemas.gift import ( GiftPurchaseRequest, GiftPurchaseResponse, GiftPurchaseStatusResponse, + PendingGiftResponse, ) @@ -141,6 +144,18 @@ async def create_gift_purchase( detail='Gift feature is not enabled', ) + # Rate limit: 5 gift purchases per 60 seconds per user + is_limited = await RateLimitCache.is_rate_limited(user.id, 'gift_purchase', limit=5, window=60) + if is_limited: + raise HTTPException(status_code=status.HTTP_429_TOO_MANY_REQUESTS, detail='Too many requests') + + # Check if user has purchase restrictions + if getattr(user, 'restriction_subscription', False): + raise HTTPException( + status_code=status.HTTP_403_FORBIDDEN, + detail='Purchases are restricted for this account', + ) + # Validate recipient format if body.recipient_type == 'email' and not _EMAIL_RE.match(body.recipient_value): raise HTTPException( @@ -153,6 +168,21 @@ async def create_gift_purchase( detail='Invalid Telegram username format', ) + # Prevent self-gift + if body.recipient_type == 'telegram': + normalized_recipient = body.recipient_value.lstrip('@').lower() + if user.username and user.username.lower() == normalized_recipient: + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail='Cannot gift to yourself', + ) + elif body.recipient_type == 'email': + if user.email and user.email.lower() == body.recipient_value.lower(): + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail='Cannot gift to yourself', + ) + # Find tariff and validate period tariff = await get_tariff_by_id(db, body.tariff_id) if tariff is None or not tariff.is_active: @@ -193,6 +223,27 @@ async def create_gift_purchase( detail='Insufficient balance', ) + # Pre-check: try to resolve Telegram username via Bot API + # Placed after validation gates to prevent zero-cost enumeration. + # The resolved ID is passed to fulfill_purchase to avoid a duplicate API call. + recipient_warning: str | None = None + pre_resolved_telegram_id: int | None = None + if body.recipient_type == 'telegram': + tg_username = body.recipient_value.lstrip('@') + try: + from aiogram import Bot + + async with Bot(token=settings.BOT_TOKEN) as bot: + chat = await asyncio.wait_for(bot.get_chat(chat_id=f'@{tg_username}'), timeout=5.0) + pre_resolved_telegram_id = chat.id + except Exception: + recipient_warning = 'telegram_unresolvable' + logger.warning( + 'Telegram username not resolvable for gift', + username=tg_username, + buyer_id=user.id, + ) + # Create purchase record try: purchase = await create_purchase( @@ -231,7 +282,7 @@ async def create_gift_purchase( ) # Create transaction record - await create_transaction( + transaction = await create_transaction( db, user_id=user.id, type=TransactionType.GIFT_PAYMENT, @@ -247,9 +298,23 @@ async def create_gift_purchase( await db.commit() + # Emit deferred side-effects after atomic commit + await emit_transaction_side_effects( + db, + transaction, + amount_kopeks=price_kopeks, + user_id=user.id, + type=TransactionType.GIFT_PAYMENT, + payment_method=PaymentMethod.BALANCE, + description=f'Gift: {tariff.name} ({body.period_days}d) -> {body.recipient_value}', + ) + + # Capture token before fulfill_purchase — session state may change after rollback inside fulfill + purchase_token = purchase.token + # Fulfill the purchase (find/create recipient user, create subscription, notify) try: - await fulfill_purchase(db, purchase.token) + await fulfill_purchase(db, purchase_token, pre_resolved_telegram_id=pre_resolved_telegram_id) except Exception: logger.exception( 'Gift purchase fulfillment failed (purchase is paid, will retry)', @@ -258,10 +323,49 @@ async def create_gift_purchase( return GiftPurchaseResponse( status='ok', - purchase_token=purchase.token, + purchase_token=purchase_token, + warning=recipient_warning, ) +@router.get('/pending', response_model=list[PendingGiftResponse]) +async def get_pending_gifts( + user: User = Depends(get_current_cabinet_user), + db: AsyncSession = Depends(get_cabinet_db), +): + """Get pending gift purchases that the current user can activate.""" + result = await db.execute( + select(GuestPurchase) + .where( + GuestPurchase.user_id == user.id, + GuestPurchase.is_gift.is_(True), + GuestPurchase.status == GuestPurchaseStatus.PENDING_ACTIVATION.value, + ) + .order_by(GuestPurchase.created_at.desc()) + ) + purchases = result.scalars().all() + + pending: list[PendingGiftResponse] = [] + for p in purchases: + # Determine sender display name + sender_display = None + if p.contact_value: + sender_display = p.contact_value + + pending.append( + PendingGiftResponse( + token=p.token, + tariff_name=p.tariff.name if p.tariff else None, + period_days=p.period_days, + gift_message=p.gift_message, + sender_display=sender_display, + created_at=p.created_at, + ) + ) + + return pending + + @router.get('/purchase/{token}', response_model=GiftPurchaseStatusResponse) async def get_gift_purchase_status( token: str, @@ -276,11 +380,11 @@ async def get_gift_purchase_status( detail='Purchase not found', ) - # Only the buyer can view this + # Uniform 404 prevents token existence oracle if purchase.buyer_user_id != user.id: raise HTTPException( - status_code=status.HTTP_403_FORBIDDEN, - detail='Access denied', + status_code=status.HTTP_404_NOT_FOUND, + detail='Purchase not found', ) tariff_name = purchase.tariff.name if purchase.tariff else None diff --git a/app/cabinet/schemas/gift.py b/app/cabinet/schemas/gift.py index 16ecffa5..9e98b7a8 100644 --- a/app/cabinet/schemas/gift.py +++ b/app/cabinet/schemas/gift.py @@ -2,6 +2,8 @@ from __future__ import annotations +from datetime import datetime + from pydantic import BaseModel, Field, model_validator @@ -46,8 +48,8 @@ class GiftConfigResponse(BaseModel): class GiftPurchaseRequest(BaseModel): - tariff_id: int - period_days: int + tariff_id: int = Field(gt=0) + period_days: int = Field(gt=0, le=3650) recipient_type: str = Field(pattern=r'^(email|telegram)$') recipient_value: str = Field(min_length=1, max_length=255) gift_message: str | None = Field(default=None, max_length=1000) @@ -65,6 +67,7 @@ class GiftPurchaseResponse(BaseModel): status: str purchase_token: str payment_url: str | None = None + warning: str | None = None class GiftPurchaseStatusResponse(BaseModel): @@ -74,3 +77,12 @@ class GiftPurchaseStatusResponse(BaseModel): gift_message: str | None = None tariff_name: str | None = None period_days: int | None = None + + +class PendingGiftResponse(BaseModel): + token: str + tariff_name: str | None = None + period_days: int + gift_message: str | None = None + sender_display: str | None = None + created_at: datetime | None = None diff --git a/app/database/models.py b/app/database/models.py index e9c01fcb..259e5197 100644 --- a/app/database/models.py +++ b/app/database/models.py @@ -3078,6 +3078,10 @@ class GuestPurchase(Base): Index('ix_guest_purchases_status', 'status'), Index('ix_guest_purchases_contact', 'contact_type', 'contact_value'), Index('ix_guest_purchases_landing_status_paid', 'landing_id', 'status', 'paid_at'), + Index('ix_guest_purchases_source', 'source'), + Index('ix_guest_purchases_user_gift_status', 'user_id', 'is_gift', 'status'), + Index('ix_guest_purchases_status_paid_at', 'status', 'paid_at'), + Index('ix_guest_purchases_buyer_user_id', 'buyer_user_id'), ) id = Column(Integer, primary_key=True, index=True) diff --git a/app/services/guest_purchase_service.py b/app/services/guest_purchase_service.py index 095b3a0c..ddfe1408 100644 --- a/app/services/guest_purchase_service.py +++ b/app/services/guest_purchase_service.py @@ -3,11 +3,11 @@ import asyncio import re import secrets -from datetime import UTC, datetime +from datetime import UTC, datetime, timedelta from typing import Literal import structlog -from sqlalchemy import func, select +from sqlalchemy import func, or_, select from sqlalchemy.exc import IntegrityError from sqlalchemy.ext.asyncio import AsyncSession @@ -165,13 +165,21 @@ async def create_purchase( return purchase -async def fulfill_purchase(db: AsyncSession, purchase_token: str) -> GuestPurchase | None: +async def fulfill_purchase( + db: AsyncSession, + purchase_token: str, + pre_resolved_telegram_id: int | None = None, +) -> GuestPurchase | None: """After payment: find/create user, create subscription, send notification. Uses SELECT ... FOR UPDATE to prevent concurrent fulfillment of the same purchase. The PENDING_ACTIVATION path commits early and returns (terminal for this call). The DELIVERED path commits after subscription creation. Returns the updated purchase or None if not found. + + Args: + pre_resolved_telegram_id: If caller already resolved the recipient's telegram_id + via Bot API, pass it here to avoid a duplicate API call. """ result = await db.execute(select(GuestPurchase).where(GuestPurchase.token == purchase_token).with_for_update()) purchase = result.scalars().first() @@ -193,7 +201,10 @@ async def fulfill_purchase(db: AsyncSession, purchase_token: str) -> GuestPurcha recipient_type, recipient_value = _get_recipient_contact(purchase) # Find or create user for the recipient (no commit — stays within our transaction) - user, is_new_account = await _find_or_create_user(db, recipient_type, recipient_value, purchase=purchase) + user, is_new_account = await _find_or_create_user( + db, recipient_type, recipient_value, purchase=purchase, + pre_resolved_telegram_id=pre_resolved_telegram_id, + ) # Load tariff early — needed for both PENDING_ACTIVATION and DELIVERED paths tariff = await get_tariff_by_id(db, purchase.tariff_id) @@ -364,6 +375,7 @@ async def _find_or_create_user( contact_type: Literal['email', 'telegram'], contact_value: str, purchase: GuestPurchase | None = None, + pre_resolved_telegram_id: int | None = None, ) -> tuple[User, bool]: """Find user by email/telegram username or create a new one. @@ -372,6 +384,10 @@ async def _find_or_create_user( Returns (user, is_new_account) where is_new_account means a new password was generated. + Args: + pre_resolved_telegram_id: If caller already resolved the telegram_id via Bot API, + pass it here to skip the redundant API call. + NOTE: Does NOT commit — caller is responsible for committing the transaction. This preserves FOR UPDATE locks held by the caller. """ @@ -443,22 +459,23 @@ async def _find_or_create_user( normalized = username.lower() # Try to resolve telegram_id via Bot API (works if user has interacted with the bot) - resolved_telegram_id: int | None = None - try: - from aiogram import Bot + resolved_telegram_id: int | None = pre_resolved_telegram_id + if resolved_telegram_id is None: + try: + from aiogram import Bot - async with Bot(token=settings.BOT_TOKEN) as bot: - chat = await asyncio.wait_for( - bot.get_chat(chat_id=f'@{username}'), - timeout=5.0, - ) - resolved_telegram_id = chat.id - # Use the canonical username from Telegram if available - if chat.username: - username = chat.username - normalized = username.lower() - except Exception as exc: - logger.debug('Could not resolve telegram_id for username', username=username, error=str(exc)) + async with Bot(token=settings.BOT_TOKEN) as bot: + chat = await asyncio.wait_for( + bot.get_chat(chat_id=f'@{username}'), + timeout=5.0, + ) + resolved_telegram_id = chat.id + # Use the canonical username from Telegram if available + if chat.username: + username = chat.username + normalized = username.lower() + except Exception as exc: + logger.debug('Could not resolve telegram_id for username', username=username, error=str(exc)) # Search by telegram_id first (most reliable), then by username (case-insensitive) user = None @@ -714,8 +731,8 @@ async def send_guest_notification( notification_type=notification_type.value, ) - # Send separate credentials email for new/upgraded accounts (non-gift self-purchases) - if purchase.cabinet_password and not purchase.is_gift: + # Send separate credentials email for new accounts (self-purchases and gifts) + if purchase.cabinet_password: cred_template = None try: from app.cabinet.services.email_template_overrides import get_rendered_override @@ -729,8 +746,8 @@ async def send_guest_notification( 'subject': cred_subject, 'body_html': cred_body, } - except Exception: - pass + except Exception as e: + logger.debug('Failed to check credentials template override', e=e) if not cred_template: cred_template = templates.get_template(NotificationType.GUEST_CABINET_CREDENTIALS, language, context) if cred_template: @@ -874,3 +891,52 @@ async def activate_purchase(db: AsyncSession, purchase_token: str, *, skip_notif raise GuestPurchaseError('Activation failed, please try again', status_code=500) return purchase + + +async def retry_stuck_paid_purchases( + db: AsyncSession, + stale_minutes: int = 5, + limit: int = 10, + max_age_hours: int = 24, +) -> int: + """Retry fulfillment for purchases stuck in PAID status. + + Finds purchases that have been in PAID status for longer than stale_minutes + (but not older than max_age_hours) and attempts to fulfill them in isolated + sessions. Returns the number of successfully retried purchases. + + Purchases older than max_age_hours are left for manual investigation. + """ + from app.database.database import AsyncSessionLocal + + cutoff = datetime.now(UTC) - timedelta(minutes=stale_minutes) + max_age = datetime.now(UTC) - timedelta(hours=max_age_hours) + + # Collect tokens only — each retry gets its own session. + # NULL paid_at is included via or_() as a safety net for data anomalies. + result = await db.execute( + select(GuestPurchase.token) + .where( + GuestPurchase.status == GuestPurchaseStatus.PAID.value, + or_(GuestPurchase.paid_at < cutoff, GuestPurchase.paid_at.is_(None)), + or_(GuestPurchase.paid_at > max_age, GuestPurchase.paid_at.is_(None)), + ) + .order_by(GuestPurchase.paid_at.asc().nulls_first()) + .limit(limit) + ) + tokens = result.scalars().all() + + if not tokens: + return 0 + + retried = 0 + for token in tokens: + try: + async with AsyncSessionLocal() as retry_db: + await fulfill_purchase(retry_db, token) + retried += 1 + logger.info('Retried stuck purchase successfully', token_prefix=token[:5]) + except Exception: + logger.exception('Failed to retry stuck purchase', token_prefix=token[:5]) + + return retried diff --git a/app/services/monitoring_service.py b/app/services/monitoring_service.py index ffe528f8..6011abe7 100644 --- a/app/services/monitoring_service.py +++ b/app/services/monitoring_service.py @@ -226,6 +226,7 @@ class MonitoringService: await self._check_trial_expiring_soon(db) await self._check_trial_channel_subscriptions(db) await self._check_expired_subscription_followups(db) + await self._retry_stuck_guest_purchases(db) await self._cleanup_inactive_users(db) await self._sync_with_remnawave(db) @@ -1679,6 +1680,16 @@ class MonitoringService: 'Ошибка отправки уведомления о неудачном автоплатеже пользователю', telegram_id=user.telegram_id, e=e ) + async def _retry_stuck_guest_purchases(self, db: AsyncSession): + try: + from app.services.guest_purchase_service import retry_stuck_paid_purchases + + retried = await retry_stuck_paid_purchases(db, stale_minutes=5, limit=10) + if retried: + logger.info('Retried stuck guest purchases', retried=retried) + except Exception: + logger.error('Error retrying stuck guest purchases', exc_info=True) + async def _cleanup_inactive_users(self, db: AsyncSession): try: now = datetime.now(UTC) diff --git a/migrations/alembic/versions/0033_guest_purchase_gift_index.py b/migrations/alembic/versions/0033_guest_purchase_gift_index.py new file mode 100644 index 00000000..ee24d8f5 --- /dev/null +++ b/migrations/alembic/versions/0033_guest_purchase_gift_index.py @@ -0,0 +1,44 @@ +"""Add indexes for gift pending queries and retry on guest_purchases + +Adds three indexes: +- (user_id, is_gift, status) for dashboard pending gifts query +- (status, paid_at) for retry_stuck_paid_purchases query +- (buyer_user_id) for FK lookup performance + +Revision ID: 0033 +Revises: 0032 +""" + +from typing import Sequence, Union + +import sqlalchemy as sa +from alembic import op + +revision: str = '0033' +down_revision: Union[str, None] = '0032' +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + +INDEXES = [ + ('ix_guest_purchases_user_gift_status', ['user_id', 'is_gift', 'status']), + ('ix_guest_purchases_status_paid_at', ['status', 'paid_at']), + ('ix_guest_purchases_buyer_user_id', ['buyer_user_id']), +] + + +def _has_index(table: str, index_name: str) -> bool: + conn = op.get_bind() + inspector = sa.inspect(conn) + return index_name in [idx['name'] for idx in inspector.get_indexes(table)] + + +def upgrade() -> None: + for index_name, columns in INDEXES: + if not _has_index('guest_purchases', index_name): + op.create_index(index_name, 'guest_purchases', columns) + + +def downgrade() -> None: + for index_name, _ in reversed(INDEXES): + if _has_index('guest_purchases', index_name): + op.drop_index(index_name, table_name='guest_purchases')