Refactor/edge apps - Complete POST appinstance and DELETE /appinstances/{appInstanceId} async lifecycle

George Papathanailrequested to merge
refactor/edge-apps into develop

Complete the POST /appinstances and DELETE /appinstances/{appInstanceId} async lifecycle Context A spec-compliance review against architecture/oeg/ found that POST /appinstances accepted requests and published to the DataBus, but skipped nearly every oeg_db write the flow doc requires, and the event.srm.operation.completed consumer was a stub that only logged incoming events — meaning no async operation ever became observable, retryable, or terminal. DELETE /appinstances/{appInstanceId} had the same gap in reverse: no persistence on the request side, and (once the completion consumer became real) no way to distinguish a terminate completion from a deploy completion.

This MR closes both gaps end-to-end, plus a couple of pre-existing tooling issues found along the way.

POST /appinstances:

  • Resolve appId against app_registrations before accepting the request; 400 INVALID_ARGUMENT for an unregistered app (matches the vendored spec's documented response codes for this operation no 404 is listed).
  • Wire Operation/AppInstance/CallbackRegistration repositories into the service and DI (previously only AppRegistrationRepository was connected, despite the other repos/adapters already existing).
  • Persist a PENDING operations row and an instantiating app_instances row before publishing command.srm.service.deploy, per the flow doc's write-before-publish requirement.
  • Idempotency-Key header support (a platform extension, not in the vendored CAMARA spec): a repeated key returns the existing AppInstanceInfo reflecting current instance status via a live oeg_db read, not a stale echo instead of creating a duplicate operation or republishing.
  • Persist callback_registrations from subscriptionRequest; sinkCredential is stored only as a secret:// reference, never the raw value (no secret-resolution mechanism exists yet, so nothing beyond the reference can be done with it today).

UC7 completion path (event.srm.operation.completed) The consumer previously did nothing but log. Now:

  • EdgeApplicationManagementService.handle_completed updates the operations row (status/result/error/completed_at) and the corresponding app_instances row(s) (ready/failed), including a fallback via operation_id for the case where a total failure carries an empty instances[].
  • NatsOperationConsumer is now a thin deserialize-and-delegate adapter instead of doing the DB work itself; main.py opens a session per message since there's no HTTP request to hang one off (mirrors get_session's commit/rollback shape by hand).
  • Webhook delivery: a new CallbackDeliveryPort + HttpCallbackClient (deliberately separate from SRMClientPort a per-subscription external URL, not the fixed internal SRM peer) deliver one CloudEvent per updated instance when an active callback_registrations row exists, recording the outcome in callback_deliveries. Delivery failures are caught and recorded, never propagated, so they can't roll back the operations/app_instances updates already made in the same transaction. No Authorization header is sent — the vendored spec's security requirement ([{}, notificationsBearerAuth: []]) makes unauthenticated delivery valid, not just a stopgap for the missing secret-resolution mechanism.

DELETE /appinstances/{appInstanceId}

  • Look up the app_instances row first (404 if unknown a documented response code for this operation); write a PENDING operations row (operation_type=TERMINATE) and set app_instances.state=TERMINATING, before publishing command.srm.service.terminate.
  • Fixed a bug this surfaced in the completion handler: it previously mapped every completed instance to ready, regardless of what kind of operation produced the completion so a successful termination would have been flipped back to ready instead of removed. handle_completed now branches on operation.operation_type: a TERMINATE completion deletes the app_instances row (no terminal "terminated" state exists in the domain model to mark it with instead); a failed termination still maps to failed, since the instance isn't necessarily gone.
Edited by George Papathanail

Merge request reports

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

+25 −1
Changes for src/open_exposure_gateway/adapters/database/repos/app_instances.py: 25 added lines, 1 removed line.
Original line number Diff line number Diff line
@@ -5,9 +5,11 @@ from sqlalchemy.ext.asyncio import AsyncSession 


from open_exposure_gateway.adapters.database.mappers import AppInstanceMapper

from open_exposure_gateway.adapters.database.sql import AppInstanceRow

from open_exposure_gateway.domain.models import AppInstance

from open_exposure_gateway.domain.models import AppInstance, AppInstanceState

from open_exposure_gateway.ports.database.instances import AppInstanceRepository



_TERMINAL_STATES = (AppInstanceState.FAILED, AppInstanceState.TERMINATED)





class SqlAppInstanceRepository(AppInstanceRepository):

    def __init__(self, session: AsyncSession) -> None:
@@ -18,6 +20,28 @@ class SqlAppInstanceRepository(AppInstanceRepository): 
        row = await self._session.scalar(stmt)

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



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

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

        row = await self._session.scalar(stmt)

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



    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,

            AppInstanceRow.state.notin_(_TERMINAL_STATES),

        )

        row = await self._session.scalar(stmt.limit(1))

        return row is not None



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

        stmt = select(AppInstanceRow.app_instance_id).where(

            AppInstanceRow.app_registration_id == app_registration_id,

            AppInstanceRow.edge_cloud_zone_id == edge_cloud_zone_id,

            AppInstanceRow.state.notin_(_TERMINAL_STATES),

        )

        row = await self._session.scalar(stmt.limit(1))

        return row is not None



    async def save(self, app_instance: AppInstance) -> AppInstance:

        merged = await self._session.merge(AppInstanceMapper.to_row(app_instance))

        await self._session.flush()

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

+18 −3
Changes for src/open_exposure_gateway/adapters/database/repos/app_registrations.py: 18 added lines, 3 removed lines.
Original line number Diff line number Diff line 
from uuid import UUID



from sqlalchemy import 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
@@ -41,3 +44,15 @@ class SqlAppRegistrationRepository(AppRegistrationRepository): 
        if saved is None:

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

        return saved



    async def delete_by_app_id(self, app_id: UUID) -> None:

        """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
Changes for src/open_exposure_gateway/adapters/database/sql.py: 11 added lines, 2 removed lines.
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/adapters/databus/nats_adapter.py

+20 −4
Changes for src/open_exposure_gateway/adapters/databus/nats_adapter.py: 20 added lines, 4 removed lines.
Original line number Diff line number Diff line 
import json

from collections.abc import Awaitable, Callable

from typing import Any, Protocol



import nats

import structlog

from nats.aio.client import Client

from nats.aio.subscription import Subscription

from pydantic import ValidationError



from open_exposure_gateway.core.config import NatsSettings

from open_exposure_gateway.domain.edge_application_management import SRMOperationCompleted

from open_exposure_gateway.ports.databus_port import DataBusPort



logger: structlog.BoundLogger = structlog.get_logger(__name__)
@@ -71,9 +74,15 @@ class _Msg(Protocol): 




class NatsOperationConsumer:

    def __init__(self, client: Client, subject: str) -> None:

    def __init__(

        self,

        client: Client,

        subject: str,

        handler: Callable[[SRMOperationCompleted], Awaitable[None]],

    ) -> None:

        self._client = client

        self._subject = subject

        self._handler = handler

        self._subscription: Subscription | None = None



    async def start(self) -> None:
@@ -90,6 +99,13 @@ class NatsOperationConsumer: 
            logger.warning("invalid_json", subject=msg.subject)

            return



        # TODO: parse operation.completed payload, update oeg_db operation record to

        # COMPLETED or FAILED, and trigger webhook callback if registered

        logger.info("operation_completed_received", subject=msg.subject, payload=raw)
 
        try:

            event = SRMOperationCompleted.model_validate(raw)

        except ValidationError as exc:

            logger.warning("invalid_operation_completed_event", subject=msg.subject, error=str(exc))

            return



        try:

            await self._handler(event)

        except Exception:

            logger.exception("operation_completed_handler_failed", operation_id=event.operation_id)

src/open_exposure_gateway/adapters/http/callback_client.py

0 → 100644
+24 −0
Changes for src/open_exposure_gateway/adapters/http/callback_client.py: 24 added lines, 0 removed lines.
Original line number Diff line number Diff line 
import httpx

import structlog



from open_exposure_gateway.core.config import get_settings

from open_exposure_gateway.domain.edge_application_management import (

    AppInstanceStatusChangeCloudEvent,

)



logger = structlog.get_logger(__name__)





class HttpCallbackClient:

    def __init__(self) -> None:

        settings = get_settings()

        self.timeout = settings.callback_settings.timeout



    async def deliver(self, sink: str, event: AppInstanceStatusChangeCloudEvent) -> None:

        async with httpx.AsyncClient(timeout=self.timeout) as client:

            response = await client.post(

                sink,

                content=event.model_dump_json(),

                headers={"Content-Type": "application/cloudevents+json"},

            )

        response.raise_for_status()