"""Marketplace purchase service: atomic escrow, fee split, item granting, refunds. docs/planning/economy.md US-4 (task bc90d3f481b4). """ from __future__ import annotations from datetime import datetime, timedelta, timezone import math from typing import Any from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from app.models.marketplace import ( CatalogItem, CreatorBalance, ListingStatus, MarketplaceListing, Purchase, PurchaseStatus, ) from app.models.wallet import TransactionType from app.services.wallet import InsufficientFundsError, WalletService MARKETPLACE_FEE_RATE = 0.30 # 30% platform fee DEFAULT_ESCROW_HOLD_HOURS = 72 class ListingNotFoundError(Exception): """Raised when the specified listing does not exist.""" pass class ListingNotAvailableError(Exception): """Raised when the listing is not in active status.""" pass class SelfPurchaseError(Exception): """Raised when a seller attempts to buy their own listing.""" pass class PurchaseNotFoundError(Exception): """Raised when the specified purchase does not exist.""" pass class InvalidEscrowStateError(Exception): """Raised when an operation cannot be performed on the current escrow state.""" pass class MatureContentRestrictedError(Exception): """Raised when an unverified minor attempts to purchase mature content.""" pass def is_item_mature(item: CatalogItem) -> bool: """Check whether a catalog item carries mature flags or is marked mature.""" if not item.metadata_json or not isinstance(item.metadata_json, dict): return False if item.metadata_json.get("is_mature") or item.metadata_json.get("mature"): return True flags = item.metadata_json.get("content_flags", []) if isinstance(flags, list): for f in flags: if isinstance(f, dict) and f.get("mature"): return True if isinstance(f, str) and any( m in f for m in ("gore", "sexual", "gambling", "hate", "explicit") ): return True return False class PurchaseService: @staticmethod def compute_fee_split(price: int, fee_rate: float = MARKETPLACE_FEE_RATE) -> tuple[int, int]: """Compute platform fee and seller net amount. Returns (platform_fee, seller_net). """ platform_fee = math.floor(price * fee_rate) seller_net = price - platform_fee return platform_fee, seller_net @classmethod async def get_or_create_creator_balance( cls, db: AsyncSession, user_id: int, for_update: bool = False ) -> CreatorBalance: """Fetch or lazily create a creator balance row for user_id.""" stmt = select(CreatorBalance).where(CreatorBalance.user_id == user_id) if for_update: stmt = stmt.with_for_update() result = await db.execute(stmt) cb = result.scalar_one_or_none() if cb is not None: return cb new_cb = CreatorBalance( user_id=user_id, available=0, pending=0, lifetime_earned=0, ) db.add(new_cb) await db.flush() return new_cb @classmethod async def purchase_listing( cls, db: AsyncSession, buyer_id: int, listing_id: int, buyer_is_adult: bool = True, idempotency_key: str | None = None, escrow_hours: int = DEFAULT_ESCROW_HOLD_HOURS, ) -> tuple[Purchase, CatalogItem]: """Perform atomic purchase of a marketplace listing. Debits buyer, holds escrow, grants item to buyer, records purchase, and updates creator pending balance. """ # 1. Lock and validate listing stmt_listing = ( select(MarketplaceListing) .where(MarketplaceListing.id == listing_id) .with_for_update() ) res_listing = await db.execute(stmt_listing) listing = res_listing.scalar_one_or_none() if listing is None: raise ListingNotFoundError(f"Listing {listing_id} not found") if listing.status != ListingStatus.active: raise ListingNotAvailableError(f"Listing {listing_id} is not active ({listing.status.value})") if listing.seller_id == buyer_id: raise SelfPurchaseError("Cannot purchase your own listing") # 2. Fetch catalog item catalog_item = await db.get(CatalogItem, listing.catalog_item_id) if catalog_item is None: raise ListingNotFoundError("Catalog item referenced by listing does not exist") if is_item_mature(catalog_item) and not buyer_is_adult: raise MatureContentRestrictedError( "MATURE_CONTENT_REQUIRES_VERIFICATION: Buyer must be 18+ to purchase mature items" ) # 3. Calculate fee split price = listing.price platform_fee, seller_net = cls.compute_fee_split(price) # 4. Debit buyer wallet atomically (commit=False) tx_idemp = f"mkt-purchase-{idempotency_key}" if idempotency_key else None await WalletService.debit( db=db, user_id=buyer_id, amount=price, tx_type=TransactionType.marketplace_purchase, reference_type="marketplace_listing", reference_id=str(listing.id), idempotency_key=tx_idemp, commit=False, ) # 5. Grant catalog item to buyer granted_item = CatalogItem( owner_id=buyer_id, asset_id=catalog_item.asset_id, item_type=catalog_item.item_type, name=catalog_item.name, description=catalog_item.description, metadata_json=catalog_item.metadata_json, ) db.add(granted_item) # 6. Mark listing as sold now = datetime.now(timezone.utc) listing.status = ListingStatus.sold listing.updated_at = now # 7. Create purchase record hold_until = now + timedelta(hours=escrow_hours) purchase = Purchase( listing_id=listing.id, buyer_id=buyer_id, seller_id=listing.seller_id, price=price, platform_fee=platform_fee, seller_net=seller_net, status=PurchaseStatus.pending_escrow, escrow_hold_until=hold_until, ) db.add(purchase) # 8. Update seller's CreatorBalance (pending escrow) creator_balance = await cls.get_or_create_creator_balance( db, listing.seller_id, for_update=True ) creator_balance.pending += seller_net creator_balance.updated_at = now # Commit everything atomically await db.commit() await db.refresh(purchase) await db.refresh(granted_item) return purchase, granted_item @classmethod async def release_escrow(cls, db: AsyncSession, purchase_id: int) -> Purchase: """Release funds from escrow to seller's available creator balance.""" stmt = select(Purchase).where(Purchase.id == purchase_id).with_for_update() res = await db.execute(stmt) purchase = res.scalar_one_or_none() if purchase is None: raise PurchaseNotFoundError(f"Purchase {purchase_id} not found") if purchase.status != PurchaseStatus.pending_escrow: raise InvalidEscrowStateError( f"Cannot release escrow in status {purchase.status.value}" ) cb = await cls.get_or_create_creator_balance(db, purchase.seller_id, for_update=True) cb.pending = max(0, cb.pending - purchase.seller_net) cb.available += purchase.seller_net cb.lifetime_earned += purchase.seller_net cb.updated_at = datetime.now(timezone.utc) purchase.status = PurchaseStatus.completed await db.commit() await db.refresh(purchase) return purchase @classmethod async def refund_purchase( cls, db: AsyncSession, purchase_id: int, reason: str | None = None ) -> Purchase: """Refund a purchase during the escrow hold window.""" stmt = select(Purchase).where(Purchase.id == purchase_id).with_for_update() res = await db.execute(stmt) purchase = res.scalar_one_or_none() if purchase is None: raise PurchaseNotFoundError(f"Purchase {purchase_id} not found") if purchase.status != PurchaseStatus.pending_escrow: raise InvalidEscrowStateError( f"Cannot refund purchase in status {purchase.status.value}" ) # Reverse seller's pending creator balance cb = await cls.get_or_create_creator_balance(db, purchase.seller_id, for_update=True) cb.pending = max(0, cb.pending - purchase.seller_net) cb.updated_at = datetime.now(timezone.utc) # Credit buyer wallet await WalletService.credit( db=db, user_id=purchase.buyer_id, amount=purchase.price, tx_type=TransactionType.refund, reference_type="purchase_refund", reference_id=str(purchase.id), metadata={"reason": reason} if reason else None, commit=False, ) purchase.status = PurchaseStatus.refunded await db.commit() await db.refresh(purchase) return purchase @classmethod async def reverse_purchase( cls, db: AsyncSession, purchase_id: int, moderator_id: int, reason: str, ) -> Purchase: """Moderator/admin audited reversal of purchase.""" stmt = select(Purchase).where(Purchase.id == purchase_id).with_for_update() res = await db.execute(stmt) purchase = res.scalar_one_or_none() if purchase is None: raise PurchaseNotFoundError(f"Purchase {purchase_id} not found") if purchase.status in (PurchaseStatus.reversed, PurchaseStatus.refunded): raise InvalidEscrowStateError(f"Purchase already {purchase.status.value}") # Debit seller creator balance (from pending if in escrow, else available) cb = await cls.get_or_create_creator_balance(db, purchase.seller_id, for_update=True) if purchase.status == PurchaseStatus.pending_escrow: cb.pending = max(0, cb.pending - purchase.seller_net) else: cb.available = max(0, cb.available - purchase.seller_net) cb.updated_at = datetime.now(timezone.utc) # Refund buyer wallet await WalletService.credit( db=db, user_id=purchase.buyer_id, amount=purchase.price, tx_type=TransactionType.refund, reference_type="purchase_reversal", reference_id=str(purchase.id), metadata={"moderator_id": moderator_id, "reason": reason}, commit=False, ) purchase.status = PurchaseStatus.reversed await db.commit() await db.refresh(purchase) return purchase purchase_service = PurchaseService()