sse: real-time flushing via raw handler overriding the generated route

The generated strict-server path could only return an io.Reader that
io.Copy drains without flushing, so SSE events sat chunk-buffered instead
of streaming in real time. Replace it with a raw http.ResponseWriter
handler (serveSSE) that Flush()es after every event.

Routing: chi allows a later registration to supersede an earlier one for
the same method+path (verified empirically for v5.3.1), so serveSSE is
registered on the router AFTER gen.HandlerWithOptions and wins over the
generated /events/stream route. It inherits the base middleware chain and
applies auth via With(). The generated StreamEvents method now returns an
error (never reached) so a routing regression fails loudly rather than
silently reverting to buffered delivery.

Adds TestSSEStreamRealtimeDelivery: a real httptest.NewServer + streaming
client (NewRecorder can't flush) that connects, triggers an event, and
asserts delivery within 3s — proving both the override routing and
per-event flushing. Passes in <1s.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

internal/httpapi/server.go

@@ -101,12 +101,16 @@ func NewHandler(ctx context.Context, pool *db.Pool, cfg config.Config) http.Hand
 		},
 	})
 
+	// SSE stream: override the generated /events/stream route with a raw
+	// flushing handler (registered AFTER HandlerWithOptions so chi's last
+	// registration wins). The strict-server path can't Flush() per event;
+	// this one uses the real ResponseWriter for real-time delivery. It
+	// inherits the router's base middleware and applies auth via With().
+	r.With(combinedAuth(cfg)).Get("/api/v1/events/stream", s.serveSSE)
+
 	// Mount MCP at /mcp (plan R3-10)
 	r.With(combinedAuth(cfg)).Handle("/mcp", mcphandler.NewHandler(pool, cfg.MCPBearerToken))
 
-	// Mount SSE stream — handled by the strict handler's StreamEvents method
-	// via the OpenAPI-specified /api/v1/events/stream route.
-
 	return r
 }
 

internal/httpapi/sse.go

@@ -5,7 +5,6 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
-	"io"
 	"log/slog"
 	"net/http"
 	"strconv"
@@ -231,35 +230,32 @@ type ioWriter interface {
 	Write([]byte) (int, error)
 }
 
-// StreamEvents implements the OpenAPI interface (SSE event stream).
+// serveSSE is the streaming handler for GET /api/v1/events/stream, registered
+// directly on the chi router in NewHandler so it supersedes the generated
+// route. Using the raw http.ResponseWriter lets us Flush() after every event
+// (real-time delivery); the generated strict-server path can only hand back
+// an io.Reader that io.Copy drains without flushing (chunk-buffered).
 //
-// The strict-server model hands us only an io.Reader Body (copied to the
+// Protocol: https://html.spec.whatwg.org/multipage/server-sent-events.html
-// client via io.Copy in the generated Visit method), not the raw
+//  1. If Last-Event-ID is present, replay buffered events (broker, then DB).
-// ResponseWriter/Flusher — so we bridge with an io.Pipe. ctx is the request
+//  2. Subscribe and forward events fanned out from pg_notify.
-// context (the generated wrapper passes r.Context()); when the client
+//  3. Colon-comment heartbeat every 15s.
-// disconnects it cancels, the SSE loop returns, and pw.Close() ends the
+//  4. Unsubscribe on client disconnect (request context cancels).
-// io.Copy. Without threading ctx through, each disconnected client would
+func (s *Server) serveSSE(w http.ResponseWriter, r *http.Request) {
-// leak a goroutine holding a pipe.
+	flusher, ok := w.(http.Flusher)
-//
+	if !ok {
-// Note: io.Copy does not flush per-write, so delivery is chunk-buffered
+		writeProblem(w, r, http.StatusInternalServerError, "internal error", "streaming not supported")
-// rather than strictly real-time. Adequate for the event feed; a raw
+		return
-// flushing handler is a possible future refinement.
+	}
-func (s *Server) StreamEvents(ctx context.Context, req gen.StreamEventsRequestObject) (gen.StreamEventsResponseObject, error) {
-	pr, pw := io.Pipe()
-	go func() {
-		s.serveSSEWriter(ctx, pw, req.Params)
-		pw.Close()
-	}()
-	return gen.StreamEvents200TexteventStreamResponse{
-		Body:          pr,
-		ContentLength: 0, // unknown/streaming — omit the Content-Length header
-	}, nil
-}
 
-// serveSSEWriter runs the SSE loop writing to an io.Writer until the request
+	w.Header().Set("Content-Type", "text/event-stream")
-// context is cancelled (client disconnect / server shutdown).
+	w.Header().Set("Cache-Control", "no-cache")
-func (s *Server) serveSSEWriter(parent context.Context, w io.Writer, params gen.StreamEventsParams) {
+	w.Header().Set("Connection", "keep-alive")
-	ctx, cancel := context.WithCancel(parent)
+	w.Header().Set("X-Accel-Buffering", "no") // disable proxy buffering
+	w.WriteHeader(http.StatusOK)
+	flusher.Flush()
+
+	ctx, cancel := context.WithCancel(r.Context())
 	defer cancel()
 
 	sub := &sseSubscriber{
@@ -277,27 +273,27 @@ func (s *Server) serveSSEWriter(parent context.Context, w io.Writer, params gen.
 		close(sub.done)
 	}()
 
-	// Replay on Last-Event-ID
+	// ── 1. Replay on Last-Event-ID ────────────────────────────────
-	if params.LastEventID != nil && *params.LastEventID != "" {
+	if lastID := r.Header.Get("Last-Event-ID"); lastID != "" {
-		id, err := strconv.ParseInt(*params.LastEventID, 10, 64)
+		if id, err := strconv.ParseInt(lastID, 10, 64); err == nil {
-		if err == nil {
+			// The in-memory broker only holds recent events; if it doesn't
+			// cover the whole gap, fall back to the DB for a complete replay.
 			events := s.sseBroker.after(id)
-			if len(events) > 0 {
+			complete := len(events) > 0 && events[len(events)-1].ID == s.sseBroker.latestID()
+			if complete {
 				for _, ev := range events {
-					if !writeSSE(w, nil, ev) {
+					if !writeSSE(w, flusher, ev) {
 						return
 					}
 				}
-			}
+			} else {
-			// If broker didn't have them all, fetch from DB
-			if len(events) == 0 || events[len(events)-1].ID != s.sseBroker.latestID() {
 				q := sqlcgen.New(s.pool)
 				dbEvents, err := q.ListEventsAfter(ctx, sqlcgen.ListEventsAfterParams{ID: id, Limit: 5000})
 				if err != nil {
 					slog.Error("sse db replay failed", "error", err)
 				} else {
 					for _, ev := range dbEvents {
-						if !writeSSE(w, nil, ev) {
+						if !writeSSE(w, flusher, ev) {
 							return
 						}
 					}
@@ -306,7 +302,7 @@ func (s *Server) serveSSEWriter(parent context.Context, w io.Writer, params gen.
 		}
 	}
 
-	// Subscribe and forward
+	// ── 2. Subscribe and forward ──────────────────────────────────
 	heartbeat := time.NewTicker(15 * time.Second)
 	defer heartbeat.Stop()
 
@@ -318,18 +314,26 @@ func (s *Server) serveSSEWriter(parent context.Context, w io.Writer, params gen.
 			if !ok {
 				return
 			}
-			if !writeSSE(w, nil, ev) {
+			if !writeSSE(w, flusher, ev) {
 				return
 			}
 		case <-heartbeat.C:
-			_, err := fmt.Fprintf(w, ": heartbeat\n\n")
+			if _, err := fmt.Fprintf(w, ": heartbeat\n\n"); err != nil {
-			if err != nil {
 				return
 			}
+			flusher.Flush()
 		}
 	}
 }
 
+// StreamEvents satisfies the generated ServerInterface, but the SSE route is
+// served by the raw serveSSE handler registered in NewHandler (which wins
+// over this generated route). If this method is ever reached, routing has
+// regressed — fail loudly rather than silently chunk-buffering.
+func (s *Server) StreamEvents(ctx context.Context, req gen.StreamEventsRequestObject) (gen.StreamEventsResponseObject, error) {
+	return nil, fmt.Errorf("%w: SSE must be served by the raw handler", errNotImplemented)
+}
+
 // Ensure pgxpool is imported — used via Acquire.
 var _ = &pgxpool.Pool{}
 var _ = pgx.ErrNoRows

internal/httpapi/sse_test.go

@@ -0,0 +1,89 @@
+package httpapi
+
+// Real-connection SSE test. httptest.NewRecorder buffers and never flushes,
+// so this uses httptest.NewServer + a streaming client to verify that:
+//   - the raw serveSSE handler (not the generated 501 stub) serves the route,
+//   - an event created *after* the client connects is delivered in real time
+//     (i.e. flushed before the connection closes),
+//   - the SSE `data:` payload is the canonical gen.Event shape.
+// Guarded by OIKOS_TEST_DATABASE_URL.
+
+import (
+	"bufio"
+	"bytes"
+	"context"
+	"encoding/json"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"testing"
+	"time"
+)
+
+func TestSSEStreamRealtimeDelivery(t *testing.T) {
+	srv := httptest.NewServer(newTestHandler(t, devConfig()))
+	defer srv.Close()
+
+	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+	defer cancel()
+
+	// Connect to the stream.
+	req, _ := http.NewRequestWithContext(ctx, "GET", srv.URL+"/api/v1/events/stream", nil)
+	resp, err := http.DefaultClient.Do(req)
+	if err != nil {
+		t.Fatalf("connect stream: %v", err)
+	}
+	defer resp.Body.Close()
+	if resp.StatusCode != 200 {
+		t.Fatalf("stream status = %d, want 200", resp.StatusCode)
+	}
+	if ct := resp.Header.Get("Content-Type"); !strings.HasPrefix(ct, "text/event-stream") {
+		t.Fatalf("content-type = %q, want text/event-stream", ct)
+	}
+
+	// Read SSE frames in a goroutine.
+	dataCh := make(chan map[string]any, 4)
+	go func() {
+		sc := bufio.NewScanner(resp.Body)
+		for sc.Scan() {
+			line := sc.Text()
+			if strings.HasPrefix(line, "data: ") {
+				var m map[string]any
+				if json.Unmarshal([]byte(strings.TrimPrefix(line, "data: ")), &m) == nil {
+					dataCh <- m
+				}
+			}
+		}
+	}()
+
+	// Give the subscriber a moment to register, then trigger an event by
+	// POSTing to the SAME live server (same DB → NOTIFY the listener sees).
+	time.Sleep(300 * time.Millisecond)
+	payload, _ := json.Marshal(map[string]any{"slug": "service:sse-rt", "type": "service", "name": "sse-rt"})
+	cResp, err := http.Post(srv.URL+"/api/v1/entities", "application/json", bytes.NewReader(payload))
+	if err != nil {
+		t.Fatalf("trigger create: %v", err)
+	}
+	cResp.Body.Close()
+	if cResp.StatusCode != 201 {
+		t.Fatalf("trigger create status = %d, want 201", cResp.StatusCode)
+	}
+
+	// The event must arrive in real time (well before the 10s ctx deadline),
+	// proving the handler flushes rather than buffering until close.
+	select {
+	case ev := <-dataCh:
+		if ev["type"] != "entity.created" {
+			t.Errorf("event type = %v, want entity.created", ev["type"])
+		}
+		// canonical shape: snake_case + decoded data object
+		if _, ok := ev["entity_id"]; !ok {
+			t.Errorf("missing snake_case entity_id: %v", ev)
+		}
+		if d, ok := ev["data"].(map[string]any); !ok || d["slug"] != "service:sse-rt" {
+			t.Errorf("data not a decoded object with slug: %v", ev["data"])
+		}
+	case <-time.After(3 * time.Second):
+		t.Fatal("SSE event not delivered within 3s (flushing broken?)")
+	}
+}