#!/usr/bin/env python3
"""NW AL Scanner process cleanup / preflight helper.

Purpose:
- STOP_* must stop the launcher *and* any orphan service processes that survived it.
- START_* must not silently adopt an old parser/server left over from a previous build.

This script is intentionally conservative: it only force-stops listeners whose
command lines identify them as one of this platform's known services.
"""
from __future__ import annotations

import json
import os
import subprocess
import sys
import time
from pathlib import Path

BASE_DIR = Path(__file__).resolve().parent
TENANTS_DIR = BASE_DIR / "tenants"

KNOWN_PORT_SIGNATURES = {
    "core": ("serve_v097_incident_video.py",),
    "peet": ("peet_server_hazcams_wu.py",),
    "lightning": ("server.js", "lightning-radar-dashboard"),
    "operations": ("platform_launcher.py",),
    "radio_intelligence": ("radio_intelligence_server.py",),
}


def load_tenant(tenant_id: str) -> dict:
    path = TENANTS_DIR / tenant_id / "station.json"
    if not path.exists():
        raise SystemExit(f"Tenant not found: {tenant_id}")
    return json.loads(path.read_text(encoding="utf-8-sig"))


def runtime_dir(tenant_id: str) -> Path:
    return BASE_DIR / "runtime" / "instances" / tenant_id


def pid_alive(pid: int) -> bool:
    if os.name != "nt":
        try:
            os.kill(pid, 0)
            return True
        except OSError:
            return False
    cp = subprocess.run(
        ["tasklist", "/FI", f"PID eq {pid}", "/FO", "CSV", "/NH"],
        capture_output=True, text=True, errors="replace"
    )
    return f'"{pid}"' in cp.stdout


def process_info(pid: int) -> dict | None:
    if os.name != "nt":
        return {"ProcessId": pid, "Name": "", "CommandLine": ""}
    ps = (
        f"$p=Get-CimInstance Win32_Process -Filter \"ProcessId = {int(pid)}\" -ErrorAction SilentlyContinue; "
        "if($p){$p | Select-Object ProcessId,Name,ExecutablePath,CommandLine,ParentProcessId | ConvertTo-Json -Compress}"
    )
    cp = subprocess.run(["powershell", "-NoProfile", "-Command", ps], capture_output=True, text=True, errors="replace")
    text = cp.stdout.strip()
    if not text:
        return None
    try:
        return json.loads(text)
    except Exception:
        return None


def listener_pids(port: int) -> list[int]:
    if os.name != "nt":
        return []
    ps = (
        f"Get-NetTCPConnection -LocalPort {int(port)} -State Listen -ErrorAction SilentlyContinue | "
        "Select-Object -ExpandProperty OwningProcess -Unique | ConvertTo-Json -Compress"
    )
    cp = subprocess.run(["powershell", "-NoProfile", "-Command", ps], capture_output=True, text=True, errors="replace")
    text = cp.stdout.strip()
    if not text:
        return []
    try:
        data = json.loads(text)
    except Exception:
        return []
    if isinstance(data, int):
        return [data]
    return [int(x) for x in (data or [])]


def taskkill(pid: int, tree: bool = True) -> bool:
    if not pid_alive(pid):
        return True
    if os.name == "nt":
        cmd = ["taskkill", "/PID", str(pid)]
        if tree:
            cmd.append("/T")
        cmd.append("/F")
        cp = subprocess.run(cmd, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
        return cp.returncode == 0 or not pid_alive(pid)
    try:
        os.kill(pid, 15)
        return True
    except OSError:
        return True


def signature_matches(info: dict | None, signatures: tuple[str, ...]) -> bool:
    if not info:
        return False
    command = str(info.get("CommandLine") or "").lower()
    # For the node dashboard, both server.js and the module path must be present.
    if signatures == KNOWN_PORT_SIGNATURES["lightning"]:
        return all(s.lower() in command for s in signatures)
    return any(s.lower() in command for s in signatures)


def stop_launcher_from_pidfile(tenant_id: str) -> None:
    pidfile = runtime_dir(tenant_id) / "platform_launcher.pid"
    if not pidfile.exists():
        return
    try:
        pid = int(pidfile.read_text(encoding="utf-8").strip())
    except Exception:
        pid = 0
    if pid and pid_alive(pid):
        print(f"Stopping platform launcher PID {pid} and its child tree...")
        taskkill(pid, tree=True)
        # Give Windows a moment to release child listeners.
        for _ in range(20):
            if not pid_alive(pid):
                break
            time.sleep(0.1)
    try:
        pidfile.unlink(missing_ok=True)
    except Exception:
        pass


def stop_known_port_owners(tenant: dict, *, verbose: bool = True) -> int:
    ports = tenant.get("ports") or {}
    services = tenant.get("services") or {}
    role_enabled = {
        "core": bool(services.get("core", True)),
        "peet": bool(services.get("peet", False)),
        "lightning": bool(services.get("lightning_dashboard", True)),
        "operations": True,
        "radio_intelligence": bool(services.get("radio_intelligence", False)),
    }
    stopped = 0
    for role in ("core", "peet", "lightning", "radio_intelligence", "operations"):
        if not role_enabled.get(role):
            continue
        port = ports.get(role)
        if port is None:
            continue
        for pid in listener_pids(int(port)):
            info = process_info(pid)
            if signature_matches(info, KNOWN_PORT_SIGNATURES[role]):
                if verbose:
                    print(f"Stopping stale {role} listener on port {port}: PID {pid} ({info.get('Name','process')})")
                if taskkill(pid, tree=True):
                    stopped += 1
            elif verbose:
                cmd = str((info or {}).get("CommandLine") or "unknown process")
                print(f"WARNING: port {port} is occupied by an unrecognized process PID {pid}; leaving it untouched.")
                print(f"         {cmd}")
    return stopped


def stop_spark_for_tenant(tenant_id: str) -> int:
    """Spark has no listen port, so identify this tenant by its --output-dir argument."""
    if os.name != "nt":
        return 0
    marker = str((runtime_dir(tenant_id) / "lightning").resolve()).lower().replace("'", "''")
    ps = (
        "$items=Get-CimInstance Win32_Process | Where-Object { "
        "$_.Name -match '^py(?:thon(?:w)?(?:3(?:\\.\\d+)?)?)?\\.exe$' -and "
        "$_.CommandLine -like '*spark_lightning_bridge_v003.py*' -and "
        f"$_.CommandLine.ToLower().Contains('{marker}') "
        "}; $items | Select-Object -ExpandProperty ProcessId | ConvertTo-Json -Compress"
    )
    cp = subprocess.run(["powershell", "-NoProfile", "-Command", ps], capture_output=True, text=True, errors="replace")
    text = cp.stdout.strip()
    if not text:
        return 0
    try:
        data = json.loads(text)
    except Exception:
        return 0
    pids = [data] if isinstance(data, int) else list(data or [])
    count = 0
    for pid in pids:
        print(f"Stopping stale Spark bridge PID {pid} for {tenant_id}...")
        if taskkill(int(pid), tree=True):
            count += 1
    return count


def verify_ports_free(tenant: dict) -> bool:
    ports = tenant.get("ports") or {}
    services = tenant.get("services") or {}
    roles = ["core", "operations"]
    if services.get("peet", False):
        roles.append("peet")
    if services.get("lightning_dashboard", True):
        roles.append("lightning")
    if services.get("radio_intelligence", False):
        roles.append("radio_intelligence")
    busy = []
    for role in roles:
        port = ports.get(role)
        if port is not None and listener_pids(int(port)):
            busy.append((role, port, listener_pids(int(port))))
    if busy:
        print("One or more required ports are still occupied:")
        for role, port, pids in busy:
            print(f"  {role}: port {port}, PID(s) {', '.join(map(str,pids))}")
        return False
    return True


def do_stop(tenant_id: str) -> int:
    tenant = load_tenant(tenant_id)
    print(f"NWAL process cleanup: {tenant.get('display_name', tenant_id)}")
    stop_launcher_from_pidfile(tenant_id)
    # A launcher killed unexpectedly in an older build may have orphaned children.
    stop_known_port_owners(tenant)
    stop_spark_for_tenant(tenant_id)
    # Re-run port cleanup once, because child teardown can race with the first scan.
    time.sleep(0.5)
    stop_known_port_owners(tenant, verbose=False)
    ok = verify_ports_free(tenant)
    if ok:
        print("All managed listeners are stopped and required ports are free.")
        return 0
    print("STOP completed with warnings. Review the occupied ports above.")
    return 2


def do_preflight(tenant_id: str) -> int:
    tenant = load_tenant(tenant_id)
    pidfile = runtime_dir(tenant_id) / "platform_launcher.pid"
    if pidfile.exists():
        try:
            pid = int(pidfile.read_text(encoding="utf-8").strip())
        except Exception:
            pid = 0
        if pid and pid_alive(pid):
            print(f"Platform launcher is already running as PID {pid}.")
            return 3
        try:
            pidfile.unlink(missing_ok=True)
        except Exception:
            pass
    # No active launcher owns this tenant. Remove any stale known services from an
    # older build before the new launcher is allowed to claim the ports.
    stop_known_port_owners(tenant)
    stop_spark_for_tenant(tenant_id)
    time.sleep(0.35)
    if not verify_ports_free(tenant):
        print("Preflight failed. A required port is occupied by a process we did not recognize.")
        return 2
    print("Preflight OK: no stale NWAL service processes remain.")
    return 0


def main() -> int:
    if len(sys.argv) != 3 or sys.argv[1] not in {"stop", "preflight"}:
        print("Usage: py -X utf8 platform_process_control.py stop|preflight TENANT_ID")
        return 64
    action, tenant_id = sys.argv[1], sys.argv[2]
    return do_stop(tenant_id) if action == "stop" else do_preflight(tenant_id)


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