#!/usr/bin/env bash
# Live M6.5 smoke test. Run from repo root:
#   bash scripts/m6.5_smoke.sh
#
# Walks through the 4 scenarios in M6.5_VERIFICATION.md:
#
#   Step 2 — single collapse: 100 alerts with the same
#            --dedupe-key → faketgmd /admin/sent shows
#            count=1, text contains '(×100)'. (This is the
#            headline M6.5 win: the recipient gets ONE
#            message, not 100.)
#   Step 3 — tail visibility: the WS tail still shows
#            every arrival (operators see the storm live),
#            even though only 1 delivery went out.
#   Step 4 — max-wait re-flush: a 6s continuous burst
#            re-flushes every 2s. The recipient gets 3+
#            messages, each with the count-so-far.
#   Step 5 — per-source isolation: same --dedupe-key from
#            acme (prom-prod) and globex (grafana) does
#            NOT collapse into one — separate state, two
#            messages.
#   Step 6 — empty key passthrough: the M2 path is
#            preserved — alerts with empty dedupe_key
#            deliver immediately. (The loadgen normal
#            profile gives each alert a unique key, so
#            3 alerts produce 3 deliveries — i.e. no
#            collapse happens for keys we never see twice.)
#
# Exit code is the number of failed checks.

set -e
cd "$(dirname "$0")/.."

INGESTD_METRICS=http://localhost:8800/metrics
FAKETGMD=http://localhost:8830
WS_INGEST=ws://localhost:8800/v1/ingest/ws
WS_TAIL=ws://localhost:8800/v1/tail/ws
SRC_ACME=acme-001:prom-prod:s3cret-acme
SRC_GLOBEX=globex-002:grafana:s3cret-globex
TAIL_TOKEN=tail-dev-token-please-change-in-prod
BA_FLUSH_MS=${BA_ROUTERD_DEDUPE_FLUSH_MS:-2000}

fails=0
pass() { echo "  ✅ $*"; }
fail() { echo "  ❌ $*"; fails=$((fails+1)); }

# Seed globex company + source + individual + subscription.
# Done once at the top of the script (idempotent), not in
# reset_state — the schema's foreign keys make per-call seed
# noisy. Step 5 needs globex to actually have a recipient.
PGCMD="docker exec -i broad-announce-postgres-1 psql -U ba -d ba -A -t"
$PGCMD -c \
  "INSERT INTO companies (id, name) VALUES ('globex-002', 'Globex Corp') ON CONFLICT (id) DO NOTHING;" \
  >/dev/null 2>&1 || true
$PGCMD -c \
  "INSERT INTO sources (id, company_id, name) VALUES ('grafana', 'globex-002', 'Globex Grafana') ON CONFLICT (company_id, id) DO NOTHING;" \
  >/dev/null 2>&1 || true
$PGCMD -c \
  "INSERT INTO individuals (id, company_id, full_name, telegram_chat_id) VALUES ('ind-globex-001', 'globex-002', 'Globex Test', 2001) ON CONFLICT (id) DO UPDATE SET telegram_chat_id = 2001;" \
  >/dev/null 2>&1 || true
$PGCMD -c \
  "INSERT INTO subscriptions (individual_id, company_id, source_id, channel_mask, min_severity, tz, status) VALUES ('ind-globex-001', 'globex-002', 'grafana', '[\"fcm\",\"telegram\"]'::jsonb, '', 'UTC', 'active') ON CONFLICT (individual_id, source_id) DO UPDATE SET channel_mask = '[\"fcm\",\"telegram\"]'::jsonb;" \
  >/dev/null 2>&1 || true

reset_state() {
  curl -sS -X POST $FAKETGMD/admin/reset >/dev/null
  # deliveries are kept (we want to count them across steps)
}

# ── Setup: build loadgen-ws and tailcount ──
if [[ ! -x /tmp/loadgen-ws-m65 ]]; then
  echo "▸ Building /tmp/loadgen-ws-m65"
  (cd loadgen && CGO_ENABLED=0 go build -o /tmp/loadgen-ws-m65 ./cmd/ws)
fi
if [[ ! -x /tmp/tailcount ]]; then
  echo "▸ Building /tmp/tailcount"
  (cd testfakes/tailcount && CGO_ENABLED=0 go build -o /tmp/tailcount)
fi

# ─────────────────────────────────────────────────────────────────
echo "── M6.5 smoke — Router-level dedupe collapse ──"
echo ""

# ── Step 2: single collapse (100 alerts → 1 message) ─────
echo "── Step 2: 100 alerts with same --dedupe-key → 1 message, (×100) ──"
reset_state
# Drain the dedupe state for our key. We use a timestamp
# suffix so each run is isolated.
KEY="m65-step2-$(date +%s%N)"
/tmp/loadgen-ws-m65 --target "$WS_INGEST" --api-key "$SRC_ACME" \
  --count 100 --rate 50 --dedupe-key "$KEY" 2>&1 | tail -2
# Wait for the router's flush (BA_ROUTERD_DEDUPE_FLUSH_MS)
# plus a margin for the deliverer round-trip.
sleep_secs=$(awk "BEGIN { print int($BA_FLUSH_MS/1000) + 3 }")
sleep "$sleep_secs"

count=$(curl -sS $FAKETGMD/admin/sent 2>/dev/null | python3 -c 'import json,sys; d=json.load(sys.stdin); print(d.get("count",0))' 2>/dev/null)
if [[ "$count" == "1" ]]; then
  pass "faketgmd /admin/sent count = 1 (100 alerts collapsed to 1 delivery)"
else
  fail "faketgmd /admin/sent count = $count (expected 1)"
fi

last_text=$(curl -sS $FAKETGMD/admin/sent 2>/dev/null | python3 -c 'import json,sys; d=json.load(sys.stdin); items=d.get("items",[]); print(items[-1]["text"] if items else "")' 2>/dev/null)
if echo "$last_text" | grep -q '(×100)'; then
  pass "faketgmd last message contains '(×100)'"
else
  fail "faketgmd last message: $last_text (expected '(×100)' suffix)"
fi

# ── Step 3: tail still shows the storm ────────────────────
echo ""
echo "── Step 3: WS tail sees every arrival (operators see the storm live) ──"
reset_state
KEY="m65-step3-$(date +%s%N)"

# Subscribe to the tail in the background, filtered to the
# title pattern that the loadgen's makeAlert puts in when
# --dedupe-key is set ("LG M5 burst <key> #N"). The compact
# tail event includes title but NOT dedupe_key, so we filter
# on the title.
/tmp/tailcount -token "$TAIL_TOKEN" -company "acme-001" \
  -filter "burst $KEY" -timeout 12s > /tmp/m65-step3-tail.out 2>&1 &
TAIL_PID=$!

# Give the tail 1.5s to connect.
sleep 1.5

/tmp/loadgen-ws-m65 --target "$WS_INGEST" --api-key "$SRC_ACME" \
  --count 25 --rate 50 --dedupe-key "$KEY" 2>&1 | tail -1

# Wait for the tailcount to finish.
wait $TAIL_PID 2>/dev/null || true

tail_count=$(grep -oE 'count=[0-9]+' /tmp/m65-step3-tail.out | tail -1 | cut -d= -f2)
tail_count=${tail_count:-0}
if [[ $tail_count -ge 20 ]]; then
  pass "tail subscriber saw $tail_count events (≥20 of 25 alerts streamed live)"
else
  fail "tail saw $tail_count events (expected ≥20, output=$(cat /tmp/m65-step3-tail.out))"
fi

# And the recipient still only got 1.
sleep "$sleep_secs"
count=$(curl -sS $FAKETGMD/admin/sent 2>/dev/null | python3 -c 'import json,sys; d=json.load(sys.stdin); print(d.get("count",0))' 2>/dev/null)
if [[ "$count" == "1" ]]; then
  pass "faketgmd count = 1 (collapse still applied — tail ≠ delivery)"
else
  fail "faketgmd count = $count (expected 1)"
fi

# ── Step 4: max-wait re-flush (continuous stream) ────────
echo ""
echo "── Step 4: 6s continuous burst at 30/s → ≥3 re-flushes ──"
reset_state
KEY="m65-step4-$(date +%s%N)"

/tmp/loadgen-ws-m65 --target "$WS_INGEST" --api-key "$SRC_ACME" \
  --count 180 --rate 30 --dedupe-key "$KEY" 2>&1 | tail -1
# 180 / 30 = 6 seconds of streaming, well above 3 flush windows
# at BA_ROUTERD_DEDUPE_FLUSH_MS=2000.
sleep "$sleep_secs"

count=$(curl -sS $FAKETGMD/admin/sent 2>/dev/null | python3 -c 'import json,sys; d=json.load(sys.stdin); print(d.get("count",0))' 2>/dev/null)
if [[ $count -ge 2 && $count -le 5 ]]; then
  pass "faketgmd count = $count (expected 2..5 re-flushes over 6s @ flushMs=$BA_FLUSH_MS)"
else
  fail "faketgmd count = $count (expected 2..5 re-flushes)"
fi

# ── Step 5: per-source isolation ─────────────────────────
echo ""
echo "── Step 5: same --dedupe-key from acme and globex → 2 separate collapses ──"
reset_state
KEY="m65-step5-$(date +%s%N)"
# 5 from acme, 5 from globex, same key.
# The two sources have separate Collapser state, so each
# gets 1 delivery. Total: 2 messages.
# acme subscription is on 'prom-prod'; globex is on 'grafana'.
/tmp/loadgen-ws-m65 --target "$WS_INGEST" --api-key "$SRC_ACME" \
  --count 5 --rate 5 --dedupe-key "$KEY" 2>&1 | tail -1
sleep 1
/tmp/loadgen-ws-m65 --target "$WS_INGEST" --api-key "$SRC_GLOBEX" \
  --count 5 --rate 5 --dedupe-key "$KEY" 2>&1 | tail -1
sleep "$sleep_secs"

count=$(curl -sS $FAKETGMD/admin/sent 2>/dev/null | python3 -c 'import json,sys; d=json.load(sys.stdin); print(d.get("count",0))' 2>/dev/null)
# We expect 2: one from acme's chat (1001) and one from
# globex's chat (2001). The faketgmd log groups by
# (bot_token, chat_id), so 2 distinct recipients show
# up as 2 items in /admin/sent.
if [[ "$count" == "2" ]]; then
  pass "faketgmd count = 2 (one per source — per-source isolation holds)"
else
  fail "faketgmd count = $count (expected 2: one per source)"
fi

# Verify the messages are from different chats.
chats=$(curl -sS $FAKETGMD/admin/sent 2>/dev/null | python3 -c 'import json,sys; d=json.load(sys.stdin); items=d.get("items",[]); print(",".join(sorted(set(str(i.get("chat_id")) for i in items))))' 2>/dev/null)
if [[ "$chats" == "1001,2001" ]]; then
  pass "chat_ids = $chats (acme=1001, globex=2001)"
else
  fail "chat_ids = $chats (expected 1001,2001)"
fi

# ── Step 6: passthrough — unique keys don't collapse ──────
echo ""
echo "── Step 6: loadgen normal profile (unique keys per alert) → no collapse ──"
reset_state
# The loadgen's normal profile gives each alert a unique
# dedupe_key (lg-m5-0, lg-m5-1, ...) — 30% share, but with
# idx>0 and only 3 alerts, in practice they're all unique.
# The Collapser's Passthrough path is for empty dedupe_key;
# the unique-keys path uses CollapseNew, which still
# publishes after the first observe. So 3 alerts → 3
# messages (one per key).
/tmp/loadgen-ws-m65 --target "$WS_INGEST" --api-key "$SRC_ACME" \
  --count 3 --rate 5 2>&1 | tail -1
sleep "$sleep_secs"

count=$(curl -sS $FAKETGMD/admin/sent 2>/dev/null | python3 -c 'import json,sys; d=json.load(sys.stdin); print(d.get("count",0))' 2>/dev/null)
if [[ "$count" == "3" ]]; then
  pass "faketgmd count = 3 (3 unique keys → 3 deliveries, no collapse)"
else
  fail "faketgmd count = $count (expected 3)"
fi

# ── Summary ─────────────────────────────────────────────────
echo ""
if [[ $fails -eq 0 ]]; then
  echo "🟢 M6.5 smoke PASS — all checks green"
  exit 0
else
  echo "🔴 M6.5 smoke FAIL — $fails check(s) failed"
  exit $fails
fi