Add notification for controller connect/startup

garden-app/controller/controller.go

@@ -297,6 +297,19 @@ func (c *Controller) publishTemperatureHumidityData() {
 	}
 }
 
+// PublishStartupLog publishes the message that controllers use to signal that they started up
+func (c *Controller) PublishStartupLog(topicPrefix string) error {
+	topic := fmt.Sprintf("%s/data/logs", topicPrefix)
+	msg := "logs message=\"garden-controller setup complete\""
+
+	err := c.mqttClient.Publish(topic, []byte(msg))
+	if err != nil {
+		return fmt.Errorf("error publishing startup log %w", err)
+	}
+
+	return nil
+}
+
 // addNoise will take a base value and introduce some += variance based on the provided percentage range. This will
 // produce sensor data that is relatively consistent but not totally flat
 func addNoise(baseValue float64, percentRange float64) float64 {

garden-app/integration_tests/main_test.go

@@ -14,6 +14,8 @@ import (
 	"github.com/calvinmclean/automated-garden/garden-app/controller"
 	"github.com/calvinmclean/automated-garden/garden-app/pkg"
 	"github.com/calvinmclean/automated-garden/garden-app/pkg/action"
+	"github.com/calvinmclean/automated-garden/garden-app/pkg/notifications"
+	fake_notification "github.com/calvinmclean/automated-garden/garden-app/pkg/notifications/fake"
 	"github.com/calvinmclean/automated-garden/garden-app/pkg/weather"
 	"github.com/calvinmclean/automated-garden/garden-app/pkg/weather/fake"
 	"github.com/calvinmclean/automated-garden/garden-app/server"
@@ -59,6 +61,7 @@ func TestIntegration(t *testing.T) {
 	t.Run("Garden", GardenTests)
 	t.Run("Zone", ZoneTests)
 	t.Run("WaterSchedule", WaterScheduleTests)
+	t.Run("ControllerStartupNotification", ControllerStartupNotificationTest)
 }
 
 func getConfigs(t *testing.T) (server.Config, controller.Config) {
@@ -409,8 +412,8 @@ func ZoneTests(t *testing.T) {
 	})
 }
 
-func floatPointer(n float32) *float32 {
-	return &n
+func pointer[T any](v T) *T {
+	return &v
 }
 
 func WaterScheduleTests(t *testing.T) {
@@ -443,9 +446,9 @@ func WaterScheduleTests(t *testing.T) {
 		status, err = makeRequest(http.MethodPatch, "/water_schedules/"+waterScheduleID, pkg.WaterSchedule{
 			WeatherControl: &weather.Control{
 				Rain: &weather.ScaleControl{
-					BaselineValue: floatPointer(0),
-					Factor:        floatPointer(0),
-					Range:         floatPointer(25.4),
+					BaselineValue: pointer[float32](0),
+					Factor:        pointer[float32](0),
+					Range:         pointer[float32](25.4),
 					ClientID:      weatherClientWithRain,
 				},
 			},
@@ -468,6 +471,49 @@ func WaterScheduleTests(t *testing.T) {
 	})
 }
 
+func ControllerStartupNotificationTest(t *testing.T) {
+	var g server.GardenResponse
+	t.Run("CreateGarden", func(t *testing.T) {
+		status, err := makeRequest(http.MethodPost, "/gardens", `{
+				"name": "Notification",
+				"topic_prefix": "notification",
+				"max_zones": 3
+			}`, &g)
+		assert.NoError(t, err)
+		assert.Equal(t, http.StatusCreated, status)
+	})
+
+	var nc notifications.Client
+	t.Run("CreateNotificationClient", func(t *testing.T) {
+		status, err := makeRequest(http.MethodPost, "/notification_clients", `{
+				"name": "fake client",
+				"type": "fake",
+				"options": {}
+			}`, &nc)
+		assert.NoError(t, err)
+		assert.Equal(t, http.StatusCreated, status)
+	})
+
+	t.Run("EnableNotificationsForGarden", func(t *testing.T) {
+		status, err := makeRequest(http.MethodPatch, "/gardens/"+g.GetID(), pkg.Garden{
+			NotificationClientID: pointer(nc.GetID()),
+		}, &g)
+		assert.NoError(t, err)
+		assert.Equal(t, http.StatusOK, status)
+	})
+
+	t.Run("PublishStartupLogAndCheckNotification", func(t *testing.T) {
+		err := c.PublishStartupLog(g.TopicPrefix)
+		require.NoError(t, err)
+
+		time.Sleep(500 * time.Millisecond)
+
+		lastMsg := fake_notification.LastMessage()
+		require.Equal(t, "Notification connected", lastMsg.Title)
+		require.Equal(t, "garden-controller setup complete", lastMsg.Message)
+	})
+}
+
 func makeRequest(method, path string, body, response interface{}) (int, error) {
 	// TODO: Use babyapi Client
 	var reqBody io.Reader

garden-app/pkg/mqtt/mqtt.go

@@ -1,5 +1,7 @@
 package mqtt
 
+//go:generate mockery --all --inpackage
+
 import (
 	"bytes"
 	"errors"
@@ -12,6 +14,8 @@ import (
 	"github.com/prometheus/client_golang/prometheus"
 )
 
+const QOS = byte(1)
+
 var mqttClientSummary = prometheus.NewSummaryVec(prometheus.SummaryOpts{
 	Namespace: "garden_app",
 	Name:      "mqtt_client_duration_seconds",
@@ -39,12 +43,16 @@ type Client interface {
 	LightTopic(string) (string, error)
 	Connect() error
 	Disconnect(uint)
+	AddHandler(TopicHandler)
 }
 
 // client is a wrapper struct for connecting our config and MQTT Client. It implements the Client interface
 type client struct {
 	mu sync.Mutex
 	mqtt.Client
+
+	handlers []TopicHandler
+
 	Config
 }
 
@@ -58,17 +66,21 @@ type TopicHandler struct {
 // using the supplied functions to handle incoming messages. It really should be used with only one function,
 // but I wanted to make it an optional argument, which required using the variadic function argument
 func NewClient(config Config, defaultHandler mqtt.MessageHandler, handlers ...TopicHandler) (Client, error) {
+	client := &client{
+		Config:   config,
+		handlers: handlers,
+	}
+
 	opts := mqtt.NewClientOptions().AddBroker(fmt.Sprintf("tcp://%s:%d", config.Broker, config.Port))
 	opts.ClientID = config.ClientID
 	opts.AutoReconnect = true
 	opts.CleanSession = false
-	if len(handlers) > 0 {
-		opts.OnConnect = func(c mqtt.Client) {
-			for _, handler := range handlers {
-				if token := c.Subscribe(handler.Topic, byte(1), handler.Handler); token.Wait() && token.Error() != nil {
-					// TODO: can I return an error instead of panicking (recover maybe?)
-					panic(token.Error())
-				}
+	opts.OnConnect = func(c mqtt.Client) {
+		for _, handler := range client.handlers {
+			token := c.Subscribe(handler.Topic, QOS, handler.Handler)
+			if token.Wait() && token.Error() != nil {
+				// TODO: can I return an error instead of panicking (recover maybe?)
+				panic(token.Error())
 			}
 		}
 	}
@@ -79,7 +91,13 @@ func NewClient(config Config, defaultHandler mqtt.MessageHandler, handlers ...To
 		return nil, err
 	}
 
-	return &client{Client: mqtt.NewClient(opts), Config: config}, nil
+	client.Client = mqtt.NewClient(opts)
+
+	return client, nil
+}
+
+func (c *client) AddHandler(handler TopicHandler) {
+	c.handlers = append(c.handlers, handler)
 }
 
 // Connect uses the MQTT Client's Connect function but returns the error instead of Token

garden-app/server/api.go

@@ -117,10 +117,7 @@ func (api *API) Setup(cfg Config, validateData bool) error {
 		"broker", cfg.MQTTConfig.Broker,
 		"port", cfg.MQTTConfig.Port,
 	).Info("initializing MQTT client")
-	mqttClient, err := mqtt.NewClient(cfg.MQTTConfig, mqtt.DefaultHandler(logger), mqtt.TopicHandler{
-		Topic:   "+/data/water",
-		Handler: NewWaterNotificationHandler(storageClient, logger).HandleMessage,
-	})
+	mqttClient, err := mqtt.NewClient(cfg.MQTTConfig, mqtt.DefaultHandler(logger))
 	if err != nil {
 		return fmt.Errorf("unable to initialize MQTT client: %v", err)
 	}

garden-app/server/notification_clients.go

@@ -12,8 +12,7 @@ import (
 )
 
 const (
-	notificationClientsBasePath  = "/notification_clients"
-	notificationClientIDLogField = "notification_client_id"
+	notificationClientsBasePath = "/notification_clients"
 )
 
 // NotificationClientsAPI encapsulates the structs and dependencies necessary for the NotificationClients API