Merge branch 'agent/bc90d3f481b4-purchase-escrow-service' into main

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
+

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()

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()
+        if commit:
-        await db.refresh(wallet)
+            await db.commit()
-        await db.refresh(tx)
+            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()
+        if commit:
-        await db.refresh(wallet)
+            await db.commit()
-        await db.refresh(tx)
+            await db.refresh(wallet)
+            await db.refresh(tx)
+        else:
+            await db.flush()
         return wallet, tx
 
     @staticmethod

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