#!/usr/bin/env bash
# spark-submit compatibility launcher for spark-rust client-mode applications.
set -euo pipefail
ROOT="$(cd "$(dirname "$0")/.." && pwd)"
MASTER="${SPARK_REMOTE:-sc://127.0.0.1:15002}"
MAIN_CLASS=""
DRY_RUN=0

usage() {
  cat <<'EOF'
Usage: bin/spark-submit [options] <application> [application arguments]

Options:
  --master <sc://host:port>   spark-rust Spark Connect endpoint
  --class <main-class>        main class for a runnable/fat JAR
  --dry-run                   print the resolved command without executing
  --help                      show this help

Application types:
  *.jar   Java/Scala client-mode application (requires --class)
  *.rs    Rust source file (source checkout only; requires Cargo + SDK sources)
  *.ts    TypeScript source file (source checkout only; requires SDK + Node)
  other   prebuilt native executable

The application driver runs locally and submits relational work to spark-rust
over Spark Connect. Start target/release/spark-connect-server first, or point
--master at an existing spark-rust driver.
EOF
}

while (($#)); do
  case "$1" in
    --master) MASTER="${2:?--master requires a value}"; shift 2 ;;
    --class) MAIN_CLASS="${2:?--class requires a value}"; shift 2 ;;
    --dry-run) DRY_RUN=1; shift ;;
    --help|-h) usage; exit 0 ;;
    --conf) shift 2 ;; # accepted for CLI compatibility; app/server owns conf
    --*) echo "spark-submit: unsupported option: $1" >&2; exit 2 ;;
    *) break ;;
  esac
done

APP="${1:-}"
[[ -n "$APP" ]] || { usage >&2; exit 2; }
shift
[[ -e "$APP" ]] || { echo "spark-submit: application not found: $APP" >&2; exit 2; }
APP="$(realpath "$APP")"
export SPARK_REMOTE="$MASTER"

run() {
  printf 'SPARK_REMOTE=%q' "$SPARK_REMOTE"
  printf ' %q' "$@"
  printf '\n'
  ((DRY_RUN)) || exec "$@"
}

case "$APP" in
  *.jar)
    [[ -n "$MAIN_CLASS" ]] || { echo "spark-submit: --class is required for JAR applications" >&2; exit 2; }
    run java --add-opens=java.base/java.nio=ALL-UNNAMED -cp "$APP" "$MAIN_CLASS" "$@"
    ;;
  *.rs)
    [[ -f "$ROOT/Cargo.toml" && -d "$ROOT/crates/spark-rust" ]] || {
      echo "spark-submit: Rust source submission requires a developer source checkout; use a prebuilt native executable in the administrator distribution" >&2
      exit 2
    }
    TMP="$(mktemp -d "${TMPDIR:-/tmp}/spark-submit-rust.XXXXXX")"
    trap 'rm -rf "$TMP"' EXIT
    cat >"$TMP/Cargo.toml" <<EOF
[package]
name = "spark-submit-rust-app"
version = "0.1.0"
edition = "2021"

[[bin]]
name = "app"
path = "$APP"

[dependencies]
spark-rust = { path = "$ROOT/crates/spark-rust", features = ["connect"] }
tokio = { version = "1", features = ["macros", "rt-multi-thread"] }

[patch.crates-io]
tonic = { path = "$ROOT/../tonic" }
EOF
    run cargo run --quiet --manifest-path "$TMP/Cargo.toml" -- "$@"
    ;;
  *.ts)
    TSROOT="$ROOT/clients/typescript"
    [[ -d "$TSROOT/src" ]] || {
      echo "spark-submit: TypeScript source submission requires a developer source checkout; use a packaged JVM client or prebuilt executable" >&2
      exit 2
    }
    [[ -x "$TSROOT/node_modules/.bin/tsc" ]] || {
      echo "spark-submit: TypeScript dependencies missing; run: npm --prefix clients/typescript install" >&2
      exit 2
    }
    [[ -f "$TSROOT/dist/src/index.js" ]] || npm --prefix "$TSROOT" run build >/dev/null
    TMP="$(mktemp -d "${TMPDIR:-/tmp}/spark-submit-ts.XXXXXX")"
    trap 'rm -rf "$TMP"' EXIT
    mkdir -p "$TMP/node_modules/@spark-rust" "$TMP/node_modules/@types"
    ln -s "$TSROOT" "$TMP/node_modules/@spark-rust/connect"
    ln -s "$TSROOT/node_modules/@types/node" "$TMP/node_modules/@types/node"
    cp "$APP" "$TMP/app.ts"
    printf '{"type":"module"}\n' >"$TMP/package.json"
    cat >"$TMP/tsconfig.json" <<'EOF'
{
  "compilerOptions": {
    "target": "ES2022",
    "module": "NodeNext",
    "moduleResolution": "NodeNext",
    "outDir": "dist",
    "types": ["node"],
    "skipLibCheck": true,
    "strict": true
  },
  "include": ["app.ts"]
}
EOF
    "$TSROOT/node_modules/.bin/tsc" -p "$TMP/tsconfig.json"
    run node "$TMP/dist/app.js" "$@"
    ;;
  *)
    [[ -x "$APP" ]] || { echo "spark-submit: application is not executable: $APP" >&2; exit 2; }
    run "$APP" "$@"
    ;;
esac
