324 lines
11 KiB
Python
324 lines
11 KiB
Python
"""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()
|