"""Task: run the two-plane hosting pipeline for a project deploy row. Flow: 1. Ensure CF Pages project exists (deployment plane) 2. Poll CF Pages for a live deployment at revision git_sha on the correct branch 3. Set up / update W4P proxy Worker + KV routing record (routing plane) 4. Persist deployment outcome Retries: if CF Pages hasn't finished building yet, the task raises ``Retry`` so arq re-queues it with a 30-second delay (up to ``max_tries`` in WorkerSettings). """ from __future__ import annotations from dataclasses import replace from typing import Literal, cast from urllib.parse import urlparse from uuid import UUID import structlog from src.config import get_settings from src.domain.hosting.deployment_targets import ( decide_target_kind, pages_preview_branch_url, pages_production_url, pages_project_name, ) from src.domain.hosting.ci_activation import WAITING_FOR_CI_STEP from src.domain.hosting.factory import get_deployment_provider, get_routing_provider, get_w4p_static_provider from src.domain.hosting.execution import uses_worker_ci from src.domain.hosting.migration_state import ( destination_deploy_allowed, resolve_execution_provider, ) from src.domain.hosting.project_provider import W4P_STATIC, is_pages_project_limit_error from src.domain.hosting.hostnames import canonical_live_url, canonical_preview_url from src.domain.hosting.types import DeployJobContext logger = structlog.get_logger() _DEPLOY_POLL_DEFER_SECONDS = 30 async def _dispatch_deploy_workflow( *, db_factory, workspace_id: UUID, github_repo: str, branch: str, ) -> bool: """Nudge a missing Pages build without creating a second deployment row. A project deploy normally starts from the repository push. If GitHub's workflow event was dropped, polling CF Pages forever leaves the customer seeing ``Deploying`` for hours. Dispatch is idempotent at the workflow level and is only attempted by the first/periodic poll in the caller. """ from src.integrations.github_client import GitHubAppClient from src.integrations.github_installation import resolve_installation_id_for_workspace try: installation_id = await resolve_installation_id_for_workspace(db_factory, workspace_id) github = GitHubAppClient(installation_id=installation_id) return await github.trigger_workflow_dispatch( repo_full_name=github_repo, workflow_file="deploy.yml", ref=branch, ) except Exception as exc: # pragma: no cover - provider failures are retried by the job logger.warning( "project_deploy_workflow_dispatch_failed", repo=github_repo, branch=branch, error=str(exc), ) return False def _normalize_env(raw: str) -> Literal["preview", "production", "staging"]: e = (raw or "").strip().lower() if e in ("preview", "production", "staging"): return cast(Literal["preview", "production", "staging"], e) return "production" def _hostname_for(env: str, project_ref: str, apex: str) -> str: if env == "preview": url = canonical_preview_url(project_ref, apex) else: url = canonical_live_url(project_ref, apex) return urlparse(url).netloc or "" def _cf_pages_url_for(env: str, project_ref: str, preview_branch: str) -> str: if env == "preview": return pages_preview_branch_url(project_ref, preview_branch) return pages_production_url(project_ref) async def run_project_deploy(ctx: dict, deploy_id: str) -> dict: from arq.worker import Retry from src.core.database import async_session_factory from src.domain.deployments.repository import DeploymentTargetRepository, DeployRepository from src.domain.projects.repository import ProjectRepository from src.domain.revisions.repository import RevisionRepository from src.domain.workspaces.repository import WorkspaceRepository db_factory = ctx.get("db_factory", async_session_factory) settings = get_settings() deploy_settings = settings did = UUID(deploy_id) async with db_factory() as db: deploy_repo = DeployRepository(db) project_repo = ProjectRepository(db) revision_repo = RevisionRepository(db) workspace_repo = WorkspaceRepository(db) deploy = await deploy_repo.get_by_id(did) if not deploy: logger.error("project_deploy_missing", deploy_id=deploy_id) return {"ok": False, "reason": "deploy_not_found"} project = await project_repo.get_with_repo(deploy.project_id) if not project or not getattr(project, "repo", None): await deploy_repo.apply_hosting_outcome(did, status="failed", metadata_patch={"error": "project_or_repo_missing"}) await db.commit() return {"ok": False, "reason": "project_or_repo_missing"} project_deploy_history = await deploy_repo.list_for_project(project.id) has_successful_pages_deployment = any( d.status == "live" and (d.hosting_provider or "").strip().lower() in {"cloudflare_pages", "cloudflare"} for d in project_deploy_history ) if not deploy.revision_id: await deploy_repo.apply_hosting_outcome(did, status="failed", metadata_patch={"error": "no_revision"}) await db.commit() return {"ok": False, "reason": "no_revision"} revision = await revision_repo.get_by_id(deploy.revision_id) if not revision or revision.project_id != project.id: await deploy_repo.apply_hosting_outcome(did, status="failed", metadata_patch={"error": "revision_not_found"}) await db.commit() return {"ok": False, "reason": "revision_not_found"} workspace = await workspace_repo.get_by_id(project.workspace_id) tenant_ref = (workspace.tenant_ref or "").strip() if workspace else "" from src.domain.hosting.deploy_target import DeployAccountError, settings_for_project from src.domain.cloud_infra.repository import CFAccountRepository, CloudInstanceRepository, WorkspaceCloudAssignmentRepository from src.domain.cloud_infra.service import CloudInfraService async def _resolve_workspace_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: deploy_settings = await settings_for_project(project.workspace_id, project.created_at, settings, _resolve_workspace_credentials) except DeployAccountError as exc: await deploy_repo.apply_hosting_outcome(did, status="failed", metadata_patch={"error": str(exc)}) await db.commit() return {"ok": False, "reason": str(exc)} project_ref = (project.project_ref or "").strip() env_norm = _normalize_deploy_env(deploy.environment) apex = deploy_settings.hosting_public_apex or "sites.p2hs.com" preview_branch = getattr(project.repo, "preview_branch", None) or settings.hosting_preview_branch or "preview" target_kind = decide_target_kind(project.type) current_hosting_provider = (getattr(project, "primary_hosting_provider", None) or "").strip().lower() hosting_migration = getattr(project, "hosting_migration", None) selected_provider = resolve_execution_provider( project.created_at, settings, primary_hosting_provider=getattr(project, "primary_hosting_provider", None), hosting_migration=hosting_migration, assume_unassigned_w4p=True, has_successful_pages_deployment=has_successful_pages_deployment, ) is_w4p = target_kind == "cloudflare_pages" and uses_worker_ci(selected_provider) job_ctx = DeployJobContext( deploy_id=did, project_id=project.id, workspace_id=project.workspace_id, revision_id=revision.id, git_sha=revision.git_sha, environment=env_norm, tenant_ref=tenant_ref, project_ref=project_ref, target_hostname=_hostname_for(env_norm, project_ref, apex), repo_full_name=project.repo.repo_full_name, default_branch=project.repo.default_branch or "main", framework=project.type, ) await db.commit() needs_w4p_provider_persist = is_w4p and current_hosting_provider not in { W4P_STATIC, "cloudflare_w4p", "w4p", } repair_w4p_workflow = ( needs_w4p_provider_persist and not (deploy.deployment_metadata or {}).get("w4p_workflow_repair_requested") ) if needs_w4p_provider_persist: project.primary_hosting_provider = W4P_STATIC deploy.deployment_metadata = { **(deploy.deployment_metadata or {}), "w4p_workflow_repair_requested": True, } await db.commit() if repair_w4p_workflow: from src.workers.enqueue import enqueue_fix_deploy_workflows await enqueue_fix_deploy_workflows(str(job_ctx.project_id)) logger.info("project_deploy_waiting_for_w4p_workflow_repair", deploy_id=deploy_id) raise Retry(defer=_DEPLOY_POLL_DEFER_SECONDS) # External origin: record metadata, no real deployment if target_kind == "external_origin": async with db_factory() as db: deploy_repo = DeployRepository(db) target_repo = DeploymentTargetRepository(db) await target_repo.upsert( job_ctx.project_id, env_norm, target_kind, repo_linkage_status="external_manual_required", ) await deploy_repo.apply_hosting_outcome( did, status="live", last_completed_step="external_origin_noop", metadata_patch={"target_kind": target_kind, "message": "External origin — deploy managed outside P2HS"}, ) await db.commit() return {"ok": True, "deploy_id": deploy_id, "target_kind": target_kind} # W4P projects have no Pages project to ensure. if not is_w4p: # --- Deployment plane: ensure CF Pages project -------------------------------- deployment_provider = get_deployment_provider(deploy_settings) ensure_out = await deployment_provider.ensure_project_environment(job_ctx) if not ensure_out.ok: if is_pages_project_limit_error(ensure_out.message) and destination_deploy_allowed( hosting_migration ): async with db_factory() as db: from src.domain.projects.repository import ProjectRepository migrated = await ProjectRepository(db).get_by_id(job_ctx.project_id) if migrated: migrated.primary_hosting_provider = W4P_STATIC await db.commit() logger.warning( "project_pages_limit_promoting_to_w4p", deploy_id=deploy_id, project_id=str(job_ctx.project_id), project_ref=job_ctx.project_ref, error=ensure_out.message, ) is_w4p = True deployment_provider = get_w4p_static_provider(deploy_settings, environment=env_norm) from src.workers.enqueue import enqueue_fix_deploy_workflows await enqueue_fix_deploy_workflows(str(job_ctx.project_id)) raise Retry(defer=_DEPLOY_POLL_DEFER_SECONDS) else: async with db_factory() as db: await DeployRepository(db).apply_hosting_outcome( did, status="failed", last_completed_step="ensure_project_environment", metadata_patch={**(ensure_out.metadata or {}), "message": ensure_out.message}, ) await db.commit() return {"ok": False, "reason": "ensure_failed"} if not is_w4p: pages_name = (ensure_out.metadata or {}).get("pages_project_name") or pages_project_name(project_ref) linkage = (ensure_out.metadata or {}).get("repo_linkage_status", "unknown") # --- For production: merge preview → main so one push starts deploy.yml ------ deploy_branch = preview_branch if env_norm == "preview" else job_ctx.default_branch from src.domain.hosting.deploy_trigger import ( ci_already_started, should_dispatch_workflow, should_promote_production, ) deploy_metadata = dict(deploy.deployment_metadata or {}) ci_just_pushed = False if env_norm == "production" and should_promote_production(deploy_metadata): from src.integrations.github_client import GitHubAppClient from src.integrations.github_installation import resolve_installation_id_for_workspace from src.workers.repo_ops import get_project_repo_path, promote_preview_to_production installation_id = await resolve_installation_id_for_workspace(db_factory, job_ctx.workspace_id) github = GitHubAppClient(installation_id=installation_id) clone_url = await github.get_authenticated_clone_url(job_ctx.repo_full_name) repo_path = get_project_repo_path(settings.repo_workspace, job_ctx.workspace_id, job_ctx.project_id) promotion_source_ref = (deploy.deployment_metadata or {}).get("rollback_source_ref") try: promoted_sha, previous_production_sha = await promote_preview_to_production( repo_path=repo_path, clone_url=clone_url, preview_branch=preview_branch, default_branch=job_ctx.default_branch, deploy_id=deploy_id, source_ref=promotion_source_ref, git_name=settings.git_committer_name, git_email=settings.git_committer_email, ) ci_just_pushed = True logger.info("deploy_preview_promoted", deploy_id=deploy_id, sha=promoted_sha[:12] if promoted_sha else "") except RuntimeError as merge_err: async with db_factory() as db: await DeployRepository(db).apply_hosting_outcome( did, status="failed", last_completed_step="promote_preview_to_main", metadata_patch={"error": str(merge_err)}, ) await db.commit() logger.error("deploy_promote_failed", deploy_id=deploy_id, error=str(merge_err)) return {"ok": False, "reason": "merge_failed", "detail": str(merge_err)} async with db_factory() as db: await DeployRepository(db).apply_hosting_outcome( did, status="deploying", last_completed_step="preview_promoted_to_main", metadata_patch={ "promoted_sha": promoted_sha, "previous_production_sha": previous_production_sha, "production_backup_ref": f"refs/p2hs/backups/{deploy_id}", }, ) await db.commit() if promoted_sha.strip(): job_ctx = replace(job_ctx, git_sha=promoted_sha.strip()) logger.info( "deploy_git_sha_synced_to_main_tip", deploy_id=deploy_id, git_sha=job_ctx.git_sha[:12], ) # --- W4P: the run's workflow builds, uploads the version and calls /api/hosting/w4p/activate. # That call moves the route and marks the deployment live, so there is nothing to poll here. if is_w4p: # The push that updated preview or main already starts deploy.yml. # Dispatch only on a later attempt when that workflow was never observed. job_try = int(ctx.get("job_try", 1) or 1) if should_dispatch_workflow( deploy_metadata, job_try=job_try, ci_just_pushed=ci_just_pushed ): dispatched = await _dispatch_deploy_workflow( db_factory=db_factory, workspace_id=job_ctx.workspace_id, github_repo=job_ctx.repo_full_name, branch=deploy_branch, ) if not dispatched: logger.warning( "project_deploy_w4p_workflow_dispatch_pending", deploy_id=deploy_id, branch=deploy_branch, ) raise Retry(defer=_DEPLOY_POLL_DEFER_SECONDS) elif not ci_already_started(deploy_metadata, ci_just_pushed=ci_just_pushed): raise Retry(defer=_DEPLOY_POLL_DEFER_SECONDS) async with db_factory() as db: await DeployRepository(db).apply_hosting_outcome( did, status="deploying", source_git_sha=job_ctx.git_sha, hosting_provider=W4P_STATIC, last_completed_step=WAITING_FOR_CI_STEP, metadata_patch={"target_kind": target_kind, "environment": env_norm}, ) await db.commit() logger.info( "project_deploy_waiting_for_push_triggered_ci", deploy_id=deploy_id, sha=job_ctx.git_sha[:12], environment=env_norm, branch=deploy_branch, workflow_dispatch=should_dispatch_workflow( deploy_metadata, job_try=job_try, ci_just_pushed=ci_just_pushed ), ) return {"ok": True, "deploy_id": deploy_id, "status": WAITING_FOR_CI_STEP} # --- Poll CF Pages for live deployment at this SHA ---------------------------- cf_pages_url = _cf_pages_url_for(env_norm, project_ref, preview_branch) status_out = await deployment_provider.get_deployment_status(job_ctx) if not status_out.ok and status_out.activation_status in ("not_found", "building"): # CF Pages hasn't finished building yet — retry after delay job_try = int(ctx.get("job_try", 1) or 1) if should_dispatch_workflow( deploy_metadata, job_try=job_try, ci_just_pushed=ci_just_pushed ) and job_try % 10 == 0: dispatched = await _dispatch_deploy_workflow( db_factory=db_factory, workspace_id=job_ctx.workspace_id, github_repo=job_ctx.repo_full_name, branch=deploy_branch, ) logger.info( "project_deploy_workflow_dispatch_attempted", deploy_id=deploy_id, branch=deploy_branch, job_try=job_try, accepted=dispatched, ) logger.info( "project_deploy_waiting_for_cf_pages", deploy_id=deploy_id, sha=job_ctx.git_sha[:12], branch=deploy_branch, ) async with db_factory() as db: await DeployRepository(db).apply_hosting_outcome( did, status="deploying", last_completed_step="waiting_for_cf_pages", metadata_patch={"cf_pages_status": status_out.activation_status}, ) await db.commit() raise Retry(defer=_DEPLOY_POLL_DEFER_SECONDS) if not status_out.ok: async with db_factory() as db: await DeployRepository(db).apply_hosting_outcome( did, status="failed", last_completed_step="cf_pages_deployment_failed", metadata_patch={**(status_out.metadata or {}), "message": status_out.message}, ) await db.commit() return {"ok": False, "reason": "cf_pages_deployment_failed"} cf_deployment_id = status_out.provider_resource_id # --- Routing plane: upsert W4P proxy Worker + KV ---------------------------- routing_provider = get_routing_provider(deploy_settings) routing_out = await routing_provider.activate_route( job_ctx, target_url=cf_pages_url, deployment_target_id=None, active_deployment_id=cf_deployment_id, ) if not routing_out.ok: async with db_factory() as db: await DeployRepository(db).apply_hosting_outcome( did, status="failed", last_completed_step="routing_activation_failed", metadata_patch={**(routing_out.metadata or {}), "message": routing_out.message}, ) await db.commit() return {"ok": False, "reason": "routing_failed"} # --- Wire verified custom domains for this project+environment --------------- # Any domain the user has verified gets an additional KV entry pointing to the # same proxy Worker script. No extra Worker upload needed — just a KV alias. try: from src.domain.deployments.repository import DomainRepository as _DomainRepo async with db_factory() as db: custom_domains = await _DomainRepo(db).list_verified_for_environment( job_ctx.project_id, env_norm ) custom_domain_names = [d.domain for d in custom_domains] if custom_domain_names: for custom_hostname in custom_domain_names: try: await routing_provider.alias_custom_hostname( custom_hostname=custom_hostname, project_ref=project_ref, environment=env_norm, target_url=cf_pages_url, active_deployment_id=cf_deployment_id, ) except Exception as ce: logger.warning( "custom_domain_alias_failed", deploy_id=deploy_id, custom_hostname=custom_hostname, error=str(ce), ) logger.info( "custom_domains_aliased", deploy_id=deploy_id, count=len(custom_domain_names), domains=custom_domain_names, ) except Exception as exc: # Non-fatal: canonical routing already succeeded logger.warning("custom_domains_alias_step_failed", deploy_id=deploy_id, error=str(exc)) # --- Persist final state ------------------------------------------------------ canonical_hostname = _hostname_for(env_norm, project_ref, apex) canonical_url = f"https://{canonical_hostname}/" async with db_factory() as db: deploy_repo = DeployRepository(db) target_repo = DeploymentTargetRepository(db) project_repo = ProjectRepository(db) target_row = await target_repo.upsert( job_ctx.project_id, env_norm, target_kind, provider_project_name=pages_name, provider_deployment_id=cf_deployment_id, provider_preview_url=pages_preview_branch_url(project_ref, preview_branch), provider_production_url=pages_production_url(project_ref), latest_commit_sha=job_ctx.git_sha, latest_revision_id=job_ctx.revision_id, repo_linkage_status=linkage, domain_status="active", ) await deploy_repo.apply_hosting_outcome( did, status="live", deploy_url=canonical_url, source_git_sha=job_ctx.git_sha, activation_status="live", last_completed_step="routing_activated", hosting_provider="cloudflare_pages", metadata_patch={ "target_kind": target_kind, "cf_pages_url": cf_pages_url, "canonical_url": canonical_url, "deployment_target_id": str(target_row.id), **(routing_out.metadata or {}), }, ) if env_norm == "preview": await project_repo.update_hostnames(job_ctx.project_id, preview_hostname=canonical_hostname) else: await project_repo.update_hostnames(job_ctx.project_id, live_hostname=canonical_hostname) await db.commit() logger.info( "project_deploy_complete", deploy_id=deploy_id, environment=env_norm, canonical_url=canonical_url, cf_pages_url=cf_pages_url, ) return {"ok": True, "deploy_id": deploy_id, "url": canonical_url} def _normalize_deploy_env(raw: str) -> Literal["preview", "production", "staging"]: return _normalize_env(raw)