"""Activate a W4P static version from the project's own CI run. The deploy task leaves the deployment waiting. The run builds the site, uploads the version script, then calls this endpoint with its GitHub OIDC token and its project ref. Core finds the project, verifies the run (repository, audience, and the commit named in the token), finds the waiting deployment for that commit, and checks that the version script is the one that deployment produces. Only then does it move the hostname's KV route. Core stays the source of truth for routing. """ from __future__ import annotations import json from typing import Any from urllib.parse import urlparse import httpx import structlog from fastapi import APIRouter, Depends, Header, HTTPException from pydantic import BaseModel, Field from sqlalchemy.ext.asyncio import AsyncSession from src.config import get_settings from src.core.database import get_db from src.domain.deployments.repository import DeployRepository from src.domain.hosting.adapters.w4p_static import version_script_name from src.domain.hosting.ci_activation import ( GITHUB_OIDC_JWKS_URL, WAITING_FOR_CI_STEP, CiActivationError, verify_ci_run_token, ) from src.domain.cloud_infra.repository import ( CFAccountRepository, CloudInstanceRepository, WorkspaceCloudAssignmentRepository, ) from src.domain.cloud_infra.service import CloudInfraService from src.domain.hosting.deploy_target import DeployAccountError, settings_for_project from src.domain.hosting.factory import get_w4p_static_provider from src.domain.hosting.hostnames import canonical_live_url, canonical_preview_url from src.domain.hosting.project_provider import W4P_STATIC from src.domain.hosting.types import DeployJobContext from src.domain.projects.repository import ProjectRepository from src.domain.revisions.repository import RevisionRepository router = APIRouter(prefix="/api/hosting/w4p", tags=["hosting"]) logger = structlog.get_logger() _STATUS_FOR_CODE = { "activation_not_configured": 503, "deploy_account_unresolved": 503, "account_kv_not_configured": 503, "invalid_ci_token": 401, "repository_mismatch": 403, "project_not_found": 404, "deployment_not_found": 404, "deployment_not_waiting": 409, "custom_domain_activation_failed": 503, "migration_destination_unverified": 409, "runtime_health_failed": 503, "runtime_secrets_failed": 503, "routing_write_failed": 503, "activation_internal_error": 500, } def _live_name(env: str, project_ref: str) -> str: return f"p2hs-{'prev' if env == 'preview' else 'prod'}-{project_ref}" def runtime_health_decision(status: int, body: Any, text: str) -> str: """Classify a same-origin health response. ``pending`` means the dispatcher is still on the Cloud proxy or the hostname is not answering as the project runtime yet. Do not roll back or delete scripts. ``unhealthy`` means the project runtime answered and failed its own check. """ if status != 200 or not isinstance(body, dict) or body.get("runtime") != "p2hs-application-runtime": return "pending" lowered = (text or "").lower() if any(marker in lowered for marker in ("cloudgate.", "beapipie.", "/v1/cloud/")): return "unhealthy" return "healthy" if body.get("ok") is True else "unhealthy" def _host_for(env: str, project_ref: str, apex: str) -> str: url = canonical_preview_url(project_ref, apex) if env == "preview" else canonical_live_url(project_ref, apex) return urlparse(url).netloc or "" async def _fetch_jwks() -> dict[str, Any]: async with httpx.AsyncClient(timeout=10.0) as client: r = await client.get(GITHUB_OIDC_JWKS_URL) r.raise_for_status() return r.json() def routing_policy_for_activation(previous_route: str | None) -> str: """First hostname uses manifest routing. An existing route keeps its policy.""" if not previous_route: return "manifest" try: record = json.loads(previous_route) except (TypeError, json.JSONDecodeError): return "legacy" if not isinstance(record, dict): return "legacy" policy = record.get("routing_policy") if policy in {"manifest", "legacy"}: return policy return "legacy" async def _probe_runtime_health(hostname: str) -> str: try: async with httpx.AsyncClient(timeout=4.0) as client: response = await client.get(f"https://{hostname}/api/__p2hs/health") raw_body = response.text try: parsed = response.json() except ValueError: parsed = None return runtime_health_decision(response.status_code, parsed, raw_body) except httpx.HTTPError: return "pending" async def activate_from_ci( db: AsyncSession, *, project_ref: str, version_script: str, bearer: str, settings, fetch_jwks, runtime_health=None, ) -> dict[str, Any]: """Verify the run, then point the hostname at ``version_script``. Raises CiActivationError.""" if not settings.w4p_activation_audience: raise CiActivationError("activation_not_configured") project_repo = ProjectRepository(db) found = await project_repo.get_by_project_ref(project_ref) if found is None: raise CiActivationError("project_not_found") project = await project_repo.get_with_repo(found.id) if not project or not getattr(project, "repo", None): raise CiActivationError("project_or_repo_missing") claims = verify_ci_run_token( bearer, audience=settings.w4p_activation_audience, jwks=await fetch_jwks(), expected_repository=project.repo.repo_full_name, ) sha = claims["sha"].strip() expected_environment = ( "preview" if version_script.startswith(f"p2hs-prev-{project_ref}-") else "production" ) deploy_repo = DeployRepository(db) # Newest first: the waiting deployment for this commit is the one to activate. deployments = await deploy_repo.list_for_project(project.id) # A single commit can legitimately have both a preview and a production # deployment (the two workflows often run at the same time). Never choose # by SHA alone: selecting the other environment makes the generated script # name look wrong and turns a valid activation into version_script_mismatch. matches = [ d for d in deployments if (d.source_git_sha or "") == sha and d.environment == expected_environment ] waiting_matches = [ d for d in matches if d.status == "deploying" and d.last_completed_step == WAITING_FOR_CI_STEP ] if waiting_matches: matches = waiting_matches else: # Activation is intentionally idempotent. A previous attempt may have # persisted the exact CI SHA as live after the route write, while the # workflow itself was retried or the dashboard callback was missed. # Re-run the verified route write rather than rejecting the same exact # artifact as "not waiting". live_matches = [d for d in matches if d.status == "live"] if live_matches: matches = live_matches if not matches: # Project setup creates the bootstrap deployment immediately after the # initial GitHub push. A failed CI run can then be repaired by a new # push before that deployment has a matching SHA. It is safe to adopt # the current verified run only for that one, still-waiting bootstrap # deployment; normal codegen/deploy runs remain exact-SHA only. bootstrap = [ d for d in deployments if d.status in {"deploying", "live"} and ( d.last_completed_step == WAITING_FOR_CI_STEP or d.status == "live" ) and d.environment == expected_environment and (getattr(d, "deployment_metadata", None) or {}).get("bootstrap_initial_hosting") is True and (d.source_git_sha or "") ] if len(bootstrap) == 1: if bootstrap[0].status == "live": await deploy_repo.apply_hosting_outcome( bootstrap[0].id, status="deploying", last_completed_step=WAITING_FOR_CI_STEP, ) bootstrap[0].source_git_sha = sha matches = bootstrap if not matches and expected_environment == "preview" and getattr(project, "active_revision_id", None): # Older projects were created before initial preview deployment rows # were queued. Once the CI token has been verified against the project # repository, create that missing bootstrap row on demand so preview # activation is recoverable without a manual database repair. preview = await deploy_repo.create( project.id, project.active_revision_id, "preview", created_by=None, deployment_metadata={ "bootstrap_initial_hosting": True, "github_repo_full_name": project.repo.repo_full_name, "created_by_ci_activation": True, }, ) await deploy_repo.apply_hosting_outcome( preview.id, status="deploying", source_git_sha=sha, hosting_provider=W4P_STATIC, last_completed_step=WAITING_FOR_CI_STEP, metadata_patch={"target_kind": "w4p_static", "environment": "preview"}, ) matches = [preview] if not matches: # GitHub workflow_run webhooks are useful for normal reconciliation, # but activation must remain recoverable when that webhook is absent # or delayed. The OIDC token already authenticates the repository and # exact commit, so materialize the missing Core revision/deployment # here rather than making CI depend on an unrelated webhook delivery. revision_repo = RevisionRepository(db) revision = await revision_repo.get_by_git_sha(project.id, sha) if revision is None and expected_environment == "preview": fallback = await revision_repo.get_by_source(project.id, "codegen") if fallback is not None: revision = await revision_repo.create( project.id, git_sha=sha, source="codegen", source_id=fallback.source_id, commit_message=f"[p2hs] W4P deployment commit {sha[:12]}", ) logger.info( "w4p_activation_revision_created", project_id=str(project.id), revision_id=str(revision.id), sha=sha[:12], ) if revision is not None and expected_environment == "preview": deployment = await deploy_repo.create( project.id, revision.id, "preview", created_by=None, deployment_metadata={ "created_by_ci_activation": True, "github_repo_full_name": project.repo.repo_full_name, "target_kind": "w4p_static", "environment": "preview", }, ) await deploy_repo.apply_hosting_outcome( deployment.id, status="deploying", source_git_sha=sha, hosting_provider=W4P_STATIC, last_completed_step=WAITING_FOR_CI_STEP, metadata_patch={"target_kind": "w4p_static", "environment": "preview"}, ) matches = [deployment] if not matches: raise CiActivationError("deployment_not_found") deploy = matches[0] if deploy.status != "live" and ( deploy.status != "deploying" or deploy.last_completed_step != WAITING_FOR_CI_STEP ): raise CiActivationError("deployment_not_waiting") env = "preview" if deploy.environment == "preview" else "production" if version_script != version_script_name(_live_name(env, project_ref), sha): raise CiActivationError("version_script_mismatch") # Persist the waiting row before any Cloudflare call. A dropped connection # (Cloudflare's own "error code: 502") otherwise rolls the new row back, # and the workflow retry cannot find the commit it just uploaded. await db.commit() # The route goes to the account this project deploys to, with that account's KV namespace and apex. # Same resolution as the deploy task, so the two cannot disagree about the account. async def _resolve_credentials(workspace_id): infra = CloudInfraService( cf_account_repo=CFAccountRepository(db), instance_repo=CloudInstanceRepository(db), assignment_repo=WorkspaceCloudAssignmentRepository(db), ) return await infra.get_credentials_for_workspace(workspace_id) try: account = await settings_for_project(project.workspace_id, project.created_at, settings, _resolve_credentials) except DeployAccountError as exc: raise CiActivationError("deploy_account_unresolved") from exc if not account.cloudflare_dispatch_kv_namespace_id: raise CiActivationError("account_kv_not_configured") hostname = _host_for(env, project_ref, account.hosting_public_apex) ctx = DeployJobContext( deploy_id=deploy.id, project_id=project.id, workspace_id=project.workspace_id, revision_id=deploy.revision_id, git_sha=sha, environment=env, tenant_ref="", project_ref=project_ref, target_hostname=hostname, repo_full_name=project.repo.repo_full_name, default_branch=project.repo.default_branch or "main", framework=project.type, ) provider = get_w4p_static_provider(account, environment=env) domains = [] route_snapshots: dict[str, str | None] = {} if hasattr(provider, "read_route"): from src.domain.deployments.repository import DomainRepository domains = await DomainRepository(db).list_verified_for_environment(project.id, env) for route_hostname in [hostname, *(domain.domain for domain in domains)]: route_snapshots[route_hostname] = await provider.read_route(route_hostname) async def _restore_routes() -> None: if not hasattr(provider, "restore_route"): return for route_hostname, raw in route_snapshots.items(): try: await provider.restore_route(route_hostname, raw) except Exception: logger.error("w4p_route_restore_failed", hostname=route_hostname, exc_info=True) # The project runtime calls this Cloud origin server-side. The browser does # not. api_execution=project_runtime keeps /api on the project Worker. # bff_origin stays on the record so a rollback to cloud_proxy still has a target. bff_origin = ( getattr(account, "cloud_public_api_url", "") or getattr(account, "cloud_control_api_url", "") or "" ).strip().rstrip("/") or None if bff_origin: try: await provider.bind_runtime_secrets(version_script, { "PROJECT_ID": str(project.id), "CLOUD_GATEWAY_URL": bff_origin, }) except Exception as exc: raise CiActivationError("runtime_secrets_failed") from exc from src.domain.hosting.migration_state import ( mark_cutover_verified, route_activation_allowed, ) if not route_activation_allowed(getattr(project, "hosting_migration", None)): raise CiActivationError("migration_destination_unverified") try: out = await provider.activate_deployment( ctx, version_script, _live_name(env, project_ref), hostname, { "target_kind": "w4p_static", "routing_provider_kind": "cloudflare_w4p", "runtime_contract": "p2hs-bff-v1", "project_id": str(project.id), "project_ref": project_ref, "environment": env, "api_execution": "project_runtime", "routing_policy": routing_policy_for_activation(route_snapshots.get(hostname)), **({"bff_origin": bff_origin} if bff_origin else {}), }, ) except Exception as exc: raise CiActivationError("routing_write_failed") from exc if not out.ok: raise CiActivationError(out.message or "activation_failed") health = await (runtime_health(hostname) if runtime_health else _probe_runtime_health(hostname)) # A brand-new hostname often answers no_route until the dispatcher sees the # KV write. That is pending, not a bad build. Leave the route in place and # finish activation; rolling it back makes every retry miss too. if health == "pending": logger.info( "w4p_activation_runtime_pending", hostname=hostname, project_ref=project_ref, environment=env, ) elif health != "healthy": await _restore_routes() raise CiActivationError("runtime_health_failed") if health == "healthy": previous_name = "" current_raw = await provider.read_route(hostname) if hasattr(provider, "read_route") else None if current_raw: try: previous_name = str(json.loads(json.loads(current_raw).get("previous_record") or "{}").get("script_name") or "") except (TypeError, json.JSONDecodeError): previous_name = "" keep = {version_script} if previous_name: keep.add(previous_name) try: await provider.prune_old_versions(_live_name(env, project_ref), keep) except Exception as exc: logger.warning("w4p_version_cleanup_skipped", hostname=hostname, error=str(exc)) for domain in domains: try: await provider.copy_route(hostname, domain.domain) except Exception as exc: await _restore_routes() raise CiActivationError("custom_domain_activation_failed") from exc canonical_url = f"https://{hostname}/" await deploy_repo.apply_hosting_outcome( deploy.id, status="live", deploy_url=canonical_url, hosting_provider=W4P_STATIC, provider_version_id=sha, source_git_sha=sha, activation_status="live", last_completed_step="routing_activated", metadata_patch={"target_kind": "w4p_static", "version_script": version_script, "canonical_url": canonical_url}, ) if env == "preview": await project_repo.update_hostnames(project.id, preview_hostname=hostname) await _finalize_waiting_preview( db, project.id, deploy.revision_id, canonical_url, active_preview_id=getattr(project, "active_preview_id", None), ) else: await project_repo.update_hostnames(project.id, live_hostname=hostname) verified = mark_cutover_verified( getattr(project, "hosting_migration", None), deploy_id=str(deploy.id), ) if verified is not None: project.hosting_migration = verified await db.commit() await _finalize_waiting_codegen_run(db, project.id, sha) logger.info( "w4p_activation_complete", deploy_id=str(deploy.id), project_id=str(project.id), project_ref=project_ref, environment=env, sha=sha[:12], url=canonical_url, ) logger.info( "project_deploy_complete", deploy_id=str(deploy.id), environment=env, canonical_url=canonical_url, hosting_provider=W4P_STATIC, ) return {"ok": True, "deploy_id": str(deploy.id), "url": canonical_url} async def _finalize_waiting_preview( db: AsyncSession, project_id, revision_id, canonical_url: str, *, active_preview_id=None, ) -> None: """Mark the dashboard's preview card ``running`` now that the site is actually live. Nothing else in the W4P path touches ``preview_environments`` -- it's created once by the codegen worker (``workers/tasks/codegen.py``) as "provisioning" and, unlike the CF Pages flow, nothing ever advances it, so it sits until the 30-minute stale-preview sweep (``project_recovery.py``) marks it "failed" regardless of whether the deploy succeeded. Never raises: a missed update must not fail the activation that already succeeded. """ from src.domain.previews.repository import PreviewRepository try: preview_repo = PreviewRepository(db) pe = ( await preview_repo.get_latest_for_revision(project_id, revision_id) if revision_id is not None else None ) if pe is None: # Older W4P deploy rows can carry a revision different from the # preview row created by the codegen pipeline. Activation is still # authoritative: converge the active/latest pending preview rather # than leaving the dashboard in ``building`` forever. previews = await preview_repo.list_for_project(project_id) pending = [p for p in previews if p.status == "provisioning"] if active_preview_id is not None: pe = next((p for p in pending if p.id == active_preview_id), None) if pe is None and pending: pe = pending[0] if pe is None: logger.warning( "w4p_activation_preview_not_found", project_id=str(project_id), revision_id=str(revision_id) if revision_id else None, active_preview_id=str(active_preview_id) if active_preview_id else None, ) return if pe.status == "running": return await PreviewRepository(db).update_status( pe.id, "running", base_url=canonical_url, provider_preview_url=canonical_url, ) # Persist the authoritative preview state before best-effort timeline # and realtime notifications. A notification failure must not leave a # successfully activated preview stuck in ``provisioning``. await db.commit() try: from src.domain.builder.repository import TimelineEventRepository await TimelineEventRepository(db).create( project_id, "preview_refresh_succeeded", revision_id=revision_id, preview_id=pe.id, phase="preview", title="Preview is live", detail_text=canonical_url, ) await db.commit() except Exception: logger.warning( "w4p_activation_preview_timeline_failed", project_id=str(project_id), preview_id=str(pe.id), exc_info=True, ) logger.info( "w4p_activation_preview_finalized", project_id=str(project_id), preview_id=str(pe.id), revision_id=str(revision_id) if revision_id else None, ) # The CF Pages path publishes this from provision_preview_for_revision.py's # success tail once the preview goes "running"; W4P activation is a separate # completion path that never reached that code (by design -- see 3671dea) and # so never emitted it either. Without it, the dashboard has nothing to react # to and falls back to its own polling/timeout, which is the real cause of # "preview stuck building for minutes" for W4P projects even though the DB # already shows the preview running. try: from src.core.pubsub import get_pubsub_redis, publish_event as _publish_event _pubsub_redis = await get_pubsub_redis() try: await _publish_event(_pubsub_redis, str(project_id), "preview.updated", { "preview_id": str(pe.id), "revision_id": str(revision_id) if revision_id else None, "url": canonical_url, }) finally: await _pubsub_redis.aclose() except Exception: logger.warning( "w4p_activation_preview_ws_publish_failed", project_id=str(project_id), exc_info=True, ) except Exception: logger.warning("w4p_activation_preview_finalize_failed", project_id=str(project_id), exc_info=True) async def _finalize_waiting_codegen_run(db: AsyncSession, project_id, commit_sha: str) -> None: """Hand a codegen run that's waiting on this exact deploy to the normal finalize path. W4P deploys and activates in a separate workflow from the codegen job (see ``w4p_workflow.py``), so unlike the Pages flow there is no in-workflow completion callback once the site actually goes live -- the run would otherwise sit on ``preview_refreshing`` forever and the dashboard with it. This mirrors the short-circuit branch in ``github_webhooks.handlers._finalize_successful_actions_codegen_from_webhook`` (worker_checkpoint == "github_actions_deploying"), triggered here instead of from a workflow_run webhook, since the W4P codegen workflow has no deploy job for that webhook to key off. Never raises: a missed finalize must not fail the activation that already succeeded. """ from sqlalchemy import select from src.domain.codegen.models import CodegenRun from src.domain.codegen.repository import CodegenRunRepository from src.workers.enqueue import enqueue_codegen_run try: result = await db.execute( select(CodegenRun) .where(CodegenRun.project_id == project_id) .where(CodegenRun.status.in_(("running", "preview_refreshing"))) .where(CodegenRun.worker_checkpoint == "github_actions_deploying") .order_by(CodegenRun.created_at.desc()) .limit(1) ) run = result.scalar_one_or_none() if not run: return saved = dict(run.checkpoint_payload or {}) if not saved.get("commit_sha") or saved["commit_sha"] != commit_sha: return repo = CodegenRunRepository(db) await repo.update_status(run.id, "running") await repo.update_checkpoint(run.id, "pushed", saved) await db.commit() await enqueue_codegen_run(str(run.id), project_id=str(project_id)) except Exception: logger.warning("w4p_activation_codegen_finalize_failed", project_id=str(project_id), exc_info=True) class ActivateRequest(BaseModel): project_ref: str = Field(min_length=1, max_length=64) version_script: str = Field(min_length=1, max_length=256) @router.post("/activate") async def activate( body: ActivateRequest, authorization: str | None = Header(default=None), db: AsyncSession = Depends(get_db), ) -> dict[str, Any]: if not authorization or not authorization.startswith("Bearer "): raise HTTPException(status_code=401, detail={"code": "missing_token"}) bearer = authorization.removeprefix("Bearer ").strip() try: return await activate_from_ci( db, project_ref=body.project_ref, version_script=body.version_script, bearer=bearer, settings=get_settings(), fetch_jwks=_fetch_jwks, ) except CiActivationError as exc: raise HTTPException( status_code=_STATUS_FOR_CODE.get(exc.code, 422), detail={"code": exc.code}, ) from exc except Exception as exc: # An unhandled error here becomes a dropped connection. Cloudflare then # answers the workflow with plaintext "error code: 502" and the real # failure never appears in the Actions log. logger.exception( "w4p_activation_unhandled", project_ref=body.project_ref, error_type=type(exc).__name__, ) raise HTTPException( status_code=500, detail={"code": "activation_internal_error"}, ) from exc