#!/usr/bin/env python3
import json
import os
import re
import shutil
import subprocess
import sys
import time
from datetime import datetime, timezone


def now_iso():
    return datetime.now(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z")


def command(args, timeout=60):
    started = time.time()
    try:
        result = subprocess.run(args, text=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, timeout=timeout)
        return {
            "exitCode": result.returncode,
            "elapsedMs": int((time.time() - started) * 1000),
            "stdoutTail": result.stdout[-1200:],
            "stderrTail": result.stderr[-1200:],
        }
    except subprocess.TimeoutExpired as exc:
        return {
            "exitCode": 124,
            "elapsedMs": int((time.time() - started) * 1000),
            "stdoutTail": (exc.stdout or "")[-1200:] if isinstance(exc.stdout, str) else "",
            "stderrTail": (exc.stderr or "")[-1200:] if isinstance(exc.stderr, str) else "",
            "timedOut": True,
        }


def load_config(path):
    with open(path, "r", encoding="utf-8") as handle:
        return json.load(handle)


def safe_int(value, default=0):
    try:
        return int(value)
    except Exception:
        return default


def du_size(path):
    result = command(["du", "-sb", path], 15)
    if result["exitCode"] != 0:
        return 0
    try:
        return int((result["stdoutTail"] or "0").split()[0])
    except Exception:
        return 0


def path_size(path):
    try:
        stat = os.lstat(path)
    except OSError:
        return 0
    if os.path.isdir(path) and not os.path.islink(path):
        return du_size(path)
    return int(getattr(stat, "st_blocks", 0)) * 512


def path_has_open_fd(path):
    resolved = os.path.realpath(path)
    prefix = resolved.rstrip("/") + "/"
    try:
        pids = [name for name in os.listdir("/proc") if name.isdigit()]
    except OSError:
        return True
    for pid in pids:
        base = os.path.join("/proc", pid)
        for name in ["cwd", "root"]:
            try:
                target = os.path.realpath(os.readlink(os.path.join(base, name)))
            except OSError:
                continue
            if target == resolved or target.startswith(prefix):
                return True
        fd_dir = os.path.join(base, "fd")
        try:
            fds = os.listdir(fd_dir)
        except OSError:
            continue
        for fd in fds:
            try:
                target = os.path.realpath(os.readlink(os.path.join(fd_dir, fd)))
            except OSError:
                continue
            if target == resolved or target.startswith(prefix):
                return True
    return False


def run_json(args, timeout=45):
    started = time.time()
    try:
        process = subprocess.run(args, text=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, timeout=timeout)
    except subprocess.TimeoutExpired as exc:
        return None, {
            "exitCode": 124,
            "elapsedMs": int((time.time() - started) * 1000),
            "stdoutTail": (exc.stdout or "")[-1200:] if isinstance(exc.stdout, str) else "",
            "stderrTail": (exc.stderr or "")[-1200:] if isinstance(exc.stderr, str) else "",
            "timedOut": True,
        }
    result = {
        "exitCode": process.returncode,
        "elapsedMs": int((time.time() - started) * 1000),
        "stdoutTail": process.stdout[-1200:],
        "stderrTail": process.stderr[-1200:],
    }
    if process.returncode != 0:
        return None, result
    try:
        return json.loads(process.stdout or "{}"), result
    except Exception:
        return None, result


def write_state(config, payload):
    state_path = config.get("statePath")
    if not state_path:
        return
    os.makedirs(os.path.dirname(state_path), mode=0o755, exist_ok=True)
    tmp = "%s.tmp.%d" % (state_path, os.getpid())
    with open(tmp, "w", encoding="utf-8") as handle:
        json.dump(payload, handle, ensure_ascii=False, sort_keys=True)
        handle.write("\n")
    os.replace(tmp, state_path)


def run_journal(config):
    target = safe_int(config.get("journalTargetBytes"), 268435456)
    result = command(["journalctl", "--vacuum-size=%d" % target], 30)
    return {"id": "journal", "ok": result["exitCode"] == 0, "command": result}


def run_apt(config):
    if not config.get("includeAptCache"):
        return {"id": "apt-cache", "ok": True, "skipped": True}
    before = du_size("/var/cache/apt/archives")
    result = command(["apt-get", "clean"], 30)
    after = du_size("/var/cache/apt/archives")
    return {"id": "apt-cache", "ok": result["exitCode"] == 0, "reclaimedBytes": max(0, before - after), "command": result}


def run_tmp(config):
    prefixes = list(config.get("tmpPrefixAllowlist") or [])
    protected = set(config.get("tmpExactProtect") or [])
    cutoff = time.time() - float(config.get("tmpMinAgeHours") or 24) * 3600.0
    deleted = 0
    reclaimed = 0
    if not os.path.isdir("/tmp"):
        return {"id": "tmp-allowlist", "ok": True, "deletedCount": 0, "reclaimedBytes": 0}
    for name in os.listdir("/tmp"):
        path = os.path.join("/tmp", name)
        if path in protected or not any(name.startswith(prefix) for prefix in prefixes):
            continue
        try:
            stat = os.lstat(path)
        except OSError:
            continue
        if stat.st_mtime >= cutoff or path_has_open_fd(path):
            continue
        before = path_size(path)
        if os.path.isdir(path) and not os.path.islink(path):
            shutil.rmtree(path, ignore_errors=True)
        else:
            try:
                os.unlink(path)
            except FileNotFoundError:
                pass
        deleted += 1
        reclaimed += before
    return {"id": "tmp-allowlist", "ok": True, "deletedCount": deleted, "reclaimedBytes": reclaimed}


def run_tool_caches(config):
    paths = list(config.get("toolCachePaths") or [])
    reclaimed = 0
    deleted = 0
    for path in paths:
        resolved = os.path.abspath(path)
        if resolved != path or os.path.islink(resolved) or resolved in ["/", "/root", "/root/.npm", "/root/.bun"]:
            continue
        before = path_size(resolved)
        if os.path.isdir(resolved):
            shutil.rmtree(resolved, ignore_errors=True)
        elif os.path.exists(resolved):
            try:
                os.unlink(resolved)
            except FileNotFoundError:
                pass
        else:
            continue
        deleted += 1
        reclaimed += before
    return {"id": "tool-caches", "ok": True, "deletedCount": deleted, "reclaimedBytes": reclaimed}


def pid_alive(pid):
    try:
        value = int(pid)
    except Exception:
        return False
    return value > 0 and os.path.exists("/proc/%d" % value)


def read_json_file(path):
    try:
        with open(path, "r", encoding="utf-8") as handle:
            return json.load(handle)
    except Exception:
        return {}


def run_web_observe(config):
    cfg = config.get("webObserve") or {}
    if not cfg.get("enabled"):
        return {"id": "web-observe-artifacts", "ok": True, "skipped": True}
    roots = [os.path.abspath(item) for item in (cfg.get("observeStateRoots") or cfg.get("webObserveRoots") or []) if isinstance(item, str) and item.startswith("/")]
    stale_hours = float(cfg.get("staleRunMaxAgeHours") or 6)
    cutoff = time.time() - stale_hours * 3600.0
    deleted = 0
    reclaimed = 0
    protected = 0
    for root in roots:
        if not os.path.isdir(root):
            continue
        for name in os.listdir(root):
            path = os.path.abspath(os.path.join(root, name))
            if os.path.dirname(path) != root or os.path.islink(path) or not os.path.isdir(path):
                continue
            manifest = read_json_file(os.path.join(path, "manifest.json"))
            heartbeat = read_json_file(os.path.join(path, "heartbeat.json"))
            pid = manifest.get("pid") or heartbeat.get("pid") or read_json_file(os.path.join(path, "pid")).get("pid")
            try:
                stat = os.lstat(path)
            except OSError:
                continue
            completed = bool(manifest.get("completedAt") or heartbeat.get("completedAt") or manifest.get("status") in ["completed", "failed", "blocked"])
            stale = completed or stat.st_mtime < cutoff
            if pid_alive(pid) or not stale or path_has_open_fd(path):
                protected += 1
                continue
            before = path_size(path)
            shutil.rmtree(path, ignore_errors=True)
            deleted += 1
            reclaimed += before
    return {"id": "web-observe-artifacts", "ok": True, "deletedCount": deleted, "protectedCount": protected, "reclaimedBytes": reclaimed}


def active_claims(namespace):
    pods, _ = run_json(["kubectl", "-n", namespace, "get", "pod", "-o", "json"], 30)
    claims = {}
    for pod in (pods or {}).get("items") or []:
        if ((pod.get("status") or {}).get("phase")) in ["Succeeded", "Failed"]:
            continue
        pod_name = ((pod.get("metadata") or {}).get("name")) or ""
        for volume in ((pod.get("spec") or {}).get("volumes") or []):
            claim = (volume.get("persistentVolumeClaim") or {}).get("claimName")
            if claim:
                claims.setdefault(claim, []).append(pod_name)
    return claims


def run_session_pvcs(config):
    cfg = config.get("agentrunSessionPvcs") or {}
    if not cfg.get("enabled"):
        return {"id": "agentrun-session-pvcs", "ok": True, "skipped": True}
    namespace = cfg.get("namespace")
    prefixes = list(cfg.get("prefixes") or [])
    limit = max(1, min(safe_int(cfg.get("maxDeletePerRun"), 100), 1000))
    if not namespace or not prefixes:
        return {"id": "agentrun-session-pvcs", "ok": False, "error": "namespace-or-prefixes-missing"}
    pv_data, pv_cmd = run_json(["kubectl", "get", "pv", "-o", "json"], 30)
    pvc_data, pvc_cmd = run_json(["kubectl", "-n", namespace, "get", "pvc", "-o", "json"], 30)
    if pv_data is None or pvc_data is None:
        return {"id": "agentrun-session-pvcs", "ok": False, "pvCommand": pv_cmd, "pvcCommand": pvc_cmd}
    pvs = {((pv.get("metadata") or {}).get("name")): pv for pv in pv_data.get("items") or []}
    active = active_claims(namespace)
    selected = []
    protected = 0
    for pvc in pvc_data.get("items") or []:
        name = ((pvc.get("metadata") or {}).get("name")) or ""
        if not any(name.startswith(prefix) for prefix in prefixes):
            continue
        pv = pvs.get((pvc.get("spec") or {}).get("volumeName")) or {}
        storage_class = (pvc.get("spec") or {}).get("storageClassName") or (pv.get("spec") or {}).get("storageClassName")
        reclaim_policy = (pv.get("spec") or {}).get("persistentVolumeReclaimPolicy")
        if active.get(name) or storage_class != "local-path" or reclaim_policy != "Delete":
            protected += 1
            continue
        selected.append(name)
    selected = selected[:limit]
    if selected:
        result = command(["kubectl", "-n", namespace, "delete", "pvc", "--wait=false"] + selected, 45)
        ok = result["exitCode"] == 0
    else:
        result = None
        ok = True
    return {"id": "agentrun-session-pvcs", "ok": ok, "deletedPvcCount": len(selected), "protectedCount": protected, "command": result}


def ci_active(config):
    namespaces = list((config.get("k3sImageCache") or {}).get("ciNamespaces") or [])
    active = []
    for namespace in namespaces:
        result = command(["sh", "-lc", "KUBECONFIG=/etc/rancher/k3s/k3s.yaml kubectl get pipelinerun,taskrun,job -n %s --no-headers 2>/dev/null | awk '$2 != \"True\" && $2 != \"False\" && $2 != \"Complete\" && $2 != \"Failed\" {print}' | head -20" % namespace], 15)
        for line in (result.get("stdoutTail") or "").splitlines():
            if line.strip():
                active.append({"namespace": namespace, "line": line.strip()})
    return active


def run_k3s_images(config):
    cfg = config.get("k3sImageCache") or {}
    if not cfg.get("enabled"):
        return {"id": "k3s-image-cache", "ok": True, "skipped": True}
    active = ci_active(config)
    if active:
        return {"id": "k3s-image-cache", "ok": True, "skipped": True, "reason": "ci-active", "activePreview": active[:5]}
    endpoint = cfg.get("runtimeEndpoint") or "unix:///run/k3s/containerd/containerd.sock"
    before = du_size("/var/lib/rancher/k3s/agent/containerd")
    result = command(["crictl", "--runtime-endpoint", endpoint, "rmi", "--prune"], 300)
    after = du_size("/var/lib/rancher/k3s/agent/containerd")
    return {"id": "k3s-image-cache", "ok": result["exitCode"] == 0, "reclaimedBytes": max(0, before - after), "command": result}


def table_data_lines(stdout, header_prefix):
    return [line.strip() for line in str(stdout or "").splitlines() if line.strip() and not line.startswith(header_prefix)]


def host_containerd_empty(config):
    cfg = config.get("hostContainerdCache") or {}
    address = cfg.get("address") or "/run/containerd/containerd.sock"
    namespaces = cfg.get("namespaces") or ["default"]
    active = []
    for namespace in namespaces:
        base = ["ctr", "--address", address, "-n", namespace]
        for kind, args, header in [
            ("image", ["images", "list", "-q"], ""),
            ("container", ["containers", "list", "-q"], ""),
            ("task", ["tasks", "list", "-q"], ""),
            ("lease", ["leases", "list", "-q"], ""),
            ("snapshot", ["snapshots", "list"], "KEY"),
            ("content", ["content", "list"], "DIGEST"),
        ]:
            result = command(base + args, 20)
            if result["exitCode"] != 0:
                active.append({"namespace": namespace, "kind": kind, "state": "unknown"})
                continue
            lines = table_data_lines(result.get("stdoutTail") or "", header) if header else [line for line in (result.get("stdoutTail") or "").splitlines() if line.strip()]
            for line in lines:
                active.append({"namespace": namespace, "kind": kind, "name": line.split()[0] if line.split() else line})
    return active


def direct_child_rows(root, predicate):
    rows = []
    if not root or not os.path.isdir(root) or os.path.islink(root):
        return rows
    real_root = os.path.realpath(os.path.abspath(root))
    for name in sorted(os.listdir(real_root)):
        path = os.path.realpath(os.path.abspath(os.path.join(real_root, name)))
        if os.path.dirname(path) != real_root or not predicate(name, path):
            continue
        rows.append({"name": name, "path": path, "sizeBytes": path_size(path)})
    rows.sort(key=lambda item: item["sizeBytes"], reverse=True)
    return rows


def run_host_containerd(config):
    cfg = config.get("hostContainerdCache") or {}
    if not cfg.get("enabled"):
        return {"id": "host-containerd-orphans", "ok": True, "skipped": True}
    active = host_containerd_empty(config)
    if active:
        return {"id": "host-containerd-orphans", "ok": True, "skipped": True, "reason": "metadata-not-empty", "activePreview": active[:8]}
    orphan = cfg.get("orphanCleanup") or {}
    if not orphan.get("enabled"):
        return {"id": "host-containerd-orphans", "ok": True, "skipped": True, "reason": "orphan-cleanup-disabled"}
    overlay_root = os.path.realpath(os.path.abspath(orphan.get("overlaySnapshotsRoot") or ""))
    content_root = os.path.realpath(os.path.abspath(orphan.get("contentBlobRoot") or ""))
    root = os.path.realpath(os.path.abspath(cfg.get("root") or ""))
    if not root or not overlay_root.startswith(root + "/") or not content_root.startswith(root + "/"):
        return {"id": "host-containerd-orphans", "ok": False, "error": "orphan-root-outside-containerd-root"}
    for root_path in [overlay_root, content_root]:
        if os.path.exists(root_path) and path_has_open_fd(root_path):
            return {"id": "host-containerd-orphans", "ok": True, "skipped": True, "reason": "open-fd", "root": root_path}
    rows = direct_child_rows(overlay_root, lambda name, path: os.path.isdir(path) and not os.path.islink(path) and re.match(r"^[0-9]+$", name) is not None)
    rows += direct_child_rows(content_root, lambda name, path: os.path.isfile(path) and not os.path.islink(path) and re.match(r"^[0-9a-f]{64}$", name) is not None)
    rows.sort(key=lambda item: item["sizeBytes"], reverse=True)
    limit = max(1, min(safe_int(orphan.get("maxDeletePerRun"), 100), 1000))
    reclaimed = 0
    deleted = 0
    for row in rows[:limit]:
        before = path_size(row["path"])
        if os.path.isdir(row["path"]):
            shutil.rmtree(row["path"], ignore_errors=True)
        else:
            try:
                os.unlink(row["path"])
            except FileNotFoundError:
                continue
        reclaimed += before
        deleted += 1
    return {"id": "host-containerd-orphans", "ok": True, "deletedCount": deleted, "candidateCount": len(rows), "reclaimedBytes": reclaimed}


def run_local_path_orphans(config):
    cfg = config.get("localPathStorage") or {}
    if not cfg.get("enabled"):
        return {"id": "local-path-orphans", "ok": True, "skipped": True}
    root = os.path.realpath(os.path.abspath(cfg.get("root") or ""))
    prefixes = list(cfg.get("orphanDirPrefixes") or [])
    if not root or not prefixes or not os.path.isdir(root):
        return {"id": "local-path-orphans", "ok": False, "error": "root-or-prefix-missing"}
    pv_data, pv_cmd = run_json(["kubectl", "get", "pv", "-o", "json"], 30)
    if pv_data is None:
        return {"id": "local-path-orphans", "ok": False, "pvCommand": pv_cmd}
    referenced = set()
    for pv in pv_data.get("items") or []:
        spec = pv.get("spec") or {}
        path = ((spec.get("hostPath") or {}).get("path")) or ((spec.get("local") or {}).get("path"))
        if path:
            referenced.add(os.path.realpath(os.path.abspath(path)))
    rows = []
    for name in os.listdir(root):
        path = os.path.realpath(os.path.abspath(os.path.join(root, name)))
        if os.path.dirname(path) != root or not any(name.startswith(prefix) for prefix in prefixes) or path in referenced or path_has_open_fd(path):
            continue
        if os.path.isdir(path) and not os.path.islink(path):
            rows.append({"name": name, "path": path, "sizeBytes": path_size(path)})
    rows.sort(key=lambda item: item["sizeBytes"], reverse=True)
    limit = max(1, min(safe_int(cfg.get("maxDeletePerRun"), 100), 1000))
    reclaimed = 0
    deleted = 0
    for row in rows[:limit]:
        before = path_size(row["path"])
        shutil.rmtree(row["path"], ignore_errors=True)
        reclaimed += before
        deleted += 1
    return {"id": "local-path-orphans", "ok": True, "deletedCount": deleted, "candidateCount": len(rows), "reclaimedBytes": reclaimed}


def main():
    config_path = sys.argv[1] if len(sys.argv) > 1 else ""
    config = load_config(config_path)
    results = []
    for fn in [run_journal, run_apt, run_tmp, run_tool_caches, run_web_observe, run_session_pvcs, run_k3s_images, run_host_containerd, run_local_path_orphans]:
        try:
            results.append(fn(config))
        except Exception as exc:
            results.append({"id": getattr(fn, "__name__", "stage"), "ok": False, "error": str(exc)})
    payload = {
        "ok": all(item.get("ok") for item in results),
        "observedAt": now_iso(),
        "providerId": config.get("providerId"),
        "unitName": config.get("unitName"),
        "resultCount": len(results),
        "failedCount": len([item for item in results if not item.get("ok")]),
        "reclaimedBytes": sum(safe_int(item.get("reclaimedBytes")) for item in results),
        "results": results,
    }
    write_state(config, payload)
    print(json.dumps(payload, ensure_ascii=False))
    return 0 if payload["ok"] else 1


if __name__ == "__main__":
    raise SystemExit(main())