#!/usr/bin/env python3
"""CUST-WP-0071-T05 in-cluster weekly runner: read-only collection, join, retain.

Standard library only, so it runs on a pinned python:3.12-slim image with the
scripts mounted from a ConfigMap. It reads the Kubernetes API with the pod's
ServiceAccount (get/list only), queries Prometheus over plain HTTP, reuses the
existing owner code unchanged (railiance-cluster build_observation, custodian
reconcile join, weekly_allocation_report build_report/retain) and writes only
under the data directory. It never resizes, creates tasks or delivers messages.

Week identity is the ISO week of (run time - 6h); the telemetry window always
ends at Monday 06:00 UTC of that week, so consecutive reports are exactly
604800 s apart regardless of DST or scheduler jitter. 06:00 UTC is at or before
the Monday 08:00 Europe/Berlin schedule in both CET and CEST and keeps the
window start as close as possible to the seven-day Prometheus retention edge. A week whose report is already
retained is a no-op success (idempotent retry); a failed attempt leaves no
report, so the Job's retry or the next manual run collects again.

Delivery (founder decision 2026-10-21, "Hub message + repo tasks"): after a
week is retained, one State Hub message goes from custodian-allocation-review
to the-custodian with a short summary, report path and sha256. The message id
is recorded in receipts.jsonl; a week that already has a message id is never
posted again, and a retained-but-undelivered week is posted by the next run.
"""
from __future__ import annotations

import argparse
from datetime import datetime, timedelta, timezone
import hashlib
import importlib.util
import json
import os
from pathlib import Path
import re
import ssl
import sys
import urllib.error
import urllib.parse
import urllib.request

HERE = Path(__file__).resolve().parent
SA_DIR = Path("/var/run/secrets/kubernetes.io/serviceaccount")
WEEK = re.compile(r"^(\d{4})-W(\d{2})\.json$")
# Exactly the reads the collector performs; mirrored by the ClusterRole.
API_PATHS = {
    "nodes": "/api/v1/nodes",
    "pods": "/api/v1/pods",
    "pvcs": "/api/v1/persistentvolumeclaims",
    "pvs": "/api/v1/persistentvolumes",
    "storageclasses": "/apis/storage.k8s.io/v1/storageclasses",
}
METRICS_PATHS = {
    "node_top": "/apis/metrics.k8s.io/v1beta1/nodes",
    "pod_top": "/apis/metrics.k8s.io/v1beta1/pods",
}


def load(path: Path, name: str):
    spec = importlib.util.spec_from_file_location(name, path)
    if spec is None or spec.loader is None:
        raise RuntimeError(f"cannot load {path}")
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)
    return module


def sha256(path: Path) -> str:
    return hashlib.sha256(path.read_bytes()).hexdigest()


def iso(value: datetime) -> str:
    return value.astimezone(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z")


WINDOW_END_HOUR = 6


def week_of(now: datetime) -> tuple[str, datetime]:
    """ISO week label and its Monday 06:00 UTC (the telemetry window end)."""
    shifted = now.astimezone(timezone.utc) - timedelta(hours=WINDOW_END_HOUR)
    year, week, weekday = shifted.isocalendar()
    monday = (shifted - timedelta(days=weekday - 1)).replace(hour=0, minute=0, second=0, microsecond=0)
    return f"{year}-W{week:02d}", monday + timedelta(hours=WINDOW_END_HOUR)


def missed_weeks(previous: str | None, current: str) -> list[str]:
    if previous is None:
        return []
    def monday(label):
        year, week = int(label[:4]), int(label[6:8])
        return datetime.fromisocalendar(year, week, 1).replace(tzinfo=timezone.utc, hour=WINDOW_END_HOUR)
    out, cursor = [], monday(previous) + timedelta(days=7)
    while cursor < monday(current):
        out.append(week_of(cursor)[0])
        cursor += timedelta(days=7)
    return out


def previous_report(reports: Path, label: str) -> Path | None:
    earlier = sorted(p for p in reports.glob("*.json") if WEEK.match(p.name) and p.stem < label)
    return earlier[-1] if earlier else None


class Kube:
    """Minimal in-cluster API reader; the token is never logged or persisted."""

    def __init__(self, base: str, token_path: Path, ca_path: Path):
        self.base = base.rstrip("/")
        self.token_path = token_path
        self.context = ssl.create_default_context(cafile=str(ca_path))

    def get(self, path: str):
        request = urllib.request.Request(self.base + path, headers={
            "Authorization": "Bearer " + self.token_path.read_text().strip(),
            "Accept": "application/json"})
        with urllib.request.urlopen(request, context=self.context, timeout=60) as response:
            return json.load(response)


def top_lines(metrics: dict, *, pods: bool, observer) -> str:
    """Render metrics.k8s.io lists in the `kubectl top` shape parse_top reads."""
    lines = []
    for item in metrics.get("items") or []:
        meta = item.get("metadata") or {}
        if pods:
            cpu = sum(observer.cpu_millicores((c.get("usage") or {}).get("cpu")) for c in item.get("containers") or [])
            mem = sum(observer.bytes_value((c.get("usage") or {}).get("memory")) for c in item.get("containers") or [])
            lines.append(f"{meta.get('namespace')} {meta.get('name')} {cpu}m {mem}")
        else:
            usage = item.get("usage") or {}
            cpu = observer.cpu_millicores(usage.get("cpu"))
            mem = observer.bytes_value(usage.get("memory"))
            lines.append(f"{meta.get('name')} {cpu}m 0% {mem} 0%")
    return "\n".join(lines) + ("\n" if lines else "")


def collect_snapshots(get, observer) -> dict:
    snapshots = {name: get(path) for name, path in API_PATHS.items()}
    for name, path in METRICS_PATHS.items():
        try:
            snapshots[name] = top_lines(get(path), pods=name == "pod_top", observer=observer)
        except (OSError, ValueError, urllib.error.URLError):
            snapshots[name] = None  # recorded by build_observation as a measurement gap
    return snapshots


def prometheus_query(base: str, promql: str, at: datetime, opener=urllib.request.urlopen) -> dict:
    """Return Prometheus' JSON response, including error bodies, never raise on 4xx/5xx."""
    url = base.rstrip("/") + "/api/v1/query?" + urllib.parse.urlencode(
        {"query": promql, "time": f"{at.timestamp():.3f}"})
    try:
        with opener(url, timeout=120) as response:
            return json.load(response)
    except urllib.error.HTTPError as exc:
        try:
            return json.loads(exc.read())
        except ValueError:
            return {"status": "error", "errorType": "http", "error": f"HTTP {exc.code}"}
    except (OSError, ValueError) as exc:
        return {"status": "error", "errorType": "transport", "error": type(exc).__name__}


def write_json(path: Path, value) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    temporary = path.with_name(path.name + ".tmp")
    temporary.write_text(json.dumps(value, indent=2, sort_keys=True) + "\n")
    os.replace(temporary, path)


def append_receipt(data: Path, receipt: dict) -> None:
    data.mkdir(parents=True, exist_ok=True)
    with (data / "receipts.jsonl").open("a") as stream:
        stream.write(json.dumps(receipt, sort_keys=True) + "\n")
        stream.flush()
        os.fsync(stream.fileno())
    print(json.dumps(receipt, sort_keys=True))


def delivered_message(data: Path, week: str) -> str | None:
    path = data / "receipts.jsonl"
    if not path.exists():
        return None
    for line in reversed(path.read_text().splitlines()):
        try:
            entry = json.loads(line)
        except ValueError:
            continue
        if entry.get("week") == week and entry.get("message_id"):
            return entry["message_id"]
    return None


def summary(receipt: dict) -> dict:
    report = json.loads(Path(receipt["report"]).read_text())
    decisions: dict[str, int] = {}
    for item in report["recommendations"]:
        decisions[item["decision"]] = decisions.get(item["decision"], 0) + 1
    return {"week": receipt["week"], "report": receipt["report"], "report_sha256": receipt["report_sha256"],
            "capture_end": report["period"]["capture_end"], "workloads": len(report["workloads"]),
            "namespaces_with_metrics": len(report["namespace_context"]),
            "recommendations": dict(sorted(decisions.items())),
            "comparison_compatible": report["comparison"]["compatible"],
            "measurement_gaps": len(report["measurement_gaps"]),
            "failed_queries": sorted(n for n, q in report["queries"].items() if q.get("response_status") != "success"),
            "next_step": "Custodian session reviews and records owner-repo tasks (ADR-001); no automatic resizing."}


def hub_post(hub: str, body: dict) -> str:
    request = urllib.request.Request(hub.rstrip("/") + "/messages/", data=json.dumps(body).encode(),
                                     headers={"Content-Type": "application/json"}, method="POST")
    with urllib.request.urlopen(request, timeout=30) as response:
        return str(json.load(response)["id"])


def deliver(data: Path, receipt: dict, hub: str | None, post=hub_post) -> tuple[dict, bool]:
    """Post one hub message per retained week; returns (receipt, ok)."""
    if receipt.get("status") not in {"retained", "already-retained"}:
        return receipt, True
    existing = delivered_message(data, receipt["week"])
    if existing:
        return {**receipt, "delivery": "already-posted", "message_id": existing}, True
    if not hub:
        return {**receipt, "delivery": "not configured"}, True
    content = summary(receipt)
    body = {"from_agent": "custodian-allocation-review", "to_agent": "the-custodian",
            "subject": f"Weekly allocation assessment {receipt['week']}",
            "body": json.dumps(content, sort_keys=True)}
    try:
        return {**receipt, "delivery": "posted", "message_id": post(hub, body)}, True
    except (OSError, ValueError, KeyError) as exc:
        return {**receipt, "delivery": f"failed: {type(exc).__name__}"}, False


def run(args, *, now: datetime, get, query) -> dict:
    sources = {
        "weekly_allocation_report": args.source_dir / "weekly_allocation_report.py",
        "reconcile_railiance_allocation": args.source_dir / "reconcile_railiance_allocation.py",
        "observe_cluster_resources": args.observer,
        "weekly_allocation_incluster": Path(__file__).resolve(),
        "queries": args.queries, "ownership": args.ownership, "inventory": args.inventory,
    }
    release_index_path = getattr(args, "release_index", None)
    if release_index_path is None:
        raise ValueError("release index must be bound")
    release_index_path = Path(release_index_path)
    sources["release_index"] = release_index_path
    release_index = json.loads(release_index_path.read_text())
    if not isinstance(release_index, dict) or (
        "units" not in release_index and "findings" not in release_index
    ):
        raise ValueError("release index is not a composed index or coverage report")
    weekly = load(sources["weekly_allocation_report"], "weekly_allocation_report")
    reconcile = load(sources["reconcile_railiance_allocation"], "reconcile_railiance_allocation")
    observer = load(sources["observe_cluster_resources"], "observe_cluster_resources")
    label, window_end = week_of(now)
    reports = args.data_dir / "reports"
    target = reports / f"{label}.json"
    receipt = {"schema": "custodian.weekly-allocation-receipt.v1", "week": label,
               "run_at": iso(now), "report": str(target), "delivery": "not emitted",
               "acknowledgment": None}
    if target.exists():
        return {**receipt, "status": "already-retained", "report_sha256": sha256(target)}
    previous_path = previous_report(reports, label)
    previous_label = previous_path.stem if previous_path else None

    observation = observer.build_observation(collect_snapshots(get, observer), iso(now))
    owners = json.loads(args.ownership.read_text())
    allocation = reconcile.join(observation, owners)
    spec = json.loads(args.queries.read_text())
    if spec.get("schema") != "custodian.weekly-allocation-queries.v1":
        raise ValueError("unsupported query specification")
    telemetry = {"captured_at": window_end.isoformat(), "window": spec["window"], "step": spec["step"],
                 "source": args.prometheus_url,
                 "queries": {name: {"promql": promql, "response": query(promql, window_end)}
                             for name, promql in sorted(spec["queries"].items())}}
    provenance = {
        "schema": "custodian.allocation-capture-provenance.v1", "workplan": "CUST-WP-0071-T05",
        "source_observation": observation["captured_at"], "read_only": True,
        "secret_surfaces_read": False, "executor": args.executor,
        "kubernetes_reads": sorted(API_PATHS.values()) + sorted(METRICS_PATHS.values()),
        "prometheus": args.prometheus_url, "telemetry_window_end": window_end.isoformat(),
        "inputs": [{"name": n, "path": str(p), "sha256": sha256(p)} for n, p in sorted(sources.items())],
    }
    inventory = json.loads(args.inventory.read_text())
    inputs = {"cluster": observation, "allocation": allocation, "telemetry": telemetry,
              "inventory": inventory, "provenance": provenance}
    attempt = args.data_dir / "runs" / label / iso(now).replace(":", "")
    for name, value in inputs.items():
        write_json(attempt / f"{name}.json", value)
    report = weekly.build_report(
        **inputs, as_of=iso(now), max_age_hours=args.max_age_hours,
        previous=json.loads(previous_path.read_text()) if previous_path else None,
        reviews=json.loads(args.reviews.read_text()) if args.reviews and args.reviews.exists() else None,
        release_index=release_index)
    weekly.retain(target, report)
    failed = sorted(n for n, q in telemetry["queries"].items() if q["response"].get("status") != "success")
    return {**receipt, "status": "retained", "report_sha256": sha256(target),
            "inputs_dir": str(attempt), "previous_week": previous_label,
            "missed_weeks": missed_weeks(previous_label, label),
            "comparison_compatible": report["comparison"]["compatible"],
            "workloads": len(report["workloads"]), "failed_queries": failed,
            "measurement_gaps": len(report["measurement_gaps"])}


def parser() -> argparse.ArgumentParser:
    env = os.environ.get
    p = argparse.ArgumentParser(description=__doc__)
    p.add_argument("--data-dir", type=Path, default=Path(env("ASSESSMENT_DATA_DIR", "/data")))
    p.add_argument("--source-dir", type=Path, default=Path(env("ASSESSMENT_SOURCE_DIR", str(HERE))))
    p.add_argument("--observer", type=Path, default=Path(env("ASSESSMENT_OBSERVER",
                   str(Path(env("ASSESSMENT_SOURCE_DIR", str(HERE))) / "observe_cluster_resources.py"))))
    p.add_argument("--queries", type=Path, default=Path(env("ASSESSMENT_QUERIES", str(HERE.parent / "config/weekly-allocation-queries.json"))))
    p.add_argument("--ownership", type=Path, default=env("ASSESSMENT_OWNERSHIP") and Path(env("ASSESSMENT_OWNERSHIP")))
    p.add_argument("--inventory", type=Path, default=env("ASSESSMENT_INVENTORY") and Path(env("ASSESSMENT_INVENTORY")))
    p.add_argument("--reviews", type=Path, default=env("ASSESSMENT_REVIEWS") and Path(env("ASSESSMENT_REVIEWS")))
    p.add_argument("--release-index", type=Path, default=Path(env(
        "ASSESSMENT_RELEASE_INDEX", str(HERE.parent / "config/release-index.json"))))
    p.add_argument("--prometheus-url", default=env("PROMETHEUS_URL", "http://telemetry-prometheus.telemetry.svc:9090"))
    p.add_argument("--hub-url", default=env("STATE_HUB_URL"), help="State Hub API; unset disables delivery")
    p.add_argument("--executor", default=env("ASSESSMENT_EXECUTOR", "unspecified"))
    p.add_argument("--max-age-hours", type=float, default=float(env("ASSESSMENT_MAX_AGE_HOURS", "48")))
    return p


def main() -> int:
    args = parser().parse_args()
    now = datetime.now(timezone.utc)
    label = week_of(now)[0]
    try:
        if not args.ownership or not args.inventory:
            raise ValueError("ownership and inventory inputs must be bound")
        host = os.environ["KUBERNETES_SERVICE_HOST"]
        port = os.environ.get("KUBERNETES_SERVICE_PORT", "443")
        kube = Kube(f"https://{host}:{port}", SA_DIR / "token", SA_DIR / "ca.crt")
        receipt = run(args, now=now, get=kube.get,
                      query=lambda promql, at: prometheus_query(args.prometheus_url, promql, at))
    except Exception as exc:  # noqa: BLE001 - every failure must leave a receipt
        append_receipt(args.data_dir, {"schema": "custodian.weekly-allocation-receipt.v1",
                       "week": label, "run_at": iso(now), "status": "failed",
                       "error": f"{type(exc).__name__}: {exc}"[:500]})
        return 1
    receipt, delivered = deliver(args.data_dir, receipt, args.hub_url)
    append_receipt(args.data_dir, receipt)
    return 0 if delivered else 1  # undelivered: Job retry posts it (report already retained)


if __name__ == "__main__":
    sys.exit(main())