From d3189ac8f4240b0e935555402370951b0df02897 Mon Sep 17 00:00:00 2001 From: Amirhossein Khalili Date: Fri, 21 Aug 2026 01:50:11 +0330 Subject: [PATCH] fix(growth): bound SSE authorization reads --- internal/api/growth.go | 22 +++++++++++----------- internal/api/growth_test.go | 29 ++++++++++++++++++++++++++--- 2 files changed, 37 insertions(+), 14 deletions(-) diff --git a/internal/api/growth.go b/internal/api/growth.go index 4ab6e28..edac4d0 100644 --- a/internal/api/growth.go +++ b/internal/api/growth.go @@ -142,7 +142,17 @@ func (s *Server) growthReportEvents(w http.ResponseWriter, r *http.Request) { s.problem(w, r, http.StatusTooManyRequests, "RATE_LIMITED", "Too many growth report streams") return } - record, err := s.growth.Get(r.Context(), merchantID, id) + 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() + record, err := s.growth.Get(streamContext, merchantID, id) if errors.Is(err, growthreport.ErrNotFound) { s.problem(w, r, 404, "RESOURCE_NOT_FOUND", "Resource not found") return @@ -159,16 +169,6 @@ 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") - 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) diff --git a/internal/api/growth_test.go b/internal/api/growth_test.go index 0103e0f..0b9205a 100644 --- a/internal/api/growth_test.go +++ b/internal/api/growth_test.go @@ -54,14 +54,19 @@ func TestGrowthReportIDPatternRejectsTenantEnumerationInputs(t *testing.T) { } type fakeGrowthService struct { - get func(int64, string) (growthreport.Record, error) - create func() (growthreport.Record, bool, error) + get func(int64, string) (growthreport.Record, error) + getContext func(context.Context, int64, string) (growthreport.Record, error) + create func() (growthreport.Record, bool, error) } func (f fakeGrowthService) Create(_ context.Context, _, _ int64, _ growthreport.CreateRequest, _ string) (growthreport.Record, bool, error) { return f.create() } -func (f fakeGrowthService) Get(_ context.Context, merchantID int64, id string) (growthreport.Record, error) { + +func (f fakeGrowthService) Get(ctx context.Context, merchantID int64, id string) (growthreport.Record, error) { + if f.getContext != nil { + return f.getContext(ctx, merchantID, id) + } return f.get(merchantID, id) } func (f fakeGrowthService) Latest(_ context.Context, merchantID int64, terminal string) (growthreport.Record, error) { @@ -169,3 +174,21 @@ func TestGrowthSSEEndsAtConfiguredDeadline(t *testing.T) { t.Fatalf("database reads=%d", reads) } } + +func TestGrowthSSEBoundsInitialDatabaseRead(t *testing.T) { + t.Parallel() + service := fakeGrowthService{getContext: func(ctx context.Context, _ int64, _ string) (growthreport.Record, error) { + <-ctx.Done() + return growthreport.Record{}, ctx.Err() + }} + server := &Server{logger: slog.New(slog.NewTextHandler(io.Discard, nil)), growth: service, growthSSESlots: make(chan struct{}, 1), growthSSEMaxDuration: 5 * 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 || response.Code != http.StatusServiceUnavailable { + t.Fatalf("elapsed=%s status=%d", elapsed, response.Code) + } +}