diff --git a/backend/app/main.py b/backend/app/main.py index 7c46a9f..b7b32c0 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -6,7 +6,11 @@ from fastapi import FastAPI from fastapi.middleware.cors import CORSMiddleware from app.config import settings -from app.routers import auth, users, worlds, assets, avatar, world_content, internal, presence, groups, social_friends, profiles, chat_dm, chat_reports, chat_world +from app.routers import ( + auth, users, worlds, assets, avatar, world_content, internal, + presence, groups, social_friends, profiles, chat_dm, chat_reports, + chat_world, wallet, +) from app.gateway import ws @@ -54,6 +58,7 @@ app.include_router(chat_reports.chat_router, prefix="/api/chat", tags=["chat-rep app.include_router(chat_reports.moderation_router, prefix="/api/moderation", tags=["moderation"]) app.include_router(chat_world.router, prefix="/api/chat", tags=["chat-world"]) app.include_router(chat_world.router, prefix="/v1/chat", tags=["chat-world"]) +app.include_router(wallet.router, prefix="/api/wallet", tags=["wallet"]) app.include_router(ws.router, prefix="/api", tags=["websocket"]) app.include_router(ws.router, prefix="/v1", tags=["websocket"]) diff --git a/backend/app/routers/wallet.py b/backend/app/routers/wallet.py new file mode 100644 index 0000000..1267c0e --- /dev/null +++ b/backend/app/routers/wallet.py @@ -0,0 +1,78 @@ +"""Wallet & transactions API endpoints. + +docs/planning/economy.md US-1 (task 909e3e9a59af). +""" + +from datetime import datetime +from typing import Annotated, Any + +from fastapi import APIRouter, Depends, HTTPException, Query, status +from pydantic import BaseModel, Field +from sqlalchemy.ext.asyncio import AsyncSession + +from app.database import get_db +from app.dependencies import CurrentUser +from app.models.wallet import TransactionType +from app.services.wallet import wallet_service + +router = APIRouter() + +DbDep = Annotated[AsyncSession, Depends(get_db)] + + +class WalletResponse(BaseModel): + id: int + user_id: int + balance: int + created_at: datetime + updated_at: datetime + + model_config = {"from_attributes": True} + + +class TransactionResponse(BaseModel): + id: int + wallet_id: int + type: TransactionType + amount: int + balance_after: int + reference_type: str | None = None + reference_id: str | None = None + idempotency_key: str | None = None + metadata_json: dict[str, Any] | None = None + created_at: datetime + + model_config = {"from_attributes": True} + + +class TransactionListResponse(BaseModel): + transactions: list[TransactionResponse] + total: int + limit: int + offset: int + + +@router.get("", response_model=WalletResponse) +@router.get("/", response_model=WalletResponse) +async def get_wallet(current_user: CurrentUser, db: DbDep): + """Retrieve the current user's wallet balance and metadata.""" + wallet = await wallet_service.get_or_create_wallet(db, current_user.id) + return wallet + + +@router.get("/transactions", response_model=TransactionListResponse) +async def get_wallet_transactions( + current_user: CurrentUser, + db: DbDep, + limit: int = Query(50, ge=1, le=100), + offset: int = Query(0, ge=0), +): + """List the current user's ledger transactions, ordered newest-first.""" + wallet = await wallet_service.get_or_create_wallet(db, current_user.id) + txs, total = await wallet_service.get_transactions(db, wallet.id, limit=limit, offset=offset) + return TransactionListResponse( + transactions=txs, + total=total, + limit=limit, + offset=offset, + ) diff --git a/backend/app/services/wallet.py b/backend/app/services/wallet.py new file mode 100644 index 0000000..583ea24 --- /dev/null +++ b/backend/app/services/wallet.py @@ -0,0 +1,199 @@ +"""Wallet service: atomic credit/debit, row locking, and idempotency. + +docs/planning/economy.md US-1 (task 909e3e9a59af). +""" + +from __future__ import annotations + +import logging +from typing import Any + +from sqlalchemy import func, select +from sqlalchemy.exc import IntegrityError +from sqlalchemy.ext.asyncio import AsyncSession + +from app.models.wallet import Transaction, TransactionType, Wallet + +logger = logging.getLogger(__name__) + + +class InsufficientFundsError(Exception): + """Raised when a debit would cause wallet balance to drop below zero.""" + pass + + +class InvalidAmountError(ValueError): + """Raised when an invalid (non-positive) amount is supplied.""" + pass + + +class WalletService: + @staticmethod + async def get_or_create_wallet( + db: AsyncSession, user_id: int, for_update: bool = False + ) -> Wallet: + """Fetch existing wallet or lazily create one for user_id. + + If for_update is True, locks the row with SELECT ... FOR UPDATE. + """ + stmt = select(Wallet).where(Wallet.user_id == user_id) + if for_update: + stmt = stmt.with_for_update() + + result = await db.execute(stmt) + wallet = result.scalar_one_or_none() + if wallet is not None: + return wallet + + # If not found and not in update mode, create lazily + try: + new_wallet = Wallet(user_id=user_id, balance=0) + db.add(new_wallet) + await db.commit() + await db.refresh(new_wallet) + + if for_update: + stmt_locked = select(Wallet).where(Wallet.id == new_wallet.id).with_for_update() + res = await db.execute(stmt_locked) + return res.scalar_one() + + return new_wallet + except IntegrityError: + # Concurrent insert collision -- rollback and fetch the existing one + await db.rollback() + stmt = select(Wallet).where(Wallet.user_id == user_id) + if for_update: + stmt = stmt.with_for_update() + result = await db.execute(stmt) + return result.scalar_one() + + @classmethod + async def credit( + cls, + db: AsyncSession, + user_id: int, + amount: int, + tx_type: TransactionType | str, + reference_type: str | None = None, + reference_id: str | None = None, + idempotency_key: str | None = None, + metadata: dict[str, Any] | None = None, + ) -> tuple[Wallet, Transaction]: + """Credit a user's wallet atomically with row locking. + + If an idempotency_key is provided and already exists, returns the original + outcome without double-crediting. + """ + if amount <= 0: + raise InvalidAmountError("Credit amount must be strictly positive") + + if isinstance(tx_type, str): + tx_type = TransactionType(tx_type) + + # Check idempotency first + if idempotency_key: + stmt = select(Transaction).where(Transaction.idempotency_key == idempotency_key) + existing_tx = (await db.execute(stmt)).scalar_one_or_none() + if existing_tx is not None: + wallet = await cls.get_or_create_wallet(db, user_id, for_update=False) + return wallet, existing_tx + + # Lock wallet row + wallet = await cls.get_or_create_wallet(db, user_id, for_update=True) + wallet.balance += amount + + tx = Transaction( + wallet_id=wallet.id, + type=tx_type, + amount=amount, + balance_after=wallet.balance, + reference_type=reference_type, + reference_id=reference_id, + idempotency_key=idempotency_key, + metadata_json=metadata, + ) + db.add(tx) + await db.commit() + await db.refresh(wallet) + await db.refresh(tx) + return wallet, tx + + @classmethod + async def debit( + cls, + db: AsyncSession, + user_id: int, + amount: int, + tx_type: TransactionType | str, + reference_type: str | None = None, + reference_id: str | None = None, + idempotency_key: str | None = None, + metadata: dict[str, Any] | None = None, + ) -> tuple[Wallet, Transaction]: + """Debit a user's wallet atomically with row locking and balance check. + + Raises InsufficientFundsError if wallet balance < amount. + If idempotency_key is provided and already exists, returns the original + outcome without double-debiting. + """ + if amount <= 0: + raise InvalidAmountError("Debit amount must be strictly positive") + + if isinstance(tx_type, str): + tx_type = TransactionType(tx_type) + + # Check idempotency first + if idempotency_key: + stmt = select(Transaction).where(Transaction.idempotency_key == idempotency_key) + existing_tx = (await db.execute(stmt)).scalar_one_or_none() + if existing_tx is not None: + wallet = await cls.get_or_create_wallet(db, user_id, for_update=False) + return wallet, existing_tx + + # Lock wallet row + wallet = await cls.get_or_create_wallet(db, user_id, for_update=True) + cur_balance = wallet.balance + if cur_balance < amount: + await db.rollback() + raise InsufficientFundsError( + f"Insufficient funds: available {cur_balance}, required {amount}" + ) + + wallet.balance -= amount + + tx = Transaction( + wallet_id=wallet.id, + type=tx_type, + amount=-amount, + balance_after=wallet.balance, + reference_type=reference_type, + reference_id=reference_id, + idempotency_key=idempotency_key, + metadata_json=metadata, + ) + db.add(tx) + await db.commit() + await db.refresh(wallet) + await db.refresh(tx) + return wallet, tx + + @staticmethod + async def get_transactions( + db: AsyncSession, wallet_id: int, limit: int = 50, offset: int = 0 + ) -> tuple[list[Transaction], int]: + """Fetch paged transactions for a wallet ordered newest-first, plus total count.""" + count_stmt = select(func.count()).select_from(Transaction).where(Transaction.wallet_id == wallet_id) + total = (await db.execute(count_stmt)).scalar() or 0 + + stmt = ( + select(Transaction) + .where(Transaction.wallet_id == wallet_id) + .order_by(Transaction.created_at.desc(), Transaction.id.desc()) + .offset(offset) + .limit(limit) + ) + res = await db.execute(stmt) + return list(res.scalars().all()), total + + +wallet_service = WalletService() diff --git a/backend/tests/test_wallet_api.py b/backend/tests/test_wallet_api.py new file mode 100644 index 0000000..86ae9a3 --- /dev/null +++ b/backend/tests/test_wallet_api.py @@ -0,0 +1,135 @@ +"""Integration tests for Wallet service and API endpoints. + +docs/planning/economy.md US-1 (task 909e3e9a59af). +""" + +import pytest +from fastapi.testclient import TestClient +from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine +from sqlalchemy.pool import NullPool + +from app.config import settings +from app.main import app +from app.models.wallet import TransactionType +from app.services.wallet import InsufficientFundsError, InvalidAmountError, wallet_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 _session_factory(): + engine = create_async_engine(settings.DATABASE_URL, poolclass=NullPool) + return engine, async_sessionmaker(engine, expire_on_commit=False) + + +def test_wallet_unauthorized(): + client = TestClient(app) + resp = client.get("/api/wallet") + assert resp.status_code == 401 + + resp = client.get("/api/wallet/transactions") + assert resp.status_code == 401 + + +def test_get_wallet_lazy_creation(): + client = TestClient(app) + uid, headers = _register(client, "wallet_api_user_1") + + resp = client.get("/api/wallet", headers=headers) + assert resp.status_code == 200, resp.text + data = resp.json() + assert data["user_id"] == uid + assert data["balance"] == 0 + assert data["id"] is not None + + +@pytest.mark.asyncio +async def test_wallet_service_credit_and_debit_flow(): + client = TestClient(app) + uid, headers = _register(client, "wallet_api_user_2") + + engine, sf = _session_factory() + try: + async with sf() as session: + # 1. Credit + wallet, tx = await wallet_service.credit( + session, + user_id=uid, + amount=1000, + tx_type=TransactionType.credit_purchase, + reference_type="test_ref", + reference_id="ref_1", + idempotency_key="credit_idemp_1", + ) + assert wallet.balance == 1000 + assert tx.amount == 1000 + assert tx.balance_after == 1000 + + # 2. Idempotent re-credit does not increase balance + wallet2, tx2 = await wallet_service.credit( + session, + user_id=uid, + amount=1000, + tx_type=TransactionType.credit_purchase, + idempotency_key="credit_idemp_1", + ) + assert wallet2.balance == 1000 + assert tx2.id == tx.id + + # 3. Debit + wallet3, tx3 = await wallet_service.debit( + session, + user_id=uid, + amount=400, + tx_type=TransactionType.marketplace_purchase, + reference_type="purchase", + reference_id="item_42", + idempotency_key="debit_idemp_1", + ) + assert wallet3.balance == 600 + assert tx3.amount == -400 + assert tx3.balance_after == 600 + + # 4. Over-debit raises InsufficientFundsError + with pytest.raises(InsufficientFundsError): + await wallet_service.debit( + session, + user_id=uid, + amount=1000, + tx_type=TransactionType.marketplace_purchase, + ) + + # Balance remains 600 + w_check = await wallet_service.get_or_create_wallet(session, uid) + assert w_check.balance == 600 + finally: + await engine.dispose() + + # Verify via API + resp = client.get("/api/wallet", headers=headers) + assert resp.status_code == 200 + assert resp.json()["balance"] == 600 + + resp_tx = client.get("/api/wallet/transactions", headers=headers) + assert resp_tx.status_code == 200 + tx_data = resp_tx.json() + assert tx_data["total"] == 2 + assert len(tx_data["transactions"]) == 2 + # Newest first: debit (-400) first, then credit (+1000) + assert tx_data["transactions"][0]["amount"] == -400 + assert tx_data["transactions"][1]["amount"] == 1000