"""Creator Payout Service: requests, review, pluggable PayoutProvider, and audit trail. docs/planning/economy.md US-5 (tasks f5b4684f2e65, b52008923e93). """ from __future__ import annotations from dataclasses import dataclass from datetime import datetime, timezone import logging from typing import Protocol import uuid from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from app.models.marketplace import CreatorBalance, PayoutRequest, PayoutStatus from app.models.wallet import Transaction, TransactionType from app.services.purchase import PurchaseService from app.services.wallet import WalletService logger = logging.getLogger(__name__) MIN_PAYOUT_AMOUNT = 100 MAX_PAYOUT_AMOUNT = 100_000 LARGE_PAYOUT_THRESHOLD = 5000 class InsufficientCreatorBalanceError(Exception): """Raised when creator available balance is less than requested payout amount.""" pass class InvalidPayoutAmountError(ValueError): """Raised when requested payout amount is out of valid bounds.""" pass class PayoutNotFoundError(Exception): """Raised when payout request is not found.""" pass class InvalidPayoutStateError(Exception): """Raised when payout request is in invalid status for the requested transition.""" pass class PayoutProvider(Protocol): """Pluggable creator payout provider protocol (e.g. Stripe Connect, PayPal, Noop).""" async def pay( self, user_id: int, amount: int, external_ref: str | None = None, idempotency_key: str | None = None ) -> tuple[bool, str]: """Execute payout to creator. Returns (success, provider_ref).""" ... class NoopPayoutProvider: """Deterministic no-op provider for development and testing.""" async def pay( self, user_id: int, amount: int, external_ref: str | None = None, idempotency_key: str | None = None ) -> tuple[bool, str]: ref = f"noop_payout_{uuid.uuid4().hex[:12]}" logger.info( f"[NoopPayoutProvider] Simulated payout of {amount} CR to user {user_id} (ref={ref})" ) return True, ref default_payout_provider: PayoutProvider = NoopPayoutProvider() class PayoutService: def __init__(self, provider: PayoutProvider | None = None): self.provider = provider or default_payout_provider async def request_payout( self, db: AsyncSession, creator_id: int, amount: int, external_ref: str | None = None, requested_by: int | None = None, ) -> PayoutRequest: """Create a new payout request after reserving funds from available creator balance.""" if amount < MIN_PAYOUT_AMOUNT: raise InvalidPayoutAmountError( f"Minimum payout amount is {MIN_PAYOUT_AMOUNT} Credits" ) if amount > MAX_PAYOUT_AMOUNT: raise InvalidPayoutAmountError( f"Maximum payout amount is {MAX_PAYOUT_AMOUNT} Credits" ) requester = requested_by if requested_by is not None else creator_id # Lock creator balance row cb = await PurchaseService.get_or_create_creator_balance(db, creator_id, for_update=True) if cb.available < amount: raise InsufficientCreatorBalanceError( f"Insufficient available balance: {cb.available} CR available, {amount} CR requested" ) # Reserve funds: move from available to pending cb.available -= amount cb.pending += amount cb.updated_at = datetime.now(timezone.utc) payout = PayoutRequest( creator_id=creator_id, amount=amount, status=PayoutStatus.pending, external_ref=external_ref, requested_by=requester, ) db.add(payout) await db.commit() await db.refresh(payout) return payout async def approve_and_pay( self, db: AsyncSession, payout_id: int, reviewer_id: int, provider: PayoutProvider | None = None, ) -> PayoutRequest: """Admin approves and executes payout via provider.""" stmt = select(PayoutRequest).where(PayoutRequest.id == payout_id).with_for_update() res = await db.execute(stmt) payout = res.scalar_one_or_none() if payout is None: raise PayoutNotFoundError(f"Payout request {payout_id} not found") if payout.status != PayoutStatus.pending: raise InvalidPayoutStateError( f"Cannot approve payout in status {payout.status.value}" ) active_provider = provider or self.provider success, provider_ref = await active_provider.pay( user_id=payout.creator_id, amount=payout.amount, external_ref=payout.external_ref, idempotency_key=f"payout-{payout.id}", ) if not success: raise RuntimeError("Payout provider rejected transaction") now = datetime.now(timezone.utc) cb = await PurchaseService.get_or_create_creator_balance(db, payout.creator_id, for_update=True) cb.pending = max(0, cb.pending - payout.amount) cb.updated_at = now payout.status = PayoutStatus.paid payout.provider_ref = provider_ref payout.reviewed_by = reviewer_id payout.reviewed_at = now # Record immutable ledger audit row wallet = await WalletService.get_or_create_wallet(db, payout.creator_id, for_update=True) tx = Transaction( wallet_id=wallet.id, type=TransactionType.payout, amount=-payout.amount, balance_after=wallet.balance, reference_type="payout_request", reference_id=str(payout.id), idempotency_key=f"payout-audit-{payout.id}", metadata_json={ "provider_ref": provider_ref, "reviewed_by": reviewer_id, "external_ref": payout.external_ref, }, ) db.add(tx) await db.commit() await db.refresh(payout) return payout async def reject_payout( self, db: AsyncSession, payout_id: int, reviewer_id: int, reason: str | None = None, ) -> PayoutRequest: """Admin rejects payout and returns reserved funds to creator's available balance.""" stmt = select(PayoutRequest).where(PayoutRequest.id == payout_id).with_for_update() res = await db.execute(stmt) payout = res.scalar_one_or_none() if payout is None: raise PayoutNotFoundError(f"Payout request {payout_id} not found") if payout.status != PayoutStatus.pending: raise InvalidPayoutStateError( f"Cannot reject payout in status {payout.status.value}" ) now = datetime.now(timezone.utc) cb = await PurchaseService.get_or_create_creator_balance(db, payout.creator_id, for_update=True) cb.pending = max(0, cb.pending - payout.amount) cb.available += payout.amount cb.updated_at = now payout.status = PayoutStatus.rejected payout.reviewed_by = reviewer_id payout.reviewed_at = now await db.commit() await db.refresh(payout) return payout payout_service = PayoutService()