"""Platform admin console routes (users, admin grants). **Platform admin HTTP surface** (all require ``Authorization: Bearer`` for a user in ``admins``, except auth entrypoints under ``/api/admin/auth/*`` where noted): Auth (admin SPA — tokens only for platform admins after verify): - ``POST /api/admin/auth/magic/request`` — magic link only if email is already a platform admin - ``POST /api/admin/auth/magic/verify`` — verify link; 403 + revoke if not admin - ``GET /api/admin/auth/magic/verify`` — same (query ``token``) - ``POST /api/admin/auth/mfa/totp/verify`` — complete TOTP after admin magic flow - ``GET /api/admin/auth/bootstrap`` — bootstrap JSON; 403 if not admin - ``POST /api/admin/auth/refresh`` — refresh; 403 + revoke if not admin Console: - ``GET /api/admin/users`` — paginated users (``limit``, ``offset``, ``search``) - ``POST /api/admin/admins`` — add platform admin by ``user_id`` or create user from ``email`` Plans & billing: - ``POST /api/admin/plans`` — create plan (syncs Stripe) - ``GET /api/admin/plans`` — list plans - ``GET /api/admin/plans/{plan_id}`` — get plan - ``PATCH /api/admin/plans/{plan_id}`` — update plan Hosting (W4P dispatcher KV): - ``POST /api/admin/hosting/dispatch-routes`` — upsert dispatcher KV route (script must already exist in NS) - ``DELETE /api/admin/hosting/dispatch-routes/{subdomain}`` — remove route Templates (demos + starter branches; CF Pages + W4P via worker): - ``POST /api/admin/templates/create`` — create DB row + GitHub starter repo + enqueue demo deploys - ``/api/admin/templates`` — register, list, update, deploy, delete - ``/api/templates`` — list/get for authenticated users (plan-filtered) """ from __future__ import annotations from datetime import datetime, timezone from typing import Optional import structlog from fastapi import APIRouter, Depends, HTTPException, Query from sqlalchemy.ext.asyncio import AsyncSession from src.core.database import get_db from src.core.exceptions import NotFoundError from src.domain.admin.repository import AdminRepository from src.domain.admin.router import require_admin from src.domain.admin.schemas import ( AdminEmailComposerOptions, AdminEmailSendRequest, AdminEmailSendResponse, AdminUserListItem, AdminUserListResponse, CreatePlatformAdminBody, CreatePlatformAdminResponse, ) from src.domain.auth.models import User from src.domain.auth.repository import UserRepository logger = structlog.get_logger() router = APIRouter(prefix="/api/admin", tags=["admin", "admin-console"]) @router.get("/users", response_model=AdminUserListResponse) async def list_users( limit: int = Query(50, ge=1, le=200), offset: int = Query(0, ge=0), search: Optional[str] = Query(None, description="Filter by email substring (case-insensitive)"), _admin: User = Depends(require_admin), db: AsyncSession = Depends(get_db), ): """Paginated user list for operators.""" user_repo = UserRepository(db) admin_repo = AdminRepository(db) total = await user_repo.count_for_admin_list(search) rows = await user_repo.list_for_admin(limit=limit, offset=offset, search=search) user_ids = [u.id for u in rows] admin_ids = await admin_repo.filter_platform_admin_ids(user_ids) billing_summaries = await admin_repo.list_owned_workspace_billing_summaries(user_ids) items = [ AdminUserListItem( id=u.id, email=u.email, display_name=u.display_name, first_name=u.first_name, last_name=u.last_name, created_at=u.created_at, email_verified=u.email_verified_at is not None, is_platform_admin=(u.id in admin_ids), is_paying_customer=billing_summaries.get(u.id).is_paying_customer if u.id in billing_summaries else False, subscribed_plans=billing_summaries.get(u.id).subscribed_plans if u.id in billing_summaries else [], ) for u in rows ] return AdminUserListResponse(items=items, total=total, limit=limit, offset=offset) def _admin_email_from_addresses() -> list[str]: from src.config import get_settings settings = get_settings() candidates = [settings.from_email, *settings.admin_email_from_addresses.split(",")] return list(dict.fromkeys(address.strip().lower() for address in candidates if address.strip())) @router.get("/email-composer/options", response_model=AdminEmailComposerOptions) async def get_admin_email_composer_options(_admin: User = Depends(require_admin)): """Return the verified sender addresses available to the platform composer.""" return AdminEmailComposerOptions(from_addresses=_admin_email_from_addresses()) @router.post("/emails", response_model=AdminEmailSendResponse) async def queue_admin_email( body: AdminEmailSendRequest, admin: User = Depends(require_admin), db: AsyncSession = Depends(get_db), ): """Queue an admin-authored HTML email for immediate or deferred delivery.""" from src.config import get_settings from src.workers.enqueue import enqueue_admin_email from src.integrations.email_provider import email_transport_configured settings = get_settings() if not email_transport_configured(settings): raise HTTPException(status_code=503, detail="Email transport is not configured") from_email = str(body.from_email).lower() if from_email not in _admin_email_from_addresses(): raise HTTPException(status_code=422, detail="Sender address is not allowed") now = datetime.now(timezone.utc) scheduled_at = body.scheduled_at if scheduled_at is not None: if scheduled_at.tzinfo is None: raise HTTPException(status_code=422, detail="Scheduled time must include a timezone") scheduled_at = scheduled_at.astimezone(timezone.utc) if scheduled_at <= now: raise HTTPException(status_code=422, detail="Scheduled time must be in the future") users = await UserRepository(db).list_by_ids(body.user_ids) if len(users) != len(body.user_ids): raise NotFoundError("One or more selected users no longer exist") queued = await enqueue_admin_email( recipient_emails=[user.email for user in users], from_email=from_email, subject=body.subject.strip(), html_body=body.html_body, requested_by=str(admin.id), scheduled_at=scheduled_at, ) if not queued: raise HTTPException(status_code=503, detail="Email queue is unavailable") logger.info( "admin_email_queued", admin_id=str(admin.id), recipient_count=len(users), scheduled_at=scheduled_at.isoformat() if scheduled_at else None, ) return AdminEmailSendResponse(recipient_count=len(users), scheduled_at=scheduled_at) @router.post("/admins", response_model=CreatePlatformAdminResponse) async def create_platform_admin( body: CreatePlatformAdminBody, admin: User = Depends(require_admin), db: AsyncSession = Depends(get_db), ): """Add a platform admin: link an existing user, or create a minimal user from email then link.""" user_repo = UserRepository(db) admin_repo = AdminRepository(db) created_new_user = False if body.user_id is not None: user = await user_repo.get_by_id(body.user_id) if not user: raise NotFoundError("User not found") else: assert body.email is not None email = str(body.email).lower() user = await user_repo.get_by_email(email) if not user: user = await user_repo.create( email=email, first_name=body.first_name, last_name=body.last_name, phone=body.phone, company=body.company, ) created_new_user = True _, created_admin_record = await admin_repo.ensure_platform_admin( user.id, created_by=admin.id, ) return CreatePlatformAdminResponse( user_id=user.id, email=user.email, created_new_user=created_new_user, created_admin_record=created_admin_record, )