from __future__ import annotations

import hashlib
import json
import os
import stat
import subprocess
from datetime import datetime
from pathlib import Path

import pytest


ROOT = Path(__file__).resolve().parent
CONTROLLER = ROOT / "first_customer_invite_controller.py"
CONTRACT = ROOT / "first_customer_invite_contract.py"
OPERATIONS = ROOT / "first_customer_invite_operations.py"
PERMISSION_MODULE = ROOT / "first_customer_invite_permission.py"
PYTHON = Path("/home/cube/projects/richard/hermes-agent/.venv/bin/python")
SOURCE = Path("/home/cube/projects/richard/hermes-agent-dualcoach-v1.1")
MODULE = SOURCE / "gateway/platforms/telegram_customer_bootstrap.py"
APPROVAL = ROOT / "approval-event.json"
AUTHORIZATION = ROOT / "launch-authorization.json"


def _write_json(path: Path, value: dict[str, object], mode: int = 0o600) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(
        json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":")),
        encoding="utf-8",
    )
    path.chmod(mode)


def _fixture(
    tmp_path: Path,
) -> tuple[Path, Path, Path, dict[str, str], dict[str, Path]]:
    profile = tmp_path / "profile"
    data = profile / "data"
    customers = profile / "customers"
    data_customers = data / "customers"
    for path in (profile, data, customers, data_customers):
        path.mkdir(mode=0o700)
    _write_json(
        customers / "registry.json",
        {
            "version": 1,
            "registry_mode": "ordinary_v1",
            "owner": {"user_id": "8693203710"},
            "customers": [
                {
                    "customer_key": "historical-synthetic",
                    "enabled": False,
                    "ai_processing_consent": {"granted": False},
                }
            ],
        },
    )
    draft = tmp_path / "draft.json"
    _write_json(
        draft,
        {
            "customer_key": "pilot_20260820_01",
            "display_name": "DualCoach Pilot Customer",
            "starts_on": "2026-08-21",
            "daily_time": "08:00",
            "weekly_weekday": 0,
            "monthly_day": 1,
            "calories_kcal": 2000,
            "protein_g": 150,
            "meals": ["breakfast", "lunch", "dinner"],
            "customer_user_id": None,
        },
    )
    unit = tmp_path / "gateway.service"
    unit.write_text(
        f"ExecStart={PYTHON} -m hermes_cli.main --profile dualcoachtest gateway run\n",
        encoding="utf-8",
    )
    unit.chmod(0o600)
    fake_bin = tmp_path / "bin"
    fake_bin.mkdir()
    systemctl = fake_bin / "systemctl"
    systemctl.write_text(
        "#!/bin/sh\nprintf 'MainPID=0\\nActiveState=inactive\\nSubState=dead\\n'\n",
        encoding="utf-8",
    )
    systemctl.chmod(0o700)
    permission = tmp_path / "permission.json"
    outputs = {
        "preflight": tmp_path / "preflight.json",
        "prepared": tmp_path / "prepared.json",
        "handoff": tmp_path / "handoff.private.json",
        "verified": tmp_path / "verified.json",
        "install": tmp_path / "install.json",
        "provider": tmp_path / "provider.json",
        "start": tmp_path / "start.json",
    }
    _write_json(
        outputs["install"],
        {
            "schema": "dualcoach-first-customer-install-receipt-v1",
            "status": "PASS_INSTALLED",
            "candidate_digest": (
                "0e383539aea1b83205771772f1e8b417840defe60f5ea6ce184f80e3af8d25f9"
            ),
            "bootstrap_module_sha256": hashlib.sha256(MODULE.read_bytes()).hexdigest(),
            "invite_ttl_seconds": 86_400,
            "venv_mode": "0700",
            "link_mode": "copy",
            "installer": "pip",
            "installed_regular_file_count": 1,
            "unit_sha256": hashlib.sha256(unit.read_bytes()).hexdigest(),
        },
    )
    _write_json(
        outputs["provider"],
        {
            "schema": "dualcoach-first-customer-provider-receipt-v1",
            "status": "PASS_PROVIDER_READY",
            "candidate_digest": (
                "0e383539aea1b83205771772f1e8b417840defe60f5ea6ce184f80e3af8d25f9"
            ),
            "install_receipt_sha256": hashlib.sha256(
                outputs["install"].read_bytes()
            ).hexdigest(),
            "unit_sha256": hashlib.sha256(unit.read_bytes()).hexdigest(),
        },
    )
    _write_json(
        outputs["start"],
        {
            "schema": "dualcoach-first-customer-start-receipt-v1",
            "status": "PASS_STARTED",
            "candidate_digest": (
                "0e383539aea1b83205771772f1e8b417840defe60f5ea6ce184f80e3af8d25f9"
            ),
            "main_pid": 123,
        },
    )
    _write_json(
        permission,
        {
            "schema": "dualcoach-first-customer-launch-permission-v1",
            "candidate_digest": "0e383539aea1b83205771772f1e8b417840defe60f5ea6ce184f80e3af8d25f9",
            "hermes_wheel_sha256": "ec160d3d0e29bc747f463923d31bfe736aefb5840a3d32fcb04842aa5139ddb0",
            "profile_wheel_sha256": "a56da2417df0912f3fe407c0b78befd7362207d1c8a35271000af46701ba79d2",
            "tagged_seal_sha256": "8f26a2929dd349967ecc39df3b750fdd84a2ded1dfd09ffe4ea27263746a7604",
            "profile": str(profile),
            "service": "hermes-gateway-dualcoachtest.service",
            "expected_python": str(PYTHON),
            "unit_path": str(unit),
            "unit_sha256": hashlib.sha256(unit.read_bytes()).hexdigest(),
            "draft_sha256": hashlib.sha256(draft.read_bytes()).hexdigest(),
            "module_sha256": hashlib.sha256(MODULE.read_bytes()).hexdigest(),
            "controller_sha256": hashlib.sha256(CONTROLLER.read_bytes()).hexdigest()
            if CONTROLLER.exists()
            else "0" * 64,
            "contract_sha256": hashlib.sha256(CONTRACT.read_bytes()).hexdigest(),
            "operations_sha256": hashlib.sha256(OPERATIONS.read_bytes()).hexdigest(),
            "permission_module_sha256": hashlib.sha256(
                PERMISSION_MODULE.read_bytes()
            ).hexdigest(),
            "approval_event_path": str(APPROVAL),
            "approval_event_sha256": hashlib.sha256(APPROVAL.read_bytes()).hexdigest(),
            "launch_authorization_path": str(AUTHORIZATION),
            "launch_authorization_sha256": hashlib.sha256(
                AUTHORIZATION.read_bytes()
            ).hexdigest(),
            "maximum_invites": 1,
            "preflight_receipt_path": str(outputs["preflight"]),
            "prepare_receipt_path": str(outputs["prepared"]),
            "handoff_path": str(outputs["handoff"]),
            "verify_receipt_path": str(outputs["verified"]),
            "install_receipt_path": str(outputs["install"]),
            "provider_receipt_path": str(outputs["provider"]),
            "start_receipt_path": str(outputs["start"]),
            "allowed_modes": ["preflight", "prepare", "verify"],
        },
    )
    env = {
        **os.environ,
        "PATH": f"{fake_bin}:{os.environ['PATH']}",
        "PYTHONPATH": str(SOURCE),
        "PYTHONDONTWRITEBYTECODE": "1",
    }
    return profile, draft, permission, env, outputs


def _run(
    mode: str,
    profile: Path,
    draft: Path,
    permission: Path,
    env: dict[str, str],
    *extra: str,
) -> subprocess.CompletedProcess[str]:
    return subprocess.run(
        [
            str(PYTHON),
            "-B",
            str(CONTROLLER),
            mode,
            "--profile",
            str(profile),
            "--draft",
            str(draft),
            "--permission",
            str(permission),
            *extra,
        ],
        check=False,
        capture_output=True,
        text=True,
        env=env,
    )


def test_preflight_is_read_only_on_clean_historical_baseline(tmp_path: Path) -> None:
    profile, draft, permission, env, outputs = _fixture(tmp_path)
    before = {
        path.relative_to(profile): hashlib.sha256(path.read_bytes()).hexdigest()
        for path in profile.rglob("*")
        if path.is_file()
    }
    receipt = outputs["preflight"]

    result = _run(
        "preflight",
        profile,
        draft,
        permission,
        env,
        "--receipt",
        str(receipt),
    )

    assert result.returncode == 0, result.stderr
    assert json.loads(receipt.read_text())["status"] == "PASS_READY"
    assert not (profile / "data/onboarding").exists()
    after = {
        path.relative_to(profile): hashlib.sha256(path.read_bytes()).hexdigest()
        for path in profile.rglob("*")
        if path.is_file()
    }
    assert after == before


def test_prepare_then_verify_keeps_one_24_hour_unclaimed_invite(
    tmp_path: Path,
) -> None:
    profile, draft, permission, env, outputs = _fixture(tmp_path)
    handoff = outputs["handoff"]
    prepared = outputs["prepared"]

    result = _run(
        "prepare",
        profile,
        draft,
        permission,
        env,
        "--handoff",
        str(handoff),
        "--receipt",
        str(prepared),
    )

    assert result.returncode == 0, result.stderr
    private = json.loads(handoff.read_text())
    receipt = json.loads(prepared.read_text())
    assert stat.S_IMODE(handoff.stat().st_mode) == 0o600
    assert receipt["status"] == "PASS_PREPARED"
    assert "start_token" not in receipt
    assert "customer_link" not in receipt
    created = datetime.fromisoformat(receipt["created_at"])
    expires = datetime.fromisoformat(receipt["expires_at"])
    assert int((expires - created).total_seconds()) == 86_400
    assert private["customer_link"].startswith(
        "https://t.me/nutricoach_kr_bot?start=rc1_"
    )
    ledger = profile / "data/onboarding/telegram-customer-bootstrap-v1/ledger.json"
    before = hashlib.sha256(ledger.read_bytes()).hexdigest()
    verified = outputs["verified"]

    verify = _run(
        "verify",
        profile,
        draft,
        permission,
        env,
        "--handoff",
        str(handoff),
        "--session-id",
        receipt["session_id"],
        "--sid-hash",
        receipt["sid_hash"],
        "--receipt",
        str(verified),
    )

    assert verify.returncode == 0, verify.stderr
    assert json.loads(verified.read_text())["status"] == "PASS_AVAILABLE"
    assert hashlib.sha256(ledger.read_bytes()).hexdigest() == before


@pytest.mark.parametrize("attack", ["active-customer", "wrong-controller"])
def test_permission_or_baseline_mismatch_fails_closed(
    tmp_path: Path,
    attack: str,
) -> None:
    profile, draft, permission, env, outputs = _fixture(tmp_path)
    if attack == "active-customer":
        registry = profile / "customers/registry.json"
        value = json.loads(registry.read_text())
        value["customers"][0]["enabled"] = True
        _write_json(registry, value)
    else:
        value = json.loads(permission.read_text())
        value["controller_sha256"] = "f" * 64
        _write_json(permission, value)

    result = _run(
        "preflight",
        profile,
        draft,
        permission,
        env,
        "--receipt",
        str(outputs["preflight"]),
    )

    assert result.returncode == 2
    assert not (profile / "data/onboarding").exists()


def test_invite_preflight_rejects_zero_installed_file_counter(
    tmp_path: Path,
) -> None:
    profile, draft, permission, env, outputs = _fixture(tmp_path)
    installed = json.loads(outputs["install"].read_text())
    installed["installed_regular_file_count"] = 0
    _write_json(outputs["install"], installed)

    result = _run(
        "preflight",
        profile,
        draft,
        permission,
        env,
        "--receipt",
        str(outputs["preflight"]),
    )

    assert result.returncode == 2
    assert not (profile / "data/onboarding").exists()


@pytest.mark.parametrize(
    ("field", "value"),
    [
        ("approval_event_sha256", "a" * 64),
        ("launch_authorization_sha256", "b" * 64),
        ("maximum_invites", 2),
    ],
)
def test_real_approval_and_single_invite_limit_are_mandatory(
    tmp_path: Path,
    field: str,
    value: str | int,
) -> None:
    profile, draft, permission, env, outputs = _fixture(tmp_path)
    seal = json.loads(permission.read_text())
    seal[field] = value
    _write_json(permission, seal)

    result = _run(
        "preflight",
        profile,
        draft,
        permission,
        env,
        "--receipt",
        str(outputs["preflight"]),
    )

    assert result.returncode == 2
    assert not (profile / "data/onboarding").exists()


def test_unsealed_output_path_is_rejected_before_invite_creation(
    tmp_path: Path,
) -> None:
    profile, draft, permission, env, outputs = _fixture(tmp_path)

    result = _run(
        "prepare",
        profile,
        draft,
        permission,
        env,
        "--handoff",
        str(tmp_path / "alternate.private.json"),
        "--receipt",
        str(outputs["prepared"]),
    )

    assert result.returncode == 2
    assert not (profile / "data/onboarding").exists()


def test_reserved_output_collision_is_rejected_before_invite_creation(
    tmp_path: Path,
) -> None:
    profile, draft, permission, env, outputs = _fixture(tmp_path)
    outputs["handoff"].write_text("collision", encoding="utf-8")
    outputs["handoff"].chmod(0o600)

    result = _run(
        "prepare",
        profile,
        draft,
        permission,
        env,
        "--handoff",
        str(outputs["handoff"]),
        "--receipt",
        str(outputs["prepared"]),
    )

    assert result.returncode == 2
    assert not (profile / "data/onboarding").exists()