from __future__ import annotations

import base64
import os
import pickle
import shutil
import subprocess
import sys
from datetime import date
from pathlib import Path
from tempfile import TemporaryDirectory

from checkin_cli.models import EventType
from checkin_cli.store import CanonicalEventTransaction
from checkin_cli.weekly_operations_correlation import (
    CanonicalCheckinCorrelationTransaction,
    CorrelationAction,
    CorrelationRequest,
    CorrelationResult,
    CorrelationScope,
)
from checkin_cli.weekly_operations_customer_authority import (
    CanonicalCheckinCustomerAuthority,
)
from checkin_cli.weekly_operations_customer_authority_identity import (
    CanonicalCustomerNames,
)
from checkin_cli.weekly_operations_store import WeeklyOperationsStore
from tests._weekly_operations_correlation_support import (
    CUSTOMER,
    StoreFixture,
    DAY,
    event,
    registered_source_at,
    store_at,
)

_CHILD_PROBE = """
import sys
from pathlib import Path
from checkin_cli.customer_coaching import load_customer_registry
from checkin_cli.weekly_operations import WeeklyOperationsAuthorityCompromise, WeeklyOperationsConflict
from checkin_cli.weekly_operations_registration_handoff import (
    begin_canonical_authority_registration,
)
from checkin_cli.weekly_operations_authority import open_authority_root
from checkin_cli.weekly_operations_parent import reacquire_parent_authority
import base64, pickle
profile = Path(sys.argv[1])
authority_binding, parent_binding = pickle.loads(base64.b64decode(sys.argv[2]))
parent = reacquire_parent_authority(profile.parent / 'sidecar', parent_binding)
registry = open_authority_root(parent, authority_binding)
runtime = load_customer_registry(profile / 'registry.json', profile).customers[0]
try:
    with begin_canonical_authority_registration(runtime, registry) as transaction:
        authority = transaction.authority
        _ = transaction.binding
        transaction.acknowledge_binding()
except (WeeklyOperationsConflict, WeeklyOperationsAuthorityCompromise):
    raise SystemExit(23)
authority.close()
registry.close()
parent.close()
raise SystemExit(0)
"""

_MISSING_BINDING_PROBE = """
import sys
from pathlib import Path
from checkin_cli.customer_coaching import load_customer_registry
from checkin_cli.weekly_operations_customer_authority_factory import open_canonical_checkin_customer_authority
profile = Path(sys.argv[1])
runtime = load_customer_registry(profile / 'registry.json', profile).customers[0]
try:
    open_canonical_checkin_customer_authority(runtime)
except TypeError:
    raise SystemExit(23)
raise SystemExit(0)
"""


def _probe(
    script: str, profile: Path, fixture: StoreFixture | None = None
) -> str:
    environment = dict(os.environ)
    environment["PYTHONPATH"] = "dualcoach/profile"
    arguments = [sys.executable, "-c", script, str(profile)]
    if fixture is not None:
        payload = pickle.dumps((fixture.authority.binding, fixture.parent.binding))
        arguments.append(base64.b64encode(payload).decode())
    process = subprocess.Popen(
        arguments,
        cwd=Path.cwd(),
        env=environment,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
        text=True,
    )
    try:
        _stdout, stderr = process.communicate(timeout=1)
    except subprocess.TimeoutExpired as error:
        process.kill()
        _ = process.wait(timeout=1)
        raise AssertionError("canonical authority probe blocked") from error
    assert process.returncode == 23, stderr
    return stderr


def _commit(
    store: WeeklyOperationsStore, source: CanonicalCheckinCustomerAuthority
) -> CorrelationResult:
    scope = CorrelationScope(
        source.customer_identity_digest,
        date.fromisoformat(DAY),
        CorrelationAction.CHECKIN,
    )
    return CanonicalCheckinCorrelationTransaction(
        store, CorrelationRequest(scope, source)
    ).commit()



def test_customer_authority_identity_has_fixed_child_names(tmp_path: Path) -> None:
    """Given a root, when identity is built, then canonical child names are fixed."""
    names = CanonicalCustomerNames(tmp_path)
    assert (names.wizard, names.plans, names.events, names.sequence, names.lock) == (
        "wizard", "nutrition-plans", "events.jsonl", "canonical-sequence.jsonl", ".events.lock"
    )

def test_missing_binding_cannot_reauthorize_replaced_root_twice() -> None:
    with TemporaryDirectory(prefix="task4-r4-binding-") as raw:
        root = Path(raw)
        base = event("r4_binding_base_root01")
        fixture = store_at(root / "sidecar")
        canonical = registered_source_at(
            root / "profile", CUSTOMER, base, registry_authority=fixture.authority
        )
        _ = _commit(fixture.store, canonical.source)
        canonical.source.close()
        customer_root = canonical.runtime.data_root
        sealed = customer_root.with_name("sealed-root")
        _ = customer_root.rename(sealed)
        _ = shutil.copytree(sealed, customer_root)
        replacement = CanonicalEventTransaction.for_customer_runtime(canonical.runtime)
        correction = event(
            "r4_binding_rogue_correct1", EventType.CORRECTION, supersedes=base.event_id
        )
        _ = replacement.append_one(correction)

        assert "for_runtime" not in CanonicalCheckinCustomerAuthority.__dict__
        for _attempt in range(2):
            _ = _probe(_MISSING_BINDING_PROBE, root / "profile")
        assert len(fixture.store.read()) == 1
        fixture.close()


def _assert_fifo_rejected(child: str) -> None:
    with TemporaryDirectory(prefix=f"task4-r4-fifo-{child}-") as raw:
        root = Path(raw)
        fixture = store_at(root / "sidecar")
        canonical = registered_source_at(
            root / "profile",
            CUSTOMER,
            event(f"r4_fifo_{child}_root01"),
            registry_authority=fixture.authority,
        )
        paths = {
            "events": canonical.transaction.events_path,
            "sequence": canonical.transaction.sequence_path,
            "lock": canonical.transaction.lock_path,
        }
        target = paths[child]
        canonical.source.close()
        _ = target.rename(target.with_suffix(target.suffix + ".regular"))
        os.mkfifo(target, mode=0o600)
        for _attempt in range(2):
            _ = _probe(_CHILD_PROBE, root / "profile", fixture)
        assert fixture.store.read() == ()
        fixture.close()


def test_events_fifo_is_rejected_twice_without_blocking() -> None:
    _assert_fifo_rejected("events")


def test_sequence_fifo_is_rejected_twice_without_blocking() -> None:
    _assert_fifo_rejected("sequence")


def test_lock_fifo_is_rejected_twice_without_blocking() -> None:
    _assert_fifo_rejected("lock")


def test_events_directory_is_rejected_without_blocking() -> None:
    with TemporaryDirectory(prefix="task4-r4-directory-") as raw:
        root = Path(raw)
        fixture = store_at(root / "sidecar")
        canonical = registered_source_at(
            root / "profile",
            CUSTOMER,
            event("r4_directory_root01"),
            registry_authority=fixture.authority,
        )
        target = canonical.transaction.events_path
        canonical.source.close()
        _ = target.rename(target.with_suffix(".regular"))
        target.mkdir(mode=0o700)

        _ = _probe(_CHILD_PROBE, root / "profile", fixture)
        assert fixture.store.read() == ()
        fixture.close()


def test_sequence_socket_is_rejected_without_blocking() -> None:
    import socket

    with TemporaryDirectory(prefix="s-") as raw:
        root = Path(raw)
        fixture = store_at(root / "sidecar")
        canonical = registered_source_at(
            root / "profile",
            CUSTOMER,
            event("r4_socket_root_001"),
            registry_authority=fixture.authority,
        )
        target = canonical.transaction.sequence_path
        canonical.source.close()
        _ = target.rename(target.with_suffix(".regular"))
        boundary = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
        try:
            boundary.bind(str(target))
            _ = _probe(_CHILD_PROBE, root / "profile", fixture)
        finally:
            boundary.close()
        assert fixture.store.read() == ()
        fixture.close()