SRM Refactor: Implementation of service terminate command

Paris Stentoumisrequested to merge
refactor/service-terminate-message into refactor/tf-sdk-ports

Summary

Implemented handling for command.srm.service.terminate using the same command workflow structure as the existing deploy message.

Changes

  • Added TerminateServiceCommandCoordinator
    • Performs stage-2 validation with SrmServiceTerminateV1
    • Publishes failed_before_start events for invalid answerable messages
    • Commits accepted state before publishing accepted
    • Runs completion in a separate transaction
    • Converts post-accept failures into terminal failed completion events
  • Added TerminateServiceUseCase
    • Creates terminate_service orders
    • Supports idempotent replay by operation_id
    • Resolves the target service_instance
    • Marks child capability_instance rows as terminated
    • Marks the parent service_instance as terminated
    • Marks the service_order as completed
    • Includes a TF-SDK termination placeholder before child capability state updates
  • Wired terminate handling into DataBus composition
    • Added terminate use-case factory wiring
    • Replaced the previous terminate no-op subscriber callback with the terminate coordinator
  • Extended runtime inventory repositories
    • Added capability instance update_state(...)
    • Added optional terminated_at support for service/capability state updates
  • Added focused tests
    • Terminate use-case behavior
    • Terminate command coordinator behavior
    • DataBus dependency wiring
    • NATS subscriber coordinator delegation
    • Repository persistence coverage for capability termination state updates

Merge request reports

src/srm/adapters/database/repos/runtime_inventory.py

+28 −1
Changes for src/srm/adapters/database/repos/runtime_inventory.py: 28 added lines, 1 removed line.
Original line number Diff line number Diff line
@@ -20,7 +20,11 @@ from srm.domain.models.runtime_inventory import ( 
    ServiceInstance,

    ServiceOrder,

)

from srm.domain.models.runtime_inventory.enums import ServiceInstanceState, ServiceOrderState

from srm.domain.models.runtime_inventory.enums import (

    CapabilityInstanceState,

    ServiceInstanceState,

    ServiceOrderState,

)

from srm.domain.ports.database.runtime_inventory import (

    CapabilityInstanceRepository,

    ServiceInstanceRepository,
@@ -140,11 +144,15 @@ class SqlServiceInstanceRepository(ServiceInstanceRepository): 
        self,

        id: UUID,

        state: ServiceInstanceState,

        *,

        terminated_at: datetime | None = None,

    ) -> ServiceInstance:

        row = await self._session.get(ServiceInstanceRow, id)

        if row is None:

            raise RuntimeError("Service instance could not be reloaded")

        row.state = state

        if terminated_at is not None:

            row.terminated_at = terminated_at

        await self._session.flush()

        saved = await self.get_by_id(id)

        if saved is None:
@@ -174,3 +182,22 @@ class SqlCapabilityInstanceRepository(CapabilityInstanceRepository): 
        if saved is None:

            raise RuntimeError("Created capability instance could not be reloaded")

        return saved



    async def update_state(

        self,

        id: UUID,

        state: CapabilityInstanceState,

        *,

        terminated_at: datetime | None = None,

    ) -> CapabilityInstance:

        row = await self._session.get(CapabilityInstanceRow, id)

        if row is None:

            raise RuntimeError("Capability instance could not be reloaded")

        row.state = state

        if terminated_at is not None:

            row.terminated_at = terminated_at

        await self._session.flush()

        saved = await self.get_by_id(id)

        if saved is None:

            raise RuntimeError("Updated capability instance could not be reloaded")

        return saved

src/srm/api/databus/dependencies.py

+32 −0
Changes for src/srm/api/databus/dependencies.py: 32 added lines, 0 removed lines.
Original line number Diff line number Diff line
@@ -14,7 +14,9 @@ from srm.adapters.database.repos.topology import SqlZoneRepository 
from srm.adapters.databus.nats_connection_manager import NatsConnectionManager

from srm.adapters.databus.nats_publisher import NatsPublisher

from srm.application.command_handlers.deploy_service import DeployServiceCommandCoordinator

from srm.application.command_handlers.terminate_service import TerminateServiceCommandCoordinator

from srm.application.use_cases.deploy_service import DeployServiceUseCase

from srm.application.use_cases.terminate_service import TerminateServiceUseCase

from srm.domain.ports.databus.publisher import DataBusPublisher
@@ -50,3 +52,33 @@ def get_deploy_service_command_coordinator( 
        use_case_factory=build_deploy_service_use_case,

        publisher=NatsPublisher(connection_manager),

    )





def get_terminate_service_use_case(

    session: AsyncSession,

    connection_manager: NatsConnectionManager,

) -> TerminateServiceUseCase:

    return build_terminate_service_use_case(session, NatsPublisher(connection_manager))





def build_terminate_service_use_case(

    session: AsyncSession,

    publisher: DataBusPublisher,

) -> TerminateServiceUseCase:

    return TerminateServiceUseCase(

        service_orders=SqlServiceOrderRepository(session),

        service_instances=SqlServiceInstanceRepository(session),

        capability_instances=SqlCapabilityInstanceRepository(session),

        publisher=publisher,

    )





def get_terminate_service_command_coordinator(

    session_maker: async_sessionmaker[AsyncSession],

    connection_manager: NatsConnectionManager,

) -> TerminateServiceCommandCoordinator:

    return TerminateServiceCommandCoordinator(

        session_maker=session_maker,

        use_case_factory=build_terminate_service_use_case,

        publisher=NatsPublisher(connection_manager),

    )

src/srm/api/databus/nats_subscriber.py

+14 −3
Changes for src/srm/api/databus/nats_subscriber.py: 14 added lines, 3 removed lines.
Original line number Diff line number Diff line
@@ -7,7 +7,10 @@ from pydantic import ValidationError 
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker



from srm.adapters.databus.nats_connection_manager import NatsConnectionManager

from srm.api.databus.dependencies import get_deploy_service_command_coordinator

from srm.api.databus.dependencies import (

    get_deploy_service_command_coordinator,

    get_terminate_service_command_coordinator,

)

from srm.api.databus.schemas import AnswerableCommandV1, InboundMessage



logger: structlog.BoundLogger = structlog.get_logger(__name__)
@@ -27,13 +30,16 @@ async def _noop_callback(envelope: AnswerableCommandV1) -> None: 


COMMAND_HANDLERS: dict[str, MessageHandler] = {

    "command.srm.service.scale": _noop_callback,

    "command.srm.service.terminate": _noop_callback,

    "command.srm.network.capability.activate": _noop_callback,

    "command.srm.network.capability.update": _noop_callback,

    "command.srm.network.capability.deactivate": _noop_callback,

}



COMMAND_SUBJECTS = ("command.srm.service.deploy", *COMMAND_HANDLERS.keys())

COMMAND_SUBJECTS = (

    "command.srm.service.deploy",

    "command.srm.service.terminate",

    *COMMAND_HANDLERS.keys(),

)





async def subscribe_to_subjects(
@@ -44,9 +50,14 @@ async def subscribe_to_subjects( 
        session_maker,

        connection_manager,

    )

    terminate_coordinator = get_terminate_service_command_coordinator(

        session_maker,

        connection_manager,

    )



    command_handlers = {

        "command.srm.service.deploy": deploy_coordinator.handle,

        "command.srm.service.terminate": terminate_coordinator.handle,

        **COMMAND_HANDLERS,

    }

    subscribers = [

src/srm/application/command_handlers/terminate_service.py

0 → 100644
+257 −0
Changes for src/srm/application/command_handlers/terminate_service.py: 257 added lines, 0 removed lines.
Original line number Diff line number Diff line 
from __future__ import annotations



from collections.abc import Callable

from datetime import UTC, datetime

from uuid import UUID



from pydantic import ValidationError

from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker



from srm.api.databus.schemas import AnswerableCommandV1, SrmServiceTerminateV1

from srm.application.use_cases.terminate_service import (

    TerminatePayloadCommand,

    TerminateServiceCommand,

    TerminateServiceResult,

    TerminateServiceUseCase,

)

from srm.domain.ports.databus.events import (

    OperationCompletedStatus,

    OperationStatusState,

    SrmOperationCompletedV1,

    SrmOperationStatusV1,

)

from srm.domain.ports.databus.publisher import DataBusPublisher



TerminateUseCaseFactory = Callable[[AsyncSession, DataBusPublisher], TerminateServiceUseCase]





class TerminateServiceCommandCoordinator:

    def __init__(

        self,

        session_maker: async_sessionmaker[AsyncSession],

        use_case_factory: TerminateUseCaseFactory,

        publisher: DataBusPublisher,

    ) -> None:

        self._session_maker = session_maker

        self._use_case_factory = use_case_factory

        self._publisher = publisher



    async def handle(self, envelope: AnswerableCommandV1) -> None:

        command = await self._parse_command(envelope)

        if command is None:

            return



        accepted = await self._accept(command)

        if accepted.completed_event is not None:

            await self._publisher.publish(

                "event.srm.operation.completed",

                accepted.completed_event.model_dump(mode="json"),

            )

            return

        if accepted.duplicate_order_id is not None:

            await self._publish_accepted(envelope, accepted.duplicate_order_id)

            return

        if accepted.accepted_order_id is None:

            return



        await self._publish_accepted(envelope, accepted.accepted_order_id)



        try:

            completed_event = await self._complete(command, accepted)

        except Exception as exc:

            completed_event = await self._fail_after_accept(command, accepted, exc)

        if completed_event is not None:

            await self._publisher.publish(

                "event.srm.operation.completed",

                completed_event.model_dump(mode="json"),

            )



    async def _parse_command(

        self,

        envelope: AnswerableCommandV1,

    ) -> TerminateServiceCommand | None:

        try:

            terminate_message = SrmServiceTerminateV1.model_validate(

                envelope.model_dump(mode="python")

            )

        except ValidationError:

            await self._publish_failed_before_start_for_invalid_command(envelope)

            return None



        return TerminateServiceCommand(

            operation_id=terminate_message.operation_id,

            correlation_id=terminate_message.correlation_id,

            requested_at=terminate_message.requested_at,

            app_provider_id=terminate_message.app_provider_id,

            federation_partner_ref=terminate_message.federation_partner_ref,

            source=terminate_message.source,

            service_instance_id=terminate_message.service_instance_id,

            service_specification_id=terminate_message.service_specification_id,

            terminate=TerminatePayloadCommand(

                grace_period_seconds=terminate_message.terminate.grace_period_seconds,

            ),

            payload_snapshot=terminate_message.model_dump(mode="json"),

        )



    async def _accept(self, command: TerminateServiceCommand) -> TerminateServiceResult:

        async with self._session_maker() as session:

            try:

                use_case = self._use_case_factory(session, self._publisher)

                accepted = await use_case.accept(command)

                await session.commit()

                return accepted

            except Exception:

                await session.rollback()

                raise



    async def _complete(

        self,

        command: TerminateServiceCommand,

        accepted: TerminateServiceResult,

    ) -> SrmOperationCompletedV1 | None:

        async with self._session_maker() as session:

            use_case = self._use_case_factory(session, self._publisher)

            plan = await use_case.prepare_fulfill(command, accepted)



        if plan is None:

            return None



        terminated_at = datetime.now(UTC)

        for target in plan.capabilities:

            # Outside any session — TF-SDK side effect.

            # On raise: no new DB write; prior committed caps stay TERMINATED;

            # handle()'s except → fail_after_accept.

            await TerminateServiceUseCase.fulfill_one(target)



            async with self._session_maker() as session:

                try:

                    use_case = self._use_case_factory(session, self._publisher)

                    await use_case.persist_capability_terminated(

                        target.capability_instance_id,

                        terminated_at,

                    )

                    await session.commit()

                except Exception:

                    await session.rollback()

                    raise



        async with self._session_maker() as session:

            try:

                use_case = self._use_case_factory(session, self._publisher)

                completed_event = await use_case.persist_completed(

                    command,

                    accepted,

                    terminated_at=terminated_at,

                )

                await session.commit()

                return completed_event

            except Exception:

                await session.rollback()

                raise



    async def _fail_after_accept(

        self,

        command: TerminateServiceCommand,

        accepted: TerminateServiceResult,

        error: Exception,

    ) -> SrmOperationCompletedV1:

        async with self._session_maker() as session:

            try:

                use_case = self._use_case_factory(session, self._publisher)

                completed_event = await use_case.fail_after_accept(command, accepted, error)

                await session.commit()

                return completed_event

            except Exception:

                await session.rollback()

                return self._failed_after_accept_event(command, accepted, error)



    async def _publish_accepted(

        self,

        envelope: AnswerableCommandV1,

        service_order_id: UUID,

    ) -> None:

        model_extra = envelope.model_extra or {}

        status_event = SrmOperationStatusV1(

            schema_version="1.0",

            operation_id=envelope.operation_id,

            service_order_id=service_order_id,

            service_instance_id=model_extra.get("service_instance_id"),

            capability="terminate",

            state=OperationStatusState.ACCEPTED,

            correlation_id=envelope.correlation_id,

            emitted_at=datetime.now(UTC),

        )

        await self._publisher.publish(

            "event.srm.operation.status",

            status_event.model_dump(mode="json"),

        )



    async def _publish_failed_before_start_for_invalid_command(

        self,

        envelope: AnswerableCommandV1,

    ) -> None:

        now = datetime.now(UTC)

        status_event = SrmOperationStatusV1(

            schema_version="1.0",

            operation_id=envelope.operation_id,

            service_order_id=None,

            service_instance_id=None,

            capability="terminate",

            state=OperationStatusState.FAILED_BEFORE_START,

            correlation_id=envelope.correlation_id,

            emitted_at=now,

        )

        completed_event = SrmOperationCompletedV1(

            schema_version="1.0",

            operation_id=envelope.operation_id,

            status=OperationCompletedStatus.FAILED,

            service_order_id=None,

            error={

                "type": "about:blank",

                "title": "Terminate service command is invalid.",

                "status": 400,

            },

            correlation_id=envelope.correlation_id,

            completed_at=now,

        )

        await self._publisher.publish(

            "event.srm.operation.status",

            status_event.model_dump(mode="json"),

        )

        await self._publisher.publish(

            "event.srm.operation.completed",

            completed_event.model_dump(mode="json"),

        )



    def _failed_after_accept_event(

        self,

        command: TerminateServiceCommand,

        accepted: TerminateServiceResult,

        error: Exception,

    ) -> SrmOperationCompletedV1:

        now = datetime.now(UTC)

        problem = {

            "type": "about:blank",

            "title": "Terminate service failed after acceptance.",

            "status": 500,

            "detail": str(error),

        }

        return SrmOperationCompletedV1(

            schema_version="1.0",

            operation_id=command.operation_id,

            status=OperationCompletedStatus.FAILED,

            service_order_id=accepted.accepted_order_id,

            instances=[

                {

                    "service_instance_id": str(command.service_instance_id),

                    "zone_id": str(accepted.target_zone_id),

                    "status": "failed",

                    "error": problem,

                }

            ]

            if accepted.target_zone_id is not None

            else [],

            error=None if accepted.target_zone_id is not None else problem,

            correlation_id=command.correlation_id,

            completed_at=now,

        )

src/srm/application/use_cases/terminate_service.py

0 → 100644
+389 −0
Changes for src/srm/application/use_cases/terminate_service.py: 389 added lines, 0 removed lines.
Original line number Diff line number Diff line 
from __future__ import annotations



from dataclasses import dataclass, field

from datetime import UTC, datetime

from typing import Any, Literal

from uuid import UUID



import structlog



from srm.domain.models.runtime_inventory import ServiceOrder

from srm.domain.models.runtime_inventory.enums import (

    CapabilityInstanceState,

    ServiceInstanceState,

    ServiceOrderState,

    ServiceOrderType,

)

from srm.domain.ports.database.runtime_inventory import (

    CapabilityInstanceRepository,

    ServiceInstanceRepository,

    ServiceOrderRepository,

)

from srm.domain.ports.databus.events import (

    OperationCompletedStatus,

    OperationStatusState,

    SrmOperationCompletedV1,

    SrmOperationStatusV1,

)

from srm.domain.ports.databus.publisher import DataBusPublisher





@dataclass(frozen=True, slots=True)

class TerminatePayloadCommand:

    grace_period_seconds: int = 0





@dataclass(frozen=True, slots=True)

class TerminateServiceCommand:

    operation_id: UUID

    correlation_id: str

    requested_at: datetime

    app_provider_id: str

    source: Literal["nbi_camara", "nbi_tmf", "operator_portal", "federation"]

    service_instance_id: UUID

    terminate: TerminatePayloadCommand

    service_specification_id: UUID | None = None

    federation_partner_ref: str | None = None

    payload_snapshot: dict[str, Any] = field(default_factory=dict)





@dataclass(frozen=True, slots=True)

class TerminateServiceResult:

    accepted_order_id: UUID | None = None

    duplicate_order_id: UUID | None = None

    target_service_instance_id: UUID | None = None

    target_zone_id: UUID | None = None

    completed_event: SrmOperationCompletedV1 | None = None





@dataclass(frozen=True, slots=True)

class TerminateCapabilityTarget:

    capability_instance_id: UUID

    external_id: str | None

    control_path_ref_snapshot: str | None





@dataclass(frozen=True, slots=True)

class TerminateFulfillmentPlan:

    service_instance_id: UUID

    service_order_id: UUID

    zone_id: UUID

    capabilities: tuple[TerminateCapabilityTarget, ...]





class TerminateServiceUseCase:

    def __init__(

        self,

        service_orders: ServiceOrderRepository,

        service_instances: ServiceInstanceRepository,

        capability_instances: CapabilityInstanceRepository,

        publisher: DataBusPublisher,

    ) -> None:

        self._service_orders = service_orders

        self._service_instances = service_instances

        self._capability_instances = capability_instances

        self._publisher = publisher



    async def accept(self, command: TerminateServiceCommand) -> TerminateServiceResult:

        existing_order = await self._service_orders.get_by_operation_id(command.operation_id)

        if existing_order is not None:

            return await self._replay_existing_order(command, existing_order)



        service_instance = await self._service_instances.get_by_id(command.service_instance_id)

        if service_instance is None:

            await self._publish_failed_before_start(command, service_order_id=None)

            return TerminateServiceResult()

        if (

            service_instance.state == ServiceInstanceState.TERMINATED

            or service_instance.state == ServiceInstanceState.TERMINATING

        ):

            await self._publish_failed_before_start(command, service_order_id=None)

            return TerminateServiceResult()

        if service_instance.app_provider_id != command.app_provider_id:

            await self._publish_failed_before_start(command, service_order_id=None)

            return TerminateServiceResult()



        service_order = await self._service_orders.create(

            ServiceOrder(

                operation_id=command.operation_id,

                correlation_id=command.correlation_id,

                order_type=ServiceOrderType.TERMINATE_SERVICE,

                service_specification_id=(

                    command.service_specification_id or service_instance.service_specification_id

                ),

                target_service_instance_id=service_instance.id,

                state=ServiceOrderState.ACCEPTED,

                payload_snapshot=command.payload_snapshot,

                app_provider_id=command.app_provider_id,

                federation_partner_ref=command.federation_partner_ref,

            )

        )



        structlog.get_logger(__name__).info(

            "service_terminate_accepted",

            operation_id=str(command.operation_id),

            service_instance_id=str(command.service_instance_id),

        )

        return TerminateServiceResult(

            accepted_order_id=service_order.id,

            target_service_instance_id=service_instance.id,

            target_zone_id=service_instance.zone_id,

        )



    async def prepare_fulfill(

        self,

        command: TerminateServiceCommand,

        accepted: TerminateServiceResult,

    ) -> TerminateFulfillmentPlan | None:

        if accepted.accepted_order_id is None:

            return None



        service_order = await self._service_orders.get_by_id(accepted.accepted_order_id)

        if service_order is None:

            raise RuntimeError("Accepted terminate service order could not be reloaded")



        service_instance = await self._service_instances.get_by_id(command.service_instance_id)

        if service_instance is None:

            raise RuntimeError("Accepted terminate service instance could not be reloaded")



        return TerminateFulfillmentPlan(

            service_instance_id=service_instance.id,

            service_order_id=service_order.id,

            zone_id=service_instance.zone_id,

            capabilities=tuple(

                TerminateCapabilityTarget(

                    capability_instance_id=capability.id,

                    external_id=capability.external_id,

                    control_path_ref_snapshot=capability.control_path_ref_snapshot,

                )

                for capability in service_instance.capabilities

            ),

        )



    @staticmethod

    async def fulfill_one(target: TerminateCapabilityTarget) -> None:

        # TODO: TF-SDK CapabilityExecutionAdapter.deactivate_capability/terminate(...)

        # using target.external_id and target.control_path_ref_snapshot.

        # Must not touch repositories — runs outside any DB session.

        _ = target



    async def persist_capability_terminated(

        self,

        capability_instance_id: UUID,

        terminated_at: datetime,

    ) -> None:

        await self._capability_instances.update_state(

            capability_instance_id,

            CapabilityInstanceState.TERMINATED,

            terminated_at=terminated_at,

        )



    async def persist_completed(

        self,

        command: TerminateServiceCommand,

        accepted: TerminateServiceResult,

        *,

        terminated_at: datetime | None = None,

    ) -> SrmOperationCompletedV1 | None:

        if accepted.accepted_order_id is None:

            return None



        service_order = await self._service_orders.get_by_id(accepted.accepted_order_id)

        if service_order is None:

            raise RuntimeError("Accepted terminate service order could not be reloaded")



        service_instance = await self._service_instances.get_by_id(command.service_instance_id)

        if service_instance is None:

            raise RuntimeError("Accepted terminate service instance could not be reloaded")



        completed_at = terminated_at or datetime.now(UTC)

        await self._service_instances.update_state(

            service_instance.id,

            ServiceInstanceState.TERMINATED,

            terminated_at=completed_at,

        )

        await self._service_orders.update_state(

            service_order.id,

            ServiceOrderState.COMPLETED,

            completed_at=completed_at,

        )



        structlog.get_logger(__name__).info(

            "service_terminate_completed",

            operation_id=str(command.operation_id),

            service_instance_id=str(command.service_instance_id),

        )

        return SrmOperationCompletedV1(

            schema_version="1.0",

            operation_id=command.operation_id,

            status=OperationCompletedStatus.COMPLETED,

            service_order_id=service_order.id,

            instances=[

                {

                    "service_instance_id": str(service_instance.id),

                    "zone_id": str(service_instance.zone_id),

                    "status": "completed",

                }

            ],

            correlation_id=command.correlation_id,

            completed_at=completed_at,

        )



    async def fail_after_accept(

        self,

        command: TerminateServiceCommand,

        accepted: TerminateServiceResult,

        error: Exception,

    ) -> SrmOperationCompletedV1:

        if accepted.accepted_order_id is None:

            raise RuntimeError("Cannot fail terminate after accept without a service order")



        service_order = await self._service_orders.get_by_id(accepted.accepted_order_id)

        if service_order is None:

            raise RuntimeError("Accepted terminate service order could not be reloaded")



        completed_at = datetime.now(UTC)

        problem = {

            "type": "about:blank",

            "title": "Terminate service failed after acceptance.",

            "status": 500,

            "detail": str(error),

        }

        await self._service_orders.update_state(

            service_order.id,

            ServiceOrderState.FAILED,

            completed_at=completed_at,

            failure_detail=problem,

        )



        return SrmOperationCompletedV1(

            schema_version="1.0",

            operation_id=command.operation_id,

            status=OperationCompletedStatus.FAILED,

            service_order_id=service_order.id,

            instances=[

                {

                    "service_instance_id": str(command.service_instance_id),

                    "zone_id": str(accepted.target_zone_id),

                    "status": "failed",

                    "error": problem,

                }

            ]

            if accepted.target_zone_id is not None

            else [],

            error=None if accepted.target_zone_id is not None else problem,

            correlation_id=command.correlation_id,

            completed_at=completed_at,

        )



    async def _replay_existing_order(

        self,

        command: TerminateServiceCommand,

        service_order: ServiceOrder,

    ) -> TerminateServiceResult:

        if service_order.state == ServiceOrderState.COMPLETED:

            return TerminateServiceResult(

                duplicate_order_id=service_order.id,

                completed_event=await self._completed_event_from_order(

                    command,

                    service_order,

                    status=OperationCompletedStatus.COMPLETED,

                ),

            )

        if service_order.state == ServiceOrderState.FAILED:

            return TerminateServiceResult(

                duplicate_order_id=service_order.id,

                completed_event=await self._completed_event_from_order(

                    command,

                    service_order,

                    status=OperationCompletedStatus.FAILED,

                ),

            )



        return TerminateServiceResult(

            duplicate_order_id=service_order.id,

            target_service_instance_id=service_order.target_service_instance_id,

        )



    async def _completed_event_from_order(

        self,

        command: TerminateServiceCommand,

        service_order: ServiceOrder,

        *,

        status: OperationCompletedStatus,

    ) -> SrmOperationCompletedV1:

        service_instance = (

            await self._service_instances.get_by_id(service_order.target_service_instance_id)

            if service_order.target_service_instance_id is not None

            else None

        )

        instance_status = "completed" if status == OperationCompletedStatus.COMPLETED else "failed"

        problem = service_order.failure_detail or {

            "type": "about:blank",

            "title": "Terminate service failed.",

            "status": 500,

        }



        return SrmOperationCompletedV1(

            schema_version="1.0",

            operation_id=command.operation_id,

            status=status,

            service_order_id=service_order.id,

            instances=[

                {

                    "service_instance_id": str(service_instance.id),

                    "zone_id": str(service_instance.zone_id),

                    "status": instance_status,

                    **({"error": problem} if status == OperationCompletedStatus.FAILED else {}),

                }

            ]

            if service_instance is not None

            else [],

            error=(

                problem

                if status == OperationCompletedStatus.FAILED and service_instance is None

                else None

            ),

            correlation_id=command.correlation_id,

            completed_at=service_order.completed_at or datetime.now(UTC),

        )



    async def _publish_failed_before_start(

        self,

        command: TerminateServiceCommand,

        *,

        service_order_id: UUID | None,

    ) -> None:

        now = datetime.now(UTC)

        problem = {

            "type": "about:blank",

            "title": "Terminate service failed before start.",

            "status": 400,

        }

        status_event = SrmOperationStatusV1(

            schema_version="1.0",

            operation_id=command.operation_id,

            service_order_id=service_order_id,

            service_instance_id=command.service_instance_id,

            capability="terminate",

            state=OperationStatusState.FAILED_BEFORE_START,

            correlation_id=command.correlation_id,

            emitted_at=now,

        )

        completed_event = SrmOperationCompletedV1(

            schema_version="1.0",

            operation_id=command.operation_id,

            status=OperationCompletedStatus.FAILED,

            service_order_id=service_order_id,

            error=problem,

            correlation_id=command.correlation_id,

            completed_at=now,

        )

        await self._publisher.publish(

            "event.srm.operation.status",

            status_event.model_dump(mode="json"),

        )

        await self._publisher.publish(

            "event.srm.operation.completed",

            completed_event.model_dump(mode="json"),

        )