sientia-dataops-opc-ingestor/ingestor/gateway_runtime.py

"""GatewayRuntime (Task 18 F4 capstone): builds and owns the write/active-read/gRPC
object graph and wires it into the app's lifecycle.

Reuses the existing ingestor's logger/notification_handler/metrics_controller and
its already-connected DataManager (so CertProvider shares the existing Mongo
connection instead of opening a second one -- MongoDBRepository.find_one has the
exact (collection_name, filters, metadata) signature CertProvider.fetch calls).

# ponytail: subscription v2 swap + scheduled_tick wiring happen at 1974
# integration; when swapping, build ONE SubscriptionEngine per OPC server (its
# non_receive_count is engine-wide -- see plan ledger F1 carry-forward).
"""

from os import getenv

from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler
from sientia_do.notifications.models import NotificationLevel
from sientia_do.observability.logger import Logger
from sientia_do.observability.metrics_controller import MetricsController
from sientia_do.observability.sientia_monitoring import SientiaMonitoring
from sientia_do.repository.minio_repository import MinioRepository
from sientia_do.temporal.activities.postgres import Postgres
from sientia_do.utils.connectors_config import build_postgres_config

from ingestor.adapters.grpc_server import GrpcServer
from ingestor.adapters.nats_consumer import NatsWriteConsumer
from ingestor.config import GatewayConfig, _env_bool
from ingestor.connection.certs import CertProvider
from ingestor.connection.pool import OpcConnectionPool
from ingestor.engines.active_read import ActiveReadEngine
from ingestor.engines.write import WriteEngine
from ingestor.io.data_sink import DataSink
from ingestor.managers.data_manager import DataManager


class GatewayRuntime(SientiaMonitoring):
    """Owns the gateway write/active-read/gRPC object graph. Resilient to NATS
    being unreachable at startup: the read path (existing subscription loop,
    gRPC on-demand reads) must keep working even if the write path can't
    connect yet -- so a NATS connect failure is logged/notified, not raised,
    and retried from `idle_sweep()` on each main-loop iteration."""

    def __init__(
        self,
        config: GatewayConfig,
        data_manager: DataManager,
        logger: Logger,
        notification_handler: NotificationHandler,
        metrics_controller: MetricsController,
        metadata: dict | None = None,
    ):
        SientiaMonitoring.__init__(self, logger, notification_handler, metrics_controller)
        self.config = config
        self.metadata = metadata or {}
        self._nats_started = False
        self._nats_notified = False  # notify once per outage, not per 5s retry

        minio = MinioRepository(
            endpoint=getenv('MINIO_ENDPOINT', 'localhost:9000'),
            access_key=getenv('MINIO_ACCESS_KEY', 'minioadmin'),
            secret_key=getenv('MINIO_SECRET_KEY', 'minioadmin'),
            secure=_env_bool('MINIO_SECURE', True),
            bucket=config.cert_bucket,
            logger=logger,
            notification_handler=notification_handler,
            metrics_controller=metrics_controller,
        )
        cert_provider = CertProvider(
            mongo=data_manager.mongo_repository,
            minio=minio,
            cache_dir=config.cert_cache_dir,
            bucket=config.cert_bucket,
            metadata=self.metadata,
        )
        self.pool = OpcConnectionPool(cert_provider, logger, notification_handler, metrics_controller)

        postgres = Postgres(
            **build_postgres_config(),
            logger=logger,
            notification_handler=notification_handler,
            metrics_controller=metrics_controller,
        )
        self.sink = DataSink(data_manager, postgres)

        self.write_engine = WriteEngine(
            self.pool,
            self.sink,
            logger,
            notification_handler,
            metrics_controller,
            write_timestamp=config.write_timestamp,
            write_quality=config.write_quality,
            min_pause_s=config.write_min_pause_s,
        )
        self.nats_consumer = NatsWriteConsumer(
            config.nats_url,
            self.write_engine,
            logger,
            subject=config.nats_write_subject,
            queue=config.nats_queue,
        )
        self.active_read = ActiveReadEngine(
            self.pool, self.sink, logger, notification_handler, metrics_controller
        )

        # slot_lookup=None for now: address -> names resolution needs slot v2
        # (1974); until then ReadTag responses just aren't persisted.
        self.grpc_server = (
            GrpcServer(self.active_read, logger, port=config.grpc_port, slot_lookup=None)
            if config.grpc_enabled
            else None
        )

    async def start(self) -> None:
        await self._start_nats()
        if self.grpc_server is not None:
            await self.grpc_server.start()
        self.logger.info(
            f'GatewayRuntime started: nats_started={self._nats_started} '
            f'grpc_enabled={self.grpc_server is not None}'
        )

    async def stop(self) -> None:
        if self.grpc_server is not None:
            await self.grpc_server.stop()
        if self._nats_started:
            await self.nats_consumer.stop()
            self._nats_started = False
        await self.pool.close_all()

    async def idle_sweep(self) -> None:
        await self.pool.release_idle(self.config.idle_conn_max_s)
        if not self._nats_started:
            await self._start_nats()

    async def _start_nats(self) -> None:
        if self._nats_started:
            return
        try:
            await self.nats_consumer.start()
            self._nats_started = True
            self._nats_notified = False
        except Exception as e:
            self.logger.error(f'GatewayRuntime: failed to start NATS write consumer: {e}')
            if self._nats_notified:
                return
            self._nats_notified = True
            await self.send_notification_async(
                metadata=self.metadata,
                notification_id='GATEWAY_NATS_CONSUMER_START_FAILED',
                message=f'Failed to start NATS write consumer: {e}',
                block='gateway_runtime',
                level=NotificationLevel.ERROR,
            )