"""Durable PostgreSQL approval and idempotency ledger for Phase 2B writes.""" from __future__ import annotations from collections.abc import Iterator, Sequence from contextlib import contextmanager from typing import Any from urllib.parse import urlsplit import psycopg from psycopg.rows import dict_row from psycopg.types.json import Jsonb from ..errors import ContractViolation, InvalidConfiguration from ..noco.write import DuplicateRecord from ..writer.models import InsertCandidate, ShadowReportDocument from ..writer.service import ReservationState _LOCK_NAME = "osm-lead-source-service:noco-insert-writer" def _database_dsn(database_url: str) -> str: raw = database_url.strip() if not raw: raise InvalidConfiguration("sidecar database URL is missing") parsed = urlsplit(raw) if parsed.scheme not in {"postgresql", "postgresql+psycopg"}: raise InvalidConfiguration("sidecar database URL must use PostgreSQL") if not parsed.hostname or not parsed.path or parsed.path == "/": raise InvalidConfiguration("sidecar database URL is incomplete") if parsed.fragment: raise InvalidConfiguration("sidecar database URL cannot contain a fragment") if parsed.scheme == "postgresql+psycopg": return "postgresql" + raw[len("postgresql+psycopg") :] return raw def _bounded_text(value: str, field: str, *, maximum: int = 256) -> str: result = value.strip() if not result or len(result) > maximum: raise ContractViolation(f"{field} must be nonblank and bounded") return result class PostgresInsertLedger: """Fail-closed cutover approval, advisory lease, and action ledger.""" def __init__(self, *, database_url: str, approval_id: str) -> None: self._dsn = _database_dsn(database_url) self._approval_id = _bounded_text(approval_id, "approval_id") def _connect(self) -> psycopg.Connection[dict[str, Any]]: return psycopg.connect(self._dsn, row_factory=dict_row) 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") bounded_table = _bounded_text(table_id, "table_id", maximum=128) if isinstance(max_inserts, bool) or not 1 <= max_inserts <= 10_000: raise ContractViolation("max_inserts exceeds the bounded range") with self._connect() as connection: row = connection.execute( """ SELECT report_hash, region_id, profile_id, profile_version, noco_table_id, approved_max_inserts FROM osm_lead_source.writer_cutover_approval WHERE approval_id = %s AND revoked_at IS NULL AND legacy_writer_disabled_at IS NOT NULL AND CURRENT_TIMESTAMP < expires_at """, (self._approval_id,), ).fetchone() if row is None: raise InvalidConfiguration("active writer cutover approval was not found") expected = { "report_hash": document.report_hash, "region_id": document.region_id, "profile_id": document.profile_id, "profile_version": document.profile_version, "noco_table_id": bounded_table, } for field, value in expected.items(): if row[field] != value: raise InvalidConfiguration(f"writer cutover approval does not match {field}") approved_max = row["approved_max_inserts"] if not isinstance(approved_max, int) or max_inserts > approved_max: raise InvalidConfiguration("requested inserts exceed the approved cutover cap") @contextmanager def run_lease(self, *, owner: str) -> Iterator[None]: _bounded_text(owner, "lease owner") connection = self._connect() connection.autocommit = True try: row = connection.execute( "SELECT pg_try_advisory_lock(hashtextextended(%s, 0)) AS acquired", (_LOCK_NAME,), ).fetchone() if row is None or row["acquired"] is not True: raise InvalidConfiguration("another insert writer currently holds the lease") try: yield finally: connection.execute( "SELECT pg_advisory_unlock(hashtextextended(%s, 0))", (_LOCK_NAME,), ) finally: connection.close() def reserve(self, candidate: InsertCandidate) -> ReservationState: if not isinstance(candidate, InsertCandidate): raise ContractViolation("candidate must be InsertCandidate") evidence = Jsonb( { "source": "controlled-insert-writer", "report_hash": candidate.report_hash, "payload_hash": candidate.payload_hash, } ) with self._connect() as connection: inserted = connection.execute( """ INSERT INTO osm_lead_source.insert_action_ledger (approval_id, idempotency_key, report_hash, region_id, profile_id, profile_version, osm_type, osm_id, payload_hash, state, evidence_json) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, 'claimed', %s) ON CONFLICT (idempotency_key) DO NOTHING RETURNING state """, ( self._approval_id, candidate.idempotency_key, candidate.report_hash, candidate.region_id, candidate.profile_id, candidate.profile_version, candidate.osm_type, candidate.osm_id, candidate.payload_hash, evidence, ), ).fetchone() if inserted is not None: return ReservationState.CLAIMED existing = connection.execute( """ SELECT approval_id, report_hash, region_id, profile_id, profile_version, osm_type, osm_id, payload_hash, state FROM osm_lead_source.insert_action_ledger WHERE idempotency_key = %s FOR UPDATE """, (candidate.idempotency_key,), ).fetchone() if existing is None: raise InvalidConfiguration("insert ledger conflict could not be resolved") expected: dict[str, object] = { "approval_id": self._approval_id, "report_hash": candidate.report_hash, "region_id": candidate.region_id, "profile_id": candidate.profile_id, "profile_version": candidate.profile_version, "osm_type": candidate.osm_type, "osm_id": candidate.osm_id, "payload_hash": candidate.payload_hash, } for field, value in expected.items(): if existing[field] != value: raise InvalidConfiguration( f"insert ledger idempotency conflict does not match {field}" ) state = existing["state"] if state in {"applied", "duplicate"}: return ReservationState.ALREADY_APPLIED if state == "claimed": return ReservationState.IN_PROGRESS if state != "failed": raise InvalidConfiguration("insert ledger contains an unsupported state") updated = connection.execute( """ UPDATE osm_lead_source.insert_action_ledger SET state = 'claimed', attempt_count = attempt_count + 1, failure_code = NULL, finalized_at = NULL, claimed_at = CURRENT_TIMESTAMP WHERE idempotency_key = %s AND state = 'failed' RETURNING id """, (candidate.idempotency_key,), ).fetchone() if updated is None: raise InvalidConfiguration("failed insert action could not be reclaimed") return ReservationState.CLAIMED def mark_applied(self, candidate: InsertCandidate, *, table_id: str, record_id: int) -> None: bounded_table = _bounded_text(table_id, "table_id", maximum=128) if isinstance(record_id, bool) or record_id <= 0: raise ContractViolation("record_id must be positive") with self._connect() as connection: row = connection.execute( """ UPDATE osm_lead_source.insert_action_ledger SET state = 'applied', noco_table_id = %s, noco_record_id = %s, finalized_at = CURRENT_TIMESTAMP WHERE idempotency_key = %s AND approval_id = %s AND state = 'claimed' RETURNING id """, (bounded_table, record_id, candidate.idempotency_key, self._approval_id), ).fetchone() if row is None: raise InvalidConfiguration("claimed insert action could not be marked applied") def mark_duplicate( self, candidate: InsertCandidate, *, duplicates: Sequence[DuplicateRecord], ) -> None: if not duplicates: raise ContractViolation("duplicate evidence cannot be empty") evidence = Jsonb( [ {"record_id": item.record_id, "match_paths": list(item.match_paths)} for item in duplicates ] ) with self._connect() as connection: row = connection.execute( """ UPDATE osm_lead_source.insert_action_ledger SET state = 'duplicate', duplicate_evidence_json = %s, finalized_at = CURRENT_TIMESTAMP WHERE idempotency_key = %s AND approval_id = %s AND state = 'claimed' RETURNING id """, (evidence, candidate.idempotency_key, self._approval_id), ).fetchone() if row is None: raise InvalidConfiguration("claimed insert action could not be marked duplicate") def mark_failed(self, candidate: InsertCandidate, *, reason: str) -> None: failure_code = _bounded_text(reason, "failure reason", maximum=128) with self._connect() as connection: row = connection.execute( """ UPDATE osm_lead_source.insert_action_ledger SET state = 'failed', failure_code = %s, finalized_at = CURRENT_TIMESTAMP WHERE idempotency_key = %s AND approval_id = %s AND state = 'claimed' RETURNING id """, (failure_code, candidate.idempotency_key, self._approval_id), ).fetchone() if row is None: raise InvalidConfiguration("claimed insert action could not be marked failed") __all__ = ["PostgresInsertLedger"]