fix: env var decoding for bool config fields, redis sentinel multi-node support and change fs.FS path construction

app/config/sink.go

@@ -176,5 +176,5 @@ func ConfigureSink(v *viper.Viper) {
 	ConfigureKafkaConfiguration(v, "sink")
 
 	// Override Kafka configuration defaults
-	v.SetDefault("sink.kafka.consumerGroupId", "openmeter-sink-worker")
+	v.SetDefault("sink.kafka.consumerGroupID", "openmeter-sink-worker")
 }

app/config/viper.go

@@ -11,6 +11,7 @@ func DecodeHook() mapstructure.DecodeHookFunc {
 		mapstructure.TextUnmarshallerHookFunc(),
 		mapstructure.StringToTimeDurationHookFunc(),
 		mapstructure.StringToSliceHookFunc(","),
+		mapstructure.StringToBasicTypeHookFunc(),
 	)
 }
 

pkg/redis/client.go

@@ -3,6 +3,7 @@ package redis
 import (
 	"crypto/tls"
 	"fmt"
+	"strings"
 
 	"github.com/redis/go-redis/extra/redisotel/v9"
 	"github.com/redis/go-redis/v9"
@@ -57,9 +58,19 @@ func NewClient(o Options, opts ...Option) (*redis.Client, error) {
 	// Initialize Redis Client
 	var client *redis.Client
 	if o.Sentinel.Enabled {
+		// Address may be a comma-separated list of sentinel nodes.
+		var sentinelAddrs []string
+		for _, a := range strings.Split(o.Address, ",") {
+			if trimmed := strings.TrimSpace(a); trimmed != "" {
+				sentinelAddrs = append(sentinelAddrs, trimmed)
+			}
+		}
+		if len(sentinelAddrs) == 0 {
+			return nil, fmt.Errorf("sentinel mode enabled but no valid sentinel addresses provided")
+		}
 		client = redis.NewFailoverClient(&redis.FailoverOptions{
 			MasterName:    o.Sentinel.MasterName,
-			SentinelAddrs: []string{o.Address},
+			SentinelAddrs: sentinelAddrs,
 			DB:            o.Database,
 			Username:      o.Username,
 			Password:      o.Password,

tools/migrate/fs.go

@@ -4,7 +4,7 @@ import (
 	"bufio"
 	"bytes"
 	"io/fs"
-	"path/filepath"
+	"path"
 	"strings"
 )
 
@@ -29,15 +29,15 @@ func (s *SourceWrapper) Open(name string) (fs.File, error) {
 	return s.fsys.Open(name)
 }
 
-func (s *SourceWrapper) ReadDir(path string) ([]fs.DirEntry, error) {
-	entries, err := fs.ReadDir(s.fsys, path)
+func (s *SourceWrapper) ReadDir(dir string) ([]fs.DirEntry, error) {
+	entries, err := fs.ReadDir(s.fsys, dir)
 	if err != nil {
 		return nil, err
 	}
 
 	results := make([]fs.DirEntry, 0, len(entries))
 	for _, entry := range entries {
-		filePath := filepath.Join(path, entry.Name())
+		filePath := path.Join(dir, entry.Name())
 
 		if entry.IsDir() {
 			r, err := s.ReadDir(filePath)