"""Explicit operator cutover from committed customer activation to ACTIVE bootstrap."""

from __future__ import annotations

import importlib
import hashlib
import json
import os
import stat
import sys
from collections.abc import Sequence
from dataclasses import dataclass
from datetime import UTC, date, datetime, timedelta
from pathlib import Path
from typing import Callable, ContextManager, Protocol, cast

from .telegram_customer_bootstrap import (
    BootstrapConflict,
    BootstrapError,
    BootstrapState,
    GenerationConflict,
    Role,
    RoomBootstrapSession,
    RoomBootstrapStore,
    room_bootstrap_state_dir,
)


class ActivationCutoverError(BootstrapError):
    pass


class _RuntimeAuthority(Protocol):
    def authorize(
        self, candidate: str, stage: str, predecessor: object | None = None
    ) -> ContextManager[dict[str, object]]: ...


class _CustomerAdmin(Protocol):
    __file__: str | None

    def activate_customer(
        self,
        profile_root: Path,
        data_root: Path,
        customer_id: str,
        checklist_evidence_path: Path,
        staff_membership_evidence_path: Path,
        *,
        kst_date: date | None,
    ) -> object: ...

    def validate_committed_activation(
        self,
        profile_root: Path,
        registry_path: Path | None = None,
        customer_id: str | None = None,
    ) -> bool: ...


@dataclass(frozen=True, slots=True)
class ActivationCutoverResult:
    customer_id: str
    session_id: str
    generation: int
    state: BootstrapState
    reconciled: bool
    runtime_authority_snapshot: dict[str, object] | None = None

    def to_dict(self) -> dict[str, object]:
        return {
            "customer_id": self.customer_id,
            "session_id": self.session_id,
            "generation": self.generation,
            "state": self.state.value,
            "reconciled": self.reconciled,
            "runtime_authority_snapshot": self.runtime_authority_snapshot,
        }


def activate_customer_cutover(
    profile_root: Path | str,
    data_root: Path | str,
    customer_id: str,
    checklist_evidence_path: Path | str,
    staff_membership_evidence_path: Path | str,
    *,
    bootstrap_session_id: str,
    expected_generation: int,
    deployment_receipt_path: Path | str,
    package_root: Path | str | None = None,
    kst_date: date | None = None,
    task26_authority_source: _RuntimeAuthority | None = None,
    task26_candidate_digest: str | None = None,
    task26_authority_predecessor: object | None = None,
    task26_runtime_required: bool = False,
) -> ActivationCutoverResult:
    """Hold external Task26 authority through activation and bootstrap mutation."""
    if task26_authority_source is None:
        if task26_runtime_required:
            raise ActivationCutoverError("Task26 external runtime authority is required")
        return _activate_customer_cutover_authorized(
            profile_root,
            data_root,
            customer_id,
            checklist_evidence_path,
            staff_membership_evidence_path,
            bootstrap_session_id=bootstrap_session_id,
            expected_generation=expected_generation,
            deployment_receipt_path=deployment_receipt_path,
            package_root=package_root,
            kst_date=kst_date,
        )
    if task26_candidate_digest is None:
        raise ActivationCutoverError("Task26 candidate identity is unavailable")
    try:
        with task26_authority_source.authorize(
            task26_candidate_digest,
            "activation",
            task26_authority_predecessor,
        ) as snapshot:
            result = _activate_customer_cutover_authorized(
                profile_root,
                data_root,
                customer_id,
                checklist_evidence_path,
                staff_membership_evidence_path,
                bootstrap_session_id=bootstrap_session_id,
                expected_generation=expected_generation,
                deployment_receipt_path=deployment_receipt_path,
                package_root=package_root,
                kst_date=kst_date,
            )
            return ActivationCutoverResult(
                result.customer_id,
                result.session_id,
                result.generation,
                result.state,
                result.reconciled,
                snapshot,
            )
    except ValueError as exc:
        raise ActivationCutoverError("Task26 activation authority rejected") from exc


def _activate_customer_cutover_authorized(
    profile_root: Path | str,
    data_root: Path | str,
    customer_id: str,
    checklist_evidence_path: Path | str,
    staff_membership_evidence_path: Path | str,
    *,
    bootstrap_session_id: str,
    expected_generation: int,
    deployment_receipt_path: Path | str,
    package_root: Path | str | None = None,
    kst_date: date | None = None,
) -> ActivationCutoverResult:
    """Commit customer activation, then its exact bootstrap lifecycle."""
    root = _profile_root(profile_root)
    state_dir = room_bootstrap_state_dir(root)
    if not state_dir.is_dir() or not (state_dir / "ledger.json").is_file():
        raise ActivationCutoverError("customer bootstrap authority is unavailable")
    store = RoomBootstrapStore(state_dir)
    session = _preflight_bootstrap(
        store,
        bootstrap_session_id,
        expected_generation=expected_generation,
        customer_id=customer_id,
    )
    admin = _profile_admin(root, package_root)
    registry_path = _admin_registry_path(admin, root)
    document = _admin_read(admin, registry_path)
    customer = _preflight_registry(document, session, customer_id)
    enabled = getattr(customer, "enabled", None)
    if enabled is True:
        _validate_committed(admin, root, registry_path, customer_id)
        reconciled = session.state is not BootstrapState.ACTIVE
    elif enabled is False:
        if session.state is BootstrapState.ACTIVE:
            raise ActivationCutoverError(
                "active bootstrap has no committed customer activation"
            )
        _preflight_membership_evidence(
            root,
            registry_path,
            Path(staff_membership_evidence_path),
            Path(deployment_receipt_path),
            session,
            customer_id=customer_id,
        )
        _ = admin.activate_customer(
            root,
            Path(data_root),
            customer_id,
            Path(checklist_evidence_path),
            Path(staff_membership_evidence_path),
            kst_date=kst_date,
        )
        _validate_committed(admin, root, registry_path, customer_id)
        reconciled = False
    else:
        raise ActivationCutoverError("customer activation state is invalid")
    _finalize_membership_activation_binding(
        root,
        Path(staff_membership_evidence_path),
        customer_id=customer_id,
    )
    committed = _activate_bootstrap(
        store,
        session,
        customer_id=customer_id,
    )
    return ActivationCutoverResult(
        customer_id,
        committed.session_id,
        committed.generation,
        committed.state,
        reconciled,
    )


def _preflight_bootstrap(
    store: RoomBootstrapStore,
    session_id: str,
    *,
    expected_generation: int,
    customer_id: str,
) -> RoomBootstrapSession:
    session = store.get(session_id)
    allowed_generation = (
        {session.generation, session.generation - 1}
        if session.state is BootstrapState.ACTIVE
        else {session.generation}
    )
    customer = session.role_claim(Role.CUSTOMER)
    if (
        expected_generation not in allowed_generation
        or session.customer_key != customer_id
        or session.state
        not in {BootstrapState.AWAITING_ACTIVATION, BootstrapState.ACTIVE}
        or customer is None
        or len(session.role_claims) != 1
        or customer.user_id != customer.chat_id
        or customer.topic_id != "0"
        or customer.user_id != session.customer_draft.customer_user_id
        or customer.user_id == session.owner_id
    ):
        raise GenerationConflict("customer activation cutover authority is stale")
    competing = tuple(
        item
        for item in store.list_sessions()
        if item.session_id != session.session_id
        and item.customer_key == customer_id
        and item.state
        in {BootstrapState.AWAITING_ACTIVATION, BootstrapState.ACTIVE}
    )
    if competing:
        raise BootstrapConflict("customer has multiple activation lifecycles")
    return session


def _preflight_registry(
    document: object,
    session: RoomBootstrapSession,
    customer_id: str,
) -> object:
    owner = getattr(document, "owner", None)
    raw_customers = getattr(document, "customers", None)
    if not isinstance(raw_customers, (tuple, list)):
        raise ActivationCutoverError("customer registry identity is unavailable")
    customers = cast(Sequence[object], raw_customers)
    matches = tuple(
        item
        for item in customers
        if getattr(item, "customer_key", None) == customer_id
    )
    claim = session.role_claim(Role.CUSTOMER)
    if len(matches) != 1 or claim is None:
        raise ActivationCutoverError("customer registry identity is unavailable")
    telegram = getattr(matches[0], "telegram", None)
    if (
        str(getattr(owner, "user_id", "")) != session.owner_id
        or str(getattr(telegram, "user_id", "")) != claim.user_id
        or str(getattr(telegram, "chat_id", "")) != claim.chat_id
        or str(getattr(telegram, "topic_id", "")) != claim.topic_id
    ):
        raise ActivationCutoverError("customer registry identity is stale")
    return matches[0]


def _private_json(path: Path, label: str) -> dict[str, object]:
    try:
        metadata = path.lstat()
        value = json.loads(path.read_text(encoding="utf-8"))
    except (OSError, UnicodeDecodeError, json.JSONDecodeError) as exc:
        raise ActivationCutoverError(f"{label} is unavailable") from exc
    if (
        path.is_symlink()
        or not stat.S_ISREG(metadata.st_mode)
        or stat.S_IMODE(metadata.st_mode) != 0o600
        or metadata.st_nlink != 1
        or not isinstance(value, dict)
    ):
        raise ActivationCutoverError(f"{label} is invalid")
    return value


def _preflight_membership_evidence(
    profile_root: Path,
    registry_path: Path,
    evidence_path: Path,
    deployment_path: Path,
    session: RoomBootstrapSession,
    *,
    customer_id: str,
) -> None:
    """Reject stale or mismatched evidence before the profile can write."""
    from .telegram_staff_membership_gate import sha256_json

    from .task26_candidate_derivation import validate_deployment_receipt

    evidence = _private_json(evidence_path, "staff membership evidence")
    deployment = _private_json(deployment_path, "deployment receipt")
    if deployment.get("schema") == "task26-source-deployment-receipt-v2":
        product_binding = _private_json(
            profile_root / "preexecution-product-binding.json",
            "preexecution product binding",
        )
        if deployment.get("product_binding") != product_binding:
            raise ActivationCutoverError("deployment product binding is stale")
        try:
            validate_deployment_receipt(deployment, product_binding)
        except ValueError as exc:
            raise ActivationCutoverError("deployment wheel binding is invalid") from exc
    body = {
        key: value for key, value in evidence.items() if key != "evidence_sha256"
    }
    claim = session.role_claim(Role.CUSTOMER)
    config_path = profile_root / "config.yaml"
    required = {
        "schema": "telegram-staff-membership-evidence-v1",
        "phase": "pre_activation",
        "customer_id": customer_id,
        "customer_user_id": claim.user_id if claim is not None else None,
        "bootstrap_session_id": session.session_id,
        "bootstrap_generation": session.generation,
        "registry_sha256": hashlib.sha256(registry_path.read_bytes()).hexdigest(),
        "config_sha256": hashlib.sha256(config_path.read_bytes()).hexdigest(),
        "deployment_receipt_sha256": hashlib.sha256(
            deployment_path.read_bytes()
        ).hexdigest(),
    }
    if any(evidence.get(key) != value for key, value in required.items()):
        raise ActivationCutoverError("staff membership evidence binding is stale")
    if evidence.get("evidence_sha256") != sha256_json(body):
        raise ActivationCutoverError("staff membership evidence digest is invalid")
    for field in (
        "candidate_digest",
        "candidate_core_digest",
        "candidate_inventory_digest",
        "candidate_product_binding_sha256",
        "hermes_wheel_sha256",
        "profile_wheel_sha256",
    ):
        if evidence.get(field) != deployment.get(field):
            raise ActivationCutoverError("staff membership deployment binding is stale")
    try:
        armed = datetime.fromisoformat(str(evidence["subscription_armed_at_utc"]))
        started = datetime.fromisoformat(str(evidence["request_started_at_utc"]))
        observed = datetime.fromisoformat(str(evidence["observed_at_utc"]))
    except (KeyError, ValueError) as exc:
        raise ActivationCutoverError("staff membership evidence timestamp is invalid") from exc
    current = datetime.now(UTC)
    if (
        any(value.tzinfo is None for value in (armed, started, observed))
        or not armed <= started <= observed <= current
        or current - observed > timedelta(minutes=5)
    ):
        raise ActivationCutoverError("staff membership evidence is stale")


def _finalize_membership_activation_binding(
    profile_root: Path,
    evidence_path: Path,
    *,
    customer_id: str,
) -> None:
    """Bind the committed activation journal before advancing bootstrap state."""
    from .telegram_staff_membership_gate import MembershipJournal

    evidence = _private_json(evidence_path, "staff membership evidence")
    activation_path = profile_root / "data/customer-activation-journal.json"
    activation = _private_json(activation_path, "customer activation journal")
    if activation.get("state") != "committed" or activation.get("customer_id") != customer_id:
        raise ActivationCutoverError("committed customer activation is unavailable")
    activation_digest = hashlib.sha256(activation_path.read_bytes()).hexdigest()
    evidence_digest = hashlib.sha256(evidence_path.read_bytes()).hexdigest()
    journal = MembershipJournal(
        profile_root / "data/onboarding/telegram-staff-membership-v1/events.jsonl"
    )
    matches = [
        row
        for row in journal.verify()
        if row.get("event") == "activation_binding"
        and row.get("customer_id") == customer_id
    ]
    expected = {
        "activation_journal_sha256": activation_digest,
        "membership_evidence_sha256": evidence_digest,
        "subscription_epoch_id": evidence.get("subscription_epoch_id"),
    }
    if matches:
        if all(matches[-1].get(key) == value for key, value in expected.items()):
            return
        raise ActivationCutoverError("membership activation binding conflicts")
    journal.append(
        {
            "event": "activation_binding",
            "customer_id": customer_id,
            "observed_at_utc": datetime.now(UTC).isoformat(),
            **expected,
        }
    )


def _activate_bootstrap(
    store: RoomBootstrapStore,
    session: RoomBootstrapSession,
    *,
    customer_id: str,
) -> RoomBootstrapSession:
    customer = session.role_claim(Role.CUSTOMER)
    if customer is None:
        raise ActivationCutoverError("customer bootstrap identity is unavailable")
    return store.activate_bound_customer(
        session.session_id,
        expected_generation=session.generation,
        customer_key=customer_id,
        customer_user_id=customer.user_id,
        owner_id=session.owner_id,
    )


def _admin_registry_path(admin: _CustomerAdmin, root: Path) -> Path:
    resolve = cast(
        Callable[[Path], Path],
        getattr(admin, "_resolve_registry_path"),
    )
    return resolve(root)


def _admin_read(admin: _CustomerAdmin, path: Path) -> object:
    read = cast(Callable[[Path], object], getattr(admin, "_read"))
    return read(path)


def _validate_committed(
    admin: _CustomerAdmin,
    root: Path,
    registry_path: Path,
    customer_id: str,
) -> None:
    if admin.validate_committed_activation(root, registry_path, customer_id) is not True:
        raise ActivationCutoverError("committed customer activation is unavailable")


def _profile_root(value: Path | str) -> Path:
    root = Path(value)
    if root.is_symlink() or not root.is_dir():
        raise ActivationCutoverError("activation profile root is unavailable")
    return root.resolve()


def _profile_admin(
    root: Path, package_root: Path | str | None
) -> _CustomerAdmin:
    configured = Path(
        package_root
        or os.environ.get("DUALCOACH_PROFILE_PACKAGE")
        or root / "workspace" / "checkin_cli"
    ).resolve()
    package = configured / "checkin_cli"
    source_path = package / "customer_admin.py"
    if not package.is_dir() or not source_path.is_file() or source_path.is_symlink():
        raise ActivationCutoverError("profile-local checkin_cli is unavailable")
    root_text = str(configured)
    if root_text not in sys.path:
        sys.path.insert(0, root_text)
    admin = importlib.import_module("checkin_cli.customer_admin")
    source = Path(str(admin.__file__)).resolve()
    if source != source_path.resolve():
        raise ActivationCutoverError("loaded checkin_cli is not profile-local")
    return cast(_CustomerAdmin, cast(object, admin))


__all__ = [
    "ActivationCutoverError",
    "ActivationCutoverResult",
    "activate_customer_cutover",
]