feat: implement multi-zone edge application deployments (/deployments)

George Papathanailrequested to merge
refactor/deployments into develop

What this does

Implements the EAM /deployments resource end to end, adding multi-zone edge application deployment support on top of the existing single-instance flow.

Endpoints

Method Path Status in this MR
POST /deployments Implemented
GET /deployments (list) Implemented
DELETE /deployments/{id} Implemented (request + completion finalisation)
PATCH /deployments/{id} 501 stub (message only)

Key changes

  • Domain model: new AppDeployment aggregate for multi-zone deployments, with a state rollup over its child app_instances and a deployments enums module.
  • Persistence: new app_deployments table, FK'd from app_instances; AppDeploymentRepository port + SQL adapter; AppInstanceRepository gains list_by_app_deployment_id. Repo is wired through the DI chain.
  • Service layer: EdgeApplicationManagementService.create_app_deployment and get_app_deployments; multi-zone deploy command / deploy target mappers; build_app_deployment_info read mapper.
  • Completion handling: app_deployments.state is rolled up when an SRM operation completes, including multi-zone total-failure fallback (all app_instances flipped) and a guard against stale redelivered completions.
  • DELETE flow: request path issues the SRM teardown; completion path finalises the deployment (DELETED / ERROR).
  • Callbacks: CallbackDeliveryPort.deliver now takes a CloudEvent array; POST /deployments callback is delivered as a single CloudEvents array. New AppDeploymentStatusChange CloudEvent domain types + build_app_deployment_status_change_events mapper.
  • Validation / guards: POST /deployments rejects duplicate zones with 409; deployment sinkCredential is rejected with 501 rather than silently dropped.

Testing

  • New: test_app_deployment_repo.py, test_eam_endpoints.py
  • Extended: test_eam_service.py, test_eam_flows.py (e2e POST/DELETE flows), test_eam_mapper.py, test_app_instance_repo.py
  • Fakes updated: fake SRM worker + wire_app_deployment_repo for completion paths

Follow-ups (not in this MR)

  • Real PATCH /deployments logic (currently 501)
  • GET /clusters (still 501)
  • Real secret store for sinkCredential (currently 501)

Merge request reports

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

+10 −0
Changes for src/open_exposure_gateway/adapters/database/repos/app_instances.py: 10 added lines, 0 removed lines.
Original line number Diff line number Diff line
@@ -25,6 +25,16 @@ class SqlAppInstanceRepository(AppInstanceRepository): 
        row = await self._session.scalar(stmt)

        return AppInstanceMapper.to_domain(row) if row is not None else None



    async def list_by_operation_id(self, operation_id: UUID) -> list[AppInstance]:

        stmt = select(AppInstanceRow).where(AppInstanceRow.operation_id == operation_id)

        rows = await self._session.scalars(stmt)

        return [AppInstanceMapper.to_domain(row) for row in rows]



    async def list_by_app_deployment_id(self, app_deployment_id: UUID) -> list[AppInstance]:

        stmt = select(AppInstanceRow).where(AppInstanceRow.app_deployment_id == app_deployment_id)

        rows = await self._session.scalars(stmt)

        return [AppInstanceMapper.to_domain(row) for row in rows]



    async def exists_for_app_registration(self, app_registration_id: UUID) -> bool:

        stmt = select(AppInstanceRow.app_instance_id).where(

            AppInstanceRow.app_registration_id == app_registration_id,

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

0 → 100644
+46 −0
Changes for src/open_exposure_gateway/adapters/database/repos/deployments.py: 46 added lines, 0 removed lines.
Original line number Diff line number Diff line 
from uuid import UUID



from sqlalchemy import select

from sqlalchemy.ext.asyncio import AsyncSession



from open_exposure_gateway.adapters.database.mappers import AppDeploymentMapper

from open_exposure_gateway.adapters.database.sql import AppDeploymentRow

from open_exposure_gateway.domain.models import AppDeployment

from open_exposure_gateway.ports.database.deployments import AppDeploymentRepository





class SqlAppDeploymentRepository(AppDeploymentRepository):

    def __init__(self, session: AsyncSession) -> None:

        self._session = session



    async def get_by_id(self, app_deployment_id: UUID) -> AppDeployment | None:

        stmt = select(AppDeploymentRow).where(

            AppDeploymentRow.app_deployment_id == app_deployment_id

        )

        row = await self._session.scalar(stmt)

        return AppDeploymentMapper.to_domain(row) if row is not None else None



    async def get_by_operation_id(self, operation_id: UUID) -> AppDeployment | None:

        stmt = select(AppDeploymentRow).where(AppDeploymentRow.operation_id == operation_id)

        row = await self._session.scalar(stmt)

        return AppDeploymentMapper.to_domain(row) if row is not None else None



    async def list_by_app_registration_id(self, app_registration_id: UUID) -> list[AppDeployment]:

        stmt = select(AppDeploymentRow).where(

            AppDeploymentRow.app_registration_id == app_registration_id

        )

        rows = await self._session.scalars(stmt)

        return [AppDeploymentMapper.to_domain(row) for row in rows]



    async def list_all(self) -> list[AppDeployment]:

        stmt = select(AppDeploymentRow)

        rows = await self._session.scalars(stmt)

        return [AppDeploymentMapper.to_domain(row) for row in rows]



    async def save(self, app_deployment: AppDeployment) -> AppDeployment:

        merged = await self._session.merge(AppDeploymentMapper.to_row(app_deployment))

        await self._session.flush()

        saved = await self.get_by_id(merged.app_deployment_id)

        if saved is None:

            raise RuntimeError("Saved app deployment could not be reloaded")

        return saved

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

+21 −1
Changes for src/open_exposure_gateway/adapters/database/repos/operations.py: 21 added lines, 1 removed line.
Original line number Diff line number Diff line 
from collections.abc import Collection

from datetime import datetime

from uuid import UUID



from sqlalchemy import select
@@ -7,7 +9,7 @@ from sqlalchemy.ext.asyncio import AsyncSession 
from open_exposure_gateway.adapters.database.mappers import OperationMapper

from open_exposure_gateway.adapters.database.sql import OperationRow

from open_exposure_gateway.adapters.errors import DuplicateOperationError

from open_exposure_gateway.domain.models import Operation

from open_exposure_gateway.domain.models import Operation, OperationStatus, OperationType

from open_exposure_gateway.ports.database.operations import OperationRepository



_UNIQUE_VIOLATION = "23505"
@@ -44,3 +46,21 @@ class SqlOperationRepository(OperationRepository): 
        if saved is None:

            raise RuntimeError("Saved operation could not be reloaded")

        return saved



    async def list_stale(

        self,

        statuses: Collection[OperationStatus],

        operation_types: Collection[OperationType],

        older_than: datetime,

    ) -> list[Operation]:

        stmt = (

            select(OperationRow)

            .where(

                OperationRow.status.in_(statuses),

                OperationRow.operation_type.in_(operation_types),

                OperationRow.created_at < older_than,

            )

            .order_by(OperationRow.created_at)

        )

        rows = await self._session.scalars(stmt)

        return [OperationMapper.to_domain(row) for row in rows]

src/open_exposure_gateway/adapters/database/mappers.py

+30 −0
Changes for src/open_exposure_gateway/adapters/database/mappers.py: 30 added lines, 0 removed lines.
Original line number Diff line number Diff line 
from uuid import UUID



from open_exposure_gateway.adapters.database.sql import (

    AppDeploymentRow,

    AppInstanceRow,

    AppRegistrationRow,

    CallbackDeliveryRow,
@@ -7,6 +10,7 @@ from open_exposure_gateway.adapters.database.sql import ( 
    QodSessionRow,

)

from open_exposure_gateway.domain.models import (

    AppDeployment,

    AppInstance,

    AppRegistration,

    CallbackDelivery,
@@ -110,6 +114,32 @@ class AppInstanceMapper: 
        )





class AppDeploymentMapper:

    @staticmethod

    def to_domain(row: AppDeploymentRow) -> AppDeployment:

        return AppDeployment(

            app_deployment_id=row.app_deployment_id,

            operation_id=row.operation_id,

            app_registration_id=row.app_registration_id,

            app_deployment_name=row.app_deployment_name,

            edge_cloud_zones=[UUID(zone_id) for zone_id in row.edge_cloud_zones],

            state=row.state,

            created_at=row.created_at,

            updated_at=row.updated_at,

        )



    @staticmethod

    def to_row(domain: AppDeployment) -> AppDeploymentRow:

        return AppDeploymentRow(

            app_deployment_id=domain.app_deployment_id,

            operation_id=domain.operation_id,

            app_registration_id=domain.app_registration_id,

            app_deployment_name=domain.app_deployment_name,

            edge_cloud_zones=[str(zone_id) for zone_id in domain.edge_cloud_zones],

            state=domain.state,

        )





class QodSessionMapper:

    @staticmethod

    def to_domain(row: QodSessionRow) -> QodSession:

src/open_exposure_gateway/adapters/database/sql.py

+25 −3
Changes for src/open_exposure_gateway/adapters/database/sql.py: 25 added lines, 3 removed lines.
Original line number Diff line number Diff line
@@ -21,6 +21,7 @@ from sqlalchemy.dialects.postgresql import UUID as PG_UUID 
from sqlalchemy.ext.asyncio import AsyncAttrs

from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column



from open_exposure_gateway.domain.models.deployments.enums import AppDeploymentState

from open_exposure_gateway.domain.models.instances.enums import AppInstanceState

from open_exposure_gateway.domain.models.operations.enums import (

    OperationStatus,
@@ -165,6 +166,27 @@ class CallbackDeliveryRow(Base): 
    state: Mapped[str] = mapped_column(String(32), nullable=False)





class AppDeploymentRow(AuditedMixin, Base):

    __tablename__ = "app_deployments"

    __table_args__ = (

        Index("idx_app_deployments_operation", "operation_id"),

        Index("idx_app_deployments_registration", "app_registration_id"),

    )



    app_deployment_id: Mapped[UUID] = mapped_column(PG_UUID(as_uuid=True), primary_key=True)

    operation_id: Mapped[UUID] = mapped_column(

        ForeignKey("operations.operation_id"), nullable=False

    )

    app_registration_id: Mapped[UUID] = mapped_column(

        ForeignKey("app_registrations.app_registration_id"), nullable=False

    )

    app_deployment_name: Mapped[str] = mapped_column(String(64), nullable=False)

    edge_cloud_zones: Mapped[list[str]] = mapped_column(JSONB, nullable=False)

    state: Mapped[AppDeploymentState] = mapped_column(

        _enum_type(AppDeploymentState), nullable=False

    )





class AppInstanceRow(AuditedMixin, Base):

    __tablename__ = "app_instances"

    __table_args__ = (
@@ -177,9 +199,9 @@ class AppInstanceRow(AuditedMixin, Base): 
    operation_id: Mapped[UUID] = mapped_column(

        ForeignKey("operations.operation_id"), nullable=False

    )

    # No FK yet: app_deployments (multi-zone POST /deployments) is out of

    # scope for this MR.

    app_deployment_id: Mapped[UUID | None] = mapped_column(PG_UUID(as_uuid=True))

    app_deployment_id: Mapped[UUID | None] = mapped_column(

        ForeignKey("app_deployments.app_deployment_id")

    )

    app_registration_id: Mapped[UUID] = mapped_column(

        ForeignKey("app_registrations.app_registration_id"), nullable=False

    )