fathom/internal/web/events_test.go

package web

import (
	"bufio"
	"context"
	"encoding/json"
	"io"
	"log"
	"net/http"
	"net/http/httptest"
	"strings"
	"testing"
	"time"

	"git.z.gd/zzg/fathom/internal/ingest"
	"git.z.gd/zzg/fathom/internal/model"
	"git.z.gd/zzg/fathom/internal/store"
)

// liveServer wires a hub to a store the way cmd/fathom does, so the test
// exercises the real path from flush to browser rather than calling the hub
// directly.
func liveServer(t *testing.T) (*httptest.Server, *store.Store, *Hub) {
	t.Helper()
	hub := NewHub()
	st, err := store.Open(store.Options{
		Dir: t.TempDir(), FlushInterval: time.Hour, FlushRows: 1 << 20,
		OnFlush: hub.NotifyFlush,
	})
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { st.Close() })

	srv := NewServer(st, ingest.NewDecoder(), hub, log.New(io.Discard, "", 0))
	ts := httptest.NewServer(srv.Handler())
	t.Cleanup(ts.Close)
	return ts, st, hub
}

// readEvent reads SSE frames until it finds a data line, or times out.
func readEvent(t *testing.T, r *bufio.Reader) Event {
	t.Helper()
	deadline := time.Now().Add(5 * time.Second)
	for time.Now().Before(deadline) {
		line, err := r.ReadString('\n')
		if err != nil {
			t.Fatalf("reading event stream: %v", err)
		}
		line = strings.TrimSpace(line)
		if !strings.HasPrefix(line, "data: ") {
			continue // comment, retry directive, or blank separator
		}
		var ev Event
		if err := json.Unmarshal([]byte(strings.TrimPrefix(line, "data: ")), &ev); err != nil {
			t.Fatalf("event is not JSON: %v", err)
		}
		return ev
	}
	t.Fatal("timed out waiting for an event")
	return Event{}
}

func TestEventStreamAnnouncesFlush(t *testing.T) {
	ts, st, hub := liveServer(t)

	res, err := http.Get(ts.URL + "/api/events")
	if err != nil {
		t.Fatal(err)
	}
	defer res.Body.Close()
	if ct := res.Header.Get("Content-Type"); ct != "text/event-stream" {
		t.Fatalf("Content-Type = %q, want text/event-stream", ct)
	}

	r := bufio.NewReader(res.Body)

	// The stream opens with a hello so the UI can show "connected" immediately
	// rather than only after the first trace arrives.
	if ev := readEvent(t, r); ev.Type != "hello" {
		t.Fatalf("first event is %q, want hello", ev.Type)
	}

	// Wait for the handler to register before flushing, or the event is
	// broadcast to nobody.
	waitFor(t, func() bool { return hub.Clients() == 1 })

	base := time.Date(2026, 9, 15, 12, 0, 0, 0, time.UTC).UnixMicro()
	if err := st.AddSpans([]model.Span{{
		TraceID: "t-live", SpanID: "s1", Name: "x",
		StartTS: base, EndTS: base + 1000, SampleRate: 1,
	}}); err != nil {
		t.Fatal(err)
	}
	if err := st.Flush(); err != nil {
		t.Fatal(err)
	}

	ev := readEvent(t, r)
	if ev.Type != "flush" {
		t.Fatalf("event type = %q, want flush", ev.Type)
	}
	if ev.Spans != 1 {
		t.Errorf("event reports %d spans, want 1", ev.Spans)
	}
	// The trace id is what lets the detail view refetch only when the trace on
	// screen actually changed.
	if len(ev.TraceIDs) != 1 || ev.TraceIDs[0] != "t-live" {
		t.Errorf("trace_ids = %v, want [t-live]", ev.TraceIDs)
	}
	if ev.Seq == 0 {
		t.Error("event has no sequence number")
	}
}

// TestNoEventForEmptyFlush guards the quiet case: an idle server calls Flush on
// a timer, and a notification per tick would make the UI refetch forever for
// nothing.
func TestNoEventForEmptyFlush(t *testing.T) {
	var got int
	st, err := store.Open(store.Options{
		Dir: t.TempDir(), FlushInterval: time.Hour, FlushRows: 1 << 20,
		OnFlush: func(store.FlushInfo) { got++ },
	})
	if err != nil {
		t.Fatal(err)
	}
	defer st.Close()

	for i := 0; i < 5; i++ {
		if err := st.Flush(); err != nil {
			t.Fatal(err)
		}
	}
	if got != 0 {
		t.Errorf("empty flushes produced %d notifications, want 0", got)
	}
}

// TestSlowClientDoesNotBlockIngest is the property that keeps a stalled browser
// tab from becoming a stalled ingest path.
func TestSlowClientDoesNotBlockIngest(t *testing.T) {
	hub := NewHub()
	stuck := make(chan []byte) // unbuffered and never read
	hub.add(stuck)

	done := make(chan struct{})
	go func() {
		for i := 0; i < 100; i++ {
			hub.NotifyFlush(store.FlushInfo{Spans: 1, TraceIDs: []string{"t"}})
		}
		close(done)
	}()

	select {
	case <-done:
	case <-time.After(5 * time.Second):
		t.Fatal("a client that never reads blocked the flush path")
	}
}

func TestClientDisconnectIsCleanedUp(t *testing.T) {
	ts, _, hub := liveServer(t)

	req, _ := http.NewRequest("GET", ts.URL+"/api/events", nil)
	ctx, cancel := context.WithCancel(context.Background())
	res, err := http.DefaultClient.Do(req.WithContext(ctx))
	if err != nil {
		t.Fatal(err)
	}
	waitFor(t, func() bool { return hub.Clients() == 1 })

	cancel()
	res.Body.Close()

	waitFor(t, func() bool { return hub.Clients() == 0 })
}

func waitFor(t *testing.T, cond func() bool) {
	t.Helper()
	deadline := time.Now().Add(5 * time.Second)
	for time.Now().Before(deadline) {
		if cond() {
			return
		}
		time.Sleep(10 * time.Millisecond)
	}
	t.Fatal("condition never became true")
}