feat(natsstore): connect to NATS with the configured credentials and TLS

go.mod

@@ -16,10 +16,12 @@ require (
 	github.com/gorilla/websocket v1.5.3
 	github.com/jackc/pgx/v5 v5.9.2
 	github.com/joho/godotenv v1.5.1
-	github.com/klauspost/compress v1.17.4
+	github.com/klauspost/compress v1.20.0
 	github.com/libp2p/zeroconf/v2 v2.2.0
 	github.com/miekg/dns v1.1.43
 	github.com/minio/minio-go/v7 v7.0.66
+	github.com/nats-io/nats.go v1.55.0
+	github.com/nats-io/nkeys v0.4.17
 	github.com/pion/ice/v4 v4.3.0
 	github.com/pion/interceptor v0.1.45
 	github.com/pion/rtp v1.10.4
@@ -46,7 +48,7 @@ replace github.com/MoeForge/moego v0.1.0-alpha.1 => github.com/router-for-me/moe
 require (
 	github.com/cespare/xxhash/v2 v2.3.0 // indirect
 	github.com/dlclark/regexp2/v2 v2.5.1 // indirect
-	github.com/nats-io/nkeys v0.4.17 // indirect
+	github.com/nats-io/nuid v1.0.1 // indirect
 	github.com/pion/datachannel v1.6.2 // indirect
 	github.com/pion/dtls/v3 v3.1.5 // indirect
 	github.com/pion/logging v0.2.4 // indirect

go.sum

@@ -110,8 +110,8 @@ github.com/json-iterator/go v1.1.12 h1:PV8peI4a0ysnczrg+LtxykD8LfKY9ML6u2jnxaEnr
 github.com/json-iterator/go v1.1.12/go.mod h1:e30LSqwooZae/UwlEbR2852Gd8hjQvJoHmT4TnhNGBo=
 github.com/kevinburke/ssh_config v1.6.0 h1:J1FBfmuVosPHf5GRdltRLhPJtJpTlMdKTBjRgTaQBFY=
 github.com/kevinburke/ssh_config v1.6.0/go.mod h1:q2RIzfka+BXARoNexmF9gkxEX7DmvbW9P4hIVx2Kg4M=
-github.com/klauspost/compress v1.17.4 h1:Ej5ixsIri7BrIjBkRZLTo6ghwrEtHFk7ijlczPW4fZ4=
-github.com/klauspost/compress v1.17.4/go.mod h1:/dCuZOvVtNoHsyb+cuJD3itjs3NbnF6KH9zAO4BDxPM=
+github.com/klauspost/compress v1.20.0 h1:a3C1ke2ohxFymNlb2HWAHjDeKCI90scRskErZkR0ezA=
+github.com/klauspost/compress v1.20.0/go.mod h1:LUdAzn7YLVvxLpc7y3V1m40wESHTgc1422pwwBSKYuI=
 github.com/klauspost/cpuid/v2 v2.0.1/go.mod h1:FInQzS24/EEf25PyTYn52gqo7WaD8xa0213Md/qVLRg=
 github.com/klauspost/cpuid/v2 v2.0.9/go.mod h1:FInQzS24/EEf25PyTYn52gqo7WaD8xa0213Md/qVLRg=
 github.com/klauspost/cpuid/v2 v2.3.0 h1:S4CRMLnYUhGeDFDqkGriYKdfoFlDnMtqTiI/sFzhA9Y=
@@ -155,8 +155,12 @@ github.com/muesli/cancelreader v0.2.2 h1:3I4Kt4BQjOR54NavqnDogx/MIoWBFa0StPA8ELU
 github.com/muesli/cancelreader v0.2.2/go.mod h1:3XuTXfFS2VjM+HTLZY9Ak0l6eUKfijIfMUZ4EgX0QYo=
 github.com/muesli/termenv v0.16.0 h1:S5AlUN9dENB57rsbnkPyfdGuWIlkmzJjbFf0Tf5FWUc=
 github.com/muesli/termenv v0.16.0/go.mod h1:ZRfOIKPFDYQoDFF4Olj7/QJbW60Ol/kL1pU3VfY/Cnk=
+github.com/nats-io/nats.go v1.55.0 h1:Ng8QKHdvdHIWoaiq0+d/aH1BWkxvcJxWJ0Y/9BPKx3M=
+github.com/nats-io/nats.go v1.55.0/go.mod h1:y+DZoD1oBOYfZTU681eTUiUjI0vbqYGixNVFHcjHJ0k=
 github.com/nats-io/nkeys v0.4.17 h1:JAJrlDhOyQqOt0a1eg4cbIZ1Kc7+vmkCigAltpwQpVA=
 github.com/nats-io/nkeys v0.4.17/go.mod h1:llLgWoI0o4z/Q57q2R1kHfmocyhGV6VG/U18Glg1Afs=
+github.com/nats-io/nuid v1.0.1 h1:5iA8DT8V7q8WK2EScv2padNa/rTESc1KdnPw4TC2paw=
+github.com/nats-io/nuid v1.0.1/go.mod h1:19wcPz3Ph3q0Jbyiqsd0kePYG7A95tJPxeL+1OSON2c=
 github.com/pelletier/go-toml/v2 v2.4.3-0.20260624044523-35f78d5908f1 h1:gIZQQ2+wbsdQGWLPDSdJFptR25tBCjSC0KbSIQZ1rgE=
 github.com/pelletier/go-toml/v2 v2.4.3-0.20260624044523-35f78d5908f1/go.mod h1:2gIqNv+qfxSVS7cM2xJQKtLSTLUE9V8t9Stt+h56mCY=
 github.com/pierrec/xxHash v0.1.5 h1:n/jBpwTHiER4xYvK3/CdPVnLDPchj8eTJFFLUb4QHBo=

internal/store/natsstore/connect.go

@@ -0,0 +1,210 @@
+package natsstore
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"strings"
+	"sync/atomic"
+	"time"
+
+	"github.com/nats-io/nats.go"
+	"github.com/nats-io/nkeys"
+	log "github.com/sirupsen/logrus"
+)
+
+const (
+	// DefaultConnectTimeout bounds one connection attempt and the initial wait.
+	DefaultConnectTimeout = 30 * time.Second
+	// MaxReconnectDelay caps the exponential reconnect backoff.
+	MaxReconnectDelay = 30 * time.Second
+
+	clientName = "cliproxyapi"
+)
+
+// ErrNotConnected is returned when the initial connection is not established in time.
+var ErrNotConnected = errors.New("not connected to NATS")
+
+// reconnectDelay doubles per attempt from one second and stops at MaxReconnectDelay.
+func reconnectDelay(attempts int) time.Duration {
+	if attempts < 1 {
+		attempts = 1
+	}
+	if attempts > 6 {
+		return MaxReconnectDelay
+	}
+	return min(MaxReconnectDelay, time.Duration(1<<uint(attempts-1))*time.Second)
+}
+
+// buildOptions turns the resolved configuration into client options. Secrets are
+// resolved here so a bad reference fails before any network traffic, and no error
+// it returns carries a secret value.
+func buildOptions(cfg Config, env Lookup) ([]nats.Option, error) {
+	var opts []nats.Option
+	switch auth := cfg.Auth.(type) {
+	case CredentialsAuth:
+		path, err := auth.File.Path()
+		if err != nil {
+			return nil, &SettingError{Setting: EnvCreds, Err: err}
+		}
+		opts = append(opts, nats.UserCredentials(path))
+	case NKeyAuth:
+		seed, err := auth.Seed.resolve()
+		if err != nil {
+			return nil, &SettingError{Setting: EnvNKey, Err: err}
+		}
+		pair, err := nkeys.FromSeed([]byte(seed))
+		if err != nil {
+			return nil, &SettingError{Setting: EnvNKey, Err: fmt.Errorf("invalid NKey seed: %w", err)}
+		}
+		pub, err := pair.PublicKey()
+		if err != nil {
+			return nil, &SettingError{Setting: EnvNKey, Err: err}
+		}
+		opts = append(opts, nats.Nkey(pub, pair.Sign))
+	case UserPasswordAuth:
+		password, err := auth.Password.resolve(env)
+		if err != nil {
+			return nil, &SettingError{Setting: EnvPassword, Err: err}
+		}
+		opts = append(opts, nats.UserInfo(auth.User, password))
+	case JWTAuth:
+		jwt, err := auth.JWT.resolve(env)
+		if err != nil {
+			return nil, &SettingError{Setting: EnvJWT, Err: err}
+		}
+		jwt = strings.TrimSpace(jwt)
+		sign := func([]byte) ([]byte, error) { return nil, nil }
+		if auth.Seed != nil {
+			seed, err := auth.Seed.resolve(env)
+			if err != nil {
+				return nil, &SettingError{Setting: EnvSeed, Err: err}
+			}
+			pair, err := nkeys.FromSeed([]byte(strings.TrimSpace(seed)))
+			if err != nil {
+				return nil, &SettingError{Setting: EnvSeed, Err: fmt.Errorf("invalid NKey seed: %w", err)}
+			}
+			sign = pair.Sign
+		}
+		opts = append(opts, nats.UserJWT(func() (string, error) { return jwt, nil }, sign))
+	case TokenAuth:
+		token, err := auth.Token.resolve(env)
+		if err != nil {
+			return nil, &SettingError{Setting: EnvToken, Err: err}
+		}
+		opts = append(opts, nats.Token(token))
+	}
+
+	if !cfg.TLS.Configured() {
+		return opts, nil
+	}
+	opts = append(opts, nats.Secure())
+	pair, err := cfg.TLS.clientCertificate()
+	if err != nil {
+		return nil, err
+	}
+	if pair != nil {
+		cert, err := pair[0].Path()
+		if err != nil {
+			return nil, &SettingError{Setting: EnvCert, Err: err}
+		}
+		key, err := pair[1].Path()
+		if err != nil {
+			return nil, &SettingError{Setting: EnvKey, Err: err}
+		}
+		opts = append(opts, nats.ClientCert(cert, key))
+	}
+	if cfg.TLS.CA != nil {
+		ca, err := cfg.TLS.CA.Path()
+		if err != nil {
+			return nil, &SettingError{Setting: EnvCA, Err: err}
+		}
+		opts = append(opts, nats.RootCAs(ca))
+	}
+	return opts, nil
+}
+
+// Connect opens a client connection that keeps reconnecting with capped backoff,
+// and waits until the first connection is up or the timeout or context ends.
+// Server addresses are logged with any URL userinfo removed, and secrets never are.
+func Connect(ctx context.Context, cfg Config, env Lookup, timeout time.Duration) (*nats.Conn, error) {
+	if timeout <= 0 {
+		timeout = DefaultConnectTimeout
+	}
+	opts, err := buildOptions(cfg, env)
+	if err != nil {
+		return nil, err
+	}
+
+	var lastErr atomic.Value
+	remember := func(err error) {
+		if err != nil {
+			lastErr.Store(err.Error())
+		}
+	}
+	servers := redactServers(cfg.Servers)
+	opts = append(opts,
+		nats.Name(clientName),
+		nats.RetryOnFailedConnect(true),
+		nats.MaxReconnects(-1),
+		nats.Timeout(timeout),
+		nats.CustomReconnectDelay(reconnectDelay),
+		nats.DisconnectErrHandler(func(_ *nats.Conn, err error) {
+			remember(err)
+			log.WithError(err).Warn("NATS disconnected, will attempt reconnect")
+		}),
+		nats.ReconnectHandler(func(*nats.Conn) { log.Info("NATS reconnected") }),
+		nats.ClosedHandler(func(*nats.Conn) { log.Info("NATS connection closed") }),
+		nats.ErrorHandler(func(_ *nats.Conn, _ *nats.Subscription, err error) {
+			remember(err)
+			log.WithError(err).Warn("NATS error")
+		}),
+	)
+
+	log.Infof("connecting to NATS servers %v with %s authentication", servers, cfg.Auth.Description())
+	nc, err := nats.Connect(strings.Join(cfg.Servers, ","), opts...)
+	if err != nil {
+		return nil, fmt.Errorf("failed to connect to NATS servers %v: %w", servers, scrub(err, cfg.Servers))
+	}
+
+	deadline := time.NewTimer(timeout)
+	defer deadline.Stop()
+	tick := time.NewTicker(25 * time.Millisecond)
+	defer tick.Stop()
+	for !nc.IsConnected() {
+		var reason error
+		select {
+		case <-ctx.Done():
+			reason = ctx.Err()
+		case <-deadline.C:
+			reason = ErrNotConnected
+		case <-tick.C:
+			if nc.IsClosed() {
+				reason = ErrNotConnected
+			}
+		}
+		if reason != nil {
+			nc.Close()
+			if last, ok := lastErr.Load().(string); ok {
+				return nil, fmt.Errorf("failed to connect to NATS servers %v: %w (last error: %s)", servers, reason, last)
+			}
+			return nil, fmt.Errorf("failed to connect to NATS servers %v: %w", servers, reason)
+		}
+	}
+	log.Infof("connected to NATS server %s", nc.ConnectedUrlRedacted())
+	return nc, nil
+}
+
+// scrub removes every configured server string from an error message so URL userinfo cannot leak.
+func scrub(err error, servers []string) error {
+	msg := err.Error()
+	for _, server := range servers {
+		if safe := redactServer(server); safe != server {
+			msg = strings.ReplaceAll(msg, server, safe)
+		}
+	}
+	if msg == err.Error() {
+		return err
+	}
+	return errors.New(msg)
+}