#!/usr/bin/env python3
"""Offline installed-wheel QA for the r63 first-claim customer lifecycle."""

from __future__ import annotations

import asyncio
import hashlib
import importlib.metadata
import json
import os
import re
import shutil
import socket
import subprocess
import sys
import threading
from concurrent.futures import ThreadPoolExecutor
from dataclasses import dataclass
from pathlib import Path
from types import SimpleNamespace
from urllib.parse import parse_qs, urlparse

ROOT = Path(__file__).resolve().parents[1]
RUN_ROOT = ROOT / "workspace" / "disposable-root"
GUARD_ROOT = ROOT / "guard"
RESULT_PATH = ROOT / "r63-first-claim-wheel-qa-result.json"
SITE_PACKAGES = Path(sys.prefix) / "lib" / f"python{sys.version_info.major}.{sys.version_info.minor}" / "site-packages"
ASSERTIONS = 0
NETWORK_ATTEMPTS: list[str] = []


def check(condition: object, label: str) -> None:
    global ASSERTIONS
    ASSERTIONS += 1
    if not condition:
        raise AssertionError(label)


def expect_raises(exc_type: type[BaseException], label: str, callback: object) -> BaseException:
    global ASSERTIONS
    ASSERTIONS += 1
    try:
        callback()  # type: ignore[operator]
    except exc_type as exc:
        return exc
    except Exception as exc:
        raise AssertionError(f"{label}: expected {exc_type.__name__}, got {type(exc).__name__}") from exc
    raise AssertionError(f"{label}: expected {exc_type.__name__}")


def canonical_bytes(value: object) -> bytes:
    return json.dumps(
        value,
        ensure_ascii=False,
        sort_keys=True,
        separators=(",", ":"),
    ).encode("utf-8")


def write_private_json(path: Path, value: object) -> None:
    path.parent.mkdir(parents=True, exist_ok=True, mode=0o700)
    for directory in (path.parent, *path.parent.parents):
        if directory == RUN_ROOT.parent:
            break
        if directory.exists():
            directory.chmod(0o700)
    path.write_bytes(canonical_bytes(value))
    path.chmod(0o600)


def profile(root: Path) -> Path:
    root.mkdir(mode=0o700)
    root.chmod(0o700)
    (root / "data").mkdir(mode=0o700)
    (root / "data").chmod(0o700)
    (root / "customers").mkdir(mode=0o700)
    (root / "customers").chmod(0o700)
    (root / "config.yaml").write_text(
        "platforms:\n"
        "  telegram:\n"
        "    extra:\n"
        "      production_preflight:\n"
        "        expected_bot_username: dualcoachtestbot\n",
        encoding="utf-8",
    )
    (root / "config.yaml").chmod(0o600)
    write_private_json(
        root / "customers" / "registry.json",
        {
            "version": 1,
            "owner": {"user_id": "9001", "chat_id": "9002", "topic_id": "0"},
            "customers": [],
            "admission_policy": {
                "schema": "nutricoach-multi-customer-v1",
                "candidate_digest": "a" * 64,
                "max_enabled_customers": 6,
            },
        },
    )
    return root


def command_environment() -> dict[str, str]:
    environment = dict(os.environ)
    existing = environment.get("PYTHONPATH")
    environment["PYTHONPATH"] = str(GUARD_ROOT) + (os.pathsep + existing if existing else "")
    environment["QA_NETWORK_LOG"] = str(ROOT / "logs" / "blocked-network-attempts.log")
    return environment


def run_cli(*args: str) -> dict[str, object]:
    completed = subprocess.run(
        [str(Path(sys.prefix) / "bin" / "dualcoach_admin"), *args],
        cwd=RUN_ROOT,
        env=command_environment(),
        capture_output=True,
        text=True,
        check=False,
    )
    check(completed.returncode == 0, f"candidate CLI exits successfully: {completed.stderr}")
    check(completed.stderr == "", "candidate CLI emits no stderr")
    try:
        decoded = json.loads(completed.stdout)
    except json.JSONDecodeError as exc:
        raise AssertionError("candidate CLI emits exactly one JSON object") from exc
    check(isinstance(decoded, dict), "candidate CLI JSON is an object")
    return decoded


def token_from_link(link: object) -> str:
    check(isinstance(link, str), "invite has a customer link")
    parsed = urlparse(link)
    values = parse_qs(parsed.query, keep_blank_values=True)
    check(parsed.scheme == "https" and parsed.netloc == "t.me", "invite uses the Telegram HTTPS surface")
    check(parsed.path == "/dualcoachtestbot", "invite targets the configured bot only")
    check(parsed.params == "" and parsed.fragment == "", "invite has no non-token URL components")
    check(set(values) == {"start"} and len(values["start"]) == 1, "invite query contains only one start token")
    token = values["start"][0]
    check(re.fullmatch(r"rc1_[A-Za-z0-9_-]{22}", token) is not None, "invite start value is one opaque rc1 token")
    return token


@dataclass
class FakeActor:
    id: int
    is_bot: bool = False


@dataclass
class FakeChat:
    id: int
    type: str


class FakeStartMessage:
    def __init__(self, *, user_id: int, chat_id: int, chat_type: str, is_bot: bool) -> None:
        self.from_user = FakeActor(user_id, is_bot)
        self.chat = FakeChat(chat_id, chat_type)
        self.message_id = 77
        self.sender_chat = None
        self.replies: list[str] = []

    async def reply_text(self, text: str) -> None:
        self.replies.append(text)


def main() -> int:
    global ASSERTIONS
    if RUN_ROOT.exists():
        shutil.rmtree(RUN_ROOT)
    RUN_ROOT.mkdir(mode=0o700)
    GUARD_ROOT.mkdir(exist_ok=True)
    (ROOT / "logs" / "blocked-network-attempts.log").unlink(missing_ok=True)

    original_connect = socket.socket.connect

    def block_connect(_socket: socket.socket, address: object) -> None:
        NETWORK_ATTEMPTS.append(repr(address))
        raise AssertionError(f"network connection blocked: {address!r}")

    socket.socket.connect = block_connect
    try:
        import checkin_cli
        import gateway
        from checkin_cli.cli import app as profile_app
        from gateway.platforms.dualcoach_activation_cutover import (
            ActivationCutoverError,
            validate_first_claim_owner_review,
        )
        from gateway.platforms.telegram import TelegramAdapter
        from gateway.platforms.telegram_customer_bootstrap import (
            BootstrapError,
            BootstrapState,
            CustomerDraft,
            RoomBootstrapStore,
            room_bootstrap_state_dir,
        )
        from gateway.platforms.telegram_customer_bootstrap_registration import (
            TelegramCustomerBootstrapRegistration,
        )
        from gateway.platforms.telegram_nutrition_onboarding_publication_outbox import (
            GatewayOnboardingPublicationOutbox,
        )

        for module, label in (
            (checkin_cli, "checkin_cli"),
            (gateway, "gateway"),
            (sys.modules[validate_first_claim_owner_review.__module__], "activation cutover"),
            (sys.modules[GatewayOnboardingPublicationOutbox.__module__], "publication outbox"),
        ):
            source = Path(module.__file__).resolve()
            check(source.is_relative_to(SITE_PACKAGES), f"{label} imports from the fresh installed wheel venv")

        check(importlib.metadata.version("hermes-agent") == "0.17.0", "installed Hermes distribution is the candidate version")
        check(importlib.metadata.version("physique-checkin-cli") == "0.1.0", "installed profile distribution is the candidate version")

        primary_profile = profile(RUN_ROOT / "first-claim-profile")
        invite = run_cli(
            "customer",
            "invite",
            "--profile-root",
            str(primary_profile),
            "--first-claim",
            "--json",
        )
        check(set(invite) == {"customer_key", "customer_link", "draft_digest", "expires_at", "generation", "session_id"}, "first-claim CLI response has the expected machine contract")
        check(isinstance(invite["customer_key"], str) and re.fullmatch(r"lead_[a-f0-9]{16}", invite["customer_key"]) is not None, "first-claim CLI generates an opaque lead customer key")
        token = token_from_link(invite["customer_link"])
        link = str(invite["customer_link"])
        for forbidden in (str(invite["customer_key"]), "9001", "9002", str(invite["draft_digest"])):
            check(forbidden not in link, "customer link excludes profile identity and draft material")

        state_dir = room_bootstrap_state_dir(primary_profile)
        store = RoomBootstrapStore(state_dir)
        prepared = store.rehearsal_customer_invite_session(token)
        check(prepared.first_claim is True, "CLI first-claim invite persists first-claim policy")
        check(prepared.customer_draft.customer_user_id is None, "CLI first-claim invite does not require or persist a draft identity")
        check(prepared.state is BootstrapState.PREPARED and prepared.role_claims == (), "CLI first-claim invite starts unclaimed")

        adapter = SimpleNamespace(
            _get_room_bootstrap_transport=lambda: SimpleNamespace(store=store),
            _get_nutrition_coaching=lambda: object(),
        )
        invalid_messages = (
            ("owner", FakeStartMessage(user_id=9001, chat_id=9001, chat_type="private", is_bot=False)),
            ("group", FakeStartMessage(user_id=1101, chat_id=-100777, chat_type="group", is_bot=False)),
            ("bot", FakeStartMessage(user_id=1102, chat_id=1102, chat_type="private", is_bot=True)),
        )
        for kind, message in invalid_messages:
            asyncio.run(TelegramAdapter._handle_rehearsal_customer_start(adapter, message, token))
            current = store.get(prepared.session_id)
            check(message.replies == ["유효하지 않거나 만료된 가입 링크입니다."], f"{kind}-style claim receives the denial surface")
            check(current.state is BootstrapState.PREPARED and current.role_claims == (), f"{kind}-style claim does not consume the invite")
            check(current.customer_draft.customer_user_id is None, f"{kind}-style claim does not bind an identity")

        barrier = threading.Barrier(3)

        def concurrent_claim(user_id: str) -> tuple[str, object]:
            contender = RoomBootstrapStore(state_dir)
            barrier.wait(timeout=10)
            try:
                claimed = contender.claim_rehearsal_customer_invite(
                    token,
                    user_id=user_id,
                    chat_id=user_id,
                    message_id="77",
                )
            except BootstrapError as error:
                return "rejected", type(error).__name__
            return "winner", claimed

        with ThreadPoolExecutor(max_workers=2) as executor:
            futures = [executor.submit(concurrent_claim, user) for user in ("1001", "1002")]
            barrier.wait(timeout=10)
            claims = [future.result(timeout=10) for future in futures]
        winners = [value for outcome, value in claims if outcome == "winner"]
        rejected = [value for outcome, value in claims if outcome == "rejected"]
        check(len(winners) == 1 and len(rejected) == 1, "two concurrent eligible private claimants produce exactly one winner")
        winning_session = winners[0]
        winner_id = winning_session.customer_draft.customer_user_id  # type: ignore[union-attr]
        check(winner_id in {"1001", "1002"}, "single winner is one eligible private claimant")
        check(winning_session.role_claims[0].user_id == winner_id and winning_session.role_claims[0].chat_id == winner_id, "winner claim is bound to one private identity")  # type: ignore[union-attr]

        restarted_store = RoomBootstrapStore(state_dir)
        restarted = restarted_store.get(prepared.session_id)
        check(restarted.first_claim is True, "restart preserves first-claim policy")
        check(restarted.state is BootstrapState.REGISTERING, "restart preserves the consumed registration state")
        check(restarted.customer_draft.customer_user_id == winner_id, "restart preserves the winning customer identity")
        check(len(restarted.role_claims) == 1 and restarted.role_claims[0].user_id == winner_id, "restart preserves one durable customer claim")

        prior_package = os.environ.get("DUALCOACH_PROFILE_PACKAGE")
        os.environ["DUALCOACH_PROFILE_PACKAGE"] = str(SITE_PACKAGES)
        try:
            registration = TelegramCustomerBootstrapRegistration(
                primary_profile,
                restarted_store,
                package_root=SITE_PACKAGES,
            )
            disabled = registration.handoff_rehearsal_customer(restarted)
        finally:
            if prior_package is None:
                os.environ.pop("DUALCOACH_PROFILE_PACKAGE", None)
            else:
                os.environ["DUALCOACH_PROFILE_PACKAGE"] = prior_package
        check(disabled.session.state is BootstrapState.AWAITING_CONSENT, "first-claim registration enters consent waiting")
        registered_rows = json.loads((primary_profile / "customers" / "registry.json").read_text(encoding="utf-8"))["customers"]
        matching_rows = [row for row in registered_rows if row["customer_key"] == restarted.customer_key]
        check(len(matching_rows) == 1 and matching_rows[0]["enabled"] is False, "first-claim registration remains disabled")
        check(matching_rows[0]["telegram"]["user_id"] == winner_id, "disabled registration retains the claimed private identity")

        targeted_draft = CustomerDraft(
            customer_key="targeted_001",
            display_name="Targeted Customer",
            starts_on="2026-08-31",
            daily_time="08:00",
            weekly_weekday=0,
            monthly_day=1,
            calories_kcal=2000,
            protein_g=100,
            meals=("breakfast", "lunch", "dinner"),
            customer_user_id="3001",
        )
        targeted_path = RUN_ROOT / "targeted-draft.json"
        write_private_json(targeted_path, targeted_draft.to_dict())
        targeted_invite = run_cli(
            "customer",
            "invite",
            "--profile-root",
            str(primary_profile),
            "--draft",
            str(targeted_path),
            "--json",
        )
        targeted_token = token_from_link(targeted_invite["customer_link"])
        targeted_session = restarted_store.rehearsal_customer_invite_session(targeted_token)
        check(targeted_session.first_claim is False, "draft-backed invite remains targeted")
        expect_raises(
            BootstrapError,
            "wrong private targeted claimant is rejected",
            lambda: restarted_store.claim_rehearsal_customer_invite(targeted_token, user_id="3002", chat_id="3002", message_id="90"),
        )
        check(restarted_store.get(targeted_session.session_id).state is BootstrapState.PREPARED, "wrong targeted claimant does not consume the invite")
        targeted_claimed = restarted_store.claim_rehearsal_customer_invite(targeted_token, user_id="3001", chat_id="3001", message_id="91")
        check(targeted_claimed.first_claim is False and targeted_claimed.customer_draft.customer_user_id == "3001", "intended targeted claimant remains accepted")

        legacy_seed = RoomBootstrapStore(RUN_ROOT / "legacy-seed")
        legacy_prepared = legacy_seed.prepare_rehearsal_customer_invite(
            CustomerDraft(
                customer_key="legacy_targeted_001",
                display_name="Legacy Targeted Customer",
                starts_on="2026-08-31",
                daily_time="08:00",
                weekly_weekday=0,
                monthly_day=1,
                calories_kcal=2000,
                protein_g=100,
                meals=("breakfast", "lunch", "dinner"),
                customer_user_id="4001",
            ),
            bot_username="dualcoachtestbot",
            owner_id="9001",
        )
        seed_ledger = json.loads(legacy_seed.ledger_path.read_text(encoding="utf-8"))
        for row in seed_ledger["sessions"]:
            row.pop("first_claim", None)
        digest_payload = {"schema": seed_ledger["schema"], "sessions": seed_ledger["sessions"]}
        seed_ledger["digest"] = hashlib.sha256(canonical_bytes(digest_payload)).hexdigest()
        legacy_root = RUN_ROOT / "legacy-r62-state"
        legacy_root.mkdir(mode=0o700)
        write_private_json(legacy_root / "ledger.json", seed_ledger)
        legacy_store = RoomBootstrapStore(legacy_root)
        legacy = legacy_store.rehearsal_customer_invite_session(token_from_link(legacy_prepared.customer_link))
        check(legacy.first_claim is False and legacy.customer_draft.customer_user_id == "4001", "r62 session missing first_claim loads as a targeted session")
        legacy_token = token_from_link(legacy_prepared.customer_link)
        expect_raises(
            BootstrapError,
            "legacy targeted session rejects a non-intended claimant",
            lambda: legacy_store.claim_rehearsal_customer_invite(legacy_token, user_id="4002", chat_id="4002", message_id="1"),
        )
        legacy_claimed = legacy_store.claim_rehearsal_customer_invite(legacy_token, user_id="4001", chat_id="4001", message_id="2")
        check(legacy_claimed.first_claim is False and legacy_claimed.customer_draft.customer_user_id == "4001", "legacy targeted session retains the pre-existing identity")

        onboarding_root = primary_profile / "data" / "customers" / restarted.customer_key / "nutrition-onboarding"
        owner_digest = "c" * 64
        write_private_json(onboarding_root / "baseline-v1.json", {"customer_key": restarted.customer_key, "owner_review_receipt": owner_digest})
        write_private_json(onboarding_root / "readiness-receipt-v1.json", {"owner_review_receipt": owner_digest, "delivery_enabled": False, "activation_enabled": False})
        outbox = GatewayOnboardingPublicationOutbox(primary_profile)
        expect_raises(
            ActivationCutoverError,
            "forged matching owner-review digests without an authenticated callback are rejected",
            lambda: validate_first_claim_owner_review(primary_profile, disabled.session),
        )
        check(matching_rows[0]["enabled"] is False, "forged review cannot enable the customer")

        route = ("9002", "0")
        payload = {"kind": "owner-review", "customer_key": restarted.customer_key}
        render_identity = "d" * 64
        generation = 42
        claimed_publication, created = outbox.claim(
            session_id=disabled.session.session_id,
            generation=generation,
            payload=payload,
            route=route,
            role="owner",
            render_identity=render_identity,
        )
        check(created is True and claimed_publication.state == "DISPATCHING", "owner-review publication is claimed before its synthetic receipt")
        receipted = outbox.record_receipt(
            session_id=disabled.session.session_id,
            generation=generation,
            chat_id=route[0],
            topic_id=route[1],
            message_id=4242,
        )
        check(receipted.state == "RECEIPTED", "synthetic local owner publication receipt is authenticated")
        committed = outbox.mark_committed(
            session_id=disabled.session.session_id,
            generation=generation,
            payload=payload,
            route=route,
            role="owner",
            render_identity=render_identity,
            message_id=4242,
        )
        check(committed.state == "COMMITTED", "owner publication is committed before the callback receipt")
        callback = outbox.record_owner_callback(
            session_id=disabled.session.session_id,
            customer_key=disabled.session.customer_key,
            action="Approve",
            actor_user_id=9001,
            authority=(9001, 9002, 0),
            route=route,
            message_id=4242,
            update_id=7001,
            consumed_updates_before=(),
            callback_data="local-owner-approve",
            publication_generation=generation,
        )
        check(callback.action == "Approve" and callback.session_id == disabled.session.session_id, "authenticated Gateway owner callback is persisted")
        validate_first_claim_owner_review(primary_profile, disabled.session)
        check(matching_rows[0]["enabled"] is False, "authenticated owner callback unlocks only activation preflight, not activation")
        check(not (primary_profile / "data" / "customer-activation-journal.json").exists(), "owner callback alone creates no activation journal")
        check(not (primary_profile / "data" / "customer-activation-receipts" / f"{restarted.customer_key}.json").exists(), "owner callback alone creates no activation receipt")
        check(restarted_store.get(disabled.session.session_id).state is BootstrapState.AWAITING_CONSENT, "owner callback alone does not advance the disabled bootstrap lifecycle")

        profile_distribution = importlib.metadata.distribution("physique-checkin-cli")
        profile_entrypoints = [path for path in profile_distribution.files or () if path.name == "entry_points.txt"]
        check(profile_entrypoints == [], "profile wheel exposes no direct activation console entry point")
        profile_commands = {
            command.name or command.callback.__name__
            for command in profile_app.registered_commands
            if command.callback is not None
        }
        check(profile_commands == {"record", "import-history", "import-baseline"}, "profile CLI has only its check-in commands")
        check("activate" not in profile_commands, "direct profile activation CLI is absent")
        check(NETWORK_ATTEMPTS == [], "in-process QA attempted no socket connection")
        network_log = ROOT / "logs" / "blocked-network-attempts.log"
        check(not network_log.exists() or network_log.read_text(encoding="utf-8") == "", "candidate subprocess attempted no socket connection")
    except Exception as error:
        RESULT_PATH.write_text(
            json.dumps(
                {
                    "status": "FAIL",
                    "blocker_count": 1,
                    "explicit_assertions": ASSERTIONS,
                    "error": f"{type(error).__name__}: {error}",
                    "network": "socket connect blocked in-process and via subprocess sitecustomize",
                    "run_root": str(RUN_ROOT),
                },
                indent=2,
                sort_keys=True,
            )
            + "\n",
            encoding="utf-8",
        )
        raise
    finally:
        socket.socket.connect = original_connect
    RESULT_PATH.write_text(
        json.dumps(
            {
                "status": "PASS",
                "blocker_count": 0,
                "explicit_assertions": ASSERTIONS,
                "scenarios": [
                    "first-claim CLI no-draft opaque token",
                    "owner/group/bot invalid non-consumption",
                    "concurrent private single winner and restart durability",
                    "disabled first-claim registration and targeted invite preservation",
                    "r62 missing-first_claim targeted migration",
                    "forged owner-review rejection and authenticated callback preflight-only gate",
                    "direct profile activation CLI absent",
                ],
                "network": "socket connect blocked in-process and via subprocess sitecustomize",
                "run_root": str(RUN_ROOT),
            },
            indent=2,
            sort_keys=True,
        )
        + "\n",
        encoding="utf-8",
    )
    return 0


if __name__ == "__main__":
    raise SystemExit(main())