"""Fail-closed central lifecycle gate used before controlled Noco inserts.""" from __future__ import annotations import http.client import json import ssl import urllib.parse from dataclasses import dataclass, field from typing import Mapping, Protocol, runtime_checkable from ..errors import ApplicationError, ContractViolation, InvalidConfiguration from .models import InsertCandidate _MAX_RESPONSE_BYTES = 2 * 1024 * 1024 _DEFAULT_REPOSITORY_LIFECYCLE_URL = "http://lead-website-enricher:3000" _SUPPORTED_DECISIONS = frozenset({"ALLOW", "SUPPRESS", "SUPPRESS_WEBSITE", "RESTORE_CANDIDATE"}) class LifecycleGateError(ApplicationError): """Raised when central lifecycle authorization cannot be established.""" @dataclass(frozen=True, slots=True) class LifecycleDecision: """One validated central lifecycle decision.""" decision: str reason: str scope: str | None entity_id: int | None site_key: str | None replacement_allowed: bool def __post_init__(self) -> None: if self.decision not in _SUPPORTED_DECISIONS: raise ContractViolation("lifecycle decision is unsupported") if not self.reason.strip() or len(self.reason) > 512: raise ContractViolation("lifecycle reason must be nonblank and bounded") @property def suppress_business(self) -> bool: return self.decision == "SUPPRESS" @property def suppress_website(self) -> bool: return self.decision == "SUPPRESS_WEBSITE" @runtime_checkable class LifecycleGate(Protocol): """Central cross-source suppression and observation boundary.""" def evaluate(self, candidate: InsertCandidate) -> LifecycleDecision: """Return the authoritative decision before any Noco insert.""" def observe( self, candidate: InsertCandidate, *, record_id: int, include_website: bool, ) -> None: """Record a successfully inserted Noco row without retrying the insert.""" @dataclass(frozen=True, slots=True) class LifecycleApiConfig: """Secret-bearing repository lifecycle API configuration.""" base_url: str api_token: str = field(repr=False) timeout_seconds: float = 15.0 def __post_init__(self) -> None: configured = self.base_url.strip().rstrip("/") if "/api/w/" in configured or "website-lifecycle-ingestion-gate" in configured: configured = _DEFAULT_REPOSITORY_LIFECYCLE_URL parsed = urllib.parse.urlsplit(configured) if parsed.scheme not in {"http", "https"} or not parsed.hostname: raise InvalidConfiguration("lifecycle base_url must be an HTTP(S) URL") if parsed.username or parsed.password or parsed.query or parsed.fragment: raise InvalidConfiguration( "lifecycle base_url must not contain credentials or query data" ) if parsed.path not in {"", "/"}: configured = configured.removesuffix("/lifecycle/evaluate-batch") configured = configured.removesuffix("/lifecycle/observe-batch") configured = configured.removesuffix("/lifecycle/probe") if not self.api_token.strip() or len(self.api_token) > 8192: raise InvalidConfiguration("lifecycle api_token must be nonblank and bounded") if not 1 <= self.timeout_seconds <= 60: raise InvalidConfiguration("lifecycle timeout_seconds is outside the supported range") object.__setattr__(self, "base_url", configured.rstrip("/")) class HttpLifecycleGate: """Direct, bounded client for the repository lifecycle API.""" def __init__(self, config: LifecycleApiConfig) -> None: if not isinstance(config, LifecycleApiConfig): raise ContractViolation("config must be LifecycleApiConfig") self._config = config @staticmethod def _candidate_row( candidate: InsertCandidate, *, record_id: int | None = None, include_website: bool = True, ) -> dict[str, object]: row: dict[str, object] = { "source": "OSM", "source_record_id": f"{candidate.osm_type}:{candidate.osm_id}", "osm_id": str(candidate.osm_id), "name": candidate.name, "website": candidate.website if include_website and candidate.website else "", "phone": candidate.phone or "", "address": candidate.address or "", "city": candidate.city or "", "state_region": candidate.state or "", "postcode": candidate.postcode or "", "country": candidate.country, "category": candidate.category, "subcategory": candidate.subcategory or "", } if record_id is not None: if isinstance(record_id, bool) or record_id <= 0: raise ContractViolation("record_id must be positive") row["Id"] = record_id row["noco_record_id"] = record_id return row def _request( self, pathname: str, body: Mapping[str, object], ) -> Mapping[str, object]: parsed = urllib.parse.urlsplit(self._config.base_url) payload = json.dumps(dict(body), separators=(",", ":")).encode() headers = { "Accept": "application/json", "Authorization": f"Bearer {self._config.api_token}", "Content-Type": "application/json", } hostname = parsed.hostname if hostname is None: raise LifecycleGateError("lifecycle API URL has no hostname") connection: http.client.HTTPConnection if parsed.scheme == "https": connection = http.client.HTTPSConnection( hostname, parsed.port, timeout=self._config.timeout_seconds, context=ssl.create_default_context(), ) else: connection = http.client.HTTPConnection( hostname, parsed.port, timeout=self._config.timeout_seconds, ) try: base_path = parsed.path.rstrip("/") request_path = f"{base_path}{pathname}" if base_path else pathname connection.request("POST", request_path, body=payload, headers=headers) response = connection.getresponse() raw = response.read(_MAX_RESPONSE_BYTES + 1) if len(raw) > _MAX_RESPONSE_BYTES: raise LifecycleGateError("lifecycle response exceeded the bounded size") if not 200 <= response.status < 300: raise LifecycleGateError(f"lifecycle API returned HTTP {response.status}") try: decoded = json.loads(raw or b"{}") except json.JSONDecodeError as exc: raise LifecycleGateError("lifecycle API returned invalid JSON") from exc if not isinstance(decoded, dict): raise LifecycleGateError("lifecycle API returned an unsupported response") return decoded except (OSError, http.client.HTTPException) as exc: raise LifecycleGateError("lifecycle API request failed") from exc finally: connection.close() def evaluate(self, candidate: InsertCandidate) -> LifecycleDecision: if not isinstance(candidate, InsertCandidate): raise ContractViolation("candidate must be InsertCandidate") response = self._request( "/lifecycle/evaluate-batch", { "rows": [self._candidate_row(candidate)], "record_observation": False, }, ) decisions = response.get("decisions") if response.get("ok") is not True or not isinstance(decisions, list) or len(decisions) != 1: raise LifecycleGateError("lifecycle evaluation response is incomplete") raw = decisions[0] if not isinstance(raw, dict): raise LifecycleGateError("lifecycle decision is malformed") decision = str(raw.get("decision") or "").strip() reason = str(raw.get("reason") or "").strip() if reason == "lifecycle-disabled": raise LifecycleGateError("central lifecycle gate is disabled") entity_id = raw.get("entity_id") if entity_id is not None and ( isinstance(entity_id, bool) or not isinstance(entity_id, int) ): raise LifecycleGateError("lifecycle entity_id is malformed") return LifecycleDecision( decision=decision, reason=reason, scope=str(raw.get("scope")) if raw.get("scope") is not None else None, entity_id=entity_id, site_key=str(raw.get("site_key")) if raw.get("site_key") is not None else None, replacement_allowed=bool(raw.get("replacement_allowed")), ) def observe( self, candidate: InsertCandidate, *, record_id: int, include_website: bool, ) -> None: response = self._request( "/lifecycle/observe-batch", { "rows": [ self._candidate_row( candidate, record_id=record_id, include_website=include_website, ) ], }, ) observations = response.get("observations") if ( response.get("ok") is not True or not isinstance(observations, list) or len(observations) != 1 ): raise LifecycleGateError("lifecycle observation response is incomplete") observation = observations[0] if not isinstance(observation, dict) or observation.get("observed") is not True: raise LifecycleGateError("lifecycle observation was not accepted") __all__ = [ "HttpLifecycleGate", "LifecycleApiConfig", "LifecycleDecision", "LifecycleGate", "LifecycleGateError", ]