"""P2HS Cloud control-plane client. Syncs workspace / project / client / capability / integration data from Core API into the Cloud gateway's D1 database via internal control endpoints. All endpoints are bearer-token protected and never exposed to browser clients. Mirrors the payload shapes defined in packages/shared/src/control.ts in P2HSCloud. Secret / env distribution (dashboard → execution targets; Core is policy + custody): - **User app Worker (W4P / hosted Git build)**: env vars from ``project_env_vars`` and hosting pipeline — not browser-exposed. - **P2HS Cloud**: integration rows + optional ``secret_ref`` / encrypted blobs; single org R2 bucket with object prefixes ``{workspace_id}/{project_id}/...`` for tenant isolation. - **Customer-connected GitHub**: repo-only secrets when the user attaches their repo — written via GitHub Actions/secrets APIs from Core, scoped to that repository. Per-workspace routing --------------------- When multiple Cloud instances exist (multi-CF-account), use ``get_cloud_client_for_workspace(workspace_id, db)`` to obtain a client routed to the correct instance. Falls back to the global ``cloud_control_api_url`` setting when no per-workspace assignment is found (single-instance / dev). """ from __future__ import annotations import base64 from typing import TYPE_CHECKING, Any, Optional from uuid import UUID import httpx import structlog from cryptography.hazmat.primitives import serialization from cryptography.hazmat.primitives.asymmetric import rsa from src.config import get_settings if TYPE_CHECKING: from sqlalchemy.ext.asyncio import AsyncSession logger = structlog.get_logger() _TIMEOUT = 10 def _b64url_uint(value: int) -> str: raw = value.to_bytes((value.bit_length() + 7) // 8, "big") return base64.urlsafe_b64encode(raw).rstrip(b"=").decode("ascii") def _rsa_public_jwk(public_key_pem: str, *, key_id: str) -> dict[str, Any]: public_key = serialization.load_pem_public_key(public_key_pem.encode("utf-8")) if not isinstance(public_key, rsa.RSAPublicKey): raise ValueError("Project auth public key must be RSA") numbers = public_key.public_numbers() return { "kty": "RSA", "kid": key_id, "use": "sig", "alg": "RS256", "key_ops": ["verify"], "n": _b64url_uint(numbers.n), "e": _b64url_uint(numbers.e), } def _rsa_private_jwk(private_key_pem: str, *, key_id: str) -> dict[str, Any]: private_key = serialization.load_pem_private_key( private_key_pem.encode("utf-8"), password=None, ) if not isinstance(private_key, rsa.RSAPrivateKey): raise ValueError("Project auth private key must be RSA") numbers = private_key.private_numbers() public_numbers = numbers.public_numbers return { "kty": "RSA", "kid": key_id, "use": "sig", "alg": "RS256", "key_ops": ["sign"], "n": _b64url_uint(public_numbers.n), "e": _b64url_uint(public_numbers.e), "d": _b64url_uint(numbers.d), "p": _b64url_uint(numbers.p), "q": _b64url_uint(numbers.q), "dp": _b64url_uint(numbers.dmp1), "dq": _b64url_uint(numbers.dmq1), "qi": _b64url_uint(numbers.iqmp), } class CloudControlClient: """HTTP client for P2HS Cloud internal control endpoints.""" def __init__( self, api_url: Optional[str] = None, token: Optional[str] = None, ) -> None: settings = get_settings() self._url = (api_url or settings.cloud_control_api_url or "").rstrip("/") self._token = token or settings.cloud_internal_token @property def api_url(self) -> str: return self._url def _is_configured(self) -> bool: return bool(self._url and self._token) def _headers(self) -> dict: return { "Authorization": f"Bearer {self._token}", "Content-Type": "application/json", } async def _put(self, path: str, payload: dict[str, Any]) -> dict: async with httpx.AsyncClient(timeout=_TIMEOUT) as client: resp = await client.put( f"{self._url}{path}", headers=self._headers(), json=payload, ) if resp.status_code >= 400: detail = resp.text[:500] try: body = resp.json() detail = str(body.get("message") or body.get("error") or detail) except Exception: pass logger.warning( "cloud_control_put_failed", path=path, status_code=resp.status_code, detail=detail, ) resp.raise_for_status() return resp.json() async def _post(self, path: str, payload: dict[str, Any]) -> dict: async with httpx.AsyncClient(timeout=_TIMEOUT) as client: resp = await client.post( f"{self._url}{path}", headers=self._headers(), json=payload, ) resp.raise_for_status() return resp.json() async def mint_dashboard_grant( self, project_id: str, *, scopes: list[str], ttl_seconds: int = 300, ) -> dict[str, Any]: """Mint a short-lived read-only dashboard grant in Cloud. Cloud remains the owner of records, logs, and jobs. Core only authorizes the dashboard request and transports the opaque grant to the caller. """ return await self._post( f"/v1/cloud/control/projects/{project_id}/dashboard-grants", {"project_id": project_id, "scopes": scopes, "ttl_seconds": ttl_seconds}, ) async def fetch_dashboard_read( self, path: str, *, grant_token: str, query: Optional[dict[str, str]] = None, ) -> tuple[int, dict[str, Any]]: """Proxy one allowlisted dashboard read through Core to Cloud. The grant is still minted by Cloud and remains the authority for the returned data. Core only provides the authenticated transport boundary for the trusted dashboard; it does not persist or reinterpret records. """ if not path.startswith("/v1/projects/"): raise ValueError("dashboard read path must target the Cloud project API") async with httpx.AsyncClient(timeout=_TIMEOUT) as client: resp = await client.get( f"{self._url}{path}", params=query or {}, headers={ "Authorization": f"Bearer {grant_token}", "Accept": "application/json", }, ) try: payload = resp.json() except ValueError: payload = {"error": "cloud_invalid_json"} return resp.status_code, payload async def _patch(self, path: str, payload: dict[str, Any]) -> dict: async with httpx.AsyncClient(timeout=_TIMEOUT) as client: resp = await client.patch( f"{self._url}{path}", headers=self._headers(), json=payload, ) resp.raise_for_status() return resp.json() async def _get(self, path: str) -> dict: async with httpx.AsyncClient(timeout=_TIMEOUT) as client: resp = await client.get( f"{self._url}{path}", headers=self._headers(), ) resp.raise_for_status() return resp.json() async def _delete(self, path: str) -> dict: async with httpx.AsyncClient(timeout=_TIMEOUT) as client: resp = await client.delete( f"{self._url}{path}", headers=self._headers(), ) resp.raise_for_status() if not resp.content: return {"ok": True} return resp.json() async def get_wp_site_shape_summary( self, *, project_id: str, installation_id: str, ) -> dict[str, Any]: """Return compact WordPress Site Understanding context from Cloud D1. Cloud owns the D1 index and backup/rollback payload references, but Core owns assistant orchestration. This method intentionally asks for a compact planning summary rather than raw site data. """ if not self._is_configured(): return {"ok": False, "status": "missing", "error": "cloud_not_configured"} try: return await self._get( f"/v1/projects/{project_id}/wp/site-shape/summary?installation_id={installation_id}" ) except httpx.HTTPStatusError as exc: logger.warning( "cloud_wp_site_shape_summary_failed", project_id=project_id, installation_id=installation_id, status_code=exc.response.status_code, ) return { "ok": False, "status": "missing", "error": f"cloud_http_{exc.response.status_code}", } except Exception as exc: logger.warning( "cloud_wp_site_shape_summary_error", project_id=project_id, installation_id=installation_id, error=str(exc), ) return {"ok": False, "status": "missing", "error": "cloud_unavailable"} async def upsert_wp_site_shape_summary( self, *, project_id: str, payload: dict[str, Any], ) -> dict[str, Any]: """Store compact WordPress Site Understanding pointers in Cloud D1. The full sanitized JSON export lives in the managed project workspace or R2. D1 receives only query-friendly status, counts, hashes, and artifact references so it stays small and fast. """ if not self._is_configured(): return {"ok": False, "status": "skipped", "error": "cloud_not_configured"} try: return await self._put(f"/v1/projects/{project_id}/wp/site-shape/summary", payload) except httpx.HTTPStatusError as exc: logger.warning( "cloud_wp_site_shape_summary_upsert_failed", project_id=project_id, status_code=exc.response.status_code, ) return {"ok": False, "status": "failed", "error": f"cloud_http_{exc.response.status_code}"} except Exception as exc: logger.warning( "cloud_wp_site_shape_summary_upsert_error", project_id=project_id, error=str(exc), ) return {"ok": False, "status": "failed", "error": "cloud_unavailable"} async def upload_storage_file( self, *, project_id: str, bucket_name: str, filename: str, content_type: str, content: bytes, signed_url_expires_in: int = 24 * 3600, ) -> dict[str, Any]: """Upload bytes through Cloud storage and return file/ref metadata. Core uses this for codegen prompt attachments so Cloud remains the owner of R2 object keys, file rows, usage accounting, and import-compatible file IDs. """ if not self._is_configured(): return {"ok": False, "error": "cloud_not_configured"} presign = await self._post( f"/v1/projects/{project_id}/storage/{bucket_name}/presign", { "filename": filename, "content_type": content_type, "size_bytes": len(content), }, ) upload_url = presign.get("upload_url") file_id = presign.get("file_id") if not upload_url or not file_id: return {"ok": False, "error": "cloud_presign_missing_upload_url"} async with httpx.AsyncClient(timeout=_TIMEOUT) as client: put_resp = await client.put( str(upload_url), content=content, headers={"Content-Type": content_type or "application/octet-stream"}, ) put_resp.raise_for_status() confirmed = await self._post( f"/v1/projects/{project_id}/storage/{bucket_name}/files/{file_id}/confirm", {}, ) signed = await self._post( f"/v1/projects/{project_id}/storage/{bucket_name}/files/{file_id}/presign-download", {"expires_in": signed_url_expires_in}, ) return { "ok": True, "file_id": file_id, "bucket": bucket_name, "filename": confirmed.get("filename") or filename, "content_type": confirmed.get("content_type") or content_type, "size_bytes": confirmed.get("size_bytes") or len(content), "download_url": signed.get("download_url"), "download_url_expires_in": signed.get("expires_in"), } async def upsert_workspace( self, workspace_id: str, name: str, status: str, plan_slug: Optional[str] = None, tenant_id: Optional[str] = None, cloud_enabled: bool = True, suspended: bool = False, metadata: Optional[dict] = None, ) -> None: if not self._is_configured(): logger.debug("cloud_control_skip", reason="not_configured", op="upsert_workspace") return await self._put( f"/v1/cloud/control/workspaces/{workspace_id}", { "workspace_id": workspace_id, "tenant_id": tenant_id, "name": name, "status": status, "plan_slug": plan_slug, "cloud_enabled": cloud_enabled, "suspended": suspended, "metadata": metadata or {}, }, ) logger.info("cloud_workspace_synced", workspace_id=workspace_id) async def upsert_project( self, project_id: str, workspace_id: str, name: str, status: str, allowed_request_types: list[str], origins: list[str], environment: Optional[str] = None, cloud_enabled: bool = True, suspended: bool = False, metadata: Optional[dict] = None, ) -> None: if not self._is_configured(): logger.debug("cloud_control_skip", reason="not_configured", op="upsert_project") return await self._put( f"/v1/cloud/control/projects/{project_id}", { "project_id": project_id, "workspace_id": workspace_id, "name": name, "status": status, "environment": environment, "cloud_enabled": cloud_enabled, "suspended": suspended, "allowed_request_types": allowed_request_types, "origins": origins, "metadata": metadata or {}, }, ) logger.info("cloud_project_synced", project_id=project_id) async def upsert_client( self, client_id: str, workspace_id: str, project_id: str, name: str, status: str, allowed_origins: list[str], allowed_capabilities: list[str], ) -> None: if not self._is_configured(): logger.debug("cloud_control_skip", reason="not_configured", op="upsert_client") return await self._put( f"/v1/cloud/control/clients/{client_id}", { "client_id": client_id, "workspace_id": workspace_id, "project_id": project_id, "name": name, "status": status, "allowed_origins": allowed_origins, "allowed_capabilities": allowed_capabilities, "handshake_mode": "token", }, ) logger.info("cloud_client_synced", client_id=client_id, project_id=project_id) async def upsert_capabilities( self, project_id: str, version: int, capabilities: dict[str, bool], ) -> None: if not self._is_configured(): logger.debug("cloud_control_skip", reason="not_configured", op="upsert_capabilities") return await self._put( f"/v1/cloud/control/projects/{project_id}/capabilities", { "project_id": project_id, "version": version, "capabilities": capabilities, }, ) logger.info("cloud_capabilities_synced", project_id=project_id) async def upsert_integration( self, project_id: str, provider: str, status: str, secret_ref: Optional[str] = None, config: Optional[dict] = None, ) -> None: if not self._is_configured(): logger.debug("cloud_control_skip", reason="not_configured", op="upsert_integration") return await self._put( f"/v1/cloud/control/integrations/{project_id}:{provider}", { "project_id": project_id, "provider": provider, "status": status, "secret_ref": secret_ref, "config": config or {}, }, ) logger.info("cloud_integration_synced", project_id=project_id, provider=provider) async def push_integration_secrets( self, project_id: str, provider: str, secret_ref: str, secrets: dict[str, str], ) -> None: """Push decrypted integration secrets to Cloud KV (integ_secret:{secret_ref}). Core decrypts secrets_enc before calling this. Cloud stores the result in AUTH_KV and never exposes it via any API surface. Called on install and rotate_secrets; uninstall deletes the KV entry. """ if not self._is_configured(): logger.debug("cloud_control_skip", reason="not_configured", op="push_integration_secrets") return await self._post( f"/v1/cloud/control/integrations/{project_id}:{provider}/secrets", { "project_id": project_id, "provider": provider, "secret_ref": secret_ref, "secrets": secrets, }, ) logger.info("cloud_integration_secrets_pushed", project_id=project_id, provider=provider) async def delete_integration_secrets(self, project_id: str, provider: str) -> None: if not self._is_configured(): logger.debug("cloud_control_skip", reason="not_configured", op="delete_integration_secrets") return await self._delete(f"/v1/cloud/control/integrations/{project_id}:{provider}/secrets") logger.info("cloud_integration_secrets_deleted", project_id=project_id, provider=provider) async def upsert_connector_connection( self, connection_id: str, connector_key: str, workspace_id: str, project_id: Optional[str], name: str, status: str, auth_data_ref: Optional[str], scopes: list[str], config: Optional[dict] = None, permissions: Optional[list[dict]] = None, ) -> None: """Push connector connection metadata and policy to Cloud. Core remains the source of truth. Payloads intentionally contain only non-secret config plus opaque auth_data_ref. """ if not self._is_configured(): logger.debug("cloud_control_skip", reason="not_configured", op="upsert_connector_connection") return await self._put( f"/v1/cloud/control/connectors/connections/{connection_id}", { "connection_id": connection_id, "connector_key": connector_key, "workspace_id": workspace_id, "project_id": project_id, "name": name, "status": status, "auth_data_ref": auth_data_ref, "scopes": scopes, "config": config or {}, "permissions": permissions or [], }, ) logger.info( "cloud_connector_connection_synced", connection_id=connection_id, connector_key=connector_key, ) async def upsert_event_subscription( self, *, subscription_id: str, workspace_id: str, project_id: str, connection_id: str | None, event: str, event_version: str, destination: str, url: str | None, filter_json: dict, ignore_origin: bool, max_hops: int, status: str, signing_secret: str | None = None, ) -> None: if not self._is_configured(): logger.debug("cloud_control_skip", reason="not_configured", op="upsert_event_subscription") return await self._put( f"/v1/cloud/control/event-subscriptions/{subscription_id}", { "id": subscription_id, "workspace_id": workspace_id, "project_id": project_id, "connection_id": connection_id, "event": event, "event_version": event_version, "destination": destination, "url": url, "filter": filter_json, "ignore_origin": ignore_origin, "max_hops": max_hops, "status": status, "signing_secret": signing_secret, }, ) async def delete_event_subscription(self, subscription_id: str, project_id: str) -> None: if not self._is_configured(): logger.debug("cloud_control_skip", reason="not_configured", op="delete_event_subscription") return await self._delete( f"/v1/cloud/control/event-subscriptions/{subscription_id}?project_id={project_id}" ) async def delete_connector_connection(self, connection_id: str) -> None: if not self._is_configured(): logger.debug("cloud_control_skip", reason="not_configured", op="delete_connector_connection") return async with httpx.AsyncClient(timeout=_TIMEOUT) as client: resp = await client.delete( f"{self._url}/v1/cloud/control/connectors/connections/{connection_id}", headers=self._headers(), ) resp.raise_for_status() logger.info("cloud_connector_connection_deleted", connection_id=connection_id) async def upsert_connector_permission( self, connection_id: str, project_id: Optional[str], action_slug: str, mode: str, policy: Optional[dict] = None, ) -> None: if not self._is_configured(): logger.debug("cloud_control_skip", reason="not_configured", op="upsert_connector_permission") return await self._put( f"/v1/cloud/control/connectors/connections/{connection_id}/permissions/{action_slug}", { "connection_id": connection_id, "project_id": project_id, "action_slug": action_slug, "mode": mode, "policy": policy or {}, }, ) logger.info( "cloud_connector_permission_synced", connection_id=connection_id, action_slug=action_slug, mode=mode, ) async def upsert_env_vars( self, project_id: str, entries: list[dict], ) -> None: """Push project env vars to Cloud so the execution fabric can inject them into functions/queues. entries: list of {key, value, visibility, environment} - visibility "secret" → Worker Secret binding (text type, encrypted in transit) - visibility "public" → plain Worker env var - environment "all" | "production" | "preview" — Cloud uses this to match deploy env Cloud stores them per-project in D1 and injects them when dispatching edge functions or queue consumers belonging to this project. """ if not self._is_configured(): logger.debug("cloud_control_skip", reason="not_configured", op="upsert_env_vars") return await self._put( f"/v1/cloud/control/projects/{project_id}/env-vars", { "project_id": project_id, "env_vars": [ { "key": e["key"], "value": e["value"], "visibility": e.get("visibility", "secret"), "environment": e.get("environment", "all"), } for e in entries ], }, ) logger.info("cloud_env_vars_synced", project_id=project_id, count=len(entries)) async def push_email_templates(self, project_id: str, templates: list[dict]) -> None: """Push validated email-template overrides. Replaces the full set for this project — a template_key omitted here reverts to Cloud's platform default on next send, it is not left as a stale prior override. See src/domain/codegen/email_templates_manifest.py for the validation (protected placeholders, known template_keys) that must run before this is called. Cloud re-validates the payload and refuses to DELETE live rows if validation fails (unknown key / missing source placeholders). """ if not self._is_configured(): logger.debug("cloud_control_skip", reason="not_configured", op="push_email_templates") return await self._put( f"/v1/cloud/control/projects/{project_id}/email-templates", {"project_id": project_id, "templates": templates}, ) logger.info("cloud_email_templates_synced", project_id=project_id, count=len(templates)) async def list_email_templates(self, project_id: str) -> list[dict]: """Current per-project overrides only — a template_key absent here is using Cloud's platform default copy, not "unconfigured" in any error sense.""" if not self._is_configured(): return [] result = await self._get(f"/v1/cloud/control/projects/{project_id}/email-templates") templates = result.get("templates") return templates if isinstance(templates, list) else [] async def push_auth_config( self, project_id: str, config: dict[str, Any], ) -> bool: """Push auth configuration (providers, JWT policy, roles) to Cloud D1. Returns False when the push was skipped because Cloud isn't configured (dev/test) -- a deliberate no-op, not a failure -- so callers that track sync state don't record that as a confirmed push.""" if not self._is_configured(): logger.debug("cloud_control_skip", reason="not_configured", op="push_auth_config") return False await self._put( f"/v1/cloud/control/projects/{project_id}/auth-config", {"project_id": project_id, **config}, ) logger.info("cloud_auth_config_pushed", project_id=project_id) return True async def get_auth_config(self, project_id: str) -> dict[str, Any]: """Read the non-secret Cloud auth projection for readiness checks.""" if not self._is_configured(): raise ValueError("Cloud is not configured") return await self._get(f"/v1/cloud/control/projects/{project_id}/auth-config") async def get_auth_projection(self, project_id: str) -> dict[str, Any]: """Read non-secret observed auth projection flags from Cloud. This is intentionally metadata-only: the operator can distinguish missing config, public key, and AuthZ projection without reading any key material. """ if not self._is_configured(): raise ValueError("Cloud is not configured") return await self._get(f"/v1/cloud/control/projects/{project_id}/auth-projection") async def set_authz_v2( self, project_id: str, *, mode: str | None = None, enabled: bool | None = None, ) -> dict[str, Any]: """Set the per-project AuthZ-v2 rollout mode idempotently. ``enabled`` remains accepted for older callers, while new callers must preserve the three-way Core rollout state on the Cloud control plane. """ if not self._is_configured(): return {"ok": False, "error": "cloud_not_configured"} if mode is None: if enabled is None: raise ValueError("authz_rollout_mode_required") mode = "enabled" if enabled else "disabled" if mode not in {"disabled", "shadow", "enabled"}: raise ValueError("invalid_authz_rollout_mode") return await self._put( f"/v1/cloud/control/projects/{project_id}/authz-v2", { "project_id": project_id, "mode": mode, # Compatibility for older Cloud gateways that only understand # the boolean field; shadow remains fail-open there. "enabled": mode == "enabled", }, ) async def push_auth_keys( self, project_id: str, key_id: str, public_key_pem: str, private_key_pem: str, algorithm: str = "RS256", ) -> bool: """Push the active RS256 key pair to Cloud KV as JWKs. Returns ``False`` when the control plane is not configured. Required capability provisioning uses that signal to remain blocked rather than treating a local no-op as a successful projection. """ if not self._is_configured(): logger.debug("cloud_control_skip", reason="not_configured", op="push_auth_keys") return False if algorithm != "RS256": raise ValueError("P2HS Cloud project auth currently supports RS256 keys only") await self._put( f"/v1/cloud/control/projects/{project_id}/auth-keys", { "project_id": project_id, "key_id": key_id, "public_key_jwk": _rsa_public_jwk(public_key_pem, key_id=key_id), "private_key_jwk": _rsa_private_jwk(private_key_pem, key_id=key_id), }, ) logger.info("cloud_auth_keys_pushed", project_id=project_id, key_id=key_id) return True async def deploy_function( self, project_id: str, slug: str, payload: dict[str, Any], ) -> None: """Push a function deploy/update to Cloud D1 + trigger W4P script load.""" if not self._is_configured(): logger.debug("cloud_control_skip", reason="not_configured", op="deploy_function") return await self._put( f"/v1/cloud/control/projects/{project_id}/functions/{slug}", {"project_id": project_id, "slug": slug, **payload}, ) logger.info("cloud_function_deployed", project_id=project_id, slug=slug) async def delete_function(self, project_id: str, slug: str) -> None: """Remove a function from Cloud D1 (disables routing + W4P script).""" if not self._is_configured(): logger.debug("cloud_control_skip", reason="not_configured", op="delete_function") return async with httpx.AsyncClient(timeout=_TIMEOUT) as client: resp = await client.delete( f"{self._url}/v1/cloud/control/projects/{project_id}/functions/{slug}", headers=self._headers(), ) resp.raise_for_status() logger.info("cloud_function_deleted", project_id=project_id, slug=slug) async def push_function_token(self, project_id: str, token: str) -> None: """Store the project-scoped Edge Function runtime token in Cloud KV. Deployed functions authenticate their record/entity API calls with this token instead of the platform-wide internal control token, so a leaked or misbehaving function is confined to its own project. """ if not self._is_configured(): logger.debug("cloud_control_skip", reason="not_configured", op="push_function_token") return await self._put( f"/v1/cloud/control/projects/{project_id}/function-token", {"project_id": project_id, "token": token}, ) logger.info("cloud_function_token_pushed", project_id=project_id) async def delete_function_token(self, project_id: str) -> None: if not self._is_configured(): logger.debug("cloud_control_skip", reason="not_configured", op="delete_function_token") return async with httpx.AsyncClient(timeout=_TIMEOUT) as client: resp = await client.delete( f"{self._url}/v1/cloud/control/projects/{project_id}/function-token", headers=self._headers(), ) resp.raise_for_status() logger.info("cloud_function_token_deleted", project_id=project_id) async def push_storage_buckets( self, project_id: str, workspace_id: str, buckets: list[dict[str, Any]], ) -> None: """Push the full bucket list for a project to Cloud D1.""" if not self._is_configured(): logger.debug("cloud_control_skip", reason="not_configured", op="push_storage_buckets") return try: await self._put( f"/v1/cloud/control/projects/{project_id}/storage/buckets", { "project_id": project_id, "workspace_id": workspace_id, "buckets": buckets, }, ) except httpx.HTTPStatusError as exc: detail = "" try: detail = exc.response.text[:500] except Exception: pass logger.warning( "cloud_storage_buckets_push_failed", project_id=project_id, workspace_id=workspace_id, status_code=exc.response.status_code, detail=detail, ) raise logger.info("cloud_storage_buckets_pushed", project_id=project_id, count=len(buckets)) async def revoke_app_user( self, project_id: str, user_id: str, reason: str = "admin_revoke", ) -> None: """Write a revocation entry to Cloud KV for an app user.""" if not self._is_configured(): logger.debug("cloud_control_skip", reason="not_configured", op="revoke_app_user") return await self._post( f"/v1/cloud/control/projects/{project_id}/auth/revoke", {"project_id": project_id, "user_id": user_id, "reason": reason}, ) logger.info("cloud_app_user_revoked", project_id=project_id, user_id=user_id) async def upsert_app_user( self, project_id: str, *, email: str, role: str, email_verified: bool = True, user_metadata: Optional[dict[str, Any]] = None, ) -> dict[str, Any]: """Create or update a Cloud app user. Idempotent on (project, email).""" if not self._is_configured(): return {} payload = { "email": email, "role": role, "email_verified": email_verified, "user_metadata": user_metadata or {}, } try: return await self._post(f"/v1/projects/{project_id}/auth/users", payload) except httpx.HTTPStatusError as exc: if exc.response.status_code != 409: raise listed = await self._get(f"/v1/projects/{project_id}/auth/users") users = listed.get("users") if isinstance(listed, dict) else None match = next( ( u for u in (users or []) if str(u.get("email", "")).lower() == email.strip().lower() ), None, ) if not match or not match.get("id"): raise updated = await self._post( f"/v1/projects/{project_id}/auth/users/{match['id']}/role", {"role": role}, ) # Keep inherited-authority provenance in sync even when Cloud's # compatibility upsert path returns a conflict for an existing user. metadata = user_metadata or {} if metadata: await self._patch( f"/v1/projects/{project_id}/auth/users/{match['id']}", {"user_metadata": metadata, "email_verified": email_verified}, ) return updated async def preserve_legacy_app_user(self, project_id: str, user: dict, legacy: dict) -> dict: """Preserve legacy profile/ID/status through Cloud's authenticated user API.""" from urllib.parse import quote if not self._is_configured() or not user.get("id"): raise ValueError("Cloud identity must exist before preserving legacy fields") metadata = user.get("app_metadata") or {} if not isinstance(metadata, dict): raise ValueError("Invalid Cloud identity metadata") payload = { "app_metadata": {**metadata, "p2hs_legacy_user": legacy}, "status": "active" if legacy["is_active"] else "banned", } async with httpx.AsyncClient(timeout=_TIMEOUT) as client: response = await client.patch( f"{self._url}/v1/projects/{quote(project_id, safe='')}/auth/users/{quote(user['id'], safe='')}", headers=self._headers(), json=payload, ) response.raise_for_status() return response.json() async def list_app_users(self, project_id: str) -> list[dict[str, Any]]: """Read every identity page; malformed or incomplete reads are errors.""" from urllib.parse import urlencode if not self._is_configured(): raise ValueError("Cloud is not configured") users, seen_ids, seen_cursors = [], set(), set() cursor = None while True: params = {"limit": 100} if cursor: params["cursor"] = cursor listed = await self._get(f"/v1/projects/{project_id}/auth/users?{urlencode(params)}") if not isinstance(listed, dict): raise ValueError("Invalid Cloud user list") envelope = listed.get("data") if isinstance(listed.get("data"), dict) else listed rows = envelope.get("users", envelope.get("data")) if not isinstance(rows, list): raise ValueError("Invalid Cloud user list") for user in rows: if not isinstance(user, dict) or not user.get("id") or user["id"] in seen_ids: raise ValueError("Invalid Cloud user pagination") seen_ids.add(user["id"]) users.append(user) cursor = envelope.get("next_cursor") if not cursor: return users if not isinstance(cursor, str) or cursor in seen_cursors: raise ValueError("Invalid Cloud user pagination") seen_cursors.add(cursor) async def create_entity_records( self, project_id: str, entity_key: str, records: list[dict[str, Any]], ) -> list[dict[str, Any]]: """Bulk-create Cloud records. Each item must be {data, metadata?}.""" from urllib.parse import quote if not self._is_configured(): raise ValueError("Cloud is not configured") if not records: return [] created: list[dict[str, Any]] = [] for start in range(0, len(records), 100): chunk = records[start : start + 100] async with httpx.AsyncClient(timeout=30) as client: response = await client.post( ( f"{self._url}/v1/projects/{quote(project_id, safe='')}" f"/entities/{quote(entity_key, safe='')}/records" ), headers=self._headers(), json={"records": chunk}, ) response.raise_for_status() payload = response.json() rows = payload.get("records") if isinstance(payload, dict) else None if not isinstance(rows, list) or len(rows) != len(chunk): raise ValueError("Cloud did not create every imported record") for row in rows: if not isinstance(row, dict) or not row.get("id"): raise ValueError("Cloud did not return an imported record") created.extend(rows) return created async def list_entity_records(self, project_id: str, entity_key: str) -> list[dict[str, Any]]: """Read every record page for an entity. Incomplete reads are errors.""" from urllib.parse import quote, urlencode if not self._is_configured(): raise ValueError("Cloud is not configured") records, seen_ids, seen_cursors = [], set(), set() cursor = None while True: params = {"limit": 100, "status": "active"} if cursor: params["cursor"] = cursor listed = await self._get( f"/v1/projects/{quote(project_id, safe='')}/entities/" f"{quote(entity_key, safe='')}/records?{urlencode(params)}" ) if not isinstance(listed, dict): raise ValueError("Invalid Cloud record list") rows = listed.get("records", listed.get("data")) if not isinstance(rows, list): raise ValueError("Invalid Cloud record list") for row in rows: if not isinstance(row, dict) or not row.get("id") or row["id"] in seen_ids: raise ValueError("Invalid Cloud record pagination") seen_ids.add(row["id"]) records.append(row) cursor = listed.get("next_cursor") if not cursor: return records if not isinstance(cursor, str) or cursor in seen_cursors: raise ValueError("Invalid Cloud record pagination") seen_cursors.add(cursor) async def upsert_query_adapter( self, project_id: str, entity_key: str, storage_strategy: str = "p2hs_records", connection_id: Optional[str] = None, index_adapter: Optional[dict[str, Any]] = None, name: Optional[str] = None, ) -> None: """Push Core-authorized record/search adapter configuration to Cloud. The index endpoint and connector reference are policy outputs from Core; generated apps never receive this configuration. """ if not self._is_configured(): logger.debug("cloud_control_skip", reason="not_configured", op="upsert_query_adapter") return await self._put( f"/v1/cloud/control/projects/{project_id}/query-adapters/{entity_key}", { "project_id": project_id, "entity_key": entity_key, "name": name or entity_key, "storage_strategy": storage_strategy, "connection_id": connection_id, "index_adapter": index_adapter, }, ) async def delete_query_adapter(self, project_id: str, entity_key: str) -> None: if not self._is_configured(): logger.debug("cloud_control_skip", reason="not_configured", op="delete_query_adapter") return await self._delete(f"/v1/cloud/control/projects/{project_id}/query-adapters/{entity_key}") async def upsert_external_connection( self, project_id: str, connection_id: str, workspace_id: str, name: str, conn_type: str, status: str, readonly: bool, hyperdrive_id: Optional[str] = None, http_base_url: Optional[str] = None, http_auth_type: Optional[str] = None, max_rows_per_query: Optional[int] = None, ) -> None: """Push connection config (no credentials) to Cloud D1.""" if not self._is_configured(): logger.debug("cloud_control_skip", reason="not_configured", op="upsert_external_connection") return await self._put( f"/v1/cloud/control/projects/{project_id}/connections/{name}", { "connection_id": connection_id, "project_id": project_id, "workspace_id": workspace_id, "name": name, "type": conn_type, "status": status, "readonly": readonly, "hyperdrive_id": hyperdrive_id, "http_base_url": http_base_url, "http_auth_type": http_auth_type, "max_rows_per_query": max_rows_per_query, }, ) logger.info("cloud_external_connection_synced", project_id=project_id, name=name) async def delete_external_connection( self, project_id: str, name: str ) -> None: """Remove a connection from Cloud D1.""" if not self._is_configured(): logger.debug("cloud_control_skip", reason="not_configured", op="delete_external_connection") return async with httpx.AsyncClient(timeout=_TIMEOUT) as client: resp = await client.delete( f"{self._url}/v1/cloud/control/projects/{project_id}/connections/{name}", headers=self._headers(), ) resp.raise_for_status() logger.info("cloud_external_connection_deleted", project_id=project_id, name=name) async def trigger_connection_test( self, project_id: str, name: str ) -> dict: """Ask Cloud to run a SELECT 1 / connectivity test and return schema snapshot.""" if not self._is_configured(): logger.debug("cloud_control_skip", reason="not_configured", op="trigger_connection_test") return {"ok": False, "error": "cloud_not_configured"} result = await self._post( f"/v1/cloud/internal/connections/{project_id}/{name}/test", {"project_id": project_id, "name": name}, ) return result async def push_realtime_config( self, project_id: str, enabled: bool, max_concurrent_connections: int, history_enabled: bool, history_buffer_size: int, ) -> None: if not self._is_configured(): logger.debug("cloud_control_skip", reason="not_configured", op="push_realtime_config") return await self._put( f"/v1/cloud/control/projects/{project_id}/realtime/config", { "project_id": project_id, "enabled": enabled, "max_concurrent_connections": max_concurrent_connections, "history_enabled": history_enabled, "history_buffer_size": history_buffer_size, }, ) logger.info("cloud_realtime_config_pushed", project_id=project_id) async def upsert_realtime_channel( self, project_id: str, name: str, access_policy: str, rls_filter: bool, ) -> None: if not self._is_configured(): logger.debug("cloud_control_skip", reason="not_configured", op="upsert_realtime_channel") return await self._put( f"/v1/cloud/control/projects/{project_id}/realtime/channels/{name}", { "project_id": project_id, "name": name, "access_policy": access_policy, "rls_filter": rls_filter, }, ) logger.info("cloud_realtime_channel_synced", project_id=project_id, name=name) async def delete_realtime_channel(self, project_id: str, name: str) -> None: if not self._is_configured(): logger.debug("cloud_control_skip", reason="not_configured", op="delete_realtime_channel") return async with httpx.AsyncClient(timeout=_TIMEOUT) as client: resp = await client.delete( f"{self._url}/v1/cloud/control/projects/{project_id}/realtime/channels/{name}", headers=self._headers(), ) resp.raise_for_status() logger.info("cloud_realtime_channel_deleted", project_id=project_id, name=name) async def realtime_broadcast( self, project_id: str, channel: str, event: str, data: dict, workspace_id: Optional[str] = None, ) -> None: if not self._is_configured(): logger.debug("cloud_control_skip", reason="not_configured", op="realtime_broadcast") return await self._post( f"/v1/cloud/control/projects/{project_id}/realtime/broadcast", { "project_id": project_id, "workspace_id": workspace_id, "channel": channel, "event": event, "data": data, }, ) logger.info("cloud_realtime_broadcast", project_id=project_id, channel=channel) async def push_ai_config( self, project_id: str, embedding_provider: str, embedding_model: Optional[str], embedding_secret_ref: Optional[str], completion_providers: list, completion_secret_refs: dict, embed_configs: list[dict], ) -> None: """Push AI provider config + embed field configs to Cloud D1. No raw keys sent.""" if not self._is_configured(): logger.debug("cloud_control_skip", reason="not_configured", op="push_ai_config") return await self._put( f"/v1/cloud/control/projects/{project_id}/ai-config", { "project_id": project_id, "embedding_provider": embedding_provider, "embedding_model": embedding_model, "embedding_secret_ref": embedding_secret_ref, "completion_providers": completion_providers, "completion_secret_refs": completion_secret_refs, "embed_configs": embed_configs, }, ) logger.info("cloud_ai_config_pushed", project_id=project_id) async def push_voice_config( self, project_id: str, workspace_id: str, enabled: bool, max_concurrent_calls: int, default_agent_id: Optional[str] = None, billing_mode: str = "managed", credential_ref: Optional[str] = None, ) -> None: if not self._is_configured(): logger.debug("cloud_control_skip", reason="not_configured", op="push_voice_config") return await self._put( f"/v1/cloud/control/projects/{project_id}/voice/config", { "project_id": project_id, "workspace_id": workspace_id, "enabled": enabled, "max_concurrent_calls": max_concurrent_calls, "default_agent_id": default_agent_id, "billing_mode": billing_mode, "credential_ref": credential_ref, }, ) logger.info("cloud_voice_config_pushed", project_id=project_id) async def push_livechat_config( self, *, project_id: str, workspace_id: str, enabled: bool, provider: str, billing_mode: str, credential_ref: Optional[str], model: Optional[str], instructions: Optional[str], max_tokens: int, transcript_retention_days: int, retain_audio: bool, agent_id: Optional[str] = None, agent_version_id: Optional[str] = None, tools: Optional[list] = None, ) -> None: if not self._is_configured(): logger.debug("cloud_control_skip", reason="not_configured", op="push_livechat_config") return await self._put(f"/v1/cloud/control/projects/{project_id}/livechat/config", { "project_id": project_id, "workspace_id": workspace_id, "enabled": enabled, "provider": provider, "billing_mode": billing_mode, "credential_ref": credential_ref, "model": model, "instructions": instructions, "max_tokens": max_tokens, "transcript_retention_days": transcript_retention_days, "retain_audio": retain_audio, "agent_id": agent_id, "agent_version_id": agent_version_id, "tools": tools or [], }) logger.info("cloud_livechat_config_pushed", project_id=project_id) async def purge_livechat_transcripts(self, project_id: str) -> dict: if not self._is_configured(): logger.debug("cloud_control_skip", reason="not_configured", op="purge_livechat_transcripts") return {"ok": True, "deleted": 0, "remaining": 0} result = await self._delete(f"/v1/cloud/control/projects/{project_id}/livechat/transcripts") logger.info("cloud_livechat_transcripts_purged", project_id=project_id, deleted=result.get("deleted")) return result async def push_voice_agent(self, project_id: str, workspace_id: str, agent: dict) -> None: if not self._is_configured(): logger.debug("cloud_control_skip", reason="not_configured", op="push_voice_agent") return await self._put( f"/v1/cloud/control/projects/{project_id}/voice/agents/{agent['id']}", agent, ) logger.info("cloud_voice_agent_pushed", project_id=project_id, agent_id=agent.get("id")) async def push_voice_phone_number(self, project_id: str, workspace_id: str, number: dict) -> None: if not self._is_configured(): logger.debug("cloud_control_skip", reason="not_configured", op="push_voice_phone_number") return await self._put( f"/v1/cloud/control/projects/{project_id}/voice/numbers/{number['id']}", number, ) logger.info("cloud_voice_number_pushed", project_id=project_id, number_id=number.get("id")) async def push_vectorize_config( self, workspace_id: str, vectorize_index_id: str, index_name: str, dimensions: int, metric: str, ) -> None: """Push Vectorize index binding to Cloud so workers can query it.""" if not self._is_configured(): logger.debug("cloud_control_skip", reason="not_configured", op="push_vectorize_config") return await self._put( f"/v1/cloud/control/workspaces/{workspace_id}/vectorize-config", { "workspace_id": workspace_id, "vectorize_index_id": vectorize_index_id, "index_name": index_name, "dimensions": dimensions, "metric": metric, }, ) logger.info("cloud_vectorize_config_pushed", workspace_id=workspace_id) async def trigger_bulk_reembed( self, project_id: str, embed_configs: list[dict], ) -> None: """Ask Cloud to queue bulk re-embed jobs for all configured entity fields.""" if not self._is_configured(): logger.debug("cloud_control_skip", reason="not_configured", op="trigger_bulk_reembed") return await self._post( f"/v1/cloud/internal/projects/{project_id}/ai/reembed", {"project_id": project_id, "embed_configs": embed_configs}, ) logger.info("cloud_bulk_reembed_triggered", project_id=project_id) async def push_log_config( self, project_id: str, debug_enabled: bool, debug_expires_at: Optional[str], min_drain_level: str, retention_days: int, ) -> None: if not self._is_configured(): logger.debug("cloud_control_skip", reason="not_configured", op="push_log_config") return await self._put( f"/v1/cloud/control/projects/{project_id}/log-config", { "project_id": project_id, "debug_enabled": debug_enabled, "debug_expires_at": debug_expires_at, "min_drain_level": min_drain_level, "retention_days": retention_days, }, ) logger.info("cloud_log_config_pushed", project_id=project_id) async def upsert_log_drain( self, project_id: str, drain_id: str, destination_type: str, endpoint_url: str, secret_ref: Optional[str], min_level: str, sources: Optional[list], enabled: bool, ) -> None: if not self._is_configured(): logger.debug("cloud_control_skip", reason="not_configured", op="upsert_log_drain") return await self._put( f"/v1/cloud/control/projects/{project_id}/log-drains/{drain_id}", { "project_id": project_id, "drain_id": drain_id, "destination_type": destination_type, "endpoint_url": endpoint_url, "secret_ref": secret_ref, "min_level": min_level, "sources": sources, "enabled": enabled, }, ) logger.info("cloud_log_drain_synced", project_id=project_id, drain_id=drain_id) async def delete_log_drain(self, project_id: str, drain_id: str) -> None: if not self._is_configured(): logger.debug("cloud_control_skip", reason="not_configured", op="delete_log_drain") return async with httpx.AsyncClient(timeout=_TIMEOUT) as client: resp = await client.delete( f"{self._url}/v1/cloud/control/projects/{project_id}/log-drains/{drain_id}", headers=self._headers(), ) resp.raise_for_status() logger.info("cloud_log_drain_deleted", project_id=project_id, drain_id=drain_id) async def get_project_usage_snapshots( self, workspace_id: str, snapshot_date: str ) -> dict: """Fetch per-project usage snapshots from Cloud for a given date.""" if not self._is_configured(): return {"snapshots": []} result = await self._post( f"/v1/cloud/internal/workspaces/{workspace_id}/usage/snapshots", {"workspace_id": workspace_id, "snapshot_date": snapshot_date}, ) return result async def create_project_entity(self, project_id: str, entity_definition: dict) -> dict: """Create a record-engine entity for a project. Idempotent from the caller's perspective: an existing entity_key (409) is treated as already-provisioned rather than an error, since feature-pack installs must be safe to retry. """ if not self._is_configured(): return {} async with httpx.AsyncClient(timeout=_TIMEOUT) as client: resp = await client.post( f"{self._url}/v1/projects/{project_id}/entities", headers=self._headers(), json=entity_definition, ) if resp.status_code == 409: return {"already_exists": True} resp.raise_for_status() return resp.json() async def get_project_entity(self, project_id: str, entity_key: str) -> dict: """Read the full installed schema without flattening away policy/metadata.""" if not self._is_configured(): raise RuntimeError("Cloud schema is unavailable: Cloud is not configured") result = await self._get(f"/v1/projects/{project_id}/entities/{entity_key}") if not isinstance(result, dict) or not isinstance(result.get("entity"), dict): raise ValueError("Cloud returned an invalid entity") return result async def patch_project_entity( self, project_id: str, entity_key: str, payload: dict, *, expected_version: int | None = None, ) -> dict: """Additive schema update. Cloud replaces schema_json with the body provided. Callers must send the merged live+new field list. Never omit existing fields. """ if not self._is_configured(): raise RuntimeError("Cloud schema is unavailable: Cloud is not configured") headers = self._headers() if expected_version is not None: headers["If-Match"] = f'"{expected_version}"' async with httpx.AsyncClient(timeout=_TIMEOUT) as client: resp = await client.patch( f"{self._url}/v1/projects/{project_id}/entities/{entity_key}", headers=headers, json=payload, ) resp.raise_for_status() return resp.json() async def upload_application_file( self, *, project_id: str, filename: str, content_type: str, content_base64: str, ) -> dict[str, Any]: """Create a private platform application file on behalf of a form BFF.""" if not self._is_configured(): return {"ok": False, "error": "cloud_not_configured"} try: payload = await self._post( f"/v1/projects/{project_id}/application-files", { "filename": filename, "content_type": content_type, "content_base64": content_base64, }, ) return {"ok": True, **payload} except httpx.HTTPStatusError as exc: return {"ok": False, "error": f"cloud_http_{exc.response.status_code}"} except Exception: return {"ok": False, "error": "cloud_unavailable"} async def reconcile_project_entity_policy( self, project_id: str, entity_definition: dict[str, Any], ) -> dict[str, Any]: """Apply only the Core-owned policy portion of an existing entity. Existing fields are read and preserved before patching; this is a policy reconciliation path, never a destructive schema replacement. """ entity_key = str(entity_definition.get("entity_key") or "").strip() if not entity_key: raise ValueError("entity_key_required") live = await self.get_project_entity(project_id, entity_key) entity = live.get("entity") or {} live_schema = entity.get("schema") if not isinstance(live_schema, dict): raw_schema = entity.get("schema_json") if isinstance(raw_schema, str): import json live_schema = json.loads(raw_schema) else: live_schema = {} desired_schema = entity_definition.get("schema") or {} payload: dict[str, Any] = {} desired_policy = desired_schema.get("access_policy") if desired_policy is not None and live_schema.get("access_policy") != desired_policy: merged_schema = dict(live_schema) merged_schema["access_policy"] = desired_policy payload["schema"] = merged_schema for key in ("rls_enabled", "owner_managed", "owner_singleton"): if entity_definition.get(key) is True and entity.get(key) is not True: payload[key] = True if not payload: return {"changed": False, "entity_key": entity_key} version = entity.get("version") result = await self.patch_project_entity( project_id, entity_key, payload, expected_version=version if isinstance(version, int) else None, ) return {"changed": True, "entity_key": entity_key, "result": result} async def create_webhook_subscription( self, project_id: str, *, provider: str, forward_url: str, event_types: Optional[list[str]] = None, pack_slug: Optional[str] = None, ) -> Optional[dict[str, Any]]: """Provision a per-project inbound webhook route on pc-webhooks. Returns the created subscription's public fields (webhook_url, signing_secret, forward_secret, ...) for the caller to surface to codegen, or None if Cloud is unreachable/not configured — never raises, matching other best-effort provisioning calls. """ if not self._is_configured(): return None try: return await self._post( f"/v1/projects/{project_id}/webhook-subscriptions", { "provider": provider, "forward_url": forward_url, "event_types": event_types or [], "pack_slug": pack_slug, }, ) except httpx.HTTPStatusError as exc: logger.warning( "cloud_webhook_subscription_create_failed", project_id=project_id, provider=provider, status_code=exc.response.status_code, ) return None except Exception as exc: logger.warning( "cloud_webhook_subscription_create_error", project_id=project_id, provider=provider, error=str(exc), ) return None async def update_webhook_subscription( self, project_id: str, wh_token: str, *, signing_secret: str, signing_header: Optional[str] = None, ) -> Optional[dict[str, Any]]: """Supply the provider-generated signing secret for a subscription created with ``signing_secret=None`` (Stripe et al. — the owner only gets this back after creating the dashboard webhook endpoint using the ``webhook_url`` from ``create_webhook_subscription``). Returns the updated subscription's public fields, or None if Cloud is unreachable/not configured/the subscription doesn't exist — never raises. """ if not self._is_configured(): return None try: async with httpx.AsyncClient(timeout=_TIMEOUT) as client: resp = await client.patch( f"{self._url}/v1/projects/{project_id}/webhook-subscriptions/{wh_token}", headers=self._headers(), json={"signing_secret": signing_secret, "signing_header": signing_header}, ) resp.raise_for_status() return resp.json() except httpx.HTTPStatusError as exc: logger.warning( "cloud_webhook_subscription_update_failed", project_id=project_id, wh_token=wh_token, status_code=exc.response.status_code, ) return None except Exception as exc: logger.warning( "cloud_webhook_subscription_update_error", project_id=project_id, wh_token=wh_token, error=str(exc), ) return None async def deactivate_pack_webhook_subscriptions(self, project_id: str, pack_slug: str) -> bool: """Disable every webhook subscription a feature pack created, on pack uninstall. Never raises — best-effort, matching the other provisioning calls in this client. """ if not self._is_configured(): return False try: async with httpx.AsyncClient(timeout=_TIMEOUT) as client: resp = await client.delete( f"{self._url}/v1/projects/{project_id}/webhook-subscriptions/pack/{pack_slug}", headers=self._headers(), ) resp.raise_for_status() return True except Exception as exc: logger.warning( "cloud_pack_webhook_subscriptions_deactivate_failed", project_id=project_id, pack_slug=pack_slug, error=str(exc), ) return False async def get_project_schema(self, project_id: str) -> dict: """Fetch entity/field schema for a project from Cloud. Returns {"entities": [...]}. Unavailable or malformed schema is an error, never evidence that a project has no entities. """ if not self._is_configured(): raise RuntimeError("Cloud schema is unavailable: Cloud is not configured") result = await self._get(f"/v1/cloud/internal/projects/{project_id}/schema") if not isinstance(result, dict) or not isinstance(result.get("entities"), list): raise ValueError("Cloud returned an invalid project schema") from src.integrations.cloud_schema import normalize_cloud_entity entities = [] for raw in result["entities"]: try: entities.append(normalize_cloud_entity(raw)) except (TypeError, ValueError): logger.warning( "cloud_schema_entity_skipped", project_id=project_id, entity_key=(raw or {}).get("entity_key") if isinstance(raw, dict) else None, ) if result["entities"] and not entities: raise ValueError("Cloud returned entities that could not be normalized") return {**result, "entities": entities} async def query_project_records( self, project_id: str, entity_key: str, *, where: Optional[dict] = None, limit: int = 100, cursor: Optional[str] = None, ) -> dict: """One page of records for an entity, as the internal caller (sees every row regardless of owner_user_id — RLS only restricts app_user callers). Returns {"records": [...], "next_cursor": str | None}. """ if not self._is_configured(): raise RuntimeError("Cloud is not configured") body: dict[str, Any] = {"entity_key": entity_key, "limit": limit} if where is not None: body["where"] = where if cursor: body["cursor"] = cursor return await self._post(f"/v1/projects/{project_id}/records/query", body) async def set_record_owner( self, project_id: str, entity_key: str, record_id: str, owner_user_id: str ) -> dict: """Reassign a record's RLS owner — internal-caller-only on Cloud's side. For one-time backfills after enabling rls_enabled on an entity that already has data (see scripts/backfill_client_portal_ownership.py). """ if not self._is_configured(): raise RuntimeError("Cloud is not configured") return await self._patch( f"/v1/projects/{project_id}/entities/{entity_key}/records/{record_id}", {"owner_user_id": owner_user_id}, ) async def update_entity_record_data( self, project_id: str, entity_key: str, record_id: str, data: dict[str, Any] ) -> dict: """Merge-patch a record's `data` payload — internal-caller-only on Cloud's side. Used for idempotent upsert-by-identity flows (e.g. external entity ingest); callers must already have resolved which fields are safe to write (bounded field policy is a Core-side concern, not Cloud's).""" if not self._is_configured(): raise RuntimeError("Cloud is not configured") return await self._patch( f"/v1/projects/{project_id}/entities/{entity_key}/records/{record_id}", {"data": data}, ) async def invalidate_project_routing_cache(self, project_id: str) -> None: """Ask Cloud's ingress to drop its KV routing cache entry for this project. Called whenever Core changes a project's shard assignment so the ingress router stops serving the old shard mapping immediately instead of waiting for the TTL to expire. """ if not self._is_configured(): logger.debug( "cloud_control_skip", reason="not_configured", op="invalidate_project_routing_cache", ) return try: await self._post( f"/v1/cloud/control/projects/{project_id}/invalidate-routing-cache", {"project_id": project_id}, ) logger.info("project_routing_cache_invalidated", project_id=project_id) except Exception as exc: logger.warning( "project_routing_cache_invalidation_error", project_id=project_id, error=str(exc), ) async def bootstrap_project( self, workspace: dict, project: dict, clients: list[dict], capabilities: dict, integrations: Optional[list[dict]] = None, connectors: Optional[list[dict]] = None, ) -> None: """Single call that atomically upserts workspace + project + clients + capabilities + integrations + connector metadata. Preferred for initial project provisioning.""" if not self._is_configured(): logger.debug("cloud_control_skip", reason="not_configured", op="bootstrap_project") return project_id = project["project_id"] await self._put( f"/v1/cloud/control/bootstrap/project/{project_id}", { "workspace": workspace, "project": project, "clients": clients, "capabilities": capabilities, "integrations": integrations or [], "connectors": connectors or [], }, ) logger.info( "cloud_project_bootstrapped", project_id=project_id, workspace_id=workspace["workspace_id"], ) async def enqueue_control_job( self, *, project_id: str, workspace_id: str, client_id: str, request_type: str, idempotency_key: str, payload: dict[str, Any], authorization_id: str, grant_jwt: str, grant_expires_at: str, policy_snapshot: dict[str, Any], ) -> dict[str, Any]: """Enqueue a Core-pre-authorized async job (no browser handshake).""" if not self._is_configured(): return {"ok": False, "error": "cloud_not_configured"} try: return await self._post( f"/v1/cloud/control/projects/{project_id}/jobs", { "workspace_id": workspace_id, "project_id": project_id, "client_id": client_id, "request_type": request_type, "idempotency_key": idempotency_key, "payload": payload, "authorization_id": authorization_id, "grant": {"jwt": grant_jwt, "expires_at": grant_expires_at}, "policy_snapshot": policy_snapshot, }, ) except httpx.HTTPStatusError as exc: detail = f"cloud_http_{exc.response.status_code}" try: body = exc.response.json() detail = str(body.get("error") or body.get("message") or detail) except Exception: try: detail = exc.response.text[:500] or detail except Exception: pass logger.warning( "cloud_enqueue_job_failed", project_id=project_id, request_type=request_type, status_code=exc.response.status_code, detail=detail, ) return {"ok": False, "error": detail} except Exception as exc: logger.warning("cloud_enqueue_job_error", project_id=project_id, error=str(exc)) return {"ok": False, "error": "cloud_unavailable"} async def get_active_storage_file( self, *, project_id: str, file_id: str, ) -> dict[str, Any] | None: """Active [P] file by opaque id. Pending/purged/cross-project → None.""" if not self._is_configured(): return None path = f"/v1/cloud/control/projects/{project_id}/storage/files/{file_id}" try: data = await self._get(path) except httpx.HTTPStatusError as exc: if exc.response.status_code == 404: return None logger.warning( "cloud_get_active_storage_file_failed", project_id=project_id, file_id=file_id, status_code=exc.response.status_code, ) return None except Exception as exc: logger.warning( "cloud_get_active_storage_file_error", project_id=project_id, file_id=file_id, error=str(exc), ) return None if not data or data.get("ok") is False: return None if data.get("status") != "active": return None return data async def get_storage_file( self, *, project_id: str, bucket_name: str, file_id: str, ) -> dict[str, Any] | None: """Fetch a single active storage file (internal token).""" if not self._is_configured(): return None path = f"/v1/projects/{project_id}/storage/{bucket_name}/files/{file_id}" try: return await self._get(path) except httpx.HTTPStatusError as exc: if exc.response.status_code == 404: return None logger.warning( "cloud_get_storage_file_failed", project_id=project_id, file_id=file_id, status_code=exc.response.status_code, ) return None except Exception as exc: logger.warning( "cloud_get_storage_file_error", project_id=project_id, file_id=file_id, error=str(exc), ) return None async def list_storage_files( self, *, project_id: str, bucket_name: str, limit: int = 50, cursor: str | None = None, ) -> dict[str, Any]: """List active files in a project storage bucket (internal token).""" if not self._is_configured(): return {"files": [], "next_cursor": None} path = f"/v1/projects/{project_id}/storage/{bucket_name}/files?limit={min(limit, 100)}" if cursor: path += f"&cursor={cursor}" try: return await self._get(path) except httpx.HTTPStatusError as exc: if exc.response.status_code == 404: return {"files": [], "next_cursor": None} logger.warning( "cloud_list_storage_files_failed", project_id=project_id, bucket_name=bucket_name, status_code=exc.response.status_code, ) return {"files": [], "next_cursor": None} except Exception as exc: logger.warning( "cloud_list_storage_files_error", project_id=project_id, error=str(exc), ) return {"files": [], "next_cursor": None} async def presign_storage_file_download( self, *, project_id: str, bucket_name: str, file_id: str, expires_in: int = 86400, ) -> dict[str, Any]: """Presigned GET URL for a private-bucket file (internal token).""" if not self._is_configured(): return {"ok": False, "error": "cloud_not_configured"} path = ( f"/v1/projects/{project_id}/storage/{bucket_name}/files/{file_id}/presign-download" ) try: data = await self._post(path, {"expires_in": min(expires_in, 24 * 3600)}) return {"ok": True, **data} except httpx.HTTPStatusError as exc: if exc.response.status_code == 404: return {"ok": False, "error": "file_not_found"} logger.warning( "cloud_presign_download_failed", project_id=project_id, file_id=file_id, status_code=exc.response.status_code, ) return {"ok": False, "error": f"cloud_http_{exc.response.status_code}"} except Exception as exc: logger.warning( "cloud_presign_download_error", project_id=project_id, file_id=file_id, error=str(exc), ) return {"ok": False, "error": "cloud_unavailable"} async def delete_storage_file( self, *, project_id: str, bucket_name: str, file_id: str, ) -> dict[str, Any]: """Delete a storage file from Cloud D1 and R2 (internal token).""" if not self._is_configured(): return {"ok": False, "error": "cloud_not_configured"} path = f"/v1/projects/{project_id}/storage/{bucket_name}/files/{file_id}" try: return await self._delete(path) except httpx.HTTPStatusError as exc: if exc.response.status_code == 404: return {"ok": False, "error": "file_not_found"} logger.warning( "cloud_delete_storage_file_failed", project_id=project_id, file_id=file_id, status_code=exc.response.status_code, ) return {"ok": False, "error": f"cloud_http_{exc.response.status_code}"} except Exception as exc: logger.warning( "cloud_delete_storage_file_error", project_id=project_id, file_id=file_id, error=str(exc), ) return {"ok": False, "error": "cloud_unavailable"} async def get_cloud_client_for_workspace( workspace_id: str | UUID, db: "AsyncSession", ) -> "CloudControlClient": """Return a CloudControlClient routed to the Cloud instance assigned to *workspace_id*. Workflow: 1. Look up workspace_cloud_assignment in the DB. 2. If found, use that instance's gateway_url and decrypted internal_token. 3. If not found, fall back to the global cloud_control_api_url / cloud_internal_token settings (works for single-instance / dev setups). Auto-assignment is NOT performed here — if no assignment exists, callers that need to trigger auto-assignment should call CloudInfraService.auto_assign_workspace before calling this function. """ from src.domain.cloud_infra.repository import ( CloudInstanceRepository, WorkspaceCloudAssignmentRepository, ) from src.domain.cloud_infra.models import CloudInstanceStatus from src.core.security.encryption import decrypt_value ws_id = UUID(str(workspace_id)) assignment_repo = WorkspaceCloudAssignmentRepository(db) assignment = await assignment_repo.get_active_for_workspace(ws_id) if assignment: instance_repo = CloudInstanceRepository(db) instance = await instance_repo.get_by_id(assignment.cloud_instance_id) if instance and instance.status != CloudInstanceStatus.offline: token = decrypt_value(instance.internal_token_enc) return CloudControlClient( api_url=instance.gateway_url, token=token, ) logger.warning( "cloud_routing_fallback", workspace_id=str(ws_id), reason="assigned_instance_offline_or_missing", ) # Fall back to global config settings = get_settings() return CloudControlClient( api_url=settings.cloud_control_api_url, token=settings.cloud_internal_token, )