"""Task: deploy (or re-deploy) a template demo to CF Pages + W4P KV routing. Flow (per environment — production = main branch, development = dev branch): 1. Load template record from DB. 2. Ensure the CF Pages project exists (sourceless; GH Actions handles the actual build upload). 3. Push a GitHub Actions deploy workflow to the template repo if not present (so CF Pages receives wrangler direct-upload on every push to main / development). 4. Poll CF Pages for a live deployment on the relevant branch. 5. Upload a proxy Worker to the W4P dispatch namespace and write the KV routing record so {subdomain}.{apex} → proxy Worker → CF Pages URL. 6. Persist the final state (deploy_status, demo_url / dev_url) on the template record. Retries: if CF Pages hasn't built yet the task raises ``Retry`` so arq re-queues it with a 30-second delay (up to ``max_tries`` in WorkerSettings). CI/CD: the GitHub webhook handler triggers this task whenever ``main`` or ``development`` is pushed to the template repo. Registration also enqueues an initial run. """ from __future__ import annotations from typing import Literal from uuid import UUID, uuid4 import structlog logger = structlog.get_logger() _POLL_DEFER_SECONDS = 30 _NULL_UUID = UUID(int=0) async def run_template_deploy( ctx: dict, template_id: str, environment: str = "production", ) -> dict: from arq.worker import Retry from src.config import get_settings from src.core.database import async_session_factory from src.domain.hosting.adapters.cloudflare_pages import CloudflarePagesProvider from src.domain.hosting.deployment_targets import ( pages_production_url, pages_preview_branch_url, pages_project_name, ) from src.domain.hosting.routing_plane import CloudflareRoutingProvider from src.domain.hosting.types import DeployJobContext, RoutingDescriptor from src.domain.templates.repository import TemplateRepository from src.domain.templates.models import Template db_factory = ctx.get("db_factory", async_session_factory) settings = get_settings() tid = UUID(template_id) env = (environment or "production").lower() if env not in ("production", "development"): logger.error("template_deploy_invalid_env", template_id=template_id, env=env) return {"ok": False, "reason": "invalid_environment"} # ── Load template ────────────────────────────────────────────────────────── async with db_factory() as db: repo = TemplateRepository(db) template = await repo.get_by_id(tid) if not template: logger.error("template_deploy_not_found", template_id=template_id) return {"ok": False, "reason": "template_not_found"} branch = template.main_branch if env == "production" else template.dev_branch subdomain = template.demo_subdomain if env == "production" else template.dev_subdomain ref = template.ref framework = template.framework github_repo = template.github_repo installation_id = ( template.github_installation_id or settings.github_template_installation_id or settings.github_installation_id ) await repo.update(tid, deploy_status="deploying") await db.commit() # ── CF Pages: ensure project exists ─────────────────────────────────────── pages_provider = CloudflarePagesProvider(settings) cf_project_name = pages_project_name(ref) try: await pages_provider.ensure_pages_project(project_ref=ref, project_type=framework) except Exception as exc: async with db_factory() as db: await TemplateRepository(db).update(tid, deploy_status="failed") await db.commit() logger.error("template_deploy_cf_pages_ensure_failed", template_id=template_id, error=str(exc)) return {"ok": False, "reason": "cf_pages_ensure_failed", "detail": str(exc)} # ── Ensure starter branch exists (created from main if absent) ─────────── if env == "production": await _ensure_starter_branch( github_repo=github_repo, installation_id=installation_id, main_branch=template.main_branch, starter_branch=template.starter_branch, ) # ── Ensure CF secrets are set on the template repo (idempotent) ────────── await _ensure_repo_cf_secrets( github_repo=github_repo, installation_id=installation_id, settings=settings, db_factory=db_factory, ) # ── Push deploy workflow to template repo (idempotent) ──────────────────── await _ensure_deploy_workflow( github_repo=github_repo, installation_id=installation_id, ref=ref, framework=framework, main_branch=branch if env == "production" else template.main_branch, dev_branch=template.dev_branch, settings=settings, ) # ── Poll CF Pages for live deployment on this branch ────────────────────── try: deployment = await pages_provider.find_latest_live_deployment_for_branch( project_ref=ref, branch=branch, ) except Exception as exc: logger.warning("template_deploy_cf_poll_error", template_id=template_id, error=str(exc)) deployment = None if deployment is None: job_try = ctx.get("job_try", 1) # On the first attempt (and every 10th retry), trigger a workflow_dispatch so the # GitHub Actions build runs even if the file was already present but never executed # (e.g. secrets were missing, project was recreated, or no push happened since the # workflow file was committed). if job_try == 1 or job_try % 10 == 0: await _dispatch_deploy_workflow( github_repo=github_repo, installation_id=installation_id, branch=branch, ) logger.info( "deploy_waiting_for_cf_pages", template_id=template_id, branch=branch, ) async with db_factory() as db: await TemplateRepository(db).update(tid, deploy_status="deploying") await db.commit() raise Retry(defer=_POLL_DEFER_SECONDS) # ── Routing plane: proxy Worker + KV ────────────────────────────────────── apex = (settings.hosting_public_apex or "sites.p2hs.com").strip() # Reuse the W4P routing provider. For the production demo we use environment="production" # (proxy script name: p2hs-prod-{ref}); for the dev demo we use environment="preview" # (proxy script name: p2hs-prev-{ref}). route_env = "production" if env == "production" else "preview" if env == "production": cf_pages_url = pages_production_url(ref) else: cf_pages_url = pages_preview_branch_url(ref, branch) hostname = f"{subdomain}.{apex}" descriptor = RoutingDescriptor( target_kind="cloudflare_pages", project_ref=ref, project_id=str(tid), environment=route_env, target_url=cf_pages_url, ) routing_provider = CloudflareRoutingProvider(settings) try: await routing_provider.upsert_route(hostname, descriptor, route_env) except Exception as exc: async with db_factory() as db: await TemplateRepository(db).update(tid, deploy_status="failed") await db.commit() logger.error("template_deploy_routing_failed", template_id=template_id, error=str(exc)) return {"ok": False, "reason": "routing_failed", "detail": str(exc)} demo_url = f"https://{hostname}/" # ── Persist outcome ──────────────────────────────────────────────────────── async with db_factory() as db: kwargs: dict = {"deploy_status": "live", "cf_pages_project_name": cf_project_name} if env == "production": kwargs["demo_url"] = demo_url else: kwargs["dev_url"] = demo_url await TemplateRepository(db).update(tid, **kwargs) await db.commit() logger.info( "template_deploy_complete", template_id=template_id, environment=env, hostname=hostname, cf_pages_url=cf_pages_url, ) return {"ok": True, "template_id": template_id, "environment": env, "url": demo_url} async def _ensure_deploy_workflow( *, github_repo: str, installation_id: int, ref: str, framework: str, main_branch: str, dev_branch: str, settings, ) -> None: """Push a GitHub Actions deploy workflow to the template repo if it isn't there yet. The workflow triggers on pushes to ``main_branch`` and ``dev_branch`` and uploads the built output to the CF Pages project ``p2hs-{ref}`` via wrangler direct-upload. """ from src.integrations.github_client import GitHubAppClient from src.domain.templates.service import build_template_deploy_workflow workflow_path = ".github/workflows/deploy.yml" workflow_content = build_template_deploy_workflow( ref=ref, framework=framework, main_branch=main_branch, dev_branch=dev_branch, ) try: github = GitHubAppClient(installation_id=installation_id) # Check if the workflow already exists on the main branch and has the expected content. needs_push = True try: existing_bytes = await github.get_file_contents(github_repo, workflow_path, "main") existing_content = existing_bytes.decode("utf-8") if existing_bytes else "" # Re-push only if the content changed (e.g. workflow_dispatch trigger missing). needs_push = existing_content.strip() != workflow_content.strip() except Exception: pass # File not found or API error — proceed to create. if not needs_push: return await github.push_files( repo=github_repo, branch="main", files={workflow_path: workflow_content}, message="chore: update P2HS template deploy workflow", ) logger.info("template_deploy_workflow_pushed", repo=github_repo, ref=ref) except Exception as exc: # Non-fatal: the admin can push the workflow manually. logger.warning( "template_deploy_workflow_push_failed", repo=github_repo, ref=ref, error=str(exc), ) async def _ensure_starter_branch( *, github_repo: str, installation_id: int, main_branch: str, starter_branch: str, ) -> None: """Create the starter branch from main if it doesn't already exist.""" import httpx from src.integrations.github_client import GitHubAppClient, GITHUB_API try: github = GitHubAppClient(installation_id=installation_id) # Check if starter branch already exists. sha = await github.get_ref_head_sha(github_repo, starter_branch) if sha: return # Already exists. # Create branch from the tip of main. main_sha = await github.get_ref_head_sha(github_repo, main_branch) if not main_sha: logger.warning( "template_starter_branch_create_failed", repo=github_repo, reason="main_branch_sha_not_found", main_branch=main_branch, ) return headers = await github._headers() async with httpx.AsyncClient() as client: resp = await client.post( f"{GITHUB_API}/repos/{github_repo}/git/refs", headers=headers, json={"ref": f"refs/heads/{starter_branch}", "sha": main_sha}, ) if resp.status_code in (200, 201): logger.info( "template_starter_branch_created", repo=github_repo, branch=starter_branch, sha=main_sha[:12], ) elif resp.status_code == 422: pass # Already exists (race condition) — fine. else: logger.warning( "template_starter_branch_create_failed", repo=github_repo, status=resp.status_code, body=resp.text[:200], ) except Exception as exc: # Non-fatal: admin can create the branch manually. logger.warning( "template_starter_branch_ensure_failed", repo=github_repo, error=str(exc), ) async def _ensure_repo_cf_secrets( *, github_repo: str, installation_id: int, settings, db_factory=None, ) -> None: """Set CLOUDFLARE_API_TOKEN and CLOUDFLARE_ACCOUNT_ID on the template repo. Prefers primary CF account credentials from DB; falls back to settings env vars. Idempotent — safe to call on every task run. """ from src.integrations.github_client import GitHubAppClient cf_token = (settings.cloudflare_api_token or "").strip() cf_account = (settings.cloudflare_account_id or "").strip() if db_factory is not None: from src.domain.cloud_infra.repository import ( CFAccountRepository, CloudInstanceRepository, WorkspaceCloudAssignmentRepository, ) from src.domain.cloud_infra.service import CloudInfraService from src.core.exceptions import NotFoundError async with db_factory() as db: try: svc = CloudInfraService( cf_account_repo=CFAccountRepository(db), instance_repo=CloudInstanceRepository(db), assignment_repo=WorkspaceCloudAssignmentRepository(db), ) cf_creds = await svc.get_primary_credentials() cf_token = cf_creds.api_token cf_account = cf_creds.account_id except (NotFoundError, ValueError): pass # fall back to settings env vars if not cf_token or not cf_account: logger.warning( "template_deploy_cf_secrets_missing", repo=github_repo, ) return try: github = GitHubAppClient(installation_id=installation_id) await github.set_repo_action_secrets( github_repo, {"CLOUDFLARE_API_TOKEN": cf_token, "CLOUDFLARE_ACCOUNT_ID": cf_account}, ) except Exception as exc: logger.warning( "template_deploy_cf_secrets_set_failed", repo=github_repo, error=str(exc), ) async def _dispatch_deploy_workflow( *, github_repo: str, installation_id: int, branch: str, ) -> None: """Trigger a workflow_dispatch event on deploy.yml so CF Pages gets a fresh build.""" from src.integrations.github_client import GitHubAppClient try: github = GitHubAppClient(installation_id=installation_id) await github.trigger_workflow_dispatch( repo_full_name=github_repo, workflow_file="deploy.yml", ref=branch, ) except Exception as exc: logger.warning( "template_deploy_workflow_dispatch_failed", repo=github_repo, branch=branch, error=str(exc), )