#!/usr/bin/env bash
# spark-local-cluster — convenience wrapper around sbin/.
#
# Subcommands:
#   start     → sbin/start-all.sh
#   stop      → sbin/stop-all.sh
#   status    → process + listener summary
#   logs <n>  → tail -F a role log (driver|history|exec-N)
#
# Defaults (override via conf/spark-env.sh or env vars):
#   SPARK_EXECUTOR_INSTANCES=2
#   SPARK_EXECUTOR_MEMORY=2g
#   SPARK_EXECUTOR_CORES=2
#
# This is a thin convenience layer; all real work happens in sbin/.
set -euo pipefail
DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)"
SBIN="$DIR/sbin"
# shellcheck disable=SC1091
. "$SBIN/spark-config.sh"

cmd="${1:-}"; shift || true
case "$cmd" in
  start)  exec "$SBIN/start-all.sh" "$@" ;;
  stop)   exec "$SBIN/stop-all.sh"  "$@" ;;
  status)
    bold "==> spark-rust local cluster status"
    declare -A roles=(
      [history-server]="$SPARK_PID_DIR/spark-history-server.pid"
      [connect-server]="$SPARK_PID_DIR/spark-connect-server.pid"
    )
    up=0; want=0
    for name in history-server connect-server; do
      pf="${roles[$name]}"; want=$((want+1))
      if is_pid_alive "$pf"; then
        green "  ✓ $name (pid $(cat "$pf"))"; up=$((up+1))
      else
        red   "  ✗ $name"
      fi
    done
    shopt -s nullglob
    for pf in "$SPARK_PID_DIR"/spark-worker-*.pid; do
      idx=$(basename "$pf" .pid); idx="${idx#spark-worker-}"
      want=$((want+1))
      if is_pid_alive "$pf"; then
        green "  ✓ exec-$idx (pid $(cat "$pf"), :$((SPARK_WORKER_PORT_BASE + idx)))"; up=$((up+1))
      else
        red   "  ✗ exec-$idx"
      fi
    done
    bold "==> $up / $want processes up"
    if [ "$up" -gt 0 ]; then
      gray "  Connect gRPC : sc://localhost:$SPARK_CONNECT_PORT"
      gray "  Driver UI    : http://localhost:$SPARK_UI_PORT/"
      gray "  History UI   : http://localhost:$SPARK_HISTORY_UI_PORT/"
      n=$(curl -s --max-time 2 "http://127.0.0.1:$SPARK_UI_PORT/api/topology" 2>/dev/null \
          | grep -oE '"role":"executor"' | wc -l || true)
      gray "  /api/topology reports $n executor(s)"
    fi
    ;;
  logs)
    name="${1:?usage: spark-local-cluster logs <driver|history|exec-N>}"
    case "$name" in
      driver)  f="$SPARK_LOG_DIR/connect-server.log" ;;
      history) f="$SPARK_LOG_DIR/history-server.log" ;;
      exec-*)  f="$SPARK_LOG_DIR/$name.log" ;;
      *)       red "unknown log: $name"; exit 1 ;;
    esac
    [ -f "$f" ] || { red "no such log: $f"; exit 1; }
    exec tail -F "$f"
    ;;
  -h|--help|"")
    sed -n '2,18p' "$0"
    ;;
  *) red "unknown subcommand: $cmd"; exit 1 ;;
esac
