"""W4P routing plane — KV-backed hostname routing with thin proxy Workers.
Responsibilities (routing plane only):
- Ensure the W4P dispatch namespace exists
- Upload a tiny proxy Worker for each project/environment into the dispatch namespace
(the Worker forwards requests to the CF Pages URL — it does NOT host app code)
- Write / update ``host:{hostname}`` KV entries pointing to the proxy Worker script name
- Remove routing entries on project teardown
The proxy Worker script is ~500 bytes; W4P script size limits are not a concern here.
"""
from __future__ import annotations
import json
import re
from typing import Any
from urllib.parse import quote
import httpx
import structlog
from src.config import Settings, get_settings
from src.domain.hosting.types import DeployJobContext, DeploymentOutcome, RoutingDescriptor
from src.domain.hosting.routing_contract import compile_routing_manifest
logger = structlog.get_logger()
# Matches :meth:`CloudflareRoutingProvider._proxy_script_name` (``p2hs-prod-…`` / ``p2hs-prev-…``).
_P2HS_PROXY_SCRIPT_RE = re.compile(r"^p2hs-(prod|prev)-(.+)$", re.IGNORECASE)
def parse_project_ref_from_p2hs_proxy_script_name(script_name: str, environment: str) -> str | None:
"""If ``script_name`` matches our managed proxy pattern, return the embedded ``project_ref``."""
m = _P2HS_PROXY_SCRIPT_RE.match((script_name or "").strip())
if not m:
return None
arm, ref = m.group(1).lower(), m.group(2).strip()
if not ref:
return None
env = (environment or "production").lower()
if env == "preview":
if arm != "prev":
return None
else:
if arm != "prod":
return None
return ref
def resolve_dispatch_route_project_ref(
*,
subdomain: str,
script_name: str,
environment: str,
explicit_project_ref: str | None,
) -> str:
"""Resolve ``project_ref`` for ``project:*`` KV keys (explicit > parsed script > tpl fallback)."""
if (explicit_project_ref or "").strip():
return explicit_project_ref.strip()
parsed = parse_project_ref_from_p2hs_proxy_script_name(script_name, environment)
if parsed:
return parsed
label = (subdomain or "").strip().lower()
return f"tpl-{label}"
class CloudflareAPIError(RuntimeError):
pass
class CloudflareRoutingProvider:
"""W4P control-plane: namespace + proxy Workers + KV routing records."""
def __init__(self, settings: Settings | None = None):
self._s = settings or get_settings()
# ------------------------------------------------------------------
# settings
# ------------------------------------------------------------------
@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()
@property
def _kv_namespace_id(self) -> str:
return str(self._s.cloudflare_dispatch_kv_namespace_id).strip()
# Resolved KV namespace UUID (cached per instance after first lookup)
_kv_id_resolved: str | None = None
async def _resolve_kv_namespace_id(self) -> str:
"""Return the KV namespace UUID, resolving from name if needed.
CLOUDFLARE_DISPATCH_KV_NAMESPACE_ID can be either the UUID (32-char hex)
or the namespace title/name — this method handles both. The resolved UUID
is cached on the instance so subsequent calls are free.
"""
if self._kv_id_resolved:
return self._kv_id_resolved
configured = self._kv_namespace_id
if not configured:
raise CloudflareAPIError("CLOUDFLARE_DISPATCH_KV_NAMESPACE_ID is not set")
# If it looks like a UUID (32 hex chars), use it directly
if re.fullmatch(r"[0-9a-fA-F]{32}", configured):
self._kv_id_resolved = configured
return configured
# Treat it as a namespace title and look up the actual UUID
logger.debug("kv_namespace_id_lookup_by_name", name=configured)
_, namespaces = await self._request("GET", "/storage/kv/namespaces", expected={200})
ns_list = namespaces if isinstance(namespaces, list) else []
for ns in ns_list:
if ns.get("title", "").lower() == configured.lower():
resolved = str(ns["id"])
self._kv_id_resolved = resolved
logger.debug("kv_namespace_id_resolved", name=configured, id=resolved)
return resolved
raise CloudflareAPIError(
f"KV namespace '{configured}' not found. "
"Set CLOUDFLARE_DISPATCH_KV_NAMESPACE_ID to the namespace UUID "
"(32-char hex) from the CF Dashboard → Workers & Pages → KV."
)
@property
def _compat_date(self) -> str:
import datetime as dt
v = (self._s.cloudflare_worker_compatibility_date or "").strip()
return v or dt.date.today().isoformat()
def _dispatch_namespace_name(self, environment: str) -> str:
env = (environment or "").lower()
umbrella = (self._s.cloudflare_w4p_namespace or "").strip()
if env == "preview":
raw = (self._s.cloudflare_w4p_namespace_preview or umbrella or "p2hs-preview").strip()
elif env == "staging":
raw = (self._s.cloudflare_w4p_namespace_staging or umbrella or "p2hs-staging").strip()
else:
raw = (self._s.cloudflare_w4p_namespace_production or umbrella or "p2hs-production").strip()
s = re.sub(r"[^A-Za-z0-9-]+", "-", raw)
s = re.sub(r"-{2,}", "-", s).strip("-")
return s or "p2hs-production"
def _base(self) -> str:
return f"https://api.cloudflare.com/client/v4/accounts/{self._account_id}"
def _default_bff_origin(self) -> str | None:
"""Return the public runtime origin for same-origin ``/api`` requests.
W4P project Workers serve static assets only. They must not point the
BFF leg back at ``target_url``: that creates a self-referential fetch
and Cloudflare 1101 responses for every application API request. An
explicit descriptor origin remains authoritative; this is only the
safe default used when callers omit it.
``cloud_public_api_url`` is preferred because it is the public runtime
gateway. The control URL is retained as a backwards-compatible
fallback for installations that have not split the two origins yet.
"""
origin = (
getattr(self._s, "cloud_public_api_url", "")
or getattr(self._s, "cloud_control_api_url", "")
or ""
).strip().rstrip("/")
return origin or None
# ------------------------------------------------------------------
# HTTP helper
# ------------------------------------------------------------------
async def _request(
self,
method: str,
path: str,
*,
json_body: Any = None,
files: Any = None,
content: Any = None,
extra_headers: dict[str, str] | None = 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: dict[str, str] = {"Authorization": f"Bearer {self._token}"}
if extra_headers:
headers.update(extra_headers)
url = f"{self._base()}{path}"
async with httpx.AsyncClient(timeout=90.0) as client:
resp = await client.request(
method.upper(), url,
json=json_body, files=files, content=content, 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 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
# ------------------------------------------------------------------
# dispatch namespace
# ------------------------------------------------------------------
async def _list_namespaces(self) -> list[dict[str, Any]]:
_, result = await self._request("GET", "/workers/dispatch/namespaces", expected={200})
if isinstance(result, list):
return [x for x in result if isinstance(x, dict)]
if isinstance(result, dict):
inner = result.get("result")
if isinstance(inner, list):
return inner
return []
async def _find_namespace(self, name: str) -> dict[str, Any] | None:
for row in await self._list_namespaces():
n = row.get("namespace_name") or row.get("name") or ""
if n.lower() == name.lower():
return row
return None
async def ensure_dispatch_namespace(self, environment: str) -> dict[str, Any]:
name = self._dispatch_namespace_name(environment)
existing = await self._find_namespace(name)
if existing:
return existing
# CF requires lowercase for new namespace names
create_name = name.lower()
try:
_, result = await self._request(
"POST", "/workers/dispatch/namespaces",
json_body={"name": create_name, "namespace_name": create_name, "trusted_workers": bool(self._s.cloudflare_w4p_trusted_workers)},
expected={200, 201},
)
logger.debug("dispatch_namespace_creation_request_sent", name=create_name, result=result)
return result or {}
except CloudflareAPIError:
existing = await self._find_namespace(name)
if existing:
return existing
raise
# ------------------------------------------------------------------
# proxy Worker (thin forwarder to CF Pages URL)
# ------------------------------------------------------------------
def _proxy_script_name(self, project_ref: str, environment: str) -> str:
env = "prev" if (environment or "").lower() == "preview" else "prod"
raw = f"p2hs-{env}-{project_ref}"
slug = re.sub(r"[^a-z0-9-]+", "-", raw.lower()).strip("-")
slug = re.sub(r"-{2,}", "-", slug)
return slug[:63].rstrip("-")
def _proxy_worker_source(self, target_url: str, project_ref: str) -> str:
"""Tiny ES-module Worker that proxies all requests to ``target_url`` (CF Pages).
Browser clients call same-origin ``/api/*``. This proxy keeps the public
hostname; Pages Functions remain the BFF translation layer until a
project-specific Cloud binding is attached. The physical Cloud origin
must never be written into customer Git or the frontend bundle.
"""
origin = json.dumps(target_url.rstrip("/"))
ref = json.dumps(project_ref)
return f"""export default {{
async fetch(request, env, ctx) {{
const incoming = new URL(request.url);
const base = new URL({origin});
const upstream = new URL(incoming.pathname + incoming.search, base);
const headers = new Headers(request.headers);
headers.set("x-p2hs-project-ref", {ref});
headers.set("x-forwarded-host", incoming.host);
headers.set("x-forwarded-proto", incoming.protocol.replace(":", "") || "https");
headers.set("x-p2hs-public-origin", incoming.origin);
const init = {{ method: request.method, headers, redirect: "manual" }};
if (request.method !== "GET" && request.method !== "HEAD") init.body = request.body;
const response = await fetch(new Request(upstream.toString(), init));
const location = response.headers.get("location");
if (!location) return response;
try {{
const loc = new URL(location, upstream);
const locHost = loc.hostname.toLowerCase();
const rewrite = locHost.endsWith(".pages.dev")
|| /^prev-[a-z0-9]+\\./.test(locHost)
|| /^proj-[a-z0-9]+\\./.test(locHost);
if (rewrite && loc.origin !== incoming.origin) {{
const out = new Headers(response.headers);
out.set("location", incoming.origin + loc.pathname + loc.search + loc.hash);
return new Response(response.body, {{ status: response.status, statusText: response.statusText, headers: out }});
}}
}} catch {{ /* keep provider Location */ }}
return response;
}},
}};""".strip()
async def _upload_proxy_worker(
self,
namespace_name: str,
script_name: str,
target_url: str,
project_ref: str,
) -> None:
source = self._proxy_worker_source(target_url, project_ref)
metadata = json.dumps({"main_module": "worker.mjs", "compatibility_date": self._compat_date})
files = {
"metadata": (None, metadata, "application/json"),
"worker.mjs": ("worker.mjs", source, "application/javascript+module"),
}
# Use the canonical namespace name as it appears in the API (preserves case)
existing = await self._find_namespace(namespace_name)
ns_path = quote((existing.get("namespace_name") or namespace_name).strip(), safe="")
await self._request(
"PUT",
f"/workers/dispatch/namespaces/{ns_path}/scripts/{quote(script_name, safe='')}",
files=files,
expected={200, 201},
)
# ------------------------------------------------------------------
# KV routing
# ------------------------------------------------------------------
async def _kv_get(self, key: str) -> str | None:
"""Return the value for ``key`` from the routing KV namespace, or None if missing."""
kv_id = await self._resolve_kv_namespace_id()
encoded_key = quote(key, safe="")
# The KV values endpoint returns the value as raw text, not a Cloudflare
# API JSON envelope. Use one request and preserve the raw response;
# routing callers need to distinguish a missing key from a stored value.
headers = {"Authorization": f"Bearer {self._token}"}
url = f"{self._base()}/storage/kv/namespaces/{kv_id}/values/{encoded_key}"
try:
async with httpx.AsyncClient(timeout=30.0) as client:
resp = await client.get(url, headers=headers)
if resp.status_code == 404:
return None
resp.raise_for_status()
return resp.text
except Exception:
return None
async def _kv_put(self, key: str, value: str) -> None:
kv_id = await self._resolve_kv_namespace_id()
encoded_key = quote(key, safe="")
await self._request(
"PUT",
f"/storage/kv/namespaces/{kv_id}/values/{encoded_key}",
content=value.encode("utf-8"),
extra_headers={"Content-Type": "text/plain; charset=utf-8"},
expected={200},
)
async def _kv_delete(self, key: str) -> None:
kv_id = await self._resolve_kv_namespace_id()
encoded_key = quote(key, safe="")
await self._request(
"DELETE",
f"/storage/kv/namespaces/{kv_id}/values/{encoded_key}",
expected={200},
allow_statuses={404},
)
# ------------------------------------------------------------------
# dispatcher Worker (shared infrastructure, deployed once on startup)
# ------------------------------------------------------------------
def _dispatcher_source(self, apex: str) -> str:
"""ES-module dispatcher Worker: reads KV, dispatches to per-project proxy Worker.
Also handles visual-edit mode: injects the P2HS visual-edit script into HTML
responses when ``?__p2hs_ve=1`` is present or the ``__p2hs_ve`` cookie is set.
The param is stripped before dispatching so proxy Workers and apps never see it.
"""
manifest_json = json.dumps(compile_routing_manifest(), separators=(",", ":"))
from src.domain.adopt.edge_dispatch import edge_dispatch_source
from src.domain.hosting.runtime.worker_source import session_bridge_source
# The bridge and the adopt split are plain script, so their braces stay intact.
return session_bridge_source() + "\n" + edge_dispatch_source() + "\n" + f"""/**
* P2HS Dispatcher — routes *.{apex} to per-project proxy Workers.
* Bindings: ROUTING_KV (KV), DISPATCH_NS_PRODUCTION (W4P), DISPATCH_NS_PREVIEW (W4P)
*
* Visual edit: __p2hs_ve param carries the dashboard origin (e.g. https://app.p2hs.com).
* The Worker injects the visual-edit script from that origin into HTML responses and
* persists the origin in a cookie so subsequent iframe navigations keep the mode active.
* Pass __p2hs_ve=0 to exit visual edit and clear the cookie.
*/
export default {{
async fetch(request, env) {{
const incoming = new URL(request.url);
if (incoming.protocol === "http:") {{
incoming.protocol = "https:";
return Response.redirect(incoming.toString(), 301);
}}
const hostname = incoming.hostname;
const routeJson = await env.ROUTING_KV.get(`host:${{hostname}}`);
if (!routeJson) {{
return new Response(JSON.stringify({{ error: "no_route", hostname }}),
{{ status: 404, headers: {{ "content-type": "application/json" }} }});
}}
let route;
try {{ route = JSON.parse(routeJson); }} catch {{
return new Response(JSON.stringify({{ error: "invalid_route_record" }}),
{{ status: 502, headers: {{ "content-type": "application/json" }} }});
}}
if (!route.script_name) {{
return new Response(JSON.stringify({{ error: "missing_script_name" }}),
{{ status: 502, headers: {{ "content-type": "application/json" }} }});
}}
// The manifest is compiled from the canonical platform port catalog. A
// project opts in through Core-controlled KV state; callers cannot select
// this policy with headers or query parameters.
const ROUTING_MANIFEST = {manifest_json};
const policyKey = route.project_id ? `routing_policy:${{route.project_id}}` : "";
const policy = route.routing_policy === "manifest"
? "manifest"
: (policyKey ? await env.ROUTING_KV.get(policyKey) : null) || "legacy";
// Visual edit: param value is the dashboard origin, or "0" to stop
const veParam = incoming.searchParams.get("__p2hs_ve");
const cookies = request.headers.get("Cookie") || "";
const veCookieEntry = cookies.split(";").map(c => c.trim()).find(c => c.startsWith("__p2hs_ve="));
const veCookieValue = veCookieEntry ? veCookieEntry.slice(10) : null;
let veOrigin = null;
if (veParam && veParam !== "0") {{
veOrigin = veParam;
}} else if (!veParam && veCookieValue && veCookieValue !== "0") {{
veOrigin = veCookieValue;
}}
const visualEdit = !!veOrigin;
const ownerParam = incoming.searchParams.get("__p2hs_owner");
const ownerCookieEntry = cookies.split(";").map(c => c.trim()).find(c => c.startsWith("__p2hs_cloud_bypass="));
const ownerCookieValue = ownerCookieEntry ? ownerCookieEntry.slice(21) : null;
const isPreview = (route.environment || "").toLowerCase() === "preview";
if (!isPreview && !visualEdit && route.project_id) {{
const blocked = await env.ROUTING_KV.get(`cloud_public_blocked:${{route.project_id}}`);
const expected = await env.ROUTING_KV.get(`cloud_owner_bypass:${{route.project_id}}`);
const ownerOk = !!(expected && (ownerParam === expected || ownerCookieValue === expected));
if (blocked === "1" && !ownerOk) {{
return new Response(
`UnavailableThe Web App is not available
If you are the owner of the site please contact support
`,
{{ status: 503, headers: {{ "content-type": "text/html; charset=utf-8", "cache-control": "no-store" }} }}
);
}}
}}
// Strip __p2hs_ve before dispatching so proxy Workers and apps never see it
incoming.searchParams.delete("__p2hs_ve");
incoming.searchParams.delete("__p2hs_owner");
const fwdHeaders = new Headers(request.headers);
fwdHeaders.set("x-forwarded-host", hostname);
fwdHeaders.set("x-forwarded-proto", "https");
fwdHeaders.set("x-p2hs-public-origin", incoming.origin);
if (route.project_id) fwdHeaders.set("x-p2hs-project-id", String(route.project_id));
if (route.project_ref) fwdHeaders.set("x-p2hs-project-ref", String(route.project_ref));
// The browser session is an HttpOnly cookie. Promote it before either the
// Cloud proxy or the project Worker, both of which can lose Cookie on the
// way to Cloud while still forwarding Authorization.
p2hsPromoteSessionHeaders(fwdHeaders);
const sessionCarrier = new Request(incoming.toString(), request);
const rewrittenSession = await p2hsSessionBody(sessionCarrier);
if (rewrittenSession != null) {{
fwdHeaders.delete("content-length");
fwdHeaders.set("content-type", "application/json");
}}
const sessionInit = (method) => {{
const init = {{ method, headers: fwdHeaders, redirect: "manual" }};
if (method !== "GET" && method !== "HEAD") {{
init.body = rewrittenSession != null ? rewrittenSession : sessionCarrier.body;
}}
return init;
}};
const adoptDecision = await p2hsAdoptDispatch(incoming, request, route, fetch);
if (adoptDecision instanceof Response) return adoptDecision;
const adoptProjectWorker = !!(adoptDecision && adoptDecision.projectWorker);
// Set the session cookies on this response. A project Worker can set them
// too, but that Set-Cookie does not survive dispatch back to the browser.
if (incoming.pathname === "/api/auth/magic-link/consume" && request.method === "GET" && route.bff_origin && route.project_id) {{
return p2hsConsumeMagicLink(sessionCarrier, {{
CLOUD_GATEWAY_URL: String(route.bff_origin),
PROJECT_ID: String(route.project_id),
}}, fetch);
}}
if (incoming.pathname === "/api/auth/oauth/callback" && request.method === "GET") {{
const oauthResponse = await p2hsConsumeOAuthCallback(sessionCarrier);
if (oauthResponse) return oauthResponse;
}}
// cloud_proxy (missing api_execution, or any value other than project_runtime)
// keeps /api on the public Cloud gateway. project_runtime sends every path,
// including /api, to the project script. Existing hostnames stay on the proxy
// until Core writes api_execution=project_runtime for that hostname alone.
const apiPath = incoming.pathname === "/api" || incoming.pathname.startsWith("/api/");
if (route.api_execution !== "project_runtime" && !adoptProjectWorker) {{
const routeMatch = (path, method) => {{
for (const entry of ROUTING_MANIFEST) {{
if (entry.method !== "*" && entry.method !== method) continue;
const names = [];
const pattern = entry.public_path.split("/").map(part => {{
if (part.startsWith(":") && part.endsWith("*")) {{ names.push(part.slice(1, -1)); return "(.*)"; }}
if (part.startsWith(":")) {{ names.push(part.slice(1)); return "([^/]+)"; }}
return part.replace(/[.*+?^${{}}()|[\\]\\\\]/g, "\\\\$&");
}}).join("/");
const match = new RegExp("^" + pattern + "/?$").exec(path);
if (match) {{
const params = {{}};
names.forEach((name, index) => params[name] = decodeURIComponent(match[index + 1]));
return {{ entry, params }};
}}
}}
return null;
}};
const contract = policy === "manifest" && apiPath ? routeMatch(incoming.pathname, request.method.toUpperCase()) : null;
if (policy === "manifest" && apiPath && contract && contract.entry.owner === "cloud" && route.bff_origin && route.project_id) {{
try {{
const bffBase = new URL(String(route.bff_origin).replace(/\\/+$/, "") + "/");
let cloudPath = contract.entry.upstream_path.replace(":project_id", encodeURIComponent(String(route.project_id)));
for (const [name, value] of Object.entries(contract.params)) {{
const encoded = String(value).split("/").map(encodeURIComponent).join("/");
cloudPath = cloudPath.replace(`:${{name}}`, encoded);
}}
const projectPath = cloudPath;
const upstream = new URL(projectPath + incoming.search, bffBase);
const bffResponse = await fetch(new Request(upstream.toString(), sessionInit(request.method)));
if (!visualEdit && (route.environment || "").toLowerCase() !== "preview") {{
const length = Number(bffResponse.headers.get("content-length") || 0);
env.HOSTING_USAGE.writeDataPoint({{
blobs: [String(route.project_id || ""), hostname],
doubles: [1, Number.isFinite(length) && length > 0 ? length : 0],
indexes: [String(route.project_id || hostname)],
}});
}}
return p2hsSessionResponse(sessionCarrier, bffResponse);
}} catch (err) {{
return new Response(JSON.stringify({{ error: "bff_proxy_failed", detail: String(err) }}),
{{ status: 502, headers: {{ "content-type": "application/json" }} }});
}}
}}
if (policy === "manifest" && apiPath && contract && contract.entry.owner === "cloud" && (!route.bff_origin || !route.project_id)) {{
return new Response(JSON.stringify({{ error: "cloud_route_unavailable", route: contract.entry.id }}),
{{ status: 503, headers: {{ "content-type": "application/json", "cache-control": "no-store" }} }});
}}
if (policy === "manifest" && apiPath && (!contract || contract.entry.owner === "bff")) {{
// Preserve the original application path for project BFF/custom routes.
// No Cloud fallback is permitted when a project Worker is missing.
}} else if (policy === "legacy" && apiPath && route.bff_origin && route.project_id) {{
try {{
const bffBase = new URL(String(route.bff_origin).replace(/\\/+$/, "") + "/");
const apiSuffix = incoming.pathname === "/api" ? "" : incoming.pathname.slice("/api".length);
const projectPath = `/v1/projects/${{encodeURIComponent(String(route.project_id))}}${{apiSuffix}}`;
const upstream = new URL(projectPath + incoming.search, bffBase);
const bffResponse = await fetch(new Request(upstream.toString(), sessionInit(request.method)));
if (!visualEdit && (route.environment || "").toLowerCase() !== "preview") {{
const length = Number(bffResponse.headers.get("content-length") || 0);
env.HOSTING_USAGE.writeDataPoint({{
blobs: [String(route.project_id || ""), hostname],
doubles: [1, Number.isFinite(length) && length > 0 ? length : 0],
indexes: [String(route.project_id || hostname)],
}});
}}
return p2hsSessionResponse(sessionCarrier, bffResponse);
}} catch (err) {{
return new Response(JSON.stringify({{ error: "bff_proxy_failed", detail: String(err) }}),
{{ status: 502, headers: {{ "content-type": "application/json" }} }});
}}
}}
}}
const cleanRequest = new Request(incoming.toString(), sessionInit(request.method));
const ns = (route.environment || "").toLowerCase() === "preview"
? env.DISPATCH_NS_PREVIEW
: env.DISPATCH_NS_PRODUCTION;
let response;
try {{
response = await ns.get(route.script_name).fetch(cleanRequest);
}} catch (err) {{
return new Response(JSON.stringify({{ error: "dispatch_failed", script: route.script_name, detail: String(err) }}),
{{ status: 502, headers: {{ "content-type": "application/json" }} }});
}}
const location = response.headers.get("location");
if (location) {{
try {{
const loc = new URL(location, incoming);
const locHost = loc.hostname.toLowerCase();
const rewrite = locHost.endsWith(".pages.dev")
|| /^prev-[a-z0-9]+\\./.test(locHost)
|| /^proj-[a-z0-9]+\\./.test(locHost);
if (rewrite && loc.origin !== incoming.origin) {{
const out = p2hsCloneResponseHeaders(response.headers);
out.set("location", incoming.origin + loc.pathname + loc.search + loc.hash);
response = new Response(response.body, {{ status: response.status, statusText: response.statusText, headers: out }});
}}
}} catch {{ /* keep provider Location */ }}
}}
// Monitor public app traffic without touching the request path or billing.
// The project ID is the Analytics Engine sampling key, enabling per-project totals.
if (!visualEdit && (route.environment || "").toLowerCase() !== "preview") {{
const length = Number(response.headers.get("content-length") || 0);
env.HOSTING_USAGE.writeDataPoint({{
blobs: [String(route.project_id || ""), hostname],
doubles: [1, Number.isFinite(length) && length > 0 ? length : 0],
indexes: [String(route.project_id || hostname)],
}});
}}
const needsCookieUpdate = !!veParam || !!ownerParam;
if (!visualEdit && !needsCookieUpdate) return p2hsSessionResponse(sessionCarrier, response);
const newHeaders = p2hsCloneResponseHeaders(response.headers);
if (veParam && veParam !== "0") {{
newHeaders.append("Set-Cookie", `__p2hs_ve=${{veParam}}; Path=/`);
}} else if (veParam === "0") {{
newHeaders.append("Set-Cookie", "__p2hs_ve=0; Path=/; Max-Age=0");
}}
if (ownerParam) {{
newHeaders.append("Set-Cookie", `__p2hs_cloud_bypass=${{ownerParam}}; Path=/; Secure; SameSite=Lax`);
}}
const ct = response.headers.get("content-type") || "";
if (visualEdit && response.status === 200 && ct.includes("text/html")) {{
const html = await response.text();
const tag = ``;
const idx = html.lastIndexOf(" ");
const injected = idx >= 0
? html.slice(0, idx) + tag + html.slice(idx)
: html + tag;
// Remove Content-Length — injected body is longer than the original
newHeaders.delete("content-length");
return new Response(injected, {{ status: 200, headers: newHeaders }});
}}
return p2hsSessionResponse(sessionCarrier, new Response(response.body, {{ status: response.status, statusText: response.statusText, headers: newHeaders }}));
}},
}};""".strip()
async def _actual_namespace_name(self, environment: str) -> str:
"""Return the real CF namespace name for bindings (case-preserving).
``_dispatch_namespace_name`` lowercases for new-namespace creation.
When the namespace already exists (e.g. user created it as ``P2HS``),
bindings must use the exact stored name or CF returns 404.
"""
configured = self._dispatch_namespace_name(environment) # lowercase
existing = await self._find_namespace(configured) # case-insensitive search
if existing:
return (existing.get("namespace_name") or configured).strip()
return configured # not found yet — will be created with lowercase name
async def deploy_dispatcher(self, apex: str | None = None) -> dict[str, Any]:
"""Upload the dispatcher Worker and register the wildcard route.
Called from the worker startup hook so the dispatcher is always
in sync with the current codebase. Skips the upload when the
deployed content hash matches the current source (checked via KV).
Returns a dict with ``ok``, ``script_name``, and optional ``route``.
"""
import hashlib
apex = (apex or self._s.hosting_public_apex or "sites.p2hs.com").strip()
compat_date = self._compat_date
script_name = "p2hs-dispatcher"
# Ensure both dispatch namespaces exist before referencing them in bindings.
# Creates them if absent (lowercase name); returns existing if already present.
ns_prod_row = await self.ensure_dispatch_namespace("production")
ns_prev_row = await self.ensure_dispatch_namespace("preview")
# Use the actual stored name (case-preserved) for the binding
ns_prod = (ns_prod_row.get("namespace_name") or self._dispatch_namespace_name("production")).strip()
ns_prev = (ns_prev_row.get("namespace_name") or self._dispatch_namespace_name("preview")).strip()
kv_id = await self._resolve_kv_namespace_id()
source = self._dispatcher_source(apex)
metadata = json.dumps({
"main_module": "dispatcher.mjs",
"compatibility_date": compat_date,
"bindings": [
{"type": "kv_namespace", "name": "ROUTING_KV", "namespace_id": kv_id},
{"type": "dispatch_namespace", "name": "DISPATCH_NS_PRODUCTION", "namespace": ns_prod},
{"type": "dispatch_namespace", "name": "DISPATCH_NS_PREVIEW", "namespace": ns_prev},
{
"type": "analytics_engine",
"name": "HOSTING_USAGE",
"dataset": self._s.hosting_usage_analytics_dataset,
},
],
})
# Skip upload if nothing has changed since the last deploy.
content_hash = hashlib.sha256((source + metadata).encode()).hexdigest()
_HASH_KV_KEY = "_dispatcher_hash"
stored_hash = await self._kv_get(_HASH_KV_KEY)
if stored_hash == content_hash:
logger.info("dispatcher_worker_unchanged", script=script_name, apex=apex)
result: dict[str, Any] = {"ok": True, "script_name": script_name, "skipped": True}
# Still ensure the route exists even if we skipped the script upload
else:
await self._request(
"PUT",
f"/workers/scripts/{script_name}",
files={
"metadata": (None, metadata, "application/json"),
"dispatcher.mjs": ("dispatcher.mjs", source.encode(), "application/javascript+module"),
},
expected={200, 201},
)
await self._kv_put(_HASH_KV_KEY, content_hash)
logger.info("dispatcher_worker_deployed", script=script_name, apex=apex)
result = {"ok": True, "script_name": script_name}
# Register zone Worker routes if zone_id is configured.
# Two routes are needed:
# 1. *.{apex}/* — platform URLs (proj-xxx.sites.p2hsapp.com)
# 2. */* — catch-all for customer custom domains arriving via CF for SaaS
# CF matches the most-specific pattern first, so platform URLs always hit route 1.
# Custom hostname requests (myapp.example.com) only match route 2.
zone_id = (getattr(self._s, "cloudflare_zone_id", None) or "").strip()
if zone_id:
patterns = [f"*.{apex}/*", "*/*"]
registered: list[str] = []
try:
headers = {"Authorization": f"Bearer {self._token}"}
zone_base = f"https://api.cloudflare.com/client/v4/zones/{zone_id}/workers/routes"
async with httpx.AsyncClient(timeout=30.0) as client:
list_resp = await client.get(zone_base, headers=headers)
list_resp.raise_for_status()
existing: dict[str, str] = {
r["pattern"]: r["id"]
for r in (list_resp.json().get("result") or [])
}
for pattern in patterns:
if pattern in existing:
await client.put(
f"{zone_base}/{existing[pattern]}",
headers=headers,
json={"pattern": pattern, "script": script_name},
)
else:
cre = await client.post(
zone_base,
headers=headers,
json={"pattern": pattern, "script": script_name},
)
if cre.status_code not in (200, 201):
logger.warning(
"dispatcher_route_register_failed",
pattern=pattern, status=cre.status_code, body=cre.text,
)
continue
registered.append(pattern)
logger.debug("dispatcher_route_registered", pattern=pattern)
result["routes"] = registered
except Exception as exc:
logger.warning("dispatcher_route_error", error=str(exc))
else:
logger.info("dispatcher_route_skipped", reason="CLOUDFLARE_ZONE_ID not configured")
return result
# ------------------------------------------------------------------
# Template / pre-deployed Worker routes (KV only — script must exist in dispatch NS)
# ------------------------------------------------------------------
_SUBDOMAIN_LABEL_RE = re.compile(r"^[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?$")
def hostname_from_subdomain(self, subdomain: str) -> str:
"""Build ``{label}.{apex}`` from a single DNS label and ``hosting_public_apex``."""
label = (subdomain or "").strip().lower()
if not label or not self._SUBDOMAIN_LABEL_RE.fullmatch(label):
raise ValueError(
"Invalid subdomain label: use 1–63 chars, letters, digits, hyphens; "
"must not start or end with a hyphen."
)
apex = (self._s.hosting_public_apex or "sites.p2hs.com").strip().lower().rstrip(".")
return f"{label}.{apex}"
async def upsert_dispatch_route_for_existing_script(
self,
*,
subdomain: str,
script_name: str,
environment: str = "production",
project_ref: str | None = None,
project_id: str = "",
target_url: str | None = None,
) -> dict[str, Any]:
"""Write ``host:{hostname}`` KV to point at an existing W4P user Worker script.
Does **not** upload a proxy Worker — the script must already exist in the dispatch
namespace for ``environment`` (e.g. template app deployed out-of-band).
"""
hostname = self.hostname_from_subdomain(subdomain)
sn = (script_name or "").strip()
if not sn or len(sn) > 63:
raise ValueError("script_name is required and must be at most 63 characters.")
env = (environment or "production").lower()
if env not in ("production", "preview", "staging"):
raise ValueError("environment must be production, preview, or staging.")
pref = resolve_dispatch_route_project_ref(
subdomain=subdomain,
script_name=sn,
environment=env,
explicit_project_ref=project_ref,
)
tu = (target_url or "").strip() or f"https://{hostname}/"
descriptor = RoutingDescriptor(
target_kind="cloudflare_pages",
project_ref=pref,
project_id=str(project_id or ""),
environment=env,
target_url=tu,
# The target URL is the hosted asset origin. It must never be used
# as the BFF origin: doing so makes /api requests fetch the same
# dispatcher route recursively and produces Cloudflare 1101 errors.
bff_origin=self._default_bff_origin(),
deployment_target_id=None,
active_deployment_id=None,
)
kv_value = json.dumps({**descriptor.as_dict(), "script_name": sn})
await self._kv_put(f"host:{hostname}", kv_value)
await self._kv_put(f"project:{pref}:{env}", hostname)
logger.info(
"dispatch_kv_route_upserted",
hostname=hostname,
script_name=sn,
environment=env,
)
return {
"hostname": hostname,
"kv_key": f"host:{hostname}",
"script_name": sn,
"environment": env,
"project_ref": pref,
"project_id": str(project_id or ""),
}
async def delete_dispatch_route_by_subdomain(
self,
subdomain: str,
*,
project_ref: str | None = None,
script_name: str | None = None,
environment: str = "production",
) -> dict[str, Any]:
"""Remove KV entries for a template route built from ``subdomain`` + apex."""
hostname = self.hostname_from_subdomain(subdomain)
env = (environment or "production").lower()
if env not in ("production", "preview", "staging"):
raise ValueError("environment must be production, preview, or staging.")
pref = resolve_dispatch_route_project_ref(
subdomain=subdomain,
script_name=(script_name or "").strip(),
environment=env,
explicit_project_ref=project_ref,
)
await self.delete_route(hostname, pref, env)
return {"hostname": hostname, "removed": True}
# ------------------------------------------------------------------
# public API
# ------------------------------------------------------------------
async def upsert_route(
self,
hostname: str,
descriptor: RoutingDescriptor,
environment: str,
) -> None:
"""Ensure namespace exists, upload proxy Worker, write KV routing record."""
ns = await self.ensure_dispatch_namespace(environment)
ns_name = ns.get("namespace_name") or self._dispatch_namespace_name(environment)
script_name = self._proxy_script_name(descriptor.project_ref, environment)
await self._upload_proxy_worker(ns_name, script_name, descriptor.target_url, descriptor.project_ref)
if not descriptor.bff_origin:
descriptor.bff_origin = self._default_bff_origin()
# KV: hostname → routing descriptor JSON
kv_value = json.dumps({**descriptor.as_dict(), "script_name": script_name})
await self._kv_put(f"host:{hostname}", kv_value)
# Secondary lookup: project ref → hostname (for internal tooling)
await self._kv_put(f"project:{descriptor.project_ref}:{environment}", hostname)
logger.info(
"routing_route_upserted",
hostname=hostname,
script_name=script_name,
target_url=descriptor.target_url,
bff_origin=descriptor.bff_origin,
)
async def alias_custom_hostname(
self,
custom_hostname: str,
project_ref: str,
environment: str,
target_url: str,
active_deployment_id: str | None = None,
) -> None:
"""Write a KV routing entry for a user's custom domain.
Reuses the same proxy Worker script already deployed for the canonical hostname
(prod/preview). The proxy Worker is identified by project_ref + environment so
both the canonical URL and any custom domain route to the same backend.
"""
script_name = self._proxy_script_name(project_ref, environment)
descriptor = RoutingDescriptor(
target_kind="cloudflare_pages",
project_ref=project_ref,
project_id="",
environment=environment,
target_url=target_url,
bff_origin=self._default_bff_origin(),
deployment_target_id=None,
active_deployment_id=active_deployment_id,
)
kv_value = json.dumps({**descriptor.as_dict(), "script_name": script_name})
await self._kv_put(f"host:{custom_hostname}", kv_value)
logger.info(
"custom_domain_route_aliased",
custom_hostname=custom_hostname,
script_name=script_name,
target_url=target_url,
)
async def remove_custom_hostname(self, custom_hostname: str) -> None:
"""Remove the KV routing entry for a custom domain."""
await self._kv_delete(f"host:{custom_hostname}")
logger.info("custom_domain_route_removed", custom_hostname=custom_hostname)
async def update_bff_origin(self, hostname: str, bff_origin: str) -> dict[str, str]:
"""Point ``/api/*`` at a new physical BFF origin without touching Git or the proxy Worker."""
key = f"host:{hostname}"
raw = await self._kv_get(key)
if not raw:
raise CloudflareAPIError(f"no routing record for {hostname}")
try:
route = json.loads(raw)
except json.JSONDecodeError as exc:
raise CloudflareAPIError(f"invalid routing record for {hostname}") from exc
if not isinstance(route, dict):
raise CloudflareAPIError(f"invalid routing record for {hostname}")
from src.domain.hosting.bff_indirection import failover_bff_origin, normalize_origin
updated = failover_bff_origin(route, normalize_origin(bff_origin))
await self._kv_put(key, json.dumps(updated))
logger.info(
"bff_origin_updated",
hostname=hostname,
bff_origin=updated.get("bff_origin"),
git_commits_required=0,
rebuild_required=0,
)
return {
"hostname": hostname,
"bff_origin": str(updated.get("bff_origin") or ""),
"git_commits_required": "0",
"rebuild_required": "0",
}
async def set_routing_policy(self, project_id: str, policy: str) -> dict[str, str]:
"""Set the Core-controlled project routing gate used by the dispatcher.
Only the two compiled policies are accepted. This key is written by
Core control-plane code, never from request headers or customer query
parameters, and is safe to repeat during a gated rollout.
"""
value = (policy or "").strip().lower()
if value not in {"legacy", "manifest"}:
raise ValueError("policy must be legacy or manifest")
project = str(project_id or "").strip()
if not project:
raise ValueError("project_id is required")
await self._kv_put(f"routing_policy:{project}", value)
logger.info("routing_policy_updated", project_id=project, policy=value)
return {"project_id": project, "policy": value}
async def delete_route(self, hostname: str, project_ref: str, environment: str) -> None:
await self._kv_delete(f"host:{hostname}")
await self._kv_delete(f"project:{project_ref}:{environment}")
logger.info("routing_route_deleted", hostname=hostname)
# ------------------------------------------------------------------
# DeploymentPlaneProvider-compatible helpers (used by project_deploy task)
# ------------------------------------------------------------------
async def ensure_project_environment(self, ctx: DeployJobContext) -> DeploymentOutcome:
"""Ensure the W4P dispatch namespace exists (idempotent)."""
try:
ns = await self.ensure_dispatch_namespace(ctx.environment)
return DeploymentOutcome(
ok=True,
last_completed_step="ensure_dispatch_namespace",
metadata={"adapter": "cloudflare_w4p_routing", "namespace": ns.get("namespace_name")},
)
except CloudflareAPIError as exc:
return DeploymentOutcome(ok=False, message=str(exc), metadata={"adapter": "cloudflare_w4p_routing"})
async def activate_route(
self,
ctx: DeployJobContext,
target_url: str,
deployment_target_id: str | None = None,
active_deployment_id: str | None = None,
) -> DeploymentOutcome:
"""Upload proxy Worker + write KV routing record for this hostname."""
from urllib.parse import urlparse
hostname = urlparse(ctx.target_hostname if "://" in ctx.target_hostname else f"https://{ctx.target_hostname}").netloc or ctx.target_hostname
descriptor = RoutingDescriptor(
target_kind="cloudflare_pages",
project_ref=ctx.project_ref,
project_id=str(ctx.project_id),
environment=ctx.environment,
target_url=target_url,
bff_origin=self._default_bff_origin(),
deployment_target_id=deployment_target_id,
active_deployment_id=active_deployment_id,
)
try:
await self.upsert_route(hostname, descriptor, ctx.environment)
return DeploymentOutcome(
ok=True,
deploy_url=f"https://{hostname}/",
activation_status="live",
last_completed_step="activate_route",
metadata={"adapter": "cloudflare_w4p_routing", "hostname": hostname, "target_url": target_url},
)
except CloudflareAPIError as exc:
return DeploymentOutcome(ok=False, message=str(exc), metadata={"adapter": "cloudflare_w4p_routing"})