Commit f6c3b2bc authored by Paris Stentoumis's avatar Paris Stentoumis
Browse files

feat: Service Specification/Application deployment via databus

parent 8f8a1554

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

+20 −0
Original line number Diff line number Diff line
@@ -147,6 +147,16 @@ class SqlServiceDeploymentUnitRepository(ServiceDeploymentUnitRepository): 
        row = await self._session.scalar(stmt)

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



    async def list_by_service_specification_id(

        self,

        service_specification_id: UUID,

    ) -> list[ServiceDeploymentUnit]:

        stmt = select(ServiceDeploymentUnitRow).where(

            ServiceDeploymentUnitRow.service_specification_id == service_specification_id

        )

        rows = (await self._session.scalars(stmt)).all()

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



    async def create(self, deployment_unit: ServiceDeploymentUnit) -> ServiceDeploymentUnit:

        row = ServiceDeploymentUnitMapper.to_row(deployment_unit)

        self._session.add(row)
@@ -173,6 +183,16 @@ class SqlServiceCapabilityRequirementRepository(ServiceCapabilityRequirementRepo 
        row = await self._session.scalar(stmt)

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



    async def list_by_service_specification_id(

        self,

        service_specification_id: UUID,

    ) -> list[ServiceCapabilityRequirement]:

        stmt = select(ServiceCapabilityRequirementRow).where(

            ServiceCapabilityRequirementRow.service_specification_id == service_specification_id

        )

        rows = (await self._session.scalars(stmt)).all()

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



    async def create(

        self,

        capability_requirement: ServiceCapabilityRequirement,

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

+56 −0
Original line number Diff line number Diff line 
from __future__ import annotations



from datetime import datetime

from uuid import UUID



import structlog
@@ -19,6 +20,7 @@ from srm.domain.models.runtime_inventory import ( 
    ServiceInstance,

    ServiceOrder,

)

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

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

    CapabilityInstanceRepository,

    ServiceInstanceRepository,
@@ -51,6 +53,11 @@ class SqlServiceOrderRepository(ServiceOrderRepository): 
        row = await self._session.scalar(stmt)

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



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

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

        row = await self._session.scalar(stmt)

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



    async def create(self, service_order: ServiceOrder) -> ServiceOrder:

        row = ServiceOrderMapper.to_row(service_order)

        self._session.add(row)
@@ -65,6 +72,26 @@ class SqlServiceOrderRepository(ServiceOrderRepository): 
            raise RuntimeError("Created service order could not be reloaded")

        return saved



    async def update_state(

        self,

        id: UUID,

        state: ServiceOrderState,

        *,

        completed_at: datetime | None = None,

        failure_detail: dict[str, object] | None = None,

    ) -> ServiceOrder:

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

        if row is None:

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

        row.state = state

        row.completed_at = completed_at

        row.failure_detail = failure_detail

        await self._session.flush()

        saved = await self.get_by_id(id)

        if saved is None:

            raise RuntimeError("Updated service order could not be reloaded")

        return saved





class SqlServiceInstanceRepository(ServiceInstanceRepository):

    def __init__(self, session: AsyncSession) -> None:
@@ -81,6 +108,20 @@ class SqlServiceInstanceRepository(ServiceInstanceRepository): 
        row = await self._session.scalar(stmt)

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



    async def list_by_originating_service_order_id(

        self,

        originating_service_order_id: UUID,

    ) -> list[ServiceInstance]:

        stmt = (

            select(ServiceInstanceRow)

            .options(

                selectinload(ServiceInstanceRow.capability_instances),

            )

            .where(ServiceInstanceRow.originating_service_order_id == originating_service_order_id)

        )

        rows = (await self._session.scalars(stmt)).all()

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



    async def create(self, service_instance: ServiceInstance) -> ServiceInstance:

        row = ServiceInstanceMapper.to_row(service_instance)

        self._session.add(row)
@@ -95,6 +136,21 @@ class SqlServiceInstanceRepository(ServiceInstanceRepository): 
            raise RuntimeError("Created service instance could not be reloaded")

        return saved



    async def update_state(

        self,

        id: UUID,

        state: ServiceInstanceState,

    ) -> ServiceInstance:

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

        if row is None:

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

        row.state = state

        await self._session.flush()

        saved = await self.get_by_id(id)

        if saved is None:

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

        return saved





class SqlCapabilityInstanceRepository(CapabilityInstanceRepository):

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

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

+15 −0
Original line number Diff line number Diff line
@@ -16,6 +16,7 @@ from srm.adapters.database.mappers import ( 
)

from srm.adapters.database.sql import CapabilityRow, ControlPathBindingRow, DomainRow, ZoneRow

from srm.domain.models.topology import Capability, ControlPathBinding, Domain, Zone

from srm.domain.models.topology.enums import ZoneKind, ZoneState

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

    CapabilityRepository,

    ControlPathBindingRepository,
@@ -57,6 +58,20 @@ class SqlZoneRepository(ZoneRepository): 
        row = await self._session.scalar(stmt)

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



    async def list_active_resource_zones(self) -> list[Zone]:

        stmt = (

            select(ZoneRow)

            .options(

                selectinload(ZoneRow.domains)

                .selectinload(DomainRow.capabilities)

                .selectinload(CapabilityRow.control_path_bindings)

            )

            .where(ZoneRow.kind == ZoneKind.RESOURCE, ZoneRow.state == ZoneState.ACTIVE)

            .order_by(ZoneRow.ref, ZoneRow.id)

        )

        rows = (await self._session.scalars(stmt)).all()

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



    async def create(self, zone: Zone) -> Zone:

        row = ZoneMapper.to_row(zone)

        self._session.add(row)

src/srm/api/databus/dependencies.py

0 → 100644
+53 −0
Original line number Diff line number Diff line 
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker



from srm.adapters.database.repos.catalog import (

    SqlServiceCapabilityRequirementRepository,

    SqlServiceDeploymentUnitRepository,

    SqlServiceSpecificationRepository,

)

from srm.adapters.database.repos.runtime_inventory import (

    SqlCapabilityInstanceRepository,

    SqlServiceInstanceRepository,

    SqlServiceOrderRepository,

)

from srm.adapters.database.repos.topology import SqlDomainRepository, 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.use_cases.deploy_service import DeployServiceUseCase

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





def get_deploy_service_use_case(

    session: AsyncSession,

    connection_manager: NatsConnectionManager,

) -> DeployServiceUseCase:

    return build_deploy_service_use_case(session, NatsPublisher(connection_manager))





def build_deploy_service_use_case(

    session: AsyncSession,

    publisher: DataBusPublisher,

) -> DeployServiceUseCase:

    return DeployServiceUseCase(

        service_specifications=SqlServiceSpecificationRepository(session),

        service_deployment_units=SqlServiceDeploymentUnitRepository(session),

        service_capability_requirements=SqlServiceCapabilityRequirementRepository(session),

        service_orders=SqlServiceOrderRepository(session),

        service_instances=SqlServiceInstanceRepository(session),

        capability_instances=SqlCapabilityInstanceRepository(session),

        zones=SqlZoneRepository(session),

        domains=SqlDomainRepository(session),

        publisher=publisher,

    )





def get_deploy_service_command_coordinator(

    session_maker: async_sessionmaker[AsyncSession],

    connection_manager: NatsConnectionManager,

) -> DeployServiceCommandCoordinator:

    return DeployServiceCommandCoordinator(

        session_maker=session_maker,

        use_case_factory=build_deploy_service_use_case,

        publisher=NatsPublisher(connection_manager),

    )

src/srm/api/databus/nats_subscriber.py

+51 −18
Original line number Diff line number Diff line
@@ -3,34 +3,59 @@ from collections.abc import Awaitable, Callable 
import structlog

from nats.aio.msg import Msg

from nats.aio.subscription import Subscription

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.schemas import InboundMessage

from srm.api.databus.dependencies import get_deploy_service_command_coordinator

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



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



COMMAND_SUBJECTS = [

    "command.srm.service.deploy",

    "command.srm.service.scale",

    "command.srm.service.terminate",

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

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

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

]



MessageHandler = Callable[[AnswerableCommandV1], Awaitable[None]]





async def _noop_callback(envelope: AnswerableCommandV1) -> None:

    # TODO: non-deploy command handlers must validate in two stages (interface-contract.md §A).

    logger.info(

        "Message Received",

        operation_id=str(envelope.operation_id),

        correlation_id=envelope.correlation_id,

    )





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,

}



async def _noop_router(message: InboundMessage) -> None:

    # TODO: the real router must validate in two stages (interface-contract.md §A, ADR-0032).

    logger.info("Message Received", subject=message.subject, data=message.payload)

    return None

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





async def subscribe_to_subjects(

    connection_manager: NatsConnectionManager,

    session_maker: async_sessionmaker[AsyncSession],

) -> list["NatsSubscriber"]:

    deploy_coordinator = get_deploy_service_command_coordinator(

        session_maker,

        connection_manager,

    )



    command_handlers = {

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

        **COMMAND_HANDLERS,

    }

    subscribers = [

        NatsSubscriber(connection_manager=connection_manager, subject=subject, router=_noop_router)

        for subject in COMMAND_SUBJECTS

        NatsSubscriber(

            connection_manager=connection_manager,

            subject=subject,

            callback=callback,

        )

        for subject, callback in command_handlers.items()

    ]



    for subscriber in subscribers:
@@ -44,11 +69,11 @@ class NatsSubscriber: 
        self,

        connection_manager: NatsConnectionManager,

        subject: str,

        router: Callable[[InboundMessage], Awaitable[None]],

        callback: MessageHandler,

    ) -> None:

        self._connection_manager = connection_manager

        self._subject = subject

        self._router = router

        self._callback = callback

        self._subscription: Subscription | None = None



    async def start(self) -> None:
@@ -64,6 +89,14 @@ class NatsSubscriber: 
            headers=dict(msg.headers) if msg.headers is not None else {},

        )

        try:

            await self._router(inbound_message)

            envelope = AnswerableCommandV1.model_validate_json(inbound_message.payload)

        except ValidationError:

            # Stage 1 failures are unanswerable: publish no SRM event;

            # route to JetStream/DLQ when wired.

            logger.exception("databus_stage1_unanswerable", subject=msg.subject)

            return



        try:

            await self._callback(envelope)

        except Exception:

            logger.exception("databus_handler_failed", subject=msg.subject)