import type { CommandResult } from "./command";

export interface EgressBenchmarkSpec {
  scope: "platform-infra" | "hwlab-control-plane";
  targetId: string;
  route: string;
  namespace: string;
  serviceName: string;
  port: number;
  noProxy: readonly string[];
  sourceType: string;
  sourceRef: string;
  sourceConfigRef: string | null;
  sourceFingerprint: string | null;
  profile: "no-mirror";
  samples: number;
  sampleTimeoutSeconds: number;
}

export function egressBenchmarkStateDir(spec: EgressBenchmarkSpec): string {
  return `/tmp/unidesk-egress-benchmark/${spec.scope}/${safeSegment(spec.targetId)}/${spec.profile}`;
}

export function egressBenchmarkStartScript(spec: EgressBenchmarkSpec): string {
  const stateDir = egressBenchmarkStateDir(spec);
  const noProxy = unique(["localhost", "127.0.0.1", "::1", ".svc", ".svc.cluster.local", ".cluster.local", ...spec.noProxy]).join(",");
  return `
set -eu
state_dir=${shQuote(stateDir)}
mkdir -p "$state_dir"
if [ -s "$state_dir/pid" ] && kill -0 "$(cat "$state_dir/pid")" >/dev/null 2>&1; then
  printf '{"started":false,"reason":"job-already-running","pid":%s,"stateDir":"%s"}\\n' "$(cat "$state_dir/pid")" "$state_dir"
  exit 0
fi
cat >"$state_dir/job.sh" <<'JOB'
#!/bin/sh
set -eu
state_dir=${shQuote(stateDir)}
status="$state_dir/status.json"
log="$state_dir/job.log"
samples=${shQuote(String(spec.samples))}
sample_timeout=${shQuote(String(spec.sampleTimeoutSeconds))}
target_id=${shQuote(spec.targetId)}
	target_safe=${shQuote(safeSegment(spec.targetId).toLowerCase())}
scope=${shQuote(spec.scope)}
profile=${shQuote(spec.profile)}
namespace=${shQuote(spec.namespace)}
service_name=${shQuote(spec.serviceName)}
	service_port=${shQuote(String(spec.port))}
	source_type=${shQuote(spec.sourceType)}
	source_ref=${shQuote(spec.sourceRef)}
	source_config_ref=${shQuote(spec.sourceConfigRef ?? "")}
	source_fingerprint=${shQuote(spec.sourceFingerprint ?? "")}
	docker_image='quay.io/prometheus/busybox:latest'
	no_proxy=${shQuote(noProxy)}
write_state() {
  state="$1"; shift
  message="$1"; shift || true
  python3 - "$status" "$state" "$message" "$target_id" "$scope" "$profile" <<'PY'
import json, pathlib, sys, time
path=pathlib.Path(sys.argv[1])
payload={"ok": sys.argv[2] == "succeeded", "state":sys.argv[2], "message":sys.argv[3], "target":sys.argv[4], "scope":sys.argv[5], "profile":sys.argv[6], "updatedAt":time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())}
if path.exists():
    try:
        old=json.loads(path.read_text())
        if isinstance(old, dict):
            old.update(payload)
            payload=old
    except Exception:
        pass
path.write_text(json.dumps(payload, ensure_ascii=False) + "\\n")
PY
}
classify_family() {
  rc="$1"
  text="$2"
  if [ "$rc" = "0" ]; then printf success; return; fi
  printf '%s' "$text" | grep -Eiq 'toomanyrequests|too many requests|rate limit' && { printf rate-limit; return; }
  printf '%s' "$text" | grep -Eiq 'timed out|timeout|i/o timeout|TLS handshake timeout' && { printf timeout; return; }
  printf '%s' "$text" | grep -Eiq 'Could not resolve|no such host|ENOTFOUND|getaddrinfo|Temporary failure in name resolution' && { printf dns; return; }
  printf '%s' "$text" | grep -Eiq '407|proxy|CONNECT|connection reset|ECONNRESET|ECONNREFUSED' && { printf proxy-or-connect; return; }
  printf exit-"$rc"
}
sample_json() {
  test_name="$1"
  sample="$2"
  cache_state="$3"
  started_ms="$4"
  rc="$5"
  family="$6"
  elapsed_ms="$7"
  python3 - "$state_dir/results.ndjson" "$test_name" "$sample" "$cache_state" "$started_ms" "$rc" "$family" "$elapsed_ms" "$source_type" "$source_ref" "$source_config_ref" "$source_fingerprint" <<'PY'
import json, pathlib, sys
path=pathlib.Path(sys.argv[1])
payload={
  "test": sys.argv[2],
  "sample": int(sys.argv[3]),
  "proxy": "on",
  "cacheState": sys.argv[4],
  "startedMs": int(sys.argv[5]),
  "exitCode": int(sys.argv[6]),
  "failureFamily": sys.argv[7],
  "elapsedMs": int(sys.argv[8]),
  "sourceType": sys.argv[9],
  "sourceRef": sys.argv[10],
  "sourceConfigRef": sys.argv[11] or None,
  "sourceFingerprint": sys.argv[12] or None,
  "valuesPrinted": False,
}
with path.open("a", encoding="utf-8") as handle:
    handle.write(json.dumps(payload, ensure_ascii=False) + "\\n")
PY
}
run_sample() {
  test_name="$1"
  sample="$2"
  cache_state="$3"
  shift 3
  out="$state_dir/\${test_name}-\${sample}.out"
  err="$state_dir/\${test_name}-\${sample}.err"
  started_ms=$(date +%s%3N 2>/dev/null || python3 - <<'PY'
import time
print(int(time.time()*1000))
PY
)
  set +e
  env HTTP_PROXY="$proxy_url" HTTPS_PROXY="$proxy_url" ALL_PROXY="$proxy_url" http_proxy="$proxy_url" https_proxy="$proxy_url" all_proxy="$proxy_url" NO_PROXY="$no_proxy" no_proxy="$no_proxy" npm_config_proxy="$proxy_url" npm_config_https_proxy="$proxy_url" npm_config_noproxy="$no_proxy" "$@" >"$out" 2>"$err"
  rc=$?
  set -e
  finished_ms=$(date +%s%3N 2>/dev/null || python3 - <<'PY'
import time
print(int(time.time()*1000))
PY
)
  elapsed_ms=$((finished_ms - started_ms))
  family=$(classify_family "$rc" "$(cat "$err" "$out" 2>/dev/null | tail -40)")
  sample_json "$test_name" "$sample" "$cache_state" "$started_ms" "$rc" "$family" "$elapsed_ms"
}
summarize() {
  python3 - "$state_dir/results.ndjson" "$status" "$target_id" "$scope" "$profile" "$proxy_url" "$source_type" "$source_ref" "$source_config_ref" "$source_fingerprint" <<'PY'
import json, pathlib, statistics, sys, time
results_path=pathlib.Path(sys.argv[1])
status_path=pathlib.Path(sys.argv[2])
items=[]
if results_path.exists():
    items=[json.loads(line) for line in results_path.read_text().splitlines() if line.strip()]
by_test={}
for item in items:
    by_test.setdefault(item["test"], []).append(item)
rows=[]
for name in sorted(by_test):
    values=by_test[name]
    elapsed=sorted(int(item["elapsedMs"]) for item in values)
    failures={}
    for item in values:
        if item["failureFamily"] != "success":
            failures[item["failureFamily"]] = failures.get(item["failureFamily"], 0) + 1
    def percentile(p):
        if not elapsed:
            return None
        idx=min(len(elapsed)-1, max(0, round((len(elapsed)-1)*p)))
        return elapsed[idx]
    rows.append({
        "test": name,
        "samples": len(values),
        "success": sum(1 for item in values if item["exitCode"] == 0),
        "successRate": (sum(1 for item in values if item["exitCode"] == 0) / len(values)) if values else 0,
        "p50Ms": percentile(0.50),
        "p95Ms": percentile(0.95),
        "maxMs": max(elapsed) if elapsed else None,
        "failureFamilies": failures,
    })
ok=bool(rows) and all(row["success"] == row["samples"] for row in rows)
payload={
  "ok": ok,
  "state": "succeeded" if ok else "failed",
  "message": "benchmark-complete" if ok else "benchmark-failed",
  "target": sys.argv[3],
  "scope": sys.argv[4],
  "profile": sys.argv[5],
  "proxy": {"mode": "on", "urlRedacted": sys.argv[6].replace("://", "://").split("@")[-1], "valuesPrinted": False},
  "source": {"sourceType": sys.argv[7], "sourceRef": sys.argv[8], "sourceConfigRef": sys.argv[9] or None, "sourceFingerprint": sys.argv[10] or None, "valuesPrinted": False},
  "rows": rows,
  "resultCount": len(items),
  "updatedAt": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
}
status_path.write_text(json.dumps(payload, ensure_ascii=False) + "\\n")
PY
}
run_job() {
  write_state running resolving-proxy
  rm -f "$state_dir/results.ndjson"
  proxy_ip="$(kubectl -n "$namespace" get svc "$service_name" -o 'jsonpath={.spec.clusterIP}' 2>/dev/null || true)"
  if [ -z "$proxy_ip" ]; then echo "egress proxy service missing: $namespace/$service_name" >&2; write_state failed proxy-service-missing; exit 41; fi
  proxy_url="http://$proxy_ip:$service_port"
  echo "benchmark target=$target_id scope=$scope profile=$profile proxy=$proxy_url sourceType=$source_type sourceRef=$source_ref sourceFingerprint=$source_fingerprint valuesPrinted=false"
	for sample in $(seq 1 "$samples"); do
	  write_state running docker-manifest
	  run_sample docker-manifest "$sample" none timeout "$sample_timeout" docker manifest inspect "$docker_image"
	  write_state running docker-pull
	  docker image rm -f "$docker_image" >/dev/null 2>&1 || true
	  cache_state=absent
	  docker image inspect "$docker_image" >/dev/null 2>&1 && cache_state=present || true
	  run_sample docker-pull "$sample" "$cache_state" timeout "$sample_timeout" docker pull "$docker_image"
	  write_state running docker-build
	  build_dir="$state_dir/build-$sample"
	  rm -rf "$build_dir"
	  mkdir -p "$build_dir"
	  printf 'FROM quay.io/prometheus/busybox:latest\\nRUN printf no-mirror-benchmark >/benchmark.txt\\n' >"$build_dir/Dockerfile"
    run_sample docker-build "$sample" no-cache timeout "$sample_timeout" docker build --pull --no-cache --network host --build-arg HTTP_PROXY="$proxy_url" --build-arg HTTPS_PROXY="$proxy_url" --build-arg ALL_PROXY="$proxy_url" --build-arg NO_PROXY="$no_proxy" -t "unidesk-egress-benchmark-$target_safe:$sample" "$build_dir"
    write_state running npm-install
    npm_dir="$state_dir/npm-$sample"
    rm -rf "$npm_dir"
    mkdir -p "$npm_dir/cache" "$npm_dir/pkg"
    printf '{"private":true,"dependencies":{}}\\n' >"$npm_dir/pkg/package.json"
    run_sample npm-install "$sample" temp-cache timeout "$sample_timeout" npm --prefix "$npm_dir/pkg" install --package-lock=false --no-save --ignore-scripts --no-audit --no-fund --omit=dev --registry=https://registry.npmjs.org/ --cache "$npm_dir/cache" --fetch-timeout 120000 --fetch-retries 2 yaml@2.8.3
  done
  summarize
}
run_job >>"$log" 2>&1 || {
  rc=$?
  summarize || true
  exit "$rc"
}
JOB
chmod +x "$state_dir/job.sh"
: >"$state_dir/job.log"
nohup "$state_dir/job.sh" >/dev/null 2>&1 &
pid=$!
printf '%s' "$pid" >"$state_dir/pid"
printf '{"started":true,"pid":%s,"stateDir":"%s","statusCommand":"%s","logsCommand":"%s"}\\n' "$pid" "$state_dir" ${shQuote(statusCommand(spec))} ${shQuote(logsCommand(spec))}
`;
}

export function egressBenchmarkStatusScript(spec: EgressBenchmarkSpec, tailLines: number): string {
  const stateDir = egressBenchmarkStateDir(spec);
  return `
set +e
state_dir=${shQuote(stateDir)}
status_file="$state_dir/status.json"
log_file="$state_dir/job.log"
pid_file="$state_dir/pid"
running=false
pid=null
if [ -s "$pid_file" ]; then
  pid_raw="$(cat "$pid_file" 2>/dev/null || true)"
  if [ -n "$pid_raw" ] && kill -0 "$pid_raw" >/dev/null 2>&1; then running=true; pid="$pid_raw"; else pid="$pid_raw"; fi
fi
python3 - "$state_dir" "$status_file" "$log_file" "$running" "$pid" ${shQuote(String(tailLines))} <<'PY'
import json, pathlib, sys
state_dir=pathlib.Path(sys.argv[1])
status_path=pathlib.Path(sys.argv[2])
log_path=pathlib.Path(sys.argv[3])
running=sys.argv[4] == "true"
pid=None if sys.argv[5] in ("", "null") else sys.argv[5]
tail_lines=int(sys.argv[6])
status=None
if status_path.exists():
    try:
        status=json.loads(status_path.read_text())
    except Exception as error:
        status={"parseError": str(error), "raw": status_path.read_text(errors="replace")[-1000:]}
if isinstance(status, dict) and not running and status.get("state") == "running":
    status["state"]="stopped"
    status["ok"]=False
    status["message"]="job-not-running"
log_tail=""
if log_path.exists():
    lines=log_path.read_text(errors="replace").splitlines()
    log_tail="\\n".join(lines[-tail_lines:])
results_path=state_dir/"results.ndjson"
items=[]
if results_path.exists():
    for line in results_path.read_text(errors="replace").splitlines():
        if not line.strip():
            continue
        try:
            items.append(json.loads(line))
        except Exception:
            pass
if items:
    by_test={}
    for item in items:
        by_test.setdefault(item.get("test", "unknown"), []).append(item)
    rows=[]
    for name in sorted(by_test):
        values=by_test[name]
        elapsed=sorted(int(item.get("elapsedMs", 0)) for item in values)
        failures={}
        for item in values:
            family=str(item.get("failureFamily", "unknown"))
            if family.startswith("exit-"):
                err_path=state_dir/(str(item.get("test", "unknown")) + "-" + str(item.get("sample", "")) + ".err")
                out_path=state_dir/(str(item.get("test", "unknown")) + "-" + str(item.get("sample", "")) + ".out")
                text=""
                for path in (err_path, out_path):
                    if path.exists():
                        text += path.read_text(errors="replace")[-2000:]
                lowered=text.lower()
                if "toomanyrequests" in lowered or "too many requests" in lowered or "rate limit" in lowered:
                    family="rate-limit"
            if family != "success":
                failures[family] = failures.get(family, 0) + 1
        def percentile(p):
            if not elapsed:
                return None
            idx=min(len(elapsed)-1, max(0, round((len(elapsed)-1)*p)))
            return elapsed[idx]
        rows.append({"test": name, "samples": len(values), "success": sum(1 for item in values if item.get("exitCode") == 0), "successRate": (sum(1 for item in values if item.get("exitCode") == 0) / len(values)) if values else 0, "p50Ms": percentile(0.50), "p95Ms": percentile(0.95), "maxMs": max(elapsed) if elapsed else None, "failureFamilies": failures})
    if status is None:
        status={"state": "running" if running else "unknown", "ok": False}
    status["rows"]=rows
    status["resultCount"]=len(items)
print(json.dumps({"stateDir": str(state_dir), "pid": pid, "running": running, "status": status, "logBytes": log_path.stat().st_size if log_path.exists() else 0, "logTail": log_tail}, ensure_ascii=False))
PY
`;
}

export function egressBenchmarkDryRun(spec: EgressBenchmarkSpec): Record<string, unknown> {
  return {
    profile: spec.profile,
    target: spec.targetId,
    route: spec.route,
    namespace: spec.namespace,
    serviceName: spec.serviceName,
    servicePort: spec.port,
	    samples: spec.samples,
	    sampleTimeoutSeconds: spec.sampleTimeoutSeconds,
	    tests: ["docker-manifest", "docker-pull", "docker-build", "npm-install"],
	    dockerImage: "quay.io/prometheus/busybox:latest",
	    forbiddenMirrors: ["DaoCloud mirror", "registry.npmmirror.com", "cnpm", "pnpm mirror", "preheated cache as success"],
    source: {
      sourceType: spec.sourceType,
      sourceRef: spec.sourceRef,
      sourceConfigRef: spec.sourceConfigRef,
      sourceFingerprint: spec.sourceFingerprint,
      valuesPrinted: false,
    },
    stateDir: egressBenchmarkStateDir(spec),
  };
}

export function egressBenchmarkCompactResult(result: CommandResult): Record<string, unknown> {
  return {
    exitCode: result.exitCode,
    stdoutBytes: Buffer.byteLength(result.stdout, "utf8"),
    stderrBytes: Buffer.byteLength(result.stderr, "utf8"),
    stdoutTail: result.stdout.slice(-2000),
    stderrTail: result.stderr.slice(-2000),
  };
}

export function statusCommand(spec: EgressBenchmarkSpec): string {
  if (spec.scope === "platform-infra") return `bun scripts/cli.ts platform-infra egress-proxy benchmark-status --target ${spec.targetId} --profile ${spec.profile}`;
  const [node, lane] = spec.targetId.split("/", 2);
  return `bun scripts/cli.ts hwlab nodes control-plane infra egress-benchmark status --node ${node} --lane ${lane} --profile ${spec.profile}`;
}

export function logsCommand(spec: EgressBenchmarkSpec): string {
  if (spec.scope === "platform-infra") return `bun scripts/cli.ts platform-infra egress-proxy benchmark-logs --target ${spec.targetId} --profile ${spec.profile}`;
  const [node, lane] = spec.targetId.split("/", 2);
  return `bun scripts/cli.ts hwlab nodes control-plane infra egress-benchmark logs --node ${node} --lane ${lane} --profile ${spec.profile}`;
}

function unique(values: readonly string[]): string[] {
  return [...new Set(values.filter((value) => value.length > 0))];
}

function safeSegment(value: string): string {
  return value.replace(/[^A-Za-z0-9._-]+/gu, "-").replace(/^-+|-+$/gu, "") || "target";
}

function shQuote(value: string): string {
  return `'${value.replaceAll("'", "'\"'\"'")}'`;
}