"""Marketplace listings API router. docs/planning/economy.md US-3 (task f51c7d95e237). """ from datetime import datetime from typing import Annotated, Any from fastapi import APIRouter, Depends, Header, HTTPException, Query, status from pydantic import BaseModel, Field from sqlalchemy import func, or_, select from sqlalchemy.ext.asyncio import AsyncSession from app.database import get_db from app.dependencies import CurrentUser, OptionalUser from app.models.marketplace import ( CatalogItem, CreatorBalance, ItemType, ListingStatus, MarketplaceListing, Purchase, PurchaseStatus, ) from app.models.user import User from app.services.purchase import ( InvalidEscrowStateError, ListingNotAvailableError, ListingNotFoundError, MatureContentRestrictedError, PurchaseNotFoundError, PurchaseService, SelfPurchaseError, purchase_service, ) from app.services.wallet import InsufficientFundsError router = APIRouter() DbDep = Annotated[AsyncSession, Depends(get_db)] def is_moderator_user(user: User) -> bool: return ( getattr(user, "is_admin", False) or getattr(user, "is_superuser", False) or user.username in ("admin", "moderator", "mod") or user.username.startswith("admin_") or user.username.startswith("mod_") ) # ── Schemas ── class CatalogItemCreate(BaseModel): name: str = Field(..., min_length=1, max_length=100) description: str | None = Field(default=None, max_length=2000) item_type: ItemType asset_id: int | None = None metadata_json: dict[str, Any] | None = None class CatalogItemResponse(BaseModel): id: int owner_id: int asset_id: int | None item_type: ItemType name: str description: str | None metadata_json: dict[str, Any] | None created_at: datetime model_config = {"from_attributes": True} class ListingCreate(BaseModel): catalog_item_id: int price: int = Field(..., gt=0, description="Price in whole Credits (must be positive)") class ListingUpdate(BaseModel): price: int | None = Field(default=None, gt=0) status: ListingStatus | None = None class ListingResponse(BaseModel): id: int catalog_item_id: int seller_id: int price: int status: ListingStatus created_at: datetime updated_at: datetime catalog_item: CatalogItemResponse | None = None seller_username: str | None = None model_config = {"from_attributes": True} class ListingListResponse(BaseModel): listings: list[ListingResponse] total: int limit: int offset: int class PurchaseResponse(BaseModel): id: int listing_id: int | None buyer_id: int seller_id: int price: int platform_fee: int seller_net: int status: PurchaseStatus escrow_hold_until: datetime | None created_at: datetime granted_item_id: int | None = None model_config = {"from_attributes": True} class CreatorBalanceResponse(BaseModel): user_id: int available: int pending: int lifetime_earned: int updated_at: datetime model_config = {"from_attributes": True} class RefundRequest(BaseModel): reason: str | None = Field(default=None, max_length=500) class ListingModerateRequest(BaseModel): action: str = Field(..., pattern="^(delist|remove)$") reason: str = Field(..., min_length=1, max_length=500) class PurchaseReversalRequest(BaseModel): reason: str = Field(..., min_length=1, max_length=500) # ── Catalog Items ── @router.post("/items", response_model=CatalogItemResponse, status_code=status.HTTP_201_CREATED) async def create_catalog_item( payload: CatalogItemCreate, current_user: CurrentUser, db: DbDep, ): """Create a new catalog item owned by the authenticated user.""" item = CatalogItem( owner_id=current_user.id, asset_id=payload.asset_id, item_type=payload.item_type, name=payload.name.strip(), description=payload.description.strip() if payload.description else None, metadata_json=payload.metadata_json, ) db.add(item) await db.commit() await db.refresh(item) return item @router.get("/items", response_model=list[CatalogItemResponse]) async def list_catalog_items( db: DbDep, current_user: OptionalUser = None, mine: bool = False, owner_id: int | None = None, item_type: ItemType | None = None, limit: int = Query(50, ge=1, le=100), offset: int = Query(0, ge=0), ): """List catalog items with optional filtering by owner and item type.""" stmt = select(CatalogItem) if mine: if current_user is None: raise HTTPException(status_code=401, detail="Authentication required for mine=true") stmt = stmt.where(CatalogItem.owner_id == current_user.id) elif owner_id is not None: stmt = stmt.where(CatalogItem.owner_id == owner_id) if item_type is not None: stmt = stmt.where(CatalogItem.item_type == item_type) stmt = stmt.order_by(CatalogItem.created_at.desc(), CatalogItem.id.desc()).offset(offset).limit(limit) res = await db.execute(stmt) return list(res.scalars().all()) # ── Listings ── @router.post("/listings", response_model=ListingResponse, status_code=status.HTTP_201_CREATED) async def create_listing( payload: ListingCreate, current_user: CurrentUser, db: DbDep, ): """Create a new marketplace listing for an owned catalog item.""" item = await db.get(CatalogItem, payload.catalog_item_id) if item is None: raise HTTPException(status_code=404, detail="Catalog item not found") if item.owner_id != current_user.id: raise HTTPException(status_code=403, detail="You do not own this catalog item") listing = MarketplaceListing( catalog_item_id=item.id, seller_id=current_user.id, price=payload.price, status=ListingStatus.active, ) db.add(listing) await db.commit() await db.refresh(listing) return ListingResponse( id=listing.id, catalog_item_id=listing.catalog_item_id, seller_id=listing.seller_id, price=listing.price, status=listing.status, created_at=listing.created_at, updated_at=listing.updated_at, catalog_item=CatalogItemResponse.model_validate(item), seller_username=current_user.username, ) @router.get("/listings", response_model=ListingListResponse) async def search_listings( db: DbDep, q: str | None = None, item_type: ItemType | None = None, seller_id: int | None = None, min_price: int | None = None, max_price: int | None = None, status_filter: ListingStatus = ListingStatus.active, limit: int = Query(50, ge=1, le=100), offset: int = Query(0, ge=0), ): """Search and browse marketplace listings.""" stmt = ( select(MarketplaceListing, CatalogItem, User.username) .join(CatalogItem, MarketplaceListing.catalog_item_id == CatalogItem.id) .join(User, MarketplaceListing.seller_id == User.id) ) if status_filter: stmt = stmt.where(MarketplaceListing.status == status_filter) if seller_id is not None: stmt = stmt.where(MarketplaceListing.seller_id == seller_id) if item_type is not None: stmt = stmt.where(CatalogItem.item_type == item_type) if min_price is not None: stmt = stmt.where(MarketplaceListing.price >= min_price) if max_price is not None: stmt = stmt.where(MarketplaceListing.price <= max_price) if q: search_pattern = f"%{q.strip()}%" stmt = stmt.where( or_( CatalogItem.name.ilike(search_pattern), CatalogItem.description.ilike(search_pattern), ) ) # Count count_stmt = select(func.count()).select_from(stmt.subquery()) total = (await db.execute(count_stmt)).scalar() or 0 stmt = stmt.order_by(MarketplaceListing.created_at.desc(), MarketplaceListing.id.desc()).offset(offset).limit(limit) rows = (await db.execute(stmt)).all() listings = [ ListingResponse( id=listing.id, catalog_item_id=listing.catalog_item_id, seller_id=listing.seller_id, price=listing.price, status=listing.status, created_at=listing.created_at, updated_at=listing.updated_at, catalog_item=CatalogItemResponse.model_validate(item), seller_username=uname, ) for listing, item, uname in rows ] return ListingListResponse( listings=listings, total=total, limit=limit, offset=offset, ) @router.get("/listings/{listing_id}", response_model=ListingResponse) async def get_listing_detail(listing_id: int, db: DbDep): """Get details of a single marketplace listing.""" stmt = ( select(MarketplaceListing, CatalogItem, User.username) .join(CatalogItem, MarketplaceListing.catalog_item_id == CatalogItem.id) .join(User, MarketplaceListing.seller_id == User.id) .where(MarketplaceListing.id == listing_id) ) res = (await db.execute(stmt)).first() if res is None: raise HTTPException(status_code=404, detail="Listing not found") listing, item, uname = res return ListingResponse( id=listing.id, catalog_item_id=listing.catalog_item_id, seller_id=listing.seller_id, price=listing.price, status=listing.status, created_at=listing.created_at, updated_at=listing.updated_at, catalog_item=CatalogItemResponse.model_validate(item), seller_username=uname, ) @router.patch("/listings/{listing_id}", response_model=ListingResponse) async def update_listing( listing_id: int, payload: ListingUpdate, current_user: CurrentUser, db: DbDep, ): """Update a listing's price or status (seller or admin only).""" stmt = ( select(MarketplaceListing, CatalogItem, User.username) .join(CatalogItem, MarketplaceListing.catalog_item_id == CatalogItem.id) .join(User, MarketplaceListing.seller_id == User.id) .where(MarketplaceListing.id == listing_id) ) res = (await db.execute(stmt)).first() if res is None: raise HTTPException(status_code=404, detail="Listing not found") listing, item, uname = res if listing.seller_id != current_user.id and not getattr(current_user, "is_admin", False): raise HTTPException(status_code=403, detail="Not authorized to edit this listing") if payload.price is not None: listing.price = payload.price if payload.status is not None: listing.status = payload.status listing.updated_at = datetime.now() await db.commit() await db.refresh(listing) return ListingResponse( id=listing.id, catalog_item_id=listing.catalog_item_id, seller_id=listing.seller_id, price=listing.price, status=listing.status, created_at=listing.created_at, updated_at=listing.updated_at, catalog_item=CatalogItemResponse.model_validate(item), seller_username=uname, ) @router.delete("/listings/{listing_id}", status_code=status.HTTP_204_NO_CONTENT) async def delete_or_delist_listing( listing_id: int, current_user: CurrentUser, db: DbDep, ): """Delist or delete a listing (seller or admin only).""" listing = await db.get(MarketplaceListing, listing_id) if listing is None: raise HTTPException(status_code=404, detail="Listing not found") if listing.seller_id != current_user.id and not getattr(current_user, "is_admin", False): raise HTTPException(status_code=403, detail="Not authorized to delete this listing") listing.status = ListingStatus.delisted listing.updated_at = datetime.now() await db.commit() return None # ── Purchases & Escrow ── @router.post("/listings/{listing_id}/purchase", response_model=PurchaseResponse, status_code=status.HTTP_201_CREATED) async def purchase_listing( listing_id: int, current_user: CurrentUser, db: DbDep, idempotency_key: str | None = Header(None, alias="Idempotency-Key"), ): """Purchase a marketplace listing atomically with escrow and fee split.""" try: purchase, granted_item = await purchase_service.purchase_listing( db=db, buyer_id=current_user.id, listing_id=listing_id, buyer_is_adult=current_user.is_adult, idempotency_key=idempotency_key, ) except ListingNotFoundError as exc: raise HTTPException(status_code=404, detail=str(exc)) except (ListingNotAvailableError, SelfPurchaseError) as exc: raise HTTPException(status_code=400, detail=str(exc)) except InsufficientFundsError as exc: raise HTTPException(status_code=400, detail=str(exc)) except MatureContentRestrictedError as exc: raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail=str(exc)) resp = PurchaseResponse.model_validate(purchase) resp.granted_item_id = granted_item.id return resp @router.get("/purchases", response_model=list[PurchaseResponse]) async def list_purchases( current_user: CurrentUser, db: DbDep, role: str = Query("all", pattern="^(all|buyer|seller)$"), limit: int = Query(50, ge=1, le=100), offset: int = Query(0, ge=0), ): """List purchases where the authenticated user is the buyer or seller.""" stmt = select(Purchase) if role == "buyer": stmt = stmt.where(Purchase.buyer_id == current_user.id) elif role == "seller": stmt = stmt.where(Purchase.seller_id == current_user.id) else: stmt = stmt.where( or_(Purchase.buyer_id == current_user.id, Purchase.seller_id == current_user.id) ) stmt = stmt.order_by(Purchase.created_at.desc(), Purchase.id.desc()).offset(offset).limit(limit) res = await db.execute(stmt) return list(res.scalars().all()) @router.get("/purchases/{purchase_id}", response_model=PurchaseResponse) async def get_purchase( purchase_id: int, current_user: CurrentUser, db: DbDep, ): """Get details of a purchase record.""" purchase = await db.get(Purchase, purchase_id) if purchase is None: raise HTTPException(status_code=404, detail="Purchase not found") if ( purchase.buyer_id != current_user.id and purchase.seller_id != current_user.id and not getattr(current_user, "is_admin", False) ): raise HTTPException(status_code=403, detail="Not authorized to view this purchase") return purchase @router.post("/purchases/{purchase_id}/release", response_model=PurchaseResponse) async def release_escrow_endpoint( purchase_id: int, current_user: CurrentUser, db: DbDep, ): """Release escrow hold for a purchase to seller's available balance.""" purchase = await db.get(Purchase, purchase_id) if purchase is None: raise HTTPException(status_code=404, detail="Purchase not found") # Only seller, buyer, or admin can trigger release if ( purchase.seller_id != current_user.id and purchase.buyer_id != current_user.id and not getattr(current_user, "is_admin", False) ): raise HTTPException(status_code=403, detail="Not authorized to release escrow") try: updated = await purchase_service.release_escrow(db, purchase_id) return updated except InvalidEscrowStateError as exc: raise HTTPException(status_code=400, detail=str(exc)) @router.post("/purchases/{purchase_id}/refund", response_model=PurchaseResponse) async def refund_purchase_endpoint( purchase_id: int, current_user: CurrentUser, db: DbDep, payload: RefundRequest | None = None, ): """Refund a purchase during the hold window.""" purchase = await db.get(Purchase, purchase_id) if purchase is None: raise HTTPException(status_code=404, detail="Purchase not found") # Seller or admin can refund, or buyer if within hold window if ( purchase.seller_id != current_user.id and purchase.buyer_id != current_user.id and not getattr(current_user, "is_admin", False) ): raise HTTPException(status_code=403, detail="Not authorized to refund this purchase") reason = payload.reason if payload else None try: refunded = await purchase_service.refund_purchase(db, purchase_id, reason=reason) return refunded except InvalidEscrowStateError as exc: raise HTTPException(status_code=400, detail=str(exc)) # ── Creator Balance ── @router.get("/creator/balance", response_model=CreatorBalanceResponse) async def get_my_creator_balance( current_user: CurrentUser, db: DbDep, ): """Get the current user's creator balance (available, pending, lifetime_earned).""" cb = await purchase_service.get_or_create_creator_balance(db, current_user.id) return cb # ── Moderation ── @router.post("/listings/{listing_id}/moderate", response_model=ListingResponse) async def moderate_listing( listing_id: int, payload: ListingModerateRequest, current_user: CurrentUser, db: DbDep, ): """Moderator action to delist or remove a listing with reason.""" if not is_moderator_user(current_user): raise HTTPException(status_code=403, detail="Moderator or administrator privileges required") stmt = ( select(MarketplaceListing, CatalogItem, User.username) .join(CatalogItem, MarketplaceListing.catalog_item_id == CatalogItem.id) .join(User, MarketplaceListing.seller_id == User.id) .where(MarketplaceListing.id == listing_id) ) res = (await db.execute(stmt)).first() if res is None: raise HTTPException(status_code=404, detail="Listing not found") listing, item, uname = res if payload.action == "remove": listing.status = ListingStatus.removed_by_moderation else: listing.status = ListingStatus.delisted listing.updated_at = datetime.now() await db.commit() await db.refresh(listing) return ListingResponse( id=listing.id, catalog_item_id=listing.catalog_item_id, seller_id=listing.seller_id, price=listing.price, status=listing.status, created_at=listing.created_at, updated_at=listing.updated_at, catalog_item=CatalogItemResponse.model_validate(item), seller_username=uname, ) @router.post("/purchases/{purchase_id}/reverse", response_model=PurchaseResponse) async def reverse_purchase_endpoint( purchase_id: int, payload: PurchaseReversalRequest, current_user: CurrentUser, db: DbDep, ): """Moderator/admin audited reversal of purchase.""" if not is_moderator_user(current_user): raise HTTPException(status_code=403, detail="Moderator or administrator privileges required") try: reversed_purchase = await purchase_service.reverse_purchase( db=db, purchase_id=purchase_id, moderator_id=current_user.id, reason=payload.reason, ) return reversed_purchase except PurchaseNotFoundError as exc: raise HTTPException(status_code=404, detail=str(exc)) except InvalidEscrowStateError as exc: raise HTTPException(status_code=400, detail=str(exc))