"""Connection settings of the OPC server a write command targets (SIENTIAPDE-1975 GW1). The Laborious only knows the `server_id` of a write tag (pipeline `write_tags`); the connection (url, server URI, security) belongs to the gateway, which owns every OPC session. It is resolved with the same rule the orchestrator uses to build the read slots (SIENTIAPDE-1974): the tag database `connectors` first, the legacy `opc-servers` otherwise. """ from typing import Any from sientia_do.observability.logger import Logger from ingestor.connection.clock import MonotonicClock from ingestor.connection.pool import OpcConnectionPool CONNECTORS = 'connectors' OPC_SERVERS = 'opc-servers' def _id_forms(server_id: str) -> list[Any]: """`server_id` as stored: the tag database keeps ints, opc-servers strings or ints.""" forms: list[Any] = [server_id] if server_id.isdigit(): forms.append(int(server_id)) return forms def _connector_cfg(connector: dict[str, Any]) -> dict[str, Any] | None: if not connector.get('url'): return None return { 'url': connector['url'], 'server_uri': connector.get('server_uri'), 'use_security': bool(connector.get('use_security', False)), } def _opc_server_cfg(server: dict[str, Any]) -> dict[str, Any] | None: if not server.get('url'): return None return { 'url': server['url'], 'server_uri': server.get('uri'), # The certificate comes from MinIO now: a cert_path only says the server uses # security, as in the orchestrator's slot (SIENTIAPDE-1974 RF4). 'use_security': bool(server.get('cert_path')), } class ServerConfigResolver: """Resolve and cache the connection settings of an OPC server by `server_id`. Order: settings carried by the command itself (an explicit override; the C2 command of the Laborious carries none), the live pooled connection (what the read path opened, so writes never fight the reads over settings), then MongoDB. Lookups are cached for `ttl_s`; a server that is not found, for `missing_ttl_s`, so a pipeline pointing at a removed server does not query MongoDB on every command.""" def __init__( self, pool: OpcConnectionPool, mongo, logger: Logger, ttl_s: float = 60.0, missing_ttl_s: float = 10.0, metadata: dict | None = None, ): self.pool = pool self.mongo = mongo self.logger = logger self.ttl_s = ttl_s self.missing_ttl_s = missing_ttl_s self.metadata = metadata or {} self.clock = MonotonicClock() self._cache: dict[str, tuple[dict[str, Any] | None, float]] = {} async def resolve(self, server_id: str, cmd: dict) -> dict[str, Any] | None: """Connection settings for `server_id`, or None when no source knows it. Raises: Exception: When MongoDB cannot be read (the caller reports the write as failed). """ if cmd.get('url'): return { 'url': cmd['url'], 'server_uri': cmd.get('server_uri'), 'use_security': cmd.get('use_security', False), } server_id = str(server_id) pooled = self.pool.config_for(server_id) if pooled is not None: return pooled now = self.clock.elapsed() cached = self._cache.get(server_id) if cached is not None: cfg, cached_at = cached if now - cached_at < (self.ttl_s if cfg is not None else self.missing_ttl_s): return dict(cfg) if cfg is not None else None cfg = await self._lookup(server_id) self._cache[server_id] = (cfg, now) return dict(cfg) if cfg is not None else None async def _lookup(self, server_id: str) -> dict[str, Any] | None: ids = {'$in': _id_forms(server_id)} connector = await self.mongo.find_one( CONNECTORS, {'connector_id': ids, 'type': 'opc_ua', 'status': 'Active'}, self.metadata, ) cfg = _connector_cfg(connector) if connector else None if cfg is not None: return cfg server = await self.mongo.find_one(OPC_SERVERS, {'id': ids}, self.metadata) cfg = _opc_server_cfg(server) if server else None if cfg is None: self.logger.warning( f'OPC server {server_id} not found in {CONNECTORS} nor {OPC_SERVERS}' ) return cfg