"""edge_functions domain — business logic.""" from __future__ import annotations import secrets from datetime import datetime, timezone from pathlib import Path from typing import Literal, Optional from uuid import UUID import structlog from pydantic import ValidationError as PydanticValidationError from src.core.exceptions import ConflictError, NotFoundError, ValidationError from src.domain.functions.library_manifest_schema import ( FunctionLibraryManifest, FunctionLibraryManifestValidator, ManifestValidationError, MUTABLE_LIBRARY_FIELDS, apply_manifest_to_row_dict, compute_code_hash, compute_config_hash, compute_manifest_hash, diff_fields, function_library_row_to_manifest, parse_manifest, ) from src.core.security.encryption import encrypt_value from src.domain.connectors.repository import ConnectorRepository from src.domain.edge_functions import cf_runtime from src.domain.edge_functions.connector_bindings import resolve_connector_bindings from src.domain.edge_functions.models import ( FunctionLibrary, FunctionLibraryVersion, ProjectFunction, ProjectFunctionVersion, ProjectWebhookEndpoint, WorkerRuntimeVersion, ) from src.domain.edge_functions.repository import ( FunctionLibraryRepository, FunctionLibraryVersionRepository, ProjectFunctionRepository, ProjectFunctionTokenRepository, ProjectFunctionVersionRepository, ProjectWebhookEndpointRepository, WorkerRuntimeVersionRepository, ) from src.domain.edge_functions.schemas import ( CreateFunctionRequest, CreateLibraryFunctionRequest, CreateLibraryVersionRequest, CreateRuntimeVersionRequest, CreateWebhookEndpointRequest, DeployFromLibraryRequest, DeprecateLibraryFunctionRequest, FunctionLibraryResponse, FunctionLibraryUpsertResult, FunctionLibraryVersionResponse, FunctionResponse, ManifestDiffResponse, ManifestValidateResponse, FunctionVersionResponse, PublishLibraryFunctionRequest, RuntimeVersionResponse, UpdateFunctionCodeRequest, UpdateFunctionSecretsRequest, UpdateLibraryFunctionRequest, UpdateRuntimeVersionRequest, WebhookEndpointResponse, ) from src.config import get_settings from src.domain.projects.repository import ProjectRepository from src.integrations.cloud_control_client import CloudControlClient logger = structlog.get_logger() _DEFAULT_MAX_FUNCTIONS = 100 _WEBHOOK_BASE_URL = "https://cloud.p2hs.com/wh" class EdgeFunctionsService: def __init__( self, *, library_repo: FunctionLibraryRepository, library_version_repo: Optional[FunctionLibraryVersionRepository] = None, fn_repo: ProjectFunctionRepository, version_repo: ProjectFunctionVersionRepository, webhook_repo: ProjectWebhookEndpointRepository, runtime_repo: WorkerRuntimeVersionRepository, token_repo: ProjectFunctionTokenRepository, project_repo: ProjectRepository, connector_repo: ConnectorRepository, cloud_client: CloudControlClient, ) -> None: self._library = library_repo self._libver = library_version_repo self._fn = fn_repo self._ver = version_repo self._wh = webhook_repo self._runtime = runtime_repo self._tokens = token_repo self._projects = project_repo self._connectors = connector_repo self._cloud = cloud_client # ------------------------------------------------------------------ # Function Library (read + admin write) # ------------------------------------------------------------------ async def list_library( self, trigger_type: Optional[str] = None, tags: Optional[list[str]] = None, integration: Optional[str] = None, category: Optional[str] = None, minimum_plan: Optional[str] = None, visibility: Optional[str] = None, ) -> list[FunctionLibraryResponse]: rows = await self._library.list_published( trigger_type=trigger_type, tags=tags, integration=integration, category=category, minimum_plan=minimum_plan, visibility=visibility or "public", ) return [_library_response(r) for r in rows] async def list_library_all(self) -> list[FunctionLibraryResponse]: """Admin view — all entries regardless of status.""" rows = await self._library.list_all() return [_library_response(r) for r in rows] async def get_library_function(self, slug: str) -> FunctionLibraryResponse: row = await self._library.get_by_slug(slug) if not row: raise NotFoundError(f"Library function {slug!r} not found") return _library_response(row) async def upsert_library_function_from_manifest( self, *, manifest: FunctionLibraryManifest, source: Literal["seed", "admin_console", "api", "migration"] = "api", publish: bool = False, create_version: bool = False, dry_run: bool = False, force: bool = False, metadata_only: bool = False, seed_path: Optional[str] = None, actor_user_id: Optional[UUID] = None, ) -> FunctionLibraryUpsertResult: """Single write path: validate manifest, upsert row, optional version snapshot, audit log.""" val = FunctionLibraryManifestValidator() if seed_path: manifest = manifest.model_copy(update={"seed_path": seed_path}) try: val.validate_or_raise(manifest) except ManifestValidationError as e: raise ValidationError("; ".join(e.errors)) effective = manifest row = await self._library.get_by_slug(manifest.slug) if row and metadata_only: effective = _merge_metadata_only_manifest(row, manifest) m_changes: list[str] = [] v_changes: list[str] = [] if row: m_changes, v_changes = diff_fields(row, effective) if ( source == "seed" and row and row.admin_override and not force ): return FunctionLibraryUpsertResult( slug=manifest.slug, created=False, updated=False, skipped=True, skip_reason="admin_override: use --force to overwrite from seed", dry_run=dry_run, ) if ( row and row.status == "published" and v_changes and not metadata_only and not create_version and not publish ): raise ValidationError( "Published function has versioned field changes; create a new library version " "or use metadata-only / draft workflow. " "For seeds: python scripts/seed_function_library.py --slug " f"{manifest.slug} --mode create-version", ) if row and not m_changes and not v_changes and not force: return FunctionLibraryUpsertResult( slug=manifest.slug, created=False, updated=False, skipped=True, skip_reason="no changes", dry_run=dry_run, manifest_hash=row.manifest_hash, ) m_hash = compute_manifest_hash(effective) if dry_run: return FunctionLibraryUpsertResult( slug=manifest.slug, created=row is None, updated=row is not None, dry_run=True, manifest_hash=m_hash, metadata_changed=m_changes, versioned_changed=v_changes, requires_new_version=bool(v_changes and row and row.status == "published"), ) data = apply_manifest_to_row_dict(effective) data["manifest_hash"] = m_hash if source == "seed": data["latest_seed_hash"] = m_hash data["last_synced_from_seed_at"] = datetime.now(timezone.utc) if seed_path: data["seed_path"] = seed_path data["managed_source"] = "seed" elif source == "admin_console": data["managed_source"] = "admin" if not row: data.setdefault("is_official", True) data.setdefault("is_active", True) row = await self._library.create(**data) created = True else: row = await self._library.update(row, **data) created = False ver_id: Optional[UUID] = None published_flag = False if self._libver and publish: row = await self._library.update(row, status="published") ver = await self._ensure_library_version_snapshot( row, changelog=None, set_current=True ) ver_id = ver.id published_flag = True elif self._libver and create_version: ver = await self._ensure_library_version_snapshot( row, changelog=None, set_current=True ) ver_id = ver.id logger.info( "library_function_manifest_upsert", slug=manifest.slug, source=source, created=created, published=published_flag, actor_id=str(actor_user_id) if actor_user_id else None, ) return FunctionLibraryUpsertResult( slug=manifest.slug, created=created, updated=not created, manifest_hash=m_hash, metadata_changed=m_changes, versioned_changed=v_changes, requires_new_version=bool( v_changes and row and row.status == "published" and not create_version ), published=published_flag, current_version_id=ver_id, ) async def diff_manifest_against_row( self, slug: str, *, incoming: Optional[FunctionLibraryManifest] = None, ) -> ManifestDiffResponse: row = await self._library.get_by_slug(slug) if not row: return ManifestDiffResponse(slug=slug, status="not_found") if incoming is None: return ManifestDiffResponse( slug=slug, status="error", message="incoming manifest required for diff (use /diff?source=seed for repo seeds)", ) m, v = diff_fields(row, incoming) return ManifestDiffResponse( slug=slug, status="unchanged" if not m and not v else "changed", metadata_changes=m, versioned_changes=v, requires_new_version=bool(v and row.status == "published"), admin_override=row.admin_override, ) async def diff_manifest_against_seed(self, slug: str) -> ManifestDiffResponse: """Compare DB row to ``seeds/function_library/{slug}.yaml`` if present.""" row = await self._library.get_by_slug(slug) if not row: return ManifestDiffResponse(slug=slug, status="not_found") root = Path(__file__).resolve().parents[3] path = root / "seeds" / "function_library" / f"{slug}.yaml" if not path.is_file(): return ManifestDiffResponse( slug=slug, status="missing_seed", message=f"No seed file at {path}", ) import yaml # local import — seed tooling only in admin path raw = yaml.safe_load(path.read_text(encoding="utf-8")) if not raw: return ManifestDiffResponse( slug=slug, status="error", message="empty seed file" ) try: incoming = parse_manifest(raw) except Exception as e: return ManifestDiffResponse( slug=slug, status="error", message=f"parse error: {e}" ) m, v = diff_fields(row, incoming) return ManifestDiffResponse( slug=slug, status="unchanged" if not m and not v else "changed", metadata_changes=m, versioned_changes=v, requires_new_version=bool(v and row.status == "published"), admin_override=row.admin_override, ) def validate_library_manifest_dict(self, raw: dict) -> ManifestValidateResponse: try: m = parse_manifest(raw) FunctionLibraryManifestValidator().validate_or_raise(m) except ManifestValidationError as e: return ManifestValidateResponse(ok=False, errors=e.errors) except PydanticValidationError as e: return ManifestValidateResponse(ok=False, errors=[str(e)]) return ManifestValidateResponse(ok=True, errors=[]) async def import_library_from_manifest_dict( self, raw: dict, *, source: Literal["api", "migration", "admin_console"] = "api", publish: bool = False, ) -> FunctionLibraryUpsertResult: m = parse_manifest(raw) return await self.upsert_library_function_from_manifest( manifest=m, source=source, publish=publish, dry_run=False ) async def sync_library_from_manifest_for_slug( self, slug: str, raw: dict ) -> FunctionLibraryUpsertResult: m = parse_manifest(raw) if m.slug != slug: raise ValidationError("manifest.slug does not match path") return await self.upsert_library_function_from_manifest( manifest=m, source="api", publish=False, dry_run=False ) async def create_library_function( self, body: CreateLibraryFunctionRequest ) -> FunctionLibraryResponse: existing = await self._library.get_by_slug(body.slug) if existing: raise ConflictError(f"Library function {body.slug!r} already exists") manifest = _create_library_request_to_manifest(body) await self.upsert_library_function_from_manifest( manifest=manifest, source="admin_console", publish=False, dry_run=False ) row = await self._library.get_by_slug(body.slug) if not row: raise NotFoundError("Failed to load new library function") return _library_response(row) async def update_library_function( self, slug: str, body: UpdateLibraryFunctionRequest ) -> FunctionLibraryResponse: row = await self._library.get_by_slug(slug) if not row: raise NotFoundError(f"Library function {slug!r} not found") if row.status == "published" and _has_disallowed_published_update(body): raise ValidationError("Cannot change billing_config or status to unsafe value via PATCH; use version workflow.") base = function_library_row_to_manifest(row) data = base.model_dump() for k, v in body.model_dump(exclude_none=True).items(): if k in data: data[k] = v manifest = parse_manifest(data) await self.upsert_library_function_from_manifest( manifest=manifest, source="admin_console", metadata_only=True, dry_run=False ) row = await self._library.get_by_slug(slug) if not row: raise NotFoundError(f"Library function {slug!r} not found") return _library_response(row) async def publish_library_function( self, slug: str, body: PublishLibraryFunctionRequest ) -> FunctionLibraryResponse: row = await self._library.get_by_slug(slug) if not row: raise NotFoundError(f"Library function {slug!r} not found") if not self._libver: raise RuntimeError("Library version repo not initialised") row = await self._library.update(row, status="published") await self._ensure_library_version_snapshot( row, changelog=body.changelog, set_current=True ) row = await self._library.get_by_slug(slug) if not row: raise NotFoundError(f"Library function {slug!r} not found") logger.info("library_function_published", slug=slug, version=row.version) return _library_response(row) async def deprecate_library_function( self, slug: str, body: DeprecateLibraryFunctionRequest ) -> FunctionLibraryResponse: row = await self._library.get_by_slug(slug) if not row: raise NotFoundError(f"Library function {slug!r} not found") updates: dict = { "status": "deprecated", "deprecated_at": datetime.now(timezone.utc), } if body.replacement_slug: updates["replacement_slug"] = body.replacement_slug row = await self._library.update(row, **updates) logger.info("library_function_deprecated", slug=slug) return _library_response(row) async def create_library_version( self, slug: str, body: CreateLibraryVersionRequest ) -> FunctionLibraryVersionResponse: if not self._libver: raise RuntimeError("Library version repo not initialised") row = await self._library.get_by_slug(slug) if not row: raise NotFoundError(f"Library function {slug!r} not found") existing = await self._libver.get_by_version(row.id, body.version) if existing: raise ConflictError(f"Version {body.version!r} already exists for {slug!r}") base = function_library_row_to_manifest(row) merged = parse_manifest( { **base.model_dump(), "version": body.version, "semver_channel": body.semver_channel, "code_template": body.code_template, "code_bundle_ref": body.code_bundle_ref, } ) m_hash = compute_manifest_hash(merged) code_h = compute_code_hash(body.code_template) cfg_h = compute_config_hash(merged) ver = await self._libver.create( library_function_id=row.id, version=body.version, semver_channel=body.semver_channel, code_template=body.code_template, code_bundle_ref=body.code_bundle_ref, changelog=body.changelog, manifest_hash=m_hash, code_hash=code_h, config_hash=cfg_h, published_at=datetime.now(timezone.utc), ) await self._library.update( row, version=body.version, semver_channel=body.semver_channel, code_template=body.code_template, code_bundle_ref=body.code_bundle_ref, manifest_hash=m_hash, current_version_id=ver.id, ) logger.info("library_function_version_created", slug=slug, version=body.version) return _library_version_response(ver) async def list_library_versions(self, slug: str) -> list[FunctionLibraryVersionResponse]: if not self._libver: raise RuntimeError("Library version repo not initialised") row = await self._library.get_by_slug(slug) if not row: raise NotFoundError(f"Library function {slug!r} not found") vers = await self._libver.list_for_function(row.id) return [_library_version_response(v) for v in vers] async def promote_library_version( self, slug: str, version_id: UUID ) -> FunctionLibraryResponse: if not self._libver: raise RuntimeError("Library version repo not initialised") row = await self._library.get_by_slug(slug) if not row: raise NotFoundError(f"Library function {slug!r} not found") ver = await self._libver.get_by_id(version_id) if not ver or ver.library_function_id != row.id: raise NotFoundError("Library version not found") row = await self._library.update( row, version=ver.version, semver_channel=ver.semver_channel, code_template=ver.code_template, code_bundle_ref=ver.code_bundle_ref, current_version_id=ver.id, status="published", ) logger.info("library_version_promoted", slug=slug, version_id=str(version_id)) return _library_response(row) async def disable_library_function(self, slug: str) -> FunctionLibraryResponse: row = await self._library.get_by_slug(slug) if not row: raise NotFoundError(f"Library function {slug!r} not found") row = await self._library.update(row, status="disabled", is_active=False) return _library_response(row) async def restore_library_function( self, slug: str, *, to_status: str = "draft" ) -> FunctionLibraryResponse: row = await self._library.get_by_slug(slug) if not row: raise NotFoundError(f"Library function {slug!r} not found") if to_status not in ("draft", "published"): raise ValidationError("to_status must be draft or published") row = await self._library.update( row, status=to_status, is_active=to_status != "disabled" ) return _library_response(row) async def _ensure_library_version_snapshot( self, row: FunctionLibrary, *, changelog: Optional[str] = None, set_current: bool = True, ) -> FunctionLibraryVersion: if not self._libver: raise RuntimeError("Library version repo not initialised") m = function_library_row_to_manifest(row) m_hash = compute_manifest_hash(m) code_h = compute_code_hash(m.code_template) cfg_h = compute_config_hash(m) now = datetime.now(timezone.utc) existing = await self._libver.get_by_version(row.id, row.version) if existing: ver = await self._libver.update( existing, manifest_hash=m_hash, code_hash=code_h, config_hash=cfg_h, code_template=row.code_template, code_bundle_ref=row.code_bundle_ref, changelog=changelog if changelog is not None else existing.changelog, published_at=existing.published_at or now, ) else: ver = await self._libver.create( library_function_id=row.id, version=row.version, semver_channel=row.semver_channel, code_template=row.code_template, code_bundle_ref=row.code_bundle_ref, changelog=changelog, manifest_hash=m_hash, code_hash=code_h, config_hash=cfg_h, published_at=now, ) if set_current: await self._library.update(row, current_version_id=ver.id) return ver # ------------------------------------------------------------------ # Project Functions # ------------------------------------------------------------------ async def list_functions(self, project_id: UUID) -> list[FunctionResponse]: rows = await self._fn.list_for_project(project_id) return [_fn_response(r) for r in rows] async def list_functions_by_pack(self, project_id: UUID, pack_slug: str) -> list[FunctionResponse]: rows = await self._fn.list_by_pack(project_id, pack_slug) return [_fn_response(r) for r in rows] async def create_function( self, project_id: UUID, body: CreateFunctionRequest ) -> FunctionResponse: existing = await self._fn.get_by_slug(project_id, body.slug) if existing: raise ConflictError(f"Function {body.slug!r} already exists for this project") count = await self._fn.count_active(project_id) if count >= _DEFAULT_MAX_FUNCTIONS: raise ConflictError(f"Project has reached the maximum of {_DEFAULT_MAX_FUNCTIONS} functions") fn = await self._fn.create( project_id=project_id, slug=body.slug, name=body.name, description=body.description, runtime=body.runtime, trigger_type=body.trigger_type, trigger_config=body.trigger_config, secret_refs=body.secret_refs, created_by=body.created_by, status="pending", ) ver = await self._ver.create( function_id=fn.id, version_number=1, code_bundle_ref=body.code_bundle_ref, status="pending", ) # Auto-deploy the initial version fn = await self._deploy_version_internal(fn, ver) logger.info("function_created", project_id=str(project_id), slug=body.slug) return _fn_response(fn) async def deploy_from_library( self, project_id: UUID, body: DeployFromLibraryRequest ) -> FunctionResponse: lib_fn = await self._library.get_by_id(body.library_fn_id) if not lib_fn: raise NotFoundError("Library function not found") existing = await self._fn.get_by_slug(project_id, body.slug) if existing: raise ConflictError(f"Function {body.slug!r} already exists for this project") count = await self._fn.count_active(project_id) if count >= _DEFAULT_MAX_FUNCTIONS: raise ConflictError(f"Project has reached the maximum of {_DEFAULT_MAX_FUNCTIONS} functions") # Resolve the pinned library version (if versioning is active) lib_ver_id: Optional[UUID] = None if self._libver: lib_ver = await self._libver.get_by_version(lib_fn.id, lib_fn.version) lib_ver_id = lib_ver.id if lib_ver else None fn = await self._fn.create( project_id=project_id, slug=body.slug, name=lib_fn.name, description=lib_fn.description, runtime=lib_fn.runtime, trigger_type=lib_fn.trigger_type, trigger_config=body.trigger_config or {}, access_policy=body.access_policy, library_fn_id=lib_fn.id, secret_refs=body.secret_refs, editable=True, code_snapshot=lib_fn.code_template, config_snapshot=body.trigger_config, created_by=body.created_by, pack_slug=body.pack_slug, pack_version=body.pack_version, status="pending", ) ver = await self._ver.create( function_id=fn.id, version_number=1, library_fn_id=lib_fn.id, library_fn_version=lib_fn.version, source_library_version_id=lib_ver_id, status="pending", ) fn = await self._deploy_version_internal(fn, ver) await self._library.increment_install_count(lib_fn.id) logger.info( "function_deployed_from_library", project_id=str(project_id), slug=body.slug, library_slug=lib_fn.slug, ) return _fn_response(fn) async def deploy_from_library_by_slug( self, project_id: UUID, library_slug: str, project_slug: str, trigger_config: Optional[dict] = None, access_policy: str = "authenticated", secret_refs: Optional[list] = None, created_by: str = "codegen", pack_slug: Optional[str] = None, pack_version: Optional[str] = None, ) -> FunctionResponse: """Upsert a library function for a project by library slug. - If the project_slug doesn't exist: create and deploy. - If it exists with the same library_fn_id, trigger_config, and pack_version: skip (idempotent). - If it exists but trigger_config or pack_version changed: update and redeploy. """ lib_fn = await self._library.get_by_slug(library_slug) if not lib_fn: raise NotFoundError(f"Library function {library_slug!r} not found") existing = await self._fn.get_by_slug(project_id, project_slug) if existing: incoming_config = trigger_config or {} existing_config = existing.trigger_config or {} same_library = existing.library_fn_id == lib_fn.id same_config = existing_config == incoming_config same_policy = existing.access_policy == access_policy same_pack_version = existing.pack_version == pack_version if same_library and same_config and same_policy and same_pack_version: return _fn_response(existing) update_fields: dict = {} if not same_config: update_fields["trigger_config"] = incoming_config if not same_policy: update_fields["access_policy"] = access_policy if not same_library: update_fields["library_fn_id"] = lib_fn.id update_fields["name"] = lib_fn.name update_fields["description"] = lib_fn.description update_fields["runtime"] = lib_fn.runtime update_fields["trigger_type"] = lib_fn.trigger_type if not same_pack_version: update_fields["pack_slug"] = pack_slug update_fields["pack_version"] = pack_version updated = await self._fn.update(existing, **update_fields) version_number = await self._ver.next_version_number(updated.id) ver = await self._ver.create( function_id=updated.id, version_number=version_number, library_fn_id=lib_fn.id, library_fn_version=lib_fn.version, status="pending", ) updated = await self._deploy_version_internal(updated, ver) logger.info( "function_upserted", project_id=str(project_id), slug=project_slug, library_swapped=not same_library, config_changed=not same_config, ) return _fn_response(updated) return await self.deploy_from_library( project_id, DeployFromLibraryRequest( library_fn_id=lib_fn.id, slug=project_slug, trigger_config=trigger_config or {}, access_policy=access_policy, secret_refs=secret_refs or [], created_by=created_by, pack_slug=pack_slug, pack_version=pack_version, ), ) async def update_code( self, project_id: UUID, slug: str, body: UpdateFunctionCodeRequest ) -> FunctionVersionResponse: fn = await self._get_fn_or_404(project_id, slug) version_number = await self._ver.next_version_number(fn.id) ver = await self._ver.create( function_id=fn.id, version_number=version_number, code_bundle_ref=body.code_bundle_ref, status="pending", ) return _version_response(ver) async def deploy_version(self, project_id: UUID, slug: str) -> FunctionResponse: fn = await self._get_fn_or_404(project_id, slug) ver = await self._ver.get_latest_pending(fn.id) if not ver: raise NotFoundError("No pending version to deploy") fn = await self._deploy_version_internal(fn, ver) return _fn_response(fn) async def rollback(self, project_id: UUID, slug: str) -> FunctionResponse: fn = await self._get_fn_or_404(project_id, slug) if not fn.active_version_id: raise NotFoundError("No active version to roll back from") current = await self._ver.get_latest_deployed(fn.id) if not current: raise NotFoundError("No deployed version found") prev = await self._ver.get_previous_deployed(fn.id, exclude_version=current.version_number) if not prev: raise NotFoundError("No previous deployed version to roll back to") # Mark current as rolled_back, re-activate previous await self._ver.update(current, status="rolled_back") rollback_ver = await self._ver.create( function_id=fn.id, version_number=await self._ver.next_version_number(fn.id), code_bundle_ref=prev.code_bundle_ref, library_fn_id=prev.library_fn_id, library_fn_version=prev.library_fn_version, status="pending", rollback_from_version=current.version_number, ) fn = await self._deploy_version_internal(fn, rollback_ver) logger.info("function_rolled_back", project_id=str(project_id), slug=slug) return _fn_response(fn) async def update_secrets( self, project_id: UUID, slug: str, body: UpdateFunctionSecretsRequest ) -> FunctionResponse: fn = await self._get_fn_or_404(project_id, slug) fn = await self._fn.update(fn, secret_refs=body.secret_refs) await self._push_function(project_id, fn) return _fn_response(fn) async def delete_function(self, project_id: UUID, slug: str) -> None: fn = await self._get_fn_or_404(project_id, slug) await self._fn.soft_delete(fn) await self._cloud.delete_function(project_id=str(project_id), slug=slug) try: await cf_runtime.delete_function_script( cf_runtime.compute_worker_name(project_id, slug) ) except cf_runtime.EdgeFunctionDeployError as exc: logger.warning( "edge_function_script_delete_failed", project_id=str(project_id), slug=slug, error=str(exc), ) logger.info("function_deleted", project_id=str(project_id), slug=slug) async def get_function(self, project_id: UUID, slug: str) -> FunctionResponse: fn = await self._get_fn_or_404(project_id, slug) return _fn_response(fn) # ------------------------------------------------------------------ # Webhook Endpoints # ------------------------------------------------------------------ async def list_webhooks(self, project_id: UUID) -> list[WebhookEndpointResponse]: rows = await self._wh.list_for_project(project_id) return [_webhook_response(r) for r in rows] async def create_webhook( self, project_id: UUID, body: CreateWebhookEndpointRequest ) -> WebhookEndpointResponse: wh_token = secrets.token_urlsafe(32) signing_enc: Optional[str] = None if body.signing_secret: signing_enc = encrypt_value(body.signing_secret) row = await self._wh.create( project_id=project_id, wh_token=wh_token, provider=body.provider, events_json=body.events, signing_secret_enc=signing_enc, function_slug=body.function_slug, auto_registered=False, ) return _webhook_response(row) # ------------------------------------------------------------------ # Runtime Versions (admin) # ------------------------------------------------------------------ async def list_runtime_versions(self) -> list[RuntimeVersionResponse]: rows = await self._runtime.list_all() return [_runtime_response(r) for r in rows] async def create_runtime_version( self, body: CreateRuntimeVersionRequest ) -> RuntimeVersionResponse: existing = await self._runtime.get_by_version(body.version) if existing: raise ConflictError(f"Runtime version {body.version!r} already exists") row = await self._runtime.create( version=body.version, bundle_ref=body.bundle_ref, release_notes=body.release_notes, status="draft", ) return _runtime_response(row) async def update_runtime_version( self, version: str, body: UpdateRuntimeVersionRequest ) -> RuntimeVersionResponse: row = await self._runtime.get_by_version(version) if not row: raise NotFoundError(f"Runtime version {version!r} not found") fields: dict = {"status": body.status} now = datetime.now(timezone.utc) if body.status == "stable" and not row.released_at: fields["released_at"] = now if body.status == "deprecated" and not row.deprecated_at: fields["deprecated_at"] = now row = await self._runtime.update(row, **fields) return _runtime_response(row) # ------------------------------------------------------------------ # Internal # ------------------------------------------------------------------ async def _get_fn_or_404(self, project_id: UUID, slug: str) -> ProjectFunction: fn = await self._fn.get_by_slug(project_id, slug) if not fn: raise NotFoundError(f"Function {slug!r} not found") return fn async def _deploy_version_internal( self, fn: ProjectFunction, ver: ProjectFunctionVersion ) -> ProjectFunction: """Mark version as deployed, update function's active_version_id, push to Cloud.""" now = datetime.now(timezone.utc) ver = await self._ver.update(ver, status="deployed", deployed_at=now) fn = await self._fn.update( fn, active_version_id=ver.id, status="active", ) await self._push_function(fn.project_id, fn, ver) return fn async def _required_integrations_for(self, fn: ProjectFunction) -> list[str]: if not fn.library_fn_id: return [] lib_fn = await self._library.get_by_id(fn.library_fn_id) if not lib_fn: return [] return list(lib_fn.required_integrations or []) async def _ensure_function_token(self, project_id: UUID) -> str: """Return the decrypted per-project Edge Function runtime token, creating it if needed.""" from src.domain.edge_functions.runtime_token import ensure_project_cloud_service_token return await ensure_project_cloud_service_token(self._tokens, self._cloud, project_id) async def _push_function( self, project_id: UUID, fn: ProjectFunction, ver: Optional[ProjectFunctionVersion] = None, ) -> None: bundle_ref: Optional[str] = None if ver: bundle_ref = ver.code_bundle_ref elif fn.active_version_id: active = await self._ver.get_latest_deployed(fn.id) if active: bundle_ref = active.code_bundle_ref project = await self._projects.get_by_id(project_id) if not project: raise NotFoundError(f"Project {project_id} not found") worker_name = cf_runtime.compute_worker_name(project_id, fn.slug) if fn.code_snapshot and fn.status == "active": settings = get_settings() gateway_base_url = settings.cloud_public_api_url or settings.cloud_control_api_url token = await self._ensure_function_token(project_id) extra_env_vars: dict[str, str] = {} extra_secrets: dict[str, str] = {} required_integrations = await self._required_integrations_for(fn) if required_integrations: extra_env_vars, extra_secrets = await resolve_connector_bindings( self._connectors, project_id, required_integrations ) # Static config values a pack resolved at install time (e.g. a # notification destination address) — plain env vars, not secrets. # See PackInstallService.install_pack's trigger_config resolution. static_env_vars = (fn.trigger_config or {}).get("static_env_vars") if isinstance(static_env_vars, dict): extra_env_vars = {**extra_env_vars, **{str(k): str(v) for k, v in static_env_vars.items()}} try: worker_name = await cf_runtime.deploy_function_script( project_id=project_id, workspace_id=project.workspace_id, slug=fn.slug, runtime=fn.runtime, code=fn.code_snapshot, gateway_base_url=gateway_base_url, project_token=token, extra_env_vars=extra_env_vars, extra_secrets=extra_secrets, ) except cf_runtime.EdgeFunctionDeployError as exc: logger.warning( "edge_function_script_deploy_failed", project_id=str(project_id), slug=fn.slug, error=str(exc), ) event_filter: Optional[list[str]] = None schedule_cron: Optional[str] = None trigger_config = fn.trigger_config or {} if fn.trigger_type == "event" and trigger_config.get("event_type"): event_filter = [trigger_config["event_type"]] elif fn.trigger_type == "schedule" and trigger_config.get("cron"): schedule_cron = trigger_config["cron"] await self._cloud.deploy_function( project_id=str(project_id), slug=fn.slug, payload={ "id": str(fn.id), "workspace_id": str(project.workspace_id), "slug": fn.slug, "name": fn.name, "runtime": fn.runtime, "trigger_type": fn.trigger_type, "trigger_config": trigger_config, "event_filter": event_filter, "schedule_cron": schedule_cron, "worker_name": worker_name, "active_version": str(ver.version_number) if ver else "1", "access_policy": fn.access_policy, "code_bundle_ref": bundle_ref, "library_fn_id": str(fn.library_fn_id) if fn.library_fn_id else None, "secret_refs": fn.secret_refs or [], "status": "active" if fn.status == "active" else "paused", "pack_slug": fn.pack_slug, "pack_version": fn.pack_version, }, ) # --------------------------------------------------------------------------- # Manifest helpers # --------------------------------------------------------------------------- def _merge_metadata_only_manifest( row: FunctionLibrary, incoming: FunctionLibraryManifest ) -> FunctionLibraryManifest: base = function_library_row_to_manifest(row) data = base.model_dump() inc = incoming.model_dump() for f in MUTABLE_LIBRARY_FIELDS: if f in inc: data[f] = inc[f] return parse_manifest(data) def _create_library_request_to_manifest( body: CreateLibraryFunctionRequest, ) -> FunctionLibraryManifest: rt: Literal["js", "ts", "python"] = "python" if body.runtime == "python" else "js" return FunctionLibraryManifest( slug=body.slug, name=body.name, description=body.description, category=body.category, tags=body.tags, status=body.status, visibility=body.visibility, version=body.version, semver_channel=body.semver_channel, minimum_plan=body.minimum_plan, runtime=rt, trigger_type=body.trigger_type, # type: ignore[arg-type] supported_trigger_types=body.supported_trigger_types, execution_mode=body.execution_mode, timeout_ms=body.timeout_ms, max_retries=body.max_retries, idempotency_required=body.idempotency_required, required_integrations=body.required_integrations, required_secrets=body.required_secrets, required_capabilities=body.required_capabilities, required_bindings=body.required_bindings, input_schema=body.input_schema, output_schema=body.output_schema, permission_scope=body.permission_scope, external_access=body.external_access, trigger_config_schema=body.trigger_config_schema, codegen_hints=body.codegen_hints, observability_config=body.observability_config, billing_config=body.billing_config, code_template=body.code_template, code_bundle_ref=body.code_bundle_ref, managed_source="admin", ) def _has_disallowed_published_update(body: UpdateLibraryFunctionRequest) -> bool: return body.billing_config is not None # --------------------------------------------------------------------------- # Response builders # --------------------------------------------------------------------------- def _library_response(row: FunctionLibrary) -> FunctionLibraryResponse: return FunctionLibraryResponse( id=row.id, name=row.name, slug=row.slug, description=row.description, category=row.category, runtime=row.runtime, trigger_type=row.trigger_type, supported_trigger_types=row.supported_trigger_types, status=row.status, visibility=row.visibility, version=row.version, semver_channel=row.semver_channel, minimum_plan=row.minimum_plan, execution_mode=row.execution_mode, timeout_ms=row.timeout_ms, max_retries=row.max_retries, idempotency_required=row.idempotency_required, required_integrations=row.required_integrations, required_secrets=row.required_secrets, required_capabilities=row.required_capabilities, required_bindings=row.required_bindings, tags=row.tags, input_schema=row.input_schema, output_schema=row.output_schema, permission_scope=row.permission_scope, external_access=row.external_access, trigger_config_schema=row.trigger_config_schema, codegen_hints=row.codegen_hints, observability_config=row.observability_config, billing_config=row.billing_config, code_template=row.code_template, deprecated_at=row.deprecated_at, replacement_slug=row.replacement_slug, install_count=row.install_count, is_official=row.is_official, ) def _library_version_response(row: FunctionLibraryVersion) -> FunctionLibraryVersionResponse: return FunctionLibraryVersionResponse( id=row.id, library_function_id=row.library_function_id, version=row.version, semver_channel=row.semver_channel, code_template=row.code_template, changelog=row.changelog, published_at=row.published_at, created_at=row.created_at, ) def _fn_response(row: ProjectFunction) -> FunctionResponse: return FunctionResponse( id=row.id, project_id=row.project_id, slug=row.slug, name=row.name, description=row.description, runtime=row.runtime, trigger_type=row.trigger_type, trigger_config=row.trigger_config, access_policy=row.access_policy, active_version_id=row.active_version_id, status=row.status, library_fn_id=row.library_fn_id, editable=row.editable, secret_refs=row.secret_refs, created_by=row.created_by, created_at=row.created_at, updated_at=row.updated_at, ) def _version_response(row: ProjectFunctionVersion) -> FunctionVersionResponse: return FunctionVersionResponse( id=row.id, function_id=row.function_id, version_number=row.version_number, code_bundle_ref=row.code_bundle_ref, library_fn_id=row.library_fn_id, library_fn_version=row.library_fn_version, status=row.status, deployed_at=row.deployed_at, rollback_from_version=row.rollback_from_version, created_at=row.created_at, ) def _webhook_response(row: ProjectWebhookEndpoint) -> WebhookEndpointResponse: return WebhookEndpointResponse( id=row.id, project_id=row.project_id, wh_token=row.wh_token, webhook_url=f"{_WEBHOOK_BASE_URL}/{row.wh_token}", provider=row.provider, events=row.events_json, auto_registered=row.auto_registered, registered_at=row.registered_at, function_slug=row.function_slug, is_active=row.is_active, created_at=row.created_at, ) def _runtime_response(row: WorkerRuntimeVersion) -> RuntimeVersionResponse: return RuntimeVersionResponse( id=row.id, version=row.version, bundle_ref=row.bundle_ref, status=row.status, release_notes=row.release_notes, released_at=row.released_at, deprecated_at=row.deprecated_at, created_at=row.created_at, )