"""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 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, 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") # 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 purchase_service = PurchaseService()