"""Self-service CCR intake: submit, standard-lane review and id checks.

RPF-WP-0061-T02. Lets a client file a CCR and obtain the platform-owner review
for in-profile lanes without a railiance-platform worker session. The rules
live in registry/ccr-standard-profile.yaml; this module only evaluates them.
Nothing here touches OpenBao or grants approval: the CCR stays `proposed`
until the founder decision is synced.
"""
from __future__ import annotations

import argparse
import hashlib
import json
import re
import shutil
import subprocess
import urllib.request
from collections import defaultdict
from datetime import datetime, timezone
from pathlib import Path
from typing import Any

import yaml

REPO_DIR = Path(__file__).resolve().parents[1]
DEFAULT_PROFILE = REPO_DIR / "registry/ccr-standard-profile.yaml"
CCR_ID_RE = re.compile(r"^CCR-(\d{4})-(\d{4})$")
CCR_FILE_RE = re.compile(r"^(CCR-\d{4}-\d{4})(-[a-z0-9-]+)?\.ya?ml$")
SLUG_RE = re.compile(r"^[a-z0-9][a-z0-9-]*$")
TTL_UNITS = {"s": 1, "m": 60, "h": 3600, "d": 86400}
PLATFORM_OWNER = "railiance-platform-owner"
PLATFORM_TOPIC_ID = "ca369340-a64e-442e-98f1-a4fa7dc74a38"


def load_profile(path: Path = DEFAULT_PROFILE) -> tuple[dict[str, Any], str]:
    raw = path.read_bytes()
    return yaml.safe_load(raw), "sha256:" + hashlib.sha256(raw).hexdigest()


def ttl_seconds(value: Any) -> int | None:
    if not isinstance(value, str) or not re.match(r"^[1-9][0-9]*[smhd]$", value):
        return None
    return int(value[:-1]) * TTL_UNITS[value[-1]]


def bound_list(auth: dict[str, Any], key: str) -> list[Any]:
    claims = auth.get("bound_claims") or {}
    return list(claims.get(key) or auth.get(f"bound_{key}") or [])


def ccr_files(directory: Path) -> list[Path]:
    return sorted(p for p in directory.glob("CCR-*.y*ml") if CCR_FILE_RE.match(p.name))


def reader_conflicts(
    ccr: dict[str, Any], directory: Path, profile: dict[str, Any], cc: Any
) -> list[str]:
    kv_path = (ccr.get("openbao") or {}).get("kv_path")
    statuses = set(profile["reader_conflict_statuses"])
    conflicts = []
    for path in ccr_files(directory):
        other = cc.load_yaml(path)
        if other.get("id") == ccr.get("id"):
            continue
        if (other.get("openbao") or {}).get("kv_path") == kv_path and other.get("status") in statuses:
            conflicts.append(f"{other.get('id')} ({other.get('status')})")
    return conflicts


def standard_violations(
    ccr: dict[str, Any], profile: dict[str, Any], directory: Path, cc: Any
) -> list[str]:
    """Return every reason the CCR falls outside the standard lane profile."""
    v: list[str] = []
    if ccr.get("request_type") != profile["request_type"]:
        return [f"request_type must be {profile['request_type']}"]
    openbao = ccr.get("openbao") or {}
    auth = openbao.get("auth") or {}
    want = profile["auth"]
    kv_path = openbao.get("kv_path") or ""
    if openbao.get("mount") != profile["mount"]:
        v.append(f"mount must be {profile['mount']}")
    if not any(re.match(p, kv_path) for p in profile["kv_path_patterns"]):
        v.append(f"kv_path {kv_path!r} is outside the standard workload/supplier prefixes")
    if openbao.get("metadata_read", True) is not profile["metadata_read"]:
        v.append("metadata_read must be false")
    if bool(openbao.get("token_capabilities_self", False)) is not profile["token_capabilities_self"]:
        v.append("token_capabilities_self is not standard")
    policy_name = openbao.get("policy_name") or ""
    if not policy_name.startswith(profile["policy_name_prefix"]):
        v.append(f"policy_name must start with {profile['policy_name_prefix']}")
    policy_file = openbao.get("policy_file")
    if not policy_file:
        v.append("policy_file is required")
    else:
        policy_path = REPO_DIR / policy_file
        if not policy_path.exists():
            v.append(f"policy_file {policy_file} does not exist")
        elif cc.normalized_policy_body(policy_path.read_text()) != cc.normalized_policy_body(
            cc.generated_policy_hcl(ccr)
        ):
            v.append("policy_file differs from the generated policy")
    if auth.get("method") != want["method"] or auth.get("mount") != want["mount"]:
        v.append(f"auth must be {want['method']} on mount {want['mount']}")
    if not str(auth.get("role") or "").startswith(want["role_prefix"]):
        v.append(f"auth role must start with {want['role_prefix']}")
    if auth.get("audience") != want["audience"]:
        v.append(f"audience must be {want['audience']}")
    for key in ("ttl", "token_max_ttl", "token_explicit_max_ttl"):
        seconds = ttl_seconds(auth.get(key))
        if seconds is None or seconds > want["max_ttl_seconds"]:
            v.append(f"{key} must be set and at most {want['max_ttl_seconds']}s")
    if auth.get("token_no_default_policy") is not want["token_no_default_policy"]:
        v.append("token_no_default_policy must be true")
    if auth.get("policies") != [policy_name]:
        v.append("auth policies must be exactly [policy_name]")
    accounts = bound_list(auth, "service_account_names")
    namespaces = bound_list(auth, "service_account_namespaces")
    if len(accounts) != want["service_accounts"] or "*" in accounts:
        v.append(f"exactly {want['service_accounts']} exact service account required")
    if len(namespaces) != want["namespaces"] or "*" in namespaces:
        v.append(f"exactly {want['namespaces']} exact namespace required")
    forbidden = sorted(set(namespaces) & set(want["forbidden_namespaces"]))
    if forbidden:
        v.append(f"namespace {forbidden} is reserved for platform components")
    risk = (ccr.get("risk") or {}).get("classification")
    if risk not in profile["allowed_risk_classifications"]:
        v.append(f"risk classification {risk!r} needs a platform review")
    conflicts = reader_conflicts(ccr, directory, profile, cc)
    if conflicts:
        v.append(f"kv_path already has a reader: {', '.join(conflicts)}")
    return v


def duplicate_ids(directory: Path, cc: Any) -> dict[str, list[str]]:
    seen: dict[str, list[str]] = defaultdict(list)
    for path in ccr_files(directory):
        seen[str(cc.load_yaml(path).get("id"))].append(path.name)
        prefix = CCR_FILE_RE.match(path.name).group(1)
        if prefix != cc.load_yaml(path).get("id"):
            seen[f"{prefix} (file name vs id)"].append(path.name)
    return {k: v for k, v in seen.items() if len(v) > 1 or "file name" in k}


def next_ccr_id(directory: Path, year: int) -> str:
    numbers = [0]
    for path in ccr_files(directory):
        match = CCR_ID_RE.match(CCR_FILE_RE.match(path.name).group(1))
        if match and int(match.group(1)) == year:
            numbers.append(int(match.group(2)))
    return f"CCR-{year}-{max(numbers) + 1:04d}"


def record_review(path: Path, ccr: dict[str, Any], violations: list[str], profile: dict[str, Any], digest: str, cc: Any) -> None:
    label = f"{profile['profile']} v{profile['version']} {digest}"
    if violations:
        decision = "needs_platform_review"
        comment = f"Outside {label}: " + "; ".join(violations)
    else:
        decision = "approved"
        comment = (f"Standard lane: passes {label}. Platform-owner source review only; "
                   "founder approval and confirm-binding still required.")
    comments = ccr.setdefault("review", {}).setdefault("comments", [])
    comments.append({
        "at": cc.utc_now(),
        "reviewer": f"{PLATFORM_OWNER} (policy-as-code {profile['profile']} v{profile['version']})",
        "decision": decision,
        "comment": comment,
    })
    ccr["updated"] = datetime.now(timezone.utc).date().isoformat()
    cc.dump_yaml(path, ccr)


def notify_platform(ccr: dict[str, Any], violations: list[str], base_url: str) -> None:
    payload = {
        "from_agent": (ccr.get("requester") or {}).get("agent", "ccr-intake"),
        "to_agent": "railiance-platform",
        "subject": f"{ccr['id']} needs platform review (outside standard lane profile)",
        "body": f"{ccr['id']} {ccr.get('title')}: " + "; ".join(violations),
    }
    request = urllib.request.Request(
        f"{base_url.rstrip('/')}/messages/", data=json.dumps(payload).encode(),
        headers={"Content-Type": "application/json"}, method="POST")
    with urllib.request.urlopen(request, timeout=15):
        pass


def review_standard(path: Path, cc: Any, profile_path: Path = DEFAULT_PROFILE,
                    write: bool = True) -> list[str]:
    ccr, _warnings = cc.validate_or_exit(path)
    if ccr.get("status") != "proposed":
        cc.fail(f"review-standard requires status proposed, got {ccr.get('status')}")
    profile, digest = load_profile(profile_path)
    violations = standard_violations(ccr, profile, path.parent, cc)
    if write:
        record_review(path, ccr, violations, profile, digest, cc)
    return violations


def command_review_standard(args: argparse.Namespace, cc: Any) -> int:
    path = cc.resolve_ccr(args.ref)
    violations = review_standard(path, cc, Path(args.profile), write=not args.check)
    if not violations:
        print(f"[OK] {path.name}: standard lane; platform-owner review recorded"
              if not args.check else f"[OK] {path.name}: standard lane")
        return 0
    for item in violations:
        print(f"[NEEDS-PLATFORM] {path.name}: {item}")
    if args.notify and not args.check:
        notify_platform(cc.load_yaml(path), violations, args.state_hub_url)
        print("[OK] railiance-platform notified")
    return 2


def command_check_ids(args: argparse.Namespace, cc: Any) -> int:
    duplicates = duplicate_ids(cc.ccr_dir(), cc)
    for key, names in duplicates.items():
        print(f"[ERROR] {key}: {names}")
    if not duplicates:
        print("[OK] CCR ids unique and matching file names")
    return 1 if duplicates else 0


def tree_digests(root: Path) -> dict[str, str]:
    return {
        str(path.relative_to(root)): hashlib.sha256(path.read_bytes()).hexdigest()
        for path in root.rglob("*")
        if path.is_file() and ".git" not in path.relative_to(root).parts
        and "__pycache__" not in path.parts
    }


def branch_gate(main_root: Path, stem: str, cc: Any, profile_path: Path = DEFAULT_PROFILE) -> list[str]:
    """CI gate for a ccr/<stem> branch checked out at REPO_DIR (RPF-WP-0061-T03).

    The branch may add exactly one new CCR and its new policy file, nothing
    else; the CCR must be an in-profile submission from a listed submitter with
    a policy-as-code review against the current profile digest.
    """
    errors: list[str] = []
    ccr_rel = f"credential-change-requests/{stem}.yaml"
    path = REPO_DIR / ccr_rel
    if not CCR_FILE_RE.match(path.name) or not path.exists():
        return [f"branch ccr/{stem} must add {ccr_rel}"]
    if (main_root / ccr_rel).exists():
        errors.append(f"{ccr_rel} already exists on main")
    ccr, validation_errors, _warnings = cc.validate_ccr(path)
    errors.extend(f"validate: {item}" for item in validation_errors)
    policy_rel = (ccr.get("openbao") or {}).get("policy_file")
    allowed = {ccr_rel} | ({policy_rel} if policy_rel else set())
    branch, main = tree_digests(REPO_DIR), tree_digests(main_root)
    changed = {k for k in branch.keys() | main.keys() if branch.get(k) != main.get(k)}
    for extra in sorted(changed - allowed):
        errors.append(f"ccr branch may not change {extra}")
    if policy_rel and policy_rel in main and main[policy_rel] != branch.get(policy_rel):
        errors.append(f"{policy_rel} exists on main; a submission may not modify it")
    if ccr.get("status") != "proposed":
        errors.append(f"status must be proposed, got {ccr.get('status')}")
    profile, digest = load_profile(profile_path)
    if (ccr.get("requester") or {}).get("agent") not in profile["submitters"]:
        errors.append(f"requester is not a listed submitter {profile['submitters']}")
    errors.extend(standard_violations(ccr, profile, path.parent, cc))
    comments = (ccr.get("review") or {}).get("comments") or []
    if len(comments) != 1 or comments[0].get("decision") != "approved" or digest not in str(
        comments[0].get("comment")
    ):
        errors.append("review must hold exactly the policy-as-code approval for the current profile")
    errors.extend(f"duplicate id: {k} {v}" for k, v in duplicate_ids(path.parent, cc).items())
    return errors


def command_check_branch(args: argparse.Namespace, cc: Any) -> int:
    stem = args.branch.removeprefix("refs/heads/").removeprefix("ccr/")
    errors = branch_gate(Path(args.main_root).resolve(), stem, cc)
    for item in errors:
        print(f"[FAIL] ccr/{stem}: {item}")
    if not errors:
        print(f"[OK] ccr/{stem}: one in-profile CCR, nothing else changed")
    return 1 if errors else 0


def decision_payload(ccr: dict[str, Any], cc: Any) -> dict[str, Any]:
    openbao = ccr["openbao"]
    auth = openbao["auth"]
    review = ccr["review"]["comments"][-1]
    description = "\n".join([
        f"Founder decision for {ccr['id']}: {ccr.get('title')}.",
        "",
        "Non-secret request summary:",
        f"- Requester: {(ccr.get('requester') or {}).get('agent')}",
        f"- OpenBao path: {openbao.get('kv_path')} fields {openbao.get('fields')}",
        f"- Policy: {openbao.get('policy_name')} ({openbao.get('policy_file')})",
        f"- Auth: {auth.get('method')} role {auth.get('role')}, SA {bound_list(auth, 'service_account_names')}"
        f" in {bound_list(auth, 'service_account_namespaces')}, ttl {auth.get('ttl')}",
        f"- Risk: {(ccr.get('risk') or {}).get('classification')}",
        f"- Platform review: {review.get('reviewer')}: {review.get('comment')}",
        "",
        cc.render_decision_templates(ccr),
        "",
        "Approval permits apply after confirm-binding; it exposes and stores no value.",
    ])
    return {
        "topic_id": PLATFORM_TOPIC_ID,
        "title": f"Review {ccr['id']} {ccr.get('title')}",
        "description": description,
        "escalation_note": "Founder decision required before OpenBao credential lane apply.",
    }


def request_decision(path: Path, cc: Any, base_url: str) -> dict[str, Any]:
    """Open the founder's State Hub decision for a platform-reviewed CCR (RPF-WP-0061-T04)."""
    ccr, _warnings = cc.validate_or_exit(path)
    if ccr.get("status") != "proposed":
        cc.fail(f"request-decision requires status proposed, got {ccr.get('status')}")
    if (ccr.get("state_hub") or {}).get("decision_id"):
        cc.fail(f"{ccr['id']} already has State Hub decision {ccr['state_hub']['decision_id']}")
    comments = (ccr.get("review") or {}).get("comments") or []
    last = comments[-1] if comments else {}
    if last.get("decision") != "approved" or not str(last.get("reviewer", "")).startswith(PLATFORM_OWNER):
        cc.fail("request-decision requires a platform-owner approval as the latest review comment")
    decision = cc.state_hub_post_json(base_url, "/decisions/", decision_payload(ccr, cc))
    if not decision.get("id") or decision.get("status") not in (None, "open"):
        cc.fail(f"State Hub returned an unexpected decision: {decision.get('id')} {decision.get('status')}")
    state_hub = ccr.setdefault("state_hub", {})
    state_hub["decision_id"] = decision["id"]
    state_hub["decision_api_url"] = f"{base_url.rstrip('/')}/decisions/{decision['id']}"
    ccr["updated"] = datetime.now(timezone.utc).date().isoformat()
    cc.dump_yaml(path, ccr)
    return decision


def command_request_decision(args: argparse.Namespace, cc: Any) -> int:
    path = cc.resolve_ccr(args.ref)
    decision = request_decision(path, cc, args.state_hub_url)
    print(f"[OK] {path.name}: State Hub decision {decision['id']} opened; commit the CCR file")
    return 0


def git(*argv: str) -> None:
    subprocess.run(["git", "-C", str(REPO_DIR), *argv], check=True)


def submit(args: argparse.Namespace, cc: Any) -> tuple[Path, list[str]]:
    profile, _digest = load_profile(Path(args.profile))
    if args.requester not in profile["submitters"]:
        cc.fail(f"requester {args.requester!r} may not self-submit; allowed: {profile['submitters']}")
    if not SLUG_RE.match(args.slug):
        cc.fail("slug must be lowercase letters, digits and dashes")
    source = cc.resolve_repo_path(args.ccr)
    errors: list[str] = []
    cc.scan_for_secrets(source, errors)
    if errors:
        cc.fail("; ".join(errors))
    ccr = cc.load_yaml(source)
    directory = cc.ccr_dir()
    today = datetime.now(timezone.utc).date()
    ccr_id = next_ccr_id(directory, today.year)
    ccr.update(id=ccr_id, status="proposed", created=today.isoformat(), updated=today.isoformat())
    ccr.setdefault("requester", {})["agent"] = args.requester
    ccr.setdefault("review", {})["comments"] = []
    openbao = ccr.get("openbao") or {}
    if args.policy:
        policy_name = openbao.get("policy_name") or cc.fail("openbao.policy_name is required")
        target = REPO_DIR / "openbao/policies" / f"{policy_name}.hcl"
        policy_text = cc.resolve_repo_path(args.policy).read_text()
        cc.reject_secret_text(policy_text, "policy file")
        if target.exists() and target.read_text() != policy_text:
            cc.fail(f"{target.relative_to(REPO_DIR)} exists with different content")
        target.write_text(policy_text)
        openbao["policy_file"] = str(target.relative_to(REPO_DIR))
    destination = directory / f"{ccr_id}-{args.slug}.yaml"
    cc.dump_yaml(destination, ccr)
    violations = review_standard(destination, cc, Path(args.profile))
    return destination, violations


def command_submit(args: argparse.Namespace, cc: Any) -> int:
    destination, violations = submit(args, cc)
    ccr = cc.load_yaml(destination)
    print(f"[OK] filed {ccr['id']} -> {destination.relative_to(REPO_DIR)}")
    for item in violations:
        print(f"[NEEDS-PLATFORM] {item}")
    if args.commit or args.push:
        branch = f"ccr/{destination.stem}"
        git("switch", "-c", branch)
        paths = [str(destination.relative_to(REPO_DIR))]
        if ccr["openbao"].get("policy_file"):
            paths.append(ccr["openbao"]["policy_file"])
        git("add", *paths)
        git("commit", "-m", f"File {ccr['id']}: {ccr.get('title')}", "--", *paths)
        if args.push:
            git("push", "-u", "origin", branch)
    if violations and args.notify:
        notify_platform(ccr, violations, args.state_hub_url)
    return 0


def add_parsers(sub: Any, cc: Any, state_hub_url: str) -> None:
    review = sub.add_parser(
        "review-standard",
        help="Check a proposed CCR against the standard lane profile; record the platform-owner review",
    )
    review.add_argument("ref")
    review.add_argument("--profile", default=str(DEFAULT_PROFILE))
    review.add_argument("--check", action="store_true", help="report only; do not write the review")
    review.add_argument("--notify", action="store_true", help="message railiance-platform when out of profile")
    review.add_argument("--state-hub-url", default=state_hub_url)
    review.set_defaults(func=lambda args: command_review_standard(args, cc))

    check_ids = sub.add_parser("check-ids", help="Fail on duplicate CCR ids or id/file name mismatch")
    check_ids.set_defaults(func=lambda args: command_check_ids(args, cc))

    check_branch = sub.add_parser(
        "check-branch", help="CI gate: a ccr/* branch adds exactly one in-profile CCR and its policy"
    )
    check_branch.add_argument("--branch", required=True, help="ccr/<id>-<slug> or refs/heads/ccr/...")
    check_branch.add_argument("--main-root", required=True, help="tree of main to compare against")
    check_branch.set_defaults(func=lambda args: command_check_branch(args, cc))

    decision_p = sub.add_parser(
        "request-decision", help="Open the founder State Hub decision for a platform-reviewed proposed CCR"
    )
    decision_p.add_argument("ref")
    decision_p.add_argument("--state-hub-url", default=state_hub_url)
    decision_p.set_defaults(func=lambda args: command_request_decision(args, cc))

    submit_p = sub.add_parser(
        "submit",
        help="File a client CCR draft under the next CCR id and run the standard review",
    )
    submit_p.add_argument("--ccr", required=True, help="client CCR draft (non-secret YAML)")
    submit_p.add_argument("--policy", help="client policy HCL; copied to openbao/policies/<policy_name>.hcl")
    submit_p.add_argument("--slug", required=True)
    submit_p.add_argument("--requester", required=True)
    submit_p.add_argument("--profile", default=str(DEFAULT_PROFILE))
    submit_p.add_argument("--commit", action="store_true", help="commit on a new ccr/<id>-<slug> branch")
    submit_p.add_argument("--push", action="store_true", help="commit and push the ccr/* branch")
    submit_p.add_argument("--notify", action="store_true")
    submit_p.add_argument("--state-hub-url", default=state_hub_url)
    submit_p.set_defaults(func=lambda args: command_submit(args, cc))