fix: centralise nats connection logic

hive/daemon/daemon.go

@@ -541,25 +541,11 @@ func (d *Daemon) Start() error {
 
 // connectNATS establishes a connection to the NATS server with reconnect handling.
 func (d *Daemon) connectNATS() error {
-	opts := []nats.Option{
-		nats.Token(d.config.NATS.ACL.Token),
-		nats.ReconnectWait(time.Second),
-		nats.MaxReconnects(-1), // Infinite reconnects
-		nats.DisconnectErrHandler(func(nc *nats.Conn, err error) {
-			slog.Warn("NATS disconnected", "err", err)
-		}),
-		nats.ReconnectHandler(func(nc *nats.Conn) {
-			slog.Info("NATS reconnected", "url", nc.ConnectedUrl())
-		}),
-	}
-
-	var err error
-	d.natsConn, err = nats.Connect(d.config.NATS.Host, opts...)
+	nc, err := utils.ConnectNATS(d.config.NATS.Host, d.config.NATS.ACL.Token)
 	if err != nil {
 		return err
 	}
-
-	slog.Info("Connected to NATS server", "host", d.config.NATS.Host)
+	d.natsConn = nc
 	return nil
 }
 

hive/services/awsgw/awsgw.go

@@ -5,12 +5,10 @@ import (
 	"log/slog"
 	"os"
 	"strings"
-	"time"
 
 	"github.com/mulgadc/hive/hive/config"
 	"github.com/mulgadc/hive/hive/gateway"
 	"github.com/mulgadc/hive/hive/utils"
-	"github.com/nats-io/nats.go"
 )
 
 var serviceName = "awsgw"
@@ -60,27 +58,13 @@ func launchService(config *config.ClusterConfig) (err error) {
 	nodeConfig := config.Nodes[config.Node]
 
 	// Connect to NATS for service communication
-	opts := []nats.Option{
-		nats.Token(nodeConfig.NATS.ACL.Token),
-		nats.ReconnectWait(time.Second),
-		nats.MaxReconnects(-1), // Infinite reconnects
-		nats.DisconnectErrHandler(func(nc *nats.Conn, err error) {
-			slog.Warn("NATS disconnected", "err", err)
-		}),
-		nats.ReconnectHandler(func(nc *nats.Conn) {
-			slog.Info("NATS reconnected", "url", nc.ConnectedUrl())
-		}),
-	}
-
-	natsConn, err := nats.Connect(nodeConfig.NATS.Host, opts...)
+	natsConn, err := utils.ConnectNATS(nodeConfig.NATS.Host, nodeConfig.NATS.ACL.Token)
 	if err != nil {
 		slog.Error("Failed to connect to NATS", "err", err)
 		return err
 	}
 	defer natsConn.Close()
 
-	slog.Info("Connected to NATS server", "host", nodeConfig.NATS.Host)
-
 	// Append Base dir if config has no leading path
 	if nodeConfig.BaseDir != "" && !strings.HasPrefix(nodeConfig.AWSGW.Config, "/") {
 		nodeConfig.AWSGW.Config = fmt.Sprintf("%s/%s", nodeConfig.BaseDir, nodeConfig.AWSGW.Config)

hive/services/viperblockd/viperblockd.go

@@ -38,6 +38,7 @@ type Config struct {
 	PluginPath     string
 	Debug          bool
 	NatsHost       string
+	NatsToken      string
 	MountedVolumes []MountedVolume
 	S3Host         string
 	Bucket         string
@@ -143,15 +144,13 @@ func (svc *Service) Reload() (err error) {
 func launchService(cfg *Config) (err error) {
 
 	// Connect to NATS
-	nc, err := nats.Connect(cfg.NatsHost)
-
+	nc, err := utils.ConnectNATS(cfg.NatsHost, cfg.NatsToken)
 	if err != nil {
-		slog.Error("Failed to connect to NATS:", "err", err)
+		slog.Error("Failed to connect to NATS", "err", err)
 		return err
 	}
 
-	// Subscribe to the viperblock.mount subject
-	slog.Info("Connected. Waiting for EBS events")
+	slog.Info("Waiting for EBS events")
 
 	if _, err := nc.QueueSubscribe("ebs.delete", "hive-workers", func(msg *nats.Msg) {
 		slog.Info("Received ebs.delete message", "data", string(msg.Data))

hive/utils/nats.go

@@ -4,11 +4,39 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
+	"log/slog"
 	"time"
 
 	"github.com/nats-io/nats.go"
 )
 
+// ConnectNATS establishes a connection to a NATS server with standard reconnect
+// handling and logging. If token is non-empty, token authentication is used.
+func ConnectNATS(host, token string) (*nats.Conn, error) {
+	opts := []nats.Option{
+		nats.ReconnectWait(time.Second),
+		nats.MaxReconnects(-1),
+		nats.DisconnectErrHandler(func(nc *nats.Conn, err error) {
+			slog.Warn("NATS disconnected", "err", err)
+		}),
+		nats.ReconnectHandler(func(nc *nats.Conn) {
+			slog.Info("NATS reconnected", "url", nc.ConnectedUrl())
+		}),
+	}
+
+	if token != "" {
+		opts = append(opts, nats.Token(token))
+	}
+
+	nc, err := nats.Connect(host, opts...)
+	if err != nil {
+		return nil, fmt.Errorf("NATS connect failed: %w", err)
+	}
+
+	slog.Info("Connected to NATS server", "host", host)
+	return nc, nil
+}
+
 // NATSRequest performs a NATS request-response with JSON marshaling.
 // It marshals the input, sends to the given subject, validates the response
 // for error payloads, and unmarshals the successful response into Out.