feat(traffic-influence): add TrafficInfluenceService (ff67d322) · Commits · OOP / code / open-exposure-gateway

src/open_exposure_gateway/application/services/traffic_influence_service.py

0 → 100644

+337 −0

Original line number	Diff line number	Diff line
		from datetime import datetime, timezone
		from typing import Optional
		from uuid import UUID, uuid4

		import structlog
		from pydantic import BaseModel

		from open_exposure_gateway.api.camara.traffic_influence.vwip.schemas import (
		PostTrafficInfluence,
		)
		from open_exposure_gateway.api.camara.traffic_influence.vwip.schemas import (
		TrafficInfluence as TrafficInfluenceResponse,
		)
		from open_exposure_gateway.api.camara.traffic_influence.vwip.schemas import (
		TrafficInfluenceState as ApiTrafficInfluenceState,
		)
		from open_exposure_gateway.application.mappers.traffic_influence_mapper import (
		build_activate_command,
		build_deactivate_command,
		)
		from open_exposure_gateway.core.exceptions import DownstreamServiceException, NotFoundException
		from open_exposure_gateway.domain.models import (
		CallbackRegistration,
		Operation,
		OperationStatus,
		OperationType,
		)
		from open_exposure_gateway.domain.models import (
		TrafficInfluence as TrafficInfluenceRecord,
		)
		from open_exposure_gateway.domain.models import (
		TrafficInfluenceState as RecordTrafficInfluenceState,
		)
		from open_exposure_gateway.domain.srm_events import SRMOperationCompleted
		from open_exposure_gateway.domain.traffic_influence import Subject
		from open_exposure_gateway.ports.database.callbacks import CallbackRegistrationRepository
		from open_exposure_gateway.ports.database.operations import OperationRepository
		from open_exposure_gateway.ports.database.traffic_influences import TrafficInfluenceRepository
		from open_exposure_gateway.ports.databus_port import DataBusPort
		from open_exposure_gateway.ports.srm_port import SRMClientPort

		logger: structlog.BoundLogger = structlog.get_logger(__name__)

		_OPERATION_COMPLETION_STATUS_MAP: dict[str, OperationStatus] = {
		"completed": OperationStatus.COMPLETED,
		"partially_completed": OperationStatus.PARTIALLY_COMPLETED,
		"failed": OperationStatus.FAILED,
		}

		# CallbackRegistration.api_family discriminator for Traffic Influence's subscriptionRequest.
		_API_FAMILY = "traffic_influence"


		class TrafficInfluenceService:
		def __init__(
		self,
		srm_client: SRMClientPort,
		publisher: Optional[DataBusPort] = None,
		operation_repo: Optional[OperationRepository] = None,
		traffic_influence_repo: Optional[TrafficInfluenceRepository] = None,
		callback_registration_repo: Optional[CallbackRegistrationRepository] = None,
		) -> None:
		self.srm_client = srm_client
		self._publisher = publisher
		self._operation_repo = operation_repo
		self._traffic_influence_repo = traffic_influence_repo
		self._callback_registration_repo = callback_registration_repo

		def _new_operation_metadata(self, x_correlator: Optional[str]) -> tuple[UUID, str, str]:
		operation_id = uuid4()
		correlation_id = x_correlator or str(uuid4())
		requested_at = datetime.now(timezone.utc).isoformat()
		return operation_id, correlation_id, requested_at

		async def _publish(self, subject: Subject, command: BaseModel, error_msg: str) -> None:
		if self._publisher is None:
		raise RuntimeError("DataBus publisher is not available")
		try:
		await self._publisher.publish(subject, command.model_dump(mode="json"))
		except Exception as exc:
		raise DownstreamServiceException(error_msg) from exc

		async def _resolve_resource_zone_id(
		self,
		request: PostTrafficInfluence,
		x_correlator: Optional[str],
		) -> str \| None:
		if request.edgeCloudZoneId is not None:
		return str(request.edgeCloudZoneId)
		if request.edgeCloudRegion is not None:
		zones = await self.srm_client.get_resource_zones(
		region=request.edgeCloudRegion,
		status=None,
		x_correlator=x_correlator,
		)
		if zones:
		return zones[0].resource_zone_id
		return None

		async def create_traffic_influence(
		self,
		request: PostTrafficInfluence,
		tenant_id: str,
		app_provider_id: str,
		x_correlator: Optional[str] = None,
		) -> TrafficInfluenceResponse:
		if self._operation_repo is None or self._traffic_influence_repo is None:
		raise RuntimeError("Operation/TrafficInfluence repositories are not available")

		operation_id, correlation_id, requested_at = self._new_operation_metadata(x_correlator)
		traffic_influence_id = uuid4()
		# edgeCloudRegion has no SRM field of its own (srm/canonical-parameters-schema.md
		# has no "region" concept); it's resolved locally to a zone before the command
		# is built, same zones lookup EAM already uses.
		resource_zone_id = await self._resolve_resource_zone_id(request, x_correlator)

		command = build_activate_command(
		request=request,
		operation_id=operation_id,
		correlation_id=correlation_id,
		requested_at=requested_at,
		# service_specification_id == appId for an already-onboarded app
		# (srm/interface-contract.md §E.4, ADR-0011) -- unlike QoD, which mints a
		# throwaway id because its sessions aren't tied to an onboarded catalog entry.
		service_specification_id=request.appId,
		app_provider_id=app_provider_id,
		resource_zone_id=resource_zone_id,
		)

		await self._operation_repo.save(
		Operation(
		operation_id=operation_id,
		correlation_id=correlation_id,
		tenant_id=tenant_id,
		app_provider_id=app_provider_id,
		operation_type=OperationType.TRAFFIC_INFLUENCE,
		status=OperationStatus.PENDING,
		subject=Subject.TASK_ACTIVATE,
		metadata={"app_id": str(request.appId)},
		)
		)

		source_filter = request.sourceTrafficFilters
		destination_filter = request.destinationTrafficFilters
		await self._traffic_influence_repo.save(
		TrafficInfluenceRecord(
		traffic_influence_id=traffic_influence_id,
		operation_id=operation_id,
		app_id=request.appId,
		edge_cloud_zone_id=request.edgeCloudZoneId,
		edge_cloud_region=request.edgeCloudRegion,
		source_port=source_filter.sourcePort if source_filter else None,
		destination_port=destination_filter.destinationPort if destination_filter else None,
		destination_protocol=(
		destination_filter.destinationProtocol if destination_filter else None
		),
		state=RecordTrafficInfluenceState.ORDERED,
		)
		)

		# subscriptionRequest is never sent to SRM (srm/canonical-parameters-schema.md:
		# "NOT stored -- OEG owns subscription/callback state"); it's persisted here into
		# the generic callback table instead. Delivery of onTrafficInfluenceChanged is a
		# separate, not-yet-built concern -- this only records the registration.
		if request.subscriptionRequest is not None and self._callback_registration_repo is not None:
		await self._callback_registration_repo.save(
		CallbackRegistration(
		id=uuid4(),
		operation_id=operation_id,
		tenant_id=tenant_id,
		api_family=_API_FAMILY,
		sink=request.subscriptionRequest.sink,
		event_types=list(request.subscriptionRequest.types),
		expires_at=request.subscriptionRequest.config.subscriptionExpireTime,
		)
		)

		await self._publish(
		Subject.TASK_ACTIVATE,
		command,
		"Failed to publish traffic influence activation",
		)

		return TrafficInfluenceResponse(
		trafficInfluenceID=traffic_influence_id,
		state=ApiTrafficInfluenceState.ORDERED,
		apiConsumerId=request.apiConsumerId,
		appId=request.appId,
		appInstanceId=request.appInstanceId,
		edgeCloudRegion=request.edgeCloudRegion,
		edgeCloudZoneId=request.edgeCloudZoneId,
		sourceTrafficFilters=request.sourceTrafficFilters,
		destinationTrafficFilters=request.destinationTrafficFilters,
		subscriptionRequest=request.subscriptionRequest,
		)

		async def handle_completed(self, event: SRMOperationCompleted) -> None:
		if self._operation_repo is None or self._traffic_influence_repo is None:
		raise RuntimeError("Operation/TrafficInfluence repositories are not available")

		operation_id = UUID(event.operation_id)
		operation = await self._operation_repo.get_by_id(operation_id)
		if operation is None:
		logger.warning(
		"operation_completed_for_unknown_operation", operation_id=event.operation_id
		)
		return

		# Not ours: event.srm.operation.completed is shared across every domain.
		if operation.operation_type not in (
		OperationType.TRAFFIC_INFLUENCE,
		OperationType.TRAFFIC_INFLUENCE_DEACTIVATE,
		):
		return

		status = _OPERATION_COMPLETION_STATUS_MAP[event.status]
		result = None
		if status != OperationStatus.FAILED:
		result = {"instances": [i.model_dump(mode="json") for i in event.instances]}
		await self._operation_repo.save(
		operation.model_copy(
		update={
		"status": status,
		"result": result,
		"error": event.error,
		"completed_at": datetime.fromisoformat(event.completed_at),
		}
		)
		)

		instance = event.instances[0] if event.instances else None
		succeeded = instance is not None and instance.status == "completed"

		if operation.operation_type == OperationType.TRAFFIC_INFLUENCE:
		traffic_influence = await self._traffic_influence_repo.get_by_operation_id(operation_id)
		if traffic_influence is None:
		logger.warning(
		"traffic_influence_completed_for_unknown_operation",
		operation_id=event.operation_id,
		)
		return
		update = (
		{"state": RecordTrafficInfluenceState.ACTIVE, "external_ref": instance.external_ref}
		if succeeded and instance is not None
		else {"state": RecordTrafficInfluenceState.ERROR}
		)
		await self._traffic_influence_repo.save(traffic_influence.model_copy(update=update))
		return

		# TRAFFIC_INFLUENCE_DEACTIVATE: the deactivate command mints its own
		# operation_id (distinct from the activate one the row is keyed by), so the
		# deactivated row is recovered from the operation's own metadata instead.
		raw_id = operation.metadata.get("traffic_influence_id")
		if raw_id is None:
		logger.warning(
		"deactivate_completed_without_traffic_influence_id",
		operation_id=event.operation_id,
		)
		return
		traffic_influence = await self._traffic_influence_repo.get_by_id(UUID(raw_id))
		if traffic_influence is None:
		return
		deactivate_update = {
		"state": RecordTrafficInfluenceState.DELETED
		if succeeded
		else RecordTrafficInfluenceState.ERROR
		}
		await self._traffic_influence_repo.save(
		traffic_influence.model_copy(update=deactivate_update)
		)

		async def get_traffic_influence(
		self,
		traffic_influence_id: str,
		x_correlator: Optional[str] = None,
		) -> TrafficInfluenceResponse:
		return await self.srm_client.get_traffic_influence(
		traffic_influence_id=traffic_influence_id,
		x_correlator=x_correlator,
		)

		async def list_traffic_influences(
		self,
		app_id: Optional[UUID] = None,
		x_correlator: Optional[str] = None,
		) -> list[TrafficInfluenceResponse]:
		return await self.srm_client.get_traffic_influences(
		app_id=app_id,
		x_correlator=x_correlator,
		)

		async def delete_traffic_influence(
		self,
		traffic_influence_id: str,
		tenant_id: str,
		app_provider_id: str,
		x_correlator: Optional[str] = None,
		) -> None:
		if self._operation_repo is None or self._traffic_influence_repo is None:
		raise RuntimeError("Operation/TrafficInfluence repositories are not available")

		traffic_influence = await self._traffic_influence_repo.get_by_id(UUID(traffic_influence_id))
		if traffic_influence is None or traffic_influence.external_ref is None:
		raise NotFoundException(message=f"Traffic influence {traffic_influence_id} not found")

		operation_id, correlation_id, requested_at = self._new_operation_metadata(x_correlator)
		command = build_deactivate_command(
		external_ref=traffic_influence.external_ref,
		operation_id=operation_id,
		correlation_id=correlation_id,
		requested_at=requested_at,
		app_provider_id=app_provider_id,
		)

		await self._operation_repo.save(
		Operation(
		operation_id=operation_id,
		correlation_id=correlation_id,
		tenant_id=tenant_id,
		app_provider_id=app_provider_id,
		operation_type=OperationType.TRAFFIC_INFLUENCE_DEACTIVATE,
		status=OperationStatus.PENDING,
		subject=Subject.TASK_DEACTIVATE,
		metadata={"traffic_influence_id": str(traffic_influence_id)},
		)
		)
		await self._traffic_influence_repo.save(
		traffic_influence.model_copy(
		update={"state": RecordTrafficInfluenceState.DELETION_IN_PROGRESS}
		)
		)

		await self._publish(
		Subject.TASK_DEACTIVATE,
		command,
		"Failed to publish traffic influence deactivation",
		)

Admin message