Как реализуешь рефереальную систему с точки зрения архитектуры?

# Как реализуешь рефереальную систему с точки зрения архитектуры?

Реферальная система — это сложный бизнес-сценарий, требующий продуманной архитектуры. Разберём правильный подход с использованием Clean Architecture и DDD.

Общая архитектура

┌─────────────────────────────────────────────────────┐
│ Presentation (API endpoints, Telegram Bot)          │
├─────────────────────────────────────────────────────┤
│ Application (Use Cases, Orchestration)              │
├─────────────────────────────────────────────────────┤
│ Domain (Entity, Value Objects, Aggregates)          │
├─────────────────────────────────────────────────────┤
│ Infrastructure (Database, Cache, Events)            │
└─────────────────────────────────────────────────────┘

Domain Layer (Бизнес-логика)

1. Entities и Value Objects

# domain/entities/referral.py
from dataclasses import dataclass
from datetime import datetime
from enum import Enum

class ReferralStatus(str, Enum):
    PENDING = "pending"      # Реферал не активирован
    ACTIVE = "active"        # Реферал активирован
    REVOKED = "revoked"      # Отозван (мошенничество)

class ReferralRewardType(str, Enum):
    BONUS = "bonus"          # Бонус после регистрации
    PERCENTAGE = "percentage"  # % от покупок реферала
    FIXED = "fixed"          # Фиксированная сумма

@dataclass(frozen=True)
class ReferralCode:
    """Value Object для уникального кода реферала"""
    code: str
    
    def __post_init__(self):
        if len(self.code) < 4:
            raise ValueError("Code too short")
        if not self.code.isalnum():
            raise ValueError("Code must be alphanumeric")

@dataclass
class Referral:
    """Entity: реферальные отношения между пользователями"""
    id: str
    referrer_id: str      # Тот, кто привёл
    referee_id: str       # Того, кто привёлся
    code: ReferralCode
    status: ReferralStatus
    created_at: datetime
    activated_at: datetime | None = None
    
    def activate(self) -> None:
        if self.status == ReferralStatus.PENDING:
            self.status = ReferralStatus.ACTIVE
            self.activated_at = datetime.now(UTC)
    
    def revoke(self, reason: str) -> None:
        """Отозвать реферал (например, при обнаружении мошенничества)"""
        self.status = ReferralStatus.REVOKED

@dataclass(frozen=True)
class ReferralReward:
    """Value Object: награда за реферал"""
    reward_type: ReferralRewardType
    amount: float
    
    def __post_init__(self):
        if self.amount <= 0:
            raise ValueError("Amount must be positive")

@dataclass
class ReferralRewardTransaction:
    """Entity: выплата награды"""
    id: str
    referral_id: str
    reward: ReferralReward
    transaction_id: str    # ID платежа в системе
    status: str            # pending, completed, failed
    created_at: datetime

2. Aggregate Root

# domain/aggregates/referrer_aggregate.py
from typing import List

@dataclass
class ReferrerAggregate:
    """Aggregate: реферер с его всеми реферальными отношениями"""
    referrer_id: str
    referral_code: ReferralCode
    referrals: List[Referral] = field(default_factory=list)
    total_earnings: float = 0.0
    
    def add_referral(self, referral: Referral) -> None:
        if referral.referrer_id != self.referrer_id:
            raise ValueError("Invalid referrer")
        self.referrals.append(referral)
    
    def get_active_referrals_count(self) -> int:
        return sum(
            1 for r in self.referrals 
            if r.status == ReferralStatus.ACTIVE
        )
    
    def apply_reward(self, reward_transaction: ReferralRewardTransaction) -> None:
        if reward_transaction.status == "completed":
            self.total_earnings += reward_transaction.reward.amount

3. Domain Services

# domain/services/referral_service.py
from typing import Protocol

class ReferralRepository(Protocol):
    """Interface для работы с реферальными данными"""
    def save(self, referral: Referral) -> None: ...
    def get_by_code(self, code: ReferralCode) -> Referral | None: ...
    def get_by_referrer_id(self, referrer_id: str) -> List[Referral]: ...

class ReferralCodeGenerator(Protocol):
    """Interface для генерации уникальных кодов"""
    def generate(self) -> ReferralCode: ...
    def validate_uniqueness(self, code: ReferralCode) -> bool: ...

class ReferralDomainService:
    """Domain Service: бизнес-логика, которая затрагивает несколько entities"""
    
    def __init__(
        self,
        referral_repo: ReferralRepository,
        code_generator: ReferralCodeGenerator,
    ):
        self.referral_repo = referral_repo
        self.code_generator = code_generator
    
    def create_referral_pair(self, referrer_id: str, referee_id: str) -> Referral:
        """Создать реферальное отношение с уникальным кодом"""
        # Генерируем уникальный код
        code = self._generate_unique_code()
        
        referral = Referral(
            id=str(uuid4()),
            referrer_id=referrer_id,
            referee_id=referee_id,
            code=code,
            status=ReferralStatus.PENDING,
            created_at=datetime.now(UTC),
        )
        
        self.referral_repo.save(referral)
        return referral
    
    def _generate_unique_code(self) -> ReferralCode:
        while True:
            code = self.code_generator.generate()
            if self.code_generator.validate_uniqueness(code):
                return code

Application Layer (Use Cases)

# application/use_cases/register_with_referral.py
from dataclasses import dataclass
from typing import Optional

@dataclass
class RegisterWithReferralRequest:
    email: str
    password: str
    referral_code: Optional[str] = None

@dataclass
class RegisterWithReferralResponse:
    user_id: str
    referral_code: str
    referrer_id: Optional[str]

class RegisterWithReferralUseCase:
    """Use Case: регистрация с использованием реферального кода"""
    
    def __init__(
        self,
        user_repo: "UserRepository",
        referral_repo: ReferralRepository,
        user_service: "UserService",
        domain_service: ReferralDomainService,
        event_bus: "EventBus",
    ):
        self.user_repo = user_repo
        self.referral_repo = referral_repo
        self.user_service = user_service
        self.domain_service = domain_service
        self.event_bus = event_bus
    
    async def execute(self, req: RegisterWithReferralRequest) -> RegisterWithReferralResponse:
        # Шаг 1: Валидация
        self._validate_request(req)
        
        # Шаг 2: Проверка реферального кода
        referrer_id = None
        if req.referral_code:
            referrer_id = await self._get_referrer_by_code(req.referral_code)
            if not referrer_id:
                raise ReferralCodeNotFoundError()
        
        # Шаг 3: Создание пользователя
        user = await self.user_service.create_user(
            email=req.email,
            password=req.password,
        )
        
        # Шаг 4: Активирование реферального отношения
        if referrer_id:
            referral = await self._activate_referral(referrer_id, user.id)
            # Публикуем domain event
            self.event_bus.publish(
                ReferralActivatedEvent(
                    referral_id=referral.id,
                    referrer_id=referrer_id,
                    referee_id=user.id,
                )
            )
        
        # Шаг 5: Генерируем реферальный код для нового пользователя
        new_referral_code = self.domain_service.create_referral_pair(
            referrer_id=user.id,
            referee_id="",  # Placeholder
        ).code
        
        return RegisterWithReferralResponse(
            user_id=user.id,
            referral_code=new_referral_code.code,
            referrer_id=referrer_id,
        )
    
    async def _get_referrer_by_code(self, code: str) -> Optional[str]:
        referral = self.referral_repo.get_by_code(ReferralCode(code))
        return referral.referrer_id if referral else None
    
    async def _activate_referral(self, referrer_id: str, referee_id: str) -> Referral:
        referral = Referral(
            id=str(uuid4()),
            referrer_id=referrer_id,
            referee_id=referee_id,
            code=ReferralCode(str(uuid4())[:8]),
            status=ReferralStatus.PENDING,
            created_at=datetime.now(UTC),
        )
        referral.activate()
        self.referral_repo.save(referral)
        return referral

# application/use_cases/process_referral_reward.py
class ProcessReferralRewardUseCase:
    """Use Case: начисление награды за реферала"""
    
    def __init__(
        self,
        referral_repo: ReferralRepository,
        reward_calculator: "RewardCalculator",
        payment_service: "PaymentService",
        event_bus: "EventBus",
    ):
        self.referral_repo = referral_repo
        self.reward_calculator = reward_calculator
        self.payment_service = payment_service
        self.event_bus = event_bus
    
    async def execute(self, referral_id: str, purchase_amount: float):
        # Получаем реферальное отношение
        referral = self.referral_repo.get_by_id(referral_id)
        
        if referral.status != ReferralStatus.ACTIVE:
            return  # Не начисляем награду за неактивные реферралы
        
        # Рассчитываем награду
        reward = self.reward_calculator.calculate(purchase_amount)
        
        # Обрабатываем платёж
        transaction = await self.payment_service.transfer(
            user_id=referral.referrer_id,
            amount=reward.amount,
            reason=f"Referral reward for {referral.referee_id}",
        )
        
        # Сохраняем информацию о вознаграждении
        reward_tx = ReferralRewardTransaction(
            id=str(uuid4()),
            referral_id=referral_id,
            reward=reward,
            transaction_id=transaction.id,
            status="completed",
            created_at=datetime.now(UTC),
        )
        
        self.event_bus.publish(
            ReferralRewardProcessedEvent(
                referral_id=referral_id,
                referrer_id=referral.referrer_id,
                reward_amount=reward.amount,
            )
        )

Infrastructure Layer

# infrastructure/persistence/referral_repository.py
from sqlalchemy.orm import Session

class SQLAlchemyReferralRepository:
    def __init__(self, db: Session):
        self.db = db
    
    def save(self, referral: Referral) -> None:
        db_referral = ReferralModel(
            id=referral.id,
            referrer_id=referral.referrer_id,
            referee_id=referral.referee_id,
            code=referral.code.code,
            status=referral.status.value,
            created_at=referral.created_at,
            activated_at=referral.activated_at,
        )
        self.db.add(db_referral)
        self.db.commit()
    
    def get_by_code(self, code: ReferralCode) -> Optional[Referral]:
        db_referral = self.db.query(ReferralModel).filter(
            ReferralModel.code == code.code
        ).first()
        
        if not db_referral:
            return None
        
        return self._to_domain(db_referral)

# infrastructure/events/referral_events.py
from dataclasses import dataclass
from datetime import datetime

@dataclass
class ReferralActivatedEvent:
    """Domain Event: реферал активирован"""
    referral_id: str
    referrer_id: str
    referee_id: str
    timestamp: datetime = field(default_factory=datetime.now)

@dataclass
class ReferralRewardProcessedEvent:
    """Domain Event: награда за реферал выплачена"""
    referral_id: str
    referrer_id: str
    reward_amount: float
    timestamp: datetime = field(default_factory=datetime.now)

Presentation Layer (API)

# presentation/api/referrals.py
from fastapi import APIRouter, Depends

router = APIRouter(prefix="/api/v1/referrals")

@router.post("/register")
async def register_with_referral(
    request: RegisterWithReferralRequest,
    use_case: RegisterWithReferralUseCase = Depends(),
):
    response = await use_case.execute(request)
    return {
        "user_id": response.user_id,
        "referral_code": response.referral_code,
    }

@router.get("/me/stats")
async def get_referral_stats(
    user_id: str = Depends(get_current_user),
    repo: ReferralRepository = Depends(),
):
    referrals = repo.get_by_referrer_id(user_id)
    return {
        "total_referrals": len(referrals),
        "active_referrals": sum(
            1 for r in referrals if r.status == ReferralStatus.ACTIVE
        ),
        "total_earnings": sum(
            r.total_earnings for r in referrals
        ),
    }

Ключевые особенности архитектуры

1. Separation of Concerns — бизнес-логика отделена от деталей реализации
2. Testability — легко тестировать каждый слой независимо
3. Scalability — структура позволяет легко добавлять новые типы наград
4. Fraud Prevention — явная обработка ошибок и отзыва рефералов
5. Event-Driven — асинхронная обработка наград через events

Вызовы при реализации

⚠️ Race Conditions — использовать SELECT FOR UPDATE при активации ⚠️ Fraud Detection — отслеживать подозрительные паттерны (multiple accounts) ⚠️ Compensation — что делать если реферал отменит покупку? ⚠️ Scalability — как обрабатывать millions of referrals?

Заключение

Правильная архитектура реферальной системы требует:

• Четкого разделения ответственности
• Независимого тестирования
• Обработки edge cases
• Мониторинга и fraud detection

Это комплексная система, но следуя DDD и Clean Architecture, можно создать масштабируемое и надёжное решение.