from __future__ import annotations import logging import math from datetime import datetime, timezone from typing import List, Optional logger = logging.getLogger(__name__) from fastapi import APIRouter, Depends, HTTPException, Query, Request, status from pydantic import BaseModel from sqlalchemy import select, func from sqlalchemy.orm import Session from src.apps.auth.utils.auth import get_current_superuser from src.apps.admin.schemas.admin_common import ( MerchantUpdateRequest, MerchantCreateRequest, AdminUserCreate, AdminUserUpdate, ) from src.core.database import get_session, get_db class AdminCreateApiKeyRequest(BaseModel): display_name: str environment: str = "live" scopes: List[str] router = APIRouter() # ─── Seed ──────────────────────────────────────────────────────────────────── @router.post("/seed", tags=["ADMIN"]) async def seed_admin_user(db: Session = Depends(get_session)): """ One-time seed endpoint — creates the default superuser if none exists. Unauthenticated. Returns 409 if an admin user is already present. """ from src.apps.users.models.user import User from src.core.utils.password import encrypt_password import uuid as uuid_module existing = db.execute( select(User).where(User.is_superuser == True, User.deleted_at.is_(None)) ).scalar_one_or_none() if existing: raise HTTPException( status_code=409, detail="Admin user already exists. Seed can only be called once.", ) DEFAULT_EMAIL = "admin@hubwallet.com" DEFAULT_PASSWORD = "Admin@HubWallet1" hashed = encrypt_password(DEFAULT_PASSWORD) if not hashed: raise HTTPException(status_code=500, detail="Password hashing failed") hashed_str = hashed.decode() if isinstance(hashed, bytes) else hashed username = "admin_" + str(uuid_module.uuid4())[:8] user = User( email=DEFAULT_EMAIL, username=username, hashed_password=hashed_str, first_name="Super", last_name="Admin", is_superuser=True, is_active=True, is_verified=True, ) db.add(user) db.commit() db.refresh(user) return { "seeded": True, "email": DEFAULT_EMAIL, "password": DEFAULT_PASSWORD, "note": "Change this password immediately after first login.", } def get_ip(request: Request) -> str: forwarded = request.headers.get("X-Forwarded-For") if forwarded: # Use the rightmost value — set by the outermost trusted proxy, not spoofable by client parts = [p.strip() for p in forwarded.split(",")] return parts[-1] return request.client.host if request.client else "" def _model_to_dict(obj) -> dict: """Convert a SQLAlchemy model instance to a plain dict (scalar columns only).""" result = {} for col in obj.__table__.columns: result[col.name] = getattr(obj, col.name) return result def _escape_like(value: str) -> str: """Escape LIKE special characters to prevent query amplification.""" return value.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_") def _get_redis(): """Return a Redis client if available, else None.""" try: import redis as redis_lib from src.core.config import settings url = getattr(settings, "REDIS_URL", None) or getattr(settings, "CELERY_RESULT_BACKEND", None) if url and "redis" in url: return redis_lib.from_url(url, decode_responses=True) except Exception: pass return None def _redis_get(key: str): """Retrieve a value from Redis, return None on miss or error.""" try: client = _get_redis() if client: return client.get(key) except Exception: pass return None def _redis_set(key: str, value, ttl: int = 300) -> None: """Store a JSON-serialisable value in Redis with TTL seconds.""" try: import json client = _get_redis() if client: client.set(key, json.dumps(value, default=str), ex=ttl) except Exception: pass # ─── Dashboard ────────────────────────────────────────────────────────────── @router.get("/dashboard/stats") async def dashboard_stats( current_user=Depends(get_current_superuser), db: Session = Depends(get_session), ): from src.apps.admin.services import get_dashboard_stats return get_dashboard_stats(db) # ─── Merchant Management ───────────────────────────────────────────────────── @router.post("/merchants", status_code=201) async def create_merchant( body: MerchantCreateRequest, request: Request, current_user=Depends(get_current_superuser), db: Session = Depends(get_session), ): """ Admin: create a new merchant and their owner user in one step. The merchant is immediately active and ready to log in to the portal. """ from src.apps.users.models.user import User from src.apps.merchants.models.merchant import Merchant from src.apps.merchants.models.merchant_users import MerchantUsers from src.apps.admin.models.admin_audit_log import AdminAuditLog from src.apps.admin.services import create_merchant_with_owner from src.core.utils.password import encrypt_password from src.apps.base.utils.functions import generate_secure_id import uuid as uuid_module # Password validation if body.password != body.confirm_password: raise HTTPException(status_code=400, detail="Passwords do not match") pwd = body.password if len(pwd) < 8: raise HTTPException(status_code=400, detail="Password must be at least 8 characters") if not any(c.isupper() for c in pwd): raise HTTPException(status_code=400, detail="Password must contain an uppercase letter") if not any(c.islower() for c in pwd): raise HTTPException(status_code=400, detail="Password must contain a lowercase letter") if not any(c.isdigit() for c in pwd): raise HTTPException(status_code=400, detail="Password must contain a digit") # Email uniqueness check existing = db.execute( select(User).where(User.email == body.owner_email, User.deleted_at.is_(None)) ).scalar_one_or_none() if existing: raise HTTPException(status_code=409, detail="A user with this email already exists") now = datetime.now(timezone.utc) hashed = encrypt_password(pwd) if not hashed: raise HTTPException(status_code=500, detail="Password hashing failed") hashed_str = hashed.decode() if isinstance(hashed, bytes) else hashed # Create Merchant — active, verified, onboarded so they can log in immediately uin = generate_secure_id(prepend="mrc", length=12) merchant = Merchant( name=body.merchant_name, email=body.merchant_email or body.owner_email, uin=uin, is_active=True, is_verified=True, is_onboarded=True, approved_by=current_user.id, approved_at=now, ) db.add(merchant) db.flush() # Create owner User username = body.owner_email.split("@")[0] + "_" + str(uuid_module.uuid4())[:8] user_id_str = generate_secure_id(prepend="usr", length=20) owner_user = User( email=body.owner_email, username=username, hashed_password=hashed_str, first_name=body.first_name, last_name=body.last_name, full_name=f"{body.first_name} {body.last_name}".strip(), is_active=True, is_verified=True, is_superuser=False, user_type="merchant", merchant_id=merchant.id, user_id=user_id_str, ) db.add(owner_user) db.flush() # Link as owner in merchant_users mu = MerchantUsers( user_id=owner_user.id, merchant_id=merchant.id, is_owner=True, ) db.add(mu) db.flush() # Bootstrap RBAC roles for the new merchant create_merchant_with_owner(db, merchant, owner_user) AdminAuditLog.log( db, admin_user_id=current_user.id, action="merchant.create", target_type="merchant", target_id=merchant.id, metadata={"merchant_name": merchant.name, "owner_email": owner_user.email}, ip_address=get_ip(request), ) db.commit() # Seed default category for the new merchant try: from src.apps.product_categories.seeder import seed_default_categories await seed_default_categories(merchant_id=merchant.id, db=db) except Exception as _seed_err: logger.warning(f"Failed to seed default categories for merchant {merchant.id}: {_seed_err}") return { "data": { "id": merchant.id, "uin": merchant.uin, "name": merchant.name, "email": merchant.email, "status": merchant.status, "owner": { "id": owner_user.id, "email": owner_user.email, "first_name": owner_user.first_name, "last_name": owner_user.last_name, }, }, "success": True, "message": "Merchant created successfully", "status_code": 201, } @router.get("/merchants") async def list_merchants( page: int = Query(1, ge=1), per_page: int = Query(20, ge=1, le=100), search: Optional[str] = Query(None), status: Optional[str] = Query(None), is_active: Optional[bool] = Query(None), is_onboarded: Optional[bool] = Query(None), current_user=Depends(get_current_superuser), db: Session = Depends(get_session), ): from src.apps.admin.crud import get_merchants_paginated items, total = get_merchants_paginated( db, page=page, per_page=per_page, search=search, status=status, is_active=is_active, is_onboarded=is_onboarded, ) # Batch-compute invoice stats per merchant (PAID + PARTIALLY_PAID only) from sqlalchemy import select as sa_select, func as sa_func from src.apps.invoices.models.invoice import Invoice as InvoiceModel from src.core.utils.enums import InvoiceStatusTypes merchant_ids = [m.id for m in items] invoice_stats: dict = {} if merchant_ids: _PAID_STATUSES = (InvoiceStatusTypes.PAID, InvoiceStatusTypes.PARTIALLY_PAID) rows = db.execute( sa_select( InvoiceModel.merchant_id, sa_func.count(InvoiceModel.id).label("invoice_count"), sa_func.coalesce(sa_func.sum(InvoiceModel.paid_amount), 0).label("total_sales"), ).where( InvoiceModel.merchant_id.in_(merchant_ids), InvoiceModel.status.in_(_PAID_STATUSES), InvoiceModel.deleted_at == None, ).group_by(InvoiceModel.merchant_id) ).all() for row in rows: invoice_stats[row.merchant_id] = { "invoice_count": row.invoice_count, "invoice_total_sales": float(row.total_sales), } merchant_items = [] for m in items: # Filter by status hybrid property after fetch if status filter provided if status and m.status != status: continue stats = invoice_stats.get(m.id, {"invoice_count": 0, "invoice_total_sales": 0.0}) merchant_items.append({ "id": m.id, "name": m.name, "email": m.email, "uin": m.uin, "is_active": m.is_active, "is_onboarded": m.is_onboarded, "status": m.status, "created_at": m.created_at, "invoice_count": stats["invoice_count"], "invoice_total_sales": stats["invoice_total_sales"], }) total_pages = math.ceil(total / per_page) if total > 0 else 1 return { "data": { "items": merchant_items, "meta": {"page": page, "per_page": per_page, "total": total, "total_pages": total_pages}, }, "success": True, "message": "Merchants fetched successfully", "status_code": 200, } @router.get("/merchants/{merchant_id}") async def get_merchant( merchant_id: int, current_user=Depends(get_current_superuser), db: Session = Depends(get_session), ): from src.apps.admin.crud import get_merchant_by_id m = get_merchant_by_id(db, merchant_id) if not m: raise HTTPException(status_code=404, detail="Merchant not found") return { "data": { "id": m.id, "name": m.name, "email": m.email, "uin": m.uin, "merchant_id": m.merchant_id, "is_active": m.is_active, "is_onboarded": m.is_onboarded, "is_verified": m.is_verified, "approved_by": m.approved_by, "approved_at": m.approved_at, "deleted_at": m.deleted_at, "created_at": m.created_at, "status": m.status, }, "success": True, "message": "Merchant fetched successfully", "status_code": 200, } @router.patch("/merchants/{merchant_id}") async def update_merchant( merchant_id: int, body: MerchantUpdateRequest, request: Request, current_user=Depends(get_current_superuser), db: Session = Depends(get_session), ): from src.apps.admin.crud import get_merchant_by_id from src.apps.admin.models.admin_audit_log import AdminAuditLog m = get_merchant_by_id(db, merchant_id) if not m: raise HTTPException(status_code=404, detail="Merchant not found") allowed = {"name", "email", "is_active", "is_onboarded", "is_verified"} before = {k: getattr(m, k) for k in allowed if hasattr(m, k)} changed = {} for key in allowed: val = getattr(body, key, None) if val is not None: setattr(m, key, val) changed[key] = {"before": before.get(key), "after": val} AdminAuditLog.log( db, admin_user_id=current_user.id, action="merchant.update", target_type="merchant", target_id=merchant_id, metadata={"changes": changed}, ip_address=get_ip(request), ) db.commit() db.refresh(m) # Emit merchants.updated domain event (best-effort; non-fatal on failure) if changed: try: from src.events.base import BaseEvent from src.events.dispatcher import EventDispatcher await EventDispatcher.dispatch( BaseEvent( event_type="merchants.updated", data={ "merchant_id": merchant_id, "changed_fields": {k: v["after"] for k, v in changed.items()}, }, ) ) except Exception as _evt_exc: import logging as _log _log.getLogger(__name__).warning( "merchants.updated event dispatch failed for merchant %s: %s", merchant_id, _evt_exc, ) return {"id": m.id, "name": m.name, "status": m.status} @router.delete("/merchants/{merchant_id}") async def delete_merchant( merchant_id: int, request: Request, current_user=Depends(get_current_superuser), db: Session = Depends(get_session), ): from src.apps.admin.crud import get_merchant_by_id from src.apps.admin.models.admin_audit_log import AdminAuditLog from src.apps.merchants.models.merchant_users import MerchantUsers from datetime import datetime, timezone m = get_merchant_by_id(db, merchant_id) if not m: raise HTTPException(status_code=404, detail="Merchant not found") now = datetime.now(timezone.utc) m.deleted_at = now # Soft-delete all merchant_users rows (MerchantUsers has no deleted_at, so just leave them) AdminAuditLog.log( db, admin_user_id=current_user.id, action="merchant.delete", target_type="merchant", target_id=merchant_id, ip_address=get_ip(request), ) db.commit() return {"deleted": True} @router.post("/merchants/{merchant_id}/activate") async def activate_merchant( merchant_id: int, request: Request, current_user=Depends(get_current_superuser), db: Session = Depends(get_session), ): from src.apps.admin.crud import get_merchant_by_id from src.apps.admin.models.admin_audit_log import AdminAuditLog from datetime import datetime, timezone m = get_merchant_by_id(db, merchant_id) if not m: raise HTTPException(status_code=404, detail="Merchant not found") m.is_active = True m.approved_by = current_user.id m.approved_at = datetime.now(timezone.utc) AdminAuditLog.log( db, admin_user_id=current_user.id, action="merchant.activate", target_type="merchant", target_id=merchant_id, ip_address=get_ip(request), ) db.commit() return {"activated": True, "status": m.status} @router.post("/merchants/{merchant_id}/deactivate") async def deactivate_merchant( merchant_id: int, request: Request, current_user=Depends(get_current_superuser), db: Session = Depends(get_session), ): from src.apps.admin.crud import get_merchant_by_id from src.apps.admin.models.admin_audit_log import AdminAuditLog m = get_merchant_by_id(db, merchant_id) if not m: raise HTTPException(status_code=404, detail="Merchant not found") m.is_active = False AdminAuditLog.log( db, admin_user_id=current_user.id, action="merchant.deactivate", target_type="merchant", target_id=merchant_id, ip_address=get_ip(request), ) db.commit() return {"deactivated": True} # ─── Merchant Data Drill-Down ──────────────────────────────────────────────── @router.get("/merchants/{merchant_id}/transactions") async def merchant_transactions( merchant_id: int, page: int = Query(1, ge=1), per_page: int = Query(20, ge=1, le=100), date_from: Optional[datetime] = Query(None), date_to: Optional[datetime] = Query(None), current_user=Depends(get_current_superuser), db: Session = Depends(get_session), ): from src.apps.transactions.models.transactions import Transactions from src.apps.admin.crud import get_merchant_by_id m = get_merchant_by_id(db, merchant_id) if not m: raise HTTPException(status_code=404, detail="Merchant not found") stmt = select(Transactions).where(Transactions.merchant_id == merchant_id) if date_from: stmt = stmt.where(Transactions.ocurred_at >= date_from) if date_to: stmt = stmt.where(Transactions.ocurred_at <= date_to) total = db.execute(select(func.count()).select_from(stmt.subquery())).scalar_one() stmt = stmt.order_by(Transactions.ocurred_at.desc()).offset((page - 1) * per_page).limit(per_page) items = db.execute(stmt).scalars().all() total_pages = math.ceil(total / per_page) if total > 0 else 1 return { "data": { "items": [_model_to_dict(t) for t in items], "meta": {"page": page, "per_page": per_page, "total": total, "total_pages": total_pages}, }, "success": True, "message": "Transactions fetched successfully", "status_code": 200, } @router.get("/merchants/{merchant_id}/payment-requests") async def merchant_payment_requests( merchant_id: int, page: int = Query(1, ge=1), per_page: int = Query(20, ge=1, le=100), status: Optional[str] = Query(None), current_user=Depends(get_current_superuser), db: Session = Depends(get_session), ): from src.apps.payment_requests.models.payment_request import PaymentRequest from src.apps.admin.crud import get_merchant_by_id m = get_merchant_by_id(db, merchant_id) if not m: raise HTTPException(status_code=404, detail="Merchant not found") stmt = select(PaymentRequest).where( PaymentRequest.merchant_id == merchant_id, PaymentRequest.deleted_at.is_(None), ) if status: stmt = stmt.where(PaymentRequest.status == status) total = db.execute(select(func.count()).select_from(stmt.subquery())).scalar_one() stmt = stmt.order_by(PaymentRequest.created_at.desc()).offset((page - 1) * per_page).limit(per_page) items = db.execute(stmt).scalars().all() total_pages = math.ceil(total / per_page) if total > 0 else 1 return { "data": { "items": [_model_to_dict(pr) for pr in items], "meta": {"page": page, "per_page": per_page, "total": total, "total_pages": total_pages}, }, "success": True, "message": "Payment requests fetched successfully", "status_code": 200, } @router.get("/merchants/{merchant_id}/invoices") async def merchant_invoices( merchant_id: int, page: int = Query(1, ge=1), per_page: int = Query(20, ge=1, le=100), search: Optional[str] = Query(None), status: Optional[str] = Query(None), sort_by: Optional[str] = Query(None), request: Request = None, current_user=Depends(get_current_superuser), db: Session = Depends(get_session), ): """ List invoices for a specific merchant (admin view). HWINV-401: Admin invoice list with search/status/sort + AdminAuditLog entry. """ from src.apps.invoices.models.invoice import Invoice from src.apps.admin.crud import get_merchant_by_id from src.apps.admin.models.admin_audit_log import AdminAuditLog m = get_merchant_by_id(db, merchant_id) if not m: raise HTTPException(status_code=404, detail="Merchant not found") conditions = [Invoice.merchant_id == merchant_id, Invoice.deleted_at.is_(None)] if search: escaped = _escape_like(search) conditions.append( Invoice.invoice_literal.ilike(f"%{escaped}%", escape="\\") | Invoice.invoice_id.ilike(f"%{escaped}%", escape="\\") ) if status: status_values = [int(s.strip()) for s in status.split(",") if s.strip().isdigit()] if status_values: conditions.append(Invoice.status.in_(status_values)) stmt = select(Invoice).where(*conditions) # Sort _SORTABLE = { "amount": Invoice.amount, "due_date": Invoice.due_date, "created_at": Invoice.created_at, "status": Invoice.status, } if sort_by: desc_flag = sort_by.startswith("-") field_name = sort_by.lstrip("-") col = _SORTABLE.get(field_name) if col is not None: stmt = stmt.order_by(col.desc() if desc_flag else col.asc()) else: stmt = stmt.order_by(Invoice.created_at.desc()) else: stmt = stmt.order_by(Invoice.created_at.desc()) total = db.execute(select(func.count()).select_from(stmt.subquery())).scalar_one() stmt = stmt.offset((page - 1) * per_page).limit(per_page) items = db.execute(stmt).scalars().all() total_pages = math.ceil(total / per_page) if total > 0 else 1 AdminAuditLog.log( db, admin_user_id=current_user.id, action="invoice.list", target_type="merchant", target_id=merchant_id, metadata={"search": search, "status": status, "page": page}, ip_address=get_ip(request) if request else None, ) db.commit() return { "data": { "items": [_model_to_dict(i) for i in items], "meta": {"page": page, "per_page": per_page, "total": total, "total_pages": total_pages}, }, "success": True, "message": "Invoices fetched successfully", "status_code": 200, } @router.get("/merchants/{merchant_id}/invoices/summary") async def merchant_invoice_summary( merchant_id: int, current_user=Depends(get_current_superuser), db: Session = Depends(get_session), ): """ Invoice summary stats for a merchant. HWINV-402: 5-minute Redis cache per merchant. """ from src.apps.admin.crud import get_merchant_by_id m = get_merchant_by_id(db, merchant_id) if not m: raise HTTPException(status_code=404, detail="Merchant not found") # Attempt Redis cache cache_key = f"admin_invoice_summary:{merchant_id}" cached = _redis_get(cache_key) if cached: import json return {"data": json.loads(cached), "success": True, "status_code": 200, "message": "Invoice summary fetched (cached)"} from src.apps.invoices import crud as invoice_crud stats = invoice_crud.get_invoice_summary_stats(db, merchant_id) _redis_set(cache_key, stats, ttl=300) # 5 min return { "data": stats, "success": True, "message": "Invoice summary fetched successfully", "status_code": 200, } @router.get("/merchants/{merchant_id}/invoices/{invoice_literal}") async def get_merchant_invoice( merchant_id: int, invoice_literal: str, request: Request, current_user=Depends(get_current_superuser), db: Session = Depends(get_session), ): """ Fetch a single invoice for a merchant (admin view). HWINV-403: Writes AdminAuditLog entry per access. """ from src.apps.admin.crud import get_merchant_by_id from src.apps.admin.models.admin_audit_log import AdminAuditLog from src.apps.invoices import crud as invoice_crud m = get_merchant_by_id(db, merchant_id) if not m: raise HTTPException(status_code=404, detail="Merchant not found") invoice = invoice_crud.get_invoice_by_literal(db, invoice_literal, merchant_id) if not invoice: raise HTTPException(status_code=404, detail="Invoice not found") AdminAuditLog.log( db, admin_user_id=current_user.id, action="invoice.view", target_type="invoice", target_id=invoice.id, metadata={"invoice_literal": invoice_literal, "merchant_id": merchant_id}, ip_address=get_ip(request), ) db.commit() from src.apps.invoices.schemas.invoice_common import InvoiceSchema return { "data": InvoiceSchema.model_validate(invoice).model_dump(), "success": True, "message": "Invoice fetched successfully", "status_code": 200, } @router.get("/merchants/{merchant_id}/invoices/{invoice_literal}/transactions") async def admin_merchant_invoice_transactions( merchant_id: int, invoice_literal: str, current_user=Depends(get_current_superuser), db: Session = Depends(get_session), ): """Admin: list transactions for a merchant's invoice. HWINV-403.""" from src.apps.admin.crud import get_merchant_by_id from src.apps.invoices import crud as invoice_crud m = get_merchant_by_id(db, merchant_id) if not m: raise HTTPException(status_code=404, detail="Merchant not found") invoice = invoice_crud.get_invoice_by_literal(db, invoice_literal, merchant_id) if not invoice: raise HTTPException(status_code=404, detail="Invoice not found") transactions = invoice_crud.get_invoice_transactions(db, invoice.id) return { "data": [_model_to_dict(t) for t in transactions], "success": True, "message": "Invoice transactions fetched successfully", "status_code": 200, } @router.get("/merchants/{merchant_id}/invoices/{invoice_literal}/authorizations") async def admin_merchant_invoice_authorizations( merchant_id: int, invoice_literal: str, current_user=Depends(get_current_superuser), db: Session = Depends(get_session), ): """Admin: list authorization records for a merchant's invoice. HWINV-403.""" from src.apps.admin.crud import get_merchant_by_id from src.apps.invoices import crud as invoice_crud m = get_merchant_by_id(db, merchant_id) if not m: raise HTTPException(status_code=404, detail="Merchant not found") invoice = invoice_crud.get_invoice_by_literal(db, invoice_literal, merchant_id) if not invoice: raise HTTPException(status_code=404, detail="Invoice not found") authorizations = invoice_crud.get_invoice_authorizations(db, invoice.id, merchant_id=merchant_id) return { "data": [_model_to_dict(a) for a in authorizations], "success": True, "message": "Invoice authorizations fetched successfully", "status_code": 200, } @router.get("/merchants/{merchant_id}/invoices/{invoice_literal}/activity") async def admin_merchant_invoice_activity( merchant_id: int, invoice_literal: str, page: int = Query(1, ge=1), per_page: int = Query(20, ge=1, le=100), current_user=Depends(get_current_superuser), db: Session = Depends(get_session), ): """Admin: activity log for a merchant's invoice. HWINV-403.""" from src.apps.admin.crud import get_merchant_by_id from src.apps.invoices import crud as invoice_crud m = get_merchant_by_id(db, merchant_id) if not m: raise HTTPException(status_code=404, detail="Merchant not found") invoice = invoice_crud.get_invoice_by_literal(db, invoice_literal, merchant_id) if not invoice: raise HTTPException(status_code=404, detail="Invoice not found") activities, total = invoice_crud.list_invoice_activities(db, invoice.id, page, per_page) total_pages = math.ceil(total / per_page) if total > 0 else 1 items = [ { "id": a.id, "activity_type": a.activity_type, "description": a.description, "actor_type": a.actor_type, "actor_id": a.actor_id, "metadata": a.metadata_, "created_at": a.created_at, } for a in activities ] return { "data": { "items": items, "meta": {"page": page, "per_page": per_page, "total": total, "total_pages": total_pages}, }, "success": True, "message": "Invoice activity fetched successfully", "status_code": 200, } @router.get("/merchants/{merchant_id}/invoices/{invoice_literal}/analytics") async def admin_merchant_invoice_analytics( merchant_id: int, invoice_literal: str, current_user=Depends(get_current_superuser), db: Session = Depends(get_session), ): """Admin: revenue/cost/profit analytics for a merchant's invoice. HWINV-403.""" from src.apps.admin.crud import get_merchant_by_id from src.apps.invoices import crud as invoice_crud m = get_merchant_by_id(db, merchant_id) if not m: raise HTTPException(status_code=404, detail="Merchant not found") invoice = invoice_crud.get_invoice_by_literal(db, invoice_literal, merchant_id) if not invoice: raise HTTPException(status_code=404, detail="Invoice not found") total_cost = sum( (li.cost or 0) * (li.quantity or 1) for li in (invoice.invoice_line_items or []) ) revenue = invoice.amount or 0 profit = revenue - total_cost margin = (profit / revenue * 100) if revenue > 0 else 0.0 transactions = invoice_crud.get_invoice_transactions(db, invoice.id) txn_total = sum(getattr(t, "amount", 0) or 0 for t in transactions) return { "data": { "revenue": revenue, "cost": total_cost, "profit": profit, "margin_pct": round(margin, 2), "transaction_count": len(transactions), "transaction_total": txn_total, }, "success": True, "message": "Invoice analytics fetched successfully", "status_code": 200, } @router.get("/merchants/{merchant_id}/subscriptions") async def merchant_subscriptions( merchant_id: int, page: int = Query(1, ge=1), per_page: int = Query(20, ge=1, le=100), status: Optional[str] = Query(None), search: Optional[str] = Query(None), current_user=Depends(get_current_superuser), db: Session = Depends(get_session), ): from src.apps.admin.crud import get_merchant_by_id from src.apps.subscriptions import crud as sub_crud from src.apps.subscriptions.services import build_list_item m = get_merchant_by_id(db, merchant_id) if not m: raise HTTPException(status_code=404, detail="Merchant not found") filters = {} if status: filters["status"] = status if search: filters["search"] = search items, total = sub_crud.list_subscriptions( db, merchant_id=merchant_id, filters=filters, page=page, per_page=per_page ) total_pages = math.ceil(total / per_page) if total > 0 else 1 return { "data": { "items": [build_list_item(s) for s in items], "meta": {"page": page, "per_page": per_page, "total": total, "total_pages": total_pages}, }, "success": True, "message": "Subscriptions fetched successfully", "status_code": 200, } @router.get("/merchants/{merchant_id}/customers") async def merchant_customers( merchant_id: int, page: int = Query(1, ge=1), per_page: int = Query(20, ge=1, le=100), current_user=Depends(get_current_superuser), db: Session = Depends(get_session), ): from src.apps.admin.crud import get_merchant_by_id m = get_merchant_by_id(db, merchant_id) if not m: raise HTTPException(status_code=404, detail="Merchant not found") try: from src.apps.customers.models.customer import Customer stmt = select(Customer).where( Customer.merchant_id == merchant_id, Customer.deleted_at.is_(None), ) total = db.execute(select(func.count()).select_from(stmt.subquery())).scalar_one() stmt = stmt.order_by(Customer.created_at.desc()).offset((page - 1) * per_page).limit(per_page) items = db.execute(stmt).scalars().all() total_pages = math.ceil(total / per_page) if total > 0 else 1 return { "data": { "items": [_model_to_dict(c) for c in items], "meta": {"page": page, "per_page": per_page, "total": total, "total_pages": total_pages}, }, "success": True, "message": "Customers fetched successfully", "status_code": 200, } except Exception: return { "data": {"items": [], "meta": {"page": page, "per_page": per_page, "total": 0, "total_pages": 1}}, "success": True, "message": "Customers fetched successfully", "status_code": 200, } # ─── Impersonation ─────────────────────────────────────────────────────────── @router.post("/merchants/{merchant_id}/impersonate") async def impersonate_merchant( merchant_id: int, request: Request, current_user=Depends(get_current_superuser), db: Session = Depends(get_session), ): from src.apps.admin.crud import get_merchant_by_id from src.apps.admin.services import create_impersonation_token m = get_merchant_by_id(db, merchant_id) if not m: raise HTTPException(status_code=404, detail="Merchant not found") token = create_impersonation_token( db, merchant_id=merchant_id, admin_user_id=current_user.id, ip_address=get_ip(request), merchant_name=getattr(m, "name", None), ) return { "access_token": token, "token_type": "bearer", "expires_in": 14400, "impersonated_merchant_id": merchant_id, } @router.delete("/merchants/{merchant_id}/impersonate") async def end_impersonation( merchant_id: int, request: Request, current_user=Depends(get_current_superuser), db: Session = Depends(get_session), ): from src.apps.admin.services import revoke_impersonation_session revoked = revoke_impersonation_session( db, merchant_id=merchant_id, admin_user_id=current_user.id, ip_address=get_ip(request), ) return {"revoked": revoked} # ─── Admin User Management ─────────────────────────────────────────────────── @router.get("/users") async def list_admin_users( page: int = Query(1, ge=1), per_page: int = Query(20, ge=1, le=100), is_active: Optional[bool] = Query(None), current_user=Depends(get_current_superuser), db: Session = Depends(get_session), ): from src.apps.users.models.user import User stmt = select(User).where(User.is_superuser == True, User.deleted_at.is_(None)) if is_active is not None: stmt = stmt.where(User.is_active == is_active) total = db.execute(select(func.count()).select_from(stmt.subquery())).scalar_one() stmt = stmt.offset((page - 1) * per_page).limit(per_page) items = db.execute(stmt).scalars().all() total_pages = math.ceil(total / per_page) if total > 0 else 1 return { "data": { "items": [ { "id": u.id, "email": u.email, "first_name": u.first_name, "last_name": u.last_name, "is_superuser": u.is_superuser, "is_user_active": u.is_active, "created_at": u.created_at, } for u in items ], "meta": {"page": page, "per_page": per_page, "total": total, "total_pages": total_pages}, }, "success": True, "message": "Admin users fetched successfully", "status_code": 200, } @router.post("/users") async def create_admin_user( body: AdminUserCreate, request: Request, current_user=Depends(get_current_superuser), db: Session = Depends(get_session), ): from src.apps.users.models.user import User from src.apps.admin.models.admin_audit_log import AdminAuditLog from src.core.utils.password import encrypt_password import uuid as uuid_module email = body.email password = body.password if not email or not password: raise HTTPException(status_code=400, detail="email and password required") # Enforce minimum password strength for admin accounts if len(password) < 12: raise HTTPException(status_code=400, detail="Password must be at least 12 characters") if not any(c.isupper() for c in password): raise HTTPException(status_code=400, detail="Password must contain an uppercase letter") if not any(c.islower() for c in password): raise HTTPException(status_code=400, detail="Password must contain a lowercase letter") if not any(c.isdigit() for c in password): raise HTTPException(status_code=400, detail="Password must contain a digit") existing = db.execute(select(User).where(User.email == email)).scalar_one_or_none() if existing: raise HTTPException(status_code=409, detail="User with this email already exists") hashed = encrypt_password(password) if hashed: hashed_str = hashed.decode() if isinstance(hashed, bytes) else hashed else: raise HTTPException(status_code=500, detail="Password hashing failed") # Generate a unique username from email username = email.split("@")[0] + "_" + str(uuid_module.uuid4())[:8] user = User( email=email, username=username, hashed_password=hashed_str, is_superuser=True, is_active=True, is_verified=True, first_name=body.first_name or "", last_name=body.last_name or "", ) db.add(user) db.flush() AdminAuditLog.log( db, admin_user_id=current_user.id, action="admin_user.create", target_type="user", target_id=user.id, ip_address=get_ip(request), ) db.commit() return {"id": user.id, "email": user.email} @router.patch("/users/{user_id}") async def update_admin_user( user_id: int, body: AdminUserUpdate, request: Request, current_user=Depends(get_current_superuser), db: Session = Depends(get_session), ): from src.apps.users.models.user import User from src.apps.admin.models.admin_audit_log import AdminAuditLog user = db.execute( select(User).where(User.id == user_id, User.is_superuser == True) ).scalar_one_or_none() if not user: raise HTTPException(status_code=404, detail="Admin user not found") if body.email is not None: user.email = body.email if body.first_name is not None: user.first_name = body.first_name if body.last_name is not None: user.last_name = body.last_name if body.is_user_active is not None: user.is_active = body.is_user_active AdminAuditLog.log( db, admin_user_id=current_user.id, action="admin_user.update", target_type="user", target_id=user_id, ip_address=get_ip(request), ) db.commit() return {"id": user.id, "email": user.email} @router.delete("/users/{user_id}") async def delete_admin_user( user_id: int, request: Request, current_user=Depends(get_current_superuser), db: Session = Depends(get_session), ): from src.apps.users.models.user import User from src.apps.admin.models.admin_audit_log import AdminAuditLog from datetime import datetime, timezone if user_id == current_user.id: raise HTTPException(status_code=400, detail="Cannot delete your own account") user = db.execute( select(User).where(User.id == user_id, User.is_superuser == True, User.deleted_at.is_(None)) ).scalar_one_or_none() if not user: raise HTTPException(status_code=404, detail="Admin user not found") user.deleted_at = datetime.now(timezone.utc) AdminAuditLog.log( db, admin_user_id=current_user.id, action="admin_user.delete", target_type="user", target_id=user_id, ip_address=get_ip(request), ) db.commit() return {"deleted": True} # ─── Password Policy CRUD ──────────────────────────────────────────────────── @router.get("/password-policies") async def list_password_policies( merchant_id: Optional[int] = Query(None), current_user=Depends(get_current_superuser), db: Session = Depends(get_session), ): """List all password policies. Optionally filter by merchant_id.""" from src.apps.user_profile.models.password_policy import PasswordPolicy stmt = select(PasswordPolicy).where(PasswordPolicy.deleted_at.is_(None)) if merchant_id is not None: stmt = stmt.where(PasswordPolicy.merchant_id == merchant_id) stmt = stmt.order_by(PasswordPolicy.created_at.desc()) policies = db.execute(stmt).scalars().all() return { "data": [_model_to_dict(p) for p in policies], "success": True, "message": "Password policies fetched", "status_code": 200, } @router.post("/password-policies", status_code=201) async def create_password_policy( request: Request, current_user=Depends(get_current_superuser), db: Session = Depends(get_session), ): """Create a new password policy.""" import json as _json from src.apps.user_profile.models.password_policy import PasswordPolicy from src.apps.user_profile.schemas.profile_schemas import PasswordPolicyCreateRequest body_bytes = await request.body() body = PasswordPolicyCreateRequest(**_json.loads(body_bytes)) policy = PasswordPolicy(**body.model_dump()) db.add(policy) db.commit() db.refresh(policy) # Invalidate Redis cache for affected merchant if body.merchant_id is not None: try: client = _get_redis() if client: client.delete(f"password_policy:{body.merchant_id}") except Exception: pass return {"data": _model_to_dict(policy), "success": True, "status_code": 201} @router.put("/password-policies/{policy_id}") async def update_password_policy( policy_id: int, request: Request, current_user=Depends(get_current_superuser), db: Session = Depends(get_session), ): """Update an existing password policy.""" import json as _json from src.apps.user_profile.models.password_policy import PasswordPolicy from src.apps.user_profile.schemas.profile_schemas import PasswordPolicyUpdateRequest policy = db.execute( select(PasswordPolicy).where(PasswordPolicy.id == policy_id, PasswordPolicy.deleted_at.is_(None)) ).scalar_one_or_none() if not policy: from fastapi import HTTPException raise HTTPException(status_code=404, detail="Password policy not found") body_bytes = await request.body() body = PasswordPolicyUpdateRequest(**_json.loads(body_bytes)) for field, value in body.model_dump(exclude_none=True).items(): setattr(policy, field, value) db.commit() db.refresh(policy) # Invalidate Redis cache try: client = _get_redis() if client: client.delete(f"password_policy:{policy.merchant_id or 0}") client.delete(f"password_policy:0") except Exception: pass return {"data": _model_to_dict(policy), "success": True, "status_code": 200} @router.delete("/password-policies/{policy_id}") async def delete_password_policy( policy_id: int, current_user=Depends(get_current_superuser), db: Session = Depends(get_session), ): """Soft delete a password policy.""" from src.apps.user_profile.models.password_policy import PasswordPolicy policy = db.execute( select(PasswordPolicy).where(PasswordPolicy.id == policy_id, PasswordPolicy.deleted_at.is_(None)) ).scalar_one_or_none() if not policy: from fastapi import HTTPException raise HTTPException(status_code=404, detail="Password policy not found") policy.deleted_at = datetime.now(timezone.utc) db.commit() try: client = _get_redis() if client: client.delete(f"password_policy:{policy.merchant_id or 0}") except Exception: pass return {"deleted": True, "success": True, "status_code": 200} @router.get("/password-policies/merchant/{merchant_id}") async def get_effective_policy_for_merchant( merchant_id: int, current_user=Depends(get_current_superuser), db: Session = Depends(get_session), ): """Get the effective (resolved) password policy for a merchant.""" from src.apps.user_profile.services import get_effective_password_policy from src.apps.user_profile.schemas.profile_schemas import PasswordPolicyResponse policy = get_effective_password_policy(db, merchant_id) return { "data": PasswordPolicyResponse.model_validate(policy).model_dump(), "success": True, "status_code": 200, } # ─── TSYS Config Endpoints ────────────────────────────────────────────────── def _log_admin_action( db, user, action: str, entity_type: str, entity_id, merchant_id, extra: dict = None, ) -> None: """Write an AdminAuditLog entry for a TSYS config action.""" from src.apps.admin.models.admin_audit_log import AdminAuditLog metadata = {"entity_type": entity_type, "merchant_id": merchant_id} if extra: metadata.update(extra) AdminAuditLog.log( db, admin_user_id=user.id, action=action, target_type=entity_type, target_id=entity_id, metadata=metadata, ) class TsysConfigUpdateRequest(BaseModel): merchant_id: str device_id: str user_id: str password: str class TsysStatusUpdateRequest(BaseModel): onboarding_status: str @router.get("/merchants/{merchant_id}/tsys-config", tags=["ADMIN"]) async def get_merchant_tsys_config( merchant_id: int, reveal: bool = False, current_user=Depends(get_current_superuser), db: Session = Depends(get_db), ): """Retrieve TSYS configuration for a merchant. Pass ?reveal=true to expose the transaction key.""" from src.apps.payment_providers.crud import get_provider_by_slug, get_merchant_provider_config, get_credentials provider = get_provider_by_slug("tsys", db) if not provider: raise HTTPException(status_code=404, detail="TSYS provider not found in system") mc = get_merchant_provider_config(merchant_id, provider.id, db) if not mc: raise HTTPException(status_code=404, detail="No TSYS configuration found for this merchant") creds = get_credentials(mc.id, db) config_data = mc.config_data or {} _log_admin_action(db, current_user, "tsys_config.read", "merchant_provider_config", mc.id, merchant_id) response = { "merchant_id": creds.get("merchant_id", ""), "device_id": creds.get("device_id", ""), "onboarding_status": mc.onboarding_status, "key_generated_at": config_data.get("key_generated_at"), "transaction_key": None, } if reveal: response["transaction_key"] = creds.get("transaction_key", "") _log_admin_action(db, current_user, "credential_reveal", "merchant_provider_config", mc.id, merchant_id) db.commit() return {"data": response, "success": True, "status_code": 200, "message": "TSYS config fetched"} @router.put("/merchants/{merchant_id}/tsys-config", tags=["ADMIN"]) async def update_merchant_tsys_config( merchant_id: int, payload: TsysConfigUpdateRequest, current_user=Depends(get_current_superuser), db: Session = Depends(get_db), ): """Re-generate the TSYS transaction key for a merchant using new credentials.""" from src.apps.payment_providers.crud import ( get_provider_by_slug, upsert_merchant_provider_config, upsert_credential, ) from src.core.providers.implementations.tsys import TSYSProvider from datetime import datetime, timezone provider = get_provider_by_slug("tsys", db) if not provider: raise HTTPException(status_code=404, detail="TSYS provider not found") tsys = TSYSProvider() try: transaction_key = tsys._generate_transaction_key( merchant_id=payload.merchant_id, user_id=payload.user_id, password=payload.password, ) except (ValueError, RuntimeError) as e: raise HTTPException(status_code=400, detail=str(e)) mc = upsert_merchant_provider_config( merchant_id=merchant_id, provider_id=provider.id, onboarding_status="active", config_data={"key_generated_at": datetime.now(timezone.utc).isoformat()}, db=db, ) upsert_credential(mc.id, "merchant_id", payload.merchant_id, db) upsert_credential(mc.id, "device_id", payload.device_id, db) upsert_credential(mc.id, "transaction_key", transaction_key, db) _log_admin_action(db, current_user, "credential_update", "merchant_provider_config", mc.id, merchant_id) db.commit() return { "data": { "merchant_id": payload.merchant_id, "device_id": payload.device_id, "onboarding_status": "active", }, "success": True, "status_code": 200, "message": "TSYS config updated", } @router.patch("/merchants/{merchant_id}/tsys-config/status", tags=["ADMIN"]) async def update_merchant_tsys_status( merchant_id: int, payload: TsysStatusUpdateRequest, current_user=Depends(get_current_superuser), db: Session = Depends(get_db), ): """Override the onboarding status of a merchant's TSYS provider config.""" from src.apps.payment_providers.crud import get_provider_by_slug, get_merchant_provider_config provider = get_provider_by_slug("tsys", db) if not provider: raise HTTPException(status_code=404, detail="TSYS provider not found") mc = get_merchant_provider_config(merchant_id, provider.id, db) if not mc: raise HTTPException(status_code=404, detail="No TSYS configuration found for this merchant") allowed_statuses = {"not_started", "in_progress", "pending_review", "active", "suspended", "rejected"} new_status = payload.onboarding_status if new_status not in allowed_statuses: raise HTTPException( status_code=422, detail=f"Invalid status. Allowed: {', '.join(sorted(allowed_statuses))}", ) mc.onboarding_status = new_status _log_admin_action( db, current_user, "status_override", "merchant_provider_config", mc.id, merchant_id, {"new_status": new_status}, ) db.commit() return { "data": {"onboarding_status": new_status}, "success": True, "status_code": 200, "message": "TSYS status updated", } # ─── Audit Log ─────────────────────────────────────────────────────────────── @router.get("/audit-log") async def get_audit_log( page: int = Query(1, ge=1), per_page: int = Query(20, ge=1, le=100), admin_user_id: Optional[int] = Query(None), action: Optional[str] = Query(None), date_from: Optional[datetime] = Query(None), date_to: Optional[datetime] = Query(None), current_user=Depends(get_current_superuser), db: Session = Depends(get_session), ): from src.apps.admin.crud import get_audit_logs items, total = get_audit_logs( db, page=page, per_page=per_page, admin_user_id=admin_user_id, action=action, date_from=date_from, date_to=date_to, ) total_pages = math.ceil(total / per_page) if total > 0 else 1 return { "data": { "items": [ { "id": e.id, "admin_user_id": e.admin_user_id, "action": e.action, "target_type": e.target_type, "target_id": e.target_id, "metadata": e.metadata_, "ip_address": e.ip_address, "created_at": e.created_at, } for e in items ], "meta": {"page": page, "per_page": per_page, "total": total, "total_pages": total_pages}, }, "success": True, "message": "Audit log fetched successfully", "status_code": 200, } # ─── Admin Subscription Endpoints ──────────────────────────────────────────── @router.get("/subscriptions") async def admin_list_subscriptions( page: int = Query(1, ge=1), per_page: int = Query(20, ge=1, le=100), merchant_id: Optional[int] = Query(None), status: Optional[str] = Query(None, description="Comma-separated status ints"), search: Optional[str] = Query(None), current_user=Depends(get_current_superuser), db: Session = Depends(get_session), ): """List all subscriptions across all merchants (admin view).""" from src.apps.subscriptions.models.subscription import Subscription from src.apps.subscriptions import services as sub_svc from src.apps.subscriptions import crud as sub_crud from sqlalchemy import select, func base = [Subscription.deleted_at == None] if merchant_id: base.append(Subscription.merchant_id == merchant_id) if status: try: statuses = [int(s.strip()) for s in status.split(",") if s.strip()] if statuses: base.append(Subscription.status.in_(statuses)) except ValueError: pass if search: base.append(Subscription.subscription_literal.ilike(f"%{search}%")) count_stmt = select(func.count(Subscription.id)).where(*base) total = db.execute(count_stmt).scalar_one() items_stmt = ( select(Subscription) .where(*base) .order_by(Subscription.created_at.desc()) .offset((page - 1) * per_page) .limit(per_page) ) from sqlalchemy.orm import joinedload items_stmt = items_stmt.options( joinedload(Subscription.customer), joinedload(Subscription.payment_request), joinedload(Subscription.merchant), ) items = list(db.execute(items_stmt).unique().scalars().all()) total_pages = max(1, math.ceil(total / per_page)) return { "data": { "items": [sub_svc.build_list_item(s) for s in items], "meta": {"page": page, "per_page": per_page, "total": total, "total_pages": total_pages}, }, "success": True, } @router.get("/subscriptions/{sub_id}") async def admin_get_subscription( sub_id: str, current_user=Depends(get_current_superuser), db: Session = Depends(get_session), ): """Get subscription detail by ID or literal (no merchant scope check).""" from src.apps.subscriptions import crud as sub_crud from src.apps.subscriptions import services as sub_svc subscription = sub_crud.get_subscription_unscoped(db, sub_id) if not subscription: raise HTTPException(status_code=404, detail=f"Subscription '{sub_id}' not found") return {"data": sub_svc.build_detail(subscription, db), "success": True} @router.patch("/subscriptions/{sub_id}") async def admin_update_subscription( sub_id: str, request: Request, current_user=Depends(get_current_superuser), db: Session = Depends(get_session), ): """Force cancel a subscription with a reason (admin only).""" from src.apps.subscriptions import crud as sub_crud from src.apps.subscriptions import services as sub_svc from src.apps.subscriptions.schemas.subscription_schemas import AdminSubscriptionUpdateRequest from src.apps.admin.models.admin_audit_log import AdminAuditLog body_data = await request.json() payload = AdminSubscriptionUpdateRequest(**body_data) subscription = sub_crud.get_subscription_unscoped(db, sub_id) if not subscription: raise HTTPException(status_code=404, detail=f"Subscription '{sub_id}' not found") if payload.action == "cancel": subscription = sub_svc.cancel_subscription(db, subscription, actor_id=current_user.id) AdminAuditLog.log( db, admin_user_id=current_user.id, action="subscription.force_cancel", target_type="subscription", target_id=subscription.id, ip_address=get_ip(request), metadata={"reason": payload.reason}, ) db.commit() return { "data": sub_svc.build_list_item(subscription), "success": True, "message": "Subscription cancelled", } raise HTTPException(status_code=400, detail=f"Unknown action: {payload.action}") @router.get("/subscriptions/{sub_id}/invoices") async def admin_get_subscription_invoices( sub_id: str, max_items: int = Query(12, ge=1, le=50), current_user=Depends(get_current_superuser), db: Session = Depends(get_session), ): """Return invoice timeline for a subscription (admin view).""" from src.apps.subscriptions import crud as sub_crud from src.apps.subscriptions import services as sub_svc subscription = sub_crud.get_subscription_unscoped(db, sub_id) if not subscription: raise HTTPException(status_code=404, detail=f"Subscription '{sub_id}' not found") result = sub_svc.get_subscription_invoices_with_projections(db, subscription, max_items) return {"data": result.model_dump(), "success": True} @router.get("/subscriptions/{sub_id}/activities") async def admin_get_subscription_activities( sub_id: str, page: int = Query(1, ge=1), per_page: int = Query(20, ge=1, le=100), current_user=Depends(get_current_superuser), db: Session = Depends(get_session), ): """Paginated activity trail for a subscription (admin view).""" from src.apps.subscriptions import crud as sub_crud subscription = sub_crud.get_subscription_unscoped(db, sub_id) if not subscription: raise HTTPException(status_code=404, detail=f"Subscription '{sub_id}' not found") activities, total = sub_crud.get_activities(db, subscription.id, page, per_page) total_pages = max(1, math.ceil(total / per_page)) return { "data": { "items": [ { "id": a.id, "activity_type": a.activity_type, "description": a.description, "actor_type": a.actor_type, "created_at": a.created_at, "metadata": a.metadata_, } for a in activities ], "meta": {"page": page, "per_page": per_page, "total": total, "total_pages": total_pages}, }, "success": True, } # ─── Tax Rate Cache Management ──────────────────────────────────────────────── @router.delete("/tax-cache", tags=["ADMIN"]) async def clear_tax_cache( db: Session = Depends(get_db), _admin=Depends(get_current_superuser), ): """Evict the entire tax rate cache (all zip/country entries).""" from src.apps.tax.models import TaxRateCache deleted = db.query(TaxRateCache).delete() db.commit() return {"deleted": deleted} @router.delete("/tax-cache/{zip}", tags=["ADMIN"]) async def evict_tax_cache_entry( zip: str, country: str = Query(default="US"), db: Session = Depends(get_db), _admin=Depends(get_current_superuser), ): """Evict a single zip/country entry from the tax rate cache.""" from src.apps.tax.models import TaxRateCache deleted = ( db.query(TaxRateCache) .filter(TaxRateCache.zip == zip, TaxRateCache.country == country) .delete() ) db.commit() return {"deleted": deleted} # ─── Admin: Developer API Key Management ───────────────────────────────────── @router.get("/merchants/{merchant_id}/api-keys", tags=["Admin - Developer"]) async def admin_list_merchant_api_keys( merchant_id: int, current_user=Depends(get_current_superuser), db: Session = Depends(get_session), ): from src.apps.developer import crud as dev_crud from src.apps.developer.schemas.api_key_schemas import ApiKeyResponse keys = dev_crud.get_api_keys_for_merchant_admin(db, merchant_id) return [ApiKeyResponse.model_validate(k) for k in keys] @router.post("/merchants/{merchant_id}/api-keys", status_code=201, tags=["Admin - Developer"]) async def admin_create_merchant_api_key( merchant_id: int, body: AdminCreateApiKeyRequest, current_user=Depends(get_current_superuser), db: Session = Depends(get_session), ): """Create an API key for a merchant (admin). Send display_name, environment, scopes as JSON body.""" from src.apps.developer import crud as dev_crud from src.apps.developer.helpers.api_key_auth import generate_api_key, generate_key_id from src.apps.developer.schemas.api_key_schemas import ApiKeyCreatedResponse raw_key, key_hash, key_salt, key_prefix = generate_api_key(body.environment) key_id = generate_key_id() key = dev_crud.create_api_key( db, merchant_id=merchant_id, display_name=body.display_name, key_id=key_id, key_hash=key_hash, key_salt=key_salt, key_prefix=key_prefix, environment=body.environment, scopes=body.scopes, created_by_user_id=current_user.id, ) return ApiKeyCreatedResponse( id=key.id, key_id=key.key_id, api_key=raw_key, key_prefix=key.key_prefix, environment=key.environment, scopes=key.scopes, display_name=key.display_name, created_at=key.created_at, ) @router.get("/merchants/{merchant_id}/webhooks", tags=["Admin - Developer"]) async def admin_list_merchant_webhooks( merchant_id: int, current_user=Depends(get_current_superuser), db: Session = Depends(get_session), ): """List webhook endpoints for a merchant (admin, read-only).""" from src.apps.developer import crud as dev_crud from src.apps.developer.schemas.webhook_schemas import WebhookResponse webhooks = dev_crud.get_webhook_endpoints_for_merchant(db, merchant_id) return [WebhookResponse.model_validate(w) for w in webhooks] @router.delete("/merchants/{merchant_id}/api-keys/{key_id}", status_code=204, tags=["Admin - Developer"]) async def admin_revoke_merchant_api_key( merchant_id: int, key_id: int, current_user=Depends(get_current_superuser), db: Session = Depends(get_session), ): from src.apps.developer import crud as dev_crud key = dev_crud.get_api_key_by_id(db, key_id, merchant_id) if not key: raise HTTPException(status_code=404, detail="API key not found") dev_crud.revoke_api_key(db, key)