"""feature_packs domain — pack install orchestration. Consumes p2hs/feature-packs.json (via src/domain/codegen/feature_packs_manifest.py) the same way EdgeFunctionsService.deploy_from_library_by_slug consumes p2hs/functions.json: for each requested pack, deploy its composed library functions, provision its record-engine entities on Cloud, and record project_feature_packs install state. Non-blocking — failures are captured per pack/component and returned, never raised, so a codegen run never fails on pack install. """ from __future__ import annotations from datetime import datetime, timezone from typing import Any, Optional from uuid import UUID import structlog from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from src.domain.connectors.repository import ConnectorRepository from src.domain.edge_functions.connector_bindings import has_active_connector_connection from src.domain.edge_functions.service import EdgeFunctionsService from src.domain.feature_packs.config import merge_pack_config, union_notify_emails from src.domain.feature_packs.crm_authorization import ( CrmAuthorizationError, ensure_installed_policy, require_desired_policy, requires_crm_authorization, ) from src.domain.feature_packs.models import FeaturePack, ProjectFeaturePack from src.domain.feature_packs.normalize import load_pack_contract from src.domain.feature_packs.notify_emails import notify_emails_from_text from src.domain.feature_packs.ownership import ( OwnershipContractError, validate_composition_ownership_contracts, ) from src.integrations.cloud_control_client import CloudControlClient logger = structlog.get_logger() _SCHEMA_TYPE_MAP: dict[str, Any] = { "string": str, "number": (int, float), "integer": int, "boolean": bool, "object": dict, "array": list, } def _validate_against_schema(config: dict[str, Any], schema: Optional[dict[str, Any]]) -> list[str]: """Minimal required-keys/type check against a pack's ``input_schema``. Not a full JSON Schema implementation (no external dependency added for this) — covers the subset packs actually declare: top-level ``required`` and ``properties..type``. """ if not schema: return [] errors: list[str] = [] for key in schema.get("required") or []: if key not in config: errors.append(f"config.{key} is required") for key, spec in (schema.get("properties") or {}).items(): if key not in config or not isinstance(spec, dict): continue expected = _SCHEMA_TYPE_MAP.get(spec.get("type")) if expected and not isinstance(config[key], expected): errors.append(f"config.{key} must be of type {spec.get('type')}") return errors class PackInstallService: def __init__( self, db: AsyncSession, edge_functions_service: EdgeFunctionsService, cloud_client: Optional[CloudControlClient] = None, ): self._db = db self._edge_functions = edge_functions_service self._cloud_client = cloud_client or CloudControlClient() async def install_pack( self, *, project_id: UUID, pack_slug: str, config: dict[str, Any], user_prompt: str = "", ) -> dict[str, Any]: """Install (or re-apply) one feature pack. Idempotent. Incoming config is merged onto the last stored ProjectFeaturePack.config. Empty ``config={}`` from a later codegen run cannot wipe notify_to_email. """ pack = await self._get_pack_by_slug(pack_slug) if pack is None or pack.status != "active": errors = ["unknown_or_inactive_pack"] return { "pack_slug": pack_slug, "ok": False, "errors": errors, "status": "blocked", "blocker": {"errors": errors}, } composition = pack.composition or {} previous = await self._installed_config(project_id, pack.slug) config = merge_pack_config(previous, config) if pack.slug == "contact-capture": # Core-resolved form fields are authoritative over whatever the agent # freshly declared in p2hs/feature-packs.json this run -- Codegen must # not invent the operational schema (see project memory # project_capability_provisioner_architecture). Per form_key Core has # an opinion on, that opinion always wins, replacing any agent-fresh # or previously-persisted value for that key. # # For a form_key Core has NOT resolved yet, preserve whatever was # already persisted (``previous``) rather than clearing it: a legacy # project backfilling onto this mechanism, or one where extraction # simply hasn't run yet, must not lose a working configuration just # because Core has no replacement ready. This still never trusts a # *fresh* agent declaration for that key -- only something already # durably stored survives. resolved = await self._resolved_form_definitions(project_id) previous_form_definitions = previous.get("form_definitions") carried = ( dict(previous_form_definitions) if isinstance(previous_form_definitions, dict) else {} ) carried.update(resolved) if carried: config["form_definitions"] = carried else: config.pop("form_definitions", None) prompt_emails = notify_emails_from_text(user_prompt) existing = str(config.get("notify_to_email") or "").strip() if prompt_emails: config["notify_to_email"] = union_notify_emails(prompt_emails, existing) elif not existing: owner_email = await self._workspace_owner_email(project_id) if owner_email: config["notify_to_email"] = owner_email await self._persist_contact_notify_env(project_id, str(config.get("notify_to_email") or "")) hard_missing, missing_connections = await self._missing_prerequisites( project_id, pack.slug, composition ) if hard_missing: errors = [f"missing_prerequisite:{slug}" for slug in hard_missing] return { "pack_slug": pack_slug, "ok": False, "errors": errors, "status": "blocked", "blocker": {"errors": errors}, } # A missing connector connection is a runtime/deployment dependency, not a build # blocker: the pack's entities/functions/webhooks still get provisioned below, and # this is folded into `status="configuration_required"` at the end instead of # stopping the install. See composer.py's PrimitiveGap for the genuinely # build-blocking case (Cloud cannot express the capability at all) -- that one # still hard-stops; "we don't have credentials yet" never should. schema_errors = _validate_against_schema(config, pack.input_schema) if schema_errors: return { "pack_slug": pack_slug, "ok": False, "errors": schema_errors, "status": "blocked", "blocker": {"errors": schema_errors}, } try: validate_composition_ownership_contracts(composition) except OwnershipContractError as exc: return { "pack_slug": pack_slug, "ok": False, "errors": [str(exc)], "status": "blocked", "blocker": exc.as_blocker(), } try: crm_entity = requires_crm_authorization(pack.slug, pack.version) if crm_entity: require_desired_policy(composition, crm_entity) except CrmAuthorizationError as exc: await self._upsert_install_state( project_id=project_id, pack=pack, config=config, status="blocked", blocker=exc.as_blocker(), ) return {"pack_slug": pack_slug, "ok": False, "status": "blocked", "errors": [str(exc)], "blocker": exc.as_blocker()} errors: list[str] = [] authorization_blocker: dict | None = None for fn in composition.get("function_slugs") or []: lib_slug = fn if isinstance(fn, str) else fn.get("library_slug") if not lib_slug: continue proj_slug = ( f"{pack_slug}-{lib_slug}" if isinstance(fn, str) else fn.get("project_slug") or f"{pack_slug}-{lib_slug}" ) trigger_config = dict((fn.get("trigger_config") if isinstance(fn, dict) else None) or {}) static_env_vars = trigger_config.get("static_env_vars") if isinstance(static_env_vars, dict): # "$config." resolves against this install's own config dict # (e.g. contact-capture's notify_to_email) so a library function's # env vars can carry an owner-supplied value, not just connector # credentials. See EdgeFunctionsService._push_function. trigger_config["static_env_vars"] = { k: (config.get(v.removeprefix("$config."), "") if isinstance(v, str) and v.startswith("$config.") else v) for k, v in static_env_vars.items() } try: await self._edge_functions.deploy_from_library_by_slug( project_id=project_id, library_slug=lib_slug, project_slug=proj_slug, trigger_config=trigger_config, access_policy=(fn.get("access_policy") if isinstance(fn, dict) else None) or "authenticated", secret_refs=[], created_by="codegen", pack_slug=pack_slug, pack_version=pack.version, ) except Exception as exc: errors.append(f"function:{lib_slug}: {str(exc)[:200]}") logger.warning( "feature_pack_function_deploy_failed", project_id=str(project_id), pack_slug=pack_slug, library_slug=lib_slug, error=str(exc), ) for entity_def in composition.get("entities") or []: entity_key = entity_def.get("entity_key") if isinstance(entity_def, dict) else None if not entity_key: continue try: created = await self._cloud_client.create_project_entity(str(project_id), entity_def) if crm_entity == entity_key: await ensure_installed_policy( self._cloud_client, str(project_id), entity_key, allow_legacy_upgrade=( entity_key in {"lead", "catalog_item"} and bool(created.get("already_exists")) ), ) # A pack version bump can flip an entity-level access flag (rls_enabled, # owner_managed, owner_singleton) on an entity a prior install already # provisioned (e.g. client-portal 1.0.0 -> 1.1.0, business-directory # 1.0.0 -> 1.1.0). create_project_entity is create-only, so an # already-existing entity never picks that up on its own -- reconcile # explicitly rather than silently leaving old installs unprotected. if created.get("already_exists") and crm_entity != entity_key: reconcile: dict[str, Any] = {} if entity_def.get("rls_enabled") is True: reconcile["rls_enabled"] = True if entity_def.get("owner_managed") is True: reconcile["owner_managed"] = True if entity_def.get("owner_singleton") is True: reconcile["owner_singleton"] = True if reconcile: await self._cloud_client.patch_project_entity( str(project_id), entity_key, reconcile ) except Exception as exc: errors.append(f"entity:{entity_key}: {str(exc)[:200]}") if crm_entity == entity_key: authorization_blocker = ( exc.as_blocker() if isinstance(exc, CrmAuthorizationError) else CrmAuthorizationError(entity_key, "delivery or verification failed").as_blocker() ) logger.warning( "feature_pack_entity_provision_failed", project_id=str(project_id), pack_slug=pack_slug, entity_key=entity_key, error=str(exc), ) live_entities_by_key: dict[str, dict] = {} try: from src.domain.projects.cloud_schema_catalog import ( ProjectCloudSchemaSnapshotRepository, ) live = (await self._cloud_client.get_project_schema(str(project_id))).get("entities") or [] live_entities_by_key = { e["key"]: e for e in live if isinstance(e, dict) and e.get("key") } if live: await ProjectCloudSchemaSnapshotRepository(self._db).upsert(project_id, live) except Exception: logger.warning( "feature_pack_schema_snapshot_failed", project_id=str(project_id), pack_slug=pack_slug, ) # create_project_entity treats a 409 as "already exists" without comparing # schema — that is not proof a newer pack version's added fields actually # landed on an existing project. Verify against the live schema we just # fetched (fail closed if that fetch itself failed, same as everywhere else # in this codebase: an outage is never treated as evidence of success). failed_entity_keys = {e.split(":", 2)[1] for e in errors if e.startswith("entity:")} for entity_def in composition.get("entities") or []: entity_key = entity_def.get("entity_key") if isinstance(entity_def, dict) else None if not entity_key or entity_key in failed_entity_keys: continue declared_fields = (entity_def.get("schema") or {}).get("fields") or [] declared_keys = { f["key"] for f in declared_fields if isinstance(f, dict) and f.get("key") } live_entity = live_entities_by_key.get(entity_key) if live_entity is None: errors.append(f"entity:{entity_key}: could not verify against live schema") continue live_keys = { f["key"] for f in (live_entity.get("fields") or []) if isinstance(f, dict) and f.get("key") } missing = declared_keys - live_keys if missing: errors.append( f"entity:{entity_key}: fields not applied on Cloud: {', '.join(sorted(missing))}" ) webhook_setup: list[dict[str, Any]] = [] for webhook_def in composition.get("webhooks") or []: if not isinstance(webhook_def, dict): continue provider = webhook_def.get("provider") if not provider: continue url_config_key = webhook_def.get("forward_url_config_key") forward_url = None if url_config_key: candidate = config.get(url_config_key) if isinstance(candidate, str) and candidate: forward_url = candidate if not forward_url: legacy = config.get("webhook_forward_url") if isinstance(legacy, str) and legacy: forward_url = legacy if not forward_url: path = webhook_def.get("forward_path") origin = await self._project_app_origin(project_id) if isinstance(path, str) and path and origin: forward_url = origin.rstrip("/") + path if not forward_url: webhook_setup.append( { "provider": provider, "target": webhook_def.get("target"), "forward_path": webhook_def.get("forward_path"), "protected": bool(webhook_def.get("protected")), "deferred": "app_origin_unavailable", } ) continue result = await self._cloud_client.create_webhook_subscription( str(project_id), provider=provider, forward_url=forward_url, event_types=webhook_def.get("event_types") or [], pack_slug=pack_slug, ) if result is None: errors.append(f"webhook:{provider}: subscription_create_failed") continue entry = { "provider": provider, "wh_token": result.get("wh_token"), "webhook_url": result.get("webhook_url"), "setup_required": result.get("setup_required", False), } if result.get("setup_required"): entry["instructions"] = ( f"Create a webhook endpoint in the {provider} dashboard pointing at " f"{result.get('webhook_url')}, then supply the signing secret {provider} " "gives you back so P2HS can verify inbound events." ) webhook_setup.append(entry) # The app's own BFF needs `forward_secret` to trust the forwarded webhook.dispatch # call (checked against the `x-p2hs-webhook-secret` header Cloud sends). This must # reach the generated app's actual hosting runtime (CF Pages env), not just Cloud's # edge-function/queue dispatch env — CloudControlClient.upsert_env_vars only does # the latter, so the receiving BFF would never see it. Write through Core's own # project_env_vars custody and the same sync path every other app secret uses # (src.workers.tasks.env_sync.sync_project_env_vars), and block readiness if that # propagation can't even be enqueued. forward_secret = result.get("forward_secret") if forward_secret: try: from src.core.security.encryption import encrypt_value from src.domain.projects.repository import ProjectEnvVarRepository from src.workers.enqueue import enqueue_env_sync # Scoped by pack, not just provider — two packs both using stripe # webhooks must not overwrite each other's forwarding secret. env_key = "P2HS_WEBHOOK_SECRET_" + "_".join( part.upper() for part in (pack_slug, provider) if part ).replace("-", "_") await ProjectEnvVarRepository(self._db).upsert_by_key( project_id, env_key, encrypt_value(forward_secret), environment="all", visibility="secret", ) await self._db.flush() if not await enqueue_env_sync(str(project_id)): errors.append(f"webhook:{provider}: secret_propagation_enqueue_failed") except Exception as exc: errors.append(f"webhook:{provider}: secret_provision_failed") logger.warning( "feature_pack_webhook_secret_provision_failed", project_id=str(project_id), pack_slug=pack_slug, provider=provider, error=str(exc), ) status = "installed" blocker: Optional[dict[str, Any]] = None if errors: status = "blocked" blocker = {**(authorization_blocker or {}), "errors": errors} else: # Neither of these prevented building the feature -- both are runtime # dependencies the owner still needs to close, not install failures. Kept # separate keys in `blocker` so the dashboard/run message can name each one. pending: dict[str, Any] = {} webhooks_pending = [w for w in webhook_setup if w.get("setup_required")] if webhooks_pending: pending["webhooks_pending_setup"] = webhooks_pending if missing_connections: pending["pending_connections"] = missing_connections if pending: status = "configuration_required" blocker = pending await self._upsert_install_state( project_id=project_id, pack=pack, config={**config, **({"webhooks": webhook_setup} if webhook_setup else {})}, status=status, blocker=blocker, ) if status == "blocked": logger.error( "feature_pack_install_blocked", project_id=str(project_id), pack_slug=pack_slug, errors=errors or None, blocker=blocker, ) elif status == "configuration_required": logger.info( "feature_pack_install_configuration_required", project_id=str(project_id), pack_slug=pack_slug, blocker=blocker, ) return { "pack_slug": pack_slug, "ok": status != "blocked", "errors": errors, "status": status, "blocker": blocker, } async def complete_webhook_setup( self, *, project_id: UUID, pack_slug: str, provider: str, signing_secret: str, signing_header: Optional[str] = None, ) -> dict[str, Any]: """Supply the provider-generated signing secret for a webhook that install_pack left blocked (``setup_required``) because the provider only hands it back after the owner creates the dashboard endpoint. Clears the pack's blocker once every pending webhook for this pack has a secret. """ result = await self._db.execute( select(ProjectFeaturePack).where( ProjectFeaturePack.project_id == project_id, ProjectFeaturePack.pack_slug == pack_slug, ) ) row = result.scalar_one_or_none() if row is None: return {"ok": False, "error": "pack_not_installed"} webhooks = list((row.config or {}).get("webhooks") or []) entry = next((w for w in webhooks if w.get("provider") == provider), None) if entry is None or not entry.get("wh_token"): return {"ok": False, "error": "webhook_not_found"} if not entry.get("setup_required"): return {"ok": True, "already_complete": True} updated = await self._cloud_client.update_webhook_subscription( str(project_id), entry["wh_token"], signing_secret=signing_secret, signing_header=signing_header, ) if updated is None: return {"ok": False, "error": "cloud_update_failed"} entry["setup_required"] = False entry.pop("instructions", None) row.config = {**(row.config or {}), "webhooks": webhooks} still_pending = [w for w in webhooks if w.get("setup_required")] blocker = dict(row.blocker or {}) if still_pending: blocker["webhooks_pending_setup"] = still_pending else: blocker.pop("webhooks_pending_setup", None) row.blocker = blocker or None if not row.blocker: row.status = "installed" await self._db.flush() return {"ok": True, "pack_slug": pack_slug, "provider": provider, "status": row.status} async def uninstall_pack(self, *, project_id: UUID, pack_slug: str) -> dict[str, Any]: """Remove a pack's executable logic (functions, webhook subscriptions). Data-safe by default: entities and records the pack provisioned are never touched here — that's the project owner's data, not the pack's to delete. The install row is kept with status=uninstalled so config/history survive a later reinstall, mirroring how a WordPress plugin deactivation keeps its database tables until the owner explicitly asks to remove them. """ result = await self._db.execute( select(ProjectFeaturePack).where( ProjectFeaturePack.project_id == project_id, ProjectFeaturePack.pack_slug == pack_slug, ) ) row = result.scalar_one_or_none() if row is None: return {"pack_slug": pack_slug, "ok": True, "already_absent": True} functions = await self._edge_functions.list_functions_by_pack(project_id, pack_slug) errors: list[str] = [] for fn in functions: try: await self._edge_functions.delete_function(project_id, fn.slug) except Exception as exc: errors.append(f"function:{fn.slug}: {str(exc)[:200]}") logger.warning( "feature_pack_function_delete_failed", project_id=str(project_id), pack_slug=pack_slug, slug=fn.slug, error=str(exc), ) await self._cloud_client.deactivate_pack_webhook_subscriptions(str(project_id), pack_slug) row.status = "uninstalled" row.blocker = {"errors": errors} if errors else None await self._db.flush() return {"pack_slug": pack_slug, "ok": not errors, "errors": errors} async def _missing_prerequisites( self, project_id: UUID, pack_slug: str, composition: dict[str, Any] ) -> tuple[list[str], list[str]]: """Check pack and provider-binding prerequisites before installing. Returns (hard_missing, missing_connections). ``hard_missing`` prevents building the feature (missing dependency pack). ``missing_connections`` is a runtime/deployment gap: the pack still installs as ``configuration_required``. Provider names never live on architectural ``requires`` — those are typed primitives or feature packs. """ contract = load_pack_contract(pack_slug, composition) hard_missing: list[str] = [] missing_connections: list[str] = [] pack_slugs = contract.required_pack_slugs() if pack_slugs: result = await self._db.execute( select(ProjectFeaturePack.pack_slug).where( ProjectFeaturePack.project_id == project_id, ProjectFeaturePack.pack_slug.in_(pack_slugs), ProjectFeaturePack.status.in_( ("installed", "configuration_required") ), ) ) present_packs = {slug for (slug,) in result.all()} hard_missing += [ f"pack:{slug}" for slug in pack_slugs if slug not in present_packs ] connector_repo = ConnectorRepository(self._db) seen_missing: set[str] = set() for binding in contract.provider_bindings: if not binding.configuration_required or not binding.providers: continue connected = False for key in binding.providers: if await has_active_connector_connection( connector_repo, project_id, key ): connected = True break if not connected: for key in binding.providers: if key not in seen_missing: seen_missing.add(key) missing_connections.append(key) return hard_missing, missing_connections async def _persist_contact_notify_env(self, project_id: UUID, notify_to_email: str) -> None: """Put owner notify addresses on the generated app so /api/contact can email.send.""" value = (notify_to_email or "").strip() if not value: return try: from src.core.security.encryption import encrypt_value from src.domain.projects.repository import ProjectEnvVarRepository from src.workers.enqueue import enqueue_env_sync await ProjectEnvVarRepository(self._db).upsert_by_key( project_id, "NOTIFY_TO_EMAIL", encrypt_value(value), environment="all", visibility="secret", ) await self._db.flush() await enqueue_env_sync(str(project_id)) except Exception as exc: logger.warning( "contact_notify_env_persist_failed", project_id=str(project_id), error=str(exc), ) async def _installed_config(self, project_id: UUID, pack_slug: str) -> dict[str, Any]: result = await self._db.execute( select(ProjectFeaturePack.config).where( ProjectFeaturePack.project_id == project_id, ProjectFeaturePack.pack_slug == pack_slug, ) ) row = result.scalar_one_or_none() return dict(row) if isinstance(row, dict) else {} async def _resolved_form_definitions(self, project_id: UUID) -> dict[str, dict]: """Core's current resolved form field requirements, projected for direct persistence into ``contact-capture``'s config. Returns ``{}`` when the project has no customization captured yet (fields are added through later build prompts, not at install time) -- see ``src.domain.projects.form_requirement_sync``.""" from src.domain.projects.discovery_repository import DiscoveryRepository from src.domain.projects.discovery_schemas import DiscoverySpec from src.domain.projects.form_requirements import resolved_form_definitions_config project = await DiscoveryRepository(self._db).get(project_id) if project is None: return {} raw_spec = ((project.discovery_json or {}).get("result") or {}).get("spec") if not isinstance(raw_spec, dict): return {} spec = DiscoverySpec.model_validate(raw_spec) return resolved_form_definitions_config(spec.form_requirements) async def _workspace_owner_email(self, project_id: UUID) -> str: """Default contact notify destination: the workspace owner's account email.""" from src.domain.auth.models import User from src.domain.projects.models import Project from src.domain.workspaces.models import Workspace project = await self._db.get(Project, project_id) if project is None or project.workspace_id is None: return "" workspace = await self._db.get(Workspace, project.workspace_id) if workspace is None or workspace.owner_id is None: return "" owner = await self._db.get(User, workspace.owner_id) email = (getattr(owner, "email", None) or "").strip() return email async def _project_app_origin(self, project_id: UUID) -> str | None: from src.domain.hosting.hostnames import canonical_preview_url from src.domain.projects.models import Project project = await self._db.get(Project, project_id) ref = getattr(project, "project_ref", None) if project is not None else None if not ref: return None return canonical_preview_url(str(ref), "").rstrip("/") async def _get_pack_by_slug(self, slug: str) -> Optional[FeaturePack]: result = await self._db.execute(select(FeaturePack).where(FeaturePack.slug == slug)) return result.scalar_one_or_none() async def _upsert_install_state( self, *, project_id: UUID, pack: FeaturePack, config: dict[str, Any], status: str, blocker: Optional[dict[str, Any]], ) -> None: result = await self._db.execute( select(ProjectFeaturePack).where( ProjectFeaturePack.project_id == project_id, ProjectFeaturePack.pack_slug == pack.slug, ) ) row = result.scalar_one_or_none() now = datetime.now(timezone.utc) verified = status == "installed" if row is None: row = ProjectFeaturePack( project_id=project_id, pack_id=pack.id, pack_slug=pack.slug, # No prior verified version to fall back to on a first, partially-failed # install — record the attempted version; verified_at is the real signal. pack_version=pack.version, status=status, config=config, blocker=blocker, installed_at=now, verified_at=now if verified else None, ) self._db.add(row) else: row.pack_id = pack.id # Only advance pack_version once this run is actually verified — a failed or # blocked re-install must not claim the new version landed. if verified: row.pack_version = pack.version row.verified_at = now row.status = status row.config = {**(row.config or {}), **config} row.blocker = blocker if row.installed_at is None: row.installed_at = now await self._db.flush()