"""Task: Provision a provider-backed preview environment for a specific revision. Flow: 1. Load project, revision, preview row 2. Poll CF Pages for a live deployment of revision.git_sha on the preview branch 3. Set up / update W4P routing for the preview hostname 4. Persist preview status + routes manifest No local dev-server is started. CF Pages handles building and hosting automatically once the preview branch is pushed to GitHub. Retries: if CF Pages hasn't finished building yet AND the task was not triggered via a workflow_run webhook, it raises ``Retry`` so arq re-queues with a 30-second delay. When webhook_triggered=True the retry loop is skipped entirely. """ from __future__ import annotations import structlog from src.config import get_settings logger = structlog.get_logger() _POLL_DEFER_SECONDS = 30 async def _dispatch_w4p_preview_workflow(*, db_factory, workspace_id, repo_full_name: str, branch: str) -> bool: """Dispatch the preview W4P workflow even when the branch is already current. A backfill can find an existing preview branch and an unchanged workflow, so there may be no GitHub push event to start the upload. The preview task is the authoritative place to nudge that workflow. """ 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=repo_full_name, workflow_file="deploy.yml", ref=branch, ) except Exception as exc: # pragma: no cover - provider failures are retried by the job logger.warning( "preview_w4p_workflow_dispatch_failed", repo=repo_full_name, branch=branch, error=str(exc), ) return False async def provision_preview_for_revision( ctx: dict, preview_id: str, project_id: str, revision_id: str, *, webhook_triggered: bool = False, ) -> dict: from uuid import UUID from arq.worker import Retry from src.core.database import async_session_factory from src.domain.deployments.repository import DeploymentTargetRepository from src.domain.deployments.repository import DeployRepository from src.domain.hosting.deployment_targets import ( decide_target_kind, pages_preview_branch_url, pages_production_url, ) 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.factory import get_deployment_provider, get_routing_provider from src.domain.hosting.hostnames import canonical_live_url, canonical_preview_url from src.domain.hosting.types import DeployJobContext, RoutingDescriptor from src.domain.previews.repository import PreviewRepository, RoutesManifestRepository from src.domain.projects.repository import ProjectRepository from src.domain.revisions.repository import RevisionRepository db_factory = ctx.get("db_factory", async_session_factory) settings = get_settings() pid = UUID(project_id) rid = UUID(revision_id) async with db_factory() as db: project_repo = ProjectRepository(db) revision_repo = RevisionRepository(db) preview_repo = PreviewRepository(db) project = await project_repo.get_with_repo(pid) if not project or not hasattr(project, "repo") or not project.repo: raise RuntimeError("Project or repo not found") revision = await revision_repo.get_by_id(rid) if not revision or revision.project_id != pid: raise RuntimeError("Revision not found") preview = await preview_repo.get_by_id(UUID(preview_id)) if not preview or preview.project_id != pid: raise RuntimeError("Preview not found") project_type = project.type project_ref = (project.project_ref or "").strip() project_created_at = project.created_at workspace_id = project.workspace_id tenant_ref = "" preview_branch = getattr(project.repo, "preview_branch", None) or settings.hosting_preview_branch or "preview" repo_full_name = project.repo.repo_full_name default_branch = project.repo.default_branch or "main" deploy_history = await DeployRepository(db).list_for_project(pid) has_successful_pages_deployment = any( d.status == "live" and (d.hosting_provider or "").strip().lower() in {"cloudflare_pages", "cloudflare"} for d in deploy_history ) from src.domain.workspaces.repository import WorkspaceRepository async with db_factory() as db: ws = await WorkspaceRepository(db).get_by_id(workspace_id) tenant_ref = (ws.tenant_ref or "").strip() if ws else "" 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, ) needs_w4p_workflow_repair = current_hosting_provider not in { W4P_STATIC, "cloudflare_w4p", "w4p", } # W4P has no CF Pages deployment to poll. The generated W4P workflow uploads # the asset version and calls Core's activation endpoint, which is the # authoritative completion signal and also finalizes the dashboard preview. # Keeping this worker on the legacy Pages polling path causes an endless retry # loop and can attempt to create a Pages project on the wrong account. if target_kind == "cloudflare_pages" and uses_worker_ci(selected_provider): async with db_factory() as db: migrated = await ProjectRepository(db).get_by_id(pid) if migrated and ( (getattr(migrated, "primary_hosting_provider", None) or "").strip().lower() not in {W4P_STATIC, "cloudflare_w4p", "w4p"} ): migrated.primary_hosting_provider = W4P_STATIC await db.commit() if needs_w4p_workflow_repair: from src.workers.enqueue import enqueue_fix_deploy_workflows await enqueue_fix_deploy_workflows(project_id) # A completed workflow_run already built this SHA. Dispatch only when # nothing else started CI (manual backfill / missing webhook). if webhook_triggered: logger.info( "preview_w4p_workflow_already_observed", preview_id=preview_id, sha=revision.git_sha[:12], ) return { "status": "waiting_for_w4p_activation", "preview_id": preview_id, "project_ref": project_ref, "workflow_dispatch": False, } dispatched = await _dispatch_w4p_preview_workflow( db_factory=db_factory, workspace_id=workspace_id, repo_full_name=repo_full_name, branch=preview_branch, ) logger.info( "preview_waiting_for_w4p_activation", preview_id=preview_id, project_ref=project_ref, sha=revision.git_sha[:12], workflow_dispatch=dispatched, ) if not dispatched: raise Retry(defer=_POLL_DEFER_SECONDS) return { "status": "waiting_for_w4p_activation", "preview_id": preview_id, "project_ref": project_ref, } # External origin: just mark preview as "running" with no URL if target_kind == "external_origin": async with db_factory() as db: await PreviewRepository(db).update_status(UUID(preview_id), "running") await db.commit() return {"status": "running", "note": "external_origin_no_preview"} apex = settings.hosting_public_apex or "sites.p2hs.com" from urllib.parse import urlparse preview_canonical = canonical_preview_url(project_ref, apex) preview_hostname = urlparse(preview_canonical).netloc or "" cf_pages_preview_url = pages_preview_branch_url(project_ref, preview_branch) # Build a minimal DeployJobContext for the providers from uuid import uuid4 job_ctx = DeployJobContext( deploy_id=uuid4(), # synthetic — no deploy row for preview polling project_id=pid, workspace_id=workspace_id, revision_id=rid, git_sha=revision.git_sha, environment="preview", tenant_ref=tenant_ref, project_ref=project_ref, target_hostname=preview_hostname, repo_full_name=repo_full_name, default_branch=default_branch, framework=project_type, ) # --- Ensure CF Pages project exists ------------------------------------- # wrangler pages deploy creates the project automatically on first push, but # it can take a few seconds to become visible in the API. Calling # ensure_project_environment here guarantees the project exists before we # start polling for the deployment, which avoids the task retrying forever # against a 404 on the projects endpoint. deployment_provider = get_deployment_provider(settings) env_out = await deployment_provider.ensure_project_environment(job_ctx) if not env_out.ok: if is_pages_project_limit_error(env_out.message) and destination_deploy_allowed( hosting_migration ): async with db_factory() as db: migrated = await ProjectRepository(db).get_by_id(pid) if migrated: migrated.primary_hosting_provider = W4P_STATIC await db.commit() from src.workers.enqueue import enqueue_fix_deploy_workflows await enqueue_fix_deploy_workflows(project_id) logger.warning( "preview_pages_limit_promoting_to_w4p", preview_id=preview_id, project_id=project_id, project_ref=project_ref, error=env_out.message, ) return { "status": "waiting_for_w4p_activation", "preview_id": preview_id, "project_ref": project_ref, } logger.warning( "preview_cf_project_ensure_failed", preview_id=preview_id, project_ref=project_ref, error=env_out.message, ) # Non-fatal: continue and let the deployment-status poll decide # --- Check CF Pages deployment status ------------------------------------ # When webhook_triggered=True the workflow_run webhook already confirmed the # deployment completed, so we skip the retry loop entirely. status_out = await deployment_provider.get_deployment_status(job_ctx) if not status_out.ok: if not webhook_triggered and status_out.activation_status in ("not_found", "building"): logger.info( "preview_waiting_for_cf_pages", preview_id=preview_id, sha=revision.git_sha[:12], branch=preview_branch, ) raise Retry(defer=_POLL_DEFER_SECONDS) if not webhook_triggered: # Non-retriable failure async with db_factory() as db: from src.domain.builder.repository import TimelineEventRepository preview_repo = PreviewRepository(db) await preview_repo.update_status(UUID(preview_id), "failed") await TimelineEventRepository(db).create( pid, "preview_refresh_failed", revision_id=rid, preview_id=UUID(preview_id), phase="preview", title="Preview unavailable", detail_text="The preview could not be built. Please try submitting your prompt again.", ) await db.commit() raise RuntimeError(f"CF Pages deployment failed: {status_out.message}") # webhook_triggered but CF Pages API not ready yet — log and proceed anyway. # wrangler pages deploy is synchronous so the deployment exists; the API may # just need a moment to index it. logger.warning( "preview_cf_pages_not_indexed_yet", preview_id=preview_id, sha=revision.git_sha[:12], status=status_out.activation_status, ) cf_deployment_id = status_out.provider_resource_id if status_out.ok else None # --- Routing plane: W4P proxy Worker + KV -------------------------------- routing_provider = get_routing_provider(settings) routing_out = await routing_provider.activate_route( job_ctx, target_url=cf_pages_preview_url, deployment_target_id=None, active_deployment_id=cf_deployment_id, ) if not routing_out.ok: logger.error("preview_routing_failed", preview_id=preview_id, error=routing_out.message) # Keep the provider deployment URL internal even when routing fails. # The canonical hostname is the only preview URL exposed to clients. base_url = preview_canonical else: base_url = preview_canonical # --- Enumerate routes from repo filesystem (no dev-server needed) -------- try: from src.domain.previews.adapters import get_adapter from src.workers.repo_ops import get_project_repo_path repo_path = get_project_repo_path(settings.repo_workspace, workspace_id, pid) adapter = get_adapter(project_type) routes = await adapter.enumerate_routes(repo_path) except Exception: routes = {"routes": [{"path": "/", "label": "Home", "type": "page"}], "source": "default"} # --- Persist preview state ----------------------------------------------- async with db_factory() as db: from src.domain.builder.repository import TimelineEventRepository preview_repo = PreviewRepository(db) manifest_repo = RoutesManifestRepository(db) project_repo = ProjectRepository(db) target_repo = DeploymentTargetRepository(db) timeline_repo = TimelineEventRepository(db) await preview_repo.update_status( UUID(preview_id), "running", base_url=base_url, provider_preview_url=cf_pages_preview_url, ) if not await manifest_repo.get_latest(pid, rid): await manifest_repo.create( project_id=pid, manifest=routes, source=routes.get("source", "default"), revision_id=rid, ) await target_repo.upsert( pid, "preview", target_kind, provider_preview_url=cf_pages_preview_url, provider_production_url=pages_production_url(project_ref), latest_commit_sha=revision.git_sha, latest_revision_id=rid, repo_linkage_status="linked", domain_status="active", ) proj = await project_repo.get_by_id(pid) if proj and (proj.active_revision_id is None or proj.active_revision_id == rid): await project_repo.update(pid, active_preview_id=UUID(preview_id)) await project_repo.update_hostnames(pid, preview_hostname=preview_hostname) await timeline_repo.create( pid, "preview_refresh_succeeded", revision_id=rid, preview_id=UUID(preview_id), phase="preview", title="Preview is live", detail_text=base_url, ) await db.commit() logger.info( "preview_provisioned_for_revision", preview_id=preview_id, url=base_url, cf_pages_url=cf_pages_preview_url, ) # Publish real-time WebSocket event so the frontend can reload the preview iframe 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, project_id, "preview.updated", { "preview_id": preview_id, "revision_id": revision_id, "url": base_url, }) finally: await _pubsub_redis.aclose() except Exception as _ws_err: logger.warning("preview_ws_publish_failed", error=str(_ws_err)) return {"status": "running", "base_url": base_url, "provider_preview_url": cf_pages_preview_url}