fix: app registration handling and improve error handling in EAM service (d72d930c) · Commits · OOP / code / open-exposure-gateway

src/open_exposure_gateway/adapters/database/repos/app_registrations.py

+15 −4

Original line number	Diff line number	Diff line
		from uuid import UUID

		from sqlalchemy import delete, select
		from sqlalchemy import select, update
		from sqlalchemy.exc import IntegrityError
		from sqlalchemy.ext.asyncio import AsyncSession

		from open_exposure_gateway.adapters.database.mappers import AppRegistrationMapper
		from open_exposure_gateway.adapters.database.sql import AppRegistrationRow
		from open_exposure_gateway.adapters.errors import DuplicateAppRegistrationError
		from open_exposure_gateway.domain.models import AppRegistration
		from open_exposure_gateway.domain.models import AppRegistration, AppRegistrationStatus
		from open_exposure_gateway.ports.database.registration import AppRegistrationRepository

		_UNIQUE_VIOLATION = "23505"
		@@ -25,7 +25,10 @@ class SqlAppRegistrationRepository(AppRegistrationRepository):
		return AppRegistrationMapper.to_domain(row) if row is not None else None

		async def get_by_app_id(self, app_id: UUID) -> AppRegistration \| None:
		stmt = select(AppRegistrationRow).where(AppRegistrationRow.app_id == app_id)
		stmt = select(AppRegistrationRow).where(
		AppRegistrationRow.app_id == app_id,
		AppRegistrationRow.status != AppRegistrationStatus.DELETED,
		)
		row = await self._session.scalar(stmt)
		return AppRegistrationMapper.to_domain(row) if row is not None else None

		@@ -43,5 +46,13 @@ class SqlAppRegistrationRepository(AppRegistrationRepository):
		return saved

		async def delete_by_app_id(self, app_id: UUID) -> None:
		stmt = delete(AppRegistrationRow).where(AppRegistrationRow.app_id == app_id)
		"""Soft-delete: flip status to DELETED, keep the row."""
		stmt = (
		update(AppRegistrationRow)
		.where(
		AppRegistrationRow.app_id == app_id,
		AppRegistrationRow.status != AppRegistrationStatus.DELETED,
		)
		.values(status=AppRegistrationStatus.DELETED)
		)
		await self._session.execute(stmt)

src/open_exposure_gateway/adapters/database/sql.py

+11 −2

Original line number	Diff line number	Diff line
		@@ -14,6 +14,7 @@ from sqlalchemy import (
		Text,
		UniqueConstraint,
		func,
		text,
		)
		from sqlalchemy.dialects.postgresql import JSONB
		from sqlalchemy.dialects.postgresql import UUID as PG_UUID
		@@ -64,10 +65,18 @@ class AuditedMixin:

		class AppRegistrationRow(AuditedMixin, Base):
		__tablename__ = "app_registrations"
		__table_args__ = (Index("idx_app_registrations_tenant", "tenant_id"),)
		__table_args__ = (
		Index("idx_app_registrations_tenant", "tenant_id"),
		Index(
		"uq_app_registrations_app_id_active",
		"app_id",
		unique=True,
		postgresql_where=text("status <> 'DELETED'"),
		),
		)

		app_registration_id: Mapped[UUID] = mapped_column(PG_UUID(as_uuid=True), primary_key=True)
		app_id: Mapped[UUID] = mapped_column(PG_UUID(as_uuid=True), unique=True, nullable=False)
		app_id: Mapped[UUID] = mapped_column(PG_UUID(as_uuid=True), nullable=False)
		tenant_id: Mapped[str] = mapped_column(String(255), nullable=False)
		name: Mapped[str] = mapped_column(String(64), nullable=False)
		version: Mapped[str] = mapped_column(String(64), nullable=False)

src/open_exposure_gateway/application/mappers/edge_application_mapper.py

+8 −4

Original line number	Diff line number	Diff line
		@@ -277,7 +277,7 @@ def build_app_manifest(catalog: SRMCatalogPayload) -> AppManifest:
		)


		def build_app_instance_info(instance: SRMServiceInstance) -> AppInstanceInfo:
		def build_app_instance_info(instance: SRMServiceInstance, app_id: UUID) -> AppInstanceInfo:
		status = _SRM_STATE_TO_APP_INSTANCE_STATUS.get(instance.state, AppInstanceStatus.UNKNOWN)

		endpoint_info: list[ComponentEndpointInfo] = []
		@@ -298,7 +298,7 @@ def build_app_instance_info(instance: SRMServiceInstance) -> AppInstanceInfo:
		return AppInstanceInfo(
		appInstanceId=UUID(instance.service_instance_id),
		name=instance.name or instance.service_instance_id,
		appId=UUID(instance.service_specification_id),
		appId=app_id,
		appProvider=instance.app_provider_id,
		status=status,
		edgeCloudZoneId=UUID(instance.resource_zone_id)
		@@ -311,6 +311,7 @@ def build_app_instance_info(instance: SRMServiceInstance) -> AppInstanceInfo:
		def build_app_registration_translation(
		manifest: AppManifest,
		app_id: UUID,
		app_registration_id: UUID,
		tenant_id: str,
		app_provider_id: str,
		) -> AppRegistrationTranslation:
		@@ -414,6 +415,7 @@ def build_app_registration_translation(

		return AppRegistrationTranslation(
		app_id=app_id,
		app_registration_id=app_registration_id,
		tenant_id=tenant_id,
		app_provider_id=app_provider_id,
		name=manifest.name,
		@@ -542,7 +544,7 @@ def build_catalog_payload(translation: AppRegistrationTranslation) -> SRMCatalog

		return SRMCatalogPayload(
		service_specification=SRMServiceSpecEntry(
		id=str(translation.app_id),
		id=str(translation.app_registration_id),
		ref=str(translation.app_id),
		name=translation.name,
		version=translation.version,
		@@ -560,6 +562,7 @@ def build_app_deployment_translation(
		request: CreateAppInstanceRequest,
		operation_id: UUID,
		app_instance_id: UUID,
		app_registration_id: UUID,
		tenant_id: str,
		app_provider_id: str,
		correlation_id: str,
		@@ -567,6 +570,7 @@ def build_app_deployment_translation(
		) -> AppDeploymentTranslation:
		return AppDeploymentTranslation(
		app_id=request.appId,
		app_registration_id=app_registration_id,
		operation_id=operation_id,
		app_instance_id=app_instance_id,
		correlation_id=correlation_id,
		@@ -591,7 +595,7 @@ def build_deploy_command(
		correlation_id=translation.correlation_id,
		requested_at=requested_at,
		app_provider_id=translation.app_provider_id,
		service_specification_id=str(translation.app_id),
		service_specification_id=str(translation.app_registration_id),
		targets=[
		SRMDeployTarget(
		app_instance_id=str(translation.app_instance_id),

src/open_exposure_gateway/application/services/edge_application_management_service.py

+119 −26

Original line number	Diff line number	Diff line
		@@ -85,6 +85,20 @@ _APP_INSTANCE_COMPLETION_STATE_MAP: dict[str, AppInstanceState] = {
		}


		def _is_final(app_instance: AppInstance) -> bool:
		"""Whether no completion may move this instance any further."""
		return app_instance.state == AppInstanceState.TERMINATED


		def _log_stale_completion(app_instance: AppInstance, operation_id: UUID) -> None:
		logger.info(
		"stale_completion_ignored_for_final_app_instance",
		app_instance_id=str(app_instance.app_instance_id),
		operation_id=str(operation_id),
		state=app_instance.state.value,
		)


		class EdgeApplicationManagementService:
		def __init__(
		self,
		@@ -126,6 +140,11 @@ class EdgeApplicationManagementService:
		return zones

		async def get_apps(self, x_correlator: Optional[str] = None) -> list[AppManifest]:
		# TODO: scope by caller.tenant_id/app_provider_id once JWT auth is wired in
		# (dependencies.py currently hardcodes both to "placeholder"). Every other
		# endpoint here already threads these through; this one doesn't yet, and
		# tenant_id isn't in SRM's catalog response, so filtering must happen via
		# app_registration_repo, not srm_client.get_apps.
		catalogs = await self.srm_client.get_apps(x_correlator=x_correlator)
		manifests = []
		for catalog in catalogs:
		@@ -148,7 +167,15 @@ class EdgeApplicationManagementService:
		async def get_app(
		self, app_id: UUID, x_correlator: Optional[str] = None
		) -> AppManifestEnvelope:
		catalog = await self.srm_client.get_app(app_id=app_id, x_correlator=x_correlator)
		if self._app_registration_repo is None:
		raise RuntimeError("AppRegistrationRepository is not available")
		app_registration = await self._app_registration_repo.get_by_app_id(app_id)
		if app_registration is None:
		raise NotFoundException(message=f"App {app_id} not found")

		catalog = await self.srm_client.get_app(
		app_id=app_registration.app_registration_id, x_correlator=x_correlator
		)
		try:
		manifest = build_app_manifest(catalog)
		except (ValueError, TypeError, IndexError) as exc:
		@@ -176,17 +203,21 @@ class EdgeApplicationManagementService:
		"is not supported in this release"
		)

		app_registration_id = uuid4()
		translation = build_app_registration_translation(
		manifest, app_id, tenant_id, app_provider_id
		manifest, app_id, app_registration_id, tenant_id, app_provider_id
		)
		catalog_payload = build_catalog_payload(translation)
		created = await self.srm_client.create_catalog_service_specification(
		payload=catalog_payload.model_dump(mode="json"), x_correlator=x_correlator
		)
		if created.id != translation.app_id:
		if created.id != translation.app_registration_id:
		raise DownstreamServiceException(
		message="SRM confirmed a different service specification id than requested",
		details={"requested_id": str(translation.app_id), "confirmed_id": str(created.id)},
		details={
		"requested_id": str(translation.app_registration_id),
		"confirmed_id": str(created.id),
		},
		)

		if self._app_registration_repo is None:
		@@ -194,7 +225,7 @@ class EdgeApplicationManagementService:
		try:
		await self._app_registration_repo.save(
		AppRegistration(
		app_registration_id=uuid4(),
		app_registration_id=app_registration_id,
		app_id=translation.app_id,
		tenant_id=translation.tenant_id,
		name=translation.name,
		@@ -236,7 +267,9 @@ class EdgeApplicationManagementService:
		raise RuntimeError("AppInstanceRepository is not available")

		app_registration = await self._app_registration_repo.get_by_app_id(app_id)
		if app_registration is not None:
		if app_registration is None:
		raise NotFoundException(message=f"App {app_id} not found")

		has_instances = await self._app_instance_repo.exists_for_app_registration(
		app_registration.app_registration_id
		)
		@@ -245,7 +278,9 @@ class EdgeApplicationManagementService:
		message="App with a running application instance cannot be deleted"
		)

		await self.srm_client.delete_app(app_id=app_id, x_correlator=x_correlator)
		await self.srm_client.delete_app(
		app_id=app_registration.app_registration_id, x_correlator=x_correlator
		)
		await self._app_registration_repo.delete_by_app_id(app_id)

		async def create_app_instance(
		@@ -293,6 +328,7 @@ class EdgeApplicationManagementService:
		request,
		operation_id,
		app_instance_id,
		app_registration_id=app_registration.app_registration_id,
		tenant_id=tenant_id,
		app_provider_id=app_provider_id,
		correlation_id=correlation_id,
		@@ -370,13 +406,44 @@ class EdgeApplicationManagementService:
		region: Optional[str] = None,
		x_correlator: Optional[str] = None,
		) -> list[AppInstanceInfo]:
		resolved_app_id: Optional[UUID] = app_id
		if app_id is not None:
		if self._app_registration_repo is None:
		raise RuntimeError("AppRegistrationRepository is not available")
		app_registration = await self._app_registration_repo.get_by_app_id(app_id)
		if app_registration is None:
		return []
		resolved_app_id = app_registration.app_registration_id

		instances = await self.srm_client.get_app_instances(
		app_id=app_id,
		app_id=resolved_app_id,
		app_instance_id=app_instance_id,
		region=region,
		x_correlator=x_correlator,
		)
		return [build_app_instance_info(i) for i in instances]

		result: list[AppInstanceInfo] = []
		app_id_cache: dict[UUID, Optional[UUID]] = {}
		for instance in instances:
		if app_id is not None:
		instance_app_id: Optional[UUID] = app_id
		else:
		if self._app_registration_repo is None:
		raise RuntimeError("AppRegistrationRepository is not available")
		app_registration_id = UUID(instance.service_specification_id)
		if app_registration_id not in app_id_cache:
		owner = await self._app_registration_repo.get_by_id(app_registration_id)
		app_id_cache[app_registration_id] = owner.app_id if owner else None
		instance_app_id = app_id_cache[app_registration_id]
		if instance_app_id is None:
		logger.warning(
		"app_instance_listed_for_unresolvable_app_registration",
		app_instance_id=instance.service_instance_id,
		app_registration_id=instance.service_specification_id,
		)
		continue
		result.append(build_app_instance_info(instance, instance_app_id))
		return result

		async def delete_app_instance(
		self,
		@@ -413,6 +480,7 @@ class EdgeApplicationManagementService:
		status=OperationStatus.PENDING,
		subject=Subject.TASK_TERMINATE,
		app_registration_id=app_instance.app_registration_id,
		metadata={"app_instance_id": str(app_instance_id)},
		)
		)
		await self._app_instance_repo.save(
		@@ -476,6 +544,9 @@ class EdgeApplicationManagementService:
		app_instance_id=instance.service_instance_id,
		)
		continue
		if _is_final(app_instance):
		_log_stale_completion(app_instance, operation_id)
		continue
		if is_terminate and instance.status == "completed":
		state = AppInstanceState.TERMINATED
		else:
		@@ -486,11 +557,24 @@ class EdgeApplicationManagementService:
		updated_instances.append(saved)
		elif status == OperationStatus.FAILED:
		# Total failure carries no instances[] entries to match against.
		# POST /appinstances always creates exactly one app_instances row
		# per operation (ADR-0005), so fall back to that link rather than
		# leaving the pre-created row stuck at instantiating forever.
		if is_terminate:
		raw_app_instance_id = operation.metadata.get("app_instance_id")
		app_instance = (
		await self._app_instance_repo.get_by_id(UUID(raw_app_instance_id))
		if raw_app_instance_id is not None
		else None
		)
		if app_instance is None:
		logger.warning(
		"terminate_total_failure_missing_app_instance_id",
		operation_id=event.operation_id,
		)
		else:
		app_instance = await self._app_instance_repo.get_by_operation_id(operation_id)
		if app_instance is not None:
		if _is_final(app_instance):
		_log_stale_completion(app_instance, operation_id)
		else:
		saved = await self._app_instance_repo.save(
		app_instance.model_copy(update={"state": AppInstanceState.FAILED})
		)
		@@ -507,13 +591,22 @@ class EdgeApplicationManagementService:
		) -> None:
		if self._callback_registration_repo is None:
		return
		registration = await self._callback_registration_repo.get_by_operation_id(operation_id)
		if registration is None or not registration.is_active:
		return
		if self._callback_delivery_port is None or self._callback_delivery_repo is None:
		raise RuntimeError("CallbackDeliveryPort/CallbackDeliveryRepository is not available")

		registration_by_operation: dict[UUID, CallbackRegistration \| None] = {}
		for app_instance in app_instances:
		creating_operation_id = app_instance.operation_id
		if creating_operation_id not in registration_by_operation:
		registration_by_operation[
		creating_operation_id
		] = await self._callback_registration_repo.get_by_operation_id(
		creating_operation_id
		)
		registration = registration_by_operation[creating_operation_id]
		if registration is None or not registration.is_active:
		continue

		app_id = await self._resolve_app_id(app_instance.app_registration_id)
		if app_id is None:
		logger.warning(

src/open_exposure_gateway/domain/edge_application_management.py

+6 −7

Original line number	Diff line number	Diff line
		@@ -69,6 +69,7 @@ class RequiredResources(BaseModel):

		class AppRegistrationTranslation(BaseModel):
		app_id: UUID
		app_registration_id: UUID
		tenant_id: str
		app_provider_id: str
		name: str
		@@ -82,6 +83,7 @@ class AppRegistrationTranslation(BaseModel):

		class AppDeploymentTranslation(BaseModel):
		app_id: UUID
		app_registration_id: UUID
		operation_id: UUID
		app_instance_id: UUID
		correlation_id: str
		@@ -159,8 +161,6 @@ class SRMServiceSpecDescriptor(BaseModel):


		class SRMServiceSpecEntry(BaseModel):
		# Client-supplied specification id; SRM adopts it as the service_specification
		# primary key, which is what makes service_specification_id == app_id (ADR-0011).
		id: str
		ref: str
		name: str
		@@ -274,9 +274,6 @@ class SRMOperationCompleted(BaseModel):
		operation_id: str
		status: Literal["completed", "partially_completed", "failed"]
		service_order_id: str \| None = None
		# One entry per app instance produced (one per targeted zone); required
		# unless status == failed, since "failed" means none were produced
		# (srm/interface-contract.md §C.2).
		instances: list[SRMCompletedInstance] = []
		metadata: dict[str, Any] \| None = None
		error: dict[str, Any] \| None = None
		@@ -285,8 +282,10 @@ class SRMOperationCompleted(BaseModel):

		@model_validator(mode="after")
		def _require_error_when_failed(self) -> SRMOperationCompleted:
		if self.status == "failed" and self.error is None:
		raise ValueError("error is required when status is failed")
		if self.status == "failed" and not self.instances and self.error is None:
		raise ValueError(
		"error is required when status is failed and no instances were produced"
		)
		return self

		@model_validator(mode="after")

Admin message