#!/usr/bin/env bash
# Runs INSIDE the Graylog LXC container as root (invoked via `pct exec <vmid> -- bash install-graylog.sh`).
# Idempotent: safe to re-run after a partial failure.
set -euo pipefail

# The container inherits LANG=en_US.UTF-8 from the calling shell (pct exec),
# but that locale is never generated here - just noisy "Setting locale
# failed" warnings from perl/apt-listchanges on every apt-get call. C.UTF-8
# is glibc-builtin (no locale-gen needed) and silences them.
export LC_ALL=C.UTF-8 LANG=C.UTF-8

SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
INSTALL_DIR="/opt/graylog"
GRAYLOG_EXTERNAL_URI="${GRAYLOG_EXTERNAL_URI:?Set GRAYLOG_EXTERNAL_URI, e.g. http://93.171.241.5:9000/}"
DISCORD_WEBHOOK_URL="${DISCORD_WEBHOOK_URL:-}"
ADMIN_USER="admin"

# Colors only when stderr is an actual terminal (never garble log files/pipes).
if [ -t 2 ]; then
  C_RESET=$'\033[0m'; C_CYAN=$'\033[36m'; C_GREEN=$'\033[32m'; C_YELLOW=$'\033[33m'; C_RED=$'\033[1;31m'
else
  C_RESET=''; C_CYAN=''; C_GREEN=''; C_YELLOW=''; C_RED=''
fi

log()  { echo "${C_CYAN}[install-graylog]${C_RESET} $*" >&2; }
ok()   { echo "${C_GREEN}[install-graylog] ✓${C_RESET} $*" >&2; }
skip() { echo "${C_YELLOW}[install-graylog] ⏭${C_RESET} $*" >&2; }
die()  { echo "${C_RED}[install-graylog] ✗ ERROR:${C_RESET} $*" >&2; exit 1; }

require_root() {
  [ "$(id -u)" -eq 0 ] || die "must run as root inside the container (use pct exec)"
}

# ---------------------------------------------------------------------------
step_packages() {
  log "Installing base packages (curl, gnupg, python3, jq, nftables)..."
  export DEBIAN_FRONTEND=noninteractive
  apt-get update -qq
  apt-get install -y -qq ca-certificates curl gnupg python3 jq nftables >/dev/null
}

step_disable_ipv6() {
  # Workaround: on this hosting network, IPv6 paths to some registries
  # (docker.io) get intercepted and served a mismatched TLS cert. Disabling
  # IPv6 in the container forces IPv4-only egress and avoids it.
  if [ "$(cat /proc/sys/net/ipv6/conf/all/disable_ipv6 2>/dev/null || echo 0)" = "1" ]; then
    skip "IPv6 already disabled"
    return
  fi
  log "Disabling IPv6 (registry TLS workaround for this network)..."
  cat > /etc/sysctl.d/99-disable-ipv6.conf <<'EOF'
net.ipv6.conf.all.disable_ipv6 = 1
net.ipv6.conf.default.disable_ipv6 = 1
EOF
  sysctl -p /etc/sysctl.d/99-disable-ipv6.conf >/dev/null
  ok "IPv6 disabled"
}

step_docker_install() {
  if command -v docker >/dev/null 2>&1; then
    skip "Docker already installed ($(docker --version))"
    return
  fi
  log "Installing Docker Engine from the official apt repo..."
  install -m 0755 -d /etc/apt/keyrings
  curl -fsSL https://download.docker.com/linux/debian/gpg -o /etc/apt/keyrings/docker.asc
  chmod a+r /etc/apt/keyrings/docker.asc
  . /etc/os-release
  echo "deb [arch=$(dpkg --print-architecture) signed-by=/etc/apt/keyrings/docker.asc] https://download.docker.com/linux/debian $VERSION_CODENAME stable" \
    > /etc/apt/sources.list.d/docker.list
  apt-get update -qq
  apt-get install -y -qq docker-ce docker-ce-cli containerd.io docker-buildx-plugin docker-compose-plugin >/dev/null
  systemctl enable --now docker >/dev/null
  ok "Docker Engine installed"
}

step_check_max_map_count() {
  local current
  current="$(cat /proc/sys/vm/max_map_count)"
  if [ "$current" -lt 262144 ]; then
    die "vm.max_map_count=$current (<262144), required by OpenSearch. This is a HOST-level kernel \
parameter and cannot be changed from inside an unprivileged LXC. Ask the Proxmox host admin to run: \
  echo 'vm.max_map_count=262144' >> /etc/sysctl.d/99-opensearch.conf && sysctl -p /etc/sysctl.d/99-opensearch.conf \
Then re-run this script."
  fi
  ok "vm.max_map_count=$current (>= 262144)"
}

step_docker_smoke_test() {
  log "Testing Docker can actually run a container (AppArmor check)..."
  if docker run --rm hello-world >/tmp/docker-test.log 2>&1; then
    ok "Docker smoke test passed"
    return
  fi
  if grep -q "ip_unprivileged_port_start" /tmp/docker-test.log; then
    die "Docker containers fail to start due to unprivileged-LXC AppArmor confinement. \
This must be fixed on the PROXMOX HOST (not from inside the container) by the host admin: \
  echo 'lxc.apparmor.profile: unconfined' >> /etc/pve/lxc/<VMID>.conf && pct reboot <VMID> \
Then re-run this script."
  fi
  cat /tmp/docker-test.log >&2
  die "docker run hello-world failed for an unrecognized reason; see output above."
}

# ---------------------------------------------------------------------------
step_compose_files() {
  mkdir -p "$INSTALL_DIR"
  cp "$SCRIPT_DIR/docker-compose.yml" "$INSTALL_DIR/docker-compose.yml"

  # Same timezone detection used by step_inputs() for Syslog input parsing.
  # GRAYLOG_ROOT_TIMEZONE governs the display timezone of the built-in
  # read-only "admin" user - it CANNOT be changed via the Users REST API
  # (confirmed live: PUT /api/users/admin with a timezone field fails with
  # "state should be: hexString has 24 characters"), only via this env var.
  local tz
  tz="$(cat /etc/timezone 2>/dev/null || echo UTC)"

  if [ -f "$INSTALL_DIR/.env" ]; then
    if grep -q '^GRAYLOG_ROOT_TIMEZONE=' "$INSTALL_DIR/.env"; then
      skip ".env already exists, keeping existing secrets"
    else
      echo "GRAYLOG_ROOT_TIMEZONE=$tz" >> "$INSTALL_DIR/.env"
      ok ".env existed but was missing GRAYLOG_ROOT_TIMEZONE - added it ($tz)"
    fi
    return
  fi
  log "Generating fresh secrets into $INSTALL_DIR/.env ..."
  local secret admin_pass admin_sha2
  secret="$(openssl rand -hex 48)"
  # GRAYLOG_ADMIN_PASSWORD lets the operator pin a known password up front
  # instead of always getting a random one - useful for CI, where it means
  # the Forgejo secret is set once and never goes stale, rather than having
  # to be re-synced every time a fresh install randomly generates a new
  # password. Falls back to random if not set, same as before.
  admin_pass="${GRAYLOG_ADMIN_PASSWORD:-$(openssl rand -base64 18 | tr -dc 'A-Za-z0-9' | head -c20)}"
  admin_sha2="$(echo -n "$admin_pass" | sha256sum | cut -d' ' -f1)"

  cat > "$INSTALL_DIR/.env" <<EOF
GRAYLOG_PASSWORD_SECRET=$secret
GRAYLOG_ROOT_PASSWORD_SHA2=$admin_sha2
GRAYLOG_HTTP_EXTERNAL_URI=$GRAYLOG_EXTERNAL_URI
GRAYLOG_ROOT_TIMEZONE=$tz
EOF
  chmod 600 "$INSTALL_DIR/.env"

  # Only shown once, at generation time - not re-printed on later runs.
  cat > "$INSTALL_DIR/.admin_credentials_ONE_TIME" <<EOF
Graylog admin user: $ADMIN_USER
Graylog admin password: $admin_pass
(This file is only written once, at first install. install-graylog.sh prints
the password above and deletes this file automatically once the run
finishes successfully. If the script dies before that, store it yourself,
then remove the file: rm $INSTALL_DIR/.admin_credentials_ONE_TIME)
EOF
  chmod 600 "$INSTALL_DIR/.admin_credentials_ONE_TIME"
  ok "Admin password generated (shown and deleted at the end of this run if the whole script succeeds)."
}

step_compose_up() {
  log "Pulling images (retrying up to 5x - this registry intermittently serves a bad TLS cert)..."
  local i
  for i in 1 2 3 4 5; do
    if (cd "$INSTALL_DIR" && docker compose pull); then
      break
    fi
    log "Pull attempt $i failed, retrying..."
    sleep 3
    [ "$i" -eq 5 ] && die "docker compose pull failed after 5 attempts."
  done

  log "Starting the stack..."
  (cd "$INSTALL_DIR" && docker compose up -d)

  log "Waiting for graylog-server to become healthy (up to 5 minutes)..."
  local waited=0
  while true; do
    status="$(docker inspect --format '{{.State.Health.Status}}' graylog-server 2>/dev/null || echo starting)"
    [ "$status" = "healthy" ] && break
    waited=$((waited + 5))
    [ "$waited" -ge 300 ] && die "graylog-server did not become healthy within 5 minutes. Check: docker logs graylog-server"
    sleep 5
  done
  ok "Stack is healthy"
}

# ---------------------------------------------------------------------------
GRAYLOG_API="http://localhost:9000/api"

# Resolves the admin password into the global ADMIN_PASSWORD exactly once.
# Must be called as a plain statement (never via `x=$(resolve_admin_password)`)
# so that `die`'s `exit` terminates the real script instead of a subshell.
ADMIN_PASSWORD=""
resolve_admin_password() {
  [ -n "$ADMIN_PASSWORD" ] && return
  if [ -f "$INSTALL_DIR/.admin_credentials_ONE_TIME" ]; then
    ADMIN_PASSWORD="$(grep -m1 -oP '(?<=^Graylog admin password: ).*' "$INSTALL_DIR/.admin_credentials_ONE_TIME")"
  elif [ -n "${GRAYLOG_ADMIN_PASSWORD:-}" ]; then
    ADMIN_PASSWORD="$GRAYLOG_ADMIN_PASSWORD"
  else
    die "Cannot find admin password: $INSTALL_DIR/.admin_credentials_ONE_TIME was deleted. \
Re-run with GRAYLOG_ADMIN_PASSWORD=<password> set."
  fi
}

gcurl() {
  local method="$1" path="$2" data="${3:-}"
  resolve_admin_password
  if [ -n "$data" ]; then
    curl -s -u "${ADMIN_USER}:${ADMIN_PASSWORD}" -H 'X-Requested-By: cli' -H 'Content-Type: application/json' \
      -X "$method" "${GRAYLOG_API}${path}" -d "$data"
  else
    curl -s -u "${ADMIN_USER}:${ADMIN_PASSWORD}" -H 'X-Requested-By: cli' -X "$method" "${GRAYLOG_API}${path}"
  fi
}

wait_for_api_ready() {
  # Docker's healthcheck can report the container "healthy" a few seconds
  # before Graylog's REST API is actually ready to serve authenticated
  # requests (indices/auth subsystems still initializing) - confirmed live:
  # the very first gcurl call (step_index_retention) intermittently got an
  # empty response, which then crashed the downstream `python3 -c
  # "json.load(sys.stdin)"` with "Expecting value: line 1 column 1".
  log "Waiting for the Graylog REST API to accept authenticated requests..."
  local waited=0
  while true; do
    gcurl GET /system/indices/index_sets | python3 -c "import json,sys;json.load(sys.stdin)" 2>/dev/null && break
    waited=$((waited + 5))
    [ "$waited" -ge 180 ] && die "Graylog REST API did not respond with valid JSON within 180s of the container reporting healthy."
    sleep 5
  done
  ok "Graylog REST API is ready"
}

step_index_retention() {
  # Graylog's factory default (as of 7.1) keeps 30-40 days of data across up
  # to 20 indices - reasonable in general, but risky on a small disk (this
  # deployment: 50GB) combined with unverified log volume (some sources,
  # e.g. accel-ppp at debug level, can be very chatty). Tightened to a
  # 14-21 day window / 15 indices for a larger safety margin. Revisit once
  # real production volume has been observed for a few weeks.
  log "Tightening default index set retention (idempotent)..."
  local idx_id current
  idx_id="$(gcurl GET /system/indices/index_sets | python3 -c "import json,sys;d=json.load(sys.stdin);print(next(s['id'] for s in d['index_sets'] if s.get('default')))")"
  current="$(gcurl GET "/system/indices/index_sets/$idx_id")"
  if echo "$current" | python3 -c "import json,sys;d=json.load(sys.stdin);sys.exit(0 if d['rotation_strategy']['index_lifetime_max']=='P21D' else 1)" 2>/dev/null; then
    skip "index retention already tightened (P14D/P21D, max 15 indices)"
    return
  fi
  gcurl PUT "/system/indices/index_sets/$idx_id" "$(python3 -c "
import json
d = json.loads('''$current''')
d['rotation_strategy']['index_lifetime_min'] = 'P14D'
d['rotation_strategy']['index_lifetime_max'] = 'P21D'
d['retention_strategy']['max_number_of_indices'] = 15
d['data_tiering']['index_lifetime_min'] = 'P14D'
d['data_tiering']['index_lifetime_max'] = 'P21D'
d['writable'] = True
print(json.dumps(d))
")" >/dev/null
  ok "index retention tightened to P14D/P21D, max 15 indices"
}

step_inputs() {
  log "Creating Syslog UDP inputs (idempotent)..."
  local existing tz
  existing="$(gcurl GET /system/inputs)"

  # RFC3164 syslog (which is what most network gear/accel-ppp send) has no
  # timezone in its timestamp ("Jul 22 09:15:13"). Without this setting,
  # Graylog defaults to treating that bare timestamp as UTC - so a device
  # logging in local Kyiv time (UTC+3) shows up 3 hours in the future.
  # Confirmed live: a raw test packet with "Jul 22 09:15:13" was stored as
  # 09:15:13Z UTC (wrong) until this was set; afterwards it correctly
  # stored as 06:15:13Z UTC (09:15:13 Kyiv time). Uses the container's own
  # configured timezone so it stays correct regardless of where this is
  # deployed.
  tz="$(cat /etc/timezone 2>/dev/null || echo UTC)"

  ensure_syslog_input() {
    local port="$1" title="$2" var_name="$3"
    local id current_tz
    id="$(echo "$existing" | python3 -c "import json,sys;d=json.load(sys.stdin);print(next((i['id'] for i in d['inputs'] if i['attributes'].get('port')==$port),''))")"
    local body="{
      \"title\": \"$title\",
      \"type\": \"org.graylog2.inputs.syslog.udp.SyslogUDPInput\",
      \"global\": true,
      \"configuration\": {\"bind_address\":\"0.0.0.0\",\"port\":$port,\"recv_buffer_size\":262144,
        \"number_worker_threads\":2,\"force_rdns\":false,\"allow_override_date\":true,
        \"store_full_message\":false,\"expand_structured_data\":true,\"charset_name\":\"UTF-8\",
        \"timezone\":\"$tz\"}
    }"
    if [ -z "$id" ]; then
      id="$(gcurl POST /system/inputs "$body" | python3 -c "import json,sys;print(json.load(sys.stdin)['id'])")"
      ok "Created $title: $id"
    else
      current_tz="$(echo "$existing" | python3 -c "import json,sys;d=json.load(sys.stdin);print(next((i['attributes'].get('timezone') for i in d['inputs'] if i['attributes'].get('port')==$port),''))")"
      if [ "$current_tz" != "$tz" ]; then
        gcurl PUT "/system/inputs/$id" "$body" >/dev/null
        ok "$title already existed - fixed timezone ($current_tz -> $tz)"
      else
        skip "$title already exists: $id"
      fi
    fi
    printf -v "$var_name" '%s' "$id"
  }

  # Port 514 is the standard syslog port and the one most network gear
  # actually sends to (confirmed live: BDCOM switches here can't be pointed
  # at a custom port). Port 1514 is kept as a secondary input for any
  # equipment that *can* be configured with a non-standard destination port.
  ensure_syslog_input 514 "Network Equipment Syslog (standard port 514)" NETWORK_INPUT_ID_514
  ensure_syslog_input 1514 "Network Equipment Syslog (Juniper-ZTE-DLink)" NETWORK_INPUT_ID_1514
  ensure_syslog_input 5140 "Servers Syslog (RADIUS-accel-ppp)" SERVERS_INPUT_ID
}

step_pipeline_rules() {
  log "Importing pipeline rules from $SCRIPT_DIR/rules/*.json (idempotent)..."
  # Compares by 'source' (not just title) so a rule whose regex/logic
  # changed under the same title gets PUT-updated instead of silently
  # skipped - this bit us live this session (rule11's session-correlation
  # regex, rule18's severity fix) before this check existed.
  local existing
  existing="$(gcurl GET /system/pipelines/rule)"

  local f title id source current_source result
  for f in "$SCRIPT_DIR"/rules/*.json; do
    title="$(python3 -c "import json;print(json.load(open('$f'))['title'])")"
    source="$(python3 -c "import json;print(json.load(open('$f'))['source'])")"
    id="$(echo "$existing" | python3 -c "import json,sys;d=json.load(sys.stdin);print(next((r['id'] for r in d if r['title']=='$title'),''))")"
    if [ -n "$id" ]; then
      current_source="$(echo "$existing" | python3 -c "import json,sys;d=json.load(sys.stdin);print(next((r['source'] for r in d if r['id']=='$id'),''))")"
      if [ "$current_source" = "$source" ]; then
        skip "rule '$title' already up to date"
        continue
      fi
      result="$(gcurl PUT "/system/pipelines/rule/$id" "$(cat "$f")")"
      if echo "$result" | python3 -c "import json,sys;d=json.load(sys.stdin);sys.exit(0 if d.get('errors') is None else 1)"; then
        ok "updated rule '$title' ($id) - source changed"
      else
        die "rule '$title' failed to compile: $result"
      fi
      continue
    fi
    result="$(gcurl POST /system/pipelines/rule "$(cat "$f")")"
    if echo "$result" | python3 -c "import json,sys;d=json.load(sys.stdin);sys.exit(0 if d.get('errors') is None else 1)"; then
      ok "created rule '$title'"
    else
      die "rule '$title' failed to compile: $result"
    fi
  done
}

step_pipelines() {
  log "Importing pipelines from $SCRIPT_DIR/pipelines/*.json (idempotent)..."
  # Same "compare by content, PUT-update if changed" fix as
  # step_pipeline_rules() - an existing pipeline's title never changes
  # even when its stage list gains new rules, so a plain title-exists
  # check would silently skip the update forever. Confirmed live this
  # session: 5 new rules were added to rules/*.json and created fine in
  # Graylog, but "Servers Parsing" kept its old stage list since this
  # function only ever checked title existence before this fix.
  local existing
  existing="$(gcurl GET /system/pipelines/pipeline)"

  local f title id source current_source result
  for f in "$SCRIPT_DIR"/pipelines/*.json; do
    title="$(python3 -c "import json;print(json.load(open('$f'))['title'])")"
    source="$(python3 -c "import json;print(json.load(open('$f'))['source'])")"
    id="$(echo "$existing" | python3 -c "import json,sys;d=json.load(sys.stdin);print(next((p['id'] for p in d if p['title']=='$title'),''))")"
    if [ -n "$id" ]; then
      current_source="$(echo "$existing" | python3 -c "import json,sys;d=json.load(sys.stdin);print(next((p['source'] for p in d if p['id']=='$id'),''))")"
      if [ "$current_source" = "$source" ]; then
        skip "pipeline '$title' already up to date ($id)"
      else
        gcurl PUT "/system/pipelines/pipeline/$id" "$(cat "$f")" >/dev/null
        ok "updated pipeline '$title' ($id) - stage list changed"
      fi
      if [ "$title" = "Network Equipment Parsing" ]; then NETWORK_PIPELINE_ID="$id"; fi
      if [ "$title" = "Servers Parsing" ]; then SERVERS_PIPELINE_ID="$id"; fi
      continue
    fi
    result="$(gcurl POST /system/pipelines/pipeline "$(cat "$f")")"
    id="$(echo "$result" | python3 -c "import json,sys;print(json.load(sys.stdin).get('id',''))")"
    [ -n "$id" ] || die "pipeline '$title' failed to create: $result"
    ok "created pipeline '$title' ($id)"
    if [ "$title" = "Network Equipment Parsing" ]; then NETWORK_PIPELINE_ID="$id"; fi
    if [ "$title" = "Servers Parsing" ]; then SERVERS_PIPELINE_ID="$id"; fi
  done
}

step_streams() {
  log "Creating streams from $SCRIPT_DIR/streams/*.json (idempotent)..."
  # Same "compare content, don't just check the title exists" fix as
  # step_pipeline_rules()/step_pipelines()/step_alerts()/step_dashboard() - a
  # stream's title never changes when its rules do, so a plain title-exists
  # check would silently skip the update forever. Unlike a dashboard, a
  # stream's rules live in their own sub-resource
  # (POST/PUT/DELETE /streams/{id}/rules/{ruleId}), not a single comparable
  # field - and delete+recreating the whole stream would briefly break live
  # message routing (alerts/dashboards go blind until it's reconnected to
  # its pipeline). So changed rules are diffed individually - identity is
  # (field, type, value, inverted) since desired rules carry no id - missing
  # ones are created, extra ones are deleted; title/description/matching_type/
  # index_set_id are PUT-updated in place same as a pipeline's 'source'.
  local index_set_id existing_file
  index_set_id="$(gcurl GET /system/indices/index_sets | python3 -c "import json,sys;d=json.load(sys.stdin);print(next(s['id'] for s in d['index_sets'] if s.get('default')))")"
  existing_file="$(mktemp)"
  gcurl GET /streams > "$existing_file"

  # $2.. are extra `sed -e` substitution expressions applied on top of the
  # index-set-id one, so each template can carry however many input-id
  # placeholders it needs.
  render_and_create() {
    local tmpl="$1" title="$2"; shift 2
    local sed_args=(-e "s/__DEFAULT_INDEX_SET_ID__/$index_set_id/")
    local expr
    for expr in "$@"; do sed_args+=(-e "$expr"); done
    local desired_file id
    desired_file="$(mktemp)"
    sed "${sed_args[@]}" "$tmpl" > "$desired_file"

    id="$(python3 -c "
import json
d = json.load(open('$existing_file'))
print(next((s['id'] for s in d['streams'] if s['title'] == '$title'), ''))
")"

    if [ -n "$id" ]; then
      local plan_file
      plan_file="$(mktemp)"
      python3 -c "
import json

d = json.load(open('$existing_file'))
current = next(s for s in d['streams'] if s['id'] == '$id')
desired = json.load(open('$desired_file'))['entity']

meta_keys = ['title', 'description', 'matching_type', 'index_set_id', 'remove_matches_from_default_stream']
metadata_changed = any(current.get(k) != desired.get(k) for k in meta_keys)
metadata_body = {k: desired[k] for k in meta_keys if k in desired}

def rule_key(r):
    return (r['field'], r['type'], r['value'], r['inverted'])

current_rules = current.get('rules') or []
desired_rules = desired.get('rules') or []
current_by_key = {rule_key(r): r['id'] for r in current_rules}
desired_key_set = {rule_key(r) for r in desired_rules}

rules_create = [r for r in desired_rules if rule_key(r) not in current_by_key]
rules_delete = [rid for k, rid in current_by_key.items() if k not in desired_key_set]

json.dump({
    'metadata_changed': metadata_changed,
    'metadata_body': metadata_body,
    'rules_create': rules_create,
    'rules_delete': rules_delete,
}, open('$plan_file', 'w'))
"
      local metadata_changed rules_create_count rules_delete_count
      metadata_changed="$(python3 -c "import json;print(json.load(open('$plan_file'))['metadata_changed'])")"
      rules_create_count="$(python3 -c "import json;print(len(json.load(open('$plan_file'))['rules_create']))")"
      rules_delete_count="$(python3 -c "import json;print(len(json.load(open('$plan_file'))['rules_delete']))")"

      if [ "$metadata_changed" = "True" ]; then
        gcurl PUT "/streams/$id" "$(python3 -c "import json;print(json.dumps(json.load(open('$plan_file'))['metadata_body']))")" >/dev/null
      fi

      if [ "$rules_delete_count" -gt 0 ]; then
        local rid
        while IFS= read -r rid; do
          [ -n "$rid" ] || continue
          gcurl DELETE "/streams/$id/rules/$rid" >/dev/null
        done < <(python3 -c "import json;print('\n'.join(json.load(open('$plan_file'))['rules_delete']))")
      fi

      if [ "$rules_create_count" -gt 0 ]; then
        local rule_body
        while IFS= read -r rule_body; do
          gcurl POST "/streams/$id/rules" "$rule_body" >/dev/null
        done < <(python3 -c "
import json
for r in json.load(open('$plan_file'))['rules_create']:
    print(json.dumps(r))
")
      fi

      if [ "$metadata_changed" = "True" ] || [ "$rules_create_count" -gt 0 ] || [ "$rules_delete_count" -gt 0 ]; then
        ok "updated stream '$title' ($id) - $rules_create_count rule(s) added, $rules_delete_count removed$([ "$metadata_changed" = "True" ] && echo ", metadata changed")"
      else
        skip "stream '$title' already up to date ($id)"
      fi

      rm -f "$plan_file" "$desired_file"
      echo "$id"
      return
    fi

    id="$(gcurl POST /streams "$(cat "$desired_file")" | python3 -c "import json,sys;print(json.load(sys.stdin)['stream_id'])")"
    ok "created stream '$title' ($id)"
    gcurl POST "/streams/$id/resume" "" >/dev/null
    rm -f "$desired_file"
    echo "$id"
  }

  NETWORK_STREAM_ID="$(render_and_create "$SCRIPT_DIR/streams/stream1_network.json" "Network Equipment" \
    "s/__NETWORK_INPUT_ID_514__/$NETWORK_INPUT_ID_514/" \
    "s/__NETWORK_INPUT_ID_1514__/$NETWORK_INPUT_ID_1514/")"
  SERVERS_STREAM_ID="$(render_and_create "$SCRIPT_DIR/streams/stream2_servers.json" "Servers" \
    "s/__SERVERS_INPUT_ID__/$SERVERS_INPUT_ID/")"
  rm -f "$existing_file"
}

step_connect_pipelines() {
  log "Connecting pipelines to streams..."
  gcurl POST /system/pipelines/connections/to_stream \
    "{\"stream_id\":\"$NETWORK_STREAM_ID\",\"pipeline_ids\":[\"$NETWORK_PIPELINE_ID\"]}" >/dev/null
  gcurl POST /system/pipelines/connections/to_stream \
    "{\"stream_id\":\"$SERVERS_STREAM_ID\",\"pipeline_ids\":[\"$SERVERS_PIPELINE_ID\"]}" >/dev/null
  ok "Pipelines connected to streams"
}

step_firewall() {
  cp "$SCRIPT_DIR/nftables.conf" /etc/nftables.conf
  systemctl enable nftables >/dev/null 2>&1 || true
  nft -f /etc/nftables.conf
  ok "nftables applied (only 9000/tcp, 514+1514/udp, 5140/udp open)"
}

step_dashboard() {
  # Three focused dashboards instead of one combined view - each targets a
  # different reader (landing/alerts overview, network-equipment-only,
  # servers-only) so opening Graylog goes straight to what's relevant
  # instead of one big mixed-stream page.
  #
  # Same "compare content, don't just check the title exists" fix as
  # step_pipeline_rules()/step_pipelines()/step_alerts() - a dashboard's
  # title never changes when a widget/query in search_*.json does, so a
  # plain title-exists check silently skips the update forever. Unlike
  # rules/pipelines/alerts there's no safe in-place PUT that keeps a view
  # and its search in sync here, so a changed dashboard is deleted and
  # recreated instead - low risk since dashboards have no subscribers or
  # history the way alerts or streams do. Compares only the 'queries' key
  # our search_*.json files author, same reasoning as step_alerts(): GET
  # fills in extra defaults (owner, created_at, requires, ...) that never
  # appear in our files, so a whole-object compare would always say
  # "changed".
  local existing_views_file
  existing_views_file="$(mktemp)"
  gcurl GET /views > "$existing_views_file"

  local search_file view_file title existing_id existing_search_id desired_search_file current_search_file search_id id body
  for search_file in "$SCRIPT_DIR"/dashboards/search_*.json; do
    view_file="${search_file/search_/view_}"
    [ -f "$view_file" ] || die "missing $view_file for $search_file"
    title="$(python3 -c "import json;print(json.load(open('$view_file'))['entity']['title'])")"

    desired_search_file="$(mktemp)"
    sed -e "s/__NETWORK_STREAM_ID__/$NETWORK_STREAM_ID/g" -e "s/__SERVERS_STREAM_ID__/$SERVERS_STREAM_ID/g" "$search_file" > "$desired_search_file"

    existing_id="$(python3 -c "
import json
d = json.load(open('$existing_views_file'))
print(next((v['id'] for v in d['views'] if v['title'] == '$title'), ''))
")"

    if [ -n "$existing_id" ]; then
      existing_search_id="$(python3 -c "
import json
d = json.load(open('$existing_views_file'))
print(next(v['search_id'] for v in d['views'] if v['id'] == '$existing_id'))
")"
      current_search_file="$(mktemp)"
      gcurl GET "/views/search/$existing_search_id" > "$current_search_file"

      if python3 -c "
import json, sys

def sort_search_types(q):
    # search_types order isn't semantically meaningful (widgets are matched
    # to view widgets by 'id', not position) but Graylog doesn't return them
    # in the same order they were created in - confirmed live 2026-07-23:
    # this caused a false 'changed' positive for every single dashboard on
    # every single run, identical to the list-order bug already worked
    # around in step_alerts().
    q = dict(q)
    if 'search_types' in q:
        q['search_types'] = sorted(q['search_types'], key=lambda st: st.get('id', ''))
    return q

def filter_to(desired_obj, current_obj):
    # Keep only the keys/structure our search_*.json files actually author -
    # confirmed live 2026-07-23: a 'messages'-type search_type (the message
    # list widget on the Overview dashboard) comes back from GET with a pile
    # of extra default keys (decorators, fields, filter, name, timerange,
    # stream_categories, ...) that never appear in our file, so a whole-
    # object compare always said 'changed' for that one dashboard even with
    # sort_search_types() already applied - same reasoning as step_alerts().
    if isinstance(desired_obj, dict):
        cur = current_obj if isinstance(current_obj, dict) else {}
        return {k: filter_to(v, cur.get(k)) for k, v in desired_obj.items()}
    if isinstance(desired_obj, list):
        cur = current_obj if isinstance(current_obj, list) else []
        return [filter_to(dv, cur[i] if i < len(cur) else None) for i, dv in enumerate(desired_obj)]
    return current_obj

current = json.load(open('$current_search_file'))
desired = json.load(open('$desired_search_file'))
current_q = [sort_search_types(q) for q in current.get('queries', [])]
desired_q = [sort_search_types(q) for q in desired.get('queries', [])]
sys.exit(0 if filter_to(desired_q, current_q) == desired_q else 1)
"; then
        skip "dashboard '$title' already up to date ($existing_id)"
        rm -f "$desired_search_file" "$current_search_file"
        continue
      fi
      rm -f "$current_search_file"

      gcurl DELETE "/views/$existing_id" >/dev/null
      gcurl DELETE "/views/search/$existing_search_id" >/dev/null
      ok "removed stale dashboard '$title' ($existing_id) - content changed, recreating"
    fi

    search_id="$(gcurl POST /views/search "$(cat "$desired_search_file")" | python3 -c "import json,sys;print(json.load(sys.stdin)['id'])")"
    [ -n "$search_id" ] || die "search creation failed for dashboard '$title'"
    rm -f "$desired_search_file"

    body="$(sed -e "s/__NETWORK_STREAM_ID__/$NETWORK_STREAM_ID/g" -e "s/__SERVERS_STREAM_ID__/$SERVERS_STREAM_ID/g" -e "s/__SEARCH_ID__/$search_id/" "$view_file")"
    id="$(gcurl POST /views "$body" | python3 -c "import json,sys;print(json.load(sys.stdin).get('id',''))")"
    [ -n "$id" ] || die "dashboard '$title' failed to create"
    ok "created dashboard '$title' ($id)"
  done
  rm -f "$existing_views_file"
}

step_alerts() {
  if [ -z "$DISCORD_WEBHOOK_URL" ]; then
    skip "Discord notification and alerts (DISCORD_WEBHOOK_URL not set)"
    return
  fi
  if [ ! -f "$SCRIPT_DIR/alerts/notification_discord.json" ]; then
    skip "Alerts (no alerts/notification_discord.json in this checkout)"
    return
  fi

  log "Creating Discord notification (idempotent)..."
  local existing_notifs notif_id
  existing_notifs="$(gcurl GET /events/notifications)"
  notif_id="$(echo "$existing_notifs" | python3 -c "import json,sys;d=json.load(sys.stdin);print(next((n['id'] for n in d['notifications'] if n['title']=='Discord - Critical Alerts'),''))")"
  if [ -n "$notif_id" ]; then
    skip "Discord notification already exists ($notif_id)"
  else
    local webhook_slack body
    webhook_slack="${DISCORD_WEBHOOK_URL%/}/slack"
    body="$(sed "s#__DISCORD_WEBHOOK_URL_SLACK__#${webhook_slack}#" "$SCRIPT_DIR/alerts/notification_discord.json")"
    notif_id="$(gcurl POST /events/notifications "{\"entity\": ${body}, \"share_request\": {\"selected_grantee_capabilities\": {}}}" \
      | python3 -c "import json,sys;print(json.load(sys.stdin)['id'])")"
    ok "created Discord notification ($notif_id)"
  fi

  log "Creating alert (event) definitions from $SCRIPT_DIR/alerts/*.json (idempotent)..."
  # Same "compare content, PUT-update if changed" fix as step_pipeline_rules()/
  # step_pipelines() - an alert's title never changes when its query/threshold
  # does, so a plain title-exists check would silently skip the update
  # forever.
  #
  # Uses files instead of interpolating JSON into python -c strings (fragile
  # with quotes/backslashes), and compares only the keys OUR file authors in
  # "config" - not the whole object. Confirmed live this is necessary:
  # Graylog's GET response fills in extra defaults (query_parameters, filters,
  # use_cron_scheduling, cron_expression, cron_timezone, ...) that never
  # appear in our alert JSON files, so a naive whole-object comparison always
  # reports "changed" and PUTs every single alert on every single run, not
  # just the one actually edited.
  local existing_defs_file
  existing_defs_file="$(mktemp)"
  gcurl GET /events/definitions > "$existing_defs_file"

  local f title id body_file
  for f in "$SCRIPT_DIR"/alerts/alert*.json; do
    [ -f "$f" ] || continue
    title="$(python3 -c "import json;print(json.load(open('$f'))['title'])")"
    body_file="$(mktemp)"
    sed \
      -e "s/__NETWORK_STREAM_ID__/$NETWORK_STREAM_ID/" \
      -e "s/__SERVERS_STREAM_ID__/$SERVERS_STREAM_ID/" \
      -e "s/__DISCORD_NOTIFICATION_ID__/$notif_id/" \
      "$f" > "$body_file"

    id="$(python3 -c "
import json
d = json.load(open('$existing_defs_file'))
print(next((e['id'] for e in d['event_definitions'] if e['title'] == '$title'), ''))
")"

    if [ -n "$id" ]; then
      if python3 -c "
import json, sys

def normalize(v):
    # List order isn't semantically meaningful here (e.g. 'streams' with
    # two entries) but a plain dict/list == comparison is order-sensitive -
    # confirmed live this caused a false 'changed' positive specifically
    # for the one alert with a 2-element streams list, every single run.
    if isinstance(v, list):
        return sorted(v, key=str)
    return v

existing = json.load(open('$existing_defs_file'))
current_full = next(e['config'] for e in existing['event_definitions'] if e['id'] == '$id')
desired = json.load(open('$body_file'))['config']
filtered_current = {k: normalize(current_full.get(k)) for k in desired}
desired_normalized = {k: normalize(v) for k, v in desired.items()}
sys.exit(0 if filtered_current == desired_normalized else 1)
"; then
        skip "alert '$title' already up to date ($id)"
        rm -f "$body_file"
        continue
      fi
      python3 -c "
import json
d = json.load(open('$body_file'))
d['id'] = '$id'
print(json.dumps(d))
" > "${body_file}.put"
      gcurl PUT "/events/definitions/$id" "$(cat "${body_file}.put")" >/dev/null
      gcurl PUT "/events/definitions/$id/schedule" "" >/dev/null
      ok "updated alert '$title' ($id) - config changed"
      rm -f "$body_file" "${body_file}.put"
      continue
    fi

    id="$(gcurl POST /events/definitions "{\"entity\": $(cat "$body_file"), \"share_request\": {\"selected_grantee_capabilities\": {}}}" \
      | python3 -c "import json,sys;print(json.load(sys.stdin).get('id',''))")"
    [ -n "$id" ] || die "alert '$title' failed to create"
    gcurl PUT "/events/definitions/$id/schedule" "" >/dev/null
    ok "created and enabled alert '$title' ($id)"
    rm -f "$body_file"
  done
  rm -f "$existing_defs_file"
}

# ---------------------------------------------------------------------------
main() {
  require_root
  step_packages
  step_disable_ipv6
  step_docker_install
  step_check_max_map_count
  step_docker_smoke_test
  step_firewall
  step_compose_files
  step_compose_up

  # Must be a plain statement (not `x=$(...)`) so a failure here truly
  # aborts the script - gcurl() is always called from inside command
  # substitutions downstream, where `die`'s `exit` would only kill a subshell.
  resolve_admin_password
  wait_for_api_ready

  step_index_retention
  step_inputs
  step_pipeline_rules
  step_pipelines
  step_streams
  step_connect_pipelines
  step_dashboard
  step_alerts

  echo >&2
  echo "${C_GREEN}==================================================${C_RESET}" >&2
  echo "${C_GREEN} DONE - Graylog is up${C_RESET}" >&2
  echo "${C_GREEN}==================================================${C_RESET}" >&2
  echo " Web UI: $GRAYLOG_EXTERNAL_URI" >&2
  echo " Config: $INSTALL_DIR/docker-compose.yml and $INSTALL_DIR/.env" >&2
  if [ -f "$INSTALL_DIR/.admin_credentials_ONE_TIME" ]; then
    echo " Admin user:     $ADMIN_USER" >&2
    echo " Admin password: $ADMIN_PASSWORD" >&2
    echo " (shown once above - store it now; the one-time file is being deleted)" >&2
    rm -f "$INSTALL_DIR/.admin_credentials_ONE_TIME"
  fi
  echo "${C_GREEN}==================================================${C_RESET}" >&2
}

main "$@"