"""Fail-closed orchestration for controlled, insert-only NocoDB writes.""" from __future__ import annotations from collections.abc import Iterator, Sequence from contextlib import AbstractContextManager, contextmanager from dataclasses import dataclass from enum import Enum from typing import Protocol, runtime_checkable from ..errors import ContractViolation, InvalidConfiguration from ..noco.write import DuplicateRecord, NocoApiInsertRepository from .lifecycle import LifecycleGate, LifecycleGateError from .models import InsertCandidate, ShadowReportDocument class ReservationState(str, Enum): CLAIMED = "claimed" ALREADY_APPLIED = "already_applied" IN_PROGRESS = "in_progress" @runtime_checkable class InsertLedger(Protocol): """Durable sidecar approval, lease, and action-ledger boundary.""" def assert_cutover_approved( self, document: ShadowReportDocument, *, table_id: str, max_inserts: int, ) -> None: """Fail unless an active approval exactly matches this run.""" def run_lease(self, *, owner: str) -> AbstractContextManager[None]: """Hold the global single-writer lease for the external write window.""" def reserve(self, candidate: InsertCandidate) -> ReservationState: """Claim the candidate idempotency key or return its durable state.""" def mark_applied(self, candidate: InsertCandidate, *, table_id: str, record_id: int) -> None: """Record a successful Noco insert.""" def mark_duplicate( self, candidate: InsertCandidate, *, duplicates: Sequence[DuplicateRecord], ) -> None: """Record that final duplicate evidence blocked the insert.""" def mark_failed(self, candidate: InsertCandidate, *, reason: str) -> None: """Record a safely retryable pre-POST failure without exposing secrets.""" class InMemoryInsertLedger: """Deterministic test ledger; production must inject a durable sidecar ledger.""" def __init__(self) -> None: self.states: dict[str, str] = {} self.record_ids: dict[str, int] = {} self._leased = False def assert_cutover_approved( self, document: ShadowReportDocument, *, table_id: str, max_inserts: int, ) -> None: if not isinstance(document, ShadowReportDocument): raise ContractViolation("document must be ShadowReportDocument") if not table_id.strip() or max_inserts <= 0: raise ContractViolation("test cutover evidence is invalid") @contextmanager def run_lease(self, *, owner: str) -> Iterator[None]: if not owner.strip(): raise ContractViolation("lease owner must be nonblank") if self._leased: raise InvalidConfiguration("insert writer lease is already held") self._leased = True try: yield finally: self._leased = False def reserve(self, candidate: InsertCandidate) -> ReservationState: state = self.states.get(candidate.idempotency_key) if state in {"applied", "duplicate"}: return ReservationState.ALREADY_APPLIED if state == "claimed": return ReservationState.IN_PROGRESS self.states[candidate.idempotency_key] = "claimed" return ReservationState.CLAIMED def mark_applied(self, candidate: InsertCandidate, *, table_id: str, record_id: int) -> None: if not table_id or record_id <= 0: raise ContractViolation("applied ledger evidence is invalid") self.states[candidate.idempotency_key] = "applied" self.record_ids[candidate.idempotency_key] = record_id def mark_duplicate( self, candidate: InsertCandidate, *, duplicates: Sequence[DuplicateRecord], ) -> None: if not duplicates: raise ContractViolation("duplicate ledger evidence cannot be empty") self.states[candidate.idempotency_key] = "duplicate" def mark_failed(self, candidate: InsertCandidate, *, reason: str) -> None: if not reason.strip(): raise ContractViolation("failure reason must be nonblank") self.states[candidate.idempotency_key] = "failed" @dataclass(frozen=True, slots=True) class InsertRunSummary: report_hash: str candidate_count: int inserted_count: int duplicate_count: int already_applied_count: int in_progress_count: int suppressed_count: int website_suppressed_count: int restore_candidate_count: int observation_failure_count: int class ControlledInsertService: """Insert only after enablement, approved cutover evidence, and final lookups.""" def __init__( self, *, repository: NocoApiInsertRepository, ledger: InsertLedger, lifecycle_gate: LifecycleGate, writer_enabled: bool, legacy_writer_disabled: bool, ) -> None: if not isinstance(repository, NocoApiInsertRepository): raise ContractViolation("repository must be NocoApiInsertRepository") if not isinstance(ledger, InsertLedger): raise ContractViolation("ledger must implement InsertLedger") if not isinstance(lifecycle_gate, LifecycleGate): raise ContractViolation("lifecycle_gate must implement LifecycleGate") self._repository = repository self._ledger = ledger self._lifecycle_gate = lifecycle_gate self._writer_enabled = writer_enabled self._legacy_writer_disabled = legacy_writer_disabled def run( self, document: ShadowReportDocument, *, run_id: str, run_timestamp: str, max_inserts: int, ) -> InsertRunSummary: if not self._writer_enabled: raise InvalidConfiguration("insert writer is disabled") if not self._legacy_writer_disabled: raise InvalidConfiguration("legacy writer must be disabled before insert-only writes") if not isinstance(document, ShadowReportDocument): raise ContractViolation("document must be ShadowReportDocument") if isinstance(max_inserts, bool) or not 1 <= max_inserts <= 10_000: raise ContractViolation("max_inserts exceeds the bounded range") lease_owner = run_id.strip() if not lease_owner or len(lease_owner) > 256: raise ContractViolation("run_id must be nonblank and bounded") self._ledger.assert_cutover_approved( document, table_id=self._repository.table_id, max_inserts=max_inserts, ) inserted = 0 duplicates = 0 already_applied = 0 in_progress = 0 suppressed = 0 website_suppressed = 0 restore_candidates = 0 observation_failures = 0 with self._ledger.run_lease(owner=lease_owner): for candidate in document.candidates: if inserted >= max_inserts: break lifecycle_decision = self._lifecycle_gate.evaluate(candidate) if lifecycle_decision.suppress_business: suppressed += 1 continue include_website = not lifecycle_decision.suppress_website if lifecycle_decision.suppress_website: website_suppressed += 1 if lifecycle_decision.decision == "RESTORE_CANDIDATE": restore_candidates += 1 reservation = self._ledger.reserve(candidate) if reservation is ReservationState.ALREADY_APPLIED: already_applied += 1 continue if reservation is ReservationState.IN_PROGRESS: in_progress += 1 continue try: duplicate_records = self._repository.find_duplicates( candidate, include_website=include_website, ) except Exception as exc: self._ledger.mark_failed(candidate, reason=type(exc).__name__) raise if duplicate_records: self._ledger.mark_duplicate(candidate, duplicates=duplicate_records) duplicates += 1 continue # Once POST begins, any transport failure is ambiguous: Noco may # have committed before the response was lost. Leave the action # claimed so an operator must reconcile it instead of retrying. record_id = self._repository.create( candidate.noco_payload( run_id=run_id, run_timestamp=run_timestamp, include_website=include_website, ) ) self._ledger.mark_applied( candidate, table_id=self._repository.table_id, record_id=record_id, ) inserted += 1 try: self._lifecycle_gate.observe( candidate, record_id=record_id, include_website=include_website, ) except LifecycleGateError: # The Noco insert is already durably applied. Surface the # missing observation without retrying an ambiguous POST. observation_failures += 1 return InsertRunSummary( report_hash=document.report_hash, candidate_count=len(document.candidates), inserted_count=inserted, duplicate_count=duplicates, already_applied_count=already_applied, in_progress_count=in_progress, suppressed_count=suppressed, website_suppressed_count=website_suppressed, restore_candidate_count=restore_candidates, observation_failure_count=observation_failures, ) __all__ = [ "ControlledInsertService", "InMemoryInsertLedger", "InsertLedger", "InsertRunSummary", "ReservationState", ]