"""Cloudflare Pages deployment plane adapter. Responsibilities: - Create / ensure CF Pages project via API (sourceless — no GitHub OAuth required) - Poll deployment status for a specific commit SHA on a given branch - Return stable CF Pages URLs (deterministic from project_ref — known before first deploy) Deployment flow: Code is pushed to GitHub by the backend. A GitHub Actions workflow in the project repo (``deploy.yml``) triggers on push, builds the project, and uploads the built output directly to CF Pages via ``wrangler pages deploy --commit-hash``. This adapter is only responsible for: 1. Ensuring the CF Pages project exists (created sourceless via API on first deploy) 2. Polling the CF Pages deployments API until the GitHub Actions upload arrives """ from __future__ import annotations import json from typing import Any, Optional from urllib.parse import quote import httpx import structlog from src.config import Settings, get_settings from src.domain.hosting.deployment_targets import ( get_pages_build_config, pages_preview_branch_url, pages_production_url, pages_project_name, ) from src.domain.hosting.types import DeployJobContext, DeploymentOutcome logger = structlog.get_logger() class CloudflareAPIError(RuntimeError): pass class CloudflarePagesProvider: """CF Pages deployment plane — create projects, poll deployment status.""" def __init__(self, settings: Settings | None = None): self._s = settings or get_settings() # ------------------------------------------------------------------ # settings helpers # ------------------------------------------------------------------ @property def _token(self) -> str: return str(self._s.cloudflare_api_token).strip() @property def _account_id(self) -> str: return str(self._s.cloudflare_account_id).strip() def _base(self) -> str: return f"https://api.cloudflare.com/client/v4/accounts/{self._account_id}" # ------------------------------------------------------------------ # low-level HTTP # ------------------------------------------------------------------ async def _request( self, method: str, path: str, *, json_body: Any = None, expected: set[int] | None = None, allow_statuses: set[int] | None = None, ) -> tuple[int, Any]: expected = expected or {200} allow_statuses = allow_statuses or set() headers = {"Authorization": f"Bearer {self._token}", "Content-Type": "application/json"} url = f"{self._base()}{path}" async with httpx.AsyncClient(timeout=60.0) as client: resp = await client.request(method.upper(), url, json=json_body, headers=headers) if resp.status_code in allow_statuses: return resp.status_code, None if resp.status_code not in expected: try: detail: Any = resp.json() except Exception: detail = resp.text raise CloudflareAPIError( f"CF Pages API {method.upper()} {path} -> {resp.status_code}: {detail}" ) if not resp.content: return resp.status_code, {} payload = resp.json() if isinstance(payload, dict) and "result" in payload: return resp.status_code, payload["result"] return resp.status_code, payload # ------------------------------------------------------------------ # project management # ------------------------------------------------------------------ async def ensure_pages_project( self, project_ref: str, project_type: str, ) -> dict[str, Any]: """Idempotent: create CF Pages project (sourceless) if it does not exist. Sourceless means CF Pages does not connect to GitHub directly — deployments arrive via ``wrangler pages deploy`` from the project's GitHub Actions workflow. The workflow is injected into the repo by the backend during project setup. Returns the CF Pages project dict with an added ``repo_linkage_status`` key: - ``"direct_upload"`` — project exists and accepts direct-upload deployments """ project_name = pages_project_name(project_ref) # Idempotent: return existing project unchanged _, existing = await self._request( "GET", f"/pages/projects/{quote(project_name, safe='')}", expected={200}, allow_statuses={404}, ) if existing: logger.info("cf_pages_project_exists", project_name=project_name) existing["repo_linkage_status"] = "direct_upload" return existing build_cfg = get_pages_build_config(project_type) body = { "name": project_name, "production_branch": "main", "build_config": { "build_command": build_cfg.build_command, "destination_dir": build_cfg.destination_dir or ".", "root_dir": build_cfg.root_dir, }, } _, result = await self._request("POST", "/pages/projects", json_body=body, expected={200, 201}) logger.info("cf_pages_project_created", project_name=project_name) result["repo_linkage_status"] = "direct_upload" return result # ------------------------------------------------------------------ # deployment status polling # ------------------------------------------------------------------ async def find_deployment_for_sha( self, project_ref: str, commit_sha: str, branch: str, ) -> Optional[dict[str, Any]]: """Return the CF Pages deployment dict for this commit SHA + branch, or None if not yet deployed.""" project_name = pages_project_name(project_ref) _, deployments = await self._request( "GET", f"/pages/projects/{quote(project_name, safe='')}/deployments", expected={200}, allow_statuses={404}, ) if not deployments: return None if isinstance(deployments, dict): deployments = deployments.get("result") or [] for d in deployments: trigger = d.get("deployment_trigger") or {} meta = trigger.get("metadata") or {} if meta.get("commit_hash", "").startswith(commit_sha[:12]) and meta.get("branch") == branch: return d return None async def find_latest_live_deployment_for_branch( self, project_ref: str, branch: str, ) -> Optional[dict[str, Any]]: """Return the most recent successful deployment on *branch*, or None. Used by the recovery scanner to reconcile projects where CF Pages has a live deployment but the backend never recorded it (e.g. manual GA re-run, SHA mismatch). """ project_name = pages_project_name(project_ref) _, deployments = await self._request( "GET", f"/pages/projects/{quote(project_name, safe='')}/deployments", expected={200}, allow_statuses={404}, ) if not deployments: return None if isinstance(deployments, dict): deployments = deployments.get("result") or [] for d in deployments: trigger = d.get("deployment_trigger") or {} meta = trigger.get("metadata") or {} if meta.get("branch") != branch: continue latest_stage = d.get("latest_stage") or {} if latest_stage.get("name") == "deploy" and latest_stage.get("status") == "success": return d return None async def is_deployment_live(self, deployment: dict[str, Any]) -> bool: """True when the deployment has successfully reached the deploy stage.""" stages: list[dict] = deployment.get("stages") or [] for stage in stages: if stage.get("name") == "deploy" and stage.get("status") == "success": return True latest = deployment.get("latest_stage") or {} return latest.get("name") == "deploy" and latest.get("status") == "success" async def retry_latest_branch_deployments( self, project_ref: str, branches: tuple[str, ...], ) -> list[str]: """Retry the latest successful deployment on each branch so env changes apply.""" retried: list[str] = [] seen: set[str] = set() project_name = pages_project_name(project_ref) for branch in branches: if not branch or branch in seen: continue seen.add(branch) deploy = await self.find_latest_live_deployment_for_branch(project_ref, branch) deploy_id = str((deploy or {}).get("id") or "") if not deploy_id: continue status, _ = await self._request( "POST", ( f"/pages/projects/{quote(project_name, safe='')}" f"/deployments/{quote(deploy_id, safe='')}/retry" ), expected={200, 202}, allow_statuses={400, 404}, ) if status in {200, 202}: retried.append(deploy_id) return retried async def is_deployment_failed(self, deployment: dict[str, Any]) -> bool: latest = deployment.get("latest_stage") or {} return latest.get("status") == "failure" # ------------------------------------------------------------------ # DeploymentPlaneProvider protocol methods # ------------------------------------------------------------------ async def ensure_project_environment(self, ctx: DeployJobContext) -> DeploymentOutcome: """Ensure CF Pages project exists (creates sourceless project if absent).""" try: result = await self.ensure_pages_project( project_ref=ctx.project_ref, project_type=ctx.framework, ) return DeploymentOutcome( ok=True, deploy_url=pages_production_url(ctx.project_ref), provider_resource_id=result.get("name"), last_completed_step="ensure_project_environment", metadata={ "adapter": "cloudflare_pages", "pages_project_name": result.get("name"), "repo_linkage_status": result.get("repo_linkage_status", "direct_upload"), "subdomain": result.get("subdomain"), }, ) except CloudflareAPIError as exc: return DeploymentOutcome( ok=False, last_completed_step="ensure_project_environment", message=str(exc), metadata={"adapter": "cloudflare_pages"}, ) async def get_deployment_status(self, ctx: DeployJobContext) -> DeploymentOutcome: """Poll for a live deployment at ctx.git_sha on ctx.environment branch.""" branch = _branch_for_env(ctx.environment, self._s.hosting_preview_branch) try: deploy = await self.find_deployment_for_sha(ctx.project_ref, ctx.git_sha, branch) except CloudflareAPIError as exc: return DeploymentOutcome(ok=False, message=str(exc), metadata={"adapter": "cloudflare_pages"}) if deploy is None: return DeploymentOutcome( ok=False, activation_status="not_found", last_completed_step="get_deployment_status", message="No CF Pages deployment found for this SHA yet", metadata={"adapter": "cloudflare_pages", "branch": branch}, ) if await self.is_deployment_failed(deploy): return DeploymentOutcome( ok=False, provider_resource_id=deploy.get("id"), activation_status="failed", last_completed_step="get_deployment_status", message="CF Pages deployment failed", metadata={"adapter": "cloudflare_pages", "deployment": deploy}, ) is_live = await self.is_deployment_live(deploy) target_url = _target_url_for_env(ctx.project_ref, ctx.environment, self._s.hosting_preview_branch) return DeploymentOutcome( ok=is_live, deploy_url=target_url, provider_resource_id=deploy.get("id"), activation_status="live" if is_live else "building", last_completed_step="get_deployment_status", metadata={ "adapter": "cloudflare_pages", "cf_pages_url": deploy.get("url"), "branch": branch, }, ) def _branch_for_env(environment: str, preview_branch: str) -> str: return preview_branch if (environment or "").lower() == "preview" else "main" def _target_url_for_env(project_ref: str, environment: str, preview_branch: str) -> str: if (environment or "").lower() == "preview": return pages_preview_branch_url(project_ref, preview_branch) return pages_production_url(project_ref)