#!/usr/bin/env python3
"""Append-only completion of the task26 reset baseline; never restores customer data."""

from __future__ import annotations
import argparse
import fcntl
import hashlib
import json
import os
import stat
import subprocess
from datetime import datetime, timezone
from pathlib import Path
from typing import Any

O_FLAGS = getattr(os, "O_NOFOLLOW", 0) | getattr(os, "O_CLOEXEC", 0)


class BaselineError(RuntimeError):
    pass


def canonical(v: Any) -> bytes:
    return (
        json.dumps(v, ensure_ascii=False, sort_keys=True, separators=(",", ":")) + "\n"
    ).encode()


def sha(raw: bytes) -> str:
    return hashlib.sha256(raw).hexdigest()


def stable(s: os.stat_result) -> tuple[int, ...]:
    return (
        s.st_dev,
        s.st_ino,
        s.st_mode,
        s.st_uid,
        s.st_gid,
        s.st_nlink,
        s.st_size,
        s.st_mtime_ns,
        s.st_ctime_ns,
    )


def read_private(p: Path) -> tuple[bytes, os.stat_result]:
    fd = os.open(p, os.O_RDONLY | O_FLAGS)
    try:
        a = os.fstat(fd)
        if (
            not stat.S_ISREG(a.st_mode)
            or a.st_uid != os.getuid()
            or a.st_nlink != 1
            or stat.S_IMODE(a.st_mode) != 0o600
        ):
            raise BaselineError(f"unsafe private file: {p}")
        chunks = []
        while chunk := os.read(fd, 1024 * 1024):
            chunks.append(chunk)
        b = os.fstat(fd)
        if stable(a) != stable(b):
            raise BaselineError(f"file changed during read: {p}")
        return b"".join(chunks), a
    finally:
        os.close(fd)


def read_stable(p: Path) -> bytes:
    fd = os.open(p, os.O_RDONLY | O_FLAGS)
    try:
        a = os.fstat(fd)
        if not stat.S_ISREG(a.st_mode):
            raise BaselineError(f"not a regular file: {p}")
        chunks = []
        while chunk := os.read(fd, 1024 * 1024):
            chunks.append(chunk)
        if stable(a) != stable(os.fstat(fd)):
            raise BaselineError(f"file changed during read: {p}")
        return b"".join(chunks)
    finally:
        os.close(fd)


def load_private(p: Path) -> dict[str, Any]:
    try:
        v = json.loads(read_private(p)[0])
    except (json.JSONDecodeError, UnicodeDecodeError) as e:
        raise BaselineError(f"invalid JSON: {p}") from e
    if not isinstance(v, dict):
        raise BaselineError(f"invalid JSON root: {p}")
    return v


def tree_digest(path: Path) -> str:
    rows = []
    for root, dirs, files in os.walk(path, topdown=True, followlinks=False):
        dirs.sort()
        files.sort()
        base = Path(root)
        for name in dirs + files:
            p = base / name
            rel = p.relative_to(path).as_posix()
            s = p.lstat()
            if stat.S_ISLNK(s.st_mode):
                rows.append(b"L\0" + rel.encode() + b"\0" + os.readlink(p).encode())
            elif stat.S_ISDIR(s.st_mode):
                rows.append(
                    b"D\0"
                    + rel.encode()
                    + b"\0"
                    + format(stat.S_IMODE(s.st_mode), "o").encode()
                )
            elif stat.S_ISREG(s.st_mode):
                rows.append(
                    b"F\0" + rel.encode() + b"\0" + sha(read_stable(p)).encode()
                )
            else:
                raise BaselineError(f"unsupported entry: {p}")
    return sha(b"\n".join(rows))


def absent(p: Path) -> None:
    try:
        s = p.lstat()
    except FileNotFoundError:
        return
    raise BaselineError(
        f"customer authority already exists: {p} ({stat.filemode(s.st_mode)})"
    )


def runtime_state(profile: Path) -> dict[str, Any]:
    unit = "hermes-gateway-dualcoachtest.service"
    out = subprocess.run(
        [
            "systemctl",
            "--user",
            "show",
            unit,
            "--property=ActiveState,SubState,MainPID",
            "--value",
        ],
        check=True,
        capture_output=True,
        text=True,
    ).stdout.splitlines()
    if len(out) != 3:
        raise BaselineError("unknown service state")
    # systemd emits MainPID first despite property order.
    pid, active, sub = out
    procs = []
    needle = ("HERMES_HOME=" + str(profile)).encode()
    for p in Path("/proc").iterdir():
        if not p.name.isdecimal():
            continue
        try:
            env = (p / "environ").read_bytes().split(b"\0")
        except (OSError, PermissionError):
            continue
        if needle in env:
            procs.append(int(p.name))
    return {
        "active": active,
        "sub": sub,
        "main_pid": int(pid),
        "profile_processes": sorted(procs),
    }


def validate_archive(archive: Path, c: dict[str, Any]) -> None:
    if (
        archive.name != c["archive_run_id"]
        or archive.is_symlink()
        or not archive.is_dir()
    ):
        raise BaselineError("reset archive path mismatch")
    mraw, _ = read_private(archive / "manifest.json")
    rraw, _ = read_private(archive / "receipt.json")
    if (
        sha(mraw) != c["archive_manifest_sha256"]
        or sha(rraw) != c["archive_receipt_sha256"]
    ):
        raise BaselineError("reset archive pin mismatch")
    m = json.loads(mraw)
    r = json.loads(rraw)
    if (
        m.get("evidence_digest") != c["archive_evidence_digest"]
        or r.get("evidence_digest") != c["archive_evidence_digest"]
        or r.get("manifest_sha256") != c["archive_manifest_sha256"]
        or r.get("status") != "PASS"
        or r.get("post_reset_empty_baseline") is not True
    ):
        raise BaselineError("reset archive is not committed PASS")
    if tree_digest(archive) != c["archive_tree_sha256"]:
        raise BaselineError("reset archive tree drift")
    entry = next(
        (x for x in m.get("entries", []) if x.get("path") == "customers/registry.json"),
        None,
    )
    if entry != {
        "mode": "0600",
        "path": "customers/registry.json",
        "sha256": c["canonical_registry_sha256"],
        "size": len(canonical(c["canonical_registry"])),
    }:
        raise BaselineError("archived registry provenance mismatch")


def validate_invite(profile: Path, c: dict[str, Any]) -> tuple[str, os.stat_result]:
    ledger = profile / "data/onboarding/telegram-customer-bootstrap-v1/ledger.json"
    raw, s = read_private(ledger)
    v = json.loads(raw)
    if (
        set(v) != {"schema", "sessions", "digest"}
        or v.get("schema") != "telegram-customer-bootstrap-v1"
        or not isinstance(v.get("sessions"), list)
        or len(v["sessions"]) != 1
    ):
        raise BaselineError("unknown bootstrap ledger authority")
    if v["digest"] != sha(
        canonical({"schema": v["schema"], "sessions": v["sessions"]})[:-1]
    ):
        raise BaselineError("bootstrap ledger digest mismatch")
    x = v["sessions"][0]
    required = {
        "bot_username",
        "consent_card_message_id",
        "consent_publication_attempt",
        "created_at",
        "customer_draft",
        "customer_draft_digest",
        "expires_at",
        "failure_code",
        "generation",
        "owner_id",
        "recovery_attempt_generation",
        "recovery_attempts",
        "role_claims",
        "session_id",
        "sid_hash",
        "state",
        "updated_at",
    }
    if set(x) != required:
        raise BaselineError("unknown invite session schema")
    if x["customer_draft_digest"] != sha(canonical(x["customer_draft"])[:-1]):
        raise BaselineError("invite customer draft digest mismatch")
    try:
        expires_at = datetime.fromisoformat(x["expires_at"])
    except (TypeError, ValueError) as e:
        raise BaselineError("invite expiry is invalid") from e
    if expires_at <= datetime.now(timezone.utc):
        raise BaselineError("prepared invite is expired")
    if (
        x["session_id"] != c["session_id"]
        or x["sid_hash"] != c["sid_hash"]
        or x["state"] != "PREPARED"
        or x["role_claims"] != []
        or x["recovery_attempts"] != []
        or x["consent_card_message_id"] is not None
        or x["customer_draft"].get("customer_user_id") is not None
    ):
        raise BaselineError("invite is not the exact unclaimed PREPARED authority")
    if sha(raw) != c["invite_ledger_sha256"]:
        raise BaselineError("prepared invite bytes drifted")
    lock = profile / "data/onboarding/telegram-customer-bootstrap-v1/ledger.lock"
    lraw, _ = read_private(lock)
    if lraw:
        raise BaselineError("bootstrap lock authority is nonempty")
    children = {p.name for p in (profile / "data/onboarding").iterdir()}
    if children != {"telegram-customer-bootstrap-v1"}:
        raise BaselineError("unknown or pending onboarding authority")
    return sha(raw), s


def validate_empty_siblings(profile: Path) -> None:
    for rel in (
        "customers",
        "registry.json",
        "data/customers",
        "data/owner-actions",
        "data/activation-completion-notices.jsonl",
        "data/customer-activation-audit.jsonl",
        "data/customer-activation-journal.json",
        "data/customer-activation-receipt.json",
        "data/nutrition-onboarding-projection-journal.jsonl",
        "data/scheduled-deliveries.jsonl",
        "data/scheduled-deliveries-fence.json",
        "data/customer-schedule-claims",
        "data/recovery-audits",
        "data/activation-readiness",
    ):
        absent(profile / rel)
    for d in (profile / "sessions", profile / "cron/output"):
        if d.exists() and any(d.iterdir()):
            raise BaselineError(f"nonempty sibling authority: {d}")


def preflight(
    profile: Path,
    other: Path,
    archive: Path,
    c: dict[str, Any],
    runtime: dict[str, Any] | None = None,
) -> dict[str, Any]:
    for p in (profile, other):
        s = p.lstat()
        if p.is_symlink() or not stat.S_ISDIR(s.st_mode) or s.st_uid != os.getuid():
            raise BaselineError(f"unsafe profile root: {p}")
    validate_archive(archive, c)
    validate_empty_siblings(profile)
    invite_hash, invite_stat = validate_invite(profile, c)
    r = runtime if runtime is not None else runtime_state(profile)
    if r != {
        "active": "inactive",
        "sub": "dead",
        "main_pid": 0,
        "profile_processes": [],
    }:
        raise BaselineError("service/process authority is active or unknown")
    other_hash = tree_digest(other)
    if other_hash != c["other_profile_sha256"]:
        raise BaselineError("other profile drift")
    return {
        "invite_sha256": invite_hash,
        "invite_stable": stable(invite_stat),
        "other_sha256": other_hash,
    }


def publish(profile: Path, raw: bytes) -> None:
    rootfd = os.open(profile, os.O_RDONLY | os.O_DIRECTORY | O_FLAGS)
    tmp = f".customers-baseline-{os.getpid()}"
    try:
        os.mkdir(tmp, 0o700, dir_fd=rootfd)
        dfd = os.open(tmp, os.O_RDONLY | os.O_DIRECTORY | O_FLAGS, dir_fd=rootfd)
        try:
            fd = os.open(
                "registry.json",
                os.O_WRONLY | os.O_CREAT | os.O_EXCL | O_FLAGS,
                0o600,
                dir_fd=dfd,
            )
            try:
                n = 0
                while n < len(raw):
                    n += os.write(fd, raw[n:])
                os.fsync(fd)
            finally:
                os.close(fd)
            os.fsync(dfd)
        finally:
            os.close(dfd)
        os.rename(tmp, "customers", src_dir_fd=rootfd, dst_dir_fd=rootfd)
        os.fsync(rootfd)
    except Exception:
        try:
            os.unlink(f"{tmp}/registry.json", dir_fd=rootfd)
        except OSError:
            pass
        try:
            os.rmdir(tmp, dir_fd=rootfd)
        except OSError:
            pass
        raise
    finally:
        os.close(rootfd)


def verify_baseline(
    profile: Path,
    other: Path,
    archive: Path,
    c: dict[str, Any],
    runtime: dict[str, Any] | None = None,
) -> dict[str, Any]:
    # Verify all preconditions except expected absence, then exact final registry.
    reg = profile / "customers/registry.json"
    raw, s = read_private(reg)
    if raw != canonical(c["canonical_registry"]):
        raise BaselineError("registry is not exact canonical empty baseline")
    if (
        not stat.S_ISDIR(reg.parent.lstat().st_mode)
        or stat.S_IMODE(reg.parent.lstat().st_mode) != 0o700
        or reg.parent.lstat().st_uid != os.getuid()
    ):
        raise BaselineError("customers directory unsafe")
    # Temporarily validate sibling list without treating customers as pending.
    validate_archive(archive, c)
    invite_hash, _ = validate_invite(profile, c)
    for rel in (
        "registry.json",
        "data/customers",
        "data/owner-actions",
        "data/activation-completion-notices.jsonl",
        "data/customer-activation-audit.jsonl",
        "data/customer-activation-journal.json",
        "data/customer-activation-receipt.json",
        "data/nutrition-onboarding-projection-journal.jsonl",
        "data/scheduled-deliveries.jsonl",
        "data/scheduled-deliveries-fence.json",
        "data/customer-schedule-claims",
        "data/recovery-audits",
        "data/activation-readiness",
    ):
        absent(profile / rel)
    r = runtime if runtime is not None else runtime_state(profile)
    if r != {
        "active": "inactive",
        "sub": "dead",
        "main_pid": 0,
        "profile_processes": [],
    }:
        raise BaselineError("service/process authority is active or unknown")
    if tree_digest(other) != c["other_profile_sha256"]:
        raise BaselineError("other profile drift")
    return {
        "registry_sha256": sha(raw),
        "customers": 0,
        "invite_sha256": invite_hash,
        "other_profile_unchanged": True,
    }


def main() -> int:
    ap = argparse.ArgumentParser()
    ap.add_argument("mode", choices=("verify", "dry-run", "complete-baseline"))
    ap.add_argument("--profile", type=Path, required=True)
    ap.add_argument("--other-profile", type=Path, required=True)
    ap.add_argument("--archive", type=Path, required=True)
    ap.add_argument("--contract", type=Path, required=True)
    ap.add_argument("--permission", type=Path, required=True)
    a = ap.parse_args()
    c = load_private(a.contract)
    p = load_private(a.permission)
    if p != {
        "schema": "task26-reset-baseline-completion-permission-v1",
        "approval": "TASK26_ARCHIVE_FIRST_PROFILE_RESET_APPROVED",
        "candidate_digest": c["candidate_digest"],
        "controller_sha256": sha(Path(__file__).read_bytes()),
        "contract_sha256": sha(a.contract.read_bytes()),
        "rollback_record_sha256": c["rollback_record_sha256"],
        "allowed_live_profile_mutation": "atomic_create_customers_directory_with_canonical_empty_registry_only",
        "execute_allowed": True,
    }:
        raise BaselineError("permission seal mismatch")
    if a.mode == "verify":
        out = verify_baseline(a.profile, a.other_profile, a.archive, c)
    else:
        lockfd = os.open(a.profile / "gateway.lock", os.O_RDONLY | O_FLAGS)
        invite_lockfd = os.open(
            a.profile / "data/onboarding/telegram-customer-bootstrap-v1/ledger.lock",
            os.O_RDONLY | O_FLAGS,
        )
        try:
            try:
                fcntl.flock(lockfd, fcntl.LOCK_EX | fcntl.LOCK_NB)
                fcntl.flock(invite_lockfd, fcntl.LOCK_EX | fcntl.LOCK_NB)
            except BlockingIOError as e:
                raise BaselineError("gateway or invite lock is held") from e
            before = preflight(a.profile, a.other_profile, a.archive, c)
            if a.mode == "complete-baseline":
                if (
                    sha(Path(c["rollback_record_path"]).read_bytes())
                    != c["rollback_record_sha256"]
                ):
                    raise BaselineError("rollback record drift")
                publish(a.profile, canonical(c["canonical_registry"]))
                out = verify_baseline(a.profile, a.other_profile, a.archive, c)
                if out["invite_sha256"] != before["invite_sha256"]:
                    raise BaselineError("invite changed across publication")
            else:
                out = {
                    **before,
                    "planned_registry_sha256": c["canonical_registry_sha256"],
                    "mutations": 0,
                }
        finally:
            os.close(invite_lockfd)
            os.close(lockfd)
    print(
        json.dumps(
            {
                "schema": "task26-reset-baseline-completion-receipt-v1",
                "mode": a.mode,
                "status": "PASS",
                **out,
            },
            sort_keys=True,
            separators=(",", ":"),
        )
    )
    return 0


if __name__ == "__main__":
    try:
        raise SystemExit(main())
    except BaselineError as e:
        print(f"FAIL: {e}", file=__import__("sys").stderr)
        raise SystemExit(2)