home_automation_emulator/internal/mqttclient/client.go

// Package mqttclient speaks the same wire protocol a real device would:
// publish to devices/{id}/telemetry and devices/{id}/ack, subscribe to
// devices/{id}/commands. Topic and payload shapes are dictated by
// ingest-service and device-control-service in the main platform repo —
// this client exists to match them, not to define them.
package mqttclient

import (
	"encoding/json"
	"fmt"
	"log/slog"
	"strings"
	"time"

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

type Config struct {
	BrokerURL string
	ClientID  string
}

// CommandHandler receives one decoded devices/{id}/commands message.
// hasLevel distinguishes "level omitted" from "level is zero".
type CommandHandler func(externalID, action string, level float64, hasLevel bool)

type Client struct {
	client mqtt.Client
	logger *slog.Logger
}

func Connect(cfg Config, logger *slog.Logger) (*Client, error) {
	opts := mqtt.NewClientOptions().
		AddBroker(cfg.BrokerURL).
		SetClientID(cfg.ClientID).
		SetAutoReconnect(true).
		SetConnectRetry(true).
		SetConnectionLostHandler(func(_ mqtt.Client, err error) {
			logger.Warn("mqtt connection lost", "error", err)
		})

	client := mqtt.NewClient(opts)
	token := client.Connect()
	token.Wait()
	if err := token.Error(); err != nil {
		return nil, fmt.Errorf("connect to mqtt broker %s: %w", cfg.BrokerURL, err)
	}

	return &Client{client: client, logger: logger}, nil
}

// SubscribeCommands listens on devices/+/commands for every device the
// emulator might represent — one subscription covers the whole board.
func (c *Client) SubscribeCommands(handler CommandHandler) error {
	token := c.client.Subscribe("devices/+/commands", 1, func(_ mqtt.Client, msg mqtt.Message) {
		externalID, ok := externalIDFromTopic(msg.Topic())
		if !ok {
			return
		}

		var cmd struct {
			Action string   `json:"action"`
			Level  *float64 `json:"level"`
		}
		if err := json.Unmarshal(msg.Payload(), &cmd); err != nil {
			c.logger.Warn("dropping invalid command payload", "device_id", externalID, "error", err)
			return
		}

		level := 0.0
		hasLevel := cmd.Level != nil
		if hasLevel {
			level = *cmd.Level
		}
		handler(externalID, cmd.Action, level, hasLevel)
	})
	token.Wait()
	return token.Error()
}

func externalIDFromTopic(topic string) (string, bool) {
	parts := strings.Split(topic, "/")
	if len(parts) != 3 || parts[0] != "devices" {
		return "", false
	}
	return parts[1], true
}

// PublishTelemetry sends one sensor_type reading, matching the payload
// shape services/ingest-service/internal/telemetry.ParseReading expects.
func (c *Client) PublishTelemetry(externalID, zoneID, sensorType string, value float64) error {
	body, err := json.Marshal(map[string]any{
		"device_id":   externalID,
		"zone_id":     zoneID,
		"sensor_type": sensorType,
		"value":       value,
		"timestamp":   time.Now().UTC().Format(time.RFC3339),
	})
	if err != nil {
		return err
	}
	return c.publish(fmt.Sprintf("devices/%s/telemetry", externalID), body)
}

// PublishAck reports an actuator's new state after a command, matching the
// shape services/device-control-service/cmd/.../ackHandler expects.
func (c *Client) PublishAck(externalID string, reportedState map[string]any) error {
	body, err := json.Marshal(map[string]any{"state": reportedState})
	if err != nil {
		return err
	}
	return c.publish(fmt.Sprintf("devices/%s/ack", externalID), body)
}

func (c *Client) publish(topic string, body []byte) error {
	token := c.client.Publish(topic, 1, false, body)
	token.Wait()
	return token.Error()
}

func (c *Client) Close() {
	c.client.Disconnect(250)
}