# app/modules/billing/routes/api/admin.py """ Billing module admin routes. Provides admin API endpoints for subscription and billing management: - Subscription tier CRUD - Merchant subscription listing and management - Billing history - Subscription statistics """ import logging from fastapi import APIRouter, Depends, Path, Query from sqlalchemy.orm import Session from app.api.deps import get_current_admin_api, require_module_access from app.core.database import get_db from app.modules.billing.schemas import ( BillingHistoryListResponse, BillingHistoryWithMerchant, MerchantSubscriptionAdminCreate, MerchantSubscriptionAdminResponse, MerchantSubscriptionAdminUpdate, MerchantSubscriptionListResponse, MerchantSubscriptionWithMerchant, SubscriptionStatsResponse, SubscriptionTierCreate, SubscriptionTierListResponse, SubscriptionTierResponse, SubscriptionTierUpdate, ) from app.modules.billing.services import ( admin_subscription_service, subscription_service, ) from app.modules.enums import FrontendType from app.modules.tenancy.schemas.auth import UserContext logger = logging.getLogger(__name__) # Admin router with module access control router = APIRouter( prefix="/subscriptions", dependencies=[Depends(require_module_access("billing", FrontendType.ADMIN))], ) # ============================================================================ # Subscription Tier Endpoints # ============================================================================ @router.get("/tiers", response_model=SubscriptionTierListResponse) def list_subscription_tiers( include_inactive: bool = Query(False, description="Include inactive tiers"), platform_id: int | None = Query(None, description="Filter tiers by platform"), current_user: UserContext = Depends(get_current_admin_api), db: Session = Depends(get_db), ): """List all subscription tiers.""" tiers = admin_subscription_service.get_tiers(db, include_inactive=include_inactive, platform_id=platform_id) platforms_map = admin_subscription_service.get_platform_names_map(db) tiers_response = [] for t in tiers: resp = SubscriptionTierResponse.model_validate(t) resp.platform_name = platforms_map.get(t.platform_id) if t.platform_id else None resp.feature_codes = sorted(t.get_feature_codes()) tiers_response.append(resp) return SubscriptionTierListResponse( tiers=tiers_response, total=len(tiers), ) @router.get("/tiers/{tier_code}", response_model=SubscriptionTierResponse) def get_subscription_tier( tier_code: str = Path(..., description="Tier code"), current_user: UserContext = Depends(get_current_admin_api), db: Session = Depends(get_db), ): """Get a specific subscription tier by code.""" tier = admin_subscription_service.get_tier_by_code(db, tier_code) resp = SubscriptionTierResponse.model_validate(tier) resp.feature_codes = sorted(tier.get_feature_codes()) return resp @router.post("/tiers", response_model=SubscriptionTierResponse, status_code=201) def create_subscription_tier( tier_data: SubscriptionTierCreate, current_user: UserContext = Depends(get_current_admin_api), db: Session = Depends(get_db), ): """Create a new subscription tier.""" tier = admin_subscription_service.create_tier(db, tier_data.model_dump()) db.commit() db.refresh(tier) resp = SubscriptionTierResponse.model_validate(tier) resp.feature_codes = sorted(tier.get_feature_codes()) return resp @router.patch("/tiers/{tier_code}", response_model=SubscriptionTierResponse) def update_subscription_tier( tier_data: SubscriptionTierUpdate, tier_code: str = Path(..., description="Tier code"), current_user: UserContext = Depends(get_current_admin_api), db: Session = Depends(get_db), ): """Update a subscription tier.""" update_data = tier_data.model_dump(exclude_unset=True) tier = admin_subscription_service.update_tier(db, tier_code, update_data) db.commit() db.refresh(tier) resp = SubscriptionTierResponse.model_validate(tier) resp.feature_codes = sorted(tier.get_feature_codes()) return resp @router.delete("/tiers/{tier_code}", status_code=204) def delete_subscription_tier( tier_code: str = Path(..., description="Tier code"), current_user: UserContext = Depends(get_current_admin_api), db: Session = Depends(get_db), ): """Soft-delete a subscription tier.""" admin_subscription_service.deactivate_tier(db, tier_code) db.commit() # ============================================================================ # Merchant Subscription Endpoints # ============================================================================ @router.get("", response_model=MerchantSubscriptionListResponse) def list_merchant_subscriptions( page: int = Query(1, ge=1), per_page: int = Query(20, ge=1, le=100), status: str | None = Query(None, description="Filter by status"), tier: str | None = Query(None, description="Filter by tier code"), search: str | None = Query(None, description="Search merchant name"), current_user: UserContext = Depends(get_current_admin_api), db: Session = Depends(get_db), ): """List all merchant subscriptions with filtering.""" data = admin_subscription_service.list_subscriptions( db, page=page, per_page=per_page, status=status, tier=tier, search=search ) platforms_map = admin_subscription_service.get_platform_names_map(db) subscriptions = [] for sub, merchant in data["results"]: sub_resp = MerchantSubscriptionAdminResponse.model_validate(sub) tier_name = sub.tier.name if sub.tier else None subscriptions.append( MerchantSubscriptionWithMerchant( **sub_resp.model_dump(), merchant_name=merchant.name, platform_name=platforms_map.get(sub.platform_id, ""), tier_name=tier_name, ) ) return MerchantSubscriptionListResponse( subscriptions=subscriptions, total=data["total"], page=data["page"], per_page=data["per_page"], pages=data["pages"], ) @router.get("/merchants/{merchant_id}") def get_merchant_subscriptions( merchant_id: int = Path(..., description="Merchant ID"), current_user: UserContext = Depends(get_current_admin_api), db: Session = Depends(get_db), ): """Get all subscriptions for a merchant with tier info and feature usage.""" results = admin_subscription_service.get_merchant_subscriptions_with_usage( db, merchant_id ) return {"subscriptions": results} # noqa: API001 @router.post( "/merchants/{merchant_id}/platforms/{platform_id}", response_model=MerchantSubscriptionAdminResponse, status_code=201, ) def create_merchant_subscription( create_data: MerchantSubscriptionAdminCreate, merchant_id: int = Path(..., description="Merchant ID"), platform_id: int = Path(..., description="Platform ID"), current_user: UserContext = Depends(get_current_admin_api), db: Session = Depends(get_db), ): """Create a subscription for a merchant on a platform.""" sub = subscription_service.get_or_create_subscription( db, merchant_id=merchant_id, platform_id=platform_id, tier_code=create_data.tier_code, trial_days=create_data.trial_days, ) # Update status if not trial if create_data.status != "trial": sub.status = create_data.status sub.is_annual = create_data.is_annual db.commit() db.refresh(sub) logger.info( f"Admin created subscription for merchant {merchant_id} " f"on platform {platform_id}: tier={create_data.tier_code}" ) return MerchantSubscriptionAdminResponse.model_validate(sub) @router.get( "/merchants/{merchant_id}/platforms/{platform_id}", response_model=MerchantSubscriptionAdminResponse, ) def get_merchant_subscription( merchant_id: int = Path(..., description="Merchant ID"), platform_id: int = Path(..., description="Platform ID"), current_user: UserContext = Depends(get_current_admin_api), db: Session = Depends(get_db), ): """Get subscription details for a specific merchant on a platform.""" sub, merchant = admin_subscription_service.get_subscription( db, merchant_id, platform_id ) return MerchantSubscriptionAdminResponse.model_validate(sub) @router.patch( "/merchants/{merchant_id}/platforms/{platform_id}", response_model=MerchantSubscriptionAdminResponse, ) def update_merchant_subscription( update_data: MerchantSubscriptionAdminUpdate, merchant_id: int = Path(..., description="Merchant ID"), platform_id: int = Path(..., description="Platform ID"), current_user: UserContext = Depends(get_current_admin_api), db: Session = Depends(get_db), ): """Update a merchant's subscription.""" data = update_data.model_dump(exclude_unset=True) sub, merchant = admin_subscription_service.update_subscription( db, merchant_id, platform_id, data ) db.commit() db.refresh(sub) return MerchantSubscriptionAdminResponse.model_validate(sub) # ============================================================================ # Store Convenience Endpoint # ============================================================================ @router.get("/store/{store_id}") def get_subscription_for_store( store_id: int = Path(..., description="Store ID"), current_user: UserContext = Depends(get_current_admin_api), db: Session = Depends(get_db), ): """ Get subscriptions + feature usage for a store (resolves to merchant). Convenience endpoint for the admin store detail page. Resolves store -> merchant -> all platform subscriptions and returns a list of subscription entries with feature usage metrics. """ results = admin_subscription_service.get_subscriptions_for_store(db, store_id) return {"subscriptions": results} # noqa: API001 # ============================================================================ # Statistics Endpoints # ============================================================================ @router.get("/stats", response_model=SubscriptionStatsResponse) def get_subscription_stats( current_user: UserContext = Depends(get_current_admin_api), db: Session = Depends(get_db), ): """Get subscription statistics for admin dashboard.""" stats = admin_subscription_service.get_stats(db) return SubscriptionStatsResponse(**stats) # ============================================================================ # Billing History Endpoints # ============================================================================ @router.get("/billing/history", response_model=BillingHistoryListResponse) def list_billing_history( page: int = Query(1, ge=1), per_page: int = Query(20, ge=1, le=100), merchant_id: int | None = Query(None, description="Filter by merchant"), status: str | None = Query(None, description="Filter by status"), current_user: UserContext = Depends(get_current_admin_api), db: Session = Depends(get_db), ): """List billing history (invoices) across all merchants.""" data = admin_subscription_service.list_billing_history( db, page=page, per_page=per_page, merchant_id=merchant_id, status=status ) invoices = [] for invoice, merchant in data["results"]: invoices.append( BillingHistoryWithMerchant( id=invoice.id, merchant_id=invoice.merchant_id, stripe_invoice_id=invoice.stripe_invoice_id, invoice_number=invoice.invoice_number, invoice_date=invoice.invoice_date, due_date=invoice.due_date, subtotal_cents=invoice.subtotal_cents, tax_cents=invoice.tax_cents, total_cents=invoice.total_cents, amount_paid_cents=invoice.amount_paid_cents, currency=invoice.currency, status=invoice.status, invoice_pdf_url=invoice.invoice_pdf_url, hosted_invoice_url=invoice.hosted_invoice_url, description=invoice.description, created_at=invoice.created_at, merchant_name=merchant.name, ) ) return BillingHistoryListResponse( invoices=invoices, total=data["total"], page=data["page"], per_page=data["per_page"], pages=data["pages"], ) # ============================================================================ # Aggregate Feature Management Routes # ============================================================================ # Include the features router to aggregate all billing-related admin routes from app.modules.billing.routes.api.admin_features import admin_features_router router.include_router(admin_features_router, tags=["admin-features"])