From 5e1a3bb16ed8189005c609ec65eecc8832c4c5d5 Mon Sep 17 00:00:00 2001 From: daddy Date: Tue, 15 Sep 2026 20:41:07 +0000 Subject: [PATCH] feat(marketplace): purchase service with atomic escrow, 30/70 fee splitting and refunds --- backend/app/routers/marketplace.py | 186 ++++++++++++++- backend/app/services/purchase.py | 251 ++++++++++++++++++++ backend/app/services/wallet.py | 23 +- backend/tests/test_marketplace_purchases.py | 222 +++++++++++++++++ 4 files changed, 674 insertions(+), 8 deletions(-) create mode 100644 backend/app/services/purchase.py create mode 100644 backend/tests/test_marketplace_purchases.py diff --git a/backend/app/routers/marketplace.py b/backend/app/routers/marketplace.py index 78b80f6..e50e215 100644 --- a/backend/app/routers/marketplace.py +++ b/backend/app/routers/marketplace.py @@ -6,7 +6,7 @@ docs/planning/economy.md US-3 (task f51c7d95e237). from datetime import datetime from typing import Annotated, Any -from fastapi import APIRouter, Depends, HTTPException, Query, status +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 @@ -15,11 +15,24 @@ 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, + PurchaseNotFoundError, + PurchaseService, + SelfPurchaseError, + purchase_service, +) +from app.services.wallet import InsufficientFundsError router = APIRouter() @@ -80,6 +93,36 @@ class ListingListResponse(BaseModel): 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) + + # ── Catalog Items ── @router.post("/items", response_model=CatalogItemResponse, status_code=status.HTTP_201_CREATED) @@ -326,3 +369,144 @@ async def delete_or_delist_listing( 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, + 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)) + + 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 + diff --git a/backend/app/services/purchase.py b/backend/app/services/purchase.py new file mode 100644 index 0000000..fb7b029 --- /dev/null +++ b/backend/app/services/purchase.py @@ -0,0 +1,251 @@ +"""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() diff --git a/backend/app/services/wallet.py b/backend/app/services/wallet.py index 583ea24..ae69527 100644 --- a/backend/app/services/wallet.py +++ b/backend/app/services/wallet.py @@ -78,6 +78,7 @@ class WalletService: reference_id: str | None = None, idempotency_key: str | None = None, metadata: dict[str, Any] | None = None, + commit: bool = True, ) -> tuple[Wallet, Transaction]: """Credit a user's wallet atomically with row locking. @@ -113,9 +114,12 @@ class WalletService: metadata_json=metadata, ) db.add(tx) - await db.commit() - await db.refresh(wallet) - await db.refresh(tx) + if commit: + await db.commit() + await db.refresh(wallet) + await db.refresh(tx) + else: + await db.flush() return wallet, tx @classmethod @@ -129,6 +133,7 @@ class WalletService: reference_id: str | None = None, idempotency_key: str | None = None, metadata: dict[str, Any] | None = None, + commit: bool = True, ) -> tuple[Wallet, Transaction]: """Debit a user's wallet atomically with row locking and balance check. @@ -154,7 +159,8 @@ class WalletService: wallet = await cls.get_or_create_wallet(db, user_id, for_update=True) cur_balance = wallet.balance if cur_balance < amount: - await db.rollback() + if commit: + await db.rollback() raise InsufficientFundsError( f"Insufficient funds: available {cur_balance}, required {amount}" ) @@ -172,9 +178,12 @@ class WalletService: metadata_json=metadata, ) db.add(tx) - await db.commit() - await db.refresh(wallet) - await db.refresh(tx) + if commit: + await db.commit() + await db.refresh(wallet) + await db.refresh(tx) + else: + await db.flush() return wallet, tx @staticmethod diff --git a/backend/tests/test_marketplace_purchases.py b/backend/tests/test_marketplace_purchases.py new file mode 100644 index 0000000..1cb4d2f --- /dev/null +++ b/backend/tests/test_marketplace_purchases.py @@ -0,0 +1,222 @@ +"""Integration tests for Marketplace Purchase Service, Escrow, and Fee Splitting. + +docs/planning/economy.md US-4 (task bc90d3f481b4). +""" + +import pytest +from fastapi.testclient import TestClient + +from app.main import app +from app.services.purchase import purchase_service + + +def _register(client: TestClient, username: str) -> tuple[int, dict[str, str]]: + resp = client.post( + "/api/auth/register", + json={ + "username": username, + "email": f"{username}@example.com", + "password": "secret123", + "date_of_birth": "1990-01-01", + }, + ) + assert resp.status_code == 201, resp.text + token = resp.json()["access_token"] + headers = {"Authorization": f"Bearer {token}"} + me = client.get("/api/users/me", headers=headers) + assert me.status_code == 200, me.text + return me.json()["id"], headers + + +def test_fee_split_calculation(): + # 30% platform fee + fee, net = purchase_service.compute_fee_split(100) + assert fee == 30 + assert net == 70 + + fee, net = purchase_service.compute_fee_split(250) + assert fee == 75 + assert net == 175 + + +def test_marketplace_purchase_atomic_flow(): + client = TestClient(app) + seller_id, seller_headers = _register(client, "purchase_seller_1") + buyer_id, buyer_headers = _register(client, "purchase_buyer_1") + + # 1. Seller creates catalog item and lists it for 200 Credits + item_resp = client.post( + "/api/marketplace/items", + headers=seller_headers, + json={ + "name": "Dragon Armor", + "description": "Legendary chestplate", + "item_type": "avatar_item", + }, + ) + assert item_resp.status_code == 201, item_resp.text + item_id = item_resp.json()["id"] + + list_resp = client.post( + "/api/marketplace/listings", + headers=seller_headers, + json={ + "catalog_item_id": item_id, + "price": 200, + }, + ) + assert list_resp.status_code == 201, list_resp.text + listing_id = list_resp.json()["id"] + + # 2. Seller cannot purchase own listing + self_buy = client.post( + f"/api/marketplace/listings/{listing_id}/purchase", + headers=seller_headers, + ) + assert self_buy.status_code == 400 + assert "own listing" in self_buy.text + + # 3. Buyer attempts purchase without enough credits -> 400 + poor_buy = client.post( + f"/api/marketplace/listings/{listing_id}/purchase", + headers=buyer_headers, + ) + assert poor_buy.status_code == 400 + assert "Insufficient funds" in poor_buy.text + + # 4. Buyer purchases 500 Credits via payments API + sess_resp = client.post( + "/api/payments/purchase", + headers=buyer_headers, + json={"credits": 500}, + ) + assert sess_resp.status_code == 201, sess_resp.text + sess_id = sess_resp.json()["session_id"] + + conf_resp = client.post( + "/api/payments/confirm", + headers=buyer_headers, + json={"session_id": sess_id}, + ) + assert conf_resp.status_code == 200, conf_resp.text + assert conf_resp.json()["new_balance"] == 500 + + # 5. Buyer purchases listing for 200 Credits + buy_headers = dict(buyer_headers) + buy_headers["Idempotency-Key"] = "buy-test-key-1" + buy_resp = client.post( + f"/api/marketplace/listings/{listing_id}/purchase", + headers=buy_headers, + ) + assert buy_resp.status_code == 201, buy_resp.text + purchase_data = buy_resp.json() + assert purchase_data["price"] == 200 + assert purchase_data["platform_fee"] == 60 + assert purchase_data["seller_net"] == 140 + assert purchase_data["status"] == "pending_escrow" + assert purchase_data["buyer_id"] == buyer_id + assert purchase_data["seller_id"] == seller_id + purchase_id = purchase_data["id"] + + # 6. Check buyer wallet balance: 500 - 200 = 300 + w_resp = client.get("/api/wallet", headers=buyer_headers) + assert w_resp.status_code == 200 + assert w_resp.json()["balance"] == 300 + + # 7. Check buyer has granted item in inventory + inv_resp = client.get("/api/marketplace/items", headers=buyer_headers, params={"mine": "true"}) + assert inv_resp.status_code == 200 + my_items = inv_resp.json() + assert any(i["name"] == "Dragon Armor" and i["owner_id"] == buyer_id for i in my_items) + + # 8. Listing is now sold and cannot be purchased again + re_buy = client.post( + f"/api/marketplace/listings/{listing_id}/purchase", + headers=buyer_headers, + ) + assert re_buy.status_code == 400 + assert "not active" in re_buy.text + + # 9. Seller's creator balance is pending escrow + cb_resp = client.get("/api/marketplace/creator/balance", headers=seller_headers) + assert cb_resp.status_code == 200, cb_resp.text + cb = cb_resp.json() + assert cb["pending"] == 140 + assert cb["available"] == 0 + + # 10. List purchases + p_resp = client.get("/api/marketplace/purchases", headers=buyer_headers) + assert p_resp.status_code == 200 + assert any(p["id"] == purchase_id for p in p_resp.json()) + + # 11. Escrow release + rel_resp = client.post( + f"/api/marketplace/purchases/{purchase_id}/release", + headers=seller_headers, + ) + assert rel_resp.status_code == 200, rel_resp.text + assert rel_resp.json()["status"] == "completed" + + # Check seller's creator balance now shows available = 140 + cb_resp2 = client.get("/api/marketplace/creator/balance", headers=seller_headers) + assert cb_resp2.status_code == 200 + cb2 = cb_resp2.json() + assert cb2["pending"] == 0 + assert cb2["available"] == 140 + assert cb2["lifetime_earned"] == 140 + + +def test_marketplace_refund(): + client = TestClient(app) + seller_id, seller_headers = _register(client, "refund_seller_1") + buyer_id, buyer_headers = _register(client, "refund_buyer_1") + + # Item & Listing + item_resp = client.post( + "/api/marketplace/items", + headers=seller_headers, + json={"name": "Magic Ring", "item_type": "avatar_item"}, + ) + item_id = item_resp.json()["id"] + + list_resp = client.post( + "/api/marketplace/listings", + headers=seller_headers, + json={"catalog_item_id": item_id, "price": 100}, + ) + listing_id = list_resp.json()["id"] + + # Fund buyer + s_resp = client.post("/api/payments/purchase", headers=buyer_headers, json={"credits": 500}) + assert s_resp.status_code == 201, s_resp.text + client.post("/api/payments/confirm", headers=buyer_headers, json={"session_id": s_resp.json()["session_id"]}) + + # Purchase + buy_resp = client.post(f"/api/marketplace/listings/{listing_id}/purchase", headers=buyer_headers) + assert buy_resp.status_code == 201 + purchase_id = buy_resp.json()["id"] + + # Buyer balance is 400 + w1 = client.get("/api/wallet", headers=buyer_headers).json() + assert w1["balance"] == 400 + + # Seller pending is 70 + cb1 = client.get("/api/marketplace/creator/balance", headers=seller_headers).json() + assert cb1["pending"] == 70 + + # Refund + ref_resp = client.post( + f"/api/marketplace/purchases/{purchase_id}/refund", + headers=seller_headers, + json={"reason": "Buyer dissatisfaction"}, + ) + assert ref_resp.status_code == 200, ref_resp.text + assert ref_resp.json()["status"] == "refunded" + + # Buyer balance refunded back to 500 + w2 = client.get("/api/wallet", headers=buyer_headers).json() + assert w2["balance"] == 500 + + # Seller pending reset to 0 + cb2 = client.get("/api/marketplace/creator/balance", headers=seller_headers).json() + assert cb2["pending"] == 0