"""PostgreSQL implementation of validate-before-write ARR daily ingestion."""

from __future__ import annotations

import json
import os
import re
import time
from dataclasses import dataclass
from datetime import date
from decimal import Decimal, InvalidOperation
from typing import Any, Callable, Dict, Mapping, Optional, Sequence, Tuple

from arr_ingestion.contracts import ArtifactRef, IngestionError, SHA256_RE
from arr_ingestion.repository import (
    IngestionOutcome,
    IngestionRepository,
    JobRegistration,
)
from arr_ingestion.validation import VerifiedDelivery


TARGET_DATABASE = "booking_test"
DATABASE_ENV = "ARR_DATABASE_URL"
ARTIFACT_STORAGE_PROVIDER = "oss"
ARTIFACT_BUCKET_ALIAS = "arr-private"
TRANSIENT_SQLSTATES = frozenset({"23505", "40001", "40P01", "55P03"})
MAX_TRANSACTION_ATTEMPTS = 4


@dataclass(frozen=True)
class DatabaseConfig:
    dsn: str

    @classmethod
    def from_environment(cls) -> "DatabaseConfig":
        dsn = os.environ.get(DATABASE_ENV, "").strip()
        if not dsn:
            raise IngestionError(
                "DATABASE_CONFIG_MISSING",
                f"{DATABASE_ENV} is required",
            )
        return cls(dsn=dsn)


def _default_connect(dsn: str) -> Any:
    try:
        import psycopg  # type: ignore[import-not-found]
    except ImportError:
        raise IngestionError(
            "DATABASE_DRIVER_UNAVAILABLE",
            "PostgreSQL driver is unavailable; install requirements-arr-ingestion.txt",
        ) from None
    try:
        return psycopg.connect(dsn, autocommit=False)
    except Exception:
        raise IngestionError(
            "DATABASE_UNAVAILABLE",
            "database connection failed",
        ) from None


def _decimal_or_none(value: Any) -> Optional[Decimal]:
    if value is None:
        return None
    try:
        number = Decimal(str(value))
    except (InvalidOperation, ValueError):
        raise IngestionError(
            "RESULT_CONTRACT_INVALID",
            "numeric result value is invalid",
        ) from None
    if not number.is_finite():
        raise IngestionError(
            "RESULT_CONTRACT_INVALID",
            "numeric result value is invalid",
        )
    return number


def _date_or_none(value: Any) -> Optional[date]:
    if value is None:
        return None
    if isinstance(value, date):
        return value
    if isinstance(value, str):
        try:
            return date.fromisoformat(value)
        except ValueError:
            pass
    raise IngestionError(
        "RESULT_CONTRACT_INVALID",
        "date result value is invalid",
    )


def _safe_failure_code(payload: Mapping[str, Any]) -> str:
    errors = payload.get("errors")
    if isinstance(errors, list) and errors and isinstance(errors[0], Mapping):
        value = errors[0].get("code")
        if isinstance(value, str) and re.fullmatch(
            r"[A-Z][A-Z0-9_]{0,63}",
            value,
        ):
            return value
    return "PROCESSING_FAILED"


def _is_transient_database_error(error: Exception) -> bool:
    sqlstate = getattr(error, "sqlstate", None)
    if not isinstance(sqlstate, str):
        diagnostic = getattr(error, "diag", None)
        sqlstate = getattr(diagnostic, "sqlstate", None)
    return isinstance(sqlstate, str) and sqlstate in TRANSIENT_SQLSTATES


def _outcome_counts(payload: Mapping[str, Any]) -> Tuple[int, int, int, int, int]:
    values = payload.get("outcome_counts")
    if not isinstance(values, Mapping):
        raise IngestionError(
            "RESULT_CONTRACT_INVALID",
            "structured outcome counts are invalid",
        )
    try:
        counts = tuple(
            int(values[name])
            for name in (
                "retained",
                "excluded_rate_code",
                "duplicate",
                "validation_failed",
                "price_unmatched",
            )
        )
    except (KeyError, TypeError, ValueError):
        raise IngestionError(
            "RESULT_CONTRACT_INVALID",
            "structured outcome counts are invalid",
        ) from None
    if any(value < 0 for value in counts):
        raise IngestionError(
            "RESULT_CONTRACT_INVALID",
            "structured outcome counts are invalid",
        )
    return counts  # type: ignore[return-value]


class PostgresIngestionRepository(IngestionRepository):
    def __init__(
        self,
        config: DatabaseConfig,
        connect: Optional[Callable[[str], Any]] = None,
    ) -> None:
        self._config = config
        self._connect = connect or _default_connect

    def _open(self) -> Any:
        try:
            return self._connect(self._config.dsn)
        except IngestionError:
            raise
        except Exception:
            raise IngestionError(
                "DATABASE_UNAVAILABLE",
                "database connection failed",
            ) from None

    def _run_transaction(
        self,
        operation: Callable[[Any], Any],
        failure_message: str,
    ) -> Any:
        for attempt_index in range(MAX_TRANSACTION_ATTEMPTS):
            connection = self._open()
            retry = False
            try:
                with connection.transaction():
                    with connection.cursor() as cursor:
                        self._begin(cursor)
                        return operation(cursor)
            except IngestionError:
                raise
            except Exception as error:
                retry = (
                    attempt_index + 1 < MAX_TRANSACTION_ATTEMPTS
                    and _is_transient_database_error(error)
                )
                if not retry:
                    raise IngestionError(
                        "DATABASE_WRITE_FAILED",
                        failure_message,
                    ) from None
            finally:
                connection.close()
            if retry:
                time.sleep(0.02 * (2**attempt_index))
        raise IngestionError(
            "DATABASE_WRITE_FAILED",
            failure_message,
        )

    @staticmethod
    def _begin(cursor: Any) -> None:
        cursor.execute("SET TRANSACTION ISOLATION LEVEL SERIALIZABLE")
        cursor.execute("SET LOCAL lock_timeout = '10s'")
        cursor.execute("SET LOCAL statement_timeout = '30s'")
        cursor.execute("SELECT current_database()")
        row = cursor.fetchone()
        if not row or row[0] != TARGET_DATABASE:
            raise IngestionError(
                "DATABASE_TARGET_INVALID",
                "database target is not booking_test",
            )

    @staticmethod
    def _ensure_artifact(cursor: Any, reference: ArtifactRef) -> int:
        cursor.execute(
            """
            SELECT
                id,
                artifact_kind,
                original_filename,
                sha256,
                byte_size,
                mime_type
            FROM ingestion.artifacts
            WHERE storage_provider = %s
              AND bucket_alias = %s
              AND object_key = %s
              AND object_version_id IS NULL
            FOR SHARE
            """,
            (
                ARTIFACT_STORAGE_PROVIDER,
                ARTIFACT_BUCKET_ALIAS,
                reference.object_key,
            ),
        )
        row = cursor.fetchone()
        if row:
            if (
                str(row[1]) != reference.file_kind
                or str(row[2]) != reference.original_filename
                or str(row[3]) != reference.sha256
                or int(row[4]) != reference.byte_size
                or (row[5] or "") != reference.mime_type
            ):
                raise IngestionError(
                    "ARTIFACT_CONFLICT",
                    "stored artifact identity conflicts",
                )
            return int(row[0])
        cursor.execute(
            """
            INSERT INTO ingestion.artifacts (
                artifact_kind,
                storage_provider,
                bucket_alias,
                object_key,
                original_filename,
                sha256,
                byte_size,
                mime_type
            )
            VALUES (%s, %s, %s, %s, %s, %s, %s, %s)
            RETURNING id
            """,
            (
                reference.file_kind,
                ARTIFACT_STORAGE_PROVIDER,
                ARTIFACT_BUCKET_ALIAS,
                reference.object_key,
                reference.original_filename,
                reference.sha256,
                reference.byte_size,
                reference.mime_type,
            ),
        )
        return int(cursor.fetchone()[0])

    def register_job(self, registration: JobRegistration) -> None:
        if (
            registration.source.role != "source_xml"
            or registration.attempt_no < 1
            or not SHA256_RE.fullmatch(registration.idempotency_key)
        ):
            raise IngestionError(
                "JOB_INVALID",
                "processing job registration is invalid",
            )
        self._run_transaction(
            lambda cursor: self._register_job(cursor, registration),
            "processing job could not be stored",
        )

    def mark_running(self, job_id: str, attempt_no: int) -> None:
        self._run_transaction(
            lambda cursor: self._mark_running(cursor, job_id, attempt_no),
            "processing job could not be started",
        )

    def _mark_running(self, cursor: Any, job_id: str, attempt_no: int) -> None:
        run_id, run_status, attempt_id, attempt_status = self._lock_attempt(
            cursor, job_id, attempt_no
        )
        if run_status in {"accepted", "rejected", "failed", "cancelled"}:
            raise IngestionError("JOB_TERMINAL", "processing job is already terminal")
        if attempt_status in {"succeeded", "failed", "cancelled"}:
            raise IngestionError("JOB_TERMINAL", "processing attempt is already terminal")
        cursor.execute(
            """
            UPDATE ingestion.processing_attempts
            SET attempt_status = 'running',
                started_at = COALESCE(started_at, now())
            WHERE id = %s
            """,
            (attempt_id,),
        )
        cursor.execute(
            """
            UPDATE ingestion.processing_runs
            SET run_status = 'running',
                failure_code = NULL,
                failure_message = NULL,
                updated_at = now()
            WHERE id = %s
            """,
            (run_id,),
        )

    def record_failure(self, job_id: str, attempt_no: int, failure_code: str) -> None:
        if not isinstance(failure_code, str) or re.fullmatch(
            r"[A-Z][A-Z0-9_]{0,63}", failure_code
        ) is None:
            failure_code = "PROCESSING_FAILED"
        self._run_transaction(
            lambda cursor: self._record_runtime_failure(
                cursor, job_id, attempt_no, failure_code
            ),
            "processing failure could not be stored",
        )

    def _record_runtime_failure(
        self,
        cursor: Any,
        job_id: str,
        attempt_no: int,
        failure_code: str,
    ) -> None:
        run_id, run_status, attempt_id, _attempt_status = self._lock_attempt(
            cursor, job_id, attempt_no
        )
        if run_status == "failed":
            return
        if run_status in {"accepted", "rejected", "cancelled"}:
            raise IngestionError("JOB_TERMINAL", "processing job is already terminal")
        cursor.execute(
            """
            UPDATE ingestion.processing_attempts
            SET attempt_status = 'failed',
                failure_code = %s,
                failure_message = 'programmatic processing failed',
                finished_at = now()
            WHERE id = %s
            """,
            (failure_code, attempt_id),
        )
        cursor.execute(
            """
            UPDATE ingestion.processing_runs
            SET run_status = 'failed',
                failure_code = %s,
                failure_message = 'programmatic processing failed',
                updated_at = now(),
                finished_at = now()
            WHERE id = %s
            """,
            (failure_code, run_id),
        )
        self._insert_outbox(
            cursor,
            f"processing-run:{run_id}:failed",
            "processing_run",
            run_id,
            "arr.processing_failed",
            {
                "job_id": job_id,
                "business_date": None,
                "daily_version_id": None,
                "failure_code": failure_code,
            },
        )

    @staticmethod
    def _lock_attempt(
        cursor: Any,
        job_id: str,
        attempt_no: int,
    ) -> Tuple[int, str, int, str]:
        cursor.execute(
            """
            SELECT run.id, run.run_status, attempt.id, attempt.attempt_status
            FROM ingestion.processing_runs AS run
            JOIN ingestion.processing_attempts AS attempt
              ON attempt.processing_run_id = run.id
             AND attempt.attempt_no = %s
            WHERE run.run_key = %s
            FOR UPDATE OF run, attempt
            """,
            (attempt_no, job_id),
        )
        row = cursor.fetchone()
        if not row:
            raise IngestionError("JOB_NOT_FOUND", "processing job was not registered")
        return int(row[0]), str(row[1]), int(row[2]), str(row[3])

    def _register_job(self, cursor: Any, registration: JobRegistration) -> None:
        source_artifact_id = self._ensure_artifact(cursor, registration.source)
        cursor.execute(
            """
            SELECT
                id,
                source_artifact_id,
                requested_processor_version,
                requested_rule_set_sha256,
                uploaded_filename,
                run_status
            FROM ingestion.processing_runs
            WHERE run_key = %s
            FOR UPDATE
            """,
            (registration.job_id,),
        )
        existing = cursor.fetchone()
        if existing is None:
            if registration.attempt_no != 1:
                raise IngestionError(
                    "JOB_CONFLICT",
                    "first processing attempt must be attempt 1",
                )
            cursor.execute(
                """
                INSERT INTO ingestion.processing_runs (
                    run_key,
                    pipeline_type,
                    source_artifact_id,
                    run_status,
                    requested_processor_version,
                    requested_rule_set_sha256,
                    uploaded_filename
                )
                VALUES (%s, 'opera_daily', %s, 'queued', %s, %s, %s)
                RETURNING id
                """,
                (
                    registration.job_id,
                    source_artifact_id,
                    registration.processor_version,
                    registration.rule_set_sha256,
                    registration.uploaded_filename,
                ),
            )
            processing_run_id = int(cursor.fetchone()[0])
        else:
            if (
                int(existing[1]) != source_artifact_id
                or existing[2] != registration.processor_version
                or str(existing[3]) != registration.rule_set_sha256
                or existing[4] != registration.uploaded_filename
            ):
                raise IngestionError(
                    "JOB_CONFLICT",
                    "processing job identity conflicts",
                )
            processing_run_id = int(existing[0])
            cursor.execute(
                """
                SELECT idempotency_key
                FROM ingestion.processing_attempts
                WHERE processing_run_id = %s
                  AND attempt_no = %s
                """,
                (processing_run_id, registration.attempt_no),
            )
            attempt = cursor.fetchone()
            if attempt is not None:
                if str(attempt[0]) != registration.idempotency_key:
                    raise IngestionError(
                        "JOB_CONFLICT",
                        "processing attempt identity conflicts",
                    )
                return
            if str(existing[5]) in {
                "accepted",
                "rejected",
                "failed",
                "cancelled",
            }:
                raise IngestionError(
                    "JOB_TERMINAL",
                    "processing job is already terminal",
                )
            cursor.execute(
                """
                SELECT COALESCE(max(attempt_no), 0) + 1
                FROM ingestion.processing_attempts
                WHERE processing_run_id = %s
                """,
                (processing_run_id,),
            )
            if int(cursor.fetchone()[0]) != registration.attempt_no:
                raise IngestionError(
                    "JOB_CONFLICT",
                    "processing attempt sequence conflicts",
                )

        cursor.execute(
            """
            INSERT INTO ingestion.processing_attempts (
                processing_run_id,
                attempt_no,
                attempt_status,
                idempotency_key
            )
            VALUES (%s, %s, 'queued', %s)
            """,
            (
                processing_run_id,
                registration.attempt_no,
                registration.idempotency_key,
            ),
        )

    def commit_delivery(self, delivery: VerifiedDelivery) -> IngestionOutcome:
        return self._run_transaction(
            lambda cursor: self._commit(cursor, delivery),
            "validated delivery was not committed",
        )

    def _commit(
        self,
        cursor: Any,
        delivery: VerifiedDelivery,
    ) -> IngestionOutcome:
        envelope = delivery.envelope
        cursor.execute(
            """
            SELECT
                run.id,
                run.source_artifact_id,
                run.requested_processor_version,
                run.requested_rule_set_sha256,
                run.run_status,
                attempt.id,
                attempt.attempt_status
            FROM ingestion.processing_runs AS run
            JOIN ingestion.processing_attempts AS attempt
              ON attempt.processing_run_id = run.id
             AND attempt.attempt_no = %s
            WHERE run.run_key = %s
            FOR UPDATE OF run, attempt
            """,
            (envelope.attempt_no, envelope.job_id),
        )
        run_row = cursor.fetchone()
        if not run_row:
            raise IngestionError(
                "JOB_NOT_FOUND",
                "processing job was not registered",
            )
        processing_run_id = int(run_row[0])
        attempt_id = int(run_row[5])

        cursor.execute(
            """
            SELECT
                delivery.id,
                delivery.envelope_sha256,
                delivery.delivery_status,
                delivery.daily_version_id,
                delivery.business_date,
                version.version_no,
                version.processing_run_id
            FROM ingestion.processing_deliveries AS delivery
            LEFT JOIN finance.daily_versions AS version
              ON version.id = delivery.daily_version_id
            WHERE delivery.delivery_key = %s
            FOR UPDATE OF delivery
            """,
            (envelope.delivery_id,),
        )
        delivery_row = cursor.fetchone()
        if delivery_row:
            if str(delivery_row[1]) != delivery.envelope_sha256:
                raise IngestionError(
                    "CALLBACK_CONFLICT",
                    "delivery identifier conflicts",
                )
            delivery_status = str(delivery_row[2])
            if delivery_status == "recorded_failure":
                return IngestionOutcome(
                    "recorded_failure",
                    envelope.job_id,
                    delivery_row[4],
                    (
                        int(delivery_row[3])
                        if delivery_row[3] is not None
                        else None
                    ),
                    None,
                )
            if delivery_status == "committed":
                disposition = (
                    "committed"
                    if (
                        delivery_row[6] is not None
                        and int(delivery_row[6]) == processing_run_id
                    )
                    else "already_committed"
                )
                return IngestionOutcome(
                    disposition,
                    envelope.job_id,
                    delivery_row[4],
                    int(delivery_row[3]),
                    int(delivery_row[5]),
                )
            if delivery_status == "rejected":
                raise IngestionError(
                    "CALLBACK_REJECTED",
                    "delivery was previously rejected",
                )
            delivery_id = int(delivery_row[0])
        else:
            delivery_id = None

        if (
            delivery_id is None
            and str(run_row[4])
            in {"accepted", "rejected", "failed", "cancelled"}
        ):
            raise IngestionError(
                "JOB_TERMINAL",
                "processing job is already terminal",
            )
        if (
            run_row[2] != envelope.processor_version
            or str(run_row[3]) != envelope.rule_set_sha256
        ):
            raise IngestionError(
                "JOB_DELIVERY_MISMATCH",
                "delivery does not match its job",
            )

        source_ref = envelope.artifacts["source_xml"]
        result_ref = envelope.artifacts["result_json"]
        structured_ref = envelope.artifacts["structured_result_json"]
        if source_ref is None or result_ref is None or structured_ref is None:
            raise IngestionError(
                "RESULT_CONTRACT_INVALID",
                "required artifact is missing",
            )
        source_artifact_id = self._ensure_artifact(cursor, source_ref)
        if source_artifact_id != int(run_row[1]):
            raise IngestionError(
                "JOB_DELIVERY_MISMATCH",
                "delivery does not match its job",
            )
        daily_ref = envelope.artifacts["daily_report"]
        exception_ref = envelope.artifacts["exception_report"]
        daily_artifact_id = (
            self._ensure_artifact(cursor, daily_ref)
            if daily_ref is not None
            else None
        )
        result_artifact_id = self._ensure_artifact(cursor, result_ref)
        structured_artifact_id = self._ensure_artifact(
            cursor,
            structured_ref,
        )
        exception_artifact_id = (
            self._ensure_artifact(cursor, exception_ref)
            if exception_ref is not None
            else None
        )
        envelope_json = json.dumps(
            envelope.to_dict(),
            ensure_ascii=False,
            separators=(",", ":"),
            sort_keys=True,
        )

        if delivery_id is None:
            cursor.execute(
                """
                INSERT INTO ingestion.processing_deliveries (
                    delivery_key,
                    processing_run_id,
                    attempt_id,
                    envelope_sha256,
                    envelope_json,
                    delivery_status,
                    result_status,
                    processor_version,
                    rule_set_sha256,
                    result_schema_version,
                    business_date,
                    source_artifact_id,
                    daily_report_artifact_id,
                    result_json_artifact_id,
                    structured_result_artifact_id,
                    exception_report_artifact_id,
                    validated_at
                )
                VALUES (
                    %s, %s, %s, %s, %s::jsonb, 'validating', %s,
                    %s, %s, %s, %s, %s, %s, %s, %s, %s, now()
                )
                RETURNING id
                """,
                (
                    envelope.delivery_id,
                    processing_run_id,
                    attempt_id,
                    delivery.envelope_sha256,
                    envelope_json,
                    envelope.status,
                    envelope.processor_version,
                    envelope.rule_set_sha256,
                    envelope.result_schema_version,
                    envelope.business_date,
                    source_artifact_id,
                    daily_artifact_id,
                    result_artifact_id,
                    structured_artifact_id,
                    exception_artifact_id,
                ),
            )
            delivery_id = int(cursor.fetchone()[0])
        else:
            cursor.execute(
                """
                UPDATE ingestion.processing_deliveries
                SET delivery_status = 'validating',
                    result_status = %s,
                    processor_version = %s,
                    rule_set_sha256 = %s,
                    result_schema_version = %s,
                    business_date = %s,
                    source_artifact_id = %s,
                    daily_report_artifact_id = %s,
                    result_json_artifact_id = %s,
                    structured_result_artifact_id = %s,
                    exception_report_artifact_id = %s,
                    validated_at = now()
                WHERE id = %s
                """,
                (
                    envelope.status,
                    envelope.processor_version,
                    envelope.rule_set_sha256,
                    envelope.result_schema_version,
                    envelope.business_date,
                    source_artifact_id,
                    daily_artifact_id,
                    result_artifact_id,
                    structured_artifact_id,
                    exception_artifact_id,
                    delivery_id,
                ),
            )

        cursor.execute(
            """
            UPDATE ingestion.processing_runs
            SET run_status = 'validating',
                result_artifact_id = %s,
                delivered_processor_version = %s,
                delivered_rule_set_sha256 = %s,
                result_schema_version = %s,
                delivery_sha256 = %s,
                delivery_json = %s::jsonb,
                business_date = %s,
                failure_code = NULL,
                failure_message = NULL,
                updated_at = now(),
                validated_at = now()
            WHERE id = %s
            """,
            (
                structured_artifact_id,
                envelope.processor_version,
                envelope.rule_set_sha256,
                envelope.result_schema_version,
                delivery.envelope_sha256,
                envelope_json,
                envelope.business_date,
                processing_run_id,
            ),
        )
        cursor.execute(
            """
            UPDATE ingestion.processing_attempts
            SET attempt_status = 'delivered'
            WHERE id = %s
            """,
            (attempt_id,),
        )

        if envelope.status == "failed":
            return self._record_failure(
                cursor,
                delivery,
                processing_run_id,
                attempt_id,
                delivery_id,
                source_artifact_id,
                result_artifact_id,
                structured_artifact_id,
                exception_artifact_id,
            )
        if envelope.business_date is None or daily_artifact_id is None:
            raise IngestionError(
                "RESULT_CONTRACT_INVALID",
                "successful delivery is incomplete",
            )
        return self._commit_success(
            cursor,
            delivery,
            processing_run_id,
            attempt_id,
            delivery_id,
            source_artifact_id,
            daily_artifact_id,
            result_artifact_id,
            structured_artifact_id,
        )

    def _record_failure(
        self,
        cursor: Any,
        delivery: VerifiedDelivery,
        processing_run_id: int,
        attempt_id: int,
        delivery_id: int,
        source_artifact_id: int,
        result_artifact_id: int,
        structured_artifact_id: int,
        exception_artifact_id: Optional[int],
    ) -> IngestionOutcome:
        envelope = delivery.envelope
        payload = delivery.structured_payload
        code = _safe_failure_code(payload)
        counts = _outcome_counts(payload)
        records = payload.get("records")
        if not isinstance(records, list):
            raise IngestionError(
                "RESULT_CONTRACT_INVALID",
                "structured records are invalid",
            )
        cursor.execute(
            """
            INSERT INTO finance.daily_versions (
                business_date,
                version_no,
                processing_run_id,
                source_artifact_id,
                result_json_artifact_id,
                structured_result_artifact_id,
                exception_report_artifact_id,
                version_status,
                processor_version,
                rule_set_sha256,
                result_schema_version,
                result_sha256,
                source_rows,
                retained_rows,
                excluded_rate_code_rows,
                duplicate_rows,
                validation_failed_rows,
                price_unmatched_rows,
                failure_code,
                failure_message,
                validated_at
            )
            VALUES (
                NULL, NULL, %s, %s, %s, %s, %s, 'rejected',
                %s, %s, %s, %s, %s, %s, %s, %s, %s, %s,
                %s, 'deterministic processing failed', now()
            )
            RETURNING id
            """,
            (
                processing_run_id,
                source_artifact_id,
                result_artifact_id,
                structured_artifact_id,
                exception_artifact_id,
                envelope.processor_version,
                envelope.rule_set_sha256,
                envelope.result_schema_version,
                envelope.artifacts["structured_result_json"].sha256,
                int(payload["source_rows"]),
                *counts,
                code,
            ),
        )
        daily_version_id = int(cursor.fetchone()[0])
        booking_links = self._load_booking_links(cursor, records)
        self._insert_records(
            cursor,
            daily_version_id,
            records,
            booking_links,
        )
        cursor.execute(
            """
            UPDATE ingestion.processing_deliveries
            SET delivery_status = 'recorded_failure',
                daily_version_id = %s,
                failure_code = %s,
                failure_message = 'deterministic processing failed',
                committed_at = now()
            WHERE id = %s
            """,
            (daily_version_id, code, delivery_id),
        )
        cursor.execute(
            """
            UPDATE ingestion.processing_attempts
            SET attempt_status = 'failed',
                failure_code = %s,
                failure_message = 'deterministic processing failed',
                finished_at = now()
            WHERE id = %s
            """,
            (code, attempt_id),
        )
        cursor.execute(
            """
            UPDATE ingestion.processing_runs
            SET run_status = 'failed',
                failure_code = %s,
                failure_message = 'deterministic processing failed',
                updated_at = now(),
                finished_at = now()
            WHERE id = %s
            """,
            (code, processing_run_id),
        )
        self._insert_outbox(
            cursor,
            f"processing-run:{processing_run_id}:failed",
            "processing_run",
            processing_run_id,
            "arr.processing_failed",
            {
                "job_id": envelope.job_id,
                "business_date": (
                    envelope.business_date.isoformat()
                    if envelope.business_date
                    else None
                ),
                "daily_version_id": daily_version_id,
                "failure_code": code,
            },
        )
        return IngestionOutcome(
            "recorded_failure",
            envelope.job_id,
            envelope.business_date,
            daily_version_id,
            None,
        )

    def _commit_success(
        self,
        cursor: Any,
        delivery: VerifiedDelivery,
        processing_run_id: int,
        attempt_id: int,
        delivery_id: int,
        source_artifact_id: int,
        daily_artifact_id: int,
        result_artifact_id: int,
        structured_artifact_id: int,
    ) -> IngestionOutcome:
        envelope = delivery.envelope
        business_date = envelope.business_date
        assert business_date is not None
        cursor.execute(
            "SELECT pg_advisory_xact_lock(hashtextextended(%s, 0))",
            (f"finance-daily:{business_date.isoformat()}",),
        )
        cursor.execute(
            """
            SELECT id, version_no, version_status
            FROM finance.daily_versions
            WHERE source_artifact_id = %s
              AND business_date = %s
              AND processor_version = %s
              AND rule_set_sha256 = %s
            FOR UPDATE
            """,
            (
                source_artifact_id,
                business_date,
                envelope.processor_version,
                envelope.rule_set_sha256,
            ),
        )
        existing = cursor.fetchone()
        if existing:
            if str(existing[2]) not in {"active", "superseded"}:
                raise IngestionError(
                    "DATABASE_STATE_INVALID",
                    "existing daily version is invalid",
                )
            daily_version_id = int(existing[0])
            version_no = int(existing[1])
            disposition = "already_committed"
        else:
            payload = delivery.structured_payload
            counts = _outcome_counts(payload)
            cursor.execute(
                """
                SELECT COALESCE(MAX(version_no), 0) + 1
                FROM finance.daily_versions
                WHERE business_date = %s
                """,
                (business_date,),
            )
            version_no = int(cursor.fetchone()[0])
            cursor.execute(
                """
                INSERT INTO finance.daily_versions (
                    business_date,
                    version_no,
                    processing_run_id,
                    source_artifact_id,
                    daily_report_artifact_id,
                    result_json_artifact_id,
                    structured_result_artifact_id,
                    version_status,
                    processor_version,
                    rule_set_sha256,
                    result_schema_version,
                    result_sha256,
                    source_rows,
                    retained_rows,
                    excluded_rate_code_rows,
                    duplicate_rows,
                    validation_failed_rows,
                    price_unmatched_rows,
                    validated_at
                )
                VALUES (
                    %s, %s, %s, %s, %s, %s, %s, 'validated',
                    %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, now()
                )
                RETURNING id
                """,
                (
                    business_date,
                    version_no,
                    processing_run_id,
                    source_artifact_id,
                    daily_artifact_id,
                    result_artifact_id,
                    structured_artifact_id,
                    envelope.processor_version,
                    envelope.rule_set_sha256,
                    envelope.result_schema_version,
                    envelope.artifacts["structured_result_json"].sha256,
                    int(payload["source_rows"]),
                    *counts,
                ),
            )
            daily_version_id = int(cursor.fetchone()[0])
            records = payload.get("records")
            if not isinstance(records, list):
                raise IngestionError(
                    "RESULT_CONTRACT_INVALID",
                    "structured records are invalid",
                )
            booking_links = self._load_booking_links(cursor, records)
            self._insert_records(
                cursor,
                daily_version_id,
                records,
                booking_links,
            )
            channels = payload.get("channels")
            if not isinstance(channels, list):
                raise IngestionError(
                    "RESULT_CONTRACT_INVALID",
                    "structured channels are invalid",
                )
            self._insert_channel_metrics(
                cursor,
                daily_version_id,
                channels,
            )
            self._activate_daily_version(
                cursor,
                business_date,
                daily_version_id,
            )
            disposition = "committed"

        cursor.execute(
            """
            UPDATE ingestion.processing_deliveries
            SET delivery_status = 'committed',
                daily_version_id = %s,
                committed_at = now()
            WHERE id = %s
            """,
            (daily_version_id, delivery_id),
        )
        cursor.execute(
            """
            UPDATE ingestion.processing_attempts
            SET attempt_status = 'succeeded',
                failure_code = NULL,
                failure_message = NULL,
                finished_at = now()
            WHERE id = %s
            """,
            (attempt_id,),
        )
        cursor.execute(
            """
            UPDATE ingestion.processing_runs
            SET run_status = 'accepted',
                business_date = %s,
                failure_code = NULL,
                failure_message = NULL,
                updated_at = now(),
                finished_at = now()
            WHERE id = %s
            """,
            (business_date, processing_run_id),
        )
        self._insert_outbox(
            cursor,
            f"processing-run:{processing_run_id}:accepted",
            "processing_run",
            processing_run_id,
            "arr.daily_version_committed",
            {
                "job_id": envelope.job_id,
                "business_date": business_date.isoformat(),
                "daily_version_id": daily_version_id,
                "version_no": version_no,
                "disposition": disposition,
            },
        )
        return IngestionOutcome(
            disposition,
            envelope.job_id,
            business_date,
            daily_version_id,
            version_no,
        )

    @staticmethod
    def _insert_channel_metrics(
        cursor: Any,
        daily_version_id: int,
        channels: Sequence[Mapping[str, Any]],
    ) -> None:
        for channel_order, channel in enumerate(channels, 1):
            cursor.execute(
                """
                INSERT INTO finance.daily_channel_metrics (
                    daily_version_id,
                    channel_key,
                    channel_order,
                    row_count
                )
                VALUES (%s, %s, %s, %s)
                """,
                (
                    daily_version_id,
                    str(channel["worksheet"]),
                    channel_order,
                    int(channel["rows"]),
                ),
            )

    @staticmethod
    def _activate_daily_version(
        cursor: Any,
        business_date: date,
        daily_version_id: int,
    ) -> None:
        cursor.execute(
            """
            SELECT daily_version_id
            FROM finance.current_daily_versions
            WHERE business_date = %s
            FOR UPDATE
            """,
            (business_date,),
        )
        current = cursor.fetchone()
        if current and int(current[0]) != daily_version_id:
            cursor.execute(
                """
                UPDATE finance.daily_versions
                SET version_status = 'superseded',
                    superseded_at = now()
                WHERE id = %s
                  AND version_status = 'active'
                """,
                (int(current[0]),),
            )
        cursor.execute(
            """
            UPDATE finance.daily_versions
            SET version_status = 'active',
                activated_at = now()
            WHERE id = %s
              AND version_status = 'validated'
            """,
            (daily_version_id,),
        )
        cursor.execute(
            """
            INSERT INTO finance.current_daily_versions (
                business_date,
                daily_version_id,
                activated_at
            )
            VALUES (%s, %s, now())
            ON CONFLICT (business_date) DO UPDATE
            SET daily_version_id = EXCLUDED.daily_version_id,
                activated_at = EXCLUDED.activated_at
            """,
            (business_date, daily_version_id),
        )

    @staticmethod
    def _load_booking_links(
        cursor: Any,
        records: Sequence[Mapping[str, Any]],
    ) -> Mapping[str, Tuple[str, int]]:
        group_codes = sorted(
            {
                str(record["group_code_key"])
                for record in records
                if record.get("group_code_key")
            }
        )
        if not group_codes:
            return {}
        cursor.execute(
            """
            SELECT
                source.group_code_key,
                count(DISTINCT source.id)
            FROM booking.source_rows AS source
            JOIN booking.current_source_batch AS active_source
              ON active_source.source_batch_id = source.source_batch_id
             AND active_source.singleton
            JOIN booking.source_batches AS batch
              ON batch.id = source.source_batch_id
             AND batch.batch_status = 'accepted'
            JOIN booking.current_row_parses AS current_parse
              ON current_parse.source_row_id = source.id
            JOIN booking.parse_versions AS parsed
              ON parsed.id = current_parse.parse_version_id
             AND parsed.parse_status = 'accepted'
            WHERE source.group_code_key = ANY(%s)
            GROUP BY source.group_code_key
            """,
            (group_codes,),
        )
        matches: Dict[str, Tuple[str, int]] = {}
        for group_code, count in cursor.fetchall():
            match_count = int(count)
            matches[str(group_code)] = ("matched", match_count)
        for group_code in group_codes:
            matches.setdefault(group_code, ("unmatched", 0))
        return matches

    @staticmethod
    def _insert_records(
        cursor: Any,
        daily_version_id: int,
        records: Sequence[Mapping[str, Any]],
        booking_links: Mapping[str, Tuple[str, int]],
    ) -> None:
        inserted_ids: Dict[int, int] = {}
        for record in records:
            source_sequence = int(record["source_sequence"])
            duplicate_sequence = record.get(
                "duplicate_of_source_sequence",
            )
            duplicate_id = (
                inserted_ids.get(int(duplicate_sequence))
                if duplicate_sequence is not None
                else None
            )
            if duplicate_sequence is not None and duplicate_id is None:
                raise IngestionError(
                    "RESULT_CONTRACT_INVALID",
                    "duplicate lineage is invalid",
                )
            group_code = record.get("group_code_key")
            if group_code is None:
                source_status, source_match_count = (
                    "missing_group_code",
                    0,
                )
            else:
                source_status, source_match_count = booking_links.get(
                    str(group_code),
                    ("unmatched", 0),
                )
            cursor.execute(
                """
                INSERT INTO finance.daily_records (
                    daily_version_id,
                    source_sequence,
                    source_location,
                    source_worksheet,
                    source_row_no,
                    outcome,
                    decision_codes,
                    duplicate_of_record_id,
                    block_code,
                    adults,
                    children,
                    company_name,
                    company_key,
                    confirmation_no,
                    disp_room_no,
                    effective_rate_amount,
                    full_name,
                    res_comment,
                    trace_text,
                    no_of_rooms,
                    products,
                    rate_code,
                    room_category_label,
                    arrival,
                    departure,
                    nights,
                    real_price,
                    total_price,
                    kb_amount,
                    channel_key,
                    pricing_method,
                    booking_source_match_status,
                    booking_source_match_count
                )
                VALUES (
                    %s, %s, %s, %s, %s, %s, %s, %s, %s, %s,
                    %s, %s, %s, %s, %s, %s, %s, %s, %s, %s,
                    %s, %s, %s, %s, %s, %s, %s, %s, %s, %s,
                    %s, %s, %s
                )
                RETURNING id
                """,
                (
                    daily_version_id,
                    source_sequence,
                    record.get("source_location"),
                    record.get("source_worksheet"),
                    record.get("source_row_no"),
                    record.get("outcome"),
                    list(record.get("decision_codes") or []),
                    duplicate_id,
                    record.get("block_code"),
                    record.get("adults"),
                    record.get("children"),
                    record.get("company_name"),
                    record.get("company_key"),
                    record.get("confirmation_no"),
                    record.get("disp_room_no"),
                    _decimal_or_none(record.get("effective_rate_amount")),
                    record.get("full_name"),
                    record.get("res_comment"),
                    record.get("trace_text"),
                    record.get("no_of_rooms"),
                    record.get("products"),
                    record.get("rate_code"),
                    record.get("room_category_label"),
                    _date_or_none(record.get("arrival")),
                    _date_or_none(record.get("departure")),
                    record.get("nights"),
                    _decimal_or_none(record.get("real_price")),
                    _decimal_or_none(record.get("total_price")),
                    _decimal_or_none(record.get("kb_amount")),
                    record.get("channel_key"),
                    record.get("pricing_method"),
                    source_status,
                    source_match_count,
                ),
            )
            inserted_ids[source_sequence] = int(cursor.fetchone()[0])

    @staticmethod
    def _insert_outbox(
        cursor: Any,
        event_key: str,
        aggregate_type: str,
        aggregate_id: int,
        event_type: str,
        payload: Mapping[str, Any],
    ) -> None:
        cursor.execute(
            """
            INSERT INTO ingestion.outbox_events (
                event_key,
                aggregate_type,
                aggregate_id,
                event_type,
                payload
            )
            VALUES (%s, %s, %s, %s, %s::jsonb)
            ON CONFLICT (event_key) DO NOTHING
            """,
            (
                event_key,
                aggregate_type,
                aggregate_id,
                event_type,
                json.dumps(
                    payload,
                    ensure_ascii=False,
                    separators=(",", ":"),
                    sort_keys=True,
                ),
            ),
        )