package web
import (
"context"
"net/http"
"net/http/httptest"
"runtime"
"strings"
"sync"
"testing"
"time"
"craftdesign.group/quarry/internal/api"
"craftdesign.group/quarry/internal/ci"
"craftdesign.group/quarry/internal/ops"
)
// firstFlush is a recorder that says when the stream has started, so
// the test cancels a stream that is actually running rather than one
// still passing its gate.
type firstFlush struct {
*httptest.ResponseRecorder
once sync.Once
started chan struct{}
}
func (f *firstFlush) Flush() {
f.ResponseRecorder.Flush()
f.once.Do(func() { close(f.started) })
}
// Every goroutine a live stream starts must end when its request does.
// A relay that loops on a ticker or a watch channel alone never sees
// the reader leave, so each closed tab used to leave one behind for the
// life of the process.
func TestAClosedTabEndsItsRelay(t *testing.T) {
ix := landingIndex(t, false)
run := api.CIRun{RunID: "run-5", Checkin: "c1", Workflow: "w",
Pipeline: "p", Status: "running", Started: "2026-08-29T01:00:00Z"}
eng := &countingEngine{first: ci.Run{RunID: "run-5", Status: "running",
Steps: map[string]ci.Step{"build": {Status: "running"}}}}
build := func(t *testing.T, changes Changes) http.Handler {
t.Helper()
r := ops.NewRegistry()
deps := api.Deps{Version: "v",
Repos: api.NewRepos(map[string]*api.Repo{"alpha": {Index: ix}}),
CI: fakeCI{runs: []api.CIRun{run}}}
if err := api.RegisterAll(r, deps); err != nil {
t.Fatal(err)
}
opts := []Option{WithSessions(fakeSessions{user: "dan"}), WithCaps(stubCaps{letters: "i8"}), WithEngine(eng)}
if changes != nil {
opts = append(opts, WithChanges(changes))
}
return HandlerWith(r, &fakeNotifier{ch: make(chan struct{})}, nil, opts)
}
watched := func() Changes { return &fakeChanges{ch: make(chan struct{})} }
cases := []struct {
name string
path string
changes func() Changes
}{
// The run page relays a ticker and the ci watch.
{"run stream", "/r/alpha/ci/run.sse/run-5", watched},
// The landing card relays a ticker and the ci watch.
{"landing stream", "/r/alpha/landing.sse/" + strings.Repeat("a1", 20), watched},
// The forum thread relays the forum watch.
{"forum stream", "/r/alpha/forum.sse/" + strings.Repeat("a1", 20), watched},
// With no event plane the CI table relays its own ticker.
{"ci table fallback", "/r/alpha/ci.sse", func() Changes { return nil }},
}
for _, tc := range cases {
t.Run(tc.name, func(t *testing.T) {
h := build(t, tc.changes())
baseline := runtime.NumGoroutine()
ctx, cancel := context.WithCancel(context.Background())
req := httptest.NewRequest("GET", tc.path, nil).WithContext(ctx)
req.AddCookie(&http.Cookie{Name: sessionCookie, Value: "good"})
rec := &firstFlush{ResponseRecorder: httptest.NewRecorder(), started: make(chan struct{})}
done := make(chan struct{})
go func() {
defer close(done)
h.ServeHTTP(rec, req)
}()
select {
case <-rec.started:
case <-done:
t.Fatalf("FAIL: %s never streamed: %d %s", tc.path, rec.Code, rec.Body)
case <-time.After(3 * time.Second):
cancel()
t.Fatalf("FAIL: %s never flushed", tc.path)
}
// Let the handler reach its relays before the reader leaves.
time.Sleep(50 * time.Millisecond)
cancel()
select {
case <-done:
case <-time.After(3 * time.Second):
t.Fatalf("FAIL: %s handler did not return after its request ended", tc.path)
}
deadline := time.Now().Add(2 * time.Second)
left := runtime.NumGoroutine()
for left > baseline && time.Now().Before(deadline) {
time.Sleep(10 * time.Millisecond)
left = runtime.NumGoroutine()
}
if left > baseline {
t.Fatalf("FAIL: %s left %d goroutines running after its request ended", tc.path, left-baseline)
}
t.Logf("ok: %s ends every goroutine it started", tc.path)
})
}
}