broad-announce/cmd/ingestd/mqtt.go

// 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