"""Repository contracts and an offline reference implementation for ARR ingestion."""

from __future__ import annotations

import copy
import hashlib
import json
import re
import threading
from dataclasses import dataclass, replace
from datetime import date
from decimal import Decimal, InvalidOperation
from typing import Any, Dict, Mapping, Optional, Protocol, Tuple

from arr_ingestion.contracts import ArtifactRef, IngestionError
from arr_ingestion.validation import VerifiedDelivery


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


@dataclass(frozen=True)
class JobRegistration:
    job_id: str
    source: ArtifactRef
    processor_version: str
    rule_set_sha256: str
    attempt_no: int
    idempotency_key: str
    uploaded_filename: Optional[str] = None


@dataclass(frozen=True)
class IngestionOutcome:
    status: str
    job_id: str
    business_date: Optional[date]
    daily_version_id: Optional[int]
    version_no: Optional[int]
    review_case_id: Optional[str] = None
    review_revision: Optional[int] = None
    review_completed_items: Optional[int] = None
    review_total_items: Optional[int] = None


@dataclass(frozen=True)
class ReviewGeneration:
    status: str
    job_id: str
    review_case_id: str
    attempt_no: Optional[int]
    business_date: date
    manual_override_bytes: Optional[bytes]
    manual_override_sha256: Optional[str]
    daily_version_id: Optional[int] = None
    version_no: Optional[int] = None
    source: Optional[ArtifactRef] = None


class IngestionRepository(Protocol):
    def register_job(self, registration: JobRegistration) -> None:
        ...

    def mark_running(self, job_id: str, attempt_no: int) -> None:
        ...

    def record_failure(self, job_id: str, attempt_no: int, failure_code: str) -> None:
        ...

    def commit_delivery(self, delivery: VerifiedDelivery) -> IngestionOutcome:
        ...

    def get_price_review(self, job_id: str, limit: int, offset: int) -> Dict[str, Any]:
        ...

    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]:
        ...

    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:
        ...

    def record_price_review_generation_failure(
        self,
        job_id: str,
        case_id: str,
        attempt_no: int,
        failure_code: str,
    ) -> None:
        ...

    def cancel_price_review(
        self,
        job_id: str,
        case_id: str,
        revision: int,
        actor_username: str,
    ) -> Dict[str, Any]:
        ...


@dataclass
class _MemoryJob:
    registration: JobRegistration
    status: str = "queued"
    failure_code: Optional[str] = None
    business_date: Optional[date] = None


@dataclass
class _MemoryReviewCase:
    case_id: str
    business_date: date
    processor_version: str
    rule_set_sha256: str
    source_sha256: str
    status: str
    revision: int
    items: list[Dict[str, Any]]
    manifest_bytes: Optional[bytes] = None
    manifest_sha256: Optional[str] = None


@dataclass
class _MemoryDailyVersion:
    id: int
    business_date: date
    version_no: int
    identity: Tuple[str, date, str, str, Optional[str]]
    status: str
    records: Tuple[Mapping[str, Any], ...]
    channels: Tuple[Mapping[str, Any], ...]


class InMemoryIngestionRepository:
    """Thread-safe vertical-slice repository; never used as the production fact store."""

    def __init__(self) -> None:
        self._lock = threading.RLock()
        self._jobs: Dict[str, _MemoryJob] = {}
        self._callbacks: Dict[str, Tuple[str, IngestionOutcome]] = {}
        self._versions: Dict[int, _MemoryDailyVersion] = {}
        self._identity_versions: Dict[Tuple[str, date, str, str, Optional[str]], int] = {}
        self._current: Dict[date, int] = {}
        self._reviews: Dict[str, _MemoryReviewCase] = {}
        self._next_review_item_id = 1
        self._next_version_id = 1

    def register_job(self, registration: JobRegistration) -> None:
        if registration.source.role != "source_xml" or registration.attempt_no < 1:
            raise IngestionError("JOB_INVALID", "processing job registration is invalid")
        with self._lock:
            existing = self._jobs.get(registration.job_id)
            if existing is not None:
                if (
                    existing.registration.source != registration.source
                    or existing.registration.processor_version != registration.processor_version
                    or existing.registration.rule_set_sha256 != registration.rule_set_sha256
                    or existing.registration.uploaded_filename != registration.uploaded_filename
                ):
                    raise IngestionError("JOB_CONFLICT", "processing job identity conflicts")
                if registration.attempt_no == existing.registration.attempt_no:
                    if existing.registration != registration:
                        raise IngestionError("JOB_CONFLICT", "processing attempt identity conflicts")
                    return
                if (
                    existing.status != "awaiting_review"
                    or registration.attempt_no != existing.registration.attempt_no + 1
                ):
                    raise IngestionError("JOB_CONFLICT", "processing attempt sequence conflicts")
                existing.registration = registration
                existing.status = "queued"
                return
            self._jobs[registration.job_id] = _MemoryJob(registration=registration)

    def mark_running(self, job_id: str, attempt_no: int) -> None:
        with self._lock:
            job = self._jobs.get(job_id)
            if job is None:
                raise IngestionError("JOB_NOT_FOUND", "processing job was not registered")
            if attempt_no != job.registration.attempt_no:
                raise IngestionError("JOB_NOT_FOUND", "processing attempt was not registered")
            if job.status in {"succeeded", "failed", "cancelled"}:
                raise IngestionError("JOB_TERMINAL", "processing job is already terminal")
            job.status = "running"

    def record_failure(self, job_id: str, attempt_no: int, failure_code: str) -> None:
        with self._lock:
            job = self._jobs.get(job_id)
            if job is None:
                raise IngestionError("JOB_NOT_FOUND", "processing job was not registered")
            if attempt_no != job.registration.attempt_no:
                raise IngestionError("JOB_NOT_FOUND", "processing attempt was not registered")
            if job.status == "succeeded":
                raise IngestionError("JOB_TERMINAL", "processing job is already terminal")
            job.status = "failed"
            job.failure_code = failure_code
            review_case = self._reviews.get(job_id)
            if review_case is not None and review_case.status == "processing":
                review_case.status = "failed"

    def commit_delivery(self, delivery: VerifiedDelivery) -> IngestionOutcome:
        envelope = delivery.envelope
        with self._lock:
            existing_callback = self._callbacks.get(envelope.delivery_id)
            if existing_callback is not None:
                existing_hash, outcome = existing_callback
                if existing_hash != delivery.envelope_sha256:
                    raise IngestionError("CALLBACK_CONFLICT", "delivery identifier conflicts")
                return outcome

            job = self._jobs.get(envelope.job_id)
            if job is None:
                raise IngestionError("JOB_NOT_FOUND", "processing job was not registered")
            registration = job.registration
            source = envelope.artifacts["source_xml"]
            if (
                source is None
                or source.sha256 != registration.source.sha256
                or source.byte_size != registration.source.byte_size
                or source.object_key != registration.source.object_key
                or envelope.attempt_no != registration.attempt_no
                or envelope.processor_version != registration.processor_version
                or envelope.rule_set_sha256 != registration.rule_set_sha256
            ):
                raise IngestionError("JOB_DELIVERY_MISMATCH", "delivery does not match its job")

            if envelope.status == "failed":
                job.status = "failed"
                errors = delivery.structured_payload.get("errors")
                first_error = errors[0] if isinstance(errors, list) and errors else {}
                code = first_error.get("code") if isinstance(first_error, Mapping) else None
                job.failure_code = str(code or "PROCESSING_FAILED")
                review_case = self._reviews.get(envelope.job_id)
                if review_case is not None and review_case.status == "processing":
                    review_case.status = "failed"
                outcome = IngestionOutcome(
                    status="recorded_failure",
                    job_id=envelope.job_id,
                    business_date=envelope.business_date,
                    daily_version_id=None,
                    version_no=None,
                )
                self._callbacks[envelope.delivery_id] = (
                    delivery.envelope_sha256,
                    outcome,
                )
                return outcome

            if envelope.status == "review_required":
                if envelope.business_date is None:
                    raise IngestionError("RESULT_CONTRACT_INVALID", "review delivery has no date")
                issues = delivery.structured_payload.get("review_issues")
                if not isinstance(issues, list) or not issues:
                    raise IngestionError("RESULT_CONTRACT_INVALID", "review issues are invalid")
                case_id = "dailyreview-" + hashlib.sha256(
                    f"{envelope.job_id}\x00{envelope.delivery_id}".encode("utf-8")
                ).hexdigest()[:32]
                if envelope.job_id in self._reviews:
                    case = self._reviews[envelope.job_id]
                else:
                    items: list[Dict[str, Any]] = []
                    for issue in issues:
                        if not isinstance(issue, Mapping):
                            raise IngestionError("RESULT_CONTRACT_INVALID", "review issue is invalid")
                        item = copy.deepcopy(dict(issue))
                        item["item_id"] = self._next_review_item_id
                        self._next_review_item_id += 1
                        item["real_price"] = None
                        item["revision"] = 0
                        items.append(item)
                    case = _MemoryReviewCase(
                        case_id=case_id,
                        business_date=envelope.business_date,
                        processor_version=envelope.processor_version,
                        rule_set_sha256=envelope.rule_set_sha256,
                        source_sha256=registration.source.sha256,
                        status="open",
                        revision=0,
                        items=items,
                    )
                    self._reviews[envelope.job_id] = case
                job.status = "awaiting_review"
                job.business_date = envelope.business_date
                completed = sum(1 for item in case.items if item["real_price"] is not None)
                outcome = IngestionOutcome(
                    status="recorded_review",
                    job_id=envelope.job_id,
                    business_date=envelope.business_date,
                    daily_version_id=None,
                    version_no=None,
                    review_case_id=case.case_id,
                    review_revision=case.revision,
                    review_completed_items=completed,
                    review_total_items=len(case.items),
                )
                self._callbacks[envelope.delivery_id] = (delivery.envelope_sha256, outcome)
                return outcome

            review_case_id = delivery.structured_payload.get("review_case_id")
            review_case: Optional[_MemoryReviewCase] = None
            if review_case_id is not None:
                review_case = self._reviews.get(envelope.job_id)
                manual_ref = envelope.artifacts.get("manual_override_json")
                if (
                    review_case is None
                    or review_case.case_id != review_case_id
                    or review_case.status != "processing"
                    or manual_ref is None
                    or review_case.manifest_sha256 != manual_ref.sha256
                ):
                    raise IngestionError("REVIEW_STATE_INVALID", "人工复核确认状态无效")

            if envelope.business_date is None:
                raise IngestionError("RESULT_CONTRACT_INVALID", "successful delivery has no date")
            manual_override_sha256 = delivery.structured_payload.get("manual_override_sha256")
            if manual_override_sha256 is not None and not isinstance(manual_override_sha256, str):
                raise IngestionError("RESULT_CONTRACT_INVALID", "人工复核清单哈希无效")
            identity = (
                registration.source.sha256,
                envelope.business_date,
                envelope.processor_version,
                envelope.rule_set_sha256,
                manual_override_sha256,
            )
            existing_version_id = self._identity_versions.get(identity)
            if existing_version_id is not None:
                version = self._versions[existing_version_id]
                job.status = "succeeded"
                job.business_date = envelope.business_date
                if review_case is not None:
                    review_case.status = "completed"
                outcome = IngestionOutcome(
                    status="already_committed",
                    job_id=envelope.job_id,
                    business_date=envelope.business_date,
                    daily_version_id=version.id,
                    version_no=version.version_no,
                )
                self._callbacks[envelope.delivery_id] = (
                    delivery.envelope_sha256,
                    outcome,
                )
                return outcome

            version_no = 1 + max(
                (
                    version.version_no
                    for version in self._versions.values()
                    if version.business_date == envelope.business_date
                ),
                default=0,
            )
            version_id = self._next_version_id
            self._next_version_id += 1
            previous_id = self._current.get(envelope.business_date)
            if previous_id is not None:
                self._versions[previous_id].status = "superseded"
            records = delivery.structured_payload.get("records")
            channels = delivery.structured_payload.get("channels")
            version = _MemoryDailyVersion(
                id=version_id,
                business_date=envelope.business_date,
                version_no=version_no,
                identity=identity,
                status="active",
                records=tuple(copy.deepcopy(records if isinstance(records, list) else [])),
                channels=tuple(copy.deepcopy(channels if isinstance(channels, list) else [])),
            )
            self._versions[version_id] = version
            self._identity_versions[identity] = version_id
            self._current[envelope.business_date] = version_id
            job.status = "succeeded"
            job.business_date = envelope.business_date
            if review_case is not None:
                review_case.status = "completed"
            outcome = IngestionOutcome(
                status="committed",
                job_id=envelope.job_id,
                business_date=envelope.business_date,
                daily_version_id=version_id,
                version_no=version_no,
            )
            self._callbacks[envelope.delivery_id] = (
                delivery.envelope_sha256,
                outcome,
            )
            return outcome

    def current_version_id(self, business_date: date) -> Optional[int]:
        with self._lock:
            return self._current.get(business_date)

    def version_status(self, version_id: int) -> Optional[str]:
        with self._lock:
            version = self._versions.get(version_id)
            return version.status if version else None

    def version_records(self, version_id: int) -> Tuple[Mapping[str, Any], ...]:
        with self._lock:
            version = self._versions.get(version_id)
            return tuple(copy.deepcopy(version.records)) if version else tuple()

    @staticmethod
    def _money(value: object, *, two_places: bool = False) -> Decimal:
        try:
            amount = Decimal(str(value))
        except (InvalidOperation, ValueError):
            raise IngestionError("REVIEW_PRICE_INVALID", "人工价格格式无效") from None
        if not amount.is_finite() or amount < 0:
            raise IngestionError("REVIEW_PRICE_INVALID", "人工价格必须为非负金额")
        if two_places:
            if not re.fullmatch(r"(?:0|[1-9][0-9]{0,15})\.[0-9]{2}", str(value)):
                raise IngestionError("REVIEW_PRICE_INVALID", "人工价格必须使用两位小数字符串")
            return amount.quantize(Decimal("0.01"))
        return amount.normalize()

    @staticmethod
    def _decimal_text(value: Decimal) -> str:
        return format(value.normalize(), "f")

    @classmethod
    def _manual_price(cls, value: object) -> Decimal:
        if not isinstance(value, str) or MANUAL_PRICE_RE.fullmatch(value) is None:
            raise IngestionError("REVIEW_PRICE_INVALID", "人工价格必须是非负整数字符串")
        return cls._money(value).quantize(Decimal("0.01"))

    @classmethod
    def _review_price_text(cls, value: object) -> str:
        amount = cls._money(value)
        if amount != amount.to_integral_value():
            raise IngestionError("DATABASE_STATE_INVALID", "人工价格不是整数")
        return format(amount.quantize(Decimal("1")), "f")

    @classmethod
    def _manifest_for(cls, job_id: str, case: _MemoryReviewCase) -> bytes:
        keys = sorted(
            (
                str(item["company_key"]),
                str(item["rate_code"]),
                cls._money(item["effective_rate_amount"]),
            )
            for item in case.items
        )
        issues = [
            {
                "company_key": company_key,
                "rate_code": rate_code,
                "effective_rate_amount": cls._decimal_text(amount),
            }
            for company_key, rate_code, amount in keys
        ]
        by_key = {
            (
                str(item["company_key"]),
                str(item["rate_code"]),
                cls._money(item["effective_rate_amount"]),
            ): item
            for item in case.items
        }
        overrides = []
        for company_key, rate_code, amount in keys:
            price = cls._money(by_key[(company_key, rate_code, amount)]["real_price"], two_places=True)
            overrides.append(
                {
                    "company_key": company_key,
                    "rate_code": rate_code,
                    "effective_rate_amount": cls._decimal_text(amount),
                    "real_price": format(price, ".2f"),
                }
            )
        payload = {
            "review_version": "1.0",
            "review_case_id": case.case_id,
            "job_id": job_id,
            "source_sha256": case.source_sha256,
            "business_date": case.business_date.isoformat(),
            "processor_version": case.processor_version,
            "rule_set_sha256": case.rule_set_sha256,
            "issues": issues,
            "overrides": overrides,
        }
        return json.dumps(
            payload,
            ensure_ascii=False,
            sort_keys=True,
            separators=(",", ":"),
            allow_nan=False,
        ).encode("utf-8")

    @classmethod
    def _review_response(cls, case: _MemoryReviewCase, limit: int, offset: int) -> Dict[str, Any]:
        if limit < 1 or offset < 0:
            raise IngestionError("REVIEW_PAGE_INVALID", "复核分页参数无效")
        total = len(case.items)
        sliced = case.items[offset : offset + limit]
        completed = sum(1 for item in case.items if item["real_price"] is not None)
        safe_items = []
        for item in sliced:
            safe_items.append(
                {
                    "item_id": item["item_id"],
                    "company_key": item["company_key"],
                    "rate_code": item["rate_code"],
                    "effective_rate_amount": item["effective_rate_amount"],
                    "candidate_prices": copy.deepcopy(item["candidate_prices"]),
                    "affected_records": item["affected_records"],
                    "affected_rooms": item["affected_rooms"],
                    "affected_room_nights": item["affected_room_nights"],
                    "real_price": (
                        cls._review_price_text(item["real_price"])
                        if item["real_price"] is not None
                        else None
                    ),
                    "revision": item["revision"],
                }
            )
        return {
            "case_id": case.case_id,
            "case_status": case.status,
            "revision": case.revision,
            "business_date": case.business_date.isoformat(),
            "completed_items": completed,
            "total_items": total,
            "items": safe_items,
            "pagination": {
                "limit": limit,
                "offset": offset,
                "total": total,
                "has_next": offset + len(safe_items) < total,
            },
        }

    def get_price_review(self, job_id: str, limit: int, offset: int) -> Dict[str, Any]:
        with self._lock:
            case = self._reviews.get(job_id)
            if case is None:
                raise IngestionError("REVIEW_NOT_FOUND", "待人工处理任务不存在")
            return self._review_response(case, limit, offset)

    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]:
        del actor_username
        with self._lock:
            case = self._reviews.get(job_id)
            if case is None or case.case_id != case_id:
                raise IngestionError("REVIEW_NOT_FOUND", "待人工处理任务不存在")
            if case.status != "open":
                raise IngestionError("REVIEW_IMMUTABLE", "复核清单已冻结，不能修改")
            if revision != case.revision:
                raise IngestionError("REVIEW_REVISION_CONFLICT", "复核清单已被其他会话更新")
            item = next((value for value in case.items if value["item_id"] == item_id), None)
            if item is None:
                raise IngestionError("REVIEW_ITEM_NOT_FOUND", "待人工处理项不存在")
            price = self._manual_price(real_price)
            item["real_price"] = format(price, ".2f")
            case.revision += 1
            item["revision"] = case.revision
            return self._review_response(case, len(case.items), 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:
        del actor_username
        with self._lock:
            job = self._jobs.get(job_id)
            case = self._reviews.get(job_id)
            if job is None or case is None or case.case_id != case_id:
                raise IngestionError("REVIEW_NOT_FOUND", "待人工处理任务不存在")
            if case.processor_version != processor_version or case.rule_set_sha256 != rule_set_sha256:
                raise IngestionError("REVIEW_RULESET_CHANGED", "处理器或规则已变更，请取消后重新上传")
            if case.status == "completed":
                version_id = self._current.get(case.business_date)
                version = self._versions.get(version_id) if version_id is not None else None
                return ReviewGeneration(
                    "completed", job_id, case.case_id, None, case.business_date, None, case.manifest_sha256,
                    version.id if version else None, version.version_no if version else None
                )
            if case.status == "processing":
                return ReviewGeneration(
                    "processing", job_id, case.case_id, job.registration.attempt_no, case.business_date,
                    None, case.manifest_sha256
                )
            if case.status not in {"open", "generation_failed"}:
                raise IngestionError("REVIEW_NOT_OPEN", "复核任务不能确认生成")
            if revision != case.revision:
                raise IngestionError("REVIEW_REVISION_CONFLICT", "复核清单已被其他会话更新")
            if any(item["real_price"] is None for item in case.items):
                raise IngestionError("REVIEW_INCOMPLETE", "请先填写全部缺价项")
            if case.manifest_bytes is None:
                case.manifest_bytes = self._manifest_for(job_id, case)
                case.manifest_sha256 = hashlib.sha256(case.manifest_bytes).hexdigest()
            next_attempt = job.registration.attempt_no + 1
            job.registration = replace(
                job.registration,
                attempt_no=next_attempt,
                idempotency_key=idempotency_key,
            )
            job.status = "queued"
            case.status = "processing"
            return ReviewGeneration(
                "ready", job_id, case.case_id, next_attempt, case.business_date,
                case.manifest_bytes, case.manifest_sha256,
                source=job.registration.source,
            )

    def record_price_review_generation_failure(
        self,
        job_id: str,
        case_id: str,
        attempt_no: int,
        failure_code: str,
    ) -> None:
        del attempt_no, failure_code
        with self._lock:
            job = self._jobs.get(job_id)
            case = self._reviews.get(job_id)
            if job is None or case is None or case.case_id != case_id:
                raise IngestionError("REVIEW_NOT_FOUND", "待人工处理任务不存在")
            if case.status == "completed":
                return
            case.status = "generation_failed"
            job.status = "awaiting_review"

    def cancel_price_review(
        self,
        job_id: str,
        case_id: str,
        revision: int,
        actor_username: str,
    ) -> Dict[str, Any]:
        del actor_username
        with self._lock:
            job = self._jobs.get(job_id)
            case = self._reviews.get(job_id)
            if job is None or case is None or case.case_id != case_id:
                raise IngestionError("REVIEW_NOT_FOUND", "待人工处理任务不存在")
            if case.status not in {"open", "generation_failed"}:
                raise IngestionError("REVIEW_NOT_CANCELLABLE", "复核任务当前不能取消")
            if revision != case.revision:
                raise IngestionError("REVIEW_REVISION_CONFLICT", "复核清单已被其他会话更新")
            case.status = "cancelled"
            case.revision += 1
            job.status = "cancelled"
            return self._review_response(case, len(case.items), 0)
