feat(CI): fix CI

go.mod

@@ -4,7 +4,7 @@ go 1.25.0
 
 require (
 	github.com/ThreeDotsLabs/watermill v1.5.2
-	github.com/go-chi/chi/v5 v5.2.5
+	github.com/go-chi/chi/v5 v5.3.0
 	github.com/go-chi/render v1.0.3
 	github.com/pkg/errors v0.9.1
 	github.com/stretchr/testify v1.11.1

go.sum

@@ -7,6 +7,8 @@ github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c
 github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/go-chi/chi/v5 v5.2.5 h1:Eg4myHZBjyvJmAFjFvWgrqDTXFyOzjj7YIm3L3mu6Ug=
 github.com/go-chi/chi/v5 v5.2.5/go.mod h1:X7Gx4mteadT3eDOMTsXzmI4/rwUpOwBHLpAfupzFJP0=
+github.com/go-chi/chi/v5 v5.3.0 h1:halUjDxhshgXHMrao5bB8eNBXo/rnzwr8m5m36glehM=
+github.com/go-chi/chi/v5 v5.3.0/go.mod h1:R+tYY2hNuVUUjxoPtqUdgBqevM9s9njzkTLutVsOCto=
 github.com/go-chi/render v1.0.3 h1:AsXqd2a1/INaIfUSKq3G5uA8weYx20FOsM7uSoCyyt4=
 github.com/go-chi/render v1.0.3/go.mod h1:/gr3hVkmYR0YlEy3LxCuVRFzEu9Ruok+gFqbIofjao0=
 github.com/google/uuid v1.2.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=

pkg/http/sse.go

@@ -3,6 +3,7 @@ package http
 import (
 	"context"
 	"net/http"
+	"sync"
 
 	"github.com/go-chi/render"
 	"github.com/pkg/errors"
@@ -169,13 +170,25 @@ func (h sseHandler) handleEventStream(w http.ResponseWriter, r *http.Request) {
 
 	responsesChan := make(chan interface{})
 
+	// The producer goroutine reads from r (including chi.URLParam via the
+	// StreamAdapter). chi pools and resets the request's *Context once
+	// ServeHTTP returns, so the goroutine must not outlive this handler.
+	// wg.Wait below keeps ServeHTTP blocked until the goroutine exits.
+	var wg sync.WaitGroup
+	wg.Add(1)
+
 	go func() {
+		defer wg.Done()
 		defer func() {
 			h.logger.Trace("Closing SSE handler", nil)
 			close(responsesChan)
 		}()
 
-		responsesChan <- response
+		select {
+		case responsesChan <- response:
+		case <-r.Context().Done():
+			return
+		}
 
 		h.logger.Trace("Listening for messages", nil)
 
@@ -189,16 +202,15 @@ func (h sseHandler) handleEventStream(w http.ResponseWriter, r *http.Request) {
 				msg.Ack()
 
 				nextResponse, ok := h.streamAdapter.NextStreamResponse(r, msg)
+				if !ok {
+					continue
+				}
 
+				h.logger.Trace("Stream responding on message", watermill.LogFields{"uuid": msg.UUID})
 				select {
+				case responsesChan <- nextResponse:
 				case <-r.Context().Done():
 					return
-				default:
-				}
-
-				if ok {
-					h.logger.Trace("Stream responding on message", watermill.LogFields{"uuid": msg.UUID})
-					responsesChan <- nextResponse
 				}
 			case <-r.Context().Done():
 				return
@@ -210,4 +222,8 @@ func (h sseHandler) handleEventStream(w http.ResponseWriter, r *http.Request) {
 		marshaler: h.config.Marshaler,
 	}
 	responder.Respond(w, r, responsesChan)
+
+	// Block ServeHTTP until the producer goroutine has fully exited, so chi
+	// never resets the request context while the goroutine still reads r.
+	wg.Wait()
 }