from __future__ import annotations from uuid import UUID import httpx from fastapi import APIRouter, Depends, HTTPException, Request from sqlalchemy.ext.asyncio import AsyncSession from src.core.database import get_db from src.core.dependencies import require_active_billing_for_workspace_id from src.domain.auth.models import User from src.domain.auth.router import get_current_user from src.domain.domains.repository import DomainRepository from src.domain.domains.schemas import DomainAssignmentCreate, DomainAssignmentResponse, DomainCapabilitiesResponse, DomainConnectRequest, DomainSearchResponse, DomainCheckRequest, DomainCheckoutRequest, WorkspaceDomainResponse from src.domain.domains.service import DomainService from src.config import get_settings from src.domain.domains.providers import CloudflareRegistrarProvider from src.integrations.stripe_client import StripeClientImpl from src.domain.billing.repository import StripeCustomerRepository from src.domain.workspaces.repository import WorkspaceRepository from src.core.app_url import resolve_checkout_redirect_urls from src.core.exceptions import ValidationError from datetime import datetime, timezone from decimal import Decimal, ROUND_HALF_UP router = APIRouter(tags=["domains"]) def svc(db: AsyncSession = Depends(get_db)) -> DomainService: return DomainService(DomainRepository(db)) async def response(repo: DomainRepository, d): payload = {**d.__dict__, "assignments": [DomainAssignmentResponse.model_validate(x) for x in await repo.assignments(d.id)]} token = (d.metadata_json or {}).get("verification_token") payload["verification_record"] = {"type": "TXT", "name": f"_p2hs-verify.{d.normalized_domain}", "value": token, "ttl": 300} if token and d.verification_status != "verified" else None return WorkspaceDomainResponse.model_validate(payload) @router.get("/api/workspaces/{workspace_id}/domains/capabilities", response_model=DomainCapabilitiesResponse) async def capabilities(workspace_id: UUID = Depends(require_active_billing_for_workspace_id), user: User = Depends(get_current_user)): s = get_settings() registrar_configured = bool((s.cloudflare_registrar_api_token or s.cloudflare_api_token) and (s.cloudflare_registrar_account_id or s.cloudflare_account_id)) enabled = s.domain_registrar_transactions_enabled and registrar_configured return DomainCapabilitiesResponse(registration=enabled, transfer_in=enabled, renewal=enabled, dns_management=bool(s.cloudflare_api_token and s.cloudflare_account_id), reason="Automatic renewal follows your P2HS billing authorization.") @router.get("/api/workspaces/{workspace_id}/domains/search", response_model=DomainSearchResponse) async def search_domains(q: str, workspace_id: UUID = Depends(require_active_billing_for_workspace_id), user: User = Depends(get_current_user)): settings = get_settings() if not (settings.cloudflare_registrar_api_token or settings.cloudflare_api_token) or not (settings.cloudflare_registrar_account_id or settings.cloudflare_account_id): raise ValidationError("Domain search is currently unavailable") try: result = await CloudflareRegistrarProvider().search(q) except httpx.HTTPStatusError as exc: raise HTTPException(status_code=502, detail=_registrar_http_error(exc)) from exc except httpx.HTTPError as exc: raise HTTPException(status_code=503, detail="Domain search is temporarily unavailable") from exc return DomainSearchResponse(result=_decorate_domain_result(result, settings)) @router.post("/api/workspaces/{workspace_id}/domains/check", response_model=DomainSearchResponse) async def check_domain(body: DomainCheckRequest, workspace_id: UUID = Depends(require_active_billing_for_workspace_id), user: User = Depends(get_current_user)): try: result = await CloudflareRegistrarProvider().check_availability(body.domain) except httpx.HTTPStatusError as exc: raise HTTPException(status_code=502, detail=_registrar_http_error(exc)) from exc except httpx.HTTPError as exc: raise HTTPException(status_code=503, detail="Domain availability is temporarily unavailable") from exc return DomainSearchResponse(result=_decorate_domain_result(result, get_settings())) def _registrar_http_error(exc: httpx.HTTPStatusError) -> str: if exc.response.status_code in (401, 403): return "The domain service rejected its account authorization. Please contact support." if exc.response.status_code == 429: return "Domain search is temporarily busy; please retry shortly." return f"The domain service returned HTTP {exc.response.status_code}." @router.post("/api/workspaces/{workspace_id}/domains/checkout") async def create_domain_checkout(request: Request, body: DomainCheckoutRequest, workspace_id: UUID = Depends(require_active_billing_for_workspace_id), user: User = Depends(get_current_user), db: AsyncSession = Depends(get_db)): settings = get_settings() if not settings.stripe_secret_key: raise ValidationError("Stripe billing is not configured") if not settings.domain_registrar_transactions_enabled: raise ValidationError("Domain purchases are currently unavailable") domain = body.domain.strip().lower().rstrip(".") try: quote = await CloudflareRegistrarProvider().check_availability(domain) except httpx.HTTPStatusError as exc: raise HTTPException(status_code=502, detail=_registrar_http_error(exc)) from exc except httpx.HTTPError as exc: raise HTTPException(status_code=503, detail="Domain pricing is temporarily unavailable") from exc item = _quote_item(quote, domain) available = item.get("available", item.get("registrable", True)) if isinstance(item, dict) else True if available is False: raise ValidationError("This domain is not available for registration") raw_registration, _raw_renewal, registration_minor, renewal_minor = _quote_prices(item) if raw_registration is None: raise ValidationError("Could not determine the current domain price") cost_minor = registration_minor + (body.years - 1) * renewal_minor charge_minor = int((Decimal(cost_minor) * (Decimal(100 + settings.domain_markup_percent) / Decimal(100)) + Decimal(settings.domain_markup_fixed_usd_cents)).quantize(Decimal("1"), rounding=ROUND_HALF_UP)) customer_repo = StripeCustomerRepository(db) customer = await customer_repo.get_by_workspace(workspace_id) workspace = await WorkspaceRepository(db).get_by_id(workspace_id) stripe = StripeClientImpl() if not customer: customer_id = await stripe.create_customer(user.email, getattr(workspace, "name", "P2HS workspace")) customer = await customer_repo.create(workspace_id, customer_id) product = await stripe.create_product(f"P2HS domain — {domain}", "Annual domain registration managed by P2HS", {"domain_name": domain, "purchase_type": "domain_registration"}) price = await stripe.create_price(product["id"], charge_minor, "usd", "year", f"P2HS domain — {domain}", interval_count=body.years, recurring=body.auto_renew) from src.domain.domains.models import DomainOrder from src.core.security.encryption import encrypt_value order = DomainOrder(workspace_id=workspace_id, order_type="registration", provider="cloudflare_registrar", quoted_amount_minor=cost_minor, charged_amount_minor=charge_minor, provider_cost_minor=cost_minor, currency="USD", idempotency_key=body.idempotency_key or f"domain:{workspace_id}:{domain}:{body.years}:{body.auto_renew}", stripe_price_id=price["id"], metadata_json={"domain_name": domain, "years": body.years, "auto_renew": body.auto_renew, "registrant_enc": encrypt_value(__import__('json').dumps(body.registrant))}) db.add(order) await db.flush() success_url, cancel_url = resolve_checkout_redirect_urls(request, None, None) checkout_metadata = {"purchase_type": "domain_registration", "workspace_id": str(workspace_id), "domain_order_id": str(order.id), "domain_name": domain, "years": str(body.years), "auto_renew": str(body.auto_renew).lower()} if body.auto_renew: url, session_id = await stripe.create_subscription_checkout(customer.stripe_customer_id, price["id"], success_url, cancel_url, checkout_metadata, order.idempotency_key) else: url, session_id = await stripe.create_payment_checkout(customer.stripe_customer_id, price["id"], success_url, cancel_url, checkout_metadata, order.idempotency_key) order.stripe_checkout_session_id = session_id await db.commit() return {"checkout_url": url, "session_id": session_id, "domain_order_id": str(order.id), "amount_minor": charge_minor, "currency": "usd"} def _quote_amount_minor(value) -> int: amount = Decimal(str(value)) return int((amount * 100).quantize(Decimal("1"), rounding=ROUND_HALF_UP)) if amount < 10000 else int(amount) def _quote_item(payload: dict, domain: str) -> dict: result = payload.get("result", payload) if isinstance(payload, dict) else {} if isinstance(result, dict): domains = result.get("domains") if isinstance(domains, list): for item in domains: if isinstance(item, dict) and (item.get("name") or item.get("domain_name")) == domain: return item return domains[0] if domains and isinstance(domains[0], dict) else {} candidate = result.get(domain) if isinstance(candidate, dict): return candidate return result return {} def _quote_prices(item: dict) -> tuple[object, object, int, int]: pricing = item.get("pricing") or {} if pricing.get("registration_cost") is not None: registration = pricing.get("registration_cost") registration_minor = _quote_amount_minor(registration) elif item.get("registration_price") is not None: registration = item.get("registration_price") registration_minor = _quote_amount_minor(registration) elif item.get("price") is not None: registration = item.get("price") registration_minor = _quote_amount_minor(registration) elif item.get("price_minor") is not None: registration = item.get("price_minor") registration_minor = int(registration) else: registration = None registration_minor = 0 if pricing.get("renewal_cost") is not None: renewal = pricing.get("renewal_cost") renewal_minor = _quote_amount_minor(renewal) elif item.get("renewal_price") is not None: renewal = item.get("renewal_price") renewal_minor = _quote_amount_minor(renewal) else: renewal = registration renewal_minor = registration_minor return registration, renewal, registration_minor, renewal_minor def _retail_minor(cost_minor: int, settings) -> int: return int((Decimal(cost_minor) * (Decimal(100 + settings.domain_markup_percent) / Decimal(100)) + Decimal(settings.domain_markup_fixed_usd_cents)).quantize(Decimal("1"), rounding=ROUND_HALF_UP)) def _decorate_domain_item(item: dict, settings) -> dict: if not isinstance(item, dict): return item registration, renewal, registration_minor, renewal_minor = _quote_prices(item) if registration is None: return item retail_registration = _retail_minor(registration_minor, settings) retail_renewal = _retail_minor(renewal_minor, settings) decorated = dict(item) decorated["p2hs_pricing"] = { "registration_minor": retail_registration, "renewal_minor": retail_renewal, "registration": f"{retail_registration / 100:.2f}", "renewal": f"{retail_renewal / 100:.2f}", } return decorated def _decorate_domain_result(payload: dict, settings) -> dict: result = payload.get("result", payload) if isinstance(payload, dict) else {} if not isinstance(result, dict): return {} if isinstance(result.get("domains"), list): decorated = dict(result) decorated["domains"] = [_decorate_domain_item(item, settings) for item in result["domains"]] return decorated return _decorate_domain_item(result, settings) @router.get("/api/workspaces/{workspace_id}/domains", response_model=list[WorkspaceDomainResponse]) async def list_domains(workspace_id: UUID = Depends(require_active_billing_for_workspace_id), user: User = Depends(get_current_user), db: AsyncSession = Depends(get_db)): repo = DomainRepository(db) return [await response(repo, d) for d in await repo.list(workspace_id)] @router.get("/api/workspaces/{workspace_id}/domains/assignable", response_model=list[WorkspaceDomainResponse]) async def assignable_domains(workspace_id: UUID = Depends(require_active_billing_for_workspace_id), user: User = Depends(get_current_user), db: AsyncSession = Depends(get_db)): repo = DomainRepository(db) return [await response(repo, d) for d in await repo.list(workspace_id) if d.verification_status == "verified"] @router.post("/api/workspaces/{workspace_id}/domains", response_model=WorkspaceDomainResponse) async def connect_domain(body: DomainConnectRequest, workspace_id: UUID = Depends(require_active_billing_for_workspace_id), user: User = Depends(get_current_user), service: DomainService = Depends(svc), db: AsyncSession = Depends(get_db)): d = await service.connect(workspace_id, body.domain, user.id) await db.commit() return await response(service.repo, d) @router.post("/api/workspaces/{workspace_id}/domains/{domain_id}/assignments", response_model=DomainAssignmentResponse) async def assign_domain(body: DomainAssignmentCreate, domain_id: UUID, workspace_id: UUID = Depends(require_active_billing_for_workspace_id), user: User = Depends(get_current_user), service: DomainService = Depends(svc), db: AsyncSession = Depends(get_db)): a = await service.assign(workspace_id, domain_id, body.project_id, body.hostname, body.subdomain, body.environment, user.id) await db.commit() return a @router.delete("/api/workspaces/{workspace_id}/domains/assignments/{assignment_id}") async def unassign_domain(assignment_id: UUID, workspace_id: UUID = Depends(require_active_billing_for_workspace_id), user: User = Depends(get_current_user), service: DomainService = Depends(svc), db: AsyncSession = Depends(get_db)): await service.unassign(workspace_id, assignment_id, user.id) await db.commit() return {"message": "Domain assignment removed"} @router.post("/api/workspaces/{workspace_id}/domains/{domain_id}/verify", response_model=WorkspaceDomainResponse) async def verify_domain(domain_id: UUID, workspace_id: UUID = Depends(require_active_billing_for_workspace_id), user: User = Depends(get_current_user), db: AsyncSession = Depends(get_db)): from src.domain.domains.service import normalize_domain from src.domain.domains.repository import DomainRepository import httpx repo = DomainRepository(db) d = await repo.get(domain_id, workspace_id) if not d: from src.core.exceptions import NotFoundError; raise NotFoundError("Domain not found") token = (d.metadata_json or {}).get("verification_token") async with httpx.AsyncClient(timeout=10) as client: r = await client.get("https://cloudflare-dns.com/dns-query", params={"name": f"_p2hs-verify.{d.normalized_domain}", "type": "TXT"}, headers={"accept": "application/dns-json"}) answers = r.json().get("Answer", []) if r.is_success else [] if any(token and token in str(a.get("data", "")) for a in answers): d.verification_status = "verified" await repo.add_event(workspace_id, "domain.verified", user.id, d.id, {}) await db.commit() return await response(repo, d)