#!/usr/bin/env bash
# kafka-orphan-topics.sh
#
# Lists Kafka topics that have NO consumer groups referencing them.
# Internal topics (e.g. __consumer_offsets) are included.
# Output: CSV to stdout, or to OUTPUT_FILE if set.
#
# ─── Required ────────────────────────────────────────────────────────────────
#   KAFKA_BOOTSTRAP_SERVERS     e.g. "broker1:9092,broker2:9092"
#   KAFKA_SECURITY_PROTOCOL     PLAINTEXT | SSL | SASL_PLAINTEXT | SASL_SSL
#                               (default: PLAINTEXT)
# ─── Optional (tooling) ──────────────────────────────────────────────────────
#   KAFKA_HOME      root of Kafka installation (bin/ is appended)
#   OUTPUT_FILE     write CSV here instead of stdout

set -euo pipefail

# ─── helpers ─────────────────────────────────────────────────────────────────
die()  { printf '\nERROR: %s\n\n' "$*" >&2; exit 1; }
info() { printf 'INFO:  %s\n' "$*" >&2; }

# ─── locate kafka CLI tools ──────────────────────────────────────────────────
find_kafka_cmd() {
  local cmd="$1"         # e.g. kafka-topics.sh
  local bare="${cmd%.sh}" # e.g. kafka-topics

  if [[ -n "${KAFKA_HOME:-}" ]]; then
    local p="${KAFKA_HOME}/bin/${cmd}"
    [[ -x "$p" ]] || die "KAFKA_HOME is set but ${p} is not executable"
    echo "$p"; return
  fi
  for try in "$cmd" "$bare"; do
    if command -v "$try" &>/dev/null; then command -v "$try"; return; fi
  done
  for dir in /usr/local/kafka/bin /opt/kafka/bin /opt/confluent/bin \
              /usr/local/bin /usr/bin; do
    [[ -x "${dir}/${cmd}" ]] && { echo "${dir}/${cmd}"; return; }
  done
  die "Cannot find ${cmd}. Install Kafka CLI tools or set KAFKA_HOME."
}

KAFKA_TOPICS_CMD=$(find_kafka_cmd kafka-topics.sh)
KAFKA_GROUPS_CMD=$(find_kafka_cmd kafka-consumer-groups.sh)
info "kafka-topics:          ${KAFKA_TOPICS_CMD}"
info "kafka-consumer-groups: ${KAFKA_GROUPS_CMD}"

# ─── required env vars ───────────────────────────────────────────────────────
: "${KAFKA_BOOTSTRAP_SERVERS:?KAFKA_BOOTSTRAP_SERVERS is required}"

# ─── temp files ──────────────────────────────────────────────────────────────
_TMPDIR="${TMPDIR:-/tmp}"
CFG_FILE=$(mktemp "${_TMPDIR}/kafka-orphan-cfg.XXXXXX.properties")
GRP_TOPICS_FILE=$(mktemp "${_TMPDIR}/kafka-orphan-grp.XXXXXX")
PMAP_FILE=$(mktemp "${_TMPDIR}/kafka-orphan-pmap.XXXXXX")
ORPHANS_FILE=$(mktemp "${_TMPDIR}/kafka-orphan-list.XXXXXX")

cleanup() { rm -f "$CFG_FILE" "$GRP_TOPICS_FILE" "$PMAP_FILE" "$ORPHANS_FILE"; }
trap cleanup EXIT

# ─── build command.config ────────────────────────────────────────────────────
# PEM files embed their content inline in properties (Java interprets \n as newline)
pem_inline() { awk '{printf "%s\\n", $0}' "$1"; }

{
  PROTOCOL="${KAFKA_SECURITY_PROTOCOL:-PLAINTEXT}"
  echo "bootstrap.servers=${KAFKA_BOOTSTRAP_SERVERS}"
  echo "security.protocol=${PROTOCOL}"
} > "$CFG_FILE"

info "Command config: ${CFG_FILE}"

# ─── step 1: describe all topics → list + partition map ──────────────────────
info "Describing all topics (including internal)..."

# Probe whether --include-internal is supported (added in Kafka 2.1).
# { || true } prevents set -o pipefail from treating --help's non-zero exit as a failure.
INCLUDE_INT_FLAG=""
{ "$KAFKA_TOPICS_CMD" --help 2>&1 || true; } | grep -q -- '--include-internal' \
  && INCLUDE_INT_FLAG="--include-internal"

# shellcheck disable=SC2086
ALL_DESCRIBE=$("$KAFKA_TOPICS_CMD" \
  --bootstrap-server "$KAFKA_BOOTSTRAP_SERVERS" \
  --command-config "$CFG_FILE" \
  --describe \
  --include-internal 2>/dev/null) \
  || die "Failed to describe topics. Verify KAFKA_BOOTSTRAP_SERVERS and credentials."

# Summary lines look like (tab-separated):
#   Topic: <name>\tTopicId: ...\tPartitionCount: N\tReplicationFactor: M\tConfigs: ...
# Per-partition lines have "Partition:" not "PartitionCount:" — we skip them.
LC_ALL=C sort > "$PMAP_FILE" < <(
  printf '%s\n' "$ALL_DESCRIBE" \
  | awk -F'\t' '
      /PartitionCount:/ {
        name  = $1; sub(/^Topic: /, "", name)
        parts = $3; sub(/^PartitionCount: /, "", parts)
        print name "," parts
      }'
)

TOTAL_TOPICS=$(wc -l < "$PMAP_FILE" | tr -d ' ')
info "Total topics discovered: ${TOTAL_TOPICS}"
[[ "$TOTAL_TOPICS" -eq 0 ]] && { info "No topics found — nothing to do."; echo "topic,partitions"; exit 0; }

# ─── step 2: collect topics referenced by any consumer group ─────────────────
info "Listing consumer groups..."
GROUPS=$("$KAFKA_GROUPS_CMD" \
  --bootstrap-server "$KAFKA_BOOTSTRAP_SERVERS" \
  --command-config "$CFG_FILE" \
  --list 2>/dev/null) \
  || die "Failed to list consumer groups. Verify connectivity and credentials."

GROUP_COUNT=$(printf '%s\n' "$GROUPS" | grep -c '[^[:space:]]' || echo 0)
info "Consumer groups found: ${GROUP_COUNT}"

# Parse describe output: locate the TOPIC column from the header, then extract it.
_parse_group_topics() {
  awk '
    # Each group block may have its own header line — re-detect column on every header.
    $1 == "GROUP" {
      topic_col = 0
      for (i = 1; i <= NF; i++) if ($i == "TOPIC") { topic_col = i; break }
      next
    }
    # Skip blank lines and placeholder-only lines
    NF == 0 { next }
    topic_col > 0 && $topic_col != "" && $topic_col != "-" { print $topic_col }
  '
}

if [[ "$GROUP_COUNT" -eq 0 ]]; then
  info "No consumer groups — all topics are orphans."
  : > "$GRP_TOPICS_FILE"  # empty
elif { "$KAFKA_GROUPS_CMD" --help 2>&1 || true; } | grep -q -- '--all-groups'; then
  # Single API call (Kafka 2.6+)
  info "Using --all-groups for a single describe call..."
  "$KAFKA_GROUPS_CMD" \
    --bootstrap-server "$KAFKA_BOOTSTRAP_SERVERS" \
    --command-config "$CFG_FILE" \
    --describe \
    --all-groups 2>/dev/null \
  | _parse_group_topics \
  | LC_ALL=C sort -u > "$GRP_TOPICS_FILE"
else
  # Per-group describe loop (older Kafka — one JVM call per group)
  info "Describing consumer groups individually (Kafka < 2.6 — may be slow for many groups)..."
  {
    while IFS= read -r group; do
      [[ -z "${group//[[:space:]]/}" ]] && continue
      "$KAFKA_GROUPS_CMD" \
        --bootstrap-server "$KAFKA_BOOTSTRAP_SERVERS" \
        --command-config "$CFG_FILE" \
        --describe \
        --group "$group" 2>/dev/null || true
    done <<< "$GROUPS"
  } | _parse_group_topics \
    | LC_ALL=C sort -u > "$GRP_TOPICS_FILE"
fi

GRP_TOPIC_COUNT=$(wc -l < "$GRP_TOPICS_FILE" | tr -d ' ')
info "Topics referenced by consumer groups: ${GRP_TOPIC_COUNT}"

# ─── step 3: orphan topics = all topics minus topics with groups ──────────────
info "Computing orphan topics..."

# Both inputs must be sorted (they are — PMAP_FILE by LC_ALL=C sort, GRP_TOPICS_FILE by sort -u)
LC_ALL=C comm -23 \
  <(cut -d, -f1 "$PMAP_FILE") \
  "$GRP_TOPICS_FILE" \
> "$ORPHANS_FILE"

ORPHAN_COUNT=$(wc -l < "$ORPHANS_FILE" | tr -d ' ')
info "Orphan topics (no consumer group): ${ORPHAN_COUNT}"

# ─── step 4: emit CSV ────────────────────────────────────────────────────────
# ORPHANS_FILE: one topic per line (sorted)
# PMAP_FILE:    topic,partitions (sorted on topic)
# join -t, -o '1.1,2.2' produces: topic,partitions

OUTPUT="${OUTPUT_FILE:-/dev/stdout}"
{
  echo "topic,partitions"
  if [[ "$ORPHAN_COUNT" -gt 0 ]]; then
    # join requires both inputs sorted on the join key (field 1, comma-delimited)
    LC_ALL=C join -t, -a 1 -e "unknown" -o '1.1,2.2' \
      "$ORPHANS_FILE" \
      "$PMAP_FILE"
  fi
} > "$OUTPUT"

[[ "${OUTPUT}" != "/dev/stdout" ]] && info "CSV written to: ${OUTPUT}"
info "Done. ${ORPHAN_COUNT} orphan topic(s) out of ${TOTAL_TOPICS} total."