feat(shutdown): add graceful shutdown to pipeline service and HTTP server

- Add Shutdown(ctx) to PipelineService interface; implementation cancels
  all in-flight pipeline contexts and waits for pollDebounceExpiry to exit
- Add stopCh/stoppedCh channels to pipelineService; pollDebounceExpiry
  now exits cleanly on close(stopCh) instead of running indefinitely
- Replace r.Run() in main with http.Server + signal.NotifyContext for
  SIGTERM/SIGINT with 10s drain budget

Author: gomessguii

cmd/server/main.go

@@ -2,7 +2,12 @@ package main
 
 import (
 	"context"
+	"errors"
 	"log"
+	"net/http"
+	"os/signal"
+	"syscall"
+	"time"
 
 	"github.com/gin-gonic/gin"
 	"github.com/go-redsync/redsync/v4"
@@ -63,9 +68,28 @@ func main() {
 	r.Use(gin.Recovery())
 	handler.RegisterRoutes(r)
 
-	// Step 10: start server
-	log.Printf("evo-bot-runtime starting on %s", cfg.ListenAddr)
-	if err := r.Run(cfg.ListenAddr); err != nil {
-		log.Fatalf("server failed: %v", err)
+	// Step 10: start server (non-blocking)
+	srv := &http.Server{Addr: cfg.ListenAddr, Handler: r}
+	go func() {
+		log.Printf("evo-bot-runtime starting on %s", cfg.ListenAddr)
+		if err := srv.ListenAndServe(); err != nil && !errors.Is(err, http.ErrServerClosed) {
+			log.Fatalf("server failed: %v", err)
+		}
+	}()
+
+	// Step 11: wait for SIGTERM or SIGINT
+	ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGTERM, syscall.SIGINT)
+	defer stop()
+	<-ctx.Done()
+	log.Printf("evo-bot-runtime shutting down")
+
+	// Step 12: graceful shutdown — 10s budget for in-flight HTTP requests
+	shutdownCtx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+	defer cancel()
+	if err := srv.Shutdown(shutdownCtx); err != nil {
+		log.Printf("HTTP server shutdown error: %v", err)
 	}
+
+	// Step 13: stop pipeline (cancel in-flight goroutines, drain poller)
+	pipeline.Shutdown(shutdownCtx)
 }

pkg/pipeline/handler/handler_test.go

@@ -56,6 +56,7 @@ type mockSvc struct{ processErr error }
 func (m *mockSvc) Process(_ context.Context, _ *model.MessageEvent) error { return m.processErr }
 func (m *mockSvc) Cancel(_, _ int64) error                                { return nil }
 func (m *mockSvc) Start() error                                           { return nil }
+func (m *mockSvc) Shutdown(_ context.Context)                             {}
 
 var _ pipelineService.PipelineService = (*mockSvc)(nil)
 

pkg/pipeline/service/pipeline_service.go

@@ -25,6 +25,9 @@ type PipelineService interface {
 	Process(ctx context.Context, event *model.MessageEvent) error
 	Cancel(contactID, conversationID int64) error
 	Start() error
+	// Shutdown stops the polling goroutine and cancels all in-flight pipeline
+	// contexts. It blocks until the poller exits or ctx is cancelled.
+	Shutdown(ctx context.Context)
 }
 
 type pipelineEntry struct {
@@ -39,7 +42,9 @@ type pipelineService struct {
 	debounce    debounceIface.DebounceEngine
 	aiAdapter   aiIface.AIAdapter
 	dispatchEng dispatchIface.DispatchEngine
-	entries     sync.Map // string → pipelineEntry
+	entries     sync.Map      // string → pipelineEntry
+	stopCh      chan struct{}  // closed by Shutdown to stop pollDebounceExpiry
+	stoppedCh   chan struct{}  // closed by pollDebounceExpiry when it exits
 }
 
 // NewPipelineService constructs the service. Returns interface (GEAR R03).
@@ -49,7 +54,14 @@ func NewPipelineService(
 	aiAdapter   aiIface.AIAdapter,
 	dispatchEng dispatchIface.DispatchEngine,
 ) PipelineService {
-	return &pipelineService{repo: repo, debounce: debounce, aiAdapter: aiAdapter, dispatchEng: dispatchEng}
+	return &pipelineService{
+		repo:        repo,
+		debounce:    debounce,
+		aiAdapter:   aiAdapter,
+		dispatchEng: dispatchEng,
+		stopCh:      make(chan struct{}),
+		stoppedCh:   make(chan struct{}),
+	}
 }
 
 // Start recovers in-progress debounce pairs from Redis, then launches the single
@@ -433,11 +445,19 @@ func (s *pipelineService) runDispatchStage(
 
 // pollDebounceExpiry is the single timer-detection mechanism (AC: #1).
 // It runs a 100ms ticker and advances expired StageDebounce pairs to StageAI.
+// It exits when s.stopCh is closed and signals s.stoppedCh before returning.
 func (s *pipelineService) pollDebounceExpiry() {
+	defer close(s.stoppedCh)
+
 	ticker := time.NewTicker(100 * time.Millisecond)
 	defer ticker.Stop()
 
-	for range ticker.C {
+	for {
+		select {
+		case <-s.stopCh:
+			return
+		case <-ticker.C:
+		}
 		s.entries.Range(func(k, v any) bool {
 			entry, ok := v.(pipelineEntry)
 			if !ok {
@@ -494,6 +514,27 @@ func (s *pipelineService) Cancel(contactID, conversationID int64) error {
 	return nil
 }
 
+// Shutdown stops the polling goroutine and cancels all in-flight pipeline
+// contexts. It blocks until the poller exits or ctx is cancelled.
+func (s *pipelineService) Shutdown(ctx context.Context) {
+	// Cancel all in-flight pipeline goroutines.
+	s.entries.Range(func(k, v any) bool {
+		if entry, ok := v.(pipelineEntry); ok {
+			entry.cancel()
+		}
+		return true
+	})
+
+	// Signal poller to stop and wait for it to exit.
+	close(s.stopCh)
+	select {
+	case <-s.stoppedCh:
+		slog.Info("pipeline.shutdown.complete")
+	case <-ctx.Done():
+		slog.Warn("pipeline.shutdown.timeout")
+	}
+}
+
 // recoverPipeline is deferred in every pipeline goroutine to handle panics safely.
 func (s *pipelineService) recoverPipeline(contactID, conversationID int64) {
 	if r := recover(); r != nil {