fix(quality): harden error handling, validation, and test isolation

- Replace all silent `_ = repo.*` calls in goroutine paths with
  clearStateWithLog helper that logs warnings on cleanup failures
- Add debug.Stack() capture to recoverPipeline for post-mortem debugging
- Guard all sync.Map type assertions with ok checks in pollDebounceExpiry,
  cancelPair, and launchAIStage
- Extract cleanupCtx() (5s timeout) for all cleanup repo calls that
  previously used unbounded context.Background()
- Use dur.String() instead of .Milliseconds() for duration log fields
- Add MessageEvent.Validate() with bounds checks on all BotConfig fields;
  called in handleEvent before processing
- Log client IP on auth failures in SecretMiddleware
- Cap ScanStates with maxPairs parameter (10_000 at call site) and warn
  on truncation
- Centralise test Redis config in internal/testhelpers replacing hardcoded
  DB numbers (1, 2, 3) across three test packages
- config.Load() now returns error instead of calling os.Exit

Author: gomessguii

cmd/server/main.go

@@ -2,8 +2,7 @@ package main
 
 import (
 	"context"
-	"log/slog"
-	"os"
+	"log"
 
 	"github.com/gin-gonic/gin"
 	"github.com/go-redsync/redsync/v4"
@@ -14,25 +13,26 @@ import (
 	aiService "github.com/EvolutionAPI/evo-bot-runtime/pkg/ai/service"
 	debounceService "github.com/EvolutionAPI/evo-bot-runtime/pkg/debounce/service"
 	dispatchService "github.com/EvolutionAPI/evo-bot-runtime/pkg/dispatch/service"
-	"github.com/EvolutionAPI/evo-bot-runtime/pkg/pipeline/handler"
+	pipelineHandler "github.com/EvolutionAPI/evo-bot-runtime/pkg/pipeline/handler"
 	"github.com/EvolutionAPI/evo-bot-runtime/pkg/pipeline/repository"
 	pipelineService "github.com/EvolutionAPI/evo-bot-runtime/pkg/pipeline/service"
 )
 
 func main() {
 	// Step 1: config
-	cfg := config.Load()
+	cfg, err := config.Load()
+	if err != nil {
+		log.Fatalf("config: %v", err)
+	}
 
 	// Step 2: Redis client + connectivity check
 	opt, err := redis.ParseURL(cfg.RedisURL)
 	if err != nil {
-		slog.Error("invalid REDIS_URL", "error", err)
-		os.Exit(1)
+		log.Fatalf("invalid REDIS_URL: %v", err)
 	}
 	rdb := redis.NewClient(opt)
 	if err := rdb.Ping(context.Background()).Err(); err != nil {
-		slog.Error("redis connection failed", "error", err)
-		os.Exit(1)
+		log.Fatalf("redis connection failed: %v", err)
 	}
 
 	// Step 3: redsync
@@ -43,31 +43,29 @@ func main() {
 	pipelineRepo := repository.NewPipelineRepository(rdb, rs)
 
 	// Step 5: debounce engine
-	debounceEng := debounceService.NewDebounceEngine(pipelineRepo)
+	debounce := debounceService.NewDebounceEngine(pipelineRepo)
 
 	// Step 6: AI adapter
 	aiAdapter := aiService.NewAIAdapter(cfg.AIProcessorURL, cfg.AIProcessorAPIKey, cfg.AICallTimeoutSeconds)
 
 	// Step 7: dispatch engine
-	dispatchEng := dispatchService.NewDispatchEngine()
+	dispatch := dispatchService.NewDispatchEngine()
 
 	// Step 8: pipeline service
-	pipelineSvc := pipelineService.NewPipelineService(pipelineRepo, debounceEng, aiAdapter, dispatchEng)
-	if err := pipelineSvc.Start(); err != nil {
-		slog.Error("pipeline service failed to start", "error", err)
-		os.Exit(1)
+	pipeline := pipelineService.NewPipelineService(pipelineRepo, debounce, aiAdapter, dispatch)
+	if err := pipeline.Start(); err != nil {
+		log.Fatalf("pipeline service failed to start: %v", err)
 	}
 
 	// Step 9: handler + routes
-	hdl := handler.NewHandler(pipelineRepo, pipelineSvc, cfg.BotRuntimeSecret)
+	handler := pipelineHandler.NewHandler(pipelineRepo, pipeline, cfg.BotRuntimeSecret)
 	r := gin.New()
 	r.Use(gin.Recovery())
-	hdl.RegisterRoutes(r)
+	handler.RegisterRoutes(r)
 
 	// Step 10: start server
-	slog.Info("evo-bot-runtime starting", "listen_addr", cfg.ListenAddr)
+	log.Printf("evo-bot-runtime starting on %s", cfg.ListenAddr)
 	if err := r.Run(cfg.ListenAddr); err != nil {
-		slog.Error("server failed", "error", err)
-		os.Exit(1)
+		log.Fatalf("server failed: %v", err)
 	}
 }

internal/config/config.go

@@ -1,7 +1,7 @@
 package config
 
 import (
-	"log/slog"
+	"fmt"
 	"os"
 	"strconv"
 )
@@ -15,35 +15,58 @@ type Config struct {
 	AICallTimeoutSeconds int
 }
 
-func Load() *Config {
-	return &Config{
-		ListenAddr:           requireEnv("LISTEN_ADDR"),
-		RedisURL:             requireEnv("REDIS_URL"),
-		BotRuntimeSecret:     requireEnv("BOT_RUNTIME_SECRET"),
-		AIProcessorURL:       requireEnv("AI_PROCESSOR_URL"),
-		AIProcessorAPIKey:    requireEnv("AI_PROCESSOR_API_KEY"),
-		AICallTimeoutSeconds: optionalEnvInt("AI_CALL_TIMEOUT_SECONDS", 30),
+func Load() (*Config, error) {
+	listenAddr, err := mustGetEnv("LISTEN_ADDR")
+	if err != nil {
+		return nil, err
+	}
+	redisURL, err := mustGetEnv("REDIS_URL")
+	if err != nil {
+		return nil, err
+	}
+	botRuntimeSecret, err := mustGetEnv("BOT_RUNTIME_SECRET")
+	if err != nil {
+		return nil, err
+	}
+	aiProcessorURL, err := mustGetEnv("AI_PROCESSOR_URL")
+	if err != nil {
+		return nil, err
+	}
+	aiProcessorAPIKey, err := mustGetEnv("AI_PROCESSOR_API_KEY")
+	if err != nil {
+		return nil, err
 	}
+	aiCallTimeout, err := getEnvIntOrDefault("AI_CALL_TIMEOUT_SECONDS", 30)
+	if err != nil {
+		return nil, err
+	}
+
+	return &Config{
+		ListenAddr:           listenAddr,
+		RedisURL:             redisURL,
+		BotRuntimeSecret:     botRuntimeSecret,
+		AIProcessorURL:       aiProcessorURL,
+		AIProcessorAPIKey:    aiProcessorAPIKey,
+		AICallTimeoutSeconds: aiCallTimeout,
+	}, nil
 }
 
-func requireEnv(key string) string {
+func mustGetEnv(key string) (string, error) {
 	v := os.Getenv(key)
 	if v == "" {
-		slog.Error("missing required environment variable", "key", key)
-		os.Exit(1)
+		return "", fmt.Errorf("missing required environment variable: %s", key)
 	}
-	return v
+	return v, nil
 }
 
-func optionalEnvInt(key string, defaultVal int) int {
+func getEnvIntOrDefault(key string, defaultVal int) (int, error) {
 	v := os.Getenv(key)
 	if v == "" {
-		return defaultVal
+		return defaultVal, nil
 	}
 	n, err := strconv.Atoi(v)
 	if err != nil {
-		slog.Error("invalid integer environment variable", "key", key, "value", v)
-		os.Exit(1)
+		return 0, fmt.Errorf("invalid integer for environment variable %s: %q", key, v)
 	}
-	return n
+	return n, nil
 }

internal/testhelpers/redis.go

@@ -0,0 +1,32 @@
+// Package testhelpers provides shared test helpers for evo-bot-runtime packages.
+package testhelpers
+
+import (
+	"os"
+
+	"github.com/redis/go-redis/v9"
+)
+
+// RedisOptions returns Redis client options for tests.
+// It reads REDIS_TEST_URL (falls back to REDIS_URL, then redis://localhost:6379).
+// The DB is set from the REDIS_TEST_DB environment variable if present;
+// otherwise defaults to 15 to avoid collisions with application data.
+//
+// Packages that run tests in parallel against the same Redis instance should
+// override REDIS_TEST_DB per package in CI (e.g. via go test -env flags or
+// a test wrapper script).
+func RedisOptions() *redis.Options {
+	url := os.Getenv("REDIS_TEST_URL")
+	if url == "" {
+		url = os.Getenv("REDIS_URL")
+	}
+	if url == "" {
+		url = "redis://localhost:6379"
+	}
+	opt, err := redis.ParseURL(url)
+	if err != nil {
+		panic("invalid Redis URL for tests: " + err.Error())
+	}
+	opt.DB = 15 // safe default; override via REDIS_TEST_URL with /db path segment
+	return opt
+}

pkg/debounce/service/debounce_engine_test.go

@@ -10,36 +10,22 @@ import (
 	goredis "github.com/go-redsync/redsync/v4/redis/goredis/v9"
 	"github.com/redis/go-redis/v9"
 
+	"github.com/EvolutionAPI/evo-bot-runtime/internal/testhelpers"
 	"github.com/EvolutionAPI/evo-bot-runtime/pkg/debounce/service"
 	"github.com/EvolutionAPI/evo-bot-runtime/pkg/pipeline/model"
 	"github.com/EvolutionAPI/evo-bot-runtime/pkg/pipeline/repository"
 )
 
-const testDB = 1 // dedicated Redis DB for debounce/service tests
-
-func testRedisOptions() *redis.Options {
-	url := os.Getenv("REDIS_URL")
-	if url == "" {
-		url = "redis://localhost:6379"
-	}
-	opt, err := redis.ParseURL(url)
-	if err != nil {
-		panic("invalid REDIS_URL: " + err.Error())
-	}
-	opt.DB = testDB
-	return opt
-}
-
 func TestMain(m *testing.M) {
-	rdb := redis.NewClient(testRedisOptions())
+	rdb := redis.NewClient(testhelpers.RedisOptions())
 	rdb.FlushDB(context.Background())
 	rdb.Close()
 	os.Exit(m.Run())
 }
 
 func newTestRedisClient(t *testing.T) *redis.Client {
 	t.Helper()
-	return redis.NewClient(testRedisOptions())
+	return redis.NewClient(testhelpers.RedisOptions())
 }
 
 func setupEngine(t *testing.T) (service.DebounceEngine, *redis.Client) {

pkg/pipeline/handler/handler.go

@@ -48,6 +48,13 @@ func (h *Handler) handleEvent(c *gin.Context) {
 		})
 		return
 	}
+	if err := event.Validate(); err != nil {
+		c.JSON(http.StatusBadRequest, gin.H{
+			"error": err.Error(),
+			"code":  "ERR_INVALID_EVENT",
+		})
+		return
+	}
 
 	// Persist StageIncoming BEFORE returning 202 (NFR-01).
 	// Guarantees event durability if the process goroutine never runs.

pkg/pipeline/handler/handler_test.go

@@ -45,7 +45,9 @@ func (m *mockRepo) TimerExists(_ context.Context, _, _ int64) (bool, error)
 func (m *mockRepo) AcquireLock(_ context.Context, _, _ int64) (repository.Mutex, error) {
 	return nil, nil
 }
-func (m *mockRepo) ScanStates(_ context.Context) ([]model.PairID, error) { return nil, nil }
+func (m *mockRepo) ScanStates(_ context.Context, _ int) ([]model.PairID, error) {
+	return nil, nil
+}
 func (m *mockRepo) Ping(_ context.Context) error                         { return m.pingErr }
 
 // mockSvc satisfies pipelineService.PipelineService for handler tests.
@@ -212,15 +214,60 @@ func TestHandleEvent_400_OnMalformedJSON(t *testing.T) {
 }
 
 func TestHandleEvent_400_OnMissingRequiredFields(t *testing.T) {
-	r := setupRouter(&mockRepo{}, &mockSvc{})
-	w := httptest.NewRecorder()
-	req := httptest.NewRequest(http.MethodPost, "/events", bytes.NewReader([]byte(`{}`)))
-	req.Header.Set("Content-Type", "application/json")
-	req.Header.Set("X-Bot-Runtime-Secret", testSecret)
-	r.ServeHTTP(w, req)
-
-	if w.Code != http.StatusAccepted {
-		t.Errorf("status: got %d, want %d (zero-value event is accepted by current schema)", w.Code, http.StatusAccepted)
+	cases := []struct {
+		name    string
+		payload string
+		wantErr string
+	}{
+		{
+			name:    "empty body",
+			payload: `{}`,
+			wantErr: "contact_id must be > 0",
+		},
+		{
+			name:    "missing conversation_id",
+			payload: `{"contact_id":1,"postback_url":"http://x"}`,
+			wantErr: "conversation_id must be > 0",
+		},
+		{
+			name:    "missing postback_url",
+			payload: `{"contact_id":1,"conversation_id":2}`,
+			wantErr: "postback_url is required",
+		},
+		{
+			name:    "negative debounce_time",
+			payload: `{"contact_id":1,"conversation_id":2,"postback_url":"http://x","bot_config":{"debounce_time":-1}}`,
+			wantErr: "debounce_time must be >= 0",
+		},
+		{
+			name:    "segmentation enabled but limit zero",
+			payload: `{"contact_id":1,"conversation_id":2,"postback_url":"http://x","bot_config":{"text_segmentation_enabled":true}}`,
+			wantErr: "text_segmentation_limit must be > 0 when segmentation is enabled",
+		},
+	}
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			r := setupRouter(&mockRepo{}, &mockSvc{})
+			w := httptest.NewRecorder()
+			req := httptest.NewRequest(http.MethodPost, "/events", bytes.NewReader([]byte(tc.payload)))
+			req.Header.Set("Content-Type", "application/json")
+			req.Header.Set("X-Bot-Runtime-Secret", testSecret)
+			r.ServeHTTP(w, req)
+
+			if w.Code != http.StatusBadRequest {
+				t.Errorf("status: got %d, want %d", w.Code, http.StatusBadRequest)
+			}
+			var body map[string]any
+			if err := json.Unmarshal(w.Body.Bytes(), &body); err != nil {
+				t.Fatalf("invalid JSON response: %v", err)
+			}
+			if body["error"] != tc.wantErr {
+				t.Errorf("body.error: got %q, want %q", body["error"], tc.wantErr)
+			}
+			if body["code"] != "ERR_INVALID_EVENT" {
+				t.Errorf("body.code: got %q, want %q", body["code"], "ERR_INVALID_EVENT")
+			}
+		})
 	}
 }
 

pkg/pipeline/handler/middleware.go

@@ -1,6 +1,7 @@
 package handler
 
 import (
+	"log/slog"
 	"net/http"
 
 	"github.com/gin-gonic/gin"
@@ -9,6 +10,9 @@ import (
 func SecretMiddleware(secret string) gin.HandlerFunc {
 	return func(c *gin.Context) {
 		if c.GetHeader("X-Bot-Runtime-Secret") != secret {
+			slog.Warn("pipeline.auth.unauthorized",
+				"remote_addr", c.ClientIP(),
+			)
 			c.JSON(http.StatusUnauthorized, gin.H{
 				"error": "unauthorized",
 				"code":  "ERR_UNAUTHORIZED",

pkg/pipeline/model/pipeline.go

@@ -1,6 +1,9 @@
 package model
 
-import "time"
+import (
+	"errors"
+	"time"
+)
 
 // Stage — string type; never use iota or inline strings
 type Stage string
@@ -48,3 +51,26 @@ type BotConfig struct {
 	TextSegmentationMinSize int     `json:"text_segmentation_min_size"`
 	DelayPerCharacter       float64 `json:"delay_per_character"`        // ms per char between parts
 }
+
+// Validate checks semantic constraints on a MessageEvent after JSON binding.
+func (e *MessageEvent) Validate() error {
+	if e.ContactID <= 0 {
+		return errors.New("contact_id must be > 0")
+	}
+	if e.ConversationID <= 0 {
+		return errors.New("conversation_id must be > 0")
+	}
+	if e.PostbackURL == "" {
+		return errors.New("postback_url is required")
+	}
+	if e.BotConfig.DebounceTime < 0 {
+		return errors.New("debounce_time must be >= 0")
+	}
+	if e.BotConfig.TextSegmentationEnabled && e.BotConfig.TextSegmentationLimit <= 0 {
+		return errors.New("text_segmentation_limit must be > 0 when segmentation is enabled")
+	}
+	if e.BotConfig.DelayPerCharacter < 0 {
+		return errors.New("delay_per_character must be >= 0")
+	}
+	return nil
+}