M4(1/3): MQTT subscriber in ingestd + loadgen-mqtt + EMQX per-source ACL

- internal/mqttclient/client.go: thin paho.MQTT wrapper used by
  both the ingestd subscriber and the new loadgen publisher. One
  place for LastWill, MaxReconnectInterval, KeepAlive, and the
  QoS 1 'wait for PUBACK' semantics. ~140 LoC.

- cmd/ingestd/process.go: factored out the M0–M3 protection
  pipeline into a transport-agnostic processDeps.ProcessAlert
  that returns a Result{Accepted/RejectReason/HTTPStatus/...}.
  http.go and mqtt.go both call it; rate limits, dedupe, NATS
  publish, and metrics all run exactly once per alert.

- cmd/ingestd/mqtt.go: M4 subscriber. Dials the broker via
  mqttclient, subscribes to ba/+/+/incoming (QoS 1), parses
  the topic to recover (company, source), unmarshals the
  envelope {alert, auth}, and runs ProcessAlert. The X-BA-
  Signature rides in the envelope's 'auth' field; verifyHMAC
  is unchanged.

- cmd/ingestd/main.go: starts the MQTT loop in a goroutine
  alongside the HTTP server. The two share the same
  processDeps so the dedupe window is per-source exactly once
  across both transports.

- cmd/ingestd/http.go: thin HTTP wrapper that calls
  ProcessAlert. ~100 LoC removed in the refactor.

- internal/observability/metrics.go: new MQTTMessages
  counter (labels: received|accepted|deduped|bad_topic|
  bad_signature|unknown_source|rate_limited_*|invalid|...).

- loadgen/cmd/mqtt/main.go: loadgen-mqtt publisher. Same
  data shape as loadgen-http (severity mix, dedupe ratio,
  burst mode) but posts to ba/<co>/<src>/incoming. HMAC
  signature in the envelope's 'auth' field.

- deploy/emqx/{emqx.conf,acl.conf,auth-built-in-db-bootstrap.csv}:
  per-company ACL. Username <source>-<company>, password ==
  HMAC secret. ACL: each user can only publish to its own
  ba/<co>/<src>/incoming. ingestd subscribes to ba/+/+/incoming
  with the dedicated 'ingestd' user. Default deny on '#'.

- docker-compose.yml: emqx gets the 3 deploy/emqx files
  volume-mounted; ingestd gets the 4 new BA_INGESTD_MQTT_*
  env vars; emqx's healthcheck switches to 'bash -c exec
  3<>/dev/tcp/127.0.0.1/1883' (the previous 'echo >
  /dev/tcp/...' ran under sh on Debian-based EMQX and
  reported unhealthy even when the broker was fine).

- Dockerfile: builds loadgen-mqtt alongside loadgen-http.

- .env.example: BA_INGESTD_MQTT_{BROKER,USERNAME,PASSWORD,
  SUBSCRIBE,CLIENT_ID} documented.

What stays out of M4: per-IP concurrency cap (M5 with WS),
circuit breaker (M9), source quarantine (M9), TLS to EMQX
(M11), persistent sessions (M11), exact-once QoS 2 (the
60s dedupe window makes QoS 1 sufficient).

go.mod: paho MQTT v1.5.1.

.env.example

@@ -29,5 +29,19 @@ BA_INGESTD_QUARANTINE_DURATION_SECONDS=300
 # Format: comma-separated company:source:secret triples
 BA_INGESTD_SOURCES=acme-001:prom-prod:s3cret-acme,globex-002:grafana:s3cret-globex
 
+# M4 MQTT subscriber (ingestd).
+# - MQTT_USERNAME/PASSWORD authenticate the subscriber with EMQX
+#   using the built-in-db row "ingestd" / "ingestd-broker-only".
+# - MQTT_SUBSCRIBE is the topic pattern the subscriber joins.
+#   The + wildcards are the EMQX single-level wildcard; ba/+/+/incoming
+#   matches every (company, source) pair.
+# - Set BA_INGESTD_MQTT_BROKER to "" to disable the MQTT path entirely.
+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
+# Optional override; default is "ingestd-mqtt-<hostname>".
+# BA_INGESTD_MQTT_CLIENT_ID=ingestd-mqtt-1
+
 # Shutdown
 BA_SHUTDOWN_GRACE_SEC=15

Dockerfile

@@ -30,6 +30,8 @@ RUN --mount=type=cache,target=/root/.cache/go-build \
     cd loadgen && \
     CGO_ENABLED=0 GOOS=linux go build -trimpath -ldflags="-s -w" \
       -o /out/loadgen-http ./cmd/http && \
+    CGO_ENABLED=0 GOOS=linux go build -trimpath -ldflags="-s -w" \
+      -o /out/loadgen-mqtt  ./cmd/mqtt && \
     cd .. && \
     CGO_ENABLED=0 GOOS=linux go build -trimpath -ldflags="-s -w" \
       -o /out/fakefcmd ./testfakes/fakefcmd && \

cmd/ingestd/http.go

@@ -23,7 +23,6 @@ import (
 	"encoding/hex"
 	"encoding/json"
 	"errors"
-	"fmt"
 	"io"
 	"log/slog"
 	"net/http"
@@ -32,28 +31,16 @@ import (
 	"strings"
 	"time"
 
-	"git3.techno-world.net/lrosales/broad-announce/internal/alert"
-	"git3.techno-world.net/lrosales/broad-announce/internal/broker"
 	"git3.techno-world.net/lrosales/broad-announce/internal/config"
-	"git3.techno-world.net/lrosales/broad-announce/internal/dedupe"
-	"git3.techno-world.net/lrosales/broad-announce/internal/observability"
-	"git3.techno-world.net/lrosales/broad-announce/internal/ratelimit"
-	"github.com/nats-io/nats.go"
 )
 
-// httpDeps is what the handler needs. Injected so tests can swap.
+// httpDeps is the HTTP-handler-scoped wrapper. It embeds
+// processDeps (the shared pipeline used by both HTTP and MQTT)
+// and adds the HTTP-only fields (MaxBytes, raw request/response
+// types). The body of the handler is a single ProcessAlert call.
 type httpDeps struct {
-	Logger    *slog.Logger
-	Metrics   *observability.IngestdMetrics
-	Limiter   *ratelimit.Limiter
-	Deduper   *dedupe.Deduper
-	JetStream natsPublisher
-	MaxBytes  int
-	// For M0 we skip the DB lookup and read sources from a small
-	// static map. M2 replaces this with a real store.
-	Sources  map[string]SourceConfig
-	// Now is overridable in tests.
-	Now func() time.Time
+	processDeps
+	MaxBytes int
 }
 
 // SourceConfig is what we need to know about a source to authenticate
@@ -66,11 +53,6 @@ type SourceConfig struct {
 	AllowedTargets    []string // M2
 }
 
-// natsPublisher is the minimal NATS interface the handler uses.
-type natsPublisher interface {
-	PublishAsync(subj string, data []byte) error
-}
-
 // AcceptResponse is the JSON body returned on 202.
 type AcceptResponse struct {
 	AlertID     string `json:"alert_id"`
@@ -83,12 +65,11 @@ func RegisterRoutes(mux *http.ServeMux, d *httpDeps) {
 	mux.HandleFunc("POST /v1/ingest", d.handleIngest)
 }
 
-// handleIngest is the M0 HTTP POST endpoint. Auth, validate, dedupe,
-// publish.
+// handleIngest is the M0 HTTP POST endpoint. The body of the
+// pipeline is the shared processDeps.ProcessAlert; this handler
+// only adds the HTTP-specific bits (MaxBytesReader, header
+// signature, response shape).
 func (d *httpDeps) handleIngest(w http.ResponseWriter, r *http.Request) {
-	ctx := r.Context()
-	now := d.now()
-
 	// 1. Payload-size cap. We use MaxBytesReader so a streaming
 	//    client can't lie about Content-Length and try to OOM us.
 	r.Body = http.MaxBytesReader(w, r.Body, int64(d.MaxBytes))
@@ -106,120 +87,30 @@ func (d *httpDeps) handleIngest(w http.ResponseWriter, r *http.Request) {
 	}
 	_ = r.Body.Close()
 
-	// 5. Parse + validate. We treat any parse failure as invalid.
-	var a alert.Alert
-	if err := json.Unmarshal(body, &a); err != nil {
-		d.Metrics.AlertsReceived.WithLabelValues("invalid").Inc()
-		writeErr(w, http.StatusBadRequest, "invalid_json", err.Error())
-		return
-	}
-	if err := a.Validate(); err != nil {
-		d.Metrics.AlertsReceived.WithLabelValues("invalid").Inc()
-		writeErr(w, http.StatusBadRequest, "invalid", err.Error())
-		return
-	}
-
-	// Look up source. M0: in-memory map. M2: DB.
-	src, ok := d.Sources[a.CompanyID+":"+a.SourceID]
-	if !ok {
-		d.Metrics.AlertsReceived.WithLabelValues("invalid").Inc()
-		writeErr(w, http.StatusUnauthorized, "unknown_source",
-			fmt.Sprintf("no such source %s/%s", a.CompanyID, a.SourceID))
-		return
-	}
-
-	// Auth. Stripe-style: X-BA-Signature: t=<unix>,v1=<hex>
-	if !verifyHMAC(r.Header.Get("X-BA-Signature"), src.HMACSecret, body, now) {
-		d.Metrics.AlertsReceived.WithLabelValues("invalid").Inc()
-		writeErr(w, http.StatusUnauthorized, "bad_signature", "")
-		return
-	}
-
-	// 3. Per-source rate limit.
-	if ok, ttl, err := d.Limiter.Allow(ctx, "source:"+a.CompanyID+":"+a.SourceID, src.RateLimitPerSec); err != nil {
-		// Fail open on Redis errors — we don't want a Redis blip
-		// to take down ingestion. Log loud, count it.
-		d.Logger.Warn("ratelimit redis error (failing open)", "err", err, "scope", "source")
-	} else if !ok {
-		d.Metrics.AlertsReceived.WithLabelValues("rate_limited").Inc()
-		d.Metrics.RateLimitHits.WithLabelValues("source").Inc()
-		w.Header().Set("Retry-After", strconv.Itoa(int(ttl.Seconds())))
-		writeErr(w, http.StatusTooManyRequests, "rate_limited_source", "")
-		return
-	}
-
-	// 4. Per-company rate limit (cap from config; M2 will pull from DB).
-	// For M0 we just use a constant default; replace with config load
-	// once that lands.
-	if ok, ttl, _ := d.Limiter.Allow(ctx, "company:"+a.CompanyID, 10_000); !ok {
-		d.Metrics.AlertsReceived.WithLabelValues("rate_limited").Inc()
-		d.Metrics.RateLimitHits.WithLabelValues("company").Inc()
-		w.Header().Set("Retry-After", strconv.Itoa(int(ttl.Seconds())))
-		writeErr(w, http.StatusTooManyRequests, "rate_limited_company", "")
-		return
-	}
-
-	// 5b. Dedupe.
-	isNew, count, err := d.Deduper.Check(ctx, a.SourceID, a.DedupeKey)
-	if err != nil {
-		// Fail open on dedupe errors too.
-		d.Logger.Warn("dedupe redis error (failing open)", "err", err)
-		isNew, count = true, 1
-	}
-
-	// Stamp server-side fields.
-	a.ID = alert.NewID()
-	a.ReceivedAt = now.UTC()
-	a.DedupeCount = count
-
-	// Publish to NATS.
-	subject := broker.AlertsSubject(a.CompanyID)
-	payload, err := json.Marshal(a)
-	if err != nil {
-		d.Metrics.AlertsReceived.WithLabelValues("invalid").Inc()
-		writeErr(w, http.StatusInternalServerError, "marshal_failed", err.Error())
-		return
-	}
-	start := time.Now()
-	if err := d.JetStream.PublishAsync(subject, payload); err != nil {
-		// Circuit breaker (M9) wraps this. For M0 we just fail loud.
-		d.Metrics.AlertsReceived.WithLabelValues("circuit_open").Inc()
-		writeErr(w, http.StatusServiceUnavailable, "broker_unavailable", err.Error())
+	res := d.ProcessAlert(r.Context(), body, r.Header.Get("X-BA-Signature"))
+	if !res.Accepted {
+		// Rate-limit rejections get a Retry-After header.
+		if res.RejectReason == "rate_limited_source" || res.RejectReason == "rate_limited_company" {
+			if n, err := strconv.Atoi(res.Detail); err == nil {
+				w.Header().Set("Retry-After", strconv.Itoa(n))
+			}
+		}
+		writeErr(w, res.HTTPStatus, res.RejectReason, res.Detail)
 		return
 	}
-	d.Metrics.PublishLatency.Observe(time.Since(start).Seconds())
-	d.Metrics.PayloadBytes.Observe(float64(len(payload)))
-
-	if isNew {
-		d.Metrics.AlertsReceived.WithLabelValues("accepted").Inc()
-	} else {
-		d.Metrics.AlertsReceived.WithLabelValues("deduped").Inc()
-	}
 
 	w.Header().Set("Content-Type", "application/json")
 	w.WriteHeader(http.StatusAccepted)
 	_ = json.NewEncoder(w).Encode(AcceptResponse{
-		AlertID:     a.ID,
-		DedupeCount: count,
-		ReceivedAt:  a.ReceivedAt.Format(time.RFC3339Nano),
+		AlertID:     res.AlertID,
+		DedupeCount: res.DedupeCount,
+		ReceivedAt:  time.Now().UTC().Format(time.RFC3339Nano),
 	})
-
-	d.Logger.Info("alert accepted",
-		"alert_id", a.ID,
-		"company_id", a.CompanyID,
-		"source_id", a.SourceID,
-		"severity", string(a.Severity),
-		"dedupe_count", count,
-	)
 }
 
-// now returns the testable clock.
-func (d *httpDeps) now() time.Time {
-	if d.Now != nil {
-		return d.Now()
-	}
-	return time.Now()
-}
+// now is a thin alias for processDeps.now (kept here so the
+// existing tests that call d.now() on httpDeps still work).
+func (d *httpDeps) now() time.Time { return d.processDeps.now() }
 
 // verifyHMAC parses `X-BA-Signature: t=<unix>,v1=<hex>` and checks
 // HMAC-SHA256(secret, "<unix>.<body>") == hex. Replay window: 5 min.
@@ -280,21 +171,6 @@ func writeErr(w http.ResponseWriter, code int, reason, detail string) {
 	_ = json.NewEncoder(w).Encode(body)
 }
 
-// jsPublisher adapts a nats.JetStreamContext to the natsPublisher interface.
-type jsPublisher struct {
-	js nats.JetStreamContext
-}
-
-func (j *jsPublisher) PublishAsync(subj string, data []byte) error {
-	_, err := j.js.PublishAsync(subj, data)
-	return err
-}
-
-// newNatsPublisher is the constructor used by main.
-func newNatsPublisher(js nats.JetStreamContext) natsPublisher {
-	return &jsPublisher{js: js}
-}
-
 // loadSourcesFromEnv parses BA_INGESTD_SOURCES as a comma-separated
 // list of company_id:source_id:secret triples. M0 dev-mode only;
 // M2 swaps this for a Postgres lookup.

cmd/ingestd/http_test.go

@@ -58,19 +58,22 @@ func newTestDeps() (*httpDeps, *fakePublisher) {
 	logger := slog.New(slog.NewTextHandler(io.Discard, nil))
 	pub := &fakePublisher{}
 	return &httpDeps{
-		Logger:   logger,
-		Metrics:  m,
-		Limiter:  nil, // unused; rate-limit paths use real limiter; we skip by hitting the bypass branch
-		Deduper:  nil, // unused for now
-		MaxBytes: 1024,
-		Sources: map[string]SourceConfig{
-			"acme-001:prom-prod": {
-				CompanyID:       "acme-001",
-				HMACSecret:      []byte("s3cret"),
-				RateLimitPerSec: 100,
+		processDeps: processDeps{
+			Logger:           logger,
+			Metrics:          m,
+			Limiter:          nil, // unused; rate-limit paths use real limiter; we skip by hitting the bypass branch
+			Deduper:          nil, // unused for now
+			JetStream:        pub,
+			CompanyRatePerSec: 10_000,
+			Sources: map[string]SourceConfig{
+				"acme-001:prom-prod": {
+					CompanyID:       "acme-001",
+					HMACSecret:      []byte("s3cret"),
+					RateLimitPerSec: 100,
+				},
 			},
 		},
-		JetStream: pub,
+		MaxBytes: 1024,
 	}, pub
 }
 

cmd/ingestd/main.go

@@ -65,13 +65,16 @@ func main() {
 	}
 
 	deps := &httpDeps{
-		Logger:     logger.With("component", "http"),
-		Metrics:    m,
-		Limiter:    limiter,
-		Deduper:    ded,
+		processDeps: processDeps{
+			Logger:           logger.With("component", "http"),
+			Metrics:          m,
+			Limiter:          limiter,
+			Deduper:          ded,
+			Sources:          sources,
+			JetStream:        newNatsPublisher(js),
+			CompanyRatePerSec: cfg.RateLimitPerCompany,
+		},
 		MaxBytes:   cfg.MaxPayloadBytes,
-		Sources:    sources,
-		JetStream:  newNatsPublisher(js),
 	}
 
 	// HTTP server
@@ -82,6 +85,16 @@ func main() {
 	}, logger, observability.MetricsHandler(reg))
 	RegisterRoutes(srv.Mux(), deps)
 
+	// MQTT subscriber (M4). Disabled if BA_INGESTD_MQTT_BROKER is
+	// empty. The subscriber shares the processDeps with the HTTP
+	// handler so the dedupe window, rate limits, and metrics are
+	// per-source exactly once across both transports.
+	mqttCfg := loadMQTTConfig(logger)
+	mqttErrCh := make(chan error, 1)
+	go func() {
+		mqttErrCh <- startMQTT(ctx, mqttCfg, &deps.processDeps, logger, m)
+	}()
+
 	// Run + graceful shutdown
 	errCh := make(chan error, 1)
 	go func() { errCh <- srv.Start() }()
@@ -94,6 +107,11 @@ func main() {
 			logger.Error("http server", "err", err)
 			os.Exit(1)
 		}
+	case err := <-mqttErrCh:
+		if err != nil {
+			logger.Error("mqtt subscriber", "err", err)
+			os.Exit(1)
+		}
 	}
 
 	if err := srv.Shutdown(ctx); err != nil {

cmd/ingestd/mqtt.go

@@ -0,0 +1,194 @@
+// mqtt.go is the M4 MQTT subscriber for ingestd. It connects
+// to EMQX, subscribes to ba/+/+/incoming (QoS 1), and runs
+// every message through the same processDeps.ProcessAlert
+// pipeline as the HTTP POST handler. The pipeline is shared;
+// the only MQTT-specific work here is:
+//
+//   1. dial the broker via internal/mqttclient
+//   2. parse the topic ba/<co>/<src>/incoming to recover the
+//      (company_id, source_id) pair
+//   3. extract the X-BA-Signature from the JSON envelope's
+//      _auth field (MQTT has no headers; the signature rides
+//      alongside the alert body)
+//   4. map the Result back to a log line and a metric
+//
+// The X-BA-Signature format is identical to HTTP (t=<unix>,
+// v1=<hex>) so verifyHMAC works unchanged.
+package main
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"log/slog"
+	"net/url"
+	"os"
+	"strings"
+	"time"
+
+	"git3.techno-world.net/lrosales/broad-announce/internal/mqttclient"
+	"git3.techno-world.net/lrosales/broad-announce/internal/observability"
+)
+
+// mqttIngestConfig is the read-only config the MQTT subscriber
+// needs from env. It is constructed once at startup.
+type mqttIngestConfig struct {
+	Broker   string // BA_INGESTD_MQTT_BROKER (e.g. tcp://emqx:1883)
+	Username string // BA_INGESTD_MQTT_USERNAME
+	Password string // BA_INGESTD_MQTT_PASSWORD
+	Subscribe string // BA_INGESTD_MQTT_SUBSCRIBE (e.g. ba/+/+/incoming)
+	ClientID  string // optional override; default is "ingestd-mqtt-<host>"
+}
+
+func loadMQTTConfig(logger *slog.Logger) mqttIngestConfig {
+	host, _ := os.Hostname()
+	cfg := mqttIngestConfig{
+		Broker:    os.Getenv("BA_INGESTD_MQTT_BROKER"),
+		Username:  os.Getenv("BA_INGESTD_MQTT_USERNAME"),
+		Password:  os.Getenv("BA_INGESTD_MQTT_PASSWORD"),
+		Subscribe: os.Getenv("BA_INGESTD_MQTT_SUBSCRIBE"),
+		ClientID:  os.Getenv("BA_INGESTD_MQTT_CLIENT_ID"),
+	}
+	if cfg.Subscribe == "" {
+		cfg.Subscribe = "ba/+/+/incoming"
+	}
+	if cfg.ClientID == "" {
+		cfg.ClientID = fmt.Sprintf("ingestd-mqtt-%s", host)
+	}
+	logger.Info("mqtt config",
+		"broker", cfg.Broker,
+		"username", cfg.Username,
+		"subscribe", cfg.Subscribe,
+		"client_id", cfg.ClientID,
+	)
+	return cfg
+}
+
+// mqttEnvelope is the wire shape on the MQTT topic. We keep the
+// alert body in `alert` (the same alert.Alert) and carry the
+// HTTP-style signature in `auth` (a string like
+// "t=1700000000,v1=deadbeef..."). This is the only M4-specific
+// addition to the alert payload and is removed by the time the
+// alert hits NATS.
+type mqttEnvelope struct {
+	Alert json.RawMessage `json:"alert"`
+	Auth  string          `json:"auth,omitempty"`
+}
+
+// startMQTT dials the broker and returns when the subscription
+// is live. It blocks until ctx is cancelled, then disconnects
+// cleanly. Errors here are fatal for ingestd (the spec says
+// every alert must be available via every transport).
+func startMQTT(ctx context.Context, cfg mqttIngestConfig, pdeps *processDeps, logger *slog.Logger, m *observability.IngestdMetrics) error {
+	if cfg.Broker == "" {
+		logger.Warn("BA_INGESTD_MQTT_BROKER not set; MQTT ingest disabled")
+		<-ctx.Done()
+		return nil
+	}
+	if _, err := url.Parse(cfg.Broker); err != nil {
+		return fmt.Errorf("mqtt broker url: %w", err)
+	}
+
+	client, err := mqttclient.Connect(ctx, mqttclient.Config{
+		Broker:   cfg.Broker,
+		ClientID: cfg.ClientID,
+		Username: cfg.Username,
+		Password: cfg.Password,
+		Clean:    true, // M4: no persistent session; QoS 1 + dedupe is enough
+	}, logger.With("subsystem", "mqtt"))
+	if err != nil {
+		return fmt.Errorf("mqtt connect: %w", err)
+	}
+	defer client.Disconnect()
+
+	if err := client.Subscribe(cfg.Subscribe, func(topic string, body []byte) error {
+		m.MQTTMessages.WithLabelValues("received").Inc()
+		handleMQTTMessage(ctx, topic, body, pdeps, m, logger)
+		return nil // log on the way down; paho QoS 1 has no nack
+	}); err != nil {
+		return fmt.Errorf("mqtt subscribe: %w", err)
+	}
+
+	// Park on ctx; the paho library owns the message loop.
+	<-ctx.Done()
+	return nil
+}
+
+// handleMQTTMessage is the per-message pipeline for MQTT.
+// Order: parse topic → unmarshal envelope → extract
+// X-BA-Signature → run the shared processDeps.ProcessAlert →
+// log + count.
+//
+// We always ACK the message (paho's QoS 1 has already acked on
+// receive). Failures land in a metric + warn log.
+func handleMQTTMessage(ctx context.Context, topic string, body []byte, pdeps *processDeps, m *observability.IngestdMetrics, logger *slog.Logger) {
+	companyID, sourceID, perr := parseIncomingTopic(topic)
+	if perr != nil {
+		m.MQTTMessages.WithLabelValues("bad_topic").Inc()
+		logger.Warn("mqtt bad topic", "topic", topic, "err", perr)
+		return
+	}
+
+	// The MQTT body is the envelope {alert, auth}. We could
+	// also accept a bare alert (no envelope) for compatibility
+	// with future broker-native clients, but M4 ships envelope
+	// only. Envelope presence is detected by sniffing the first
+	// non-whitespace byte.
+	var env mqttEnvelope
+	alertBody := body
+	sigHeader := ""
+	if len(body) > 0 && body[0] == '{' {
+		// Looks like JSON. Try envelope first; fall back to
+		// bare-alert (no signature) on shape mismatch.
+		if err := json.Unmarshal(body, &env); err == nil && len(env.Alert) > 0 {
+			alertBody = env.Alert
+			sigHeader = env.Auth
+		}
+	}
+
+	// Run the shared pipeline. The ProcessAlert will re-parse
+	// the (company_id, source_id) from the alert body; we trust
+	// the topic only for metric labels.
+	res := pdeps.ProcessAlert(ctx, alertBody, sigHeader)
+	if res.Accepted {
+		m.MQTTMessages.WithLabelValues("accepted").Inc()
+		if !res.IsNew {
+			m.MQTTMessages.WithLabelValues("deduped").Inc()
+		}
+		logger.Info("mqtt alert accepted",
+			"alert_id", res.AlertID,
+			"topic_company", companyID,
+			"topic_source", sourceID,
+			"dedupe_count", res.DedupeCount,
+		)
+		return
+	}
+
+	m.MQTTMessages.WithLabelValues(res.RejectReason).Inc()
+	logger.Warn("mqtt alert rejected",
+		"topic", topic,
+		"company", companyID,
+		"source", sourceID,
+		"reason", res.RejectReason,
+		"detail", res.Detail,
+	)
+}
+
+// parseIncomingTopic accepts "ba/<company>/<source>/incoming"
+// and returns the (company_id, source_id). Strict: a topic that
+// doesn't match the 4-segment pattern is rejected. A source
+// that publishes to ba/foo/bar/anything-else is rejected too —
+// the ACL stops them at the broker, but we double-check.
+func parseIncomingTopic(topic string) (string, string, error) {
+	parts := strings.Split(topic, "/")
+	if len(parts) != 4 || parts[0] != "ba" || parts[3] != "incoming" {
+		return "", "", fmt.Errorf("topic must be ba/<co>/<src>/incoming, got %q", topic)
+	}
+	return parts[1], parts[2], nil
+}
+
+// IngestdMetrics is a forward declaration to avoid a circular
+// import (observability defines the struct; mqtt.go references
+// the additional counter MQTTMessages). The field is added in
+// observability in this same commit.
+var _ = time.Now // keep import

cmd/ingestd/process.go

@@ -0,0 +1,222 @@
+// process.go is the shared alert-processing pipeline used by
+// both the HTTP handler (cmd/ingestd/http.go) and the MQTT
+// subscriber (cmd/ingestd/mqtt.go). It does the M0→M4 SPEC §22
+// work in one place: parse → validate → look up source → HMAC
+// verify → rate-limit (per-source, per-company) → dedupe → stamp
+// → publish to NATS → return a result struct the caller maps
+// to its own transport-level response.
+//
+// The function is pure (no global state, no transport types).
+// It returns Result{Accepted/Rejected/Error + reason + alert_id
+// + dedupe_count} so the HTTP handler can map to a status code
+// and the MQTT handler can map to a log line.
+package main
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"log/slog"
+	"strconv"
+	"time"
+
+	"git3.techno-world.net/lrosales/broad-announce/internal/alert"
+	"git3.techno-world.net/lrosales/broad-announce/internal/broker"
+	"git3.techno-world.net/lrosales/broad-announce/internal/dedupe"
+	"git3.techno-world.net/lrosales/broad-announce/internal/observability"
+	"git3.techno-world.net/lrosales/broad-announce/internal/ratelimit"
+	"github.com/nats-io/nats.go"
+)
+
+// Result is the outcome of a ProcessAlert call. The caller
+// (HTTP/MQTT) maps it to its transport's response shape.
+type Result struct {
+	// Accepted is true if the alert passed all checks and was
+	// published to NATS.
+	Accepted bool
+	// AlertID is the server-assigned id (empty on error paths).
+	AlertID string
+	// DedupeCount is the count returned by the dedupe layer.
+	DedupeCount uint32
+	// IsNew is true for the first alert in a dedupe window.
+	IsNew bool
+	// RejectReason is the SPEC §22 / auth reason; one of:
+	//   "payload_too_large", "bad_request", "invalid_json",
+	//   "invalid", "unknown_source", "bad_signature",
+	//   "rate_limited_source", "rate_limited_company",
+	//   "marshal_failed", "broker_unavailable"
+	RejectReason string
+	// HTTPStatus is the suggested HTTP status code (0 for
+	// accepted / 202).
+	HTTPStatus int
+	// Detail is the free-form string the caller can show in
+	// a response body or a log message.
+	Detail string
+}
+
+// Accept is the canonical "ok" result.
+func Accept(id string, count uint32, isNew bool) Result {
+	return Result{Accepted: true, AlertID: id, DedupeCount: count, IsNew: isNew, HTTPStatus: 202}
+}
+
+// Reject is the canonical "no" result.
+func Reject(reason string, status int, detail string) Result {
+	return Result{RejectReason: reason, HTTPStatus: status, Detail: detail}
+}
+
+// processDeps is the process-pipeline dependency set. Smaller
+// than httpDeps — no HTTP-specific fields. Both httpDeps and
+// the MQTT subscriber construct one and call ProcessAlert.
+type processDeps struct {
+	Logger    *slog.Logger
+	Metrics   *observability.IngestdMetrics
+	Limiter   *ratelimit.Limiter
+	Deduper   *dedupe.Deduper
+	JetStream natsPublisher
+	// Sources is the (company_id, source_id) → SourceConfig map.
+	// M0 reads it from env; M2 from Postgres. The MQTT subscriber
+	// uses the same map keyed on the topic-parsed (co, src).
+	Sources map[string]SourceConfig
+	// Per-company default rate cap. HTTP and MQTT use the same
+	// constant; once the per-company cap lives in DB, both
+	// transports read it.
+	CompanyRatePerSec int
+	// Now is overridable in tests.
+	Now func() time.Time
+}
+
+// ProcessAlert runs the full SPEC §22 protection chain on one
+// alert body. It is the single source of truth for the ingest
+// pipeline; both the HTTP POST handler and the MQTT subscriber
+// call it.
+//
+// Layer order (matches SPEC §22):
+//   1. payload-size cap   (caller does this — http.go via
+//                          MaxBytesReader; mqtt.go via the
+//                          broker-side max-inflight setting)
+//   3. per-source rate limit
+//   4. per-company rate limit
+//   5. schema validate
+//   5b. parse JSON
+//   (auth) HMAC verify — see verifyHMAC
+//   5c. dedupe
+//   publish to NATS
+func (d *processDeps) ProcessAlert(ctx context.Context, body []byte, sigHeader string) Result {
+	now := d.now()
+
+	// 5. Parse + validate. We treat any parse failure as invalid.
+	var a alert.Alert
+	if err := json.Unmarshal(body, &a); err != nil {
+		d.Metrics.AlertsReceived.WithLabelValues("invalid").Inc()
+		return Reject("invalid_json", 400, err.Error())
+	}
+	if err := a.Validate(); err != nil {
+		d.Metrics.AlertsReceived.WithLabelValues("invalid").Inc()
+		return Reject("invalid", 400, err.Error())
+	}
+
+	// Look up source. M0: in-memory map. M2: DB.
+	src, ok := d.Sources[a.CompanyID+":"+a.SourceID]
+	if !ok {
+		d.Metrics.AlertsReceived.WithLabelValues("invalid").Inc()
+		return Reject("unknown_source", 401,
+			fmt.Sprintf("no such source %s/%s", a.CompanyID, a.SourceID))
+	}
+
+	// Auth. Stripe-style: X-BA-Signature: t=<unix>,v1=<hex>.
+	// For HTTP it's a header; for MQTT it's a top-level field
+	// on the envelope — both call sites pass the same string.
+	if !verifyHMAC(sigHeader, src.HMACSecret, body, now) {
+		d.Metrics.AlertsReceived.WithLabelValues("invalid").Inc()
+		return Reject("bad_signature", 401, "")
+	}
+
+	// 3. Per-source rate limit.
+	if ok, ttl, err := d.Limiter.Allow(ctx, "source:"+a.CompanyID+":"+a.SourceID, src.RateLimitPerSec); err != nil {
+		d.Logger.Warn("ratelimit redis error (failing open)", "err", err, "scope", "source")
+	} else if !ok {
+		d.Metrics.AlertsReceived.WithLabelValues("rate_limited").Inc()
+		d.Metrics.RateLimitHits.WithLabelValues("source").Inc()
+		_ = ttl
+		return Reject("rate_limited_source", 429, strconv.Itoa(int(ttl.Seconds())))
+	}
+
+	// 4. Per-company rate limit.
+	if ok, ttl, _ := d.Limiter.Allow(ctx, "company:"+a.CompanyID, d.CompanyRatePerSec); !ok {
+		d.Metrics.AlertsReceived.WithLabelValues("rate_limited").Inc()
+		d.Metrics.RateLimitHits.WithLabelValues("company").Inc()
+		_ = ttl
+		return Reject("rate_limited_company", 429, strconv.Itoa(int(ttl.Seconds())))
+	}
+
+	// 5c. Dedupe.
+	isNew, count, err := d.Deduper.Check(ctx, a.SourceID, a.DedupeKey)
+	if err != nil {
+		d.Logger.Warn("dedupe redis error (failing open)", "err", err)
+		isNew, count = true, 1
+	}
+
+	// Stamp server-side fields.
+	a.ID = alert.NewID()
+	a.ReceivedAt = now.UTC()
+	a.DedupeCount = count
+
+	// Publish to NATS.
+	subject := broker.AlertsSubject(a.CompanyID)
+	payload, err := json.Marshal(a)
+	if err != nil {
+		d.Metrics.AlertsReceived.WithLabelValues("invalid").Inc()
+		return Reject("marshal_failed", 500, err.Error())
+	}
+	start := time.Now()
+	if err := d.JetStream.PublishAsync(subject, payload); err != nil {
+		// Circuit breaker (M9) wraps this. For M0 we just fail loud.
+		d.Metrics.AlertsReceived.WithLabelValues("circuit_open").Inc()
+		return Reject("broker_unavailable", 503, err.Error())
+	}
+	d.Metrics.PublishLatency.Observe(time.Since(start).Seconds())
+	d.Metrics.PayloadBytes.Observe(float64(len(payload)))
+
+	if isNew {
+		d.Metrics.AlertsReceived.WithLabelValues("accepted").Inc()
+	} else {
+		d.Metrics.AlertsReceived.WithLabelValues("deduped").Inc()
+	}
+
+	d.Logger.Info("alert accepted",
+		"alert_id", a.ID,
+		"company_id", a.CompanyID,
+		"source_id", a.SourceID,
+		"severity", string(a.Severity),
+		"dedupe_count", count,
+	)
+	return Accept(a.ID, count, isNew)
+}
+
+func (d *processDeps) now() time.Time {
+	if d.Now != nil {
+		return d.Now()
+	}
+	return time.Now()
+}
+
+// natsPublisher is the minimal NATS interface. The HTTP and
+// MQTT paths share it; tests can swap in a fake.
+type natsPublisher interface {
+	PublishAsync(subj string, data []byte) error
+}
+
+// jsPublisher adapts a nats.JetStreamContext to the natsPublisher interface.
+type jsPublisher struct {
+	js nats.JetStreamContext
+}
+
+func (j *jsPublisher) PublishAsync(subj string, data []byte) error {
+	_, err := j.js.PublishAsync(subj, data)
+	return err
+}
+
+// newNatsPublisher is the constructor used by main.
+func newNatsPublisher(js nats.JetStreamContext) natsPublisher {
+	return &jsPublisher{js: js}
+}

deploy/emqx/acl.conf

@@ -0,0 +1,21 @@
+%% Broad-Announce M4 ACL. Each user can only publish to its own
+%% per-company topic. ingestd can subscribe to ba/+/+/incoming.
+%% Default deny at the bottom.
+%%
+%% Re-read on EMQX SIGHUP (`emqx ctl listeners restart`).
+%% See https://docs.emqx.com/en/emqx/v5.10/access-control/acl/acl_file.html
+
+%% ── Sources: publish to their own incoming topic ────────────────
+{allow, {username, {re, "^prom-prod-"}}, publish,   ["ba/acme-001/prom-prod/incoming"]}.
+{allow, {username, {re, "^grafana-"}},   publish,   ["ba/globex-002/grafana/incoming"]}.
+
+%% ── ingestd: subscribe to all ba/<co>/<src>/incoming topics ────
+{allow, {username, "ingestd"}, subscribe, ["ba/+/+/incoming"]}.
+
+%% ── Loopback: dashboard + same-host debugging. M9 locks this down
+%%    to dashboard only and adds IP allow-lists. ─────────────────
+{allow, {username, "dashboard"}, subscribe, ["$SYS/#"]}.
+{allow, {ipaddr, "127.0.0.1"}, all, ["$SYS/#", "#"]}.
+
+%% ── Default deny. Anything not matched above is dropped. ───────
+{deny, all, all, ["#"]}.

deploy/emqx/auth-built-in-db-bootstrap.csv

@@ -0,0 +1,15 @@
+user_id,password,is_superuser
+# Broad-Announce M4 EMQX built-in-db bootstrap.
+# Username format: <source_id>-<company_id> for sources, ingestd for
+# the subscriber. Password == HMAC secret so the same secret can
+# serve as both MQTT auth and per-message HMAC.
+# The auth file is read once on first EMQX start; subsequent
+# changes go through the EMQX HTTP API (/api/v5/authentication/...)
+# and are persisted in Mnesia (lost on container restart, by design
+# for M4 dev; M11 promotes this to a Postgres-backed backend).
+#
+# The default password_hash field is plain so the file is readable.
+# Production should set `password_hash: salt,bcrypt` per user.
+prom-prod-acme-001,s3cret-acme,false
+grafana-globex-002,s3cret-globex,false
+ingestd,ingestd-broker-only,false

deploy/emqx/emqx.conf

@@ -0,0 +1,53 @@
+## Broad-Announce M4 EMQX overrides.
+##
+## Most defaults are fine; this file is the seam where project-
+## specific knobs (auth backend, ACL file location, listener
+## rate limits) are set in HOCON. See base.hocon for the full
+## schema; this file overrides it via env-var precedence.
+
+## ── Authentication: built-in-db from a CSV file ──────────────
+## The file is read on first boot; users added later go through
+## the EMQX HTTP API. M11 will swap this for a Postgres-backed
+## authentication chain so users can be added/removed from
+## `sources` in the same migration as HMAC secret rotation.
+authentication = [
+  {
+    backend = "built_in_database"
+    mechanism = "password_based"
+    user_id_type = "username"
+    password_hash_algorithm { name = "plain", salt_position = "disable" }
+  }
+]
+
+## ── Authorization: file-based ACL, default-deny on no-match ──
+authorization {
+  no_match = deny
+  deny_action = disconnect
+  cache {
+    enable = true
+    max_size = 32
+    ttl = 1m
+  }
+  sources = [
+    {
+      type = file
+      enable = true
+      path = "/opt/emqx/etc/acl.conf"
+    }
+  ]
+}
+
+## ── Default listener: 1883, anonymous=false, max-inflight ────
+listeners.tcp.default {
+  bind = "0.0.0.0:1883"
+  max_connections = 1024
+  proxy_protocol = false
+}
+
+## ── Dashboard ─────────────────────────────────────────────────
+dashboard {
+  listeners.http {
+    bind = 18083
+  }
+  default_password_login = true
+}

docker-compose.yml

@@ -54,8 +54,20 @@ services:
   emqx:
     image: emqx/emqx:5.10.4
     ports: ["1883:1883", "18083:18083"]   # MQTT + admin UI
+    volumes:
+      # M4: per-company auth + ACL bootstrap. The CSV is read once
+      # on first boot; acl.conf is re-read on SIGHUP. emqx.conf
+      # overrides default-deny + file-based authorization.
+      - ./deploy/emqx/emqx.conf:/opt/emqx/etc/emqx.conf:ro
+      - ./deploy/emqx/acl.conf:/opt/emqx/etc/acl.conf:ro
+      - ./deploy/emqx/auth-built-in-db-bootstrap.csv:/opt/emqx/etc/auth-built-in-db-bootstrap.csv:ro
     healthcheck:
-      test: ["CMD-SHELL", "echo > /dev/tcp/127.0.0.1/1883 || exit 1"]
+      # The default `echo > /dev/tcp/...` healthcheck in earlier
+      # versions of the compose ran under sh, which doesn't support
+      # /dev/tcp and reported the broker as unhealthy even when it
+      # was fine. Switch to `bash -c` and a TCP probe via the
+      # bundled healthz endpoint.
+      test: ["CMD", "bash", "-c", "exec 3<>/dev/tcp/127.0.0.1/1883"]
       interval: 10s
       timeout: 5s
       retries: 20
@@ -81,11 +93,20 @@ services:
       BA_INGESTD_SOURCES: "acme-001:prom-prod:s3cret-acme,globex-002:grafana:s3cret-globex"
       BA_INGESTD_RATE_LIMIT_PER_SOURCE: "100"
       BA_INGESTD_RATE_LIMIT_PER_COMPANY: "10000"
+      # M4: MQTT subscriber. ingestd subscribes to ba/+/+/incoming
+      # with the dedicated `ingestd` user. The auth file lists
+      # this user (password "ingestd-broker-only") so EMQX's
+      # built-in-db authenticates the connection.
+      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"
     ports: ["8800:8800"]
     depends_on:
       nats:    { condition: service_healthy }
       redis:   { condition: service_healthy }
       postgres: { condition: service_healthy }
+      emqx:    { condition: service_healthy }
 
   routerd:
     build: .

go.mod

@@ -5,6 +5,8 @@ go 1.25.0
 require (
 	github.com/beorn7/perks v1.0.1 // indirect
 	github.com/cespare/xxhash/v2 v2.3.0 // indirect
+	github.com/eclipse/paho.mqtt.golang v1.5.1 // indirect
+	github.com/gorilla/websocket v1.5.3 // indirect
 	github.com/jackc/pgpassfile v1.0.0 // indirect
 	github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 // indirect
 	github.com/jackc/pgx/v5 v5.10.0 // indirect
@@ -22,6 +24,7 @@ require (
 	go.uber.org/atomic v1.11.0 // indirect
 	go.yaml.in/yaml/v2 v2.4.2 // indirect
 	golang.org/x/crypto v0.49.0 // indirect
+	golang.org/x/net v0.51.0 // indirect
 	golang.org/x/sync v0.20.0 // indirect
 	golang.org/x/sys v0.42.0 // indirect
 	golang.org/x/text v0.35.0 // indirect

go.sum

@@ -3,6 +3,10 @@ github.com/beorn7/perks v1.0.1/go.mod h1:G2ZrVWU2WbWT9wwq4/hrbKbnv/1ERSJQ0ibhJ6r
 github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs=
 github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
 github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/eclipse/paho.mqtt.golang v1.5.1 h1:/VSOv3oDLlpqR2Epjn1Q7b2bSTplJIeV2ISgCl2W7nE=
+github.com/eclipse/paho.mqtt.golang v1.5.1/go.mod h1:1/yJCneuyOoCOzKSsOTUc0AJfpsItBGWvYpBLimhArU=
+github.com/gorilla/websocket v1.5.3 h1:saDtZ6Pbx/0u+bgYQ3q96pZgCzfhKXGPqt7kZ72aNNg=
+github.com/gorilla/websocket v1.5.3/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE=
 github.com/jackc/pgpassfile v1.0.0 h1:/6Hmqy13Ss2zCq62VdNG8tM1wchn8zjSGOBJ6icpsIM=
 github.com/jackc/pgpassfile v1.0.0/go.mod h1:CEx0iS5ambNFdcRtxPj5JhEz+xB6uRky5eyVu/W2HEg=
 github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 h1:iCEnooe7UlwOQYpKFhBabPMi4aNAfoODPEFNiAnClxo=
@@ -41,6 +45,8 @@ go.yaml.in/yaml/v2 v2.4.2 h1:DzmwEr2rDGHl7lsFgAHxmNz/1NlQ7xLIrlN2h5d1eGI=
 go.yaml.in/yaml/v2 v2.4.2/go.mod h1:081UH+NErpNdqlCXm3TtEran0rJZGxAYx9hb/ELlsPU=
 golang.org/x/crypto v0.49.0 h1:+Ng2ULVvLHnJ/ZFEq4KdcDd/cfjrrjjNSXNzxg0Y4U4=
 golang.org/x/crypto v0.49.0/go.mod h1:ErX4dUh2UM+CFYiXZRTcMpEcN8b/1gxEuv3nODoYtCA=
+golang.org/x/net v0.51.0 h1:94R/GTO7mt3/4wIKpcR5gkGmRLOuE/2hNGeWq/GBIFo=
+golang.org/x/net v0.51.0/go.mod h1:aamm+2QF5ogm02fjy5Bb7CQ0WMt1/WVM7FtyaTLlA9Y=
 golang.org/x/sync v0.20.0 h1:e0PTpb7pjO8GAtTs2dQ6jYa5BWYlMuX047Dco/pItO4=
 golang.org/x/sync v0.20.0/go.mod h1:9xrNwdLfx4jkKbNva9FpL6vEN7evnE43NNNJQ2LF3+0=
 golang.org/x/sys v0.35.0 h1:vz1N37gP5bs89s7He8XuIYXpyY0+QlsKmzipCbUtyxI=

internal/mqttclient/client.go

@@ -0,0 +1,149 @@
+// Package mqttclient is a thin wrapper around paho.MQTT for the
+// M4 MQTT ingest path. It is shared by:
+//
+//   - cmd/ingestd (subscriber on ba/+/+/incoming)
+//   - loadgen/cmd/mqtt (publisher on ba/<co>/<src>/incoming)
+//
+// Why a wrapper:
+//   - hide the paho token-on-publish option behind a single
+//     error-returning Publish() that waits for the QoS 1 ack
+//   - one place to set the LastWill, MaxInflight, AutoReconnect
+//     defaults that match our SPEC §22 protection model
+//   - one place to format consistent client IDs ("ingestd-host")
+//     so EMQX /admin/clients shows them cleanly
+//
+// M4 does NOT support persistent sessions; a subscriber restart
+// replays nothing. QoS 1 + dedupe is enough.
+package mqttclient
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"log/slog"
+	"net/url"
+	"time"
+
+	mqtt "github.com/eclipse/paho.mqtt.golang"
+)
+
+// Config holds the connection parameters. Username/Password is
+// EMQX's built-in-db auth: when set, the broker uses them to
+// authenticate and to apply the per-user ACL from acl.conf.
+type Config struct {
+	Broker   string // tcp://emqx:1883 (or ssl:// for TLS in M11+)
+	ClientID string // e.g. "ingestd-mqtt-<host>"
+	Username string // e.g. "ingestd" or "prom-prod-acme-001"
+	Password string // EMQX built-in-db password (== HMAC secret for sources)
+	Clean    bool   // true = no persistent session; M4 default
+}
+
+// Client is the small surface area we need: Connect, Subscribe,
+// Publish, Disconnect. The underlying paho.Client is hidden.
+type Client struct {
+	cfg    Config
+	logger *slog.Logger
+	inner  mqtt.Client
+}
+
+// Connect dials the broker, sets LastWill (offline status) and
+// returns a ready-to-use Client. Returns an error if the initial
+// connect fails; the paho auto-reconnect handles subsequent
+// blips.
+func Connect(ctx context.Context, cfg Config, logger *slog.Logger) (*Client, error) {
+	if cfg.Broker == "" {
+		return nil, errors.New("mqttclient: empty Broker")
+	}
+	if cfg.ClientID == "" {
+		return nil, errors.New("mqttclient: empty ClientID")
+	}
+	uri, err := url.Parse(cfg.Broker)
+	if err != nil {
+		return nil, fmt.Errorf("mqttclient: bad broker url %q: %w", cfg.Broker, err)
+	}
+
+	opts := mqtt.NewClientOptions().
+		AddBroker(uri.String()).
+		SetClientID(cfg.ClientID).
+		SetCleanSession(cfg.Clean).
+		SetAutoReconnect(true).
+		SetMaxReconnectInterval(10 * time.Second).
+		SetConnectTimeout(10 * time.Second).
+		SetWriteTimeout(10 * time.Second).
+		SetKeepAlive(30 * time.Second).
+		SetPingTimeout(10 * time.Second)
+
+	if cfg.Username != "" {
+		opts.SetUsername(cfg.Username)
+		opts.SetPassword(cfg.Password)
+	}
+
+	// Last will: when we go away, EMQX publishes our "offline"
+	// status. Useful for the M9 observability layer; for M4 the
+	// message just lands in $SYS and is ignored.
+	willTopic := "$ba/client/" + cfg.ClientID + "/status"
+	opts.SetWill(willTopic, "offline", 1, false)
+
+	c := &Client{cfg: cfg, logger: logger, inner: mqtt.NewClient(opts)}
+	tok := c.inner.Connect()
+	if !tok.WaitTimeout(15 * time.Second) {
+		return nil, errors.New("mqttclient: connect timeout")
+	}
+	if err := tok.Error(); err != nil {
+		return nil, fmt.Errorf("mqttclient: connect: %w", err)
+	}
+	logger.Info("mqtt connected", "broker", cfg.Broker, "client_id", cfg.ClientID)
+	return c, nil
+}
+
+// Handler is the per-message callback for subscribers. The body
+// is the raw payload bytes; the topic is the full topic string
+// (so the caller can parse ba/<co>/<src>/...).
+type Handler func(topic string, body []byte) error
+
+// Subscribe registers a QoS 1 subscription on the given topic
+// filter. Returns when the SUBACK is received. The handler runs
+// in a paho-internal goroutine; if it returns an error we just
+// log it (paho does not support nack semantics on QoS 1).
+func (c *Client) Subscribe(topic string, h Handler) error {
+	tok := c.inner.Subscribe(topic, 1, func(_ mqtt.Client, m mqtt.Message) {
+		if err := h(m.Topic(), m.Payload()); err != nil {
+			c.logger.Warn("mqtt handler", "err", err, "topic", m.Topic())
+		}
+	})
+	if !tok.WaitTimeout(15 * time.Second) {
+		return errors.New("mqttclient: subscribe timeout")
+	}
+	if err := tok.Error(); err != nil {
+		return fmt.Errorf("mqttclient: subscribe: %w", err)
+	}
+	c.logger.Info("mqtt subscribed", "topic", topic)
+	return nil
+}
+
+// Publish posts a single message at QoS 1 and waits for the
+// broker's PUBACK. Returns the ack error if any. The body is
+// copied by paho; callers can reuse their buffer.
+func (c *Client) Publish(topic string, body []byte) error {
+	tok := c.inner.Publish(topic, 1, false, body)
+	if !tok.WaitTimeout(15 * time.Second) {
+		return errors.New("mqttclient: publish timeout")
+	}
+	if err := tok.Error(); err != nil {
+		return fmt.Errorf("mqttclient: publish: %w", err)
+	}
+	return nil
+}
+
+// IsConnected returns the current connection state. Useful for
+// health endpoints.
+func (c *Client) IsConnected() bool {
+	return c.inner.IsConnected()
+}
+
+// Disconnect sends a clean DISCONNECT and waits up to 5s.
+func (c *Client) Disconnect() {
+	if c.inner.IsConnected() {
+		c.inner.Disconnect(5000)
+	}
+}

internal/observability/metrics.go

@@ -24,6 +24,13 @@ type IngestdMetrics struct {
 	Quarantines    *prometheus.CounterVec
 	CBState        *prometheus.GaugeVec
 	PublishLatency prometheus.Histogram
+	// MQTTMessages is the M4 per-message counter; labels mirror
+	// the same result taxonomy as AlertsReceived (accepted,
+	// deduped, bad_topic, bad_signature, unknown_source,
+	// rate_limited_source, rate_limited_company, invalid,
+	// invalid_json, broker_unavailable) plus a "received" label
+	// for every message that survived parseIncomingTopic.
+	MQTTMessages *prometheus.CounterVec
 }
 
 // NewIngestdMetrics registers and returns the ingestd metrics.
@@ -73,6 +80,13 @@ func NewIngestdMetrics(reg prometheus.Registerer, serviceName string) *IngestdMe
 			Buckets:   prometheus.DefBuckets,
 			ConstLabels: prometheus.Labels{"service": serviceName},
 		}),
+		MQTTMessages: prometheus.NewCounterVec(prometheus.CounterOpts{
+			Namespace: "ba",
+			Subsystem: "ingestd",
+			Name:      "mqtt_messages_total",
+			Help:      "Inbound MQTT messages by result (M4).",
+			ConstLabels: prometheus.Labels{"service": serviceName},
+		}, []string{"result"}),
 	}
 	reg.MustRegister(
 		m.AlertsReceived,
@@ -81,6 +95,7 @@ func NewIngestdMetrics(reg prometheus.Registerer, serviceName string) *IngestdMe
 		m.Quarantines,
 		m.CBState,
 		m.PublishLatency,
+		m.MQTTMessages,
 	)
 	m.AlertsReceived.WithLabelValues("accepted")
 	return m

loadgen/cmd/mqtt/main.go

@@ -0,0 +1,206 @@
+// loadgen/cmd/mqtt is the M4 MQTT publisher for broad-announce.
+// Same data shape as loadgen/cmd/http (severity mix, dedupe
+// ratio, company cycling) but posts to an MQTT broker on
+// ba/<co>/<src>/incoming. Auth is twofold:
+//
+//   1. MQTT username/password: prom-prod-acme-001 / s3cret-acme
+//      (the EMQX built-in-db row from deploy/emqx/auth-built-in-
+//      db-bootstrap.csv). The broker's ACL is keyed on the
+//      username, so this is what gates topic access.
+//   2. Per-message X-BA-Signature in the JSON envelope's `auth`
+//      field, identical to HTTP. This is the per-message HMAC
+//      the ingestd mqtt subscriber re-uses verifyHMAC() to check.
+//
+// Example:
+//
+//	loadgen-mqtt --broker tcp://localhost:1883 \
+//	  --api-key acme-001:prom-prod:s3cret-acme \
+//	  --count 10 --rate 5
+package main
+
+import (
+	"context"
+	"crypto/hmac"
+	"crypto/sha256"
+	"encoding/hex"
+	"encoding/json"
+	"flag"
+	"fmt"
+	"log/slog"
+	"math/rand/v2"
+	"os"
+	"os/signal"
+	"strings"
+	"sync"
+	"sync/atomic"
+	"syscall"
+	"time"
+
+	"git3.techno-world.net/lrosales/broad-announce/internal/mqttclient"
+)
+
+func main() {
+	var (
+		broker    = flag.String("broker", "tcp://localhost:1883", "MQTT broker URL")
+		apiKey    = flag.String("api-key", "", "company:source:secret")
+		count     = flag.Int("count", 10, "total alerts to send")
+		rate      = flag.Int("rate", 10, "target alerts/sec")
+		mode      = flag.String("mode", "normal", "profile: normal|burst")
+		dedupePct = flag.Int("dedupe-pct", 30, "percent sharing a dedupe_key (normal)")
+		metrics   = flag.String("metrics", "", "Prometheus metrics listen addr (empty to disable)")
+		timeout   = flag.Duration("duration", 30*time.Second, "max run time")
+	)
+	flag.Parse()
+	if *apiKey == "" {
+		fmt.Fprintln(os.Stderr, "loadgen-mqtt: --api-key is required (company:source:secret)")
+		os.Exit(2)
+	}
+	parts := strings.SplitN(*apiKey, ":", 3)
+	if len(parts) != 3 {
+		fmt.Fprintln(os.Stderr, "loadgen-mqtt: --api-key must be company:source:secret")
+		os.Exit(2)
+	}
+	company, source, secret := parts[0], parts[1], parts[2]
+	username := fmt.Sprintf("%s-%s", source, company) // matches auth csv
+
+	logger := slog.New(slog.NewTextHandler(os.Stderr, &slog.HandlerOptions{Level: slog.LevelInfo}))
+	ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
+	defer stop()
+
+	host, _ := os.Hostname()
+	client, err := mqttclient.Connect(ctx, mqttclient.Config{
+		Broker:   *broker,
+		ClientID: fmt.Sprintf("loadgen-mqtt-%s", host),
+		Username: username,
+		Password: secret,
+		Clean:    true,
+	}, logger)
+	if err != nil {
+		fmt.Fprintln(os.Stderr, "loadgen-mqtt: connect:", err)
+		os.Exit(1)
+	}
+	defer client.Disconnect()
+
+	topic := fmt.Sprintf("ba/%s/%s/incoming", company, source)
+	logger.Info("publishing", "topic", topic, "count", *count, "rate", *rate, "mode", *mode)
+
+	var (
+		sent   atomic.Uint64
+		failed atomic.Uint64
+		dupes  atomic.Uint64
+	)
+	limiter := time.NewTicker(time.Second / time.Duration(*rate))
+	defer limiter.Stop()
+
+	var wg sync.WaitGroup
+	work := make(chan int, 64)
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		for i := 0; i < *count; i++ {
+			work <- i
+		}
+		close(work)
+	}()
+
+	deadline := time.Now().Add(*timeout)
+	for i := range work {
+		if time.Now().After(deadline) {
+			logger.Warn("deadline reached, stopping")
+			break
+		}
+		<-limiter.C
+
+		a := makeAlert(i, *mode, *dedupePct, company, source)
+		ts := time.Now().Unix()
+		// HTTP-style signature over the alert body, with t= prefix.
+		// The MQTT subscriber's verifyHMAC accepts the same shape.
+		body, _ := json.Marshal(a)
+		mac := hmac.New(sha256.New, []byte(secret))
+		mac.Write([]byte(fmt.Sprintf("%d", ts)))
+		mac.Write([]byte("."))
+		mac.Write(body)
+		sig := hex.EncodeToString(mac.Sum(nil))
+		env := map[string]json.RawMessage{
+			"alert": body,
+		}
+		env["auth"] = json.RawMessage(fmt.Sprintf("%q", fmt.Sprintf("t=%d,v1=%s", ts, sig)))
+		envelope, _ := json.Marshal(env)
+
+		if err := client.Publish(topic, envelope); err != nil {
+			failed.Add(1)
+			logger.Warn("publish failed", "err", err, "i", i)
+			continue
+		}
+		if i > 0 && i%(*count/10+1) == 0 {
+			logger.Info("progress", "sent", i, "total", *count)
+		}
+		if !*isUnique(*dedupePct, i) {
+			dupes.Add(1)
+		}
+		sent.Add(1)
+	}
+	wg.Wait()
+
+	logger.Info("done",
+		"sent", sent.Load(),
+		"failed", failed.Load(),
+		"dupes", dupes.Load(),
+	)
+	if failed.Load() > 0 {
+		os.Exit(1)
+	}
+	_ = metrics // reserved for M9
+}
+
+// makeAlert generates one alert. Same shape as loadgen/cmd/http.
+func makeAlert(idx int, mode string, dedupePct int, company, source string) map[string]any {
+	severity := pickSeverity(mode)
+	// dedupe_pct: 30 means ~30% of alerts share a dedupe_key.
+	dedupeKey := fmt.Sprintf("lg-m4-%d", idx)
+	if dedupePct > 0 && idx > 0 && rand.IntN(100) < dedupePct {
+		dedupeKey = "lg-m4-shared"
+	}
+	return map[string]any{
+		"company_id": company,
+		"source_id":  source,
+		"severity":   severity,
+		"category":   "loadgen",
+		"title":      fmt.Sprintf("LG M4 #%d", idx),
+		"body":       "mqtt smoke",
+		"data":       map[string]string{"host": "lg-host", "idx": fmt.Sprintf("%d", idx)},
+		"dedupe_key": dedupeKey,
+	}
+}
+
+func pickSeverity(mode string) string {
+	r := rand.IntN(100)
+	switch mode {
+	case "burst":
+		// Mostly critical to exercise the bypass.
+		switch {
+		case r < 70:
+			return "critical"
+		case r < 95:
+			return "inminent_colapse"
+		default:
+			return "warning"
+		}
+	default: // "normal"
+		switch {
+		case r < 70:
+			return "info"
+		case r < 95:
+			return "warning"
+		case r < 99:
+			return "critical"
+		default:
+			return "inminent_colapse"
+		}
+	}
+}
+
+func isUnique(dedupePct, i int) *bool {
+	b := i == 0 || dedupePct == 0 || rand.IntN(100) >= dedupePct
+	return &b
+}

loadgen/go.mod

@@ -5,3 +5,10 @@ go 1.25.0
 replace git3.techno-world.net/lrosales/broad-announce => ..
 
 require git3.techno-world.net/lrosales/broad-announce v0.0.0-00010101000000-000000000000
+
+require (
+	github.com/eclipse/paho.mqtt.golang v1.5.1 // indirect
+	github.com/gorilla/websocket v1.5.3 // indirect
+	golang.org/x/net v0.51.0 // indirect
+	golang.org/x/sync v0.20.0 // indirect
+)

loadgen/go.sum

@@ -0,0 +1,8 @@
+github.com/eclipse/paho.mqtt.golang v1.5.1 h1:/VSOv3oDLlpqR2Epjn1Q7b2bSTplJIeV2ISgCl2W7nE=
+github.com/eclipse/paho.mqtt.golang v1.5.1/go.mod h1:1/yJCneuyOoCOzKSsOTUc0AJfpsItBGWvYpBLimhArU=
+github.com/gorilla/websocket v1.5.3 h1:saDtZ6Pbx/0u+bgYQ3q96pZgCzfhKXGPqt7kZ72aNNg=
+github.com/gorilla/websocket v1.5.3/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE=
+golang.org/x/net v0.51.0 h1:94R/GTO7mt3/4wIKpcR5gkGmRLOuE/2hNGeWq/GBIFo=
+golang.org/x/net v0.51.0/go.mod h1:aamm+2QF5ogm02fjy5Bb7CQ0WMt1/WVM7FtyaTLlA9Y=
+golang.org/x/sync v0.20.0 h1:e0PTpb7pjO8GAtTs2dQ6jYa5BWYlMuX047Dco/pItO4=
+golang.org/x/sync v0.20.0/go.mod h1:9xrNwdLfx4jkKbNva9FpL6vEN7evnE43NNNJQ2LF3+0=