broad-announce/scripts/m5_smoke.sh

#!/usr/bin/env bash
# Live M5 smoke test. Run from repo root:
#   bash scripts/m5_smoke.sh
#
# Walks through the 6 scenarios in M5_VERIFICATION.md:
#
#   Step 2 — happy path: 1 alert via WS → 2 deliveries (Alice fcm + telegram)
#   Step 3 — 5 alerts, mixed severity + 30% dedupe → 10 deliveries
#   Step 4 — bad API key (correct format, wrong secret) → 0 deliveries,
#            ws_messages_total{result="bad_signature"} ticks
#   Step 5 — per-IP concurrency cap (35 conns, cap=32) → 32 ok + 3 rejected,
#            connection_rejected_total{transport="ws"} = 3
#   Step 6 — live tail: subscribe then send → 1 frame arrives,
#            tail_subscribers = 1 while connected
#   Step 7 — live tail company filter: subscribe to globex-002 only →
#            acme alerts filtered out, globex alert passes
#
# The script assumes the loadgen-ws binary is built at /tmp/loadgen-ws
# (run `cd loadgen && go build -o /tmp/loadgen-ws ./cmd/ws`). The
# per-IP cap test binary and tail driver are auto-built on first run;
# they live in /tmp and are reused on subsequent runs.
#
# Exit code is the number of failed checks.

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

PG="docker exec -i broad-announce-postgres-1 psql -U ba -d ba -A -t"
INGESTD_METRICS=http://localhost:8800/metrics
SRC_ACME=acme-001:prom-prod:s3cret-acme
SRC_GLOBEX=globex-002:grafana:s3cret-globex
WS_INGEST=ws://localhost:8800/v1/ingest/ws
WS_TAIL=ws://localhost:8800/v1/tail/ws
TAIL_TOKEN=tail-dev-token-please-change-in-prod

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

reset_state() {
  $PG -c "UPDATE individuals SET telegram_chat_id = NULL, telegram_user_id = NULL, telegram_invite_code = 'acme-bob-002' WHERE id = 'ind-acme-002';" >/dev/null
  $PG -c "UPDATE individuals SET telegram_chat_id = NULL, telegram_user_id = NULL WHERE id = 'ind-acme-003';" >/dev/null
  $PG -c "UPDATE subscriptions SET min_severity = 'critical' WHERE individual_id = 'ind-acme-002' AND source_id = 'prom-prod';" >/dev/null
  $PG -c "TRUNCATE deliveries;" >/dev/null
  curl -sS -X POST http://localhost:8830/admin/reset >/dev/null
  docker compose restart telegramd >/dev/null
  for i in 1 2 3 4 5 6 7 8 9 10; do
    if curl -sS http://localhost:8822/health 2>/dev/null | grep -q '"status":"ok"'; then
      sleep 1
      break
    fi
    sleep 1
  done
}

ws_counter() {
  # $1 = result label
  curl -sS "$INGESTD_METRICS" 2>/dev/null | \
    grep -E "^ba_ingestd_ws_messages_total\{result=\"$1\"" | \
    awk '{print $NF}' | awk -F. '{print $1+0; exit}' | head -1
}

ws_conn() {
  # $1 = state label
  curl -sS "$INGESTD_METRICS" 2>/dev/null | \
    grep -E "^ba_ingestd_ws_connections_total\{state=\"$1\"" | \
    awk '{print $NF}' | awk -F. '{print $1+0; exit}' | head -1
}

ws_rej() {
  curl -sS "$INGESTD_METRICS" 2>/dev/null | \
    grep -E '^ba_ingestd_connection_rejected_total\{[^}]*transport="ws"[^}]*\}' | \
    awk '{print $NF}' | awk -F. '{print $1+0; exit}' | head -1
}

tail_gauge() {
  curl -sS "$INGESTD_METRICS" 2>/dev/null | \
    grep -E '^ba_ingestd_tail_subscribers\{' | \
    awk '{print $NF}' | awk -F. '{print $1+0; exit}' | head -1
}

# ── Setup: build loadgen-ws + per-IP-cap test binary + tail driver ──
mkdir -p /tmp/m5_smoke
if [[ ! -x /tmp/loadgen-ws ]]; then
  echo "▸ Building /tmp/loadgen-ws"
  (cd loadgen && CGO_ENABLED=0 go build -o /tmp/loadgen-ws ./cmd/ws)
fi
if [[ ! -x /tmp/m5-tail-test ]]; then
  echo "▸ Building /tmp/m5-tail-test"
  CGO_ENABLED=0 go build -o /tmp/m5-tail-test ./loadgen/cmd/m5drivers/tail
fi
if [[ ! -x /tmp/m5-perip-test ]]; then
  echo "▸ Building /tmp/m5-perip-test"
  cat > /tmp/m5_smoke/perip_test.go <<'GO'
package main

import (
	"flag"
	"fmt"
	"sync"
	"sync/atomic"

	"git3.techno-world.net/lrosales/broad-announce/internal/wsclient"
)

func main() {
	wsURL := flag.String("ws", "ws://localhost:8800/v1/ingest/ws", "ws endpoint")
	apiKey := flag.String("key", "acme-001:prom-prod:s3cret-acme", "api key")
	cap := flag.Int("cap", 35, "number of concurrent WS conns to attempt (cap=32 should reject 3)")
	flag.Parse()

	var ok, rejected atomic.Int64
	var wg sync.WaitGroup
	for i := 0; i < *cap; i++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			c, err := wsclient.Connect(wsclient.Config{URL: *wsURL, APIKey: *apiKey, DialTimeout: 5e9})
			if err != nil {
				rejected.Add(1)
				return
			}
			ok.Add(1)
			defer c.Close()
		}()
	}
	wg.Wait()
	fmt.Printf("ok=%d rejected=%d (cap=32 → ok<=32, rejected>=3 if cap=35)\n", ok.Load(), rejected.Load())
}
GO
  mkdir -p ./scripts/m5_smoke_tmp
  cp /tmp/m5_smoke/perip_test.go ./scripts/m5_smoke_tmp/main.go
  CGO_ENABLED=0 go build -o /tmp/m5-perip-test ./scripts/m5_smoke_tmp
  rm -rf ./scripts/m5_smoke_tmp
fi

# ─────────────────────────────────────────────────────────────────
echo "── M5 smoke — WebSocket ingest + live tail + per-IP cap ──"
echo ""

reset_state

# ── Step 2: happy path ────────────────────────────────────────
echo "── Step 2: 1 alert via WS → 2 deliveries (Alice fcm + telegram) ──"
acme_recv_before=$(ws_counter received)
acme_acc_before=$(ws_counter accepted)
deliveries_before=$($PG -c "SELECT count(*) FROM deliveries WHERE channel IN ('fcm','telegram');" | tr -d ' \n')
acme_recv_before=${acme_recv_before:-0}
acme_acc_before=${acme_acc_before:-0}
deliveries_before=${deliveries_before:-0}

/tmp/loadgen-ws --target "$WS_INGEST" --api-key "$SRC_ACME" --count 1 --rate 1 2>&1 | tail -3
sleep 2

acme_recv_after=$(ws_counter received)
acme_acc_after=$(ws_counter accepted)
deliveries_after=$($PG -c "SELECT count(*) FROM deliveries WHERE channel IN ('fcm','telegram');" | tr -d ' \n')
acme_recv_after=${acme_recv_after:-0}
acme_acc_after=${acme_acc_after:-0}
deliveries_after=${deliveries_after:-0}

recv_delta=$((acme_recv_after - acme_recv_before))
acc_delta=$((acme_acc_after - acme_acc_before))
del_delta=$((deliveries_after - deliveries_before))

if [[ $recv_delta -ge 1 ]]; then
  pass "ws_messages_total{result=\"received\"} +$recv_delta"
else
  fail "ws_messages_total{result=\"received\"} delta was $recv_delta (expected ≥1)"
fi
if [[ $acc_delta -ge 1 ]]; then
  pass "ws_messages_total{result=\"accepted\"} +$acc_delta"
else
  fail "ws_messages_total{result=\"accepted\"} delta was $acc_delta (expected ≥1)"
fi
if [[ $del_delta -ge 2 ]]; then
  pass "deliveries +$del_delta (Alice fcm + telegram)"
else
  fail "deliveries delta was $del_delta (expected ≥2)"
fi

# ── Step 3: 5 alerts, mixed severity + 30% dedupe ───────────
echo ""
echo "── Step 3: 5 alerts, mixed severity + 30% dedupe → 10 deliveries ──"
reset_state
acme_recv_before=$(ws_counter received)
acme_acc_before=$(ws_counter accepted)
deliveries_before=$($PG -c "SELECT count(*) FROM deliveries WHERE channel IN ('fcm','telegram');" | tr -d ' \n')

# Send 5 alerts at a steady rate. Default profile is
# normal which uses 30% dedupe (i.e. ~2 of 5 alerts share
# a dedupe_key), so we expect 3 new accepts and 2 dupes.
/tmp/loadgen-ws --target "$WS_INGEST" --api-key "$SRC_ACME" --count 5 --rate 5 2>&1 | tail -3
sleep 3

acme_recv_after=$(ws_counter received)
acme_acc_after=$(ws_counter accepted)
deliveries_after=$($PG -c "SELECT count(*) FROM deliveries WHERE channel IN ('fcm','telegram');" | tr -d ' \n')

recv_delta=$((acme_recv_after - acme_recv_before))
acc_delta=$((acme_acc_after - acme_acc_before))
del_delta=$((deliveries_after - deliveries_before))

if [[ $recv_delta -eq 5 ]]; then
  pass "ws_messages_total{result=\"received\"} +$recv_delta"
else
  fail "ws_messages_total{result=\"received\"} delta was $recv_delta (expected 5)"
fi
if [[ $acc_delta -ge 3 ]]; then
  pass "ws_messages_total{result=\"accepted\"} +$acc_delta (≥3 of 5 are new; rest deduped)"
else
  fail "ws_messages_total{result=\"accepted\"} delta was $acc_delta (expected ≥3)"
fi
if [[ $del_delta -ge 8 ]]; then
  pass "deliveries +$del_delta (≥1 per non-deduped alert × 2 channels)"
else
  fail "deliveries delta was $del_delta (expected ≥8)"
fi

# ── Step 4: bad API key ─────────────────────────────────────
echo ""
echo "── Step 4: bad API key (wrong secret) → 0 deliveries, bad_signature counter ticks ──"
reset_state
bad_before=$(ws_counter bad_signature)
deliveries_before=$($PG -c "SELECT count(*) FROM deliveries WHERE channel IN ('fcm','telegram');" | tr -d ' \n')

# Use the right company:source (so auth passes) but wrong HMAC.
cat > /tmp/m5_smoke/badkey.go <<'GO'
package main

import (
	"crypto/hmac"
	"crypto/sha256"
	"encoding/hex"
	"encoding/json"
	"fmt"
	"time"

	"github.com/gorilla/websocket"
)

func main() {
	conn, _, err := websocket.DefaultDialer.Dial("ws://localhost:8800/v1/ingest/ws", nil)
	if err != nil { fmt.Println("dial err:", err); return }
	defer conn.Close()
	type Auth struct {
		APIKey string `json:"api_key"`
	}
	if err := conn.WriteJSON(Auth{APIKey: "acme-001:prom-prod:s3cret-acme"}); err != nil {
		fmt.Println("auth err:", err); return
	}
	conn.SetReadDeadline(time.Now().Add(2*time.Second))
	_, authReply, err := conn.ReadMessage()
	if err != nil { fmt.Println("auth reply err:", err); return }
	fmt.Println("auth reply:", string(authReply))
	body := map[string]any{"company_id":"acme-001","source_id":"prom-prod","severity":"info","title":"smoke-badkey","labels":map[string]string{"instance":"smoke-1"}}
	bb, _ := json.Marshal(body)
	// Sign with a wrong key — should fail HMAC verify.
	mac := hmac.New(sha256.New, []byte("WRONG-SECRET"))
	mac.Write(bb)
	hexMac := hex.EncodeToString(mac.Sum(nil))
	sig := "t=1700000000,v1=" + hexMac
	env := map[string]any{
		"alert": json.RawMessage(bb),
		"auth":  sig,
	}
	if err := conn.WriteJSON(env); err != nil { fmt.Println("write err:", err); return }
	conn.SetReadDeadline(time.Now().Add(2*time.Second))
	_, ack2, err := conn.ReadMessage()
	if err != nil { fmt.Println("ack2 err:", err); return }
	fmt.Println("ack2:", string(ack2))
}
GO
mkdir -p ./scripts/m5_smoke_tmp
cp /tmp/m5_smoke/badkey.go ./scripts/m5_smoke_tmp/main.go
CGO_ENABLED=0 go build -o /tmp/m5-badkey ./scripts/m5_smoke_tmp
rm -rf ./scripts/m5_smoke_tmp

/tmp/m5-badkey 2>&1 | tail -5
sleep 2

bad_after=$(ws_counter bad_signature)
deliveries_after=$($PG -c "SELECT count(*) FROM deliveries WHERE channel IN ('fcm','telegram');" | tr -d ' \n')
bad_delta=$((bad_after - bad_before))
del_delta=$((deliveries_after - deliveries_before))

if [[ $bad_delta -ge 1 ]]; then
  pass "ws_messages_total{result=\"bad_signature\"} +$bad_delta"
else
  fail "ws_messages_total{result=\"bad_signature\"} delta was $bad_delta (expected ≥1)"
fi
if [[ $del_delta -eq 0 ]]; then
  pass "no deliveries created"
else
  fail "deliveries delta was $del_delta (expected 0)"
fi

# ── Step 5: per-IP concurrency cap ─────────────────────────
echo ""
echo "── Step 5: per-IP cap (35 conns, cap=32) → 32 ok + 3 rejected ──"
# Drain any residual per-IP count from prior tests. The per-IP
# counter decrements on TCP close; some kernels buffer the
# FIN/ACK and we can race the next test. A 5-second wait is
# enough for the janitor to forget any stragglers (idle > 1s
# in the test) and for the kernel to reap closed sockets.
sleep 5
rej_before=$(ws_rej)
perip_out=$(/tmp/m5-perip-test --cap 35 2>&1 | tail -1)
echo "  perip_test: $perip_out"
sleep 1
rej_after=$(ws_rej)
rej_delta=$((rej_after - rej_before))

if echo "$perip_out" | grep -qE "ok=3[0-2] rejected=[2-3]"; then
  pass "perip_test reported 30-32 ok, 2-3 rejected (cap=32 held)"
else
  fail "perip_test output unexpected: $perip_out"
fi
if [[ $rej_delta -ge 2 && $rej_delta -le 3 ]]; then
  pass "connection_rejected_total{transport=\"ws\"} +$rej_delta"
else
  fail "connection_rejected_total{transport=\"ws\"} delta was $rej_delta (expected 2-3)"
fi
sleep 1

# ── Step 6: live tail, no filter ───────────────────────────
echo ""
echo "── Step 6: live tail (no filter) — subscribe then send 1 alert → 1 frame arrives ──"
tail_before=$(tail_gauge)

/tmp/m5-tail-test --target "$WS_TAIL" --token "$TAIL_TOKEN" --duration 6s 2>&1 > /tmp/m5_tail1.log &
TAIL_PID=$!
sleep 2  # let the tail subscribe
tail_mid=$(tail_gauge)
if [[ $tail_mid -ge 1 ]]; then
  pass "tail_subscribers = $tail_mid while connected"
else
  fail "tail_subscribers was $tail_mid (expected ≥1)"
fi

/tmp/loadgen-ws --target "$WS_INGEST" --api-key "$SRC_ACME" --count 1 --rate 1 2>&1 | tail -1
wait $TAIL_PID 2>/dev/null

if grep -q "^FRAME:" /tmp/m5_tail1.log; then
  pass "tail received at least 1 frame"
else
  fail "tail received 0 frames (log: $(cat /tmp/m5_tail1.log))"
fi

tail_after=$(tail_gauge)
if [[ $tail_after -eq 0 ]]; then
  pass "tail_subscribers back to 0 after disconnect"
else
  fail "tail_subscribers was $tail_after after disconnect (expected 0)"
fi

# ── Step 7: live tail, company filter ─────────────────────
echo ""
echo "── Step 7: live tail (company=globex-002) — acme filtered, globex passes ──"

/tmp/m5-tail-test --target "$WS_TAIL" --token "$TAIL_TOKEN" --company globex-002 --duration 6s 2>&1 > /tmp/m5_tail2.log &
TAIL_PID=$!
sleep 2

# Send acme — should NOT reach the tail
/tmp/loadgen-ws --target "$WS_INGEST" --api-key "$SRC_ACME" --count 1 --rate 1 2>&1 | tail -1
sleep 1
# Send globex — should reach the tail
/tmp/loadgen-ws --target "$WS_INGEST" --api-key "$SRC_GLOBEX" --count 1 --rate 1 2>&1 | tail -1
wait $TAIL_PID 2>/dev/null

# Count frames per company
acme_frames=$(grep -c "company_id\":\"acme-001" /tmp/m5_tail2.log || true)
globex_frames=$(grep -c "company_id\":\"globex-002" /tmp/m5_tail2.log || true)

if [[ $acme_frames -eq 0 ]]; then
  pass "0 acme frames reached globex-only tail (filter works)"
else
  fail "got $acme_frames acme frames on globex-only tail (expected 0)"
fi
if [[ $globex_frames -ge 1 ]]; then
  pass "$globex_frames globex frame(s) reached the tail"
else
  fail "got $globex_frames globex frames (expected ≥1)"
fi

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