broad-announce/scripts/m4_smoke.sh

#!/usr/bin/env bash
# Live M4 smoke test. Run from repo root:
#   bash scripts/m4_smoke.sh
#
# Walks through the 5 scenarios in M4_VERIFICATION.md:
#
#   Step 2 — happy path: 1 alert via MQTT → 2 deliveries
#   Step 3 — 5 alerts, mixed severity + 30% dedupe → 10 deliveries
#   Step 4 — bad signature (correct user, wrong HMAC) → broker accepts,
#            ingestd rejects, no delivery, bad_signature counter ticks
#   Step 5 — ACL violation (prom-prod user → globex's topic) → broker
#            denies the publish, no delivery, no ingestd counter tick
#   Step 6 — non-JSON body → invalid_json counter ticks
#
# The script assumes the loadgen-mqtt binary is built at /tmp/loadgen-mqtt
# (run `cd loadgen && go build -o /tmp/loadgen-mqtt ./cmd/mqtt`). The
# failure-path test binaries (m4_badsig, m4_acl_violation, m4_badjson)
# 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
BROKER=tcp://localhost:1883
SRC=acme-001:prom-prod:s3cret-acme
SRC_USER=prom-prod-acme-001

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
}

mqtt_counter() {
  # $1 = result label
  # Reads /metrics and returns the integer value of
  #   ba_ingestd_mqtt_messages_total{result="<label>",service="ingestd"} <N>
  # Missing label → 0.
  curl -sS "$INGESTD_METRICS" 2>/dev/null | \
    grep -E "^ba_ingestd_mqtt_messages_total\{result=\"$1\"" | \
    awk '{print $NF}' | awk -F. '{print $1+0; exit}' | head -1
}

# ── Setup: build loadgen-mqtt + the 3 failure-path test binaries ──
mkdir -p /tmp/m4_smoke
if [[ ! -x /tmp/loadgen-mqtt ]]; then
  echo "▸ Building /tmp/loadgen-mqtt"
  (cd loadgen && CGO_ENABLED=0 go build -o /tmp/loadgen-mqtt ./cmd/mqtt)
fi

cat > /tmp/m4_smoke/acl_violation.go <<'GO'
package main
import (
	"fmt"
	mqtt "github.com/eclipse/paho.mqtt.golang"
	"os"
	"time"
)
func main() {
	opts := mqtt.NewClientOptions().
		AddBroker("tcp://localhost:1883").
		SetClientID("m4-acl-violation").
		SetUsername("prom-prod-acme-001").
		SetPassword("s3cret-acme").
		SetCleanSession(true)
	c := mqtt.NewClient(opts)
	if tok := c.Connect(); !tok.WaitTimeout(5*time.Second) || tok.Error() != nil {
		fmt.Println("connect err:", tok.Error()); os.Exit(0)
	}
	tok := c.Publish("ba/globex-002/grafana/incoming", 1, false, []byte("hack"))
	if !tok.WaitTimeout(5*time.Second) {
		fmt.Println("publish timeout"); os.Exit(0)
	}
	fmt.Println("published to wrong topic; err:", tok.Error())
	os.Exit(0)
}
GO
cat > /tmp/m4_smoke/badsig.go <<'GO'
package main
import (
	"encoding/json"
	"fmt"
	mqtt "github.com/eclipse/paho.mqtt.golang"
	"os"
	"time"
)
func main() {
	opts := mqtt.NewClientOptions().
		AddBroker("tcp://localhost:1883").
		SetClientID("m4-badsig").
		SetUsername("prom-prod-acme-001").
		SetPassword("s3cret-acme").
		SetCleanSession(true)
	c := mqtt.NewClient(opts)
	if tok := c.Connect(); !tok.WaitTimeout(5*time.Second) || tok.Error() != nil {
		fmt.Println("connect err:", tok.Error()); os.Exit(0)
	}
	body, _ := json.Marshal(map[string]any{
		"alert": json.RawMessage(`{"company_id":"acme-001","source_id":"prom-prod","severity":"info","title":"bad sig"}`),
		"auth":  "t=1700000000,v1=deadbeef",
	})
	tok := c.Publish("ba/acme-001/prom-prod/incoming", 1, false, body)
	if !tok.WaitTimeout(5*time.Second) {
		fmt.Println("publish timeout"); os.Exit(0)
	}
	fmt.Println("bad-sig published; err:", tok.Error())
	os.Exit(0)
}
GO
cat > /tmp/m4_smoke/badjson.go <<'GO'
package main
import (
	"fmt"
	mqtt "github.com/eclipse/paho.mqtt.golang"
	"os"
	"time"
)
func main() {
	opts := mqtt.NewClientOptions().
		AddBroker("tcp://localhost:1883").
		SetClientID("m4-badjson").
		SetUsername("prom-prod-acme-001").
		SetPassword("s3cret-acme").
		SetCleanSession(true)
	c := mqtt.NewClient(opts)
	if tok := c.Connect(); !tok.WaitTimeout(5*time.Second) || tok.Error() != nil {
		fmt.Println("connect err:", tok.Error()); os.Exit(0)
	}
	tok := c.Publish("ba/acme-001/prom-prod/incoming", 1, false, []byte("not json"))
	if !tok.WaitTimeout(5*time.Second) {
		fmt.Println("publish timeout"); os.Exit(0)
	}
	fmt.Println("bad-json published; err:", tok.Error())
	os.Exit(0)
}
GO
if [[ ! -x /tmp/m4_smoke/m4_acl_violation ]] || [[ ! -x /tmp/m4_smoke/m4_badsig ]] || [[ ! -x /tmp/m4_smoke/m4_badjson ]]; then
  echo "▸ Building the 3 failure-path test binaries"
  (cd loadgen && CGO_ENABLED=0 go build -o /tmp/m4_smoke/m4_acl_violation /tmp/m4_smoke/acl_violation.go)
  (cd loadgen && CGO_ENABLED=0 go build -o /tmp/m4_smoke/m4_badsig      /tmp/m4_smoke/badsig.go)
  (cd loadgen && CGO_ENABLED=0 go build -o /tmp/m4_smoke/m4_badjson     /tmp/m4_smoke/badjson.go)
fi

echo "═══════════════════════════════════════════════════════"
echo "  M4 smoke test — broad-announce (MQTT ingest)"
echo "═══════════════════════════════════════════════════════"

# ── Step 2: happy path ───────────────────────────────────────
echo
echo "▸ Step 2: 1 alert via MQTT"
reset_state
before_accept=$(mqtt_counter accepted)
/tmp/loadgen-mqtt --broker "$BROKER" --api-key "$SRC" --count 1 --rate 5 2>&1 | tail -2
sleep 4
n=$($PG -c "SELECT COUNT(*) FROM deliveries;")
[[ "$n" == "2" ]] && pass "2 deliveries (Alice fcm + Alice telegram)" || fail "expected 2 deliveries, got $n"
fb=$($PG -c "SELECT COUNT(*) FROM deliveries WHERE channel='fcm' AND status='sent';")
tb=$($PG -c "SELECT COUNT(*) FROM deliveries WHERE channel='telegram' AND status='sent';")
[[ "$fb" == "1" && "$tb" == "1" ]] && pass "fcm=1, telegram=1" || fail "fcm=$fb, telegram=$tb (expected 1,1)"
after_accept=$(mqtt_counter accepted)
delta=$((after_accept - before_accept))
[[ "$delta" -ge 1 ]] && pass "mqtt_messages{accepted} delta = $delta (≥ 1)" || fail "accepted counter delta = $delta"

# ── Step 3: 5 alerts, mixed severity + dedupe ────────────────
echo
echo "▸ Step 3: 5 alerts via MQTT (mixed severity, 30% dedupe)"
reset_state
before_accept=$(mqtt_counter accepted)
before_dedup=$(mqtt_counter deduped)
/tmp/loadgen-mqtt --broker "$BROKER" --api-key "$SRC" --count 5 --rate 5 --mode normal 2>&1 | tail -2
sleep 4
n=$($PG -c "SELECT COUNT(*) FROM deliveries;")
alerts=$($PG -c "SELECT COUNT(DISTINCT alert_id) FROM deliveries;")
[[ "$alerts" == "5" ]] && pass "5 distinct alerts in deliveries" || fail "expected 5 distinct alerts, got $alerts"
# Each alert is delivered to at least Alice (fcm + telegram = 2), and
# to Bob too for severity >= critical. The default 'normal' mode is
# 70% info / 25% warning / 4% critical / 1% inminent, so the total
# delivery count is at least 10 but can be up to 20.
[[ "$n" -ge 10 ]] && pass "$n deliveries (≥ 10, 5 alerts × ≥ 2 channels for Alice + maybe Bob)" || fail "expected ≥ 10 deliveries, got $n"
after_accept=$(mqtt_counter accepted)
after_dedup=$(mqtt_counter deduped)
[[ $((after_accept - before_accept)) -ge 3 ]] && pass "accepted counter advanced by ≥ 3" || fail "accepted counter delta small"
[[ $((after_dedup - before_dedup)) -ge 1 ]] && pass "deduped counter advanced by ≥ 1 (30% dedupe-pct)" || fail "deduped counter delta = $((after_dedup - before_dedup))"

# ── Step 4: bad signature ────────────────────────────────────
echo
echo "▸ Step 4: bad signature (correct user, wrong HMAC)"
reset_state
before_bs=$(mqtt_counter bad_signature)
/tmp/m4_smoke/m4_badsig 2>&1 | tail -1
sleep 3
n=$($PG -c "SELECT COUNT(*) FROM deliveries;")
[[ "$n" == "0" ]] && pass "0 deliveries (bad signature rejected)" || fail "expected 0 deliveries, got $n"
after_bs=$(mqtt_counter bad_signature)
[[ $((after_bs - before_bs)) -ge 1 ]] && pass "bad_signature counter advanced by ≥ 1" || fail "bad_signature counter delta = $((after_bs - before_bs))"

# ── Step 5: ACL violation ────────────────────────────────────
echo
echo "▸ Step 5: ACL violation (prom-prod user → globex's topic)"
reset_state
before_recv=$(mqtt_counter received)
/tmp/m4_smoke/m4_acl_violation 2>&1 | tail -1
sleep 3
n=$($PG -c "SELECT COUNT(*) FROM deliveries;")
[[ "$n" == "0" ]] && pass "0 deliveries (ACL denied at broker)" || fail "expected 0 deliveries, got $n"
after_recv=$(mqtt_counter received)
[[ $((after_recv - before_recv)) -eq 0 ]] && pass "received counter unchanged (broker dropped before subscriber)" || fail "received counter delta = $((after_recv - before_recv)) (broker should have dropped)"

# ── Step 6: non-JSON body ────────────────────────────────────
echo
echo "▸ Step 6: non-JSON body"
reset_state
before_ij=$(mqtt_counter invalid_json)
/tmp/m4_smoke/m4_badjson 2>&1 | tail -1
sleep 3
n=$($PG -c "SELECT COUNT(*) FROM deliveries;")
[[ "$n" == "0" ]] && pass "0 deliveries (invalid body rejected)" || fail "expected 0 deliveries, got $n"
after_ij=$(mqtt_counter invalid_json)
[[ $((after_ij - before_ij)) -ge 1 ]] && pass "invalid_json counter advanced by ≥ 1" || fail "invalid_json counter delta = $((after_ij - before_ij))"

echo
echo "═══════════════════════════════════════════════════════"
echo "  M4 smoke: $fails failure(s)"
echo "═══════════════════════════════════════════════════════"
exit $fails