# Copyright (c) 2026 Nous Research
"""Pure canonical registry migration for a channel inbox route."""

from __future__ import annotations

from dataclasses import asdict, replace
import hashlib
import hmac
import json
import re

from pydantic import JsonValue, TypeAdapter, ValidationError

from checkin_cli.channel_inbox_migration_models import (
    ChannelInboxMigrationError,
    ChannelInboxMigrationProposal,
    ChannelInboxMigrationRequest,
    MigrationRejectReason,
    TelegramRoute,
)
from checkin_cli.customer_coaching import CustomerRegistryError, RegistryDocument
from gateway.platforms.telegram_channel_inbox_config import (
    channel_inbox_capability_digest,
)

_DIGEST = re.compile(r"[a-f0-9]{64}")
_JSON_OBJECT = TypeAdapter(dict[str, JsonValue])
_JSON_OBJECTS = TypeAdapter(list[dict[str, JsonValue]])
_MAX_TOPIC_ID = (1 << 52) - 1


def _sha256(payload: bytes) -> str:
    return hashlib.sha256(payload).hexdigest()


def _canonical_registry(registry: dict[str, JsonValue]) -> bytes:
    return (json.dumps(registry, ensure_ascii=False, indent=2) + "\n").encode()


def _parse_registry(
    payload: bytes,
) -> tuple[dict[str, JsonValue], list[dict[str, JsonValue]]]:
    try:
        document = RegistryDocument.model_validate_json(payload)
        registry = _JSON_OBJECT.validate_json(payload)
        customers = _JSON_OBJECTS.validate_python(registry.get("customers"))
    except (CustomerRegistryError, ValidationError) as exc:
        raise ChannelInboxMigrationError(MigrationRejectReason.REGISTRY) from exc
    if document.registry_mode != "ordinary_v1":
        raise ChannelInboxMigrationError(MigrationRejectReason.REGISTRY)
    if _canonical_registry(registry) != payload:
        raise ChannelInboxMigrationError(MigrationRejectReason.REGISTRY)
    return registry, customers


def _route(value: JsonValue) -> TelegramRoute:
    try:
        route = _JSON_OBJECT.validate_python(value)
    except ValidationError as exc:
        raise ChannelInboxMigrationError(MigrationRejectReason.IDENTITY) from exc
    fields = tuple(
        str(route.get(name, "")).strip() for name in ("user_id", "chat_id", "topic_id")
    )
    if any(not field for field in fields):
        raise ChannelInboxMigrationError(MigrationRejectReason.IDENTITY)
    return TelegramRoute(*fields)


def _canonical_integer(value: str, reason: MigrationRejectReason) -> int:
    try:
        parsed = int(value)
    except ValueError as exc:
        raise ChannelInboxMigrationError(reason) from exc
    if parsed == 0 or value != str(parsed):
        raise ChannelInboxMigrationError(reason)
    return parsed


def _proposal_digest(proposal: ChannelInboxMigrationProposal) -> str:
    payload = {
        "after_sha256": proposal.after_sha256,
        "before_sha256": proposal.before_sha256,
        "authority": asdict(proposal.authority),
        "capability_digest": proposal.capability_digest,
        "customer_key": proposal.customer_key,
        "new_route": asdict(proposal.new_route),
        "old_route": asdict(proposal.old_route),
        "schema": "nutricoach-channel-inbox-route-migration-v1",
    }
    canonical = json.dumps(
        payload, ensure_ascii=True, separators=(",", ":"), sort_keys=True
    )
    return hashlib.sha256(canonical.encode("ascii")).hexdigest()


def _validate_digest(value: str) -> None:
    if _DIGEST.fullmatch(value) is None:
        raise ChannelInboxMigrationError(MigrationRejectReason.DIGEST)


def _validate_proposal(proposal: ChannelInboxMigrationProposal) -> None:
    _validate_digest(proposal.candidate_digest)
    _validate_digest(proposal.capability_digest)
    if proposal.capability_digest != channel_inbox_capability_digest(
        proposal.authority,
    ):
        raise ChannelInboxMigrationError(MigrationRejectReason.AUTHORITY)
    expected = _proposal_digest(proposal)
    if not hmac.compare_digest(expected, proposal.proposal_digest):
        raise ChannelInboxMigrationError(MigrationRejectReason.PROPOSAL)


def _migrated_bytes(
    registry_bytes: bytes,
    *,
    customer_key: str,
    new_route: TelegramRoute,
) -> tuple[bytes, TelegramRoute]:
    registry, customers = _parse_registry(registry_bytes)
    matches = [row for row in customers if row.get("customer_key") == customer_key]
    if len(matches) != 1:
        raise ChannelInboxMigrationError(MigrationRejectReason.CUSTOMER)
    customer = matches[0]
    if customer.get("enabled") is not True:
        raise ChannelInboxMigrationError(MigrationRejectReason.ENABLED)
    try:
        consent = _JSON_OBJECT.validate_python(customer.get("ai_processing_consent"))
    except ValidationError as exc:
        raise ChannelInboxMigrationError(MigrationRejectReason.CONSENT) from exc
    if consent.get("granted") is not True:
        raise ChannelInboxMigrationError(MigrationRejectReason.CONSENT)
    old_route = _route(customer.get("telegram"))
    if old_route.user_id != new_route.user_id:
        raise ChannelInboxMigrationError(MigrationRejectReason.IDENTITY)
    owner_route = _route(registry.get("owner"))
    occupied = {
        owner_route,
        *(_route(row.get("telegram")) for row in customers if row is not customer),
    }
    if new_route in occupied or (new_route.chat_id, new_route.topic_id) in {
        (route.chat_id, route.topic_id) for route in occupied
    }:
        raise ChannelInboxMigrationError(MigrationRejectReason.COLLISION)
    customer["telegram"] = asdict(new_route)
    customers_json: list[JsonValue] = [customer for customer in customers]
    registry["customers"] = customers_json
    return _canonical_registry(registry), old_route


def plan_registry_migration(
    registry_bytes: bytes,
    request: ChannelInboxMigrationRequest,
) -> ChannelInboxMigrationProposal:
    """Build an immutable migration proposal without writing files."""
    _validate_digest(request.authority.candidate_digest)
    _validate_digest(request.capability_digest)
    if request.capability_digest != channel_inbox_capability_digest(
        request.authority,
    ):
        raise ChannelInboxMigrationError(MigrationRejectReason.AUTHORITY)
    _ = _canonical_integer(
        request.authority.direct_messages_chat_id,
        MigrationRejectReason.IDENTITY,
    )
    topic_id = _canonical_integer(
        request.direct_messages_topic_id,
        MigrationRejectReason.TOPIC,
    )
    if topic_id <= 0 or topic_id > _MAX_TOPIC_ID:
        raise ChannelInboxMigrationError(MigrationRejectReason.TOPIC)
    new_route = TelegramRoute(
        request.telegram_user_id.strip(),
        request.authority.direct_messages_chat_id,
        request.direct_messages_topic_id,
    )
    migrated, old_route = _migrated_bytes(
        registry_bytes,
        customer_key=request.customer_key,
        new_route=new_route,
    )
    before_sha256 = _sha256(registry_bytes)
    after_sha256 = _sha256(migrated)
    proposal = ChannelInboxMigrationProposal(
        request.customer_key,
        before_sha256,
        after_sha256,
        old_route,
        new_route,
        request.authority,
        request.capability_digest,
        "",
    )
    return replace(proposal, proposal_digest=_proposal_digest(proposal))


def apply_registry_migration(
    registry_bytes: bytes,
    proposal: ChannelInboxMigrationProposal,
    approval: str,
) -> bytes:
    """Apply an exact approved proposal in memory."""
    _validate_proposal(proposal)
    if not hmac.compare_digest(
        approval.encode(),
        proposal.approval_phrase.encode(),
    ):
        raise ChannelInboxMigrationError(MigrationRejectReason.APPROVAL)
    if _sha256(registry_bytes) != proposal.before_sha256:
        raise ChannelInboxMigrationError(MigrationRejectReason.BEFORE)
    migrated, old_route = _migrated_bytes(
        registry_bytes,
        customer_key=proposal.customer_key,
        new_route=proposal.new_route,
    )
    if old_route != proposal.old_route or _sha256(migrated) != proposal.after_sha256:
        raise ChannelInboxMigrationError(MigrationRejectReason.AFTER)
    return migrated


def rollback_registry_migration(
    registry_bytes: bytes,
    proposal: ChannelInboxMigrationProposal,
) -> bytes:
    """Rollback only the exact migrated registry bytes."""
    _validate_proposal(proposal)
    if _sha256(registry_bytes) != proposal.after_sha256:
        raise ChannelInboxMigrationError(MigrationRejectReason.AFTER)
    rolled_back, old_route = _migrated_bytes(
        registry_bytes,
        customer_key=proposal.customer_key,
        new_route=proposal.old_route,
    )
    if (
        old_route != proposal.new_route
        or _sha256(rolled_back) != proposal.before_sha256
    ):
        raise ChannelInboxMigrationError(MigrationRejectReason.BEFORE)
    return rolled_back