feedkit: ergonomics pass (shared logger, route compiler, param helpers)

- Add logging.Logf as the canonical printf-style logger type used across feedkit. - Update scheduler and dispatch to alias their Logger types to logging.Logf. - Eliminates type-mismatch friction when wiring one log function through the system. - Add dispatch.CompileRoutes(*config.Config) ([]dispatch.Route, error) - Compiles config routes into dispatch routes with event.ParseKind normalization. - If routes: is omitted, defaults to “all sinks receive all kinds”. - Expand config param helpers for both SourceConfig and SinkConfig - Add ParamBool/ParamInt/ParamDuration/ParamStringSlice (+ Default variants). - Supports common YAML-decoded types (bool/int/float/string, []any, etc.) - Keeps driver code cleaner and reduces repeated type assertions. - Fix Postgres sink validation error prefix ("postgres sink", not "rabbitmq sink").
2026-01-13 14:40:29 -06:00
parent 0cc2862170
commit 09bc65e947
9 changed files with 896 additions and 44 deletions
--- a/config/config.go
+++ b/config/config.go
@@ -128,12 +128,3 @@ func (d *Duration) UnmarshalYAML(value *yaml.Node) error {
 	// Anything else: reject.
 	return fmt.Errorf("duration must be a string like 15m or an integer minutes, got tag %s", value.Tag)
 }
-
-func isAllDigits(s string) bool {
-	for _, r := range s {
-		if r < '0' || r > '9' {
-			return false
-		}
-	}
-	return len(s) > 0
-}
--- a/config/params.go
+++ b/config/params.go
@@ -1,32 +1,21 @@
 // feedkit/config/params.go
 package config

-import "strings"
+import (
+	"math"
+	"strconv"
+	"strings"
+	"time"
+)
+
+// ---- SourceConfig param helpers ----

 // ParamString returns the first non-empty string found for any of the provided keys.
 // Values must actually be strings in the decoded config; other types are ignored.
 //
 // This keeps cfg.Params flexible (map[string]any) while letting callers stay type-safe.
 func (cfg SourceConfig) ParamString(keys ...string) (string, bool) {
-	if cfg.Params == nil {
-		return "", false
-	}
-	for _, k := range keys {
-		v, ok := cfg.Params[k]
-		if !ok || v == nil {
-			continue
-		}
-		s, ok := v.(string)
-		if !ok {
-			continue
-		}
-		s = strings.TrimSpace(s)
-		if s == "" {
-			continue
-		}
-		return s, true
-	}
-	return "", false
+	return paramString(cfg.Params, keys...)
 }

 // ParamStringDefault returns ParamString(keys...) if present; otherwise it returns def.
@@ -38,14 +27,150 @@ func (cfg SourceConfig) ParamStringDefault(def string, keys ...string) string {
 	return strings.TrimSpace(def)
 }

+// ParamBool returns the first boolean found for any of the provided keys.
+//
+// Accepted types in Params:
+//   - bool
+//   - string: parsed via strconv.ParseBool ("true"/"false"/"1"/"0", etc.)
+func (cfg SourceConfig) ParamBool(keys ...string) (bool, bool) {
+	return paramBool(cfg.Params, keys...)
+}
+
+func (cfg SourceConfig) ParamBoolDefault(def bool, keys ...string) bool {
+	if v, ok := cfg.ParamBool(keys...); ok {
+		return v
+	}
+	return def
+}
+
+// ParamInt returns the first integer-like value found for any of the provided keys.
+//
+// Accepted types in Params:
+//   - any integer type (int, int64, uint32, ...)
+//   - float32/float64 ONLY if it is an exact integer (e.g. 15.0)
+//   - string: parsed via strconv.Atoi (e.g. "42")
+func (cfg SourceConfig) ParamInt(keys ...string) (int, bool) {
+	return paramInt(cfg.Params, keys...)
+}
+
+func (cfg SourceConfig) ParamIntDefault(def int, keys ...string) int {
+	if v, ok := cfg.ParamInt(keys...); ok {
+		return v
+	}
+	return def
+}
+
+// ParamDuration returns the first duration-like value found for any of the provided keys.
+//
+// Accepted types in Params:
+//   - time.Duration
+//   - string: parsed via time.ParseDuration (e.g. "250ms", "30s", "5m")
+//   - if the string is all digits (e.g. "30"), it is interpreted as SECONDS
+//   - numeric: interpreted as SECONDS (e.g. 30 => 30s)
+//
+// Rationale: Param durations are usually timeouts/backoffs; seconds are a sane numeric default.
+// If you want minutes/hours, prefer a duration string like "5m" or "1h".
+func (cfg SourceConfig) ParamDuration(keys ...string) (time.Duration, bool) {
+	return paramDuration(cfg.Params, keys...)
+}
+
+func (cfg SourceConfig) ParamDurationDefault(def time.Duration, keys ...string) time.Duration {
+	if v, ok := cfg.ParamDuration(keys...); ok {
+		return v
+	}
+	return def
+}
+
+// ParamStringSlice returns the first string-slice-like value found for any of the provided keys.
+//
+// Accepted types in Params:
+//   - []string
+//   - []any where each element is a string
+//   - string:
+//   - if it contains commas, split on commas (",") and trim each item
+//   - otherwise treat as a single-item list
+//
+// Empty/blank items are removed.
+func (cfg SourceConfig) ParamStringSlice(keys ...string) ([]string, bool) {
+	return paramStringSlice(cfg.Params, keys...)
+}
+
+// ---- SinkConfig param helpers ----
+
 // ParamString returns the first non-empty string found for any of the provided keys
 // in SinkConfig.Params. (Same rationale as SourceConfig.ParamString.)
 func (cfg SinkConfig) ParamString(keys ...string) (string, bool) {
-	if cfg.Params == nil {
-		return "", false
+	return paramString(cfg.Params, keys...)
+}
+
+// ParamStringDefault returns ParamString(keys...) if present; otherwise it returns def.
+// Symmetric helper for sink implementations.
+func (cfg SinkConfig) ParamStringDefault(def string, keys ...string) string {
+	if s, ok := cfg.ParamString(keys...); ok {
+		return s
+	}
+	return strings.TrimSpace(def)
+}
+
+func (cfg SinkConfig) ParamBool(keys ...string) (bool, bool) {
+	return paramBool(cfg.Params, keys...)
+}
+
+func (cfg SinkConfig) ParamBoolDefault(def bool, keys ...string) bool {
+	if v, ok := cfg.ParamBool(keys...); ok {
+		return v
+	}
+	return def
+}
+
+func (cfg SinkConfig) ParamInt(keys ...string) (int, bool) {
+	return paramInt(cfg.Params, keys...)
+}
+
+func (cfg SinkConfig) ParamIntDefault(def int, keys ...string) int {
+	if v, ok := cfg.ParamInt(keys...); ok {
+		return v
+	}
+	return def
+}
+
+func (cfg SinkConfig) ParamDuration(keys ...string) (time.Duration, bool) {
+	return paramDuration(cfg.Params, keys...)
+}
+
+func (cfg SinkConfig) ParamDurationDefault(def time.Duration, keys ...string) time.Duration {
+	if v, ok := cfg.ParamDuration(keys...); ok {
+		return v
+	}
+	return def
+}
+
+func (cfg SinkConfig) ParamStringSlice(keys ...string) ([]string, bool) {
+	return paramStringSlice(cfg.Params, keys...)
+}
+
+// ---- shared implementations (package-private) ----
+
+func paramAny(params map[string]any, keys ...string) (any, bool) {
+	if params == nil {
+		return nil, false
 	}
 	for _, k := range keys {
-		v, ok := cfg.Params[k]
+		v, ok := params[k]
+		if !ok || v == nil {
+			continue
+		}
+		return v, true
+	}
+	return nil, false
+}
+
+func paramString(params map[string]any, keys ...string) (string, bool) {
+	for _, k := range keys {
+		if params == nil {
+			return "", false
+		}
+		v, ok := params[k]
 		if !ok || v == nil {
 			continue
 		}
@@ -62,11 +187,213 @@ func (cfg SinkConfig) ParamString(keys ...string) (string, bool) {
 	return "", false
 }

-// ParamStringDefault returns ParamString(keys...) if present; otherwise it returns def.
-// Symmetric helper for sink implementations.
-func (cfg SinkConfig) ParamStringDefault(def string, keys ...string) string {
-	if s, ok := cfg.ParamString(keys...); ok {
-		return s
+func paramBool(params map[string]any, keys ...string) (bool, bool) {
+	v, ok := paramAny(params, keys...)
+	if !ok {
+		return false, false
+	}
+
+	switch t := v.(type) {
+	case bool:
+		return t, true
+	case string:
+		s := strings.TrimSpace(t)
+		if s == "" {
+			return false, false
+		}
+		parsed, err := strconv.ParseBool(s)
+		if err != nil {
+			return false, false
+		}
+		return parsed, true
+	default:
+		return false, false
 	}
-	return strings.TrimSpace(def)
+}
+
+func paramInt(params map[string]any, keys ...string) (int, bool) {
+	v, ok := paramAny(params, keys...)
+	if !ok {
+		return 0, false
+	}
+
+	switch t := v.(type) {
+	case int:
+		return t, true
+	case int8:
+		return int(t), true
+	case int16:
+		return int(t), true
+	case int32:
+		return int(t), true
+	case int64:
+		return int(t), true
+
+	case uint:
+		return int(t), true
+	case uint8:
+		return int(t), true
+	case uint16:
+		return int(t), true
+	case uint32:
+		return int(t), true
+	case uint64:
+		return int(t), true
+
+	case float32:
+		f := float64(t)
+		if math.IsNaN(f) || math.IsInf(f, 0) {
+			return 0, false
+		}
+		if math.Trunc(f) != f {
+			return 0, false
+		}
+		return int(f), true
+
+	case float64:
+		if math.IsNaN(t) || math.IsInf(t, 0) {
+			return 0, false
+		}
+		if math.Trunc(t) != t {
+			return 0, false
+		}
+		return int(t), true
+
+	case string:
+		s := strings.TrimSpace(t)
+		if s == "" {
+			return 0, false
+		}
+		n, err := strconv.Atoi(s)
+		if err != nil {
+			return 0, false
+		}
+		return n, true
+
+	default:
+		return 0, false
+	}
+}
+
+func paramDuration(params map[string]any, keys ...string) (time.Duration, bool) {
+	v, ok := paramAny(params, keys...)
+	if !ok {
+		return 0, false
+	}
+
+	switch t := v.(type) {
+	case time.Duration:
+		if t <= 0 {
+			return 0, false
+		}
+		return t, true
+
+	case string:
+		s := strings.TrimSpace(t)
+		if s == "" {
+			return 0, false
+		}
+		// Numeric strings are interpreted as seconds (see doc comment).
+		if isAllDigits(s) {
+			n, err := strconv.Atoi(s)
+			if err != nil || n <= 0 {
+				return 0, false
+			}
+			return time.Duration(n) * time.Second, true
+		}
+		d, err := time.ParseDuration(s)
+		if err != nil || d <= 0 {
+			return 0, false
+		}
+		return d, true
+
+	case int:
+		if t <= 0 {
+			return 0, false
+		}
+		return time.Duration(t) * time.Second, true
+	case int64:
+		if t <= 0 {
+			return 0, false
+		}
+		return time.Duration(t) * time.Second, true
+	case float64:
+		if math.IsNaN(t) || math.IsInf(t, 0) || t <= 0 {
+			return 0, false
+		}
+		// Allow fractional seconds.
+		secs := t * float64(time.Second)
+		return time.Duration(secs), true
+	case float32:
+		f := float64(t)
+		if math.IsNaN(f) || math.IsInf(f, 0) || f <= 0 {
+			return 0, false
+		}
+		secs := f * float64(time.Second)
+		return time.Duration(secs), true
+
+	default:
+		return 0, false
+	}
+}
+
+func paramStringSlice(params map[string]any, keys ...string) ([]string, bool) {
+	v, ok := paramAny(params, keys...)
+	if !ok {
+		return nil, false
+	}
+
+	clean := func(items []string) ([]string, bool) {
+		out := make([]string, 0, len(items))
+		for _, it := range items {
+			it = strings.TrimSpace(it)
+			if it == "" {
+				continue
+			}
+			out = append(out, it)
+		}
+		if len(out) == 0 {
+			return nil, false
+		}
+		return out, true
+	}
+
+	switch t := v.(type) {
+	case []string:
+		return clean(t)
+
+	case []any:
+		tmp := make([]string, 0, len(t))
+		for _, it := range t {
+			s, ok := it.(string)
+			if !ok {
+				continue
+			}
+			tmp = append(tmp, s)
+		}
+		return clean(tmp)
+
+	case string:
+		s := strings.TrimSpace(t)
+		if s == "" {
+			return nil, false
+		}
+		if strings.Contains(s, ",") {
+			parts := strings.Split(s, ",")
+			return clean(parts)
+		}
+		return clean([]string{s})
+
+	default:
+		return nil, false
+	}
+}
+
+func isAllDigits(s string) bool {
+	for _, r := range s {
+		if r < '0' || r > '9' {
+			return false
+		}
+	}
+	return len(s) > 0
 }