"""Manifest domain — assembly service. Builds the complete manifest document by gathering data from every Core domain. Always assembles fresh; no caching. """ from __future__ import annotations import hashlib import os import secrets from datetime import datetime, timedelta, timezone from typing import Any, Optional from uuid import UUID, uuid4 from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from src.core.exceptions import NotFoundError, ValidationError from src.domain.manifest.models import ManifestRun from src.domain.manifest.repository import ManifestRunRepository from src.domain.manifest.schemas import ( ManifestAISection, ManifestAuthSection, ManifestBackendSection, ManifestBffConventions, ManifestBucket, ManifestConnection, ManifestCredentialsSection, ManifestCustomChannel, ManifestEnvVarsSection, ManifestDatabaseSection, ManifestEmbedConfig, ManifestEntitlementsSection, ManifestEntity, ManifestFeaturePack, ManifestFeaturePacksSection, ManifestFunctionsSection, ManifestKnowledgeBaseSection, ManifestKnowledgeEntry, ManifestLibraryFunction, ManifestPlatformSDK, ManifestProjectFeaturePack, ManifestProjectFunction, ManifestProjectSection, ManifestRealtimeSection, ManifestResponse, ManifestRunResponse, ManifestSchedule, ManifestDataImportSection, ManifestEmailTemplatesSection, ManifestQueryDslSection, ManifestStorageSection, RunTokenResponse, ) _RUN_TTL_SECONDS = 3600 # Compact connector surface advertised even when the project has no connection yet. # Codegen should wire the feature path and tell the owner to connect credentials. _CONNECTOR_CATALOG_HINTS: dict[str, Any] = { "paypal": { "name": "PayPal", "category": "payments", "actions": ["paypal.create_order", "paypal.capture_order", "paypal.get_order", "paypal.refund"], "setup": "Connect PayPal in the dashboard before live capture. Sandbox is allowed for previews.", }, "twilio": { "name": "Twilio", "category": "communication", "actions": ["twilio.send_sms"], "setup": "Connect Twilio for SMS reminders. Generic transactional email uses email.send (platform or a connected Resend/Mailgun/SES adapter), not a vendor send_email slug.", }, "calendly": { "name": "Calendly", "category": "scheduling", "actions": [ "calendly.list_event_types", "calendly.create_scheduling_link", "calendly.get_event", "calendly.cancel_event", ], "setup": "Optional. Prefer Cloud bookings.reserve for first-party scheduling; use Calendly when the owner already hosts availability there.", }, "google_calendar": { "name": "Google Calendar", "category": "scheduling", "actions": ["google_calendar.create_event", "google_calendar.update_event", "google_calendar.delete_event"], "setup": "Optional calendar sync. Custom Cloud bookings remain the source of truth for reservable slots.", }, } _EMAIL_SEND_CAPABILITY = "email.send" def _capability_surfaces() -> dict[str, Any]: from src.domain.connectors.catalog import slugs_implementing from src.domain.connectors.invoke_capabilities import INVOKE_CAPABILITIES, capability_surface return { key: capability_surface(key, list(slugs_implementing(key))) for key in INVOKE_CAPABILITIES } def _email_send_capability_surface() -> dict[str, Any]: return _capability_surfaces()[_EMAIL_SEND_CAPABILITY] def _provider_action_escape_hatches() -> dict[str, Any]: from src.domain.connectors.catalog import get_action, slugs_implementing from src.domain.connectors.invoke_capabilities import ( INVOKE_CAPABILITIES, PROVIDER_ACTION_WHEN, is_implemented, ) surface: dict[str, Any] = {} for capability_id in INVOKE_CAPABILITIES: if not is_implemented(capability_id): continue for slug in slugs_implementing(capability_id): existing = surface.get(slug) implements = list(existing["implements"]) if existing else [] if capability_id not in implements: implements.append(capability_id) surface[slug] = { "role": "escape_hatch", "implements": implements, "when": PROVIDER_ACTION_WHEN.get( slug, "Only when the user explicitly requires this provider's API.", ), } for slug, when in PROVIDER_ACTION_WHEN.items(): if slug in surface: continue pair = get_action(slug) if pair is None: continue surface[slug] = { "role": "escape_hatch", "implements": list(pair[1].implements), "when": when, } return surface def _connector_action_implements(slug: str) -> tuple[str, ...]: from src.domain.connectors.catalog import implements_for_slug return implements_for_slug(slug) def _connectors_manifest_envelope( *, items: dict[str, Any], ) -> dict[str, Any]: return { "execution_contract": { "method": "p2hs.integrations.execute", "request_type": "integration.execute", "required_fields": ["connection_id", "action", "input"], "role": "provider-specific connector dispatch; not the default for queued capabilities such as email.send", }, "capabilities": _capability_surfaces(), "provider_actions": _provider_action_escape_hatches(), "catalog": _CONNECTOR_CATALOG_HINTS, "items": items, } class ManifestService: def __init__(self, db: AsyncSession, run_repo: ManifestRunRepository) -> None: self._db = db self._run_repo = run_repo # ── Public API ─────────────────────────────────────────────────────────── async def get_manifest( self, project_id: UUID, sections: Optional[list[str]] = None, compact: bool = False, ) -> ManifestResponse: return await self._assemble(project_id, sections=sections, compact=compact) async def get_manifest_diff( self, project_id: UUID, since_run_id: UUID ) -> dict[str, Any]: run = await self._run_repo.get_by_id(since_run_id) if not run or run.project_id != project_id: raise NotFoundError("Run not found") manifest = await self._assemble(project_id) current_hash = _hash_manifest(manifest) if run.manifest_snapshot_hash and run.manifest_snapshot_hash == current_hash: return { "since_run_id": str(since_run_id), "generated_at": datetime.now(timezone.utc), "changed_sections": [], "sections": {}, } manifest_dict = manifest.model_dump() skip = {"generated_at", "credentials", "backend", "bff_conventions", "platform_sdk"} changed_sections = [k for k in manifest_dict if k not in skip] return { "since_run_id": str(since_run_id), "generated_at": datetime.now(timezone.utc), "changed_sections": changed_sections, "sections": {k: manifest_dict[k] for k in changed_sections}, } async def start_run( self, project_id: UUID, user_id: Optional[UUID], requested_scopes: Optional[dict] = None, ) -> RunTokenResponse: plain_token = secrets.token_urlsafe(48) token_hash = hashlib.sha256(plain_token.encode()).hexdigest() manifest = await self._assemble(project_id) snapshot_hash = _hash_manifest(manifest) run = ManifestRun( id=uuid4(), project_id=project_id, initiated_by=user_id, status="active", requested_scopes=requested_scopes, token_hash=token_hash, expires_at=datetime.now(timezone.utc) + timedelta(seconds=_RUN_TTL_SECONDS), manifest_snapshot_hash=snapshot_hash, ) run = await self._run_repo.create(run) return RunTokenResponse( run_id=str(run.id), token=plain_token, expires_at=run.expires_at, manifest=manifest, ) async def complete_run( self, project_id: UUID, run_id: UUID, snapshot_hash: Optional[str] = None ) -> ManifestRunResponse: run = await self._get_run_for_project(project_id, run_id) if run.status not in ("active",): raise ValidationError(f"Run is already {run.status}") updated = await self._run_repo.complete(run_id, snapshot_hash) return ManifestRunResponse.model_validate(updated) async def fail_run( self, project_id: UUID, run_id: UUID, reason: str ) -> ManifestRunResponse: run = await self._get_run_for_project(project_id, run_id) if run.status not in ("active",): raise ValidationError(f"Run is already {run.status}") updated = await self._run_repo.fail(run_id, reason) return ManifestRunResponse.model_validate(updated) async def list_runs( self, project_id: UUID, limit: int = 20, offset: int = 0 ) -> list[ManifestRunResponse]: runs = await self._run_repo.list_for_project(project_id, limit=limit, offset=offset) return [ManifestRunResponse.model_validate(r) for r in runs] # ── Assembly ───────────────────────────────────────────────────────────── async def _assemble( self, project_id: UUID, sections: Optional[list[str]] = None, compact: bool = False, ) -> ManifestResponse: from src.domain.projects.models import Project from src.domain.workspaces.models import Workspace proj_row = await self._db.get(Project, project_id) if not proj_row: raise NotFoundError("Project not found") ws_row = await self._db.get(Workspace, proj_row.workspace_id) plan_slug = (ws_row.selected_plan_slug if ws_row else None) or "free" from src.config import get_settings settings = get_settings() gateway_url = await self._resolve_gateway_url( proj_row.workspace_id, fallback=settings.cloud_control_api_url or "https://api.p2hs.cloud" ) project_section = ManifestProjectSection( id=str(project_id), name=proj_row.name, workspace_id=str(proj_row.workspace_id), plan=plan_slug, cloud_gateway_url=gateway_url, ) from src.domain.cloud.facades import project_port_context from src.integrations.platform_ports import api_surface port_context = await project_port_context(self._db, proj_row) ports = port_context["ports"] surface = api_surface(ports) sdk = ManifestPlatformSDK( version=port_context["catalog_version"], available={group: True for group in surface}, ) conventions = ManifestBffConventions( catalog_version=port_context["catalog_version"], ports=ports, records=surface.get("records", {}), auth_users={ key: value for key, value in surface.get("auth", {}).items() if key.startswith("users_") or key.startswith("password_reset") }, ) now = datetime.now(timezone.utc) return ManifestResponse( generated_at=now, project=project_section, credentials=ManifestCredentialsSection(), backend=ManifestBackendSection(), env_vars=await self._build_env_vars(project_id), database=await self._build_database(project_id, compact), auth=await self._build_auth(project_id), storage=await self._build_storage(project_id), functions=await self._build_functions(project_id, compact), feature_packs=await self._build_feature_packs(project_id, compact), connectors=await self._build_connectors(project_id, compact), integrations=await self._build_integrations(project_id), connections=await self._build_connections(project_id, compact), realtime=await self._build_realtime(project_id), ai=await self._build_ai(project_id), schedules=await self._build_schedules(project_id), knowledge_base=await self._build_knowledge_base(project_id), api_surface=surface, bff_conventions=conventions, platform_sdk=sdk, entitlements=await self._build_entitlements(project_id, plan_slug), query_dsl=self._build_query_dsl(), data_import=ManifestDataImportSection( enabled=bool(surface.get("imports")), async_import=surface.get("imports", {}), ), email_templates=await self._build_email_templates(project_id, compact), ) # ── Section builders ───────────────────────────────────────────────────── async def _build_env_vars(self, project_id: UUID) -> ManifestEnvVarsSection: from src.domain.projects.models import ProjectEnvVar from sqlalchemy import select result = await self._db.execute( select(ProjectEnvVar.key, ProjectEnvVar.visibility).where( ProjectEnvVar.project_id == project_id ) ) rows = result.all() secrets: list[str] = [] public: list[str] = [] for key, visibility in rows: if visibility == "public": public.append(key) else: secrets.append(key) return ManifestEnvVarsSection( secrets=sorted(set(secrets)), public=sorted(set(public)), ) async def _build_database( self, project_id: UUID, compact: bool ) -> ManifestDatabaseSection: if compact: return ManifestDatabaseSection(schema_status="omitted", schema_source="none") from src.domain.codegen.entity_catalog import compact_entities, resolve_database_view from src.domain.projects.cloud_schema_catalog import ( ProjectCloudSchemaSnapshotRepository, ) from src.domain.projects.repository import ProjectRepository from src.integrations.cloud_control_client import ( CloudControlClient, get_cloud_client_for_workspace, ) snapshot_repo = ProjectCloudSchemaSnapshotRepository(self._db) cached = await snapshot_repo.get_entities(project_id) live_status = "error" live_entities: list = [] try: project = await ProjectRepository(self._db).get_by_id(project_id) if project: try: client = await get_cloud_client_for_workspace(project.workspace_id, self._db) except Exception: client = CloudControlClient() raw = await client.get_project_schema(str(project_id)) live_entities = compact_entities(raw.get("entities") or []) live_status = "ok" except Exception: live_status = "error" view = resolve_database_view( live_status=live_status, live_entities=live_entities, cached_entities=cached, ) if view["persist"] is not None: try: await snapshot_repo.upsert(project_id, view["persist"]) except Exception: pass entities = None if view["entities"] is not None: entities = [ ManifestEntity( key=ent["key"], label=ent.get("label") or ent["key"], fields=[ { "key": f.get("key", ""), "label": f.get("label", f.get("key", "")), "type": f.get("type", "text"), "required": f.get("required", False), "unique": f.get("unique", False), "options": f.get("options"), "relation_entity_key": f.get("relation_entity_key"), } for f in ent.get("fields") or [] ], rls_enabled=ent.get("rls_enabled", False), record_count=ent.get("record_count"), status=ent.get("status") or "active", ) for ent in view["entities"] ] return ManifestDatabaseSection( schema_status=view["schema_status"], schema_source=view["schema_source"], entities=entities, ) async def _build_auth(self, project_id: UUID) -> ManifestAuthSection: from src.domain.project_auth.models import ProjectAuthConfig, ProjectAuthRole from sqlalchemy import select staff = await self._workspace_staff_directory(project_id) owner = next((item for item in staff if item.get("app_role") == "super_admin"), None) cfg_result = await self._db.execute( select(ProjectAuthConfig).where(ProjectAuthConfig.project_id == project_id) ) cfg = cfg_result.scalar_one_or_none() if not cfg or not cfg.enabled: return ManifestAuthSection( enabled=False, workspace_owner_email=(owner or {}).get("email"), workspace_owner_name=(owner or {}).get("name"), staff_from_workspace=staff, ) from src.domain.project_auth.repository import ( ProjectAuthConfigRepository, ProjectAuthKeyRepository, ProjectAuthRoleRepository, ) from src.domain.project_auth.service import ( CUSTOMER_MEMBER_ROLE, STAFF_ADMIN_ROLE, SUPER_ADMIN_ROLE, ProjectAuthService, ) from src.domain.projects.repository import ProjectRepository from src.integrations.cloud_control_client import ( CloudControlClient, get_cloud_client_for_workspace, ) try: project_row = await ProjectRepository(self._db).get_by_id(project_id) try: cloud = ( await get_cloud_client_for_workspace(project_row.workspace_id, self._db) if project_row else CloudControlClient() ) except Exception: cloud = CloudControlClient() await ProjectAuthService( config_repo=ProjectAuthConfigRepository(self._db), key_repo=ProjectAuthKeyRepository(self._db), role_repo=ProjectAuthRoleRepository(self._db), cloud_client=cloud, ).ensure_staff_roles_and_owner_admin(project_id) except Exception: pass roles_result = await self._db.execute( select(ProjectAuthRole).where(ProjectAuthRole.project_id == project_id) ) roles = list(roles_result.scalars().all()) role_names = [r.name for r in roles] return ManifestAuthSection( enabled=True, providers=list(cfg.providers or []), allow_signup=cfg.allow_signup, custom_roles=role_names, google_oauth_mode=cfg.google_oauth_mode, super_admin_role=SUPER_ADMIN_ROLE, admin_role=STAFF_ADMIN_ROLE, customer_role=CUSTOMER_MEMBER_ROLE, owner_provisioned_as_super_admin=SUPER_ADMIN_ROLE in role_names, owner_provisioned_as_admin=SUPER_ADMIN_ROLE in role_names or STAFF_ADMIN_ROLE in role_names, workspace_owner_email=(owner or {}).get("email"), workspace_owner_name=(owner or {}).get("name"), staff_from_workspace=staff, ) async def _workspace_staff_directory(self, project_id: UUID) -> list[dict[str, Any]]: from src.domain.project_auth.staff_directory import identities_for_workspace from src.domain.projects.models import Project from src.domain.workspaces.models import Workspace project = await self._db.get(Project, project_id) if not project: return [] workspace = await self._db.get(Workspace, project.workspace_id) if not workspace: return [] try: identities = await identities_for_workspace( self._db, project.workspace_id, workspace.owner_id ) except Exception: return [] return [ { "email": item.email, "name": item.name, "workspace_role": item.workspace_role, "app_role": item.app_role, } for item in identities ] async def _build_storage(self, project_id: UUID) -> ManifestStorageSection: from src.domain.project_storage.models import ProjectStorageBucket from sqlalchemy import select result = await self._db.execute( select(ProjectStorageBucket).where( ProjectStorageBucket.project_id == project_id, ProjectStorageBucket.is_active == True, # noqa: E712 ) ) buckets = list(result.scalars().all()) return ManifestStorageSection( buckets=[ ManifestBucket( name=b.name, label=b.name.replace("-", " ").replace("_", " ").title(), access_policy=b.access_policy, rls_enabled=b.rls_enabled, cdn_url=None, max_file_size_bytes=b.max_file_size_bytes, ) for b in buckets ] ) async def _build_functions( self, project_id: UUID, compact: bool ) -> ManifestFunctionsSection: from src.domain.edge_functions.models import FunctionLibrary, ProjectFunction from sqlalchemy import select proj_result = await self._db.execute( select(ProjectFunction).where( ProjectFunction.project_id == project_id, ProjectFunction.status == "active", ProjectFunction.deleted_at == None, # noqa: E711 ) ) proj_fns = list(proj_result.scalars().all()) system_fns: list[ManifestLibraryFunction] = [] if not compact: lib_result = await self._db.execute( select(FunctionLibrary).where(FunctionLibrary.is_active == True) # noqa: E712 ) lib_fns = list(lib_result.scalars().all()) system_fns = [ ManifestLibraryFunction( id=str(fn.id), slug=fn.slug, name=fn.name, tags=list(fn.tags or []), requires=list(fn.required_integrations or []), status="available", description=fn.description, version=fn.version, trigger_type=fn.trigger_type, input_schema=fn.input_schema, output_schema=fn.output_schema, route_examples=(fn.codegen_hints or {}).get("generated_route_examples"), ) for fn in lib_fns ] project_fns = [ ManifestProjectFunction( slug=fn.slug, name=fn.name, trigger=fn.trigger_type, event=(fn.trigger_config or {}).get("event_type"), cron=(fn.trigger_config or {}).get("cron"), status=fn.status, ) for fn in proj_fns ] return ManifestFunctionsSection(system=system_fns, project=project_fns) async def _build_feature_packs( self, project_id: UUID, compact: bool ) -> ManifestFeaturePacksSection: from src.domain.feature_packs.composition import composed_library_slugs from src.domain.feature_packs.models import FeaturePack, ProjectFeaturePack from src.domain.feature_packs.normalize import load_pack_contract from sqlalchemy import select installed_result = await self._db.execute( select(ProjectFeaturePack).where(ProjectFeaturePack.project_id == project_id) ) installed_rows = list(installed_result.scalars().all()) catalog: list[ManifestFeaturePack] = [] if not compact: catalog_result = await self._db.execute( select(FeaturePack).where(FeaturePack.status == "active") ) for pack in catalog_result.scalars().all(): contract = load_pack_contract( pack.slug, pack.composition, access_policy=pack.access_policy, implementation_version=pack.version, ) catalog.append( ManifestFeaturePack( id=str(pack.id), slug=pack.slug, name=pack.name, description=pack.description, version=pack.version, tags=list(pack.tags or []), composes=composed_library_slugs(pack.composition), requires=[ f"{dep.kind}:{dep.id}" for dep in contract.requires ], provides=list(contract.provide_ids()), events=[event.id for event in contract.events], provider_bindings=[ item.model_dump() for item in contract.provider_bindings ], status="available", input_schema=pack.input_schema, ) ) installed = [ ManifestProjectFeaturePack( slug=row.pack_slug, version=row.pack_version, status=row.status, config=row.config or {}, blocker=row.blocker, ) for row in installed_rows ] return ManifestFeaturePacksSection(catalog=catalog, installed=installed) async def _build_connectors(self, project_id: UUID, compact: bool) -> dict[str, Any]: from src.domain.connectors.models import ( Connector, ConnectorAction, ConnectorConnection, ConnectorPermission, ) from src.domain.connectors.invoke_capabilities import is_implemented from sqlalchemy import select conn_result = await self._db.execute( select(ConnectorConnection).where(ConnectorConnection.project_id == project_id) ) connections = list(conn_result.scalars().all()) if not connections: return _connectors_manifest_envelope(items={}) connector_ids = {row.connector_id for row in connections} connector_result = await self._db.execute( select(Connector).where(Connector.id.in_(connector_ids)) ) connectors_by_id = {row.id: row for row in connector_result.scalars().all()} action_result = await self._db.execute( select(ConnectorAction).where(ConnectorAction.connector_id.in_(connector_ids)) ) actions_by_connector: dict[UUID, list[ConnectorAction]] = {} for action in action_result.scalars().all(): actions_by_connector.setdefault(action.connector_id, []).append(action) permission_result = await self._db.execute( select(ConnectorPermission).where( ConnectorPermission.connection_id.in_([row.id for row in connections]) ) ) permissions_by_connection: dict[UUID, dict[str, ConnectorPermission]] = {} for permission in permission_result.scalars().all(): permissions_by_connection.setdefault(permission.connection_id, {})[ permission.action_slug ] = permission items: dict[str, Any] = {} for row in connections: connector = connectors_by_id.get(row.connector_id) if not connector: continue connector_entry = items.setdefault( connector.key, { "enabled": False, "connections": {}, "actions": [], "provider_actions": [], }, ) connector_entry["enabled"] = connector_entry["enabled"] or row.status == "active" permissions = permissions_by_connection.get(row.id, {}) actions = actions_by_connector.get(row.connector_id, []) default_actions: list[dict[str, Any]] = [] escape_actions: list[dict[str, Any]] = [] secret_fields = [] metadata = connector.metadata_json or {} if isinstance(metadata.get("secret_fields"), list): secret_fields = [str(x) for x in metadata["secret_fields"]] for action in actions: permission = permissions.get(action.slug) permission_mode = permission.mode if permission else "deny" missing_scopes = sorted(set(action.required_scopes or []) - set(row.scopes or [])) has_required_credentials = not secret_fields or bool(row.secrets_enc) requires_approval = permission_mode == "ask" unavailable_reason = None if permission_mode == "deny": unavailable_reason = "denied" elif not has_required_credentials: unavailable_reason = "credentials_missing" elif missing_scopes: unavailable_reason = "scope_missing" elif requires_approval: unavailable_reason = "approval_required" implements = list(_connector_action_implements(action.slug)) entry = { "slug": action.slug, "key": action.action_key, "permission": permission_mode, "requires_approval": requires_approval, "available": unavailable_reason is None, "unavailable_reason": unavailable_reason, "execution_mode": action.execution_mode, "required_scopes": list(action.required_scopes or []), "required_capabilities": list(action.required_capabilities or []), "implements": implements, "input_schema": {} if compact else action.input_schema, "output_schema": {} if compact else action.output_schema, } # Demote only when the backed capability is already the Codegen default # (email.send). implements alone does not hide slack.post_message. if implements and any(is_implemented(cap) for cap in implements): entry["role"] = "escape_hatch" escape_actions.append(entry) else: default_actions.append(entry) connector_entry["connections"][row.name] = { "connection_id": str(row.id), "status": row.status, "has_credentials": bool(row.secrets_enc), "scopes": list(row.scopes or []), "actions": default_actions, "provider_actions": escape_actions, } connector_entry["actions"] = sorted({a["slug"] for a in default_actions}) connector_entry["provider_actions"] = sorted( set(connector_entry["provider_actions"]) | {a["slug"] for a in escape_actions} ) return _connectors_manifest_envelope(items=items) async def _build_integrations(self, project_id: UUID) -> dict[str, Any]: from src.domain.integrations.models import ProjectIntegration from src.domain.edge_functions.models import ProjectWebhookEndpoint from sqlalchemy import select int_result = await self._db.execute( select(ProjectIntegration).where(ProjectIntegration.project_id == project_id) ) integrations = list(int_result.scalars().all()) wh_result = await self._db.execute( select(ProjectWebhookEndpoint).where( ProjectWebhookEndpoint.project_id == project_id, ProjectWebhookEndpoint.is_active == True, # noqa: E712 ) ) webhooks = {wh.provider: wh for wh in wh_result.scalars().all()} result: dict[str, Any] = {} for integration in integrations: cfg = integration.config_json or {} entry: dict[str, Any] = { "enabled": integration.status == "active", } if integration.status != "active": entry["setup_required"] = True if integration.provider in webhooks: wh = webhooks[integration.provider] entry["webhook_url"] = f"wh/{wh.wh_token}" if cfg.get("capabilities"): entry["capabilities"] = cfg["capabilities"] result[integration.provider] = entry return result async def _build_connections( self, project_id: UUID, compact: bool ) -> dict[str, Any]: from src.domain.external_connections.models import ProjectExternalConnection from sqlalchemy import select result = await self._db.execute( select(ProjectExternalConnection).where( ProjectExternalConnection.project_id == project_id ) ) connections = list(result.scalars().all()) out: dict[str, Any] = {} for conn in connections: entry: dict[str, Any] = { "type": conn.type, "status": conn.status, "readonly": conn.readonly, } if not compact and conn.schema_snapshot_json: entry["schema"] = conn.schema_snapshot_json out[conn.name] = entry return out async def _build_realtime(self, project_id: UUID) -> ManifestRealtimeSection: from src.domain.realtime.models import ProjectRealtimeChannel, ProjectRealtimeConfig from sqlalchemy import select cfg_result = await self._db.execute( select(ProjectRealtimeConfig).where( ProjectRealtimeConfig.project_id == project_id ) ) cfg = cfg_result.scalar_one_or_none() channels_result = await self._db.execute( select(ProjectRealtimeChannel).where( ProjectRealtimeChannel.project_id == project_id ) ) channels = list(channels_result.scalars().all()) return ManifestRealtimeSection( built_in_channels=["records", "jobs", "auth", "storage"], custom_channels=[ ManifestCustomChannel(name=ch.name, access_policy=ch.access_policy) for ch in channels ], ) async def _build_ai(self, project_id: UUID) -> ManifestAISection: from src.domain.ai.models import ProjectAIConfig, ProjectAIEmbedConfig from sqlalchemy import select cfg_result = await self._db.execute( select(ProjectAIConfig).where(ProjectAIConfig.project_id == project_id) ) cfg = cfg_result.scalar_one_or_none() if not cfg: return ManifestAISection(enabled=False) embed_result = await self._db.execute( select(ProjectAIEmbedConfig).where( ProjectAIEmbedConfig.project_id == project_id ) ) embed_cfgs = list(embed_result.scalars().all()) providers = list(cfg.completion_providers_json or []) if not providers: providers = ["cf_workers_ai"] return ManifestAISection( enabled=True, embedding_provider=cfg.embedding_provider, available_models=providers, embed_configs=[ ManifestEmbedConfig( entity=ec.entity_key, field=ec.field_key, auto_embed=ec.auto_embed, ) for ec in embed_cfgs ], ) async def _build_email_templates( self, project_id: UUID, compact: bool ) -> ManifestEmailTemplatesSection: from src.domain.codegen.email_templates_manifest import EMAIL_TEMPLATE_CATALOG from src.domain.projects.repository import ProjectRepository from src.integrations.cloud_control_client import ( CloudControlClient, get_cloud_client_for_workspace, ) if compact: return ManifestEmailTemplatesSection(catalog=EMAIL_TEMPLATE_CATALOG, current=[]) current: list[dict] = [] try: project = await ProjectRepository(self._db).get_by_id(project_id) if project: try: client = await get_cloud_client_for_workspace(project.workspace_id, self._db) except Exception: client = CloudControlClient() current = await client.list_email_templates(str(project_id)) except Exception: current = [] return ManifestEmailTemplatesSection(catalog=EMAIL_TEMPLATE_CATALOG, current=current) async def _build_schedules(self, project_id: UUID) -> list[ManifestSchedule]: from src.domain.edge_functions.models import ProjectFunction from sqlalchemy import select result = await self._db.execute( select(ProjectFunction).where( ProjectFunction.project_id == project_id, ProjectFunction.trigger_type == "schedule", ProjectFunction.status == "active", ProjectFunction.deleted_at == None, # noqa: E711 ) ) fns = list(result.scalars().all()) return [ ManifestSchedule( name=fn.name, cron=(fn.trigger_config or {}).get("cron", ""), fn_slug=fn.slug, status=fn.status, ) for fn in fns if (fn.trigger_config or {}).get("cron") ] async def _build_knowledge_base(self, project_id: UUID) -> ManifestKnowledgeBaseSection: from src.domain.instructions.models import ( ProjectInstruction, ProjectInstructionVersion, ) from sqlalchemy import select instr_result = await self._db.execute( select(ProjectInstruction).where(ProjectInstruction.project_id == project_id) ) instr = instr_result.scalar_one_or_none() if not instr or not instr.current_version_id: return ManifestKnowledgeBaseSection() ver_result = await self._db.execute( select(ProjectInstructionVersion).where( ProjectInstructionVersion.id == instr.current_version_id ) ) version = ver_result.scalar_one_or_none() if not version: return ManifestKnowledgeBaseSection() return ManifestKnowledgeBaseSection( entries=[ ManifestKnowledgeEntry(title="Project Instructions", content=version.content) ] ) async def _build_entitlements( self, project_id: UUID, plan_slug: str ) -> ManifestEntitlementsSection: _PLAN_LIMITS: dict[str, dict] = { "free": {"edge_functions_max": 3, "storage_gb_max": 1, "auth_users_max": 1000, "credits_total": 100}, "starter": {"edge_functions_max": 20, "storage_gb_max": 5, "auth_users_max": 10000, "credits_total": 500}, "pro": {"edge_functions_max": 100, "storage_gb_max": 10, "auth_users_max": 100000, "credits_total": 1000}, "growth": {"edge_functions_max": 500, "storage_gb_max": 50, "auth_users_max": 500000, "credits_total": 5000}, } limits = _PLAN_LIMITS.get(plan_slug, _PLAN_LIMITS["free"]) credits_used = await self._get_credits_used(project_id) credits_total = limits.pop("credits_total", 100) pct = round((credits_used / credits_total) * 100) if credits_total else 0 return ManifestEntitlementsSection( plan=plan_slug, limits=limits, usage_this_period={ "credits_used": credits_used, "credits_total": credits_total, "pct": pct, }, ) async def _get_credits_used(self, project_id: UUID) -> int: try: from src.core.redis import get_redis from src.domain.projects.models import Project from sqlalchemy import select proj = await self._db.get(Project, project_id) if not proj: return 0 from datetime import date period = date.today().strftime("%Y-%m") key = f"usage:current:{proj.workspace_id}:{period}:credits" async for redis in get_redis(): val = await redis.get(key) return int(val) if val else 0 except Exception: return 0 def _build_query_dsl(self) -> ManifestQueryDslSection: """Return the static query DSL reference section. Platform-native — always enabled for cloud projects (query_dsl_v1 capability). Codegen uses this for Cloud record queries through the BFF. """ return ManifestQueryDslSection() async def _resolve_gateway_url(self, workspace_id: UUID, fallback: str) -> str: """Return the gateway URL for the workspace's active Cloud instance. Walks workspace_cloud_assignment → cloud_instances to get the per-instance gateway URL. Falls back to the global setting when no assignment exists (single-instance deployments or not-yet-assigned workspaces). """ try: from src.domain.cloud_infra.models import ( CloudInstance, CloudInstanceStatus, WorkspaceCloudAssignment, WorkspaceCloudAssignmentStatus, ) from sqlalchemy import select assign_result = await self._db.execute( select(WorkspaceCloudAssignment).where( WorkspaceCloudAssignment.workspace_id == workspace_id, WorkspaceCloudAssignment.status == WorkspaceCloudAssignmentStatus.active, ) ) assignment = assign_result.scalar_one_or_none() if not assignment: return fallback instance_result = await self._db.execute( select(CloudInstance).where( CloudInstance.id == assignment.cloud_instance_id, CloudInstance.status != CloudInstanceStatus.offline, ) ) instance = instance_result.scalar_one_or_none() if not instance: return fallback return instance.gateway_url.rstrip("/") except Exception: return fallback async def _get_run_for_project( self, project_id: UUID, run_id: UUID ) -> ManifestRun: run = await self._run_repo.get_by_id(run_id) if not run or run.project_id != project_id: raise NotFoundError("Run not found") return run def _hash_manifest(manifest: ManifestResponse) -> str: data = manifest.model_dump_json() return hashlib.sha256(data.encode()).hexdigest()[:64]