# M4 Verification — MQTT ingest

This is the live smoke test for the M4 milestone
(**MQTT ingest**, SPEC §16 + §18 + §22). It exercises the
new path end-to-end:

1. **EMQX** is up with **per-source ACLs** (default deny on
   `#`). The bootstrap file
   `deploy/emqx/auth-built-in-db-bootstrap.csv` declares the
   sources; `deploy/emqx/acl.conf` restricts each user to
   their own `ba/<co>/<src>/incoming` topic.
2. **`internal/mqttclient`** — shared paho.MQTT wrapper used
   by both the ingestd subscriber and the new loadgen
   publisher.
3. **`cmd/ingestd/mqtt.go`** — MQTT subscriber that runs
   every message through the same `processDeps.ProcessAlert`
   pipeline as the HTTP POST handler (parse → validate →
   HMAC verify → rate-limit → dedupe → publish to NATS).
4. **`loadgen/cmd/mqtt/`** — `loadgen-mqtt` publisher. Same
   data shape as `loadgen-http` (severity mix, dedupe ratio,
   burst mode).
5. **Topic namespace** — `ba/<company_id>/<source_id>/incoming`.
   Sources publish here; `ingestd` subscribes to the wildcard
   `ba/+/+/incoming`. The `+` is the EMQX single-level
   wildcard.
6. **Auth model** — twofold:
   - **MQTT username/password** for the broker-level gate.
     Username = `<source_id>-<company_id>`, password = HMAC
     secret (so the same secret serves both as MQTT auth and
     per-message HMAC). The `acl.conf` then gates which topic
     a given user can publish to.
   - **Per-message X-BA-Signature** in the JSON envelope's
     `auth` field. Same `verifyHMAC()` is reused unchanged.

## Prerequisites

- Stack is up: `docker compose up -d`
- All M1–M3 migrations applied: `docker compose up seed`
- `BA_INGESTD_SOURCES` includes `acme-001:prom-prod:s3cret-acme`
  (and optionally `globex-002:grafana:s3cret-globex` for the
  ACL-violation test in step 5)
- The M4 ingestd env vars are set:
  ```
  BA_INGESTD_MQTT_BROKER=tcp://emqx:1883
  BA_INGESTD_MQTT_USERNAME=ingestd
  BA_INGESTD_MQTT_PASSWORD=ingestd-broker-only
  BA_INGESTD_MQTT_SUBSCRIBE=ba/+/+/incoming
  ```
- The EMQX bootstrap files are mounted (in compose by
  default): `deploy/emqx/{acl.conf,auth-built-in-db-bootstrap.csv}`.
- `loadgen-mqtt` is built:
  ```bash
  cd loadgen && go build -o /tmp/loadgen-mqtt ./cmd/mqtt
  ```

## Automated smoke

```bash
bash scripts/m4_smoke.sh
```

The script:
1. Builds `loadgen-mqtt` and the 3 failure-path test
   binaries (m4_badsig, m4_acl_violation, m4_badjson) if
   not already present.
2. Walks the 5 step-by-step scenarios below, asserting the
   expected delivery counts and metric counter deltas.
3. Exits 0 on success, $fail_count on failure.

## Step-by-step (also documented in `M4_SMOKE_LOG.md`)

### Step 1 — confirm M4 wiring is up

```bash
# ingestd's MQTT subscriber is connected
docker logs broad-announce-ingestd-1 2>&1 | tail -3
# should show: "mqtt connected" + "mqtt subscribed" + "ba/+/+/incoming"

# EMQX dashboard is reachable
curl -sS http://localhost:18083/
# EMQX 5.x landing page

# emqx is healthy
docker inspect broad-announce-emqx-1 --format '{{.State.Health.Status}}'
# "healthy"
```

### Step 2 — happy path: 1 alert via MQTT

```bash
/tmp/loadgen-mqtt \
  --broker tcp://localhost:1883 \
  --api-key acme-001:prom-prod:s3cret-acme \
  --count 1 --rate 5
```

Wait 4s, then check `deliveries`:

```bash
docker exec -i broad-announce-postgres-1 psql -U ba -d ba -c "
SELECT individual_id, channel, status FROM deliveries ORDER BY id;"
```

**Expected**: 2 rows — Alice fcm (sent) + Alice telegram
(sent). Bob has no `telegram_chat_id` linked at this point
(still in M2 default state for the smoke).

Ingestd's MQTT counters tick:

```bash
curl -sS http://localhost:8800/metrics | grep ba_ingestd_mqtt
# ba_ingestd_mqtt_messages_total{result="received",service="ingestd"} <N+1>
# ba_ingestd_mqtt_messages_total{result="accepted",service="ingestd"} <M+1>
```

### Step 3 — 5 alerts, mixed severity, 30% dedupe

```bash
/tmp/loadgen-mqtt \
  --broker tcp://localhost:1883 \
  --api-key acme-001:prom-prod:s3cret-acme \
  --count 5 --rate 5 --mode normal
```

**Expected**: 5 distinct `alert_id`s, ≥ 10 deliveries
(5 alerts × 2 channels for Alice; +4 per alert with
severity ≥ critical for Bob). At least 1 dedupe (30% of
alerts share a `dedupe_key`).

Ingestd counter `mqtt_messages{result="deduped"}` advances
by ≥ 1.

### Step 4 — bad signature

Build the test binary (script does it automatically):

```bash
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) { os.Exit(0) }
  fmt.Println("bad-sig published; err:", tok.Error())
  os.Exit(0)
}
GO
(cd loadgen && CGO_ENABLED=0 go build -o /tmp/m4_smoke/m4_badsig /tmp/m4_smoke/badsig.go)
/tmp/m4_smoke/m4_badsig
```

**Expected**:
- 0 deliveries (HMAC failed, alert dropped)
- ingestd log: `WARN msg="mqtt alert rejected" reason=bad_signature`
- `ba_ingestd_mqtt_messages_total{result="bad_signature"}` ticks

### Step 5 — ACL violation

```bash
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").  // <-- authed as acme source
    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)
  }
  // Try to publish to GLOBEX's topic — should be denied by ACL
  tok := c.Publish("ba/globex-002/grafana/incoming", 1, false, []byte("hack"))
  if !tok.WaitTimeout(5*time.Second) { os.Exit(0) }
  fmt.Println("published to wrong topic; err:", tok.Error())
  os.Exit(0)
}
GO
(cd loadgen && CGO_ENABLED=0 go build -o /tmp/m4_smoke/m4_acl_violation /tmp/m4_smoke/acl_violation.go)
/tmp/m4_smoke/m4_acl_violation
```

**Expected**:
- 0 deliveries (broker dropped the publish before the
  subscriber saw it)
- EMQX log: `tag: AUTHZ, msg: cannot_publish_to_topic_due_to_not_authorized, ... topic: ba/globex-002/grafana/incoming`
- `ba_ingestd_mqtt_messages_total{result="received"}` is
  **unchanged** (broker dropped before subscriber)

### Step 6 — non-JSON body

```bash
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) { os.Exit(0) }
  fmt.Println("bad-json published; err:", tok.Error())
  os.Exit(0)
}
GO
(cd loadgen && CGO_ENABLED=0 go build -o /tmp/m4_smoke/m4_badjson /tmp/m4_smoke/badjson.go)
/tmp/m4_smoke/m4_badjson
```

**Expected**:
- 0 deliveries (alert.Alert.Unmarshal failed)
- ingestd log: `WARN msg="mqtt alert rejected" reason=invalid_json`
- `ba_ingestd_mqtt_messages_total{result="invalid_json"}` ticks

## What this proves about M4

1. **Per-source ACLs work.** The `acl.conf` file uses
   Erlang-term rules keyed on the username. A user can
   publish only to its own `ba/<co>/<src>/incoming`. Step 5
   proves a different source can't publish to another
   source's topic — the broker drops the publish before any
   subscriber sees it.
2. **The same pipeline serves HTTP and MQTT.** Steps 2 + 3
   show the same `ProcessAlert` path that ran M0–M3's
   HTTP-only smoke now runs the MQTT path with no
   code changes to the rest of the stack (rate limits,
   dedupe, NATS publish, alert.ID, alert.DedupeCount all
   behave identically).
3. **Per-message HMAC is enforced.** Step 4 proves that
   even an authenticated source can't bypass per-message
   auth by sending an arbitrary body.
4. **Topic parsing is strict.** `parseIncomingTopic` only
   accepts `ba/<co>/<src>/incoming` (4 segments, last =
   `incoming`). Anything else lands in
   `mqtt_messages{result="bad_topic"}` and is dropped. The
   broker's ACL is the first line of defense; this is the
   second.
5. **Dedupe window is per-source, not per-transport.** The
   same `dedupe.Deduper` (Redis-backed) is reused for HTTP
   and MQTT. Step 3's 30% dedupe ratio produces the
   expected dedupe count in the deliveries table.
6. **Metrics are per-message.** `ba_ingestd_mqtt_messages_total{result="..."}`
   exposes received/accepted/deduped/bad_signature/bad_topic/
   invalid_json/... as separate labels. M9 promotes these
   to the alerts-received total.

## Known quirks (and why the smoke script does what it does)

1. **faketgmd's `getUpdates` is a fake** (carried over from
   M3). The smoke script still calls
   `docker compose restart telegramd` between steps so the
   bot's in-process long-poll offset aligns with the reset
   state. Same quirk; same workaround.
2. **EMQX's `deny_action = disconnect`** is set, but in
   practice the paho client may reconnect fast enough that
   the test code sees "still connected" after a denied
   publish. The end-state is the same: the publish was
   dropped, no delivery happened, and `received` counter
   didn't tick.
3. **The 3 failure-path test binaries** (`m4_badsig`,
   `m4_acl_violation`, `m4_badjson`) are tiny, statically
   linked, and live in `/tmp/m4_smoke/`. The smoke script
   auto-builds them on first run. They are not committed
   to the repo.
4. **EMQX env-var config beats `emqx.conf`.** Earlier
   versions of the compose tried to volume-mount
   `emqx.conf`; EMQX 5.x rejects partial overrides there
   with `node.cookie required_field` (because `emqx.conf`
   is a *full* config, not a partial). The current
   compose uses env vars
   (`EMQX_AUTHENTICATION__1__BACKEND`,
   `EMQX_AUTHORIZATION__NO_MATCH`, etc.) and the volume
   mount is just `acl.conf` + the bootstrap CSV.

## What stays out of M4 (and not supposed to be)

- Per-IP concurrency cap (M5 with WS, SPEC §22 layer 2)
- Circuit breaker + quarantine (M9)
- TLS to EMQX (M11, security milestone)
- Persistent sessions (M11)
- Per-company bot token resolution for sources that share
  a single EMQX user across companies (M3+ generalization)
- HTTP-style signature in a header (MQTT has no headers;
  the `auth` field in the JSON envelope is the equivalent)