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

from __future__ import annotations

import hashlib
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,
    ReviewGeneration,
)
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",
    )


MANUAL_PRICE_RE = re.compile(r"^(?:0|[1-9][0-9]{0,15})$")


def _manual_price(value: Any) -> Decimal:
    if not isinstance(value, str) or MANUAL_PRICE_RE.fullmatch(value) is None:
        raise IngestionError(
            "REVIEW_PRICE_INVALID",
            "人工价格必须是非负整数字符串",
        )
    try:
        amount = Decimal(value)
    except InvalidOperation:
        raise IngestionError("REVIEW_PRICE_INVALID", "人工价格格式无效") from None
    if amount < 0:
        raise IngestionError("REVIEW_PRICE_INVALID", "人工价格不得为负数")
    return amount.quantize(Decimal("0.01"))


def _review_price_text(value: Any) -> str:
    amount = _decimal_or_none(value)
    if amount is None or amount < 0 or amount != amount.to_integral_value():
        raise IngestionError("DATABASE_STATE_INVALID", "人工价格不是整数")
    return format(amount.quantize(Decimal("1")), "f")


def _canonical_amount(value: Any) -> str:
    amount = _decimal_or_none(value)
    if amount is None or amount < 0:
        raise IngestionError("RESULT_CONTRACT_INVALID", "复核Opera价格无效")
    return format(amount.normalize(), "f")


def _canonical_json_bytes(value: Mapping[str, Any]) -> bytes:
    return json.dumps(
        value,
        ensure_ascii=False,
        allow_nan=False,
        sort_keys=True,
        separators=(",", ":"),
    ).encode("utf-8")


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(
            """
            SELECT id, revision
            FROM ingestion.daily_review_cases
            WHERE processing_run_id = %s
              AND case_status = 'processing'
            FOR UPDATE
            """,
            (run_id,),
        )
        active_review = cursor.fetchone()
        if active_review:
            review_case_id = int(active_review[0])
            cursor.execute(
                """
                UPDATE ingestion.daily_review_cases
                SET case_status = 'failed',
                    failure_code = %s,
                    failure_message = 'deterministic final generation failed',
                    updated_at = now()
                WHERE id = %s
                """,
                (failure_code, review_case_id),
            )
            self._review_event(
                cursor,
                review_case_id,
                "PRICE_REVIEW_FAILED",
                "system:runtime",
                int(active_review[1]),
            )
        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 _review_actor(value: str) -> str:
        if not isinstance(value, str) or not value.strip() or len(value) > 255:
            raise IngestionError("REVIEW_ACTOR_INVALID", "复核会话用户无效")
        return value.strip()

    @staticmethod
    def _review_number(value: Any) -> int | float:
        amount = _decimal_or_none(value)
        if amount is None:
            raise IngestionError("DATABASE_STATE_INVALID", "复核金额为空")
        return int(amount) if amount == amount.to_integral_value() else float(amount)

    @staticmethod
    def _review_event(
        cursor: Any,
        review_case_id: int,
        event_type: str,
        actor_username: str,
        revision: int,
        *,
        review_item_id: Optional[int] = None,
        previous_real_price: Optional[Decimal] = None,
        new_real_price: Optional[Decimal] = None,
    ) -> None:
        cursor.execute(
            """
            INSERT INTO ingestion.daily_review_events (
                review_case_id, review_item_id, event_type, actor_username,
                previous_real_price, new_real_price, revision
            )
            VALUES (%s, %s, %s, %s, %s, %s, %s)
            """,
            (
                review_case_id,
                review_item_id,
                event_type,
                actor_username,
                previous_real_price,
                new_real_price,
                revision,
            ),
        )

    @staticmethod
    def _lock_review_case(cursor: Any, job_id: str, case_key: str) -> Tuple[Any, ...]:
        cursor.execute(
            """
            SELECT
                review.id,
                review.case_key,
                review.case_status,
                review.revision,
                review.business_date,
                review.processor_version,
                review.rule_set_sha256,
                review.source_sha256,
                review.manual_override_json,
                review.manual_override_sha256,
                review.manual_override_artifact_id,
                run.id,
                run.run_status,
                run.requested_processor_version,
                run.requested_rule_set_sha256
            FROM ingestion.daily_review_cases AS review
            JOIN ingestion.processing_runs AS run
              ON run.id = review.processing_run_id
            WHERE run.run_key = %s
              AND review.case_key = %s
            FOR UPDATE OF review, run
            """,
            (job_id, case_key),
        )
        row = cursor.fetchone()
        if not row:
            raise IngestionError("REVIEW_NOT_FOUND", "待人工处理任务不存在")
        return tuple(row)

    def get_price_review(self, job_id: str, limit: int, offset: int) -> Dict[str, Any]:
        if (
            not isinstance(limit, int)
            or isinstance(limit, bool)
            or not 1 <= limit <= 200
            or not isinstance(offset, int)
            or isinstance(offset, bool)
            or offset < 0
        ):
            raise IngestionError("REVIEW_PAGE_INVALID", "复核分页参数无效")
        return self._run_transaction(
            lambda cursor: self._get_price_review(cursor, job_id, limit, offset),
            "review case could not be read",
        )

    def _get_price_review(
        self, cursor: Any, job_id: str, limit: int, offset: int
    ) -> Dict[str, Any]:
        cursor.execute(
            """
            SELECT
                review.id,
                review.case_key,
                review.case_status,
                review.revision,
                review.business_date,
                count(item.id) AS total_items,
                count(item.id) FILTER (WHERE item.real_price IS NOT NULL) AS completed_items
            FROM ingestion.daily_review_cases AS review
            JOIN ingestion.processing_runs AS run
              ON run.id = review.processing_run_id
            LEFT JOIN ingestion.daily_review_items AS item
              ON item.review_case_id = review.id
            WHERE run.run_key = %s
            GROUP BY review.id, review.case_key, review.case_status, review.revision, review.business_date
            """,
            (job_id,),
        )
        case = cursor.fetchone()
        if not case:
            raise IngestionError("REVIEW_NOT_FOUND", "待人工处理任务不存在")
        review_case_id = int(case[0])
        total = int(case[5])
        completed = int(case[6])
        cursor.execute(
            """
            SELECT
                id,
                company_key,
                rate_code,
                effective_rate_amount,
                candidate_prices,
                affected_records,
                affected_rooms,
                affected_room_nights,
                real_price,
                revision
            FROM ingestion.daily_review_items
            WHERE review_case_id = %s
            ORDER BY company_key, rate_code, effective_rate_amount, id
            LIMIT %s OFFSET %s
            """,
            (review_case_id, limit, offset),
        )
        items = []
        for row in cursor.fetchall():
            real_price = row[8]
            items.append(
                {
                    "item_id": int(row[0]),
                    "company_key": str(row[1]),
                    "rate_code": str(row[2]),
                    "effective_rate_amount": self._review_number(row[3]),
                    "candidate_prices": row[4] if isinstance(row[4], list) else [],
                    "affected_records": int(row[5]),
                    "affected_rooms": int(row[6]),
                    "affected_room_nights": int(row[7]),
                    "real_price": (_review_price_text(real_price) if real_price is not None else None),
                    "revision": int(row[9]),
                }
            )
        return {
            "case_id": str(case[1]),
            "case_status": str(case[2]),
            "revision": int(case[3]),
            "business_date": _date_or_none(case[4]).isoformat() if case[4] is not None else None,
            "completed_items": completed,
            "total_items": total,
            "items": items,
            "pagination": {
                "limit": limit,
                "offset": offset,
                "total": total,
                "has_next": offset + len(items) < total,
            },
        }

    def update_price_review_item(
        self,
        job_id: str,
        item_id: int,
        case_id: str,
        revision: int,
        real_price: str,
        actor_username: str,
    ) -> Dict[str, Any]:
        if not isinstance(item_id, int) or isinstance(item_id, bool) or item_id < 1:
            raise IngestionError("REVIEW_ITEM_NOT_FOUND", "待人工处理项不存在")
        if not isinstance(revision, int) or isinstance(revision, bool) or revision < 0:
            raise IngestionError("REVIEW_REVISION_CONFLICT", "复核版本无效")
        price = _manual_price(real_price)
        actor = self._review_actor(actor_username)
        return self._run_transaction(
            lambda cursor: self._update_price_review_item(
                cursor, job_id, item_id, case_id, revision, price, actor
            ),
            "review item could not be updated",
        )

    def _update_price_review_item(
        self,
        cursor: Any,
        job_id: str,
        item_id: int,
        case_id: str,
        revision: int,
        price: Decimal,
        actor_username: str,
    ) -> Dict[str, Any]:
        case = self._lock_review_case(cursor, job_id, case_id)
        review_case_id, _key, status, current_revision = int(case[0]), str(case[1]), str(case[2]), int(case[3])
        if status != "open":
            raise IngestionError("REVIEW_IMMUTABLE", "复核清单已冻结，不能修改")
        if revision != current_revision:
            raise IngestionError("REVIEW_REVISION_CONFLICT", "复核清单已被其他会话更新")
        cursor.execute(
            """
            SELECT real_price
            FROM ingestion.daily_review_items
            WHERE id = %s AND review_case_id = %s
            FOR UPDATE
            """,
            (item_id, review_case_id),
        )
        item = cursor.fetchone()
        if not item:
            raise IngestionError("REVIEW_ITEM_NOT_FOUND", "待人工处理项不存在")
        previous = _decimal_or_none(item[0])
        next_revision = current_revision + 1
        cursor.execute(
            """
            UPDATE ingestion.daily_review_items
            SET real_price = %s,
                revision = %s,
                updated_at = now()
            WHERE id = %s
            """,
            (price, next_revision, item_id),
        )
        cursor.execute(
            """
            UPDATE ingestion.daily_review_cases
            SET revision = %s,
                updated_at = now()
            WHERE id = %s
            """,
            (next_revision, review_case_id),
        )
        self._review_event(
            cursor,
            review_case_id,
            "PRICE_REVIEW_UPDATED",
            actor_username,
            next_revision,
            review_item_id=item_id,
            previous_real_price=previous,
            new_real_price=price,
        )
        return self._get_price_review(cursor, job_id, 200, 0)

    def begin_price_review_generation(
        self,
        job_id: str,
        case_id: str,
        revision: int,
        actor_username: str,
        processor_version: str,
        rule_set_sha256: str,
        idempotency_key: str,
    ) -> ReviewGeneration:
        if not isinstance(revision, int) or isinstance(revision, bool) or revision < 0:
            raise IngestionError("REVIEW_REVISION_CONFLICT", "复核版本无效")
        if not SHA256_RE.fullmatch(idempotency_key):
            raise IngestionError("JOB_INVALID", "复核生成幂等键无效")
        actor = self._review_actor(actor_username)
        return self._run_transaction(
            lambda cursor: self._begin_price_review_generation(
                cursor,
                job_id,
                case_id,
                revision,
                actor,
                processor_version,
                rule_set_sha256,
                idempotency_key,
            ),
            "review final generation could not be started",
        )

    def _begin_price_review_generation(
        self,
        cursor: Any,
        job_id: str,
        case_id: str,
        revision: int,
        actor_username: str,
        processor_version: str,
        rule_set_sha256: str,
        idempotency_key: str,
    ) -> ReviewGeneration:
        case = self._lock_review_case(cursor, job_id, case_id)
        (
            review_case_id,
            case_key,
            status,
            current_revision,
            business_date,
            case_processor_version,
            case_rule_set_sha256,
            source_sha256,
            stored_manifest,
            stored_manifest_sha256,
            _stored_manifest_artifact_id,
            processing_run_id,
            _run_status,
            requested_processor_version,
            requested_rule_set_sha256,
        ) = case
        review_case_id = int(review_case_id)
        business_date = _date_or_none(business_date)
        assert business_date is not None
        if (
            str(case_processor_version) != processor_version
            or str(case_rule_set_sha256) != rule_set_sha256
            or requested_processor_version != processor_version
            or str(requested_rule_set_sha256) != rule_set_sha256
        ):
            raise IngestionError(
                "REVIEW_RULESET_CHANGED",
                "处理器或规则已变更，请取消后重新上传",
            )
        if status == "completed":
            cursor.execute(
                "SELECT id, version_no FROM finance.daily_versions WHERE review_case_id = %s",
                (review_case_id,),
            )
            version = cursor.fetchone()
            return ReviewGeneration(
                "completed",
                job_id,
                str(case_key),
                None,
                business_date,
                None,
                str(stored_manifest_sha256) if stored_manifest_sha256 else None,
                int(version[0]) if version else None,
                int(version[1]) if version else None,
            )
        if status == "processing":
            cursor.execute(
                """
                SELECT attempt_no
                FROM ingestion.processing_attempts
                WHERE processing_run_id = %s
                ORDER BY attempt_no DESC
                LIMIT 1
                """,
                (int(processing_run_id),),
            )
            attempt = cursor.fetchone()
            return ReviewGeneration(
                "processing",
                job_id,
                str(case_key),
                int(attempt[0]) if attempt else None,
                business_date,
                None,
                str(stored_manifest_sha256) if stored_manifest_sha256 else None,
            )
        if status not in {"open", "generation_failed"}:
            raise IngestionError("REVIEW_NOT_OPEN", "复核任务当前不能确认生成")
        if int(current_revision) != revision:
            raise IngestionError("REVIEW_REVISION_CONFLICT", "复核清单已被其他会话更新")
        cursor.execute(
            """
            SELECT company_key, rate_code, effective_rate_amount, real_price
            FROM ingestion.daily_review_items
            WHERE review_case_id = %s
            ORDER BY company_key, rate_code, effective_rate_amount, id
            FOR UPDATE
            """,
            (review_case_id,),
        )
        items = cursor.fetchall()
        if not items or any(row[3] is None for row in items):
            raise IngestionError("REVIEW_INCOMPLETE", "请先填写全部缺价项")
        if stored_manifest is None:
            issue_rows = [
                {
                    "company_key": str(row[0]),
                    "rate_code": str(row[1]),
                    "effective_rate_amount": _canonical_amount(row[2]),
                }
                for row in items
            ]
            overrides = [
                {
                    **issue,
                    "real_price": format(Decimal(str(row[3])), ".2f"),
                }
                for issue, row in zip(issue_rows, items)
            ]
            manifest_payload: Dict[str, Any] = {
                "review_version": "1.0",
                "review_case_id": str(case_key),
                "job_id": job_id,
                "source_sha256": str(source_sha256),
                "business_date": business_date.isoformat(),
                "processor_version": processor_version,
                "rule_set_sha256": rule_set_sha256,
                "issues": issue_rows,
                "overrides": overrides,
            }
            manifest_bytes = _canonical_json_bytes(manifest_payload)
            manifest_sha256 = hashlib.sha256(manifest_bytes).hexdigest()
            cursor.execute(
                """
                UPDATE ingestion.daily_review_cases
                SET case_status = 'processing',
                    manual_override_json = %s::jsonb,
                    manual_override_sha256 = %s,
                    frozen_at = now(),
                    failure_code = NULL,
                    failure_message = NULL,
                    updated_at = now()
                WHERE id = %s
                """,
                (manifest_bytes.decode("utf-8"), manifest_sha256, review_case_id),
            )
        else:
            manifest_payload = stored_manifest if isinstance(stored_manifest, Mapping) else None
            if manifest_payload is None:
                raise IngestionError("DATABASE_STATE_INVALID", "冻结人工价格清单无效")
            manifest_bytes = _canonical_json_bytes(dict(manifest_payload))
            manifest_sha256 = hashlib.sha256(manifest_bytes).hexdigest()
            if manifest_sha256 != str(stored_manifest_sha256):
                raise IngestionError("DATABASE_STATE_INVALID", "冻结人工价格清单哈希不一致")
            cursor.execute(
                """
                UPDATE ingestion.daily_review_cases
                SET case_status = 'processing',
                    failure_code = NULL,
                    failure_message = NULL,
                    updated_at = now()
                WHERE id = %s
                """,
                (review_case_id,),
            )
        cursor.execute(
            """
            SELECT COALESCE(max(attempt_no), 0) + 1
            FROM ingestion.processing_attempts
            WHERE processing_run_id = %s
            """,
            (int(processing_run_id),),
        )
        attempt_no = int(cursor.fetchone()[0])
        cursor.execute(
            """
            INSERT INTO ingestion.processing_attempts (
                processing_run_id, attempt_no, attempt_status, idempotency_key
            )
            VALUES (%s, %s, 'queued', %s)
            """,
            (int(processing_run_id), attempt_no, idempotency_key),
        )
        self._review_event(
            cursor,
            review_case_id,
            "PRICE_REVIEW_FINALIZED",
            actor_username,
            int(current_revision),
        )
        cursor.execute(
            """
            SELECT
                artifact.artifact_kind,
                artifact.object_key,
                artifact.original_filename,
                artifact.sha256,
                artifact.byte_size,
                artifact.mime_type
            FROM ingestion.processing_runs AS run
            JOIN ingestion.artifacts AS artifact
              ON artifact.id = run.source_artifact_id
            WHERE run.id = %s
            FOR SHARE OF artifact
            """,
            (int(processing_run_id),),
        )
        source_row = cursor.fetchone()
        if (
            source_row is None
            or str(source_row[0]) != "opera_xml"
            or not isinstance(source_row[1], str)
            or not isinstance(source_row[2], str)
            or not isinstance(source_row[3], str)
            or not isinstance(source_row[4], int)
            or not isinstance(source_row[5], str)
        ):
            raise IngestionError("DATABASE_STATE_INVALID", "复核源XML制品无效")
        source_ref = ArtifactRef(
            role="source_xml",
            file_kind="opera_xml",
            object_key=source_row[1],
            original_filename=source_row[2],
            sha256=source_row[3],
            byte_size=source_row[4],
            mime_type=source_row[5],
        )
        if source_ref.sha256 != str(source_sha256):
            raise IngestionError("DATABASE_STATE_INVALID", "复核源XML身份不一致")
        return ReviewGeneration(
            "ready",
            job_id,
            str(case_key),
            attempt_no,
            business_date,
            manifest_bytes,
            manifest_sha256,
            source=source_ref,
        )

    def record_price_review_generation_failure(
        self,
        job_id: str,
        case_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 = "REVIEW_GENERATION_FAILED"
        self._run_transaction(
            lambda cursor: self._record_price_review_generation_failure(
                cursor, job_id, case_id, attempt_no, failure_code
            ),
            "review generation failure could not be recorded",
        )

    def _record_price_review_generation_failure(
        self, cursor: Any, job_id: str, case_id: str, attempt_no: int, failure_code: str
    ) -> None:
        case = self._lock_review_case(cursor, job_id, case_id)
        review_case_id, _case_key, status, revision, _date_value, _processor, _rules, _source, _json, _sha, _artifact, run_id, _run_status, _requested_processor, _requested_rules = case
        if str(status) == "completed":
            return
        cursor.execute(
            """
            SELECT id
            FROM ingestion.processing_attempts
            WHERE processing_run_id = %s AND attempt_no = %s
            FOR UPDATE
            """,
            (int(run_id), attempt_no),
        )
        attempt = cursor.fetchone()
        if not attempt:
            raise IngestionError("JOB_NOT_FOUND", "复核生成尝试不存在")
        cursor.execute(
            """
            UPDATE ingestion.processing_attempts
            SET attempt_status = 'failed',
                failure_code = %s,
                failure_message = 'manual review final generation failed',
                finished_at = now()
            WHERE id = %s
            """,
            (failure_code, int(attempt[0])),
        )
        cursor.execute(
            """
            UPDATE ingestion.daily_review_cases
            SET case_status = 'generation_failed',
                failure_code = %s,
                failure_message = 'manual review final generation failed',
                updated_at = now()
            WHERE id = %s
            """,
            (failure_code, int(review_case_id)),
        )
        cursor.execute(
            """
            UPDATE ingestion.processing_runs
            SET run_status = 'awaiting_review',
                failure_code = NULL,
                failure_message = NULL,
                finished_at = NULL,
                updated_at = now()
            WHERE id = %s
            """,
            (int(run_id),),
        )
        self._review_event(
            cursor,
            int(review_case_id),
            "PRICE_REVIEW_GENERATION_FAILED",
            "system:processor",
            int(revision),
        )

    def cancel_price_review(
        self,
        job_id: str,
        case_id: str,
        revision: int,
        actor_username: str,
    ) -> Dict[str, Any]:
        actor = self._review_actor(actor_username)
        return self._run_transaction(
            lambda cursor: self._cancel_price_review(cursor, job_id, case_id, revision, actor),
            "review case could not be cancelled",
        )

    def _cancel_price_review(
        self, cursor: Any, job_id: str, case_id: str, revision: int, actor_username: str
    ) -> Dict[str, Any]:
        case = self._lock_review_case(cursor, job_id, case_id)
        review_case_id, _case_key, status, current_revision, _date_value, _processor, _rules, _source, _json, _sha, _artifact, run_id, _run_status, _requested_processor, _requested_rules = case
        if str(status) not in {"open", "generation_failed"}:
            raise IngestionError("REVIEW_NOT_CANCELLABLE", "复核任务当前不能取消")
        if not isinstance(revision, int) or revision != int(current_revision):
            raise IngestionError("REVIEW_REVISION_CONFLICT", "复核清单已被其他会话更新")
        next_revision = int(current_revision) + 1
        cursor.execute(
            """
            UPDATE ingestion.daily_review_cases
            SET case_status = 'cancelled',
                revision = %s,
                cancelled_at = now(),
                updated_at = now()
            WHERE id = %s
            """,
            (next_revision, int(review_case_id)),
        )
        cursor.execute(
            """
            UPDATE ingestion.processing_runs
            SET run_status = 'cancelled',
                failure_code = NULL,
                failure_message = NULL,
                finished_at = now(),
                updated_at = now()
            WHERE id = %s
            """,
            (int(run_id),),
        )
        self._review_event(
            cursor,
            int(review_case_id),
            "PRICE_REVIEW_CANCELLED",
            actor_username,
            next_revision,
        )
        return self._get_price_review(cursor, job_id, 200, 0)

    @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.review_case_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[5],
                    (
                        int(delivery_row[3])
                        if delivery_row[3] is not None
                        else None
                    ),
                    None,
                )
            if delivery_status == "committed":
                disposition = (
                    "committed"
                    if (
                        delivery_row[7] is not None
                        and int(delivery_row[7]) == processing_run_id
                    )
                    else "already_committed"
                )
                return IngestionOutcome(
                    disposition,
                    envelope.job_id,
                    delivery_row[5],
                    int(delivery_row[3]),
                    int(delivery_row[6]),
                )
            if delivery_status == "recorded_review":
                return IngestionOutcome(
                    "recorded_review",
                    envelope.job_id,
                    delivery_row[5],
                    None,
                    None,
                )
            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"]
        manual_override_ref = envelope.artifacts.get("manual_override_json")
        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
        )
        manual_override_artifact_id = (
            self._ensure_artifact(cursor, manual_override_ref)
            if manual_override_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,
                    manual_override_artifact_id,
                    validated_at
                )
                VALUES (
                    %s, %s, %s, %s, %s::jsonb, 'validating', %s,
                    %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,
                    manual_override_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,
                    manual_override_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,
                    manual_override_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.status == "review_required":
            return self._record_review(
                cursor,
                delivery,
                processing_run_id,
                attempt_id,
                delivery_id,
                source_artifact_id,
                result_artifact_id,
                structured_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,
            manual_override_artifact_id,
        )

    def _record_review(
        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,
    ) -> IngestionOutcome:
        envelope = delivery.envelope
        payload = delivery.structured_payload
        business_date = envelope.business_date
        if business_date is None:
            raise IngestionError("RESULT_CONTRACT_INVALID", "review delivery has no business date")
        source_ref = envelope.artifacts["source_xml"]
        issues = payload.get("review_issues")
        if source_ref is None or not isinstance(issues, list) or not issues:
            raise IngestionError("RESULT_CONTRACT_INVALID", "review delivery is incomplete")
        case_key = "dailyreview-" + hashlib.sha256(
            f"{envelope.job_id}\x00{envelope.delivery_id}".encode("utf-8")
        ).hexdigest()[:32]
        cursor.execute(
            """
            INSERT INTO ingestion.daily_review_cases (
                case_key,
                processing_run_id,
                initial_delivery_id,
                business_date,
                source_sha256,
                processor_version,
                rule_set_sha256,
                review_version,
                case_status
            )
            VALUES (%s, %s, %s, %s, %s, %s, %s, '1.0', 'open')
            RETURNING id
            """,
            (
                case_key,
                processing_run_id,
                delivery_id,
                business_date,
                source_ref.sha256,
                envelope.processor_version,
                envelope.rule_set_sha256,
            ),
        )
        review_case_id = int(cursor.fetchone()[0])
        for issue in issues:
            if not isinstance(issue, Mapping):
                raise IngestionError("RESULT_CONTRACT_INVALID", "review issue is invalid")
            try:
                company_key = issue["company_key"]
                rate_code = issue["rate_code"]
                opera_amount = _decimal_or_none(issue["effective_rate_amount"])
                candidate_prices = issue["candidate_prices"]
                affected_records = int(issue["affected_records"])
                affected_rooms = int(issue["affected_rooms"])
                affected_room_nights = int(issue["affected_room_nights"])
            except (KeyError, TypeError, ValueError):
                raise IngestionError("RESULT_CONTRACT_INVALID", "review issue is invalid") from None
            if (
                not isinstance(company_key, str)
                or not company_key
                or not isinstance(rate_code, str)
                or not rate_code
                or opera_amount is None
                or opera_amount < 0
                or not isinstance(candidate_prices, list)
                or affected_records < 1
                or affected_rooms < 1
                or affected_room_nights < 0
            ):
                raise IngestionError("RESULT_CONTRACT_INVALID", "review issue is invalid")
            cursor.execute(
                """
                INSERT INTO ingestion.daily_review_items (
                    review_case_id,
                    company_key,
                    rate_code,
                    effective_rate_amount,
                    candidate_prices,
                    affected_records,
                    affected_rooms,
                    affected_room_nights
                )
                VALUES (%s, %s, %s, %s, %s::jsonb, %s, %s, %s)
                """,
                (
                    review_case_id,
                    company_key,
                    rate_code,
                    opera_amount,
                    json.dumps(candidate_prices, ensure_ascii=False, separators=(",", ":")),
                    affected_records,
                    affected_rooms,
                    affected_room_nights,
                ),
            )
        self._review_event(
            cursor,
            review_case_id,
            "PRICE_REVIEW_REQUIRED",
            "system:processor",
            0,
        )
        cursor.execute(
            """
            UPDATE ingestion.processing_deliveries
            SET delivery_status = 'recorded_review',
                review_case_id = %s,
                committed_at = now()
            WHERE id = %s
            """,
            (review_case_id, delivery_id),
        )
        cursor.execute(
            """
            UPDATE ingestion.processing_attempts
            SET attempt_status = 'review_required',
                failure_code = NULL,
                failure_message = NULL,
                finished_at = now()
            WHERE id = %s
            """,
            (attempt_id,),
        )
        cursor.execute(
            """
            UPDATE ingestion.processing_runs
            SET run_status = 'awaiting_review',
                result_artifact_id = %s,
                business_date = %s,
                failure_code = NULL,
                failure_message = NULL,
                finished_at = NULL,
                updated_at = now()
            WHERE id = %s
            """,
            (structured_artifact_id, business_date, processing_run_id),
        )
        return IngestionOutcome(
            "recorded_review",
            envelope.job_id,
            business_date,
            None,
            None,
            review_case_id=case_key,
            review_revision=0,
            review_completed_items=0,
            review_total_items=len(issues),
        )

    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(
            """
            SELECT id, revision
            FROM ingestion.daily_review_cases
            WHERE processing_run_id = %s
              AND case_status = 'processing'
            FOR UPDATE
            """,
            (processing_run_id,),
        )
        active_review = cursor.fetchone()
        if active_review:
            review_case_id = int(active_review[0])
            cursor.execute(
                """
                UPDATE ingestion.daily_review_cases
                SET case_status = 'failed',
                    failure_code = %s,
                    failure_message = 'deterministic final generation failed',
                    updated_at = now()
                WHERE id = %s
                """,
                (code, review_case_id),
            )
            self._review_event(
                cursor,
                review_case_id,
                "PRICE_REVIEW_FAILED",
                "system:processor",
                int(active_review[1]),
            )
        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,
        manual_override_artifact_id: Optional[int],
    ) -> IngestionOutcome:
        envelope = delivery.envelope
        business_date = envelope.business_date
        assert business_date is not None
        payload = delivery.structured_payload
        review_case_db_id: Optional[int] = None
        manual_override_sha256: Optional[str] = None
        manually_priced_rows = 0
        if manual_override_artifact_id is not None:
            review_case_key = payload.get("review_case_id")
            manual_override_sha256 = payload.get("manual_override_sha256")
            manually_priced_rows = int(payload.get("manually_priced_rows", 0))
            if (
                not isinstance(review_case_key, str)
                or not isinstance(manual_override_sha256, str)
                or not SHA256_RE.fullmatch(manual_override_sha256)
                or manually_priced_rows < 1
            ):
                raise IngestionError("RESULT_CONTRACT_INVALID", "人工复核结果lineage无效")
            case = self._lock_review_case(cursor, envelope.job_id, review_case_key)
            if (
                str(case[2]) != "processing"
                or int(case[11]) != processing_run_id
                or str(case[7]) != envelope.artifacts["source_xml"].sha256
                or str(case[5]) != envelope.processor_version
                or str(case[6]) != envelope.rule_set_sha256
                or str(case[9]) != manual_override_sha256
            ):
                raise IngestionError("REVIEW_STATE_INVALID", "人工复核确认状态无效")
            review_case_db_id = int(case[0])
        elif payload.get("review_case_id") is not None:
            raise IngestionError("RESULT_CONTRACT_INVALID", "人工复核结果缺少冻结清单产物")
        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
              AND manual_override_sha256 IS NOT DISTINCT FROM %s
            FOR UPDATE
            """,
            (
                source_artifact_id,
                business_date,
                envelope.processor_version,
                envelope.rule_set_sha256,
                manual_override_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,
                    review_case_id,
                    manual_override_sha256,
                    manually_priced_rows,
                    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, %s, %s, %s, 'validated',
                    %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, now()
                )
                RETURNING id
                """,
                (
                    business_date,
                    version_no,
                    processing_run_id,
                    review_case_db_id,
                    manual_override_sha256,
                    manually_priced_rows,
                    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,
                review_case_id = %s,
                committed_at = now()
            WHERE id = %s
            """,
            (daily_version_id, review_case_db_id, delivery_id),
        )
        if review_case_db_id is not None:
            cursor.execute(
                """
                UPDATE ingestion.daily_review_cases
                SET case_status = 'completed',
                    manual_override_artifact_id = %s,
                    failure_code = NULL,
                    failure_message = NULL,
                    completed_at = now(),
                    updated_at = now()
                WHERE id = %s
                """,
                (manual_override_artifact_id, review_case_db_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,
                ),
            ),
        )
