fix(growth): harden AI jobs and cost controls

internal/api/growth.go

@@ -131,6 +131,17 @@ func (s *Server) growthReportEvents(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 	merchantID := sessionFrom(r.Context()).MerchantID
+	slots := s.growthSSESlots
+	if slots == nil {
+		slots = defaultGrowthSSESlots
+	}
+	select {
+	case slots <- struct{}{}:
+		defer func() { <-slots }()
+	default:
+		s.problem(w, r, http.StatusTooManyRequests, "RATE_LIMITED", "Too many growth report streams")
+		return
+	}
 	record, err := s.growth.Get(r.Context(), merchantID, id)
 	if errors.Is(err, growthreport.ErrNotFound) {
 		s.problem(w, r, 404, "RESOURCE_NOT_FOUND", "Resource not found")
@@ -148,7 +159,17 @@ func (s *Server) growthReportEvents(w http.ResponseWriter, r *http.Request) {
 	w.Header().Set("Content-Type", "text/event-stream; charset=utf-8")
 	w.Header().Set("Cache-Control", "no-cache, no-transform")
 	w.Header().Set("X-Accel-Buffering", "no")
-	ticker := time.NewTicker(2 * time.Second)
+	maxDuration := s.growthSSEMaxDuration
+	if maxDuration <= 0 {
+		maxDuration = defaultGrowthSSEMaxDuration
+	}
+	pollInterval := s.growthSSEPollInterval
+	if pollInterval <= 0 {
+		pollInterval = defaultGrowthSSEPollInterval
+	}
+	streamContext, cancel := context.WithTimeout(r.Context(), maxDuration)
+	defer cancel()
+	ticker := time.NewTicker(pollInterval)
 	defer ticker.Stop()
 	heartbeat := time.NewTicker(15 * time.Second)
 	defer heartbeat.Stop()
@@ -165,10 +186,10 @@ func (s *Server) growthReportEvents(w http.ResponseWriter, r *http.Request) {
 			return
 		}
 		select {
-		case <-r.Context().Done():
+		case <-streamContext.Done():
 			return
 		case <-ticker.C:
-			record, err = s.growth.Get(r.Context(), merchantID, id)
+			record, err = s.growth.Get(streamContext, merchantID, id)
 			if err != nil {
 				return
 			}

internal/api/growth_test.go

@@ -127,3 +127,45 @@ func TestGrowthSSEAuthorizesBeforeSettingStreamHeaders(t *testing.T) {
 		t.Fatalf("status=%d content-type=%q", response.Code, response.Header().Get("Content-Type"))
 	}
 }
+
+func TestGrowthSSERejectsWhenGlobalCapacityIsFull(t *testing.T) {
+	t.Parallel()
+	slots := make(chan struct{}, 1)
+	slots <- struct{}{}
+	reads := 0
+	service := fakeGrowthService{get: func(int64, string) (growthreport.Record, error) {
+		reads++
+		return growthreport.Record{Report: growthreport.Report{ReportID: "growth_0123456789abcdef01234567", Status: growthreport.StatusQueued}}, nil
+	}}
+	server := &Server{logger: slog.New(slog.NewTextHandler(io.Discard, nil)), growth: service, growthSSESlots: slots}
+	request := httptest.NewRequest("GET", "/", nil)
+	request.SetPathValue("reportId", "growth_0123456789abcdef01234567")
+	request = request.WithContext(context.WithValue(request.Context(), sessionKey{}, identity.Session{MerchantID: 7}))
+	response := httptest.NewRecorder()
+	server.growthReportEvents(response, request)
+	if response.Code != http.StatusTooManyRequests || reads != 0 {
+		t.Fatalf("status=%d database reads=%d", response.Code, reads)
+	}
+}
+
+func TestGrowthSSEEndsAtConfiguredDeadline(t *testing.T) {
+	t.Parallel()
+	reads := 0
+	service := fakeGrowthService{get: func(int64, string) (growthreport.Record, error) {
+		reads++
+		return growthreport.Record{Report: growthreport.Report{ReportID: "growth_0123456789abcdef01234567", Status: growthreport.StatusQueued, UpdatedAt: time.Unix(int64(reads), 0)}}, nil
+	}}
+	server := &Server{logger: slog.New(slog.NewTextHandler(io.Discard, nil)), growth: service, growthSSESlots: make(chan struct{}, 1), growthSSEMaxDuration: 5 * time.Millisecond, growthSSEPollInterval: time.Millisecond}
+	request := httptest.NewRequest("GET", "/", nil)
+	request.SetPathValue("reportId", "growth_0123456789abcdef01234567")
+	request = request.WithContext(context.WithValue(request.Context(), sessionKey{}, identity.Session{MerchantID: 7}))
+	response := httptest.NewRecorder()
+	started := time.Now()
+	server.growthReportEvents(response, request)
+	if elapsed := time.Since(started); elapsed > 250*time.Millisecond {
+		t.Fatalf("stream remained open for %s", elapsed)
+	}
+	if reads < 1 || reads > 10 {
+		t.Fatalf("database reads=%d", reads)
+	}
+}

internal/api/server.go

@@ -23,22 +23,33 @@ import (
 
 const maxExportBytes = 50 << 20
 
+const (
+	defaultGrowthSSECapacity     = 64
+	defaultGrowthSSEMaxDuration  = 5 * time.Minute
+	defaultGrowthSSEPollInterval = 2 * time.Second
+)
+
+var defaultGrowthSSESlots = make(chan struct{}, defaultGrowthSSECapacity)
+
 var errExportBytesExceeded = errors.New("export exceeds byte limit")
 
 type Server struct {
-	identity       *identity.Service
-	analytics      *analytics.Service
-	growth         growthReportService
-	logger         *slog.Logger
-	frontendOrigin string
-	cookieSecure   bool
+	identity              *identity.Service
+	analytics             *analytics.Service
+	growth                growthReportService
+	logger                *slog.Logger
+	frontendOrigin        string
+	cookieSecure          bool
+	growthSSESlots        chan struct{}
+	growthSSEMaxDuration  time.Duration
+	growthSSEPollInterval time.Duration
 }
 
 type sessionKey struct{}
 type tokenKey struct{}
 
 func NewServer(identityService *identity.Service, analyticsService *analytics.Service, logger *slog.Logger, frontendOrigin string, cookieSecure bool, growthService ...growthReportService) *Server {
-	server := &Server{identity: identityService, analytics: analyticsService, logger: logger, frontendOrigin: frontendOrigin, cookieSecure: cookieSecure}
+	server := &Server{identity: identityService, analytics: analyticsService, logger: logger, frontendOrigin: frontendOrigin, cookieSecure: cookieSecure, growthSSESlots: defaultGrowthSSESlots, growthSSEMaxDuration: defaultGrowthSSEMaxDuration, growthSSEPollInterval: defaultGrowthSSEPollInterval}
 	if len(growthService) > 0 {
 		server.growth = growthService[0]
 	}