from __future__ import annotations

import json
import hashlib
import sys
from dataclasses import replace
from datetime import UTC, datetime
from pathlib import Path
from types import ModuleType
from typing import cast

import pytest

from gateway.platforms import dualcoach_admin as _dualcoach_admin
from gateway.platforms.telegram_customer_bootstrap import (
    BootstrapState,
    CustomerDraft,
    RoomBootstrapSession,
    RoomBootstrapStore,
    room_bootstrap_state_dir,
)


_ADMIN_SOURCE = '''
import json
from dataclasses import dataclass
from pathlib import Path

class CustomerAdminError(ValueError):
    pass

@dataclass
class Result:
    customer_id: str
    enabled: bool = True

def _resolve_registry_path(root):
    return Path(root) / "customers" / "registry.json"

def _read(path):
    value = json.loads(Path(path).read_text())
    class Value:
        def __init__(self, row):
            for key, item in row.items():
                if isinstance(item, dict):
                    item = Value(item)
                elif isinstance(item, list):
                    item = [Value(value) if isinstance(value, dict) else value for value in item]
                setattr(self, key, item)
    return Value(value)

def activate_customer(profile_root, data_root, customer_id, checklist_evidence_path, staff_membership_evidence_path, *, kst_date=None):
    del data_root, checklist_evidence_path, staff_membership_evidence_path, kst_date
    registry_path = _resolve_registry_path(profile_root)
    value = json.loads(registry_path.read_text())
    row = next(item for item in value["customers"] if item["customer_key"] == customer_id)
    if row["enabled"]:
        raise CustomerAdminError("customer is already enabled")
    row["enabled"] = True
    registry_path.write_text(json.dumps(value))
    data = Path(profile_root) / "data"
    with (data / "customer-activation-audit.jsonl").open("a") as stream:
        stream.write(json.dumps({"customer_id": customer_id}) + "\\n")
    with (data / "nutrition-readiness-audit.jsonl").open("a") as stream:
        stream.write(json.dumps({"customer_id": customer_id}) + "\\n")
    (data / "customer-activation-journal.json").write_text(
        json.dumps({"state": "committed", "customer_id": customer_id})
    )
    (data / "customer-activation-journal.json").chmod(0o600)
    return Result(customer_id)

def validate_committed_activation(profile_root, registry_path=None, customer_id=None):
    del registry_path
    root = Path(profile_root)
    value = json.loads((root / "customers" / "registry.json").read_text())
    row = next(item for item in value["customers"] if item["customer_key"] == customer_id)
    journal = json.loads((root / "data" / "customer-activation-journal.json").read_text())
    if not row["enabled"] or journal != {"state": "committed", "customer_id": customer_id}:
        raise CustomerAdminError("committed activation receipt is missing")
    return True
'''


def _checkin_modules() -> dict[str, ModuleType]:
    return {
        name: module
        for name, module in sys.modules.items()
        if name == "checkin_cli" or name.startswith("checkin_cli.")
    }


def _run_cli(argv: list[str]) -> int:
    modules_before = _checkin_modules()
    path_before = tuple(sys.path)
    preflight_before = _dualcoach_admin.staff_membership_preflight

    async def preserve_fixture_evidence(**kwargs: object) -> dict[str, object]:
        output_path = cast(Path, kwargs["output_path"])
        return cast(dict[str, object], json.loads(output_path.read_text()))

    _dualcoach_admin.staff_membership_preflight = preserve_fixture_evidence
    for name in modules_before:
        del sys.modules[name]
    try:
        return _dualcoach_admin.main(argv)
    finally:
        for name in tuple(_checkin_modules()):
            del sys.modules[name]
        sys.modules.update(modules_before)
        sys.path[:] = path_before
        _dualcoach_admin.staff_membership_preflight = preflight_before
        assert _checkin_modules() == modules_before
        assert tuple(sys.path) == path_before


def _fixture(
    tmp_path: Path,
) -> tuple[Path, Path, RoomBootstrapStore, RoomBootstrapSession]:
    profile = tmp_path / "profile"
    (profile / "customers").mkdir(parents=True)
    (profile / "data").mkdir(mode=0o700)
    _ = (profile / "customers" / "registry.json").write_text(
        json.dumps(
            {
                "owner": {"user_id": "12", "chat_id": "-100", "topic_id": "22"},
                "customers": [
                    {
                        "customer_key": "synthetic_001",
                        "enabled": False,
                        "telegram": {"user_id": "10", "chat_id": "10", "topic_id": "0"},
                    }
                ],
            }
        )
    )
    package_root = tmp_path / "package"
    package = package_root / "checkin_cli"
    package.mkdir(parents=True)
    _ = (package / "__init__.py").write_text("")
    _ = (package / "customer_admin.py").write_text(_ADMIN_SOURCE)
    store = RoomBootstrapStore(room_bootstrap_state_dir(profile))
    prepared = store.prepare_rehearsal_customer_invite(
        CustomerDraft(
            "synthetic_001", "Synthetic", "2026-08-14", "08:00", 0, 1, 2200, 140, ("dinner",)
        ),
        bot_username="dualcoachtestbot",
        owner_id="12",
    )
    claimed = store.claim_rehearsal_customer_invite(
        prepared.customer_link.rsplit("=", 1)[1],
        user_id="10",
        chat_id="10",
        message_id="100",
    )
    awaiting = store.transition(
        claimed.session_id,
        expected_generation=claimed.generation,
        target=BootstrapState.AWAITING_CONSENT,
    )
    awaiting = store.transition(
        awaiting.session_id,
        expected_generation=awaiting.generation,
        target=BootstrapState.AWAITING_ACTIVATION,
    )
    _ = (profile / "config.yaml").write_text("platforms: {}\n")
    deployment = profile / "deployment.json"
    deployment_payload = {
        "candidate_digest": "a" * 64,
        "candidate_core_digest": "b" * 64,
        "candidate_inventory_digest": "c" * 64,
        "hermes_wheel_sha256": "d" * 64,
        "profile_wheel_sha256": "e" * 64,
    }
    deployment.write_text(json.dumps(deployment_payload, sort_keys=True, separators=(",", ":")) + "\n")
    deployment.chmod(0o600)
    now = datetime.now(UTC).isoformat()
    evidence = {
        "schema": "telegram-staff-membership-evidence-v1",
        "phase": "pre_activation",
        **deployment_payload,
        "deployment_receipt_path": str(deployment),
        "deployment_receipt_sha256": hashlib.sha256(deployment.read_bytes()).hexdigest(),
        "customer_id": "synthetic_001",
        "customer_user_id": "10",
        "bootstrap_session_id": awaiting.session_id,
        "bootstrap_generation": awaiting.generation,
        "registry_sha256": hashlib.sha256((profile / "customers/registry.json").read_bytes()).hexdigest(),
        "config_sha256": hashlib.sha256((profile / "config.yaml").read_bytes()).hexdigest(),
        "staff_chat_inventory": [],
        "staff_chat_inventory_sha256": hashlib.sha256(b"[]").hexdigest(),
        "subscription_epoch_id": "epoch-1",
        "subscription_armed_at_utc": now,
        "request_started_at_utc": now,
        "observed_at_utc": now,
        "activation_journal_sha256": "0" * 64,
        "membership_results": [],
        "private_dm_results": [],
    }
    evidence["evidence_sha256"] = hashlib.sha256(
        json.dumps(evidence, sort_keys=True, separators=(",", ":")).encode()
    ).hexdigest()
    evidence_path = profile / "membership.json"
    evidence_path.write_text(json.dumps(evidence, sort_keys=True, separators=(",", ":")) + "\n")
    evidence_path.chmod(0o600)
    return profile, package_root, store, awaiting


def _argv(
    profile: Path, package_root: Path | None, session: RoomBootstrapSession
) -> list[str]:
    argv = [
        "customer",
        "activate",
        "--profile-root",
        str(profile),
        "--data-root",
        str(profile / "data/customers/synthetic_001"),
        "--customer-id",
        "synthetic_001",
        "--checklist-evidence",
        str(profile / "checklist.json"),
        "--staff-membership-evidence",
        str(profile / "membership.json"),
        "--deployment-receipt",
        str(profile / "deployment.json"),
        "--bootstrap-session",
        session.session_id,
        "--expected-generation",
        str(session.generation),
        "--kst-date",
        "2026-08-14",
    ]
    if package_root is not None:
        argv.extend(("--package-root", str(package_root)))
    return argv


def _lines(path: Path) -> list[str]:
    return path.read_text().splitlines() if path.exists() else []


def test_cli_activation_cutover_uses_configured_profile_package(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    profile, package_root, store, awaiting = _fixture(tmp_path)
    monkeypatch.setenv("DUALCOACH_PROFILE_PACKAGE", str(package_root))

    assert _run_cli(_argv(profile, None, awaiting)) == 0

    assert store.get(awaiting.session_id).state is BootstrapState.ACTIVE


def test_cli_activation_cutover_commits_active_and_is_idempotent(
    tmp_path: Path, capsys: pytest.CaptureFixture[str]
) -> None:
    profile, package_root, store, awaiting = _fixture(tmp_path)

    assert _run_cli(_argv(profile, package_root, awaiting)) == 0
    first = cast(dict[str, object], json.loads(capsys.readouterr().out))
    assert _run_cli(_argv(profile, package_root, awaiting)) == 0
    second = cast(dict[str, object], json.loads(capsys.readouterr().out))

    registry = cast(
        dict[str, object],
        json.loads((profile / "customers/registry.json").read_text()),
    )
    assert first["state"] == second["state"] == "ACTIVE"
    assert store.get(awaiting.session_id).state is BootstrapState.ACTIVE
    customers = cast(list[dict[str, object]], registry["customers"])
    assert sum(row["enabled"] is True for row in customers) == 1
    assert len(_lines(profile / "data/customer-activation-audit.jsonl")) == 1
    assert len(_lines(profile / "data/nutrition-readiness-audit.jsonl")) == 1


def test_cli_reconciles_crash_after_customer_activation(
    tmp_path: Path, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]
) -> None:
    profile, package_root, store, awaiting = _fixture(tmp_path)
    real_activate = RoomBootstrapStore.activate_bound_customer

    def crash_after_activation(
        self: RoomBootstrapStore,
        session_id: str,
        *,
        expected_generation: int,
        customer_key: str,
        customer_user_id: str,
        owner_id: str,
    ) -> RoomBootstrapSession:
        del (
            self,
            session_id,
            expected_generation,
            customer_key,
            customer_user_id,
            owner_id,
        )
        raise RuntimeError("crash")

    monkeypatch.setattr(
        RoomBootstrapStore,
        "activate_bound_customer",
        crash_after_activation,
    )
    with pytest.raises(RuntimeError, match="crash"):
        _ = _run_cli(_argv(profile, package_root, awaiting))
    assert store.get(awaiting.session_id).state is BootstrapState.AWAITING_ACTIVATION

    monkeypatch.setattr(RoomBootstrapStore, "activate_bound_customer", real_activate)
    assert _run_cli(_argv(profile, package_root, awaiting)) == 0
    _ = capsys.readouterr()
    assert store.get(awaiting.session_id).state is BootstrapState.ACTIVE
    assert len(_lines(profile / "data/customer-activation-audit.jsonl")) == 1
    assert len(_lines(profile / "data/nutrition-readiness-audit.jsonl")) == 1


@pytest.mark.parametrize(
    "fault", ("wrong_identity", "stale_generation", "missing_session")
)
def test_cli_cutover_preflight_failure_does_not_activate(
    tmp_path: Path, fault: str
) -> None:
    profile, package_root, store, awaiting = _fixture(tmp_path)
    argv = _argv(profile, package_root, awaiting)
    if fault == "wrong_identity":
        registry_path = profile / "customers/registry.json"
        registry = cast(dict[str, object], json.loads(registry_path.read_text()))
        customers = cast(list[dict[str, object]], registry["customers"])
        telegram = cast(dict[str, object], customers[0]["telegram"])
        telegram["user_id"] = "11"
        _ = registry_path.write_text(json.dumps(registry))
    elif fault == "stale_generation":
        argv[argv.index("--expected-generation") + 1] = str(awaiting.generation - 1)
    else:
        argv[argv.index("--bootstrap-session") + 1] = "cb_0000000000000000000000"

    with pytest.raises(ValueError):
        _ = _run_cli(argv)

    assert store.get(awaiting.session_id).state is BootstrapState.AWAITING_ACTIVATION
    assert json.loads((profile / "customers/registry.json").read_text())["customers"][0]["enabled"] is False
    assert not (profile / "data/customer-activation-audit.jsonl").exists()


def test_cli_cutover_rejects_a_second_active_customer_lifecycle(tmp_path: Path) -> None:
    profile, package_root, store, first = _fixture(tmp_path)
    _ = store.activate_bound_customer(
        first.session_id,
        expected_generation=first.generation,
        customer_key="synthetic_001",
        customer_user_id="10",
        owner_id="12",
    )
    prepared = store.prepare_rehearsal_customer_invite(
        replace(first.customer_draft, customer_user_id=None),
        bot_username="dualcoachtestbot",
        owner_id="12",
    )
    claimed = store.claim_rehearsal_customer_invite(
        prepared.customer_link.rsplit("=", 1)[1], user_id="10", chat_id="10", message_id="101"
    )
    second = store.transition(
        claimed.session_id, expected_generation=claimed.generation, target=BootstrapState.AWAITING_CONSENT
    )
    second = store.transition(
        second.session_id, expected_generation=second.generation, target=BootstrapState.AWAITING_ACTIVATION
    )

    with pytest.raises(ValueError):
        _ = _run_cli(_argv(profile, package_root, second))

    assert store.get(first.session_id).state is BootstrapState.ACTIVE
    assert store.get(second.session_id).state is BootstrapState.AWAITING_ACTIVATION
    assert not (profile / "data/customer-activation-audit.jsonl").exists()