diff --git a/app/cabinet/routes/__init__.py b/app/cabinet/routes/__init__.py
index f1adf283..d0924ddf 100644
--- a/app/cabinet/routes/__init__.py
+++ b/app/cabinet/routes/__init__.py
@@ -20,6 +20,7 @@ from .admin_pinned_messages import router as admin_pinned_messages_router
from .admin_policies import router as admin_policies_router
from .admin_promo_offers import router as admin_promo_offers_router
from .admin_promocodes import promo_groups_router as admin_promo_groups_router, router as admin_promocodes_router
+from .admin_referral_network import router as admin_referral_network_router
from .admin_remnawave import router as admin_remnawave_router
from .admin_roles import router as admin_roles_router
from .admin_sales_stats import router as admin_sales_stats_router
@@ -99,6 +100,7 @@ router.include_router(admin_wheel_router)
router.include_router(admin_tariffs_router)
router.include_router(admin_servers_router)
router.include_router(admin_stats_router)
+router.include_router(admin_referral_network_router)
router.include_router(admin_sales_stats_router)
router.include_router(admin_ban_system_router)
router.include_router(admin_broadcasts_router)
diff --git a/app/cabinet/routes/admin_referral_network.py b/app/cabinet/routes/admin_referral_network.py
new file mode 100644
index 00000000..10fecfc7
--- /dev/null
+++ b/app/cabinet/routes/admin_referral_network.py
@@ -0,0 +1,1444 @@
+"""Admin routes for referral network graph visualization."""
+
+import re
+from collections import defaultdict
+
+import structlog
+from fastapi import APIRouter, Depends, HTTPException, Query, status
+from pydantic import BaseModel
+from sqlalchemy import and_, func, literal, or_, select
+from sqlalchemy.ext.asyncio import AsyncSession
+from sqlalchemy.orm import selectinload
+
+from app.database.models import (
+ AdvertisingCampaign,
+ AdvertisingCampaignRegistration,
+ PartnerStatus,
+ ReferralEarning,
+ Subscription,
+ Tariff,
+ Transaction,
+ TransactionType,
+ User,
+)
+from app.utils.cache import RateLimitCache
+
+from ..dependencies import get_cabinet_db, require_permission
+
+
+logger = structlog.get_logger(__name__)
+
+router = APIRouter(prefix='/admin/referral-network', tags=['Cabinet Admin Referral Network'])
+
+# ============ Constants ============
+
+SPENT_TRANSACTION_TYPES: tuple[str, ...] = (TransactionType.SUBSCRIPTION_PAYMENT.value,)
+
+EDGE_TYPE_REFERRAL = 'referral'
+EDGE_TYPE_CAMPAIGN = 'campaign'
+EDGE_TYPE_PARTNER_CAMPAIGN = 'partner_campaign'
+
+NODE_PREFIX_USER = 'user_'
+NODE_PREFIX_CAMPAIGN = 'campaign_'
+
+TOP_REFERRERS_LIMIT = 5
+SEARCH_RESULTS_LIMIT = 20
+GRAPH_MAX_NODES = 5000
+
+# Rate limits (per admin user, per window)
+GRAPH_RATE_LIMIT = 10
+GRAPH_RATE_WINDOW = 60
+DETAIL_RATE_LIMIT = 30
+DETAIL_RATE_WINDOW = 60
+SEARCH_RATE_LIMIT = 30
+SEARCH_RATE_WINDOW = 60
+
+MAX_REFERRAL_DEPTH = 50
+
+# Regex to escape LIKE wildcards
+_LIKE_ESCAPE_RE = re.compile(r'([%_\\])')
+
+
+# ============ Schemas ============
+
+
+class NetworkUserNode(BaseModel):
+ id: int
+ tg_id: int | None
+ username: str | None
+ email: str | None
+ display_name: str
+ is_partner: bool
+ referrer_id: int | None
+ campaign_id: int | None
+ direct_referrals: int
+ total_branch_users: int
+ branch_revenue_kopeks: int
+ personal_revenue_kopeks: int
+ personal_spent_kopeks: int
+ subscription_name: str | None
+ subscription_end: str | None
+ registered_at: str | None
+
+
+class TopReferrer(BaseModel):
+ user_id: int
+ username: str | None
+ referral_count: int
+
+
+class NetworkCampaignNode(BaseModel):
+ id: int
+ name: str
+ start_parameter: str
+ is_active: bool
+ direct_users: int
+ total_network_users: int
+ total_revenue_kopeks: int
+ conversion_rate: float
+ avg_check_kopeks: int
+ top_referrers: list[TopReferrer]
+
+
+class NetworkEdge(BaseModel):
+ source: str
+ target: str
+ type: str
+
+
+class NetworkGraphResponse(BaseModel):
+ users: list[NetworkUserNode]
+ campaigns: list[NetworkCampaignNode]
+ edges: list[NetworkEdge]
+ total_users: int
+ total_referrers: int
+ total_campaigns: int
+ total_earnings_kopeks: int
+
+
+class NetworkUserDetail(BaseModel):
+ id: int
+ tg_id: int | None
+ username: str | None
+ email: str | None
+ display_name: str
+ is_partner: bool
+ referrer_id: int | None
+ referrer_display_name: str | None
+ campaign_id: int | None
+ campaign_name: str | None
+ direct_referrals: int
+ total_branch_users: int
+ branch_revenue_kopeks: int
+ personal_revenue_kopeks: int
+ personal_spent_kopeks: int
+ subscription_name: str | None
+ subscription_end: str | None
+ registered_at: str | None
+
+
+class NetworkCampaignDetail(BaseModel):
+ id: int
+ name: str
+ start_parameter: str
+ is_active: bool
+ direct_users: int
+ total_network_users: int
+ total_revenue_kopeks: int
+ conversion_rate: float
+ avg_check_kopeks: int
+ top_referrers: list[TopReferrer]
+
+
+class NetworkSearchResult(BaseModel):
+ users: list[NetworkUserNode]
+ campaigns: list[NetworkCampaignNode]
+
+
+class CampaignOption(BaseModel):
+ id: int
+ name: str
+ start_parameter: str
+ is_active: bool
+ direct_users: int
+
+
+class PartnerOption(BaseModel):
+ id: int
+ display_name: str
+ username: str | None
+ campaign_count: int
+
+
+class ScopeOptionsResponse(BaseModel):
+ campaigns: list[CampaignOption]
+ partners: list[PartnerOption]
+
+
+# ============ Helpers ============
+
+
+def _user_display_name(user: User) -> str:
+ """Build display name from User model."""
+ parts = [user.first_name, user.last_name]
+ name = ' '.join(filter(None, parts))
+ if name:
+ return name
+ if user.username:
+ return user.username
+ if user.telegram_id:
+ return f'ID{user.telegram_id}'
+ if user.email:
+ return user.email.split('@')[0]
+ return f'User{user.id}'
+
+
+def _format_datetime(dt: object) -> str | None:
+ """Format datetime to ISO string, handle None."""
+ if dt is None:
+ return None
+ return dt.isoformat()
+
+
+def _escape_like(value: str) -> str:
+ """Escape LIKE wildcards (%, _, \\) to prevent injection."""
+ return _LIKE_ESCAPE_RE.sub(r'\\\1', value)
+
+
+def _build_user_node(
+ user: User,
+ *,
+ direct_referral_count: int,
+ personal_revenue: int,
+ branch_revenue: int,
+ personal_spent: int,
+ campaign_id: int | None,
+ subscription_name: str | None,
+ subscription_end_str: str | None,
+) -> NetworkUserNode:
+ return NetworkUserNode(
+ id=user.id,
+ tg_id=user.telegram_id,
+ username=user.username,
+ email=user.email,
+ display_name=_user_display_name(user),
+ is_partner=user.partner_status == PartnerStatus.APPROVED.value,
+ referrer_id=user.referred_by_id,
+ campaign_id=campaign_id,
+ direct_referrals=direct_referral_count,
+ total_branch_users=direct_referral_count,
+ branch_revenue_kopeks=branch_revenue,
+ personal_revenue_kopeks=personal_revenue,
+ personal_spent_kopeks=personal_spent,
+ subscription_name=subscription_name,
+ subscription_end=subscription_end_str,
+ registered_at=_format_datetime(user.created_at),
+ )
+
+
+# ============ Data fetching ============
+
+
+async def _fetch_network_user_ids(db: AsyncSession) -> set[int]:
+ """Get IDs of all users that participate in the referral network.
+
+ A user is in the network if they:
+ - have a referrer (referred_by_id IS NOT NULL)
+ - have at least one referral (someone's referred_by_id points to them)
+ - have at least one campaign registration
+ """
+ # Users with referrer
+ referred_q = select(User.id).where(User.referred_by_id.isnot(None))
+
+ # Users who are referrers
+ referrer_q = select(User.referred_by_id).where(User.referred_by_id.isnot(None)).distinct()
+
+ # Users with campaign registration
+ campaign_user_q = select(AdvertisingCampaignRegistration.user_id).distinct()
+
+ result_referred = await db.execute(referred_q)
+ result_referrers = await db.execute(referrer_q)
+ result_campaign = await db.execute(campaign_user_q)
+
+ user_ids: set[int] = set()
+ user_ids.update(row[0] for row in result_referred)
+ user_ids.update(row[0] for row in result_referrers if row[0] is not None)
+ user_ids.update(row[0] for row in result_campaign)
+
+ return user_ids
+
+
+async def _fetch_direct_referral_counts(db: AsyncSession, user_ids: set[int] | None = None) -> dict[int, int]:
+ """Return {user_id: count_of_direct_referrals}.
+
+ When user_ids is provided, only counts referrals for those users.
+ """
+ stmt = select(User.referred_by_id, func.count(User.id)).where(User.referred_by_id.isnot(None))
+ if user_ids is not None:
+ stmt = stmt.where(User.referred_by_id.in_(user_ids))
+ stmt = stmt.group_by(User.referred_by_id)
+ result = await db.execute(stmt)
+ return {row[0]: row[1] for row in result}
+
+
+async def _fetch_personal_revenue(db: AsyncSession, user_ids: set[int]) -> dict[int, int]:
+ """Return {user_id: total_referral_earnings_kopeks} for given users."""
+ if not user_ids:
+ return {}
+
+ stmt = (
+ select(ReferralEarning.user_id, func.coalesce(func.sum(ReferralEarning.amount_kopeks), 0))
+ .where(ReferralEarning.user_id.in_(user_ids))
+ .group_by(ReferralEarning.user_id)
+ )
+ result = await db.execute(stmt)
+ return {row[0]: row[1] for row in result}
+
+
+async def _fetch_branch_revenue(db: AsyncSession, user_ids: set[int]) -> dict[int, int]:
+ """Return {referrer_id: total_spent_by_direct_referrals}.
+
+ This sums subscription payments by each user's direct referrals (one level deep).
+ """
+ if not user_ids:
+ return {}
+
+ referred_user = (
+ select(User.id, User.referred_by_id)
+ .where(and_(User.referred_by_id.isnot(None), User.referred_by_id.in_(user_ids)))
+ .subquery()
+ )
+
+ stmt = (
+ select(
+ referred_user.c.referred_by_id,
+ func.coalesce(func.sum(Transaction.amount_kopeks), 0),
+ )
+ .join(referred_user, Transaction.user_id == referred_user.c.id)
+ .where(
+ and_(
+ Transaction.type.in_(SPENT_TRANSACTION_TYPES),
+ Transaction.is_completed.is_(True),
+ )
+ )
+ .group_by(referred_user.c.referred_by_id)
+ )
+ result = await db.execute(stmt)
+ return {row[0]: row[1] for row in result}
+
+
+async def _fetch_personal_spent(db: AsyncSession, user_ids: set[int]) -> dict[int, int]:
+ """Return {user_id: total_spent_kopeks} for given users."""
+ if not user_ids:
+ return {}
+
+ stmt = (
+ select(Transaction.user_id, func.coalesce(func.sum(Transaction.amount_kopeks), 0))
+ .where(
+ and_(
+ Transaction.user_id.in_(user_ids),
+ Transaction.type.in_(SPENT_TRANSACTION_TYPES),
+ Transaction.is_completed.is_(True),
+ )
+ )
+ .group_by(Transaction.user_id)
+ )
+ result = await db.execute(stmt)
+ return {row[0]: row[1] for row in result}
+
+
+async def _fetch_campaign_registrations(db: AsyncSession, user_ids: set[int] | None = None) -> dict[int, int]:
+ """Return {user_id: first_campaign_id} (the earliest registration per user).
+
+ When user_ids is provided, only fetches registrations for those users.
+ """
+ # Use a window function to pick the first registration per user
+ row_num = (
+ func.row_number()
+ .over(
+ partition_by=AdvertisingCampaignRegistration.user_id,
+ order_by=AdvertisingCampaignRegistration.created_at.asc(),
+ )
+ .label('rn')
+ )
+
+ inner = select(
+ AdvertisingCampaignRegistration.user_id,
+ AdvertisingCampaignRegistration.campaign_id,
+ row_num,
+ )
+ if user_ids is not None:
+ inner = inner.where(AdvertisingCampaignRegistration.user_id.in_(user_ids))
+ subq = inner.subquery()
+
+ stmt = select(subq.c.user_id, subq.c.campaign_id).where(subq.c.rn == 1)
+ result = await db.execute(stmt)
+ return {row[0]: row[1] for row in result}
+
+
+async def _fetch_subscription_info(db: AsyncSession, user_ids: set[int]) -> dict[int, tuple[str | None, str | None]]:
+ """Return {user_id: (tariff_name, end_date_iso)} for given users."""
+ if not user_ids:
+ return {}
+
+ stmt = (
+ select(Subscription.user_id, Tariff.name, Subscription.end_date)
+ .outerjoin(Tariff, Subscription.tariff_id == Tariff.id)
+ .where(Subscription.user_id.in_(user_ids))
+ )
+ result = await db.execute(stmt)
+ return {row[0]: (row[1], _format_datetime(row[2]) if row[2] else None) for row in result}
+
+
+async def _fetch_campaign_stats(
+ db: AsyncSession,
+ referral_counts: dict[int, int],
+ campaign_ids: set[int] | None = None,
+) -> list[NetworkCampaignNode]:
+ """Build campaign nodes with aggregated stats. Optionally scoped to campaign_ids."""
+ stmt = select(AdvertisingCampaign)
+ if campaign_ids is not None:
+ stmt = stmt.where(AdvertisingCampaign.id.in_(campaign_ids))
+ result = await db.execute(stmt)
+ campaigns = list(result.scalars().all())
+
+ if not campaigns:
+ return []
+
+ fetched_campaign_ids = [c.id for c in campaigns]
+
+ # Users per campaign (for computing registration counts, network users, and top referrers)
+ user_campaign_stmt = select(
+ AdvertisingCampaignRegistration.campaign_id,
+ AdvertisingCampaignRegistration.user_id,
+ ).where(AdvertisingCampaignRegistration.campaign_id.in_(fetched_campaign_ids))
+ uc_result = await db.execute(user_campaign_stmt)
+ campaign_user_ids: dict[int, list[int]] = defaultdict(list)
+ for row in uc_result:
+ campaign_user_ids[row[0]].append(row[1])
+
+ # Registration counts derived from the same query
+ reg_counts: dict[int, int] = {cid: len(uids) for cid, uids in campaign_user_ids.items()}
+
+ # Total spending by users from each campaign (for revenue, conversion, avg check)
+ all_campaign_users = set()
+ for uids in campaign_user_ids.values():
+ all_campaign_users.update(uids)
+
+ user_spent: dict[int, int] = {}
+ if all_campaign_users:
+ spent_stmt = (
+ select(Transaction.user_id, func.coalesce(func.sum(Transaction.amount_kopeks), 0))
+ .where(
+ and_(
+ Transaction.user_id.in_(all_campaign_users),
+ Transaction.type.in_(SPENT_TRANSACTION_TYPES),
+ Transaction.is_completed.is_(True),
+ )
+ )
+ .group_by(Transaction.user_id)
+ )
+ spent_result = await db.execute(spent_stmt)
+ user_spent = {row[0]: row[1] for row in spent_result}
+
+ # Top referrers: users from campaign who have the most referrals
+ # Also need usernames for those users
+ top_referrer_user_ids: set[int] = set()
+ campaign_top_referrers_raw: dict[int, list[tuple[int, int]]] = {}
+
+ for cid, uids in campaign_user_ids.items():
+ scored = [(uid, referral_counts.get(uid, 0)) for uid in uids if referral_counts.get(uid, 0) > 0]
+ scored.sort(key=lambda x: x[1], reverse=True)
+ top = scored[:TOP_REFERRERS_LIMIT]
+ campaign_top_referrers_raw[cid] = top
+ top_referrer_user_ids.update(uid for uid, _ in top)
+
+ username_map: dict[int, str | None] = {}
+ if top_referrer_user_ids:
+ uname_stmt = select(User.id, User.username).where(User.id.in_(top_referrer_user_ids))
+ uname_result = await db.execute(uname_stmt)
+ username_map = {row[0]: row[1] for row in uname_result}
+
+ campaign_nodes: list[NetworkCampaignNode] = []
+ for campaign in campaigns:
+ cid = campaign.id
+ direct_users = reg_counts.get(cid, 0)
+ c_user_ids = campaign_user_ids.get(cid, [])
+
+ # Total network users: direct users + their referrals
+ network_users = direct_users
+ for uid in c_user_ids:
+ network_users += referral_counts.get(uid, 0)
+
+ # Conversion = users who spent > 0 / total registered
+ paying_users = sum(1 for uid in c_user_ids if user_spent.get(uid, 0) > 0)
+ conversion_rate = (paying_users / direct_users * 100) if direct_users > 0 else 0.0
+
+ # Total revenue = sum of subscription payments by campaign users
+ total_spent_by_campaign_users = sum(user_spent.get(uid, 0) for uid in c_user_ids)
+ avg_check = (total_spent_by_campaign_users // paying_users) if paying_users > 0 else 0
+
+ top_refs = [
+ TopReferrer(
+ user_id=uid,
+ username=username_map.get(uid),
+ referral_count=cnt,
+ )
+ for uid, cnt in campaign_top_referrers_raw.get(cid, [])
+ ]
+
+ campaign_nodes.append(
+ NetworkCampaignNode(
+ id=cid,
+ name=campaign.name,
+ start_parameter=campaign.start_parameter,
+ is_active=campaign.is_active,
+ direct_users=direct_users,
+ total_network_users=network_users,
+ total_revenue_kopeks=total_spent_by_campaign_users,
+ conversion_rate=round(conversion_rate, 2),
+ avg_check_kopeks=avg_check,
+ top_referrers=top_refs,
+ )
+ )
+
+ return campaign_nodes
+
+
+# ============ Endpoints ============
+
+
+@router.get('/', response_model=NetworkGraphResponse)
+async def get_referral_network(
+ admin: User = Depends(require_permission('stats:read')),
+ db: AsyncSession = Depends(get_cabinet_db),
+) -> NetworkGraphResponse:
+ """Return full referral network graph data for visualization."""
+ if await RateLimitCache.is_rate_limited(
+ admin.id,
+ 'referral_graph',
+ GRAPH_RATE_LIMIT,
+ GRAPH_RATE_WINDOW,
+ fail_closed=True,
+ ):
+ raise HTTPException(
+ status_code=status.HTTP_429_TOO_MANY_REQUESTS,
+ detail='Too many requests',
+ headers={'Retry-After': str(GRAPH_RATE_WINDOW)},
+ )
+ logger.info('Fetching referral network graph', admin_id=admin.id)
+
+ # Gather all IDs of users in the network
+ network_user_ids = await _fetch_network_user_ids(db)
+
+ if not network_user_ids:
+ return NetworkGraphResponse(
+ users=[],
+ campaigns=[],
+ edges=[],
+ total_users=0,
+ total_referrers=0,
+ total_campaigns=0,
+ total_earnings_kopeks=0,
+ )
+
+ # Cap to prevent excessive response sizes (deterministic: keep lowest IDs for stability)
+ if len(network_user_ids) > GRAPH_MAX_NODES:
+ logger.warning(
+ 'Referral network exceeds node limit, truncating',
+ total=len(network_user_ids),
+ limit=GRAPH_MAX_NODES,
+ )
+ network_user_ids = set(sorted(network_user_ids)[:GRAPH_MAX_NODES])
+
+ # Batch-fetch all aggregated data (scoped to network users)
+ referral_counts = await _fetch_direct_referral_counts(db, network_user_ids)
+ personal_revenue = await _fetch_personal_revenue(db, network_user_ids)
+ branch_revenue = await _fetch_branch_revenue(db, network_user_ids)
+ personal_spent = await _fetch_personal_spent(db, network_user_ids)
+ campaign_regs = await _fetch_campaign_registrations(db, network_user_ids)
+ sub_info = await _fetch_subscription_info(db, network_user_ids)
+
+ # Fetch actual user rows
+ users_stmt = select(User).where(User.id.in_(network_user_ids))
+ users_result = await db.execute(users_stmt)
+ users = list(users_result.scalars().all())
+
+ # Build user nodes
+ user_nodes: list[NetworkUserNode] = []
+ for user in users:
+ sub = sub_info.get(user.id, (None, None))
+ user_nodes.append(
+ _build_user_node(
+ user,
+ direct_referral_count=referral_counts.get(user.id, 0),
+ personal_revenue=personal_revenue.get(user.id, 0),
+ branch_revenue=branch_revenue.get(user.id, 0),
+ personal_spent=personal_spent.get(user.id, 0),
+ campaign_id=campaign_regs.get(user.id),
+ subscription_name=sub[0],
+ subscription_end_str=sub[1],
+ )
+ )
+
+ # Build campaign nodes
+ campaign_nodes = await _fetch_campaign_stats(db, referral_counts)
+
+ # Build edges
+ edges: list[NetworkEdge] = []
+
+ # Referral edges (only emit when both endpoints exist in the node set)
+ for user in users:
+ if user.referred_by_id is not None and user.referred_by_id in network_user_ids:
+ edges.append(
+ NetworkEdge(
+ source=f'{NODE_PREFIX_USER}{user.referred_by_id}',
+ target=f'{NODE_PREFIX_USER}{user.id}',
+ type=EDGE_TYPE_REFERRAL,
+ )
+ )
+
+ # Campaign edges
+ for user_id, campaign_id in campaign_regs.items():
+ if user_id in network_user_ids:
+ edges.append(
+ NetworkEdge(
+ source=f'{NODE_PREFIX_CAMPAIGN}{campaign_id}',
+ target=f'{NODE_PREFIX_USER}{user_id}',
+ type=EDGE_TYPE_CAMPAIGN,
+ )
+ )
+
+ # Partner ↔ Campaign edges (partner owns campaign)
+ partner_campaigns_stmt = select(
+ AdvertisingCampaign.id,
+ AdvertisingCampaign.partner_user_id,
+ ).where(AdvertisingCampaign.partner_user_id.isnot(None))
+ partner_campaigns_result = await db.execute(partner_campaigns_stmt)
+ for campaign_id, partner_user_id in partner_campaigns_result:
+ if partner_user_id in network_user_ids:
+ edges.append(
+ NetworkEdge(
+ source=f'{NODE_PREFIX_USER}{partner_user_id}',
+ target=f'{NODE_PREFIX_CAMPAIGN}{campaign_id}',
+ type=EDGE_TYPE_PARTNER_CAMPAIGN,
+ )
+ )
+
+ # Summary stats
+ total_referrers = len([u for u in user_nodes if u.direct_referrals > 0])
+
+ total_earnings = sum(personal_revenue.values())
+
+ return NetworkGraphResponse(
+ users=user_nodes,
+ campaigns=campaign_nodes,
+ edges=edges,
+ total_users=len(user_nodes),
+ total_referrers=total_referrers,
+ total_campaigns=len(campaign_nodes),
+ total_earnings_kopeks=total_earnings,
+ )
+
+
+@router.get('/scope-options', response_model=ScopeOptionsResponse)
+async def get_scope_options(
+ admin: User = Depends(require_permission('stats:read')),
+ db: AsyncSession = Depends(get_cabinet_db),
+) -> ScopeOptionsResponse:
+ """Return lightweight lists of campaigns and partners for the scope selector."""
+ if await RateLimitCache.is_rate_limited(
+ admin.id,
+ 'referral_scope_opts',
+ DETAIL_RATE_LIMIT,
+ DETAIL_RATE_WINDOW,
+ fail_closed=True,
+ ):
+ raise HTTPException(
+ status_code=status.HTTP_429_TOO_MANY_REQUESTS,
+ detail='Too many requests',
+ headers={'Retry-After': str(DETAIL_RATE_WINDOW)},
+ )
+
+ # Campaigns with registration counts
+ campaign_stmt = (
+ select(
+ AdvertisingCampaign.id,
+ AdvertisingCampaign.name,
+ AdvertisingCampaign.start_parameter,
+ AdvertisingCampaign.is_active,
+ func.count(AdvertisingCampaignRegistration.id).label('direct_users'),
+ )
+ .outerjoin(
+ AdvertisingCampaignRegistration,
+ AdvertisingCampaignRegistration.campaign_id == AdvertisingCampaign.id,
+ )
+ .group_by(AdvertisingCampaign.id)
+ .order_by(AdvertisingCampaign.name)
+ )
+ campaign_result = await db.execute(campaign_stmt)
+ campaign_options = [
+ CampaignOption(
+ id=row[0],
+ name=row[1],
+ start_parameter=row[2],
+ is_active=row[3],
+ direct_users=row[4],
+ )
+ for row in campaign_result
+ ]
+
+ # Partners with campaign counts
+ partner_stmt = (
+ select(
+ User.id,
+ User.username,
+ User.first_name,
+ User.last_name,
+ User.telegram_id,
+ User.email,
+ func.count(AdvertisingCampaign.id).label('campaign_count'),
+ )
+ .outerjoin(AdvertisingCampaign, AdvertisingCampaign.partner_user_id == User.id)
+ .where(User.partner_status == PartnerStatus.APPROVED.value)
+ .group_by(User.id)
+ .order_by(User.id)
+ )
+ partner_result = await db.execute(partner_stmt)
+ partner_options = []
+ for row in partner_result:
+ user_obj = User(
+ id=row[0],
+ username=row[1],
+ first_name=row[2],
+ last_name=row[3],
+ telegram_id=row[4],
+ email=row[5],
+ )
+ partner_options.append(
+ PartnerOption(
+ id=row[0],
+ display_name=_user_display_name(user_obj),
+ username=row[1],
+ campaign_count=row[6],
+ )
+ )
+
+ return ScopeOptionsResponse(campaigns=campaign_options, partners=partner_options)
+
+
+async def _get_descendant_user_ids(db: AsyncSession, root_ids: set[int]) -> set[int]:
+ """Get all user IDs in the referral trees rooted at root_ids (inclusive)."""
+ if not root_ids:
+ return set()
+
+ anchor = (
+ select(User.id, literal(0).label('depth')).where(User.id.in_(root_ids)).cte(name='descendants', recursive=True)
+ )
+ rpart = (
+ select(User.id, (anchor.c.depth + 1).label('depth'))
+ .join(anchor, User.referred_by_id == anchor.c.id)
+ .where(anchor.c.depth < MAX_REFERRAL_DEPTH)
+ )
+ descendants_cte = anchor.union_all(rpart)
+
+ result = await db.execute(select(func.distinct(descendants_cte.c.id)))
+ return {row[0] for row in result}
+
+
+async def _get_ancestor_user_ids(db: AsyncSession, start_user_ids: set[int]) -> set[int]:
+ """Walk up the referral chain from start_user_ids to root (inclusive)."""
+ if not start_user_ids:
+ return set()
+ anchor = (
+ select(User.id, User.referred_by_id, literal(0).label('depth'))
+ .where(User.id.in_(start_user_ids))
+ .cte(name='ancestors', recursive=True)
+ )
+ rpart = (
+ select(User.id, User.referred_by_id, (anchor.c.depth + 1).label('depth'))
+ .join(anchor, User.id == anchor.c.referred_by_id)
+ .where(anchor.c.depth < MAX_REFERRAL_DEPTH)
+ )
+ ancestors_cte = anchor.union_all(rpart)
+
+ result = await db.execute(select(func.distinct(ancestors_cte.c.id)))
+ return {row[0] for row in result}
+
+
+async def _build_scoped_graph(
+ db: AsyncSession,
+ scoped_user_ids: set[int],
+ campaign_ids: set[int],
+) -> NetworkGraphResponse:
+ """Build graph response for a scoped set of users and campaigns."""
+ if not scoped_user_ids:
+ # Still show campaign nodes even with no users
+ if campaign_ids:
+ campaign_nodes = await _fetch_campaign_stats(db, {}, campaign_ids=campaign_ids)
+ return NetworkGraphResponse(
+ users=[],
+ campaigns=campaign_nodes,
+ edges=[],
+ total_users=0,
+ total_referrers=0,
+ total_campaigns=len(campaign_nodes),
+ total_earnings_kopeks=0,
+ )
+ return NetworkGraphResponse(
+ users=[],
+ campaigns=[],
+ edges=[],
+ total_users=0,
+ total_referrers=0,
+ total_campaigns=0,
+ total_earnings_kopeks=0,
+ )
+
+ # Cap to prevent excessive response sizes
+ if len(scoped_user_ids) > GRAPH_MAX_NODES:
+ logger.warning(
+ 'Scoped referral network exceeds node limit, truncating',
+ total=len(scoped_user_ids),
+ limit=GRAPH_MAX_NODES,
+ )
+ scoped_user_ids = set(sorted(scoped_user_ids)[:GRAPH_MAX_NODES])
+
+ referral_counts = await _fetch_direct_referral_counts(db, scoped_user_ids)
+ personal_revenue = await _fetch_personal_revenue(db, scoped_user_ids)
+ branch_revenue = await _fetch_branch_revenue(db, scoped_user_ids)
+ personal_spent = await _fetch_personal_spent(db, scoped_user_ids)
+ campaign_regs = await _fetch_campaign_registrations(db, scoped_user_ids)
+ sub_info = await _fetch_subscription_info(db, scoped_user_ids)
+
+ users_result = await db.execute(select(User).where(User.id.in_(scoped_user_ids)))
+ users = list(users_result.scalars().all())
+
+ user_nodes: list[NetworkUserNode] = []
+ for user in users:
+ sub = sub_info.get(user.id, (None, None))
+ user_nodes.append(
+ _build_user_node(
+ user,
+ direct_referral_count=referral_counts.get(user.id, 0),
+ personal_revenue=personal_revenue.get(user.id, 0),
+ branch_revenue=branch_revenue.get(user.id, 0),
+ personal_spent=personal_spent.get(user.id, 0),
+ campaign_id=campaign_regs.get(user.id),
+ subscription_name=sub[0],
+ subscription_end_str=sub[1],
+ )
+ )
+
+ # Include campaigns from the scope + any campaigns users registered through
+ all_campaign_ids = campaign_ids | set(campaign_regs.values())
+ all_campaign_ids.discard(None)
+ campaign_nodes = (
+ await _fetch_campaign_stats(db, referral_counts, campaign_ids=all_campaign_ids) if all_campaign_ids else []
+ )
+
+ edges: list[NetworkEdge] = []
+
+ for user in users:
+ if user.referred_by_id is not None and user.referred_by_id in scoped_user_ids:
+ edges.append(
+ NetworkEdge(
+ source=f'{NODE_PREFIX_USER}{user.referred_by_id}',
+ target=f'{NODE_PREFIX_USER}{user.id}',
+ type=EDGE_TYPE_REFERRAL,
+ )
+ )
+
+ for user_id, cid in campaign_regs.items():
+ if user_id in scoped_user_ids and cid in all_campaign_ids:
+ edges.append(
+ NetworkEdge(
+ source=f'{NODE_PREFIX_CAMPAIGN}{cid}',
+ target=f'{NODE_PREFIX_USER}{user_id}',
+ type=EDGE_TYPE_CAMPAIGN,
+ )
+ )
+
+ partner_stmt = select(
+ AdvertisingCampaign.id,
+ AdvertisingCampaign.partner_user_id,
+ ).where(
+ AdvertisingCampaign.partner_user_id.isnot(None),
+ AdvertisingCampaign.id.in_(all_campaign_ids),
+ )
+ for cid, pid in await db.execute(partner_stmt):
+ if pid in scoped_user_ids:
+ edges.append(
+ NetworkEdge(
+ source=f'{NODE_PREFIX_USER}{pid}',
+ target=f'{NODE_PREFIX_CAMPAIGN}{cid}',
+ type=EDGE_TYPE_PARTNER_CAMPAIGN,
+ )
+ )
+
+ total_referrers = len([u for u in user_nodes if u.direct_referrals > 0])
+ total_earnings = sum(personal_revenue.values())
+
+ return NetworkGraphResponse(
+ users=user_nodes,
+ campaigns=campaign_nodes,
+ edges=edges,
+ total_users=len(user_nodes),
+ total_referrers=total_referrers,
+ total_campaigns=len(campaign_nodes),
+ total_earnings_kopeks=total_earnings,
+ )
+
+
+MAX_SCOPE_ITEMS = 50
+
+
+@router.get('/scoped', response_model=NetworkGraphResponse)
+async def get_scoped_referral_network(
+ campaign_ids: list[int] = Query(default=[], max_length=MAX_SCOPE_ITEMS),
+ partner_ids: list[int] = Query(default=[], max_length=MAX_SCOPE_ITEMS),
+ user_ids: list[int] = Query(default=[], max_length=MAX_SCOPE_ITEMS),
+ admin: User = Depends(require_permission('stats:read')),
+ db: AsyncSession = Depends(get_cabinet_db),
+) -> NetworkGraphResponse:
+ """Return scoped referral network graph for selected campaigns, partners, and/or users."""
+ if await RateLimitCache.is_rate_limited(
+ admin.id,
+ 'referral_scoped',
+ GRAPH_RATE_LIMIT,
+ GRAPH_RATE_WINDOW,
+ fail_closed=True,
+ ):
+ raise HTTPException(
+ status_code=status.HTTP_429_TOO_MANY_REQUESTS,
+ detail='Too many requests',
+ headers={'Retry-After': str(GRAPH_RATE_WINDOW)},
+ )
+
+ unique_campaign_ids = set(campaign_ids)
+ unique_partner_ids = set(partner_ids)
+ unique_user_ids = set(user_ids)
+
+ total_items = len(unique_campaign_ids) + len(unique_partner_ids) + len(unique_user_ids)
+ if total_items == 0:
+ raise HTTPException(
+ status_code=status.HTTP_400_BAD_REQUEST,
+ detail='At least one campaign, partner, or user must be selected',
+ )
+ if total_items > MAX_SCOPE_ITEMS:
+ raise HTTPException(
+ status_code=status.HTTP_400_BAD_REQUEST,
+ detail=f'Too many items selected (max {MAX_SCOPE_ITEMS})',
+ )
+
+ all_scoped_user_ids: set[int] = set()
+ all_campaign_ids: set[int] = set()
+
+ # --- Campaigns ---
+ if unique_campaign_ids:
+ existing = await db.execute(
+ select(AdvertisingCampaign.id).where(AdvertisingCampaign.id.in_(unique_campaign_ids))
+ )
+ valid_campaign_ids = {row[0] for row in existing}
+ if valid_campaign_ids:
+ all_campaign_ids |= valid_campaign_ids
+
+ campaign_reg_result = await db.execute(
+ select(AdvertisingCampaignRegistration.user_id).where(
+ AdvertisingCampaignRegistration.campaign_id.in_(valid_campaign_ids)
+ )
+ )
+ campaign_registered_ids = {row[0] for row in campaign_reg_result}
+ campaign_descendant_ids = await _get_descendant_user_ids(db, campaign_registered_ids)
+ all_scoped_user_ids |= campaign_descendant_ids
+
+ # --- Partners ---
+ if unique_partner_ids:
+ partner_result = await db.execute(
+ select(User.id).where(
+ User.id.in_(unique_partner_ids),
+ User.partner_status == PartnerStatus.APPROVED.value,
+ )
+ )
+ valid_partner_ids = {row[0] for row in partner_result}
+ if valid_partner_ids:
+ partner_campaigns_result = await db.execute(
+ select(AdvertisingCampaign.id).where(AdvertisingCampaign.partner_user_id.in_(valid_partner_ids))
+ )
+ partner_campaign_set = {row[0] for row in partner_campaigns_result}
+ all_campaign_ids |= partner_campaign_set
+
+ partner_registered_ids: set[int] = set()
+ if partner_campaign_set:
+ partner_reg_result = await db.execute(
+ select(AdvertisingCampaignRegistration.user_id).where(
+ AdvertisingCampaignRegistration.campaign_id.in_(partner_campaign_set)
+ )
+ )
+ partner_registered_ids = {row[0] for row in partner_reg_result}
+
+ partner_descendant_ids = await _get_descendant_user_ids(db, partner_registered_ids | valid_partner_ids)
+ all_scoped_user_ids |= partner_descendant_ids
+
+ # --- Users ---
+ if unique_user_ids:
+ user_result = await db.execute(select(User.id).where(User.id.in_(unique_user_ids)))
+ valid_user_ids = {row[0] for row in user_result}
+ if valid_user_ids:
+ ancestor_ids = await _get_ancestor_user_ids(db, valid_user_ids)
+ descendant_ids = await _get_descendant_user_ids(db, valid_user_ids)
+ all_scoped_user_ids |= ancestor_ids | descendant_ids
+
+ # Fail only if ALL provided IDs were invalid
+ if not all_scoped_user_ids and not all_campaign_ids:
+ raise HTTPException(
+ status_code=status.HTTP_404_NOT_FOUND,
+ detail='No valid items found for the provided IDs',
+ )
+
+ # Discover campaigns that scoped users registered through (runs for all scopes)
+ if all_scoped_user_ids:
+ scope_campaign_regs = await _fetch_campaign_registrations(db, all_scoped_user_ids)
+ relevant_campaigns = {cid for cid in scope_campaign_regs.values() if cid is not None}
+ all_campaign_ids |= relevant_campaigns
+
+ return await _build_scoped_graph(db, all_scoped_user_ids, all_campaign_ids)
+
+
+@router.get('/user/{user_id}', response_model=NetworkUserDetail)
+async def get_network_user_detail(
+ user_id: int,
+ admin: User = Depends(require_permission('stats:read')),
+ db: AsyncSession = Depends(get_cabinet_db),
+) -> NetworkUserDetail:
+ """Return detailed info about a specific user in the referral network."""
+ if await RateLimitCache.is_rate_limited(
+ admin.id,
+ 'referral_user_detail',
+ DETAIL_RATE_LIMIT,
+ DETAIL_RATE_WINDOW,
+ fail_closed=True,
+ ):
+ raise HTTPException(
+ status_code=status.HTTP_429_TOO_MANY_REQUESTS,
+ detail='Too many requests',
+ headers={'Retry-After': str(DETAIL_RATE_WINDOW)},
+ )
+ logger.info('Fetching network user detail', admin_id=admin.id, target_user_id=user_id)
+
+ # Fetch user with subscription eagerly loaded
+ stmt = (
+ select(User)
+ .options(selectinload(User.subscription).selectinload(Subscription.tariff))
+ .where(User.id == user_id)
+ )
+ result = await db.execute(stmt)
+ user = result.scalar_one_or_none()
+
+ if user is None:
+ raise HTTPException(
+ status_code=status.HTTP_404_NOT_FOUND,
+ detail='User not found',
+ )
+
+ # Direct referral count
+ ref_count_stmt = select(func.count(User.id)).where(User.referred_by_id == user_id)
+ ref_count_result = await db.execute(ref_count_stmt)
+ direct_referrals = ref_count_result.scalar() or 0
+
+ # Personal revenue
+ personal_rev_stmt = select(func.coalesce(func.sum(ReferralEarning.amount_kopeks), 0)).where(
+ ReferralEarning.user_id == user_id
+ )
+ personal_rev_result = await db.execute(personal_rev_stmt)
+ personal_revenue = personal_rev_result.scalar() or 0
+
+ # Branch revenue: computed after branch CTE (see below)
+ branch_revenue = 0
+
+ # Personal spent
+ spent_stmt = select(func.coalesce(func.sum(Transaction.amount_kopeks), 0)).where(
+ and_(
+ Transaction.user_id == user_id,
+ Transaction.type.in_(SPENT_TRANSACTION_TYPES),
+ Transaction.is_completed.is_(True),
+ )
+ )
+ spent_result = await db.execute(spent_stmt)
+ personal_spent = spent_result.scalar() or 0
+
+ # Campaign registration
+ campaign_reg_stmt = (
+ select(AdvertisingCampaignRegistration.campaign_id)
+ .where(AdvertisingCampaignRegistration.user_id == user_id)
+ .order_by(AdvertisingCampaignRegistration.created_at.asc())
+ .limit(1)
+ )
+ campaign_reg_result = await db.execute(campaign_reg_stmt)
+ campaign_id = campaign_reg_result.scalar_one_or_none()
+
+ # Campaign name
+ campaign_name: str | None = None
+ if campaign_id is not None:
+ camp_stmt = select(AdvertisingCampaign.name).where(AdvertisingCampaign.id == campaign_id)
+ camp_result = await db.execute(camp_stmt)
+ campaign_name = camp_result.scalar_one_or_none()
+
+ # Total branch users via recursive CTE (with depth limit to prevent cycles)
+ base = (
+ select(User.id, literal(1).label('depth'))
+ .where(User.referred_by_id == user_id)
+ .cte(name='branch', recursive=True)
+ )
+ recursive_part = (
+ select(User.id, (base.c.depth + 1).label('depth'))
+ .join(base, User.referred_by_id == base.c.id)
+ .where(base.c.depth < MAX_REFERRAL_DEPTH)
+ )
+ branch_cte = base.union_all(recursive_part) # UNION ALL + count(distinct) is faster than UNION
+ total_branch_stmt = select(func.count(func.distinct(branch_cte.c.id))).select_from(branch_cte)
+ total_branch_result = await db.execute(total_branch_stmt)
+ total_branch_users = total_branch_result.scalar() or 0
+
+ # Branch revenue: total spent by all users in the branch
+ branch_user_ids_stmt = select(branch_cte.c.id)
+ branch_rev_stmt = select(func.coalesce(func.sum(Transaction.amount_kopeks), 0)).where(
+ and_(
+ Transaction.user_id.in_(branch_user_ids_stmt),
+ Transaction.type.in_(SPENT_TRANSACTION_TYPES),
+ Transaction.is_completed.is_(True),
+ )
+ )
+ branch_rev_result = await db.execute(branch_rev_stmt)
+ branch_revenue = branch_rev_result.scalar() or 0
+
+ # Referrer info
+ referrer_display_name: str | None = None
+ if user.referred_by_id is not None:
+ referrer_stmt = select(User).where(User.id == user.referred_by_id)
+ referrer_result = await db.execute(referrer_stmt)
+ referrer = referrer_result.scalar_one_or_none()
+ if referrer is not None:
+ referrer_display_name = _user_display_name(referrer)
+
+ # Subscription info
+ subscription_name: str | None = None
+ subscription_end: str | None = None
+ if user.subscription is not None:
+ if user.subscription.tariff is not None:
+ subscription_name = user.subscription.tariff.name
+ subscription_end = _format_datetime(user.subscription.end_date)
+
+ return NetworkUserDetail(
+ id=user.id,
+ tg_id=user.telegram_id,
+ username=user.username,
+ email=user.email,
+ display_name=_user_display_name(user),
+ is_partner=user.partner_status == PartnerStatus.APPROVED.value,
+ referrer_id=user.referred_by_id,
+ referrer_display_name=referrer_display_name,
+ campaign_id=campaign_id,
+ campaign_name=campaign_name,
+ direct_referrals=direct_referrals,
+ total_branch_users=total_branch_users,
+ branch_revenue_kopeks=branch_revenue,
+ personal_revenue_kopeks=personal_revenue,
+ personal_spent_kopeks=personal_spent,
+ subscription_name=subscription_name,
+ subscription_end=subscription_end,
+ registered_at=_format_datetime(user.created_at),
+ )
+
+
+@router.get('/campaign/{campaign_id}', response_model=NetworkCampaignDetail)
+async def get_network_campaign_detail(
+ campaign_id: int,
+ admin: User = Depends(require_permission('stats:read')),
+ db: AsyncSession = Depends(get_cabinet_db),
+) -> NetworkCampaignDetail:
+ """Return detailed info about a specific advertising campaign."""
+ if await RateLimitCache.is_rate_limited(
+ admin.id,
+ 'referral_campaign_detail',
+ DETAIL_RATE_LIMIT,
+ DETAIL_RATE_WINDOW,
+ fail_closed=True,
+ ):
+ raise HTTPException(
+ status_code=status.HTTP_429_TOO_MANY_REQUESTS,
+ detail='Too many requests',
+ headers={'Retry-After': str(DETAIL_RATE_WINDOW)},
+ )
+ logger.info('Fetching network campaign detail', admin_id=admin.id, campaign_id=campaign_id)
+
+ # Fetch campaign
+ stmt = select(AdvertisingCampaign).where(AdvertisingCampaign.id == campaign_id)
+ result = await db.execute(stmt)
+ campaign = result.scalar_one_or_none()
+
+ if campaign is None:
+ raise HTTPException(
+ status_code=status.HTTP_404_NOT_FOUND,
+ detail='Campaign not found',
+ )
+
+ # Registration count
+ reg_count_stmt = select(func.count(AdvertisingCampaignRegistration.id)).where(
+ AdvertisingCampaignRegistration.campaign_id == campaign_id
+ )
+ reg_result = await db.execute(reg_count_stmt)
+ direct_users = reg_result.scalar() or 0
+
+ # User IDs from this campaign
+ user_ids_stmt = select(AdvertisingCampaignRegistration.user_id).where(
+ AdvertisingCampaignRegistration.campaign_id == campaign_id
+ )
+ user_ids_result = await db.execute(user_ids_stmt)
+ campaign_user_ids = [row[0] for row in user_ids_result]
+
+ # Referral counts for campaign users
+ referral_counts: dict[int, int] = {}
+ total_network_users = direct_users
+ if campaign_user_ids:
+ ref_stmt = (
+ select(User.referred_by_id, func.count(User.id))
+ .where(User.referred_by_id.in_(campaign_user_ids))
+ .group_by(User.referred_by_id)
+ )
+ ref_result = await db.execute(ref_stmt)
+ referral_counts = {row[0]: row[1] for row in ref_result}
+ total_network_users += sum(referral_counts.values())
+
+ # Spending by campaign users (for conversion, avg check, and total revenue)
+ paying_users = 0
+ total_spent = 0
+ if campaign_user_ids:
+ spent_stmt = (
+ select(Transaction.user_id, func.coalesce(func.sum(Transaction.amount_kopeks), 0))
+ .where(
+ and_(
+ Transaction.user_id.in_(campaign_user_ids),
+ Transaction.type.in_(SPENT_TRANSACTION_TYPES),
+ Transaction.is_completed.is_(True),
+ )
+ )
+ .group_by(Transaction.user_id)
+ )
+ spent_result = await db.execute(spent_stmt)
+ for row in spent_result:
+ if row[1] > 0:
+ paying_users += 1
+ total_spent += row[1]
+
+ conversion_rate = (paying_users / direct_users * 100) if direct_users > 0 else 0.0
+ avg_check = (total_spent // paying_users) if paying_users > 0 else 0
+
+ # Top referrers from this campaign
+ scored = [(uid, referral_counts.get(uid, 0)) for uid in campaign_user_ids if referral_counts.get(uid, 0) > 0]
+ scored.sort(key=lambda x: x[1], reverse=True)
+ top = scored[:TOP_REFERRERS_LIMIT]
+
+ top_user_ids = [uid for uid, _ in top]
+ username_map: dict[int, str | None] = {}
+ if top_user_ids:
+ uname_stmt = select(User.id, User.username).where(User.id.in_(top_user_ids))
+ uname_result = await db.execute(uname_stmt)
+ username_map = {row[0]: row[1] for row in uname_result}
+
+ top_referrers = [
+ TopReferrer(
+ user_id=uid,
+ username=username_map.get(uid),
+ referral_count=cnt,
+ )
+ for uid, cnt in top
+ ]
+
+ return NetworkCampaignDetail(
+ id=campaign.id,
+ name=campaign.name,
+ start_parameter=campaign.start_parameter,
+ is_active=campaign.is_active,
+ direct_users=direct_users,
+ total_network_users=total_network_users,
+ total_revenue_kopeks=total_spent,
+ conversion_rate=round(conversion_rate, 2),
+ avg_check_kopeks=avg_check,
+ top_referrers=top_referrers,
+ )
+
+
+@router.get('/search', response_model=NetworkSearchResult)
+async def search_referral_network(
+ q: str = Query(..., min_length=1, max_length=200, description='Search query'),
+ admin: User = Depends(require_permission('stats:read')),
+ db: AsyncSession = Depends(get_cabinet_db),
+) -> NetworkSearchResult:
+ """Search users and campaigns in the referral network by telegram_id, username, email, or campaign name."""
+ if await RateLimitCache.is_rate_limited(
+ admin.id,
+ 'referral_search',
+ SEARCH_RATE_LIMIT,
+ SEARCH_RATE_WINDOW,
+ fail_closed=True,
+ ):
+ raise HTTPException(
+ status_code=status.HTTP_429_TOO_MANY_REQUESTS,
+ detail='Too many requests',
+ headers={'Retry-After': str(SEARCH_RATE_WINDOW)},
+ )
+ logger.info('Searching referral network', admin_id=admin.id, query_len=len(q))
+
+ query_stripped = q.strip()
+ escaped_query = _escape_like(query_stripped)
+
+ # Build user search conditions (with escaped LIKE wildcards)
+ user_conditions = [
+ User.username.ilike(f'%{escaped_query}%', escape='\\'),
+ User.email.ilike(f'%{escaped_query}%', escape='\\'),
+ ]
+
+ # If query is numeric, also search by telegram_id and user id
+ if query_stripped.isdigit():
+ numeric_val = int(query_stripped)
+ user_conditions.append(User.telegram_id == numeric_val)
+ user_conditions.append(User.id == numeric_val)
+
+ # Find matching users that are part of the referral network
+ network_user_ids = await _fetch_network_user_ids(db)
+
+ user_stmt = (
+ select(User)
+ .where(
+ and_(
+ User.id.in_(network_user_ids) if network_user_ids else literal(False),
+ or_(*user_conditions),
+ )
+ )
+ .limit(SEARCH_RESULTS_LIMIT)
+ )
+ user_result = await db.execute(user_stmt)
+ matched_users = list(user_result.scalars().all())
+
+ # Batch-fetch data for matched users
+ matched_ids = {u.id for u in matched_users}
+ user_nodes: list[NetworkUserNode] = []
+
+ # Fetch referral counts scoped to matched users
+ referral_counts = await _fetch_direct_referral_counts(db, matched_ids) if matched_ids else {}
+
+ if matched_ids:
+ personal_revenue = await _fetch_personal_revenue(db, matched_ids)
+ branch_revenue = await _fetch_branch_revenue(db, matched_ids)
+ personal_spent = await _fetch_personal_spent(db, matched_ids)
+ campaign_regs = await _fetch_campaign_registrations(db, matched_ids)
+ sub_info = await _fetch_subscription_info(db, matched_ids)
+
+ for user in matched_users:
+ sub = sub_info.get(user.id, (None, None))
+ user_nodes.append(
+ _build_user_node(
+ user,
+ direct_referral_count=referral_counts.get(user.id, 0),
+ personal_revenue=personal_revenue.get(user.id, 0),
+ branch_revenue=branch_revenue.get(user.id, 0),
+ personal_spent=personal_spent.get(user.id, 0),
+ campaign_id=campaign_regs.get(user.id),
+ subscription_name=sub[0],
+ subscription_end_str=sub[1],
+ )
+ )
+
+ # Search campaigns (with escaped LIKE wildcards)
+ campaign_stmt = (
+ select(AdvertisingCampaign)
+ .where(
+ or_(
+ AdvertisingCampaign.name.ilike(f'%{escaped_query}%', escape='\\'),
+ AdvertisingCampaign.start_parameter.ilike(f'%{escaped_query}%', escape='\\'),
+ )
+ )
+ .limit(SEARCH_RESULTS_LIMIT)
+ )
+ campaign_result = await db.execute(campaign_stmt)
+ matched_campaigns = list(campaign_result.scalars().all())
+
+ # Batch campaign stats instead of N+1 queries per campaign
+ campaign_nodes: list[NetworkCampaignNode] = []
+ if matched_campaigns:
+ matched_campaign_ids = [c.id for c in matched_campaigns]
+
+ # Batch: registration counts per campaign
+ reg_count_stmt = (
+ select(
+ AdvertisingCampaignRegistration.campaign_id,
+ func.count(AdvertisingCampaignRegistration.id),
+ )
+ .where(AdvertisingCampaignRegistration.campaign_id.in_(matched_campaign_ids))
+ .group_by(AdvertisingCampaignRegistration.campaign_id)
+ )
+ reg_res = await db.execute(reg_count_stmt)
+ reg_counts: dict[int, int] = {row[0]: row[1] for row in reg_res}
+
+ # Batch: user IDs per campaign
+ user_campaign_stmt = select(
+ AdvertisingCampaignRegistration.campaign_id,
+ AdvertisingCampaignRegistration.user_id,
+ ).where(AdvertisingCampaignRegistration.campaign_id.in_(matched_campaign_ids))
+ uc_res = await db.execute(user_campaign_stmt)
+ campaign_user_map: dict[int, list[int]] = defaultdict(list)
+ for row in uc_res:
+ campaign_user_map[row[0]].append(row[1])
+
+ # Fetch referral counts and spending scoped to campaign users
+ all_campaign_user_ids: set[int] = set()
+ for uids in campaign_user_map.values():
+ all_campaign_user_ids.update(uids)
+
+ # Batch: spending per user (for campaign revenue)
+ campaign_user_spent: dict[int, int] = {}
+ if all_campaign_user_ids:
+ spent_stmt = (
+ select(Transaction.user_id, func.coalesce(func.sum(Transaction.amount_kopeks), 0))
+ .where(
+ and_(
+ Transaction.user_id.in_(all_campaign_user_ids),
+ Transaction.type.in_(SPENT_TRANSACTION_TYPES),
+ Transaction.is_completed.is_(True),
+ )
+ )
+ .group_by(Transaction.user_id)
+ )
+ spent_res = await db.execute(spent_stmt)
+ campaign_user_spent = {row[0]: row[1] for row in spent_res}
+ campaign_referral_counts = (
+ await _fetch_direct_referral_counts(db, all_campaign_user_ids) if all_campaign_user_ids else {}
+ )
+
+ for campaign in matched_campaigns:
+ cid = campaign.id
+ direct_users = reg_counts.get(cid, 0)
+ c_user_ids = campaign_user_map.get(cid, [])
+ network_users = direct_users + sum(campaign_referral_counts.get(uid, 0) for uid in c_user_ids)
+ total_revenue = sum(campaign_user_spent.get(uid, 0) for uid in c_user_ids)
+
+ campaign_nodes.append(
+ NetworkCampaignNode(
+ id=cid,
+ name=campaign.name,
+ start_parameter=campaign.start_parameter,
+ is_active=campaign.is_active,
+ direct_users=direct_users,
+ total_network_users=network_users,
+ total_revenue_kopeks=total_revenue,
+ conversion_rate=0.0,
+ avg_check_kopeks=0,
+ top_referrers=[],
+ )
+ )
+
+ return NetworkSearchResult(
+ users=user_nodes,
+ campaigns=campaign_nodes,
+ )
diff --git a/app/cabinet/routes/admin_tickets.py b/app/cabinet/routes/admin_tickets.py
index 16a3dedc..3778f42b 100644
--- a/app/cabinet/routes/admin_tickets.py
+++ b/app/cabinet/routes/admin_tickets.py
@@ -5,7 +5,7 @@ from datetime import UTC, datetime
import structlog
from fastapi import APIRouter, Depends, HTTPException, Query, status
-from pydantic import BaseModel, Field
+from pydantic import BaseModel, Field, model_validator
from sqlalchemy import desc, func, select
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.orm import selectinload
@@ -90,6 +90,19 @@ class AdminReplyRequest(BaseModel):
"""Admin reply to ticket."""
message: str = Field(..., min_length=1, max_length=4000, description='Reply message')
+ media_type: str | None = Field(None, description='Media type: photo, video, or document')
+ media_file_id: str | None = Field(None, max_length=255, description='Telegram file_id from media upload')
+ media_caption: str | None = Field(None, max_length=1000, description='Caption for media')
+
+ @model_validator(mode='after')
+ def validate_media_fields(self) -> 'AdminReplyRequest':
+ if self.media_file_id and not self.media_type:
+ raise ValueError('media_type is required when media_file_id is provided')
+ if self.media_type and not self.media_file_id:
+ raise ValueError('media_file_id is required when media_type is provided')
+ if self.media_type and self.media_type not in {'photo', 'video', 'document'}:
+ raise ValueError('media_type must be one of: photo, video, document')
+ return self
class AdminStatusUpdateRequest(BaseModel):
@@ -443,11 +456,16 @@ async def reply_to_ticket(
)
# Create admin message
+ has_media = bool(request.media_file_id)
message = TicketMessage(
ticket_id=ticket.id,
user_id=ticket.user_id,
message_text=request.message,
is_from_admin=True,
+ has_media=has_media,
+ media_type=request.media_type if has_media else None,
+ media_file_id=request.media_file_id if has_media else None,
+ media_caption=request.media_caption if has_media else None,
created_at=datetime.now(UTC),
)
db.add(message)
diff --git a/app/database/crud/transaction.py b/app/database/crud/transaction.py
index 2f5254e4..0338f9ce 100644
--- a/app/database/crud/transaction.py
+++ b/app/database/crud/transaction.py
@@ -108,7 +108,7 @@ async def create_transaction(
await maybe_assign_promo_group_by_total_spent(db, user_id)
except Exception as exc:
- logger.debug('Не удалось проверить автовыдачу промогруппы для пользователя', user_id=user_id, exc=exc)
+ logger.warning('Не удалось проверить автовыдачу промогруппы для пользователя', user_id=user_id, exc=exc)
if type == TransactionType.SUBSCRIPTION_PAYMENT and is_completed:
try:
from app.services.referral_contest_service import referral_contest_service
@@ -168,7 +168,7 @@ async def emit_transaction_side_effects(
await maybe_assign_promo_group_by_total_spent(db, user_id)
except Exception as exc:
- logger.debug('Не удалось проверить автовыдачу промогруппы для пользователя', user_id=user_id, exc=exc)
+ logger.warning('Не удалось проверить автовыдачу промогруппы для пользователя', user_id=user_id, exc=exc)
if type == TransactionType.SUBSCRIPTION_PAYMENT and is_completed:
try:
@@ -253,7 +253,7 @@ async def complete_transaction(db: AsyncSession, transaction: Transaction) -> Tr
await maybe_assign_promo_group_by_total_spent(db, transaction.user_id)
except Exception as exc:
- logger.debug(
+ logger.warning(
'Не удалось проверить автовыдачу промогруппы для пользователя', user_id=transaction.user_id, exc=exc
)
diff --git a/app/database/crud/user_promo_group.py b/app/database/crud/user_promo_group.py
index 9140b115..98329106 100644
--- a/app/database/crud/user_promo_group.py
+++ b/app/database/crud/user_promo_group.py
@@ -3,7 +3,7 @@
from datetime import UTC, datetime
import structlog
-from sqlalchemy import and_, desc, select
+from sqlalchemy import and_, desc, func, select
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.orm import selectinload
@@ -24,7 +24,7 @@ async def _sync_user_primary_promo_group(
select(UserPromoGroup.promo_group_id)
.join(PromoGroup, UserPromoGroup.promo_group_id == PromoGroup.id)
.where(UserPromoGroup.user_id == user_id)
- .order_by(desc(PromoGroup.priority), PromoGroup.id)
+ .order_by(desc(PromoGroup.priority), desc(PromoGroup.id))
)
first = result.first()
@@ -53,7 +53,12 @@ async def sync_user_primary_promo_group(
async def add_user_to_promo_group(
- db: AsyncSession, user_id: int, promo_group_id: int, assigned_by: str = 'admin'
+ db: AsyncSession,
+ user_id: int,
+ promo_group_id: int,
+ assigned_by: str = 'admin',
+ *,
+ commit: bool = True,
) -> UserPromoGroup | None:
"""
Добавляет пользователю промогруппу.
@@ -63,6 +68,7 @@ async def add_user_to_promo_group(
user_id: ID пользователя
promo_group_id: ID промогруппы
assigned_by: Кто назначил ('admin', 'system', 'auto', 'promocode')
+ commit: Коммитить транзакцию (False для батчевых операций)
Returns:
UserPromoGroup или None если уже существует
@@ -85,8 +91,9 @@ async def add_user_to_promo_group(
await _sync_user_primary_promo_group(db, user_id)
- await db.commit()
- await db.refresh(user_promo_group)
+ if commit:
+ await db.commit()
+ await db.refresh(user_promo_group)
logger.info(
'Пользователю добавлена промогруппа',
@@ -98,11 +105,19 @@ async def add_user_to_promo_group(
except Exception as error:
logger.error('Ошибка добавления промогруппы пользователю', error=error)
- await db.rollback()
- return None
+ if commit:
+ await db.rollback()
+ return None
+ raise
-async def remove_user_from_promo_group(db: AsyncSession, user_id: int, promo_group_id: int) -> bool:
+async def remove_user_from_promo_group(
+ db: AsyncSession,
+ user_id: int,
+ promo_group_id: int,
+ *,
+ commit: bool = True,
+) -> bool:
"""
Удаляет промогруппу у пользователя.
@@ -110,6 +125,7 @@ async def remove_user_from_promo_group(db: AsyncSession, user_id: int, promo_gro
db: Сессия БД
user_id: ID пользователя
promo_group_id: ID промогруппы
+ commit: Коммитить транзакцию (False для батчевых операций)
Returns:
True если удалено, False если связи не было
@@ -133,15 +149,18 @@ async def remove_user_from_promo_group(db: AsyncSession, user_id: int, promo_gro
await _sync_user_primary_promo_group(db, user_id)
- await db.commit()
+ if commit:
+ await db.commit()
logger.info('У пользователя удалена промогруппа', user_id=user_id, promo_group_id=promo_group_id)
return True
except Exception as error:
logger.error('Ошибка удаления промогруппы у пользователя', error=error)
- await db.rollback()
- return False
+ if commit:
+ await db.rollback()
+ return False
+ raise
async def get_user_promo_groups(db: AsyncSession, user_id: int) -> list[UserPromoGroup]:
@@ -155,19 +174,14 @@ async def get_user_promo_groups(db: AsyncSession, user_id: int) -> list[UserProm
Returns:
Список UserPromoGroup с загруженными PromoGroup, отсортированный по приоритету DESC
"""
- try:
- result = await db.execute(
- select(UserPromoGroup)
- .options(selectinload(UserPromoGroup.promo_group))
- .where(UserPromoGroup.user_id == user_id)
- .join(PromoGroup, UserPromoGroup.promo_group_id == PromoGroup.id)
- .order_by(desc(PromoGroup.priority), PromoGroup.id)
- )
- return list(result.scalars().all())
-
- except Exception as error:
- logger.error('Ошибка получения промогрупп пользователя', user_id=user_id, error=error)
- return []
+ result = await db.execute(
+ select(UserPromoGroup)
+ .options(selectinload(UserPromoGroup.promo_group))
+ .where(UserPromoGroup.user_id == user_id)
+ .join(PromoGroup, UserPromoGroup.promo_group_id == PromoGroup.id)
+ .order_by(desc(PromoGroup.priority), desc(PromoGroup.id))
+ )
+ return list(result.scalars().all())
async def get_primary_user_promo_group(db: AsyncSession, user_id: int) -> PromoGroup | None:
@@ -181,19 +195,14 @@ async def get_primary_user_promo_group(db: AsyncSession, user_id: int) -> PromoG
Returns:
PromoGroup с максимальным приоритетом или None
"""
- try:
- user_promo_groups = await get_user_promo_groups(db, user_id)
+ user_promo_groups = await get_user_promo_groups(db, user_id)
- if not user_promo_groups:
- return None
-
- # Первая в списке имеет максимальный приоритет (список уже отсортирован)
- return user_promo_groups[0].promo_group or None
-
- except Exception as error:
- logger.error('Ошибка получения primary промогруппы пользователя', user_id=user_id, error=error)
+ if not user_promo_groups:
return None
+ # Первая в списке имеет максимальный приоритет (список уже отсортирован)
+ return user_promo_groups[0].promo_group or None
+
async def has_user_promo_group(db: AsyncSession, user_id: int, promo_group_id: int) -> bool:
"""
@@ -207,17 +216,12 @@ async def has_user_promo_group(db: AsyncSession, user_id: int, promo_group_id: i
Returns:
True если пользователь уже имеет эту промогруппу
"""
- try:
- result = await db.execute(
- select(UserPromoGroup).where(
- and_(UserPromoGroup.user_id == user_id, UserPromoGroup.promo_group_id == promo_group_id)
- )
+ result = await db.execute(
+ select(UserPromoGroup).where(
+ and_(UserPromoGroup.user_id == user_id, UserPromoGroup.promo_group_id == promo_group_id)
)
- return result.scalar_one_or_none() is not None
-
- except Exception as error:
- logger.error('Ошибка проверки промогруппы пользователя', error=error)
- return False
+ )
+ return result.scalar_one_or_none() is not None
async def count_user_promo_groups(db: AsyncSession, user_id: int) -> int:
@@ -232,8 +236,10 @@ async def count_user_promo_groups(db: AsyncSession, user_id: int) -> int:
Количество промогрупп
"""
try:
- result = await db.execute(select(UserPromoGroup).where(UserPromoGroup.user_id == user_id))
- return len(list(result.scalars().all()))
+ result = await db.execute(
+ select(func.count()).select_from(UserPromoGroup).where(UserPromoGroup.user_id == user_id)
+ )
+ return result.scalar_one()
except Exception as error:
logger.error('Ошибка подсчета промогрупп пользователя', error=error)
@@ -257,15 +263,18 @@ async def replace_user_promo_groups(
"""
try:
# Удаляем все текущие промогруппы
- await db.execute(select(UserPromoGroup).where(UserPromoGroup.user_id == user_id))
result = await db.execute(select(UserPromoGroup).where(UserPromoGroup.user_id == user_id))
for upg in result.scalars().all():
await db.delete(upg)
+ await db.flush()
# Добавляем новые
for promo_group_id in promo_group_ids:
user_promo_group = UserPromoGroup(user_id=user_id, promo_group_id=promo_group_id, assigned_by=assigned_by)
db.add(user_promo_group)
+ await db.flush()
+
+ await _sync_user_primary_promo_group(db, user_id)
await db.commit()
logger.info('Промогруппы пользователя заменены на', user_id=user_id, promo_group_ids=promo_group_ids)
diff --git a/app/database/models.py b/app/database/models.py
index 24f63d65..0bf958eb 100644
--- a/app/database/models.py
+++ b/app/database/models.py
@@ -1574,6 +1574,7 @@ class Transaction(Base):
Index('ix_transactions_type_created_completed', 'type', 'created_at', 'is_completed'),
Index('ix_transactions_user_created', 'user_id', 'created_at'),
Index('ix_transactions_type_method_created', 'type', 'payment_method', 'created_at'),
+ Index('ix_transactions_user_type_completed_amount', 'user_id', 'type', 'is_completed', 'amount_kopeks'),
)
id = Column(Integer, primary_key=True, index=True)
@@ -2490,7 +2491,10 @@ class AdvertisingCampaign(Base):
class AdvertisingCampaignRegistration(Base):
__tablename__ = 'advertising_campaign_registrations'
- __table_args__ = (UniqueConstraint('campaign_id', 'user_id', name='uq_campaign_user'),)
+ __table_args__ = (
+ UniqueConstraint('campaign_id', 'user_id', name='uq_campaign_user'),
+ Index('ix_campaign_reg_user_created', 'user_id', 'created_at'),
+ )
id = Column(Integer, primary_key=True, index=True)
campaign_id = Column(Integer, ForeignKey('advertising_campaigns.id', ondelete='CASCADE'), nullable=False)
@@ -3283,6 +3287,7 @@ class GuestPurchase(Base):
cabinet_password = Column(Text, nullable=True)
auto_login_token = Column(Text, nullable=True)
recipient_warning = Column(String(50), nullable=True)
+ retry_count = Column(Integer, nullable=False, default=0, server_default='0')
landing = relationship('LandingPage', back_populates='guest_purchases', lazy='selectin')
tariff = relationship('Tariff', lazy='selectin')
diff --git a/app/handlers/stars_payments.py b/app/handlers/stars_payments.py
index a0c8a066..a7105a1e 100644
--- a/app/handlers/stars_payments.py
+++ b/app/handlers/stars_payments.py
@@ -354,10 +354,6 @@ async def _handle_guest_purchase_payment(
stars_amount=stars_amount,
purchase_token_prefix=purchase_token[:5],
)
- elif result is False:
- await message.answer(
- '❌ Произошла ошибка при обработке подарочной подписки. Обратитесь в поддержку.',
- )
else:
logger.error('try_fulfill_guest_purchase returned None for Stars gift', payload=payload)
await message.answer('❌ Ошибка обработки платежа. Обратитесь в поддержку.')
diff --git a/app/services/guest_purchase_service.py b/app/services/guest_purchase_service.py
index d20c16d2..e5fadc5f 100644
--- a/app/services/guest_purchase_service.py
+++ b/app/services/guest_purchase_service.py
@@ -7,7 +7,7 @@ from datetime import UTC, datetime, timedelta
from typing import Literal
import structlog
-from sqlalchemy import func, or_, select
+from sqlalchemy import func, or_, select, update
from sqlalchemy.exc import IntegrityError
from sqlalchemy.ext.asyncio import AsyncSession
@@ -392,24 +392,41 @@ async def fulfill_purchase(
return purchase
+def _resolve_base_payment_method(method_str: str | None) -> str:
+ """Resolve base payment method string by stripping sub-option suffixes.
+
+ 'yookassa_sbp' → 'yookassa', 'kassa_ai' → 'kassa_ai' (enum match keeps it),
+ 'platega_2' → 'platega'.
+ """
+ if not method_str:
+ return ''
+ # If exact enum match, return as-is (handles 'telegram_stars', 'kassa_ai', etc.)
+ try:
+ PaymentMethod(method_str)
+ return method_str
+ except ValueError:
+ pass
+ # Strip sub-option suffix
+ if '_' in method_str:
+ base = method_str.rsplit('_', 1)[0]
+ try:
+ PaymentMethod(base)
+ return base
+ except ValueError:
+ pass
+ return method_str
+
+
def _resolve_payment_method(method_str: str | None) -> PaymentMethod | None:
"""Convert payment method string from GuestPurchase to PaymentMethod enum."""
if not method_str:
return None
- # Try exact match first (handles 'telegram_stars', 'kassa_ai', 'yookassa', etc.)
+ base = _resolve_base_payment_method(method_str)
try:
- return PaymentMethod(method_str)
+ return PaymentMethod(base)
except ValueError:
- pass
- # Strip sub-option suffix ('yookassa_sbp' → 'yookassa', 'platega_2' → 'platega')
- if '_' in method_str:
- base_method = method_str.split('_')[0]
- try:
- return PaymentMethod(base_method)
- except ValueError:
- pass
- logger.debug('Unknown payment method for transaction', method=method_str)
- return None
+ logger.debug('Unknown payment method for transaction', method=method_str)
+ return None
def _mask_email(email: str) -> str:
@@ -1002,14 +1019,16 @@ async def retry_stuck_paid_purchases(
stale_minutes: int = 5,
limit: int = 10,
max_age_hours: int = 24,
+ max_retries: int = 20,
) -> 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.
+ (but not older than max_age_hours, and with retry_count < max_retries) and
+ attempts to fulfill them in isolated sessions.
- Purchases older than max_age_hours are left for manual investigation.
+ Purchases exceeding max_retries are marked FAILED and an admin alert is sent.
+ Returns the number of successfully retried purchases.
"""
from app.database.database import AsyncSessionLocal
@@ -1018,10 +1037,12 @@ async def retry_stuck_paid_purchases(
# Collect tokens only — each retry gets its own session.
# NULL paid_at is included via or_() as a safety net for data anomalies.
+ # Filter retry_count < max_retries in SQL to avoid wasting LIMIT slots.
result = await db.execute(
select(GuestPurchase.token)
.where(
GuestPurchase.status == GuestPurchaseStatus.PAID.value,
+ GuestPurchase.retry_count < max_retries,
or_(GuestPurchase.paid_at < cutoff, GuestPurchase.paid_at.is_(None)),
or_(GuestPurchase.paid_at > max_age, GuestPurchase.paid_at.is_(None)),
# Exclude code-only gifts — they stay PAID intentionally until activated
@@ -1032,6 +1053,9 @@ async def retry_stuck_paid_purchases(
)
tokens = result.scalars().all()
+ # Separately fail exhausted purchases (retry_count >= max_retries)
+ await _fail_exhausted_purchases_batch(db, GuestPurchaseStatus.PAID, max_retries, max_age)
+
if not tokens:
return 0
@@ -1039,6 +1063,7 @@ async def retry_stuck_paid_purchases(
for token in tokens:
try:
async with AsyncSessionLocal() as retry_db:
+ await _increment_retry_count(retry_db, token)
await fulfill_purchase(retry_db, token)
retried += 1
logger.info('Retried stuck purchase successfully', token_prefix=token[:5])
@@ -1053,12 +1078,15 @@ async def retry_stuck_pending_activation(
stale_minutes: int = 10,
limit: int = 10,
max_age_hours: int = 24,
+ max_retries: int = 20,
) -> int:
"""Retry activation for purchases stuck in PENDING_ACTIVATION status.
This handles the case where activate_purchase() failed after the status
was already transitioned to PENDING_ACTIVATION (e.g., Remnawave panel was
temporarily down). Each retry runs in an isolated session.
+
+ Purchases exceeding max_retries are marked FAILED and an admin alert is sent.
"""
from app.database.database import AsyncSessionLocal
@@ -1069,6 +1097,7 @@ async def retry_stuck_pending_activation(
select(GuestPurchase.token)
.where(
GuestPurchase.status == GuestPurchaseStatus.PENDING_ACTIVATION.value,
+ GuestPurchase.retry_count < max_retries,
or_(GuestPurchase.paid_at < cutoff, GuestPurchase.paid_at.is_(None)),
or_(GuestPurchase.paid_at > max_age, GuestPurchase.paid_at.is_(None)),
GuestPurchase.user_id.isnot(None),
@@ -1078,6 +1107,9 @@ async def retry_stuck_pending_activation(
)
tokens = result.scalars().all()
+ # Separately fail exhausted purchases (retry_count >= max_retries)
+ await _fail_exhausted_purchases_batch(db, GuestPurchaseStatus.PENDING_ACTIVATION, max_retries, max_age)
+
if not tokens:
return 0
@@ -1085,6 +1117,7 @@ async def retry_stuck_pending_activation(
for token in tokens:
try:
async with AsyncSessionLocal() as retry_db:
+ await _increment_retry_count(retry_db, token)
await activate_purchase(retry_db, token)
retried += 1
logger.info('Retried stuck pending_activation successfully', token_prefix=token[:5])
@@ -1092,3 +1125,369 @@ async def retry_stuck_pending_activation(
logger.exception('Failed to retry stuck pending_activation', token_prefix=token[:5])
return retried
+
+
+async def _increment_retry_count(db: AsyncSession, purchase_token: str) -> None:
+ """Atomically increment retry_count via UPDATE statement (no SELECT, no identity map pollution)."""
+ await db.execute(
+ update(GuestPurchase)
+ .where(GuestPurchase.token == purchase_token)
+ .values(retry_count=GuestPurchase.retry_count + 1)
+ )
+ await db.commit()
+
+
+async def _fail_exhausted_purchases_batch(
+ db: AsyncSession,
+ status: GuestPurchaseStatus,
+ max_retries: int,
+ max_age: datetime,
+) -> None:
+ """Find and mark exhausted purchases as FAILED, then send admin alerts."""
+ from app.database.crud.landing import update_purchase_status
+ from app.database.database import AsyncSessionLocal
+
+ result = await db.execute(
+ select(GuestPurchase.token, GuestPurchase.retry_count)
+ .where(
+ GuestPurchase.status == status.value,
+ GuestPurchase.retry_count >= max_retries,
+ or_(GuestPurchase.paid_at > max_age, GuestPurchase.paid_at.is_(None)),
+ )
+ .limit(10)
+ )
+ exhausted = result.all()
+
+ for token, retry_count in exhausted:
+ # Collect alert data before closing the session
+ alert_data: dict | None = None
+ try:
+ async with AsyncSessionLocal() as fail_db:
+ row = await fail_db.execute(select(GuestPurchase).where(GuestPurchase.token == token).with_for_update())
+ purchase = row.scalars().first()
+ if purchase and purchase.status not in (
+ GuestPurchaseStatus.DELIVERED.value,
+ GuestPurchaseStatus.FAILED.value,
+ ):
+ # Capture alert data before commit expires attributes
+ alert_data = {
+ 'id': purchase.id,
+ 'token': purchase.token,
+ 'amount_kopeks': purchase.amount_kopeks,
+ 'payment_method': purchase.payment_method,
+ 'payment_id': purchase.payment_id,
+ 'contact_type': purchase.contact_type,
+ 'contact_value': purchase.contact_value,
+ 'created_at': purchase.created_at,
+ }
+ await update_purchase_status(fail_db, token, GuestPurchaseStatus.FAILED)
+ logger.error(
+ 'Purchase exceeded max retries — marked FAILED',
+ token_prefix=token[:5],
+ retry_count=retry_count,
+ phase=status.value,
+ )
+ except Exception:
+ logger.exception('Failed to mark exhausted purchase as FAILED', token_prefix=token[:5])
+
+ # Send alert OUTSIDE the session (no row lock held)
+ if alert_data:
+ await _send_stuck_purchase_alert(alert_data, retry_count, status.value)
+
+
+async def _send_stuck_purchase_alert(data: dict, retry_count: int, phase: str) -> None:
+ """Send admin notification about a purchase that exhausted all retries.
+
+ Accepts a plain dict (not ORM object) so it can be called after the session is closed.
+ """
+ if not getattr(settings, 'ADMIN_NOTIFICATIONS_ENABLED', False) or not settings.BOT_TOKEN:
+ return
+ try:
+ import html as html_mod
+
+ from aiogram import Bot
+
+ from app.services.admin_notification_service import AdminNotificationService, NotificationCategory
+
+ amount_rub = data['amount_kopeks'] / 100
+ contact_value = html_mod.escape(str(data.get('contact_value', '?')))
+ contact_type = html_mod.escape(str(data.get('contact_type', '?')))
+ text = (
+ f'STUCK PURCHASE — retries exhausted\n\n'
+ f'Token: {data["token"][:8]}...\n'
+ f'Status: {phase} → FAILED\n'
+ f'Retries: {retry_count}\n'
+ f'Amount: {amount_rub:.0f} ₽\n'
+ f'Payment: {html_mod.escape(str(data.get("payment_method") or "?"))}\n'
+ f'Payment ID: {html_mod.escape(str(data.get("payment_id") or "?"))}\n'
+ f'Contact: {contact_type}: {contact_value}\n'
+ f'Created: {data["created_at"]:%Y-%m-%d %H:%M UTC}\n\n'
+ f'Requires manual investigation.'
+ )
+
+ async with Bot(token=settings.BOT_TOKEN) as bot:
+ service = AdminNotificationService(bot)
+ await service.send_admin_notification(text, category=NotificationCategory.ERRORS)
+ except Exception:
+ logger.warning('Failed to send stuck purchase admin alert', purchase_id=data.get('id'), exc_info=True)
+
+
+async def _send_amount_mismatch_alert(
+ purchase: GuestPurchase,
+ provider_amount_kopeks: int,
+ provider_payment_id: str,
+ payment_method: str | None,
+) -> None:
+ """Send admin alert when recovery detects an amount mismatch (possible fraud or bug)."""
+ if not getattr(settings, 'ADMIN_NOTIFICATIONS_ENABLED', False) or not settings.BOT_TOKEN:
+ return
+ try:
+ import html as html_mod
+
+ from aiogram import Bot
+
+ from app.services.admin_notification_service import AdminNotificationService, NotificationCategory
+
+ text = (
+ f'AMOUNT MISMATCH — purchase marked FAILED\n\n'
+ f'Token: {purchase.token[:8]}...\n'
+ f'Expected: {purchase.amount_kopeks / 100:.0f} ₽\n'
+ f'Provider: {provider_amount_kopeks / 100:.0f} ₽\n'
+ f'Payment: {html_mod.escape(str(payment_method or "?"))}\n'
+ f'Payment ID: {html_mod.escape(str(provider_payment_id))}\n'
+ f'Contact: {html_mod.escape(str(purchase.contact_type))}: '
+ f'{html_mod.escape(str(purchase.contact_value))}\n\n'
+ f'Requires manual investigation.'
+ )
+
+ async with Bot(token=settings.BOT_TOKEN) as bot:
+ service = AdminNotificationService(bot)
+ await service.send_admin_notification(text, category=NotificationCategory.ERRORS)
+ except Exception:
+ logger.warning('Failed to send amount mismatch alert', purchase_id=purchase.id, exc_info=True)
+
+
+async def recover_stuck_pending_purchases(
+ db: AsyncSession,
+ stale_minutes: int = 10,
+ limit: int = 10,
+ max_age_hours: int = 24,
+) -> int:
+ """Recover purchases stuck in PENDING by checking provider payment status.
+
+ Queries all payment provider tables (YooKassa, Heleket, CryptoBot, etc.)
+ for succeeded payments matching the purchase_token. If a provider payment
+ is confirmed but the GuestPurchase is still PENDING (webhook was lost or
+ processing failed), marks the purchase as PAID so retry_stuck_paid_purchases
+ can fulfill it. Includes amount verification.
+
+ Returns the number of recovered purchases.
+ """
+ from app.database.database import AsyncSessionLocal
+
+ cutoff = datetime.now(UTC) - timedelta(minutes=stale_minutes)
+ max_age = datetime.now(UTC) - timedelta(hours=max_age_hours)
+
+ # Find PENDING purchases older than stale_minutes but younger than max_age_hours
+ result = await db.execute(
+ select(GuestPurchase.token, GuestPurchase.payment_method)
+ .where(
+ GuestPurchase.status == GuestPurchaseStatus.PENDING.value,
+ GuestPurchase.created_at < cutoff,
+ GuestPurchase.created_at > max_age,
+ )
+ .order_by(GuestPurchase.created_at.asc())
+ .limit(limit)
+ )
+ pending_purchases = result.all()
+
+ if not pending_purchases:
+ return 0
+
+ recovered = 0
+ for token, payment_method in pending_purchases:
+ try:
+ async with AsyncSessionLocal() as recover_db:
+ paid = await _check_and_recover_pending_purchase(recover_db, token, payment_method)
+ if paid:
+ recovered += 1
+ except Exception:
+ logger.exception('Failed to recover pending purchase', token_prefix=token[:5])
+
+ return recovered
+
+
+async def _find_succeeded_provider_payment(
+ db: AsyncSession,
+ base_method: str,
+ purchase_token: str,
+) -> tuple[str, int | None] | None:
+ """Query provider payment tables for a succeeded payment matching purchase_token.
+
+ Returns ``(provider_payment_id, amount_kopeks)`` or ``None``.
+ ``amount_kopeks`` is ``None`` when the amount check should be skipped
+ (e.g., CryptoBot where USD→RUB conversion introduces imprecision).
+ """
+ from sqlalchemy import cast
+ from sqlalchemy.types import JSON as SA_JSON
+
+ from app.database.models import (
+ CloudPaymentsPayment,
+ CryptoBotPayment,
+ FreekassaPayment,
+ HeleketPayment,
+ KassaAiPayment,
+ MulenPayPayment,
+ Pal24Payment,
+ PlategaPayment,
+ RioPayPayment,
+ SeverPayPayment,
+ WataPayment,
+ YooKassaPayment,
+ )
+
+ # --- CryptoBot: special case — payload field (text JSON), skip amount check ---
+ if base_method == 'cryptobot':
+ result = await db.execute(
+ select(CryptoBotPayment).where(
+ CryptoBotPayment.status == 'paid',
+ cast(CryptoBotPayment.payload, SA_JSON)['purchase_token'].as_string() == purchase_token,
+ )
+ )
+ p = result.scalars().first()
+ return (p.invoice_id, None) if p else None
+
+ # --- All other providers: metadata_json['purchase_token'] + is_paid/status filters ---
+ model = None
+ payment_id_attr: str = ''
+ extra_conditions: list = []
+
+ if base_method.startswith('yookassa'):
+ model = YooKassaPayment
+ payment_id_attr = 'yookassa_payment_id'
+ extra_conditions = [YooKassaPayment.status == 'succeeded', YooKassaPayment.is_paid.is_(True)]
+ elif base_method == 'heleket':
+ model = HeleketPayment
+ payment_id_attr = 'uuid'
+ extra_conditions = [HeleketPayment.status.in_(['paid', 'paid_over'])]
+ elif base_method == 'mulenpay':
+ model = MulenPayPayment
+ payment_id_attr = 'uuid'
+ extra_conditions = [MulenPayPayment.is_paid.is_(True)]
+ elif base_method == 'pal24':
+ model = Pal24Payment
+ payment_id_attr = 'bill_id'
+ extra_conditions = [Pal24Payment.is_paid.is_(True)]
+ elif base_method == 'wata':
+ model = WataPayment
+ payment_id_attr = 'payment_link_id'
+ extra_conditions = [WataPayment.is_paid.is_(True)]
+ elif base_method == 'platega':
+ model = PlategaPayment
+ payment_id_attr = 'platega_transaction_id'
+ extra_conditions = [PlategaPayment.is_paid.is_(True)]
+ elif base_method == 'cloudpayments':
+ model = CloudPaymentsPayment
+ payment_id_attr = 'invoice_id'
+ extra_conditions = [CloudPaymentsPayment.status == 'completed', CloudPaymentsPayment.is_paid.is_(True)]
+ elif base_method == 'freekassa':
+ model = FreekassaPayment
+ payment_id_attr = 'order_id'
+ extra_conditions = [FreekassaPayment.status == 'success', FreekassaPayment.is_paid.is_(True)]
+ elif base_method == 'kassa_ai':
+ model = KassaAiPayment
+ payment_id_attr = 'order_id'
+ extra_conditions = [KassaAiPayment.status == 'success', KassaAiPayment.is_paid.is_(True)]
+ elif base_method == 'riopay':
+ model = RioPayPayment
+ payment_id_attr = 'order_id'
+ extra_conditions = [RioPayPayment.status == 'success', RioPayPayment.is_paid.is_(True)]
+ elif base_method == 'severpay':
+ model = SeverPayPayment
+ payment_id_attr = 'order_id'
+ extra_conditions = [SeverPayPayment.status == 'success', SeverPayPayment.is_paid.is_(True)]
+
+ if model is None:
+ return None
+
+ result = await db.execute(
+ select(model).where(
+ model.metadata_json['purchase_token'].as_string() == purchase_token,
+ *extra_conditions,
+ )
+ )
+ p = result.scalars().first()
+ if p is None:
+ return None
+
+ payment_id = str(getattr(p, payment_id_attr))
+ # amount_kopeks: Integer column for most providers, @property for Heleket
+ amount = getattr(p, 'amount_kopeks', None)
+ return (payment_id, amount)
+
+
+async def _check_and_recover_pending_purchase(
+ db: AsyncSession,
+ purchase_token: str,
+ payment_method: str | None,
+) -> bool:
+ """Check if a PENDING purchase has a succeeded payment and transition to PAID.
+
+ Uses SELECT ... FOR UPDATE on the GuestPurchase row to prevent concurrent
+ webhook processing from racing with the recovery.
+ Verifies amount match between provider payment and guest purchase.
+ """
+ from app.database.crud.landing import update_purchase_status
+
+ # Lock the row to prevent TOCTOU race with concurrent webhook processing
+ result = await db.execute(select(GuestPurchase).where(GuestPurchase.token == purchase_token).with_for_update())
+ purchase = result.scalars().first()
+ if purchase is None or purchase.status != GuestPurchaseStatus.PENDING.value:
+ return False
+
+ # Resolve base method: 'yookassa_sbp' → 'yookassa', 'kassa_ai' stays 'kassa_ai'
+ base_method = _resolve_base_payment_method(payment_method)
+
+ match = await _find_succeeded_provider_payment(db, base_method, purchase_token)
+ if match is None:
+ if base_method:
+ logger.debug(
+ 'No succeeded provider payment found for PENDING purchase',
+ token_prefix=purchase_token[:5],
+ payment_method=payment_method,
+ )
+ return False
+
+ provider_payment_id, provider_amount_kopeks = match
+
+ # Amount verification (skip when provider_amount_kopeks is None, e.g., crypto)
+ if provider_amount_kopeks is not None and provider_amount_kopeks != purchase.amount_kopeks:
+ logger.error(
+ 'Amount mismatch during PENDING recovery — skipping',
+ token_prefix=purchase_token[:5],
+ provider_amount=provider_amount_kopeks,
+ purchase_amount=purchase.amount_kopeks,
+ payment_method=payment_method,
+ )
+ # Mark FAILED to prevent repeated mismatch logs every cycle
+ from app.database.crud.landing import update_purchase_status as _update_status
+
+ await _update_status(db, purchase_token, GuestPurchaseStatus.FAILED)
+ await _send_amount_mismatch_alert(purchase, provider_amount_kopeks, provider_payment_id, payment_method)
+ return False
+
+ # Transition PENDING → PAID for retry_stuck_paid_purchases to handle
+ await update_purchase_status(
+ db,
+ purchase_token,
+ GuestPurchaseStatus.PAID,
+ payment_id=provider_payment_id,
+ paid_at=datetime.now(UTC),
+ )
+ logger.info(
+ 'Recovered stuck PENDING purchase → PAID',
+ token_prefix=purchase_token[:5],
+ payment_method=payment_method,
+ provider_payment_id=provider_payment_id,
+ )
+ return True
diff --git a/app/services/monitoring_service.py b/app/services/monitoring_service.py
index e5952544..69d3def8 100644
--- a/app/services/monitoring_service.py
+++ b/app/services/monitoring_service.py
@@ -1742,18 +1742,35 @@ class MonitoringService:
)
async def _retry_stuck_guest_purchases(self, db: AsyncSession):
- try:
- from app.services.guest_purchase_service import retry_stuck_paid_purchases, retry_stuck_pending_activation
+ from app.services.guest_purchase_service import (
+ recover_stuck_pending_purchases,
+ retry_stuck_paid_purchases,
+ retry_stuck_pending_activation,
+ )
+ # Phase 1: Recover PENDING purchases where provider payment already succeeded
+ try:
+ recovered = await recover_stuck_pending_purchases(db, stale_minutes=10, limit=10)
+ if recovered:
+ logger.info('Recovered stuck PENDING purchases', recovered=recovered)
+ except Exception:
+ logger.error('Error recovering stuck PENDING guest purchases', exc_info=True)
+
+ # Phase 2: Retry fulfillment for purchases in PAID status
+ try:
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 PAID guest purchases', exc_info=True)
+ # Phase 3: Retry activation for purchases in PENDING_ACTIVATION status
+ try:
retried_pa = await retry_stuck_pending_activation(db, stale_minutes=10, limit=10)
if retried_pa:
logger.info('Retried stuck pending_activation purchases', retried=retried_pa)
except Exception:
- logger.error('Error retrying stuck guest purchases', exc_info=True)
+ logger.error('Error retrying stuck PENDING_ACTIVATION guest purchases', exc_info=True)
async def _cleanup_inactive_users(self, db: AsyncSession):
try:
diff --git a/app/services/payment/common.py b/app/services/payment/common.py
index 1f0cffb4..4bfc1a0b 100644
--- a/app/services/payment/common.py
+++ b/app/services/payment/common.py
@@ -476,8 +476,7 @@ async def try_fulfill_guest_purchase(
introduces imprecision.
Returns:
- ``True`` -- guest purchase was detected and successfully fulfilled.
- ``False`` -- guest purchase was detected but fulfillment failed.
+ ``True`` -- guest purchase was detected and consumed (fulfilled or queued for retry).
``None`` -- this is NOT a guest purchase (caller should proceed normally).
"""
purchase_token = _extract_guest_purchase_token(metadata)
@@ -485,7 +484,7 @@ async def try_fulfill_guest_purchase(
return None
from app.database.crud.landing import get_purchase_by_token, update_purchase_status
- from app.database.models import GuestPurchaseStatus
+ from app.database.models import GuestPurchase, GuestPurchaseStatus
from app.services.guest_purchase_service import fulfill_purchase
try:
@@ -558,13 +557,26 @@ async def try_fulfill_guest_purchase(
provider=provider_name,
error=guest_error,
)
- # Mark as FAILED so it doesn't get retried forever
+ # Mark as PAID (not FAILED) so retry_stuck_paid_purchases can pick it up.
+ # Use a fresh session to avoid tainted-session issues after rollback.
+ # The monitoring service retries PAID purchases every 5 minutes for up to 24 hours.
try:
- await update_purchase_status(
- db,
- purchase_token,
- GuestPurchaseStatus.FAILED,
- )
+ from app.database.database import AsyncSessionLocal
+
+ async with AsyncSessionLocal() as recovery_db:
+ # Use FOR UPDATE to prevent TOCTOU race with concurrent webhook.
+ row = await recovery_db.execute(
+ select(GuestPurchase).where(GuestPurchase.token == purchase_token).with_for_update()
+ )
+ current = row.scalars().first()
+ if current and current.status in (
+ GuestPurchaseStatus.PENDING.value,
+ GuestPurchaseStatus.PAID.value,
+ ):
+ current.status = GuestPurchaseStatus.PAID.value
+ current.payment_id = provider_payment_id
+ current.paid_at = datetime.now(UTC)
+ await recovery_db.commit()
except Exception:
- logger.exception('Failed to mark guest purchase as FAILED')
- return False
+ logger.exception('Failed to mark guest purchase as PAID for retry')
+ return True
diff --git a/app/services/payment_service.py b/app/services/payment_service.py
index a9bcd8c4..18446ca3 100644
--- a/app/services/payment_service.py
+++ b/app/services/payment_service.py
@@ -721,7 +721,7 @@ class PaymentService(
payment_system_id=ps_id,
)
if result:
- await _patch_guest_metadata(result['local_payment_id'], payment_method)
+ await _patch_guest_metadata(result['local_payment_id'], 'kassa_ai')
return {
'payment_url': result.get('payment_url'),
'payment_id': result.get('order_id'),
diff --git a/app/services/promo_group_assignment.py b/app/services/promo_group_assignment.py
index b3d46fb5..468eaa63 100644
--- a/app/services/promo_group_assignment.py
+++ b/app/services/promo_group_assignment.py
@@ -8,6 +8,7 @@ from sqlalchemy.ext.asyncio import AsyncSession
from app.config import settings
from app.database.crud.transaction import get_user_total_spent_kopeks
+from app.database.crud.user import lock_user_for_update
from app.database.models import PromoGroup, User
from app.services.admin_notification_service import AdminNotificationService
@@ -90,7 +91,9 @@ async def maybe_assign_promo_group_by_total_spent(
) -> PromoGroup | None:
from app.database.crud.user_promo_group import (
add_user_to_promo_group,
+ get_user_promo_groups,
has_user_promo_group,
+ remove_user_from_promo_group,
sync_user_primary_promo_group,
)
@@ -99,6 +102,9 @@ async def maybe_assign_promo_group_by_total_spent(
logger.debug('Не удалось найти пользователя для автовыдачи промогруппы', user_id=user_id)
return None
+ # Блокируем строку пользователя для предотвращения гонок при конкурентных вебхуках
+ user = await lock_user_for_update(db, user)
+
# Получаем текущую primary промогруппу
old_group = user.get_primary_promo_group()
@@ -108,60 +114,68 @@ async def maybe_assign_promo_group_by_total_spent(
previous_threshold = user.auto_promo_group_threshold_kopeks or 0
- target_group = await _get_best_group_for_spending(
- db,
- total_spent,
- min_threshold_kopeks=previous_threshold,
- )
+ # Находим группу, соответствующую текущим тратам (без порогового фильтра,
+ # чтобы промокод-группы всегда очищались при покупке)
+ target_group = await _get_best_group_for_spending(db, total_spent)
if not target_group:
return None
try:
target_threshold = target_group.auto_assign_total_spent_kopeks or 0
- if target_threshold <= previous_threshold:
- logger.debug(
- "Порог промогруппы '' не превышает ранее назначенный для пользователя",
- target_group_name=target_group.name,
- target_threshold=target_threshold,
- previous_threshold=previous_threshold,
- telegram_id=user.telegram_id,
- )
- return None
+ # Фаза 1: Удаляем старые auto/promocode группы, отличные от целевой
+ current_groups = await get_user_promo_groups(db, user_id)
+ removed_any = False
+ for upg in current_groups:
+ if upg.promo_group_id != target_group.id and upg.assigned_by in ('auto', 'promocode'):
+ await remove_user_from_promo_group(db, user_id, upg.promo_group_id, commit=False)
+ removed_any = True
+ logger.info(
+ 'Удалена старая промогруппа перед автоназначением',
+ telegram_id=user.telegram_id,
+ old_group_name=upg.promo_group.name if upg.promo_group else upg.promo_group_id,
+ old_assigned_by=upg.assigned_by,
+ )
- # Проверяем, есть ли уже эта группа у пользователя
+ if removed_any:
+ await db.flush()
+ await db.refresh(user)
+
+ # Проверяем, есть ли уже целевая группа у пользователя
already_has_group = await has_user_promo_group(db, user_id, target_group.id)
- if user.auto_promo_group_assigned and already_has_group:
+ if user.auto_promo_group_assigned and already_has_group and not removed_any:
logger.debug(
- "Пользователь уже имеет промогруппу '', повторная выдача не требуется",
+ 'Пользователь уже имеет промогруппу, повторная выдача не требуется',
telegram_id=user.telegram_id,
target_group_name=target_group.name,
)
- await sync_user_primary_promo_group(db, user_id)
if target_threshold > previous_threshold:
user.auto_promo_group_threshold_kopeks = target_threshold
user.updated_at = datetime.now(UTC)
- await db.commit()
- await db.refresh(user)
+ await db.commit()
+ await db.refresh(user)
return target_group
user.auto_promo_group_assigned = True
- user.auto_promo_group_threshold_kopeks = target_threshold
+ if target_threshold > previous_threshold:
+ user.auto_promo_group_threshold_kopeks = target_threshold
user.updated_at = datetime.now(UTC)
+ newly_added = False
if not already_has_group:
- # Добавляем новую промогруппу к существующим
- await add_user_to_promo_group(db, user_id, target_group.id, assigned_by='auto')
+ await add_user_to_promo_group(db, user_id, target_group.id, assigned_by='auto', commit=False)
+ newly_added = True
logger.info(
- "🤖 Пользователю добавлена промогруппа '' за траты ₽",
+ 'Пользователю назначена промогруппа за траты',
telegram_id=user.telegram_id,
target_group_name=target_group.name,
total_spent=total_spent / 100,
)
else:
+ await sync_user_primary_promo_group(db, user_id)
logger.info(
- "🤖 Пользователь уже имеет промогруппу '', отмечаем автоприсвоение",
+ 'Пользователь уже имеет промогруппу, синхронизировано',
telegram_id=user.telegram_id,
target_group_name=target_group.name,
)
@@ -169,7 +183,7 @@ async def maybe_assign_promo_group_by_total_spent(
await db.commit()
await db.refresh(user)
- if not already_has_group:
+ if newly_added:
await _notify_admins_about_auto_assignment(
db,
user,
diff --git a/app/services/promocode_service.py b/app/services/promocode_service.py
index bcaa8c5b..2253ed78 100644
--- a/app/services/promocode_service.py
+++ b/app/services/promocode_service.py
@@ -119,7 +119,7 @@ class PromoCodeService:
if promo_group:
# Add promo group to user
await add_user_to_promo_group(
- db, user_id, promocode.promo_group_id, assigned_by='promocode'
+ db, user_id, promocode.promo_group_id, assigned_by='promocode', commit=False
)
logger.info(
@@ -393,7 +393,7 @@ class PromoCodeService:
has_group = await has_user_promo_group(db, user_id, promocode.promo_group_id)
if has_group:
- await remove_user_from_promo_group(db, user_id, promocode.promo_group_id)
+ await remove_user_from_promo_group(db, user_id, promocode.promo_group_id, commit=False)
logger.info(
'Снята промогруппа ID у пользователя при деактивации промокода',
promo_group_id=promocode.promo_group_id,
diff --git a/app/utils/cache.py b/app/utils/cache.py
index 8824cbf4..7624246d 100644
--- a/app/utils/cache.py
+++ b/app/utils/cache.py
@@ -400,9 +400,16 @@ return c
return fail_closed
@staticmethod
- async def is_rate_limited(user_id: int, action: str, limit: int, window: int) -> bool:
+ async def is_rate_limited(
+ user_id: int,
+ action: str,
+ limit: int,
+ window: int,
+ *,
+ fail_closed: bool = False,
+ ) -> bool:
key = cache_key('rate_limit', user_id, action)
- return await RateLimitCache._atomic_rate_check(key, limit, window)
+ return await RateLimitCache._atomic_rate_check(key, limit, window, fail_closed=fail_closed)
@staticmethod
async def reset_rate_limit(user_id: int, action: str) -> bool:
diff --git a/migrations/alembic/env.py b/migrations/alembic/env.py
index 99593cc4..effb6318 100644
--- a/migrations/alembic/env.py
+++ b/migrations/alembic/env.py
@@ -45,7 +45,11 @@ def run_migrations_offline() -> None:
def do_run_migrations(connection: Connection) -> None:
- context.configure(connection=connection, target_metadata=target_metadata)
+ context.configure(
+ connection=connection,
+ target_metadata=target_metadata,
+ transaction_per_migration=True,
+ )
with context.begin_transaction():
context.run_migrations()
diff --git a/migrations/alembic/versions/0041_add_performance_indexes.py b/migrations/alembic/versions/0041_add_performance_indexes.py
new file mode 100644
index 00000000..0b18053a
--- /dev/null
+++ b/migrations/alembic/versions/0041_add_performance_indexes.py
@@ -0,0 +1,56 @@
+"""add performance indexes for referral network queries
+
+Revision ID: 0041
+Revises: 0040
+Create Date: 2026-03-20
+
+"""
+
+from typing import Sequence, Union
+
+from alembic import op
+
+revision: str = '0041'
+down_revision: Union[str, None] = '0040'
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+ # CREATE INDEX CONCURRENTLY cannot run inside a transaction.
+ # autocommit_block() temporarily disables the transaction wrapper.
+ #
+ # NOTE: If a concurrent index creation fails midway, PostgreSQL leaves behind
+ # an INVALID index. Check with:
+ # SELECT indexrelname FROM pg_stat_user_indexes
+ # JOIN pg_index ON pg_index.indexrelid = pg_stat_user_indexes.indexrelid
+ # WHERE NOT pg_index.indisvalid;
+ # Then drop the invalid index and re-run the migration.
+ with op.get_context().autocommit_block():
+ # Index on advertising_campaign_registrations(user_id, created_at)
+ # Fixes sequential scan in _fetch_campaign_registrations which filters by user_id
+ # and uses ROW_NUMBER() OVER (PARTITION BY user_id ORDER BY created_at)
+ op.create_index(
+ 'ix_campaign_reg_user_created',
+ 'advertising_campaign_registrations',
+ ['user_id', 'created_at'],
+ if_not_exists=True,
+ postgresql_concurrently=True,
+ )
+
+ # Covering composite index on transactions(user_id, type, is_completed, amount_kopeks)
+ # Enables index-only scans for aggregation queries in referral network stats:
+ # _fetch_personal_spent, _fetch_branch_revenue, _fetch_campaign_stats
+ op.create_index(
+ 'ix_transactions_user_type_completed_amount',
+ 'transactions',
+ ['user_id', 'type', 'is_completed', 'amount_kopeks'],
+ if_not_exists=True,
+ postgresql_concurrently=True,
+ )
+
+
+def downgrade() -> None:
+ with op.get_context().autocommit_block():
+ op.execute('DROP INDEX CONCURRENTLY IF EXISTS ix_transactions_user_type_completed_amount')
+ op.execute('DROP INDEX CONCURRENTLY IF EXISTS ix_campaign_reg_user_created')
diff --git a/migrations/alembic/versions/0042_add_retry_count_and_payment_recovery_indexes.py b/migrations/alembic/versions/0042_add_retry_count_and_payment_recovery_indexes.py
new file mode 100644
index 00000000..cbb5d73a
--- /dev/null
+++ b/migrations/alembic/versions/0042_add_retry_count_and_payment_recovery_indexes.py
@@ -0,0 +1,88 @@
+"""add retry_count to guest_purchases and expression indexes for payment recovery
+
+Revision ID: 0042
+Revises: 0041
+Create Date: 2026-03-20
+
+"""
+
+from typing import Sequence, Union
+
+import sqlalchemy as sa
+from alembic import op
+
+revision: str = '0042'
+down_revision: Union[str, None] = '0041'
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+# Payment tables with metadata_json + is_paid column.
+_TABLES_WITH_IS_PAID = [
+ 'yookassa_payments',
+ 'mulenpay_payments',
+ 'pal24_payments',
+ 'wata_payments',
+ 'platega_payments',
+ 'cloudpayments_payments',
+ 'freekassa_payments',
+ 'kassa_ai_payments',
+ 'riopay_payments',
+ 'severpay_payments',
+]
+
+# All tables that get a metadata purchase_token index (for downgrade)
+_ALL_METADATA_TABLES = [*_TABLES_WITH_IS_PAID, 'heleket_payments']
+
+
+def upgrade() -> None:
+ # 1. Add retry_count column to guest_purchases (safe: has server_default)
+ op.add_column(
+ 'guest_purchases',
+ sa.Column('retry_count', sa.Integer(), nullable=False, server_default='0'),
+ )
+
+ # 2. Create expression indexes for payment recovery queries.
+ # These allow efficient lookup of succeeded payments by purchase_token
+ # stored inside the metadata_json column.
+ with op.get_context().autocommit_block():
+ # Tables with is_paid boolean column
+ for table in _TABLES_WITH_IS_PAID:
+ idx_name = f'ix_{table}_metadata_purchase_token'
+ op.execute(
+ sa.text(
+ f'CREATE INDEX CONCURRENTLY IF NOT EXISTS {idx_name} '
+ f"ON {table} ((metadata_json ->> 'purchase_token')) "
+ f'WHERE is_paid = TRUE'
+ )
+ )
+
+ # Heleket: no is_paid column (it's a Python @property), use status filter
+ op.execute(
+ sa.text(
+ 'CREATE INDEX CONCURRENTLY IF NOT EXISTS ix_heleket_payments_metadata_purchase_token '
+ "ON heleket_payments ((metadata_json ->> 'purchase_token')) "
+ "WHERE status IN ('paid', 'paid_over')"
+ )
+ )
+
+ # CryptoBot: payload (text) column with JSON inside, no metadata_json
+ op.execute(
+ sa.text(
+ 'CREATE INDEX CONCURRENTLY IF NOT EXISTS ix_cryptobot_payments_payload_purchase_token '
+ "ON cryptobot_payments ((CAST(payload AS json) ->> 'purchase_token')) "
+ "WHERE status = 'paid'"
+ )
+ )
+
+
+def downgrade() -> None:
+ with op.get_context().autocommit_block():
+ for table in _ALL_METADATA_TABLES:
+ idx_name = f'ix_{table}_metadata_purchase_token'
+ op.execute(sa.text(f'DROP INDEX CONCURRENTLY IF EXISTS {idx_name}'))
+
+ op.execute(
+ sa.text('DROP INDEX CONCURRENTLY IF EXISTS ix_cryptobot_payments_payload_purchase_token')
+ )
+
+ op.drop_column('guest_purchases', 'retry_count')
diff --git a/uv.lock b/uv.lock
index d94e229f..392c7554 100644
--- a/uv.lock
+++ b/uv.lock
@@ -1115,7 +1115,7 @@ wheels = [
[[package]]
name = "remnawave-bedolaga-telegram-bot"
-version = "3.33.0"
+version = "3.34.1"
source = { virtual = "." }
dependencies = [
{ name = "aiogram" },