broad-announce/internal/mqttclient/client.go

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