"""Explicit Phase 2B insert runtime assembled only on command execution.""" from __future__ import annotations import json import os import stat from dataclasses import dataclass, field from pathlib import Path from typing import Mapping from ..errors import ContractViolation, InvalidConfiguration from ..noco.write import HttpsJsonTransport, NocoApiInsertRepository, NocoInsertConfig from ..sidecar.writer_ledger import PostgresInsertLedger from .lifecycle import HttpLifecycleGate, LifecycleApiConfig from .models import ShadowReportDocument from .service import ControlledInsertService, InsertRunSummary _TRUE_VALUES = frozenset({"1", "true", "yes", "on"}) _FALSE_VALUES = frozenset({"0", "false", "no", "off"}) @dataclass(frozen=True, slots=True) class WriterRuntimeConfig: """Secret-bearing writer configuration loaded only by ``insert-run``.""" writer_enabled: bool legacy_writer_disabled: bool sidecar_database_url: str = field(repr=False) noco_base_url: str noco_api_token: str = field(repr=False) lifecycle_base_url: str lifecycle_api_token: str = field(repr=False) lifecycle_timeout_seconds: float @staticmethod def _boolean(environ: Mapping[str, str], name: str) -> bool: raw = environ.get(name) if raw is None: return False normalized = raw.strip().lower() if normalized in _TRUE_VALUES: return True if normalized in _FALSE_VALUES: return False allowed = ", ".join(sorted(_TRUE_VALUES | _FALSE_VALUES)) raise InvalidConfiguration(f"{name} must be a boolean value ({allowed})") @staticmethod def _required(environ: Mapping[str, str], name: str) -> str: value = environ.get(name, "").strip() if not value: raise InvalidConfiguration(f"{name} is required for insert-run") if len(value) > 8192: raise InvalidConfiguration(f"{name} exceeds the bounded length") return value @staticmethod def _timeout(environ: Mapping[str, str], name: str, default: float) -> float: raw = environ.get(name, str(default)).strip() try: value = float(raw) except ValueError as exc: raise InvalidConfiguration(f"{name} must be numeric") from exc if not 1 <= value <= 60: raise InvalidConfiguration(f"{name} must be between 1 and 60 seconds") return value @classmethod def from_env(cls, environ: Mapping[str, str] | None = None) -> "WriterRuntimeConfig": source = os.environ if environ is None else environ writer_enabled = cls._boolean(source, "OSM_LEAD_SOURCE_WRITER_ENABLED") legacy_disabled = cls._boolean( source, "OSM_LEAD_SOURCE_LEGACY_WRITER_DISABLED", ) if not writer_enabled: raise InvalidConfiguration("insert writer is disabled") if not legacy_disabled: raise InvalidConfiguration("legacy writer must be disabled before insert-run") return cls( writer_enabled=writer_enabled, legacy_writer_disabled=legacy_disabled, sidecar_database_url=cls._required( source, "OSM_LEAD_SOURCE_SIDECAR_DATABASE_URL", ), noco_base_url=cls._required(source, "OSM_LEAD_SOURCE_NOCO_BASE_URL"), noco_api_token=cls._required(source, "OSM_LEAD_SOURCE_NOCO_API_TOKEN"), lifecycle_base_url=cls._required( source, "OSM_LEAD_SOURCE_LIFECYCLE_URL", ), lifecycle_api_token=cls._required( source, "OSM_LEAD_SOURCE_LIFECYCLE_TOKEN", ), lifecycle_timeout_seconds=cls._timeout( source, "OSM_LEAD_SOURCE_LIFECYCLE_TIMEOUT_SECONDS", 15.0, ), ) def read_shadow_report(path: Path, *, max_bytes: int) -> str: """Read one bounded regular report without following a final symlink.""" if isinstance(max_bytes, bool) or not 1 <= max_bytes <= 512 * 1024 * 1024: raise ContractViolation("max_report_bytes exceeds the bounded range") source = path.expanduser() flags = os.O_RDONLY if hasattr(os, "O_NOFOLLOW"): flags |= os.O_NOFOLLOW try: descriptor = os.open(source, flags) except OSError as exc: raise InvalidConfiguration("shadow report could not be opened safely") from exc try: metadata = os.fstat(descriptor) if not stat.S_ISREG(metadata.st_mode): raise InvalidConfiguration("shadow report must be a regular file") if metadata.st_size <= 0: raise InvalidConfiguration("shadow report must not be empty") if metadata.st_size > max_bytes: raise InvalidConfiguration("shadow report exceeds max_report_bytes") with os.fdopen(descriptor, "rb", closefd=False) as handle: content = handle.read(max_bytes + 1) if len(content) > max_bytes: raise InvalidConfiguration("shadow report exceeds max_report_bytes") try: return content.decode("utf-8") except UnicodeDecodeError as exc: raise InvalidConfiguration("shadow report must be UTF-8 JSON") from exc finally: os.close(descriptor) def execute_insert_report( *, report_path: Path, expected_report_hash: str, approval_id: str, noco_table_id: str, max_report_bytes: int, max_candidates: int, max_inserts: int, run_id: str, run_timestamp: str, environ: Mapping[str, str] | None = None, ) -> InsertRunSummary: """Validate all local and durable gates, then run bounded API inserts.""" config = WriterRuntimeConfig.from_env(environ) document = ShadowReportDocument.from_json( read_shadow_report(report_path, max_bytes=max_report_bytes), expected_report_hash=expected_report_hash, max_candidates=max_candidates, ) repository = NocoApiInsertRepository( config=NocoInsertConfig( base_url=config.noco_base_url, table_id=noco_table_id, api_token=config.noco_api_token, ), transport=HttpsJsonTransport(), ) ledger = PostgresInsertLedger( database_url=config.sidecar_database_url, approval_id=approval_id, ) lifecycle_gate = HttpLifecycleGate( LifecycleApiConfig( base_url=config.lifecycle_base_url, api_token=config.lifecycle_api_token, timeout_seconds=config.lifecycle_timeout_seconds, ) ) service = ControlledInsertService( repository=repository, ledger=ledger, lifecycle_gate=lifecycle_gate, writer_enabled=config.writer_enabled, legacy_writer_disabled=config.legacy_writer_disabled, ) return service.run( document, run_id=run_id, run_timestamp=run_timestamp, max_inserts=max_inserts, ) def summary_json(summary: InsertRunSummary) -> str: """Emit a stable, secret-free insert summary.""" return json.dumps( { "already_applied_count": summary.already_applied_count, "candidate_count": summary.candidate_count, "duplicate_count": summary.duplicate_count, "in_progress_count": summary.in_progress_count, "inserted_count": summary.inserted_count, "observation_failure_count": summary.observation_failure_count, "report_hash": summary.report_hash, "restore_candidate_count": summary.restore_candidate_count, "suppressed_count": summary.suppressed_count, "website_suppressed_count": summary.website_suppressed_count, }, sort_keys=True, separators=(",", ":"), ) __all__ = [ "WriterRuntimeConfig", "execute_insert_report", "read_shadow_report", "summary_json", ]