fix: deliver POST /deployments callback as one CloudEvents array

from collections import defaultdict

from datetime import datetime, timezone

from typing import Any, Optional

from uuid import UUID, uuid4

)

from open_exposure_gateway.application.mappers.edge_application_mapper import (

    build_app_deployment_info,

    build_app_deployment_status_change_events,

    build_app_deployment_translation,

    build_app_instance_info,

    build_app_instance_status_change_event,

    OperationType,

    PackageType,

)

from open_exposure_gateway.ports.callback_delivery_port import CallbackDeliveryPort

from open_exposure_gateway.ports.callback_delivery_port import (

    CallbackDeliveryPort,

    CallbackEvent,

)

from open_exposure_gateway.ports.database.callbacks import (

    CallbackDeliveryRepository,

    CallbackRegistrationRepository,

    }

)

# Vendored EAM event type. A subscription registered with this type gets the

# onAppDeploymentStatusChange contract (one delivery, array of CloudEvents);

# anything else falls through to the per-instance onAppInstanceStatusChange path.

_DEPLOYMENT_STATUS_CHANGE_EVENT_TYPE = (

    "org.camaraproject.edge-application-management.v0.app-deployment-status-change"

)

def _is_final(app_instance: AppInstance) -> bool:

    """Whether no completion may move this instance any further."""

        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] = {}

        # Group by the operation that CREATED each instance -- that's where the

        # subscription was registered (a terminate never re-homes the row). For a

        # multi-zone deployment all N instances share one creating operation, so

        # this yields a single group / registration.

        instances_by_operation: dict[UUID, list[AppInstance]] = defaultdict(list)

        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(

            instances_by_operation[app_instance.operation_id].append(app_instance)

        for creating_operation_id, group in instances_by_operation.items():

            registration = 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)

            # Every instance in the group shares one app_registration_id.

            app_id = await self._resolve_app_id(group[0].app_registration_id)

            if app_id is None:

                logger.warning(

                    "callback_skipped_unresolvable_app_id",

                    app_instance_id=str(app_instance.app_instance_id),

                    operation_id=str(creating_operation_id),

                )

                continue

            # onAppDeploymentStatusChange -> ONE delivery carrying an array of

            # CloudEvents; onAppInstanceStatusChange -> one delivery per instance

            # (ADR-0008). Each list entry below is exactly one HTTP POST and one

            # callback_deliveries row.

            deliveries: list[CallbackEvent] = []

            if _DEPLOYMENT_STATUS_CHANGE_EVENT_TYPE in registration.event_types:

                app_deployment_id = group[0].app_deployment_id

                if app_deployment_id is None:

                    logger.warning(

                        "callback_skipped_deployment_event_without_deployment_id",

                        operation_id=str(creating_operation_id),

                    )

                    continue

                deliveries.append(

                    build_app_deployment_status_change_events(

                        app_deployment_id, group, app_id, occurred_at

                    )

                )

            else:

                deliveries.extend(

                    build_app_instance_status_change_event(app_instance, app_id, occurred_at)

                    for app_instance in group

                )

            cloud_event = build_app_instance_status_change_event(app_instance, app_id, occurred_at)

            for event in deliveries:

                last_error: Optional[str] = None

                try:

                await self._callback_delivery_port.deliver(registration.sink, cloud_event)

                    await self._callback_delivery_port.deliver(registration.sink, event)

                    state = "delivered"

                except Exception as exc:

                    state = "failed"

                    last_error = str(exc)

                logger.warning("callback_delivery_failed", sink=registration.sink, error=str(exc))

                    logger.warning(

                        "callback_delivery_failed", sink=registration.sink, error=str(exc)

                    )

                await self._callback_delivery_repo.save(

                    CallbackDelivery(