"""Deployments domain - business logic.""" from __future__ import annotations import re from typing import Optional from uuid import UUID import httpx import structlog from src.core.exceptions import ConflictError, NotFoundError, ValidationError from src.domain.deployments.repository import DeployRepository, DeploymentTargetRepository, DomainRepository from src.domain.deployments.schemas import DnsInstructions, DnsRecord logger = structlog.get_logger() _DOMAIN_RE = re.compile( r'^[a-z0-9]([a-z0-9\-]{0,61}[a-z0-9])?' r'(\.[a-z0-9]([a-z0-9\-]{0,61}[a-z0-9])?)+$' ) def _normalize_domain(raw: str) -> str: d = raw.strip().lower() for prefix in ("https://", "http://"): if d.startswith(prefix): d = d[len(prefix):] return d.rstrip("/").rstrip(".").strip() def build_dns_instructions(domain: str, verification_txt: str, cname_target: str) -> DnsInstructions: """Build the step-by-step DNS records a user must add for a custom domain.""" domain = _normalize_domain(domain) parts = domain.split(".") is_apex = len(parts) == 2 txt_host = f"_p2hs-verify.{domain}" records: list[DnsRecord] = [ DnsRecord( step=1, purpose="verify", type="TXT", host=txt_host, value=verification_txt, ttl=300, note="Proves you own this domain. Add this record first.", ), ] if is_apex: records.append(DnsRecord( step=2, purpose="route", type="ALIAS", host=domain, value=cname_target, ttl=300, note=( "Some registrars call this ANAME or CNAME Flattening. " "If your registrar doesn't support ALIAS, use Cloudflare (free) " "and add a proxied CNAME @ → " + cname_target ), )) apex_note = ( f"Apex domains ({domain}) require ALIAS or ANAME support. " "Cloudflare, Route53, and Namecheap support this. " f"Alternatively, redirect {domain} → www.{domain} and add " f"www as a CNAME to {cname_target}." ) else: records.append(DnsRecord( step=2, purpose="route", type="CNAME", host=domain, value=cname_target, ttl=300, note="Routes traffic from your domain to P2HS. Add after verification.", )) apex_note = None return DnsInstructions( records=records, is_apex=is_apex, apex_note=apex_note, ) async def check_dns_txt(hostname: str, expected_value: str) -> bool: """Check that hostname has the expected TXT record value using Cloudflare DoH.""" try: async with httpx.AsyncClient(timeout=10.0) as client: resp = await client.get( "https://cloudflare-dns.com/dns-query", params={"name": hostname, "type": "TXT"}, headers={"Accept": "application/dns-json"}, ) resp.raise_for_status() data = resp.json() answers = data.get("Answer") or [] for ans in answers: rdata = str(ans.get("data", "")).strip('"') if rdata == expected_value: return True return False except Exception as exc: logger.warning("dns_txt_check_failed", hostname=hostname, error=str(exc)) return False def domain_response(d, cname_target: str | None = None): """Build a DomainResponse with dns_instructions always populated. Imported by both builder/router.py and deployments/router.py so the response shape is consistent regardless of which router handles the request. """ from src.domain.deployments.schemas import DomainResponse if cname_target is None: from src.config import get_settings cname_target = get_settings().hosting_cname_target return DomainResponse( id=d.id, project_id=d.project_id, domain=d.domain, environment=d.environment or "production", verification_status=d.verification_status, verification_txt=d.verification_txt, is_primary=d.is_primary, tls_status=d.tls_status, created_at=d.created_at, verified_at=d.verified_at, dns_instructions=build_dns_instructions(d.domain, d.verification_txt or "", cname_target), ) class DeployService: def __init__( self, deploy_repo: DeployRepository, domain_repo: DomainRepository, target_repo: Optional[DeploymentTargetRepository] = None, project_repo=None, revision_repo=None, ): self._deploy_repo = deploy_repo self._domain_repo = domain_repo self._target_repo = target_repo self._project_repo = project_repo self._revision_repo = revision_repo async def create_deploy(self, project_id: UUID, revision_id: UUID, environment: str, created_by: Optional[UUID] = None): active = await self._deploy_repo.get_active_for_project_environment( project_id, environment ) if active: raise ConflictError( f"A {environment} deployment is already in progress. " "Wait for it to succeed or fail before starting another deployment." ) return await self._deploy_repo.create(project_id, revision_id, environment, created_by) async def list_deploys(self, project_id: UUID): return await self._deploy_repo.list_for_project(project_id) async def rollback(self, deploy_id: UUID, user_id: UUID): deploy = await self._deploy_repo.get_by_id(deploy_id) if not deploy: raise NotFoundError("Deploy not found") revision_id = deploy.revision_id previous_sha = (deploy.deployment_metadata or {}).get("previous_production_sha") if previous_sha and self._revision_repo: previous_revision = await self._revision_repo.get_by_git_sha( deploy.project_id, previous_sha ) if previous_revision: revision_id = previous_revision.id return await self._deploy_repo.create( project_id=deploy.project_id, revision_id=revision_id, environment=deploy.environment, created_by=user_id, deployment_metadata={ "rollback_of_deploy_id": str(deploy.id), "rollback_to_sha": previous_sha, "rollback_source_ref": ( f"refs/p2hs/backups/{deploy.id}" if previous_sha else None ), }, ) async def add_domain(self, project_id: UUID, domain: str, environment: str): domain = _normalize_domain(domain) if not _DOMAIN_RE.match(domain): raise ValidationError("Invalid domain name. Use a valid hostname like myapp.example.com.") existing = await self._domain_repo.get_by_domain(domain) if existing: if str(existing.project_id) == str(project_id): raise ValidationError("This domain is already added to this project.") raise ValidationError("This domain is already claimed by another project.") d = await self._domain_repo.create(project_id, domain, environment) # Wire CF for SaaS + KV routing immediately — DNS propagation is the customer's side, # not ours. When their CNAME propagates, everything is already in place. await self._activate_custom_domain_routing(d) return d async def verify_domain(self, domain_id: UUID) -> tuple: """Check DNS TXT record then mark domain verified if found. Returns (domain_row, dns_verified: bool). After successful verification, activates KV routing for the custom domain. """ d = await self._domain_repo.get_by_id(domain_id) if not d: raise NotFoundError("Domain not found") if d.verified_at: # Already verified — retry CF wiring if it wasn't completed (e.g. added before # this feature existed, or failed due to a transient error / missing deployment). if not d.cf_hostname_id: await self._activate_custom_domain_routing(d) return d, True txt_host = f"_p2hs-verify.{d.domain}" dns_ok = await check_dns_txt(txt_host, d.verification_txt or "") if dns_ok: d = await self._domain_repo.mark_verified(domain_id) logger.info("domain_verified", domain=d.domain, domain_id=str(domain_id)) # Retry wiring only if add_domain didn't complete it (e.g. project had no # deployment target yet at the time the domain was added). if not d.cf_hostname_id: await self._activate_custom_domain_routing(d) else: logger.info("domain_verify_txt_not_found", domain=d.domain, txt_host=txt_host) return d, dns_ok async def _activate_custom_domain_routing(self, domain) -> None: """Register with CF for SaaS and write KV routing for the custom hostname. Steps: 1. POST /zones/{zone_id}/custom_hostnames — CF begins CNAME validation + TLS provisioning 2. Store the CF hostname ID on the domain row (needed for status polling and deletion) 3. Write the KV routing entry so the dispatcher forwards traffic to this project """ from src.config import get_settings from src.domain.hosting.cf_saas import CFSaaSClient, CFSaaSError from src.domain.hosting.routing_plane import CloudflareAPIError, CloudflareRoutingProvider settings = get_settings() hosting_key = (settings.hosting_provider or "").strip() if hosting_key in {"", "stub"}: logger.info("custom_domain_routing_skipped", reason="hosting_provider_unset", domain=domain.domain) return if not self._target_repo or not self._project_repo: logger.warning("custom_domain_routing_skipped", reason="repos_not_injected", domain=domain.domain) return if not (settings.cloudflare_zone_id or "").strip(): logger.warning("custom_domain_routing_skipped", reason="CLOUDFLARE_ZONE_ID_not_set", domain=domain.domain) return env = domain.environment or "production" target = await self._target_repo.get(domain.project_id, env) if not target: logger.warning("custom_domain_routing_skipped", reason="no_deployment_target", domain=domain.domain, env=env) return target_url = target.provider_production_url if env == "production" else target.provider_preview_url if not target_url: logger.warning("custom_domain_routing_skipped", reason="no_provider_url", domain=domain.domain, env=env) return project = await self._project_repo.get_by_id(domain.project_id) if not project or not project.project_ref: logger.warning("custom_domain_routing_skipped", reason="no_project_ref", domain=domain.domain) return # Step 1: register with CF for SaaS (idempotent — reuse existing if already registered) saas = CFSaaSClient(settings) try: cf_obj = await saas.find_by_hostname(domain.domain) if not cf_obj: cf_obj = await saas.create(domain.domain) cf_hostname_id = cf_obj.get("id") or "" except CFSaaSError as exc: logger.error("cf_saas_create_failed", domain=domain.domain, error=str(exc)) return # Step 2: persist CF hostname ID so we can poll and delete later if cf_hostname_id: await self._domain_repo.update_cf_hostname( domain.id, cf_hostname_id=cf_hostname_id, zone_id=settings.cloudflare_zone_id, ) # Step 3: write KV routing entry. W4P copies the canonical hostname's # record so project_id, script_name, and api_execution stay identical. # Pages keeps the proxy alias. try: from src.domain.hosting.project_provider import W4P_STATIC, project_hosting_provider provider = CloudflareRoutingProvider(settings) deploy_history = await self._deploy_repo.list_for_project(domain.project_id) has_successful_pages_deployment = any( d.status == "live" and (d.hosting_provider or "").strip().lower() in {"cloudflare_pages", "cloudflare"} for d in deploy_history ) project_provider = project_hosting_provider( getattr(project, "created_at", None), settings, primary_hosting_provider=getattr(project, "primary_hosting_provider", None), assume_unassigned_w4p=not has_successful_pages_deployment, has_successful_pages_deployment=has_successful_pages_deployment, ) descriptor = target.routing_target_descriptor or {} if project_provider == W4P_STATIC or descriptor.get("target_kind") == "w4p_static": from urllib.parse import urlparse from src.domain.hosting.factory import get_w4p_static_provider from src.domain.hosting.hostnames import canonical_live_url, canonical_preview_url apex = settings.hosting_public_apex canonical = canonical_preview_url(project.project_ref, apex) if env == "preview" else canonical_live_url(project.project_ref, apex) source_host = urlparse(canonical).netloc await get_w4p_static_provider(settings, environment=env).copy_route(source_host, domain.domain) else: await provider.alias_custom_hostname( custom_hostname=domain.domain, project_ref=project.project_ref, environment=env, target_url=target_url, ) logger.info("custom_domain_activated", domain=domain.domain, cf_hostname_id=cf_hostname_id) except CloudflareAPIError as exc: logger.error("custom_domain_kv_write_failed", domain=domain.domain, error=str(exc)) async def check_dns_status(self, domain_id: UUID) -> dict: """DNS + TLS status check. Marks tls_provisioned_at when CF cert goes active.""" from src.config import get_settings from src.domain.hosting.cf_saas import CFSaaSClient, CFSaaSError d = await self._domain_repo.get_by_id(domain_id) if not d: raise NotFoundError("Domain not found") settings = get_settings() txt_host = f"_p2hs-verify.{d.domain}" txt_ok = await check_dns_txt(txt_host, d.verification_txt or "") cname_ok = False try: async with httpx.AsyncClient(timeout=10.0) as client: resp = await client.get( "https://cloudflare-dns.com/dns-query", params={"name": d.domain, "type": "CNAME"}, headers={"Accept": "application/dns-json"}, ) data = resp.json() answers = data.get("Answer") or [] for ans in answers: rdata = str(ans.get("data", "")).rstrip(".").lower() if rdata == settings.hosting_cname_target.lower(): cname_ok = True break except Exception: pass # Poll CF for SaaS cert status when we have a registered hostname ID cf_status: dict = {} if d.cf_hostname_id and (settings.cloudflare_zone_id or "").strip(): try: saas = CFSaaSClient(settings) cf_obj = await saas.get(d.cf_hostname_id) cf_status = CFSaaSClient.parse_status(cf_obj) # Transition tls_provisioned_at the moment CF reports the cert active if cf_status["tls_active"] and d.tls_provisioned_at is None: await self._domain_repo.mark_tls_provisioned(domain_id) logger.info("domain_tls_provisioned", domain=d.domain, cf_hostname_id=d.cf_hostname_id) except CFSaaSError as exc: logger.warning("cf_saas_status_poll_failed", domain=d.domain, error=str(exc)) return { "domain": d.domain, "verification_status": d.verification_status, "already_verified": d.verified_at is not None, "txt_found": txt_ok, "cname_found": cname_ok, "tls_status": d.tls_status, **cf_status, } async def set_primary_domain(self, domain_id: UUID): d = await self._domain_repo.set_primary(domain_id) if not d: raise NotFoundError("Domain not found") return d async def delete_domain(self, domain_id: UUID): d = await self._domain_repo.get_by_id(domain_id) if not d: return if d.verified_at: from src.config import get_settings from src.domain.hosting.cf_saas import CFSaaSClient, CFSaaSError from src.domain.hosting.routing_plane import CloudflareAPIError, CloudflareRoutingProvider settings = get_settings() if (settings.hosting_provider or "").strip() == "cloudflare_pages": # Remove KV routing entry try: provider = CloudflareRoutingProvider(settings) await provider.remove_custom_hostname(d.domain) logger.info("custom_domain_kv_removed", domain=d.domain) except CloudflareAPIError as exc: logger.warning("custom_domain_kv_remove_failed", domain=d.domain, error=str(exc)) # Remove CF for SaaS custom hostname (stops TLS + proxying) cf_hostname_id = d.cf_hostname_id if cf_hostname_id and (settings.cloudflare_zone_id or "").strip(): try: saas = CFSaaSClient(settings) await saas.delete(cf_hostname_id) except CFSaaSError as exc: logger.warning("cf_saas_delete_failed", domain=d.domain, cf_hostname_id=cf_hostname_id, error=str(exc)) await self._domain_repo.delete(domain_id)