#!/bin/sh
# kafka — daily Apache Kafka / Amazon MSK client ops (POSIX sh).
# Deployed to ~/.infra/utils/kafka/kafka by `infra setup kafka`.
# Prefer --bootstrap-server; use --command-config for TLS/SASL/IAM.
set -eu

SCRIPT_DIR=$(CDPATH= cd -- "$(dirname -- "$0")" && pwd)
# shellcheck source=lib/common.sh
. "$SCRIPT_DIR/lib/common.sh"

# Defaults (override with flags / env). Apache Kafka 3.9.x + Scala 2.13.
KAFKA_DEFAULT_VERSION="${KAFKA_DEFAULT_VERSION:-3.9.1}"
KAFKA_DEFAULT_SCALA="${KAFKA_DEFAULT_SCALA:-2.13}"
MSK_IAM_DEFAULT_VERSION="${MSK_IAM_DEFAULT_VERSION:-2.3.2}"

cmd_download() {
  version=$KAFKA_DEFAULT_VERSION
  scala=$KAFKA_DEFAULT_SCALA
  while [ $# -gt 0 ]; do
    case "$1" in
      --version)
        [ $# -ge 2 ] || kafka_die "download --version needs a value"
        version=$2
        shift 2
        ;;
      --scala)
        [ $# -ge 2 ] || kafka_die "download --scala needs a value"
        scala=$2
        shift 2
        ;;
      -h|--help)
        printf '%s\n' "Usage: kafka download [--version VER] [--scala VER]" >&2
        exit 0
        ;;
      *)
        kafka_die "unknown download option: $1"
        ;;
    esac
  done

  kafka_require_java
  home=$(kafka_resolve_home)
  mkdir -p "$home"
  base="kafka_${scala}-${version}"
  tarball="${base}.tgz"
  # Prefer CDN; mirror list is documented by Apache.
  url="https://dlcdn.apache.org/kafka/${version}/${tarball}"
  alt="https://archive.apache.org/dist/kafka/${version}/${tarball}"
  tmpdir=$(mktemp -d "${TMPDIR:-/tmp}/kafka-dl.XXXXXX")
  trap 'rm -rf "$tmpdir"' EXIT INT TERM
  kafka_info "downloading $url"
  if ! kafka_fetch "$url" "$tmpdir/$tarball"; then
    kafka_info "CDN failed; trying archive.apache.org"
    kafka_fetch "$alt" "$tmpdir/$tarball" || kafka_die "download failed"
  fi
  # Best-effort checksum (CDN may 404 the .sha512 on some mirrors).
  if kafka_fetch "${url}.sha512" "$tmpdir/$tarball.sha512" 2>/dev/null \
    || kafka_fetch "${alt}.sha512" "$tmpdir/$tarball.sha512" 2>/dev/null; then
    kafka_sha512_ok "$tmpdir/$tarball" "$tmpdir/$tarball.sha512"
  else
    kafka_info "no .sha512 available; continuing without checksum"
  fi
  tar -xzf "$tmpdir/$tarball" -C "$tmpdir"
  parent=$(dirname -- "$home")
  mkdir -p "$parent"
  dest="$parent/$base"
  rm -rf "$dest"
  mv "$tmpdir/$base" "$dest"
  # Point KAFKA_HOME at the versioned tree (symlink when possible).
  if [ -e "$home" ] && [ ! -L "$home" ] && [ -d "$home/bin" ]; then
    kafka_info "installed $dest (existing $home kept; set KAFKA_HOME=$dest or: ln -sfn $dest $home)"
    trap - EXIT INT TERM
    rm -rf "$tmpdir"
    printf '%s\n' "$dest"
    return 0
  fi
  ln -sfn "$dest" "$home"
  trap - EXIT INT TERM
  rm -rf "$tmpdir"
  kafka_info "installed -> $home (-> $dest)"
  printf '%s\n' "$home"
}

cmd_iam_jar() {
  version=$MSK_IAM_DEFAULT_VERSION
  while [ $# -gt 0 ]; do
    case "$1" in
      --version)
        [ $# -ge 2 ] || kafka_die "iam-jar --version needs a value"
        version=$2
        shift 2
        ;;
      -h|--help)
        printf '%s\n' "Usage: kafka iam-jar [--version VER]" >&2
        exit 0
        ;;
      *)
        kafka_die "unknown iam-jar option: $1"
        ;;
    esac
  done
  kafka_require_home
  jar="aws-msk-iam-auth-${version}-all.jar"
  url="https://github.com/aws/aws-msk-iam-auth/releases/download/v${version}/${jar}"
  dest="$KAFKA_HOME/libs/$jar"
  mkdir -p "$KAFKA_HOME/libs"
  if [ -f "$dest" ]; then
    kafka_info "already present: $dest"
    printf '%s\n' "$dest"
    return 0
  fi
  tmp=$(mktemp "${TMPDIR:-/tmp}/msk-iam.XXXXXX")
  trap 'rm -f "$tmp"' EXIT INT TERM
  kafka_info "downloading $url"
  kafka_fetch "$url" "$tmp" || kafka_die "iam-jar download failed"
  mv "$tmp" "$dest"
  chmod 0644 "$dest"
  trap - EXIT INT TERM
  kafka_info "installed $dest"
  kafka_info "use conf/client.iam.properties (SASL_SSL + AWS_MSK_IAM)"
  printf '%s\n' "$dest"
}

cmd_env() {
  home=$(kafka_resolve_home)
  bs=$(kafka_resolve_bootstrap)
  cfg=$(kafka_resolve_command_config)
  # Human text on stderr; machine-readable exports on stdout (eval-safe).
  printf 'KAFKA_HOME=%s\n' "$home" >&2
  if [ -n "$bs" ]; then
    printf 'KAFKA_BOOTSTRAP_SERVERS=%s\n' "$bs" >&2
  else
    printf 'KAFKA_BOOTSTRAP_SERVERS=(unset)\n' >&2
  fi
  if [ -n "$cfg" ]; then
    printf 'KAFKA_COMMAND_CONFIG=%s\n' "$cfg" >&2
  else
    printf 'KAFKA_COMMAND_CONFIG=(unset)\n' >&2
  fi
  if [ -d "$home/bin" ]; then
    printf 'bin=ok\n' >&2
  else
    printf 'bin=missing (kafka download)\n' >&2
  fi
  printf 'export KAFKA_HOME=%s\n' "$home"
  if [ -n "$bs" ]; then
    printf 'export KAFKA_BOOTSTRAP_SERVERS=%s\n' "$bs"
  fi
  if [ -n "$cfg" ]; then
    printf 'export KAFKA_COMMAND_CONFIG=%s\n' "$cfg"
  fi
  printf 'export PATH="%s/bin:${PATH}"\n' "$home"
}

cmd_status() {
  kafka_run_admin kafka-broker-api-versions.sh "$@"
}

cmd_topics() {
  if [ $# -eq 0 ]; then
    cat <<'EOF' >&2
Usage: kafka topics <list|describe|create|delete|alter> …
   or: kafka topics --list   # passthrough to kafka-topics.sh

  list
  describe TOPIC [TOPIC…]
  create TOPIC [--partitions N] [--replication-factor N] [--config k=v]...
  delete TOPIC
  alter TOPIC --partitions N

Best practice: production RF=3 with min.insync.replicas=2; prefer
--bootstrap-server (not ZooKeeper).
EOF
    exit 1
  fi
  case "$1" in
    -h|--help)
      cmd_topics
      ;;
    list|--list)
      shift
      kafka_run_admin kafka-topics.sh --list "$@"
      ;;
    describe|--describe)
      shift
      if [ $# -eq 0 ]; then
        kafka_die "topics describe needs TOPIC"
      fi
      if [ "$1" = "--topic" ]; then
        kafka_run_admin kafka-topics.sh --describe "$@"
      fi
      # One describe per topic (kafka-topics.sh --topic is singular).
      failed=0
      for topic in "$@"; do
        case "$topic" in
          -*)
            kafka_die "topics describe: unexpected option $topic"
            ;;
        esac
        if ! kafka_invoke_admin kafka-topics.sh --describe --topic "$topic"; then
          failed=1
        fi
      done
      exit "$failed"
      ;;
    create|--create)
      shift
      [ $# -ge 1 ] || kafka_die "topics create needs TOPIC"
      topic=$1
      shift
      if [ $# -eq 0 ]; then
        # Defaults match common MSK/Kafka production guidance (RF=3).
        kafka_info "using defaults: --partitions 3 --replication-factor 3"
        kafka_run_admin kafka-topics.sh --create --topic "$topic" \
          --partitions 3 --replication-factor 3
      fi
      # Passthrough remaining kafka-topics.sh flags (--partitions, --config, …).
      case " $* " in
        *" --replication-factor 1 "*|*" --replication-factor 2 "*)
          kafka_info "warning: replication-factor < 3; production usually uses 3"
          ;;
      esac
      kafka_run_admin kafka-topics.sh --create --topic "$topic" "$@"
      ;;
    delete|--delete)
      shift
      [ $# -ge 1 ] || kafka_die "topics delete needs TOPIC"
      topic=$1
      shift
      kafka_run_admin kafka-topics.sh --delete --topic "$topic" "$@"
      ;;
    alter|--alter)
      shift
      [ $# -ge 1 ] || kafka_die "topics alter needs TOPIC"
      topic=$1
      shift
      kafka_run_admin kafka-topics.sh --alter --topic "$topic" "$@"
      ;;
    --*)
      # Passthrough (user already speaks kafka-topics.sh).
      kafka_run_admin kafka-topics.sh "$@"
      ;;
    *)
      kafka_die "unknown topics subcommand: $1 (list|describe|create|delete|alter)"
      ;;
  esac
}

cmd_configs() {
  if [ $# -eq 0 ]; then
    cat <<'EOF' >&2
Usage: kafka configs describe --topic TOPIC
       kafka configs set --topic TOPIC --add-config k=v[,k=v…]
       kafka configs set --topic TOPIC --delete-config k[,k…]

Entity defaults to topics. For brokers use kafka-configs.sh passthrough:
  kafka configs --entity-type brokers --entity-name 1 --describe
EOF
    exit 1
  fi
  case "$1" in
    -h|--help)
      cmd_configs
      ;;
    describe)
      shift
      entity_type=topics
      entity_name=
      while [ $# -gt 0 ]; do
        case "$1" in
          --topic)
            entity_type=topics
            entity_name=$2
            shift 2
            ;;
          --entity-type)
            entity_type=$2
            shift 2
            ;;
          --entity-name)
            entity_name=$2
            shift 2
            ;;
          *)
            kafka_die "unknown configs describe option: $1"
            ;;
        esac
      done
      [ -n "$entity_name" ] || kafka_die "configs describe needs --topic NAME"
      kafka_run_admin kafka-configs.sh --describe \
        --entity-type "$entity_type" --entity-name "$entity_name"
      ;;
    set|alter)
      shift
      entity_type=topics
      entity_name=
      add_config=
      delete_config=
      while [ $# -gt 0 ]; do
        case "$1" in
          --topic)
            entity_type=topics
            entity_name=$2
            shift 2
            ;;
          --entity-type)
            entity_type=$2
            shift 2
            ;;
          --entity-name)
            entity_name=$2
            shift 2
            ;;
          --add-config)
            add_config=$2
            shift 2
            ;;
          --delete-config)
            delete_config=$2
            shift 2
            ;;
          *)
            kafka_die "unknown configs set option: $1"
            ;;
        esac
      done
      [ -n "$entity_name" ] || kafka_die "configs set needs --topic NAME"
      if [ -n "$add_config" ]; then
        kafka_run_admin kafka-configs.sh --alter \
          --entity-type "$entity_type" --entity-name "$entity_name" \
          --add-config "$add_config"
      elif [ -n "$delete_config" ]; then
        kafka_run_admin kafka-configs.sh --alter \
          --entity-type "$entity_type" --entity-name "$entity_name" \
          --delete-config "$delete_config"
      else
        kafka_die "configs set needs --add-config or --delete-config"
      fi
      ;;
    --*)
      kafka_run_admin kafka-configs.sh "$@"
      ;;
    *)
      kafka_die "unknown configs subcommand: $1 (describe|set)"
      ;;
  esac
}

cmd_groups() {
  if [ $# -eq 0 ]; then
    cat <<'EOF' >&2
Usage: kafka groups list
       kafka groups describe GROUP
EOF
    exit 1
  fi
  case "$1" in
    list|--list)
      shift
      kafka_run_admin kafka-consumer-groups.sh --list "$@"
      ;;
    describe|--describe)
      shift
      [ $# -ge 1 ] || kafka_die "groups describe needs GROUP"
      group=$1
      shift
      kafka_run_admin kafka-consumer-groups.sh --describe --group "$group" "$@"
      ;;
    --*)
      kafka_run_admin kafka-consumer-groups.sh "$@"
      ;;
    *)
      kafka_die "unknown groups subcommand: $1 (list|describe)"
      ;;
  esac
}

cmd_consume() {
  [ $# -ge 1 ] || kafka_die "consume needs TOPIC"
  topic=$1
  shift
  kafka_require_java
  kafka_require_home
  bin=$(kafka_bin kafka-console-consumer.sh)
  bs=$(kafka_require_bootstrap)
  cfg=$(kafka_resolve_command_config)
  if [ -n "$cfg" ]; then
    exec "$bin" --bootstrap-server "$bs" --consumer.config "$cfg" --topic "$topic" "$@"
  fi
  exec "$bin" --bootstrap-server "$bs" --topic "$topic" "$@"
}

cmd_produce() {
  [ $# -ge 1 ] || kafka_die "produce needs TOPIC"
  topic=$1
  shift
  kafka_require_java
  kafka_require_home
  bin=$(kafka_bin kafka-console-producer.sh)
  bs=$(kafka_require_bootstrap)
  cfg=$(kafka_resolve_command_config)
  if [ -n "$cfg" ]; then
    exec "$bin" --bootstrap-server "$bs" --producer.config "$cfg" --topic "$topic" "$@"
  fi
  exec "$bin" --bootstrap-server "$bs" --topic "$topic" "$@"
}

# --- global option parse ---------------------------------------------------

while [ $# -gt 0 ]; do
  case "$1" in
    -b|--bootstrap|--bootstrap-server)
      [ $# -ge 2 ] || kafka_die "$1 needs a value"
      KAFKA_BOOTSTRAP_SERVERS=$2
      export KAFKA_BOOTSTRAP_SERVERS
      shift 2
      ;;
    -c|--command-config|--client-config)
      [ $# -ge 2 ] || kafka_die "$1 needs a value"
      KAFKA_COMMAND_CONFIG=$2
      export KAFKA_COMMAND_CONFIG
      shift 2
      ;;
    -H|--home)
      [ $# -ge 2 ] || kafka_die "$1 needs a value"
      KAFKA_HOME=$2
      export KAFKA_HOME
      shift 2
      ;;
    -h|--help)
      kafka_usage_global
      exit 0
      ;;
    --)
      shift
      break
      ;;
    -*)
      kafka_die "unknown option: $1 (try kafka help)"
      ;;
    *)
      break
      ;;
  esac
done

if [ $# -eq 0 ]; then
  kafka_usage_global
  exit 1
fi

command=$1
shift

case "$command" in
  help|-h|--help)
    kafka_usage_global
    exit 0
    ;;
  download|install)
    cmd_download "$@"
    ;;
  iam-jar|iam|msk-iam)
    cmd_iam_jar "$@"
    ;;
  env)
    cmd_env "$@"
    ;;
  status|brokers|api-versions)
    cmd_status "$@"
    ;;
  topics|topic)
    cmd_topics "$@"
    ;;
  configs|config)
    cmd_configs "$@"
    ;;
  groups|group|consumers)
    cmd_groups "$@"
    ;;
  consume|consumer)
    cmd_consume "$@"
    ;;
  produce|producer)
    cmd_produce "$@"
    ;;
  *)
    kafka_die "unknown command: $command (try kafka help)"
    ;;
esac
