internet_use_control/internal/mqtt/link.go

// The broker link: connect, subscribe to device reports, publish retained
// config. Everything that reaches SQLite is handled here (AGENTS.md §11
// Phase 3).
package mqtt

import (
	"context"
	"crypto/tls"
	"crypto/x509"
	"errors"
	"fmt"
	"log/slog"
	"net/url"
	"os"
	"strings"
	"sync"
	"sync/atomic"
	"time"

	paho "github.com/eclipse/paho.mqtt.golang"

	"gitea.lab.audasmedia.com.au/sam/internet_use_control/internal/store"
)

// DeviceLookup finds a device from a topic it wrote to.
type DeviceLookup interface {
	DeviceByConfigTopic(ctx context.Context, topic string) (*store.Device, error)
}

// ReportSink stores what a device reported.
type ReportSink interface {
	SetPresence(ctx context.Context, id int64, online bool, at time.Time, actor string) (bool, error)
	InsertTelemetry(ctx context.Context, id int64, day string, entries []store.TelemetryEntry) (int, error)
}

// ConfigSource reads the desired state that is published to devices.
type ConfigSource interface {
	ListDevices(ctx context.Context) ([]store.Device, error)
	DeviceLists(ctx context.Context, id int64) (allow, block []string, err error)
}

// Options configures the link.
type Options struct {
	// Broker is a URL such as "tls://mosquitto.lab:8883" or, until the shared
	// broker gets its TLS listener, "tcp://host.docker.internal:1883".
	Broker string
	// ClientID identifies the Control Service on the broker.
	ClientID string
	// Username and Password are the Control Service credentials. The broker's
	// ACL gives this identity write access to every config topic and read
	// access to every status and telemetry topic — and nothing else.
	Username string
	Password string
	// CAFile is a PEM bundle used to verify the broker certificate.
	CAFile string
	// KeepAlive is the MQTT keepalive.
	KeepAlive time.Duration
	// ConnectTimeout bounds one connection attempt.
	ConnectTimeout time.Duration

	// Actor is the audit-log actor for reports that arrive over the broker.
	Actor string
	// Log receives structured logs.
	Log *slog.Logger

	// Devices and Reports are required.
	Devices DeviceLookup
	// Source is required to publish config.
	Source ConfigSource
	// Reports is required to store device reports.
	Reports ReportSink
	// OnConnected runs after every successful (re)connect. The service uses it
	// to republish every device's retained config.
	OnConnected func()
}

// Link is the live connection to the broker.
type Link struct {
	opts Options
	log  *slog.Logger

	client paho.Client

	connected  atomic.Bool
	received   atomic.Int64
	published  atomic.Int64
	lastMsg    atomic.Int64 // unix nanos
	lastErrMu  sync.Mutex
	lastErrMsg string
}

// New builds the link. It does not connect; call Start.
func New(opts Options) (*Link, error) {
	if opts.Devices == nil || opts.Source == nil || opts.Reports == nil {
		return nil, errors.New("mqtt: Devices, Source and Reports are required")
	}
	if strings.TrimSpace(opts.Broker) == "" {
		return nil, errors.New("mqtt: Broker is required")
	}
	if opts.ClientID == "" {
		opts.ClientID = "control-service"
	}
	if opts.KeepAlive <= 0 {
		opts.KeepAlive = 30 * time.Second
	}
	if opts.ConnectTimeout <= 0 {
		opts.ConnectTimeout = 10 * time.Second
	}
	if opts.Actor == "" {
		opts.Actor = "device"
	}
	if opts.Log == nil {
		opts.Log = slog.Default()
	}

	l := &Link{opts: opts, log: opts.Log}

	clientOpts := paho.NewClientOptions()
	clientOpts.AddBroker(opts.Broker)
	clientOpts.SetClientID(opts.ClientID)
	clientOpts.SetUsername(opts.Username)
	clientOpts.SetPassword(opts.Password)
	clientOpts.SetKeepAlive(opts.KeepAlive)
	clientOpts.SetPingTimeout(min(10*time.Second, opts.KeepAlive))
	clientOpts.SetConnectTimeout(opts.ConnectTimeout)
	clientOpts.SetWriteTimeout(opts.ConnectTimeout)
	// The service is a publisher of retained truth and a listener for reports:
	// reconnect forever, and re-subscribe on every connect.
	clientOpts.SetAutoReconnect(true)
	clientOpts.SetConnectRetry(true)
	clientOpts.SetConnectRetryInterval(5 * time.Second)
	clientOpts.SetMaxReconnectInterval(60 * time.Second)
	// Clean session: the broker keeps the retained config, so the service does
	// not need a queue of stale reports.
	clientOpts.SetCleanSession(true)
	clientOpts.SetOrderMatters(false)

	if isTLSBroker(opts.Broker) {
		tlsCfg, err := tlsConfig(opts.Broker, opts.CAFile)
		if err != nil {
			return nil, err
		}
		clientOpts.SetTLSConfig(tlsCfg)
	}

	clientOpts.SetOnConnectHandler(l.handleConnect)
	clientOpts.SetConnectionLostHandler(func(_ paho.Client, err error) {
		l.connected.Store(false)
		l.setErr(err)
		l.log.Warn("mqtt connection lost", "broker", opts.Broker, "err", err)
	})

	l.client = paho.NewClient(clientOpts)
	return l, nil
}

// Start connects in the background. A broker outage must never stop the HTTP
// service: the dashboard keeps working and the link reconnects on its own.
func (l *Link) Start(ctx context.Context) {
	l.log.Info("mqtt connecting", "broker", l.opts.Broker, "client_id", l.opts.ClientID,
		"tls", isTLSBroker(l.opts.Broker), "username", l.opts.Username)
	go func() {
		if token := l.client.Connect(); token.Wait() && token.Error() != nil {
			l.setErr(token.Error())
			l.log.Error("mqtt connect", "err", token.Error())
		}
	}()
}

// Stop disconnects cleanly. The broker then publishes nothing on our behalf;
// the service's own will is not registered, because only the service publishes
// config and a stale config must survive a service restart.
func (l *Link) Stop() {
	if l.client == nil {
		return
	}
	if l.client.IsConnected() {
		l.client.Unsubscribe(StatusSubscription, TelemetrySubscription)
	}
	l.client.Disconnect(250)
	l.connected.Store(false)
}

// Broker returns the configured broker URL.
func (l *Link) Broker() string { return l.opts.Broker }

// Connected reports the current connection state.
func (l *Link) Connected() bool { return l.connected.Load() }

// Stats returns counters for the dashboard.
func (l *Link) Stats() (received, published int64, lastMessageAt time.Time, lastErr string) {
	ns := l.lastMsg.Load()
	if ns > 0 {
		lastMessageAt = time.Unix(0, ns)
	}
	l.lastErrMu.Lock()
	lastErr = l.lastErrMsg
	l.lastErrMu.Unlock()
	return l.received.Load(), l.published.Load(), lastMessageAt, lastErr
}

func (l *Link) setErr(err error) {
	if err == nil {
		return
	}
	l.lastErrMu.Lock()
	l.lastErrMsg = err.Error()
	l.lastErrMu.Unlock()
}

// handleConnect runs on every (re)connect, including the first one.
func (l *Link) handleConnect(c paho.Client) {
	l.connected.Store(true)
	l.setErr(nil)
	l.log.Info("mqtt connected", "broker", l.opts.Broker)

	for _, sub := range []struct {
		topic   string
		handler paho.MessageHandler
	}{
		{StatusSubscription, l.handleStatusMessage},
		{TelemetrySubscription, l.handleTelemetryMessage},
	} {
		token := c.Subscribe(sub.topic, 1, sub.handler)
		if !token.WaitTimeout(l.opts.ConnectTimeout) {
			l.log.Warn("mqtt subscribe timed out", "topic", sub.topic)
			continue
		}
		if err := token.Error(); err != nil {
			l.setErr(err)
			l.log.Error("mqtt subscribe", "topic", sub.topic, "err", err)
			continue
		}
		l.log.Info("mqtt subscribed", "topic", sub.topic)
	}

	if l.opts.OnConnected != nil {
		l.opts.OnConnected()
	}
}

// handleStatusMessage turns a device presence report into a store update.
func (l *Link) handleStatusMessage(_ paho.Client, msg paho.Message) {
	l.noteMessage()

	configTopic, ok := deviceFromTopic(msg.Topic(), SuffixStatus)
	if !ok {
		l.log.Warn("mqtt status on an unexpected topic", "topic", msg.Topic())
		return
	}

	status, err := ParseStatus(msg.Payload())
	if err != nil {
		l.log.Warn("mqtt status rejected", "topic", msg.Topic(), "err", err)
		return
	}

	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()

	dev, err := l.opts.Devices.DeviceByConfigTopic(ctx, configTopic)
	if err != nil {
		l.log.Warn("mqtt status for an unknown device", "topic", msg.Topic(), "err", err)
		return
	}

	online := status.IsOnline()
	changed, err := l.opts.Reports.SetPresence(ctx, dev.ID, online, status.Timestamp(time.Now()), l.opts.Actor)
	if err != nil {
		l.log.Error("mqtt presence", "device", dev.Name, "err", err)
		return
	}
	if changed {
		l.log.Info("presence changed", "device", dev.Name, "topic", msg.Topic(),
			"online", online, "retained", msg.Retained())
	}
}

// handleTelemetryMessage stores a device usage report.
func (l *Link) handleTelemetryMessage(_ paho.Client, msg paho.Message) {
	l.noteMessage()

	configTopic, ok := deviceFromTopic(msg.Topic(), SuffixTelemetry)
	if !ok {
		l.log.Warn("mqtt telemetry on an unexpected topic", "topic", msg.Topic())
		return
	}

	report, err := ParseTelemetry(msg.Payload())
	if err != nil {
		l.log.Warn("mqtt telemetry rejected", "topic", msg.Topic(), "err", err)
		return
	}

	rows := report.Rows()
	if len(rows) == 0 {
		l.log.Warn("mqtt telemetry had no usable rows", "topic", msg.Topic())
		return
	}

	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()

	dev, err := l.opts.Devices.DeviceByConfigTopic(ctx, configTopic)
	if err != nil {
		l.log.Warn("mqtt telemetry for an unknown device", "topic", msg.Topic(), "err", err)
		return
	}

	day := report.DayOrDefault(time.Now())
	n, err := l.opts.Reports.InsertTelemetry(ctx, dev.ID, day, rows)
	if err != nil {
		l.log.Error("mqtt telemetry", "device", dev.Name, "err", err)
		return
	}
	l.log.Info("telemetry stored", "device", dev.Name, "day", day, "rows", n)
}

func (l *Link) noteMessage() {
	l.received.Add(1)
	l.lastMsg.Store(time.Now().UnixNano())
}

// PublishDeviceConfig publishes one device's desired state, retained, on the
// topic stored with the device. It is the single writer path for config
// (AGENTS.md §10 rule 5): only the Control Service publishes.
func (l *Link) PublishDeviceConfig(ctx context.Context, dev store.Device) error {
	allow, block, err := l.opts.Source.DeviceLists(ctx, dev.ID)
	if err != nil {
		return err
	}
	payload, err := Config{
		V:           ConfigVersion,
		DeviceID:    dev.ID,
		Name:        dev.Name,
		Platform:    dev.Platform,
		Locked:      dev.Locked,
		Until:       dev.Until,
		Allow:       allow,
		Block:       block,
		GeneratedAt: time.Now().UTC(),
	}.Marshal()
	if err != nil {
		return err
	}
	return l.Publish(dev.Topic, payload)
}

// Publish sends one retained message and waits for the broker to accept it.
func (l *Link) Publish(topic string, payload []byte) error {
	if !l.client.IsConnected() {
		return errors.New("mqtt: not connected")
	}
	token := l.client.Publish(topic, 1, true, payload)
	if !token.WaitTimeout(l.opts.ConnectTimeout) {
		return fmt.Errorf("mqtt: publish to %s timed out", topic)
	}
	if err := token.Error(); err != nil {
		return fmt.Errorf("mqtt: publish to %s: %w", topic, err)
	}
	l.published.Add(1)
	return nil
}

// PublishAll republishes every device's retained config. It runs after every
// connect, so a broker that lost its retained store is repaired from SQLite —
// the database stays the source of truth.
func (l *Link) PublishAll(ctx context.Context) (int, error) {
	devices, err := l.opts.Source.ListDevices(ctx)
	if err != nil {
		return 0, err
	}
	published := 0
	var firstErr error
	for _, dev := range devices {
		if err := l.PublishDeviceConfig(ctx, dev); err != nil {
			l.log.Error("mqtt publish config", "device", dev.Name, "err", err)
			if firstErr == nil {
				firstErr = err
			}
			continue
		}
		published++
	}
	return published, firstErr
}

// isTLSBroker reports whether the broker URL asks for TLS.
func isTLSBroker(broker string) bool {
	return strings.HasPrefix(broker, "tls://") || strings.HasPrefix(broker, "ssl://")
}

// tlsConfig verifies the broker against the lab CA. Every device and the
// Control Service use the same CA, so a rogue broker on the LAN cannot take
// over a device's rules.
func tlsConfig(broker, caFile string) (*tls.Config, error) {
	if caFile == "" {
		return nil, errors.New("mqtt: a TLS broker needs CONTROL_MQTT_CA_FILE")
	}
	pem, err := os.ReadFile(caFile)
	if err != nil {
		return nil, fmt.Errorf("mqtt: read CA %s: %w", caFile, err)
	}
	pool := x509.NewCertPool()
	if !pool.AppendCertsFromPEM(pem) {
		return nil, fmt.Errorf("mqtt: no certificates in %s", caFile)
	}

	cfg := &tls.Config{RootCAs: pool, MinVersion: tls.VersionTLS12}
	if u, err := url.Parse(broker); err == nil && u.Hostname() != "" {
		cfg.ServerName = u.Hostname()
	}
	return cfg, nil
}