Vetting/internal/httpserver/sse_e2e_test.go

package httpserver

import (
	"bufio"
	"context"
	"net/http"
	"net/http/httptest"
	"path/filepath"
	"strings"
	"testing"
	"time"

	"vetting/internal/api"
	"vetting/internal/db"
	"vetting/internal/events"
	"vetting/internal/model"
	"vetting/internal/orchestrator"
	"vetting/internal/store"
)

// TestSSE_EndToEnd is the canary test the unit-level render tests miss:
// it mounts the real chi router (with RealIP + Recoverer + Logger
// middleware, the same stack that runs in prod), opens a GET /events,
// publishes an event via Runner.PublishTileUpdate, and asserts the
// payload lands on the wire as a well-formed SSE message.
//
// Motivation: the user reported that after cdd6cae "the pipeline does
// not update without a manual refresh". Template-only tests can't catch
// that — they only check rendered HTML. This test fails if any layer
// between publish and browser (hub subscribe, writeSSE formatting,
// middleware ResponseWriter wrapping that strips Flusher) is broken.
func TestSSE_EndToEnd(t *testing.T) {
	conn, err := db.Open(filepath.Join(t.TempDir(), "vetting.db"))
	if err != nil {
		t.Fatalf("open db: %v", err)
	}
	t.Cleanup(func() { _ = conn.Close() })

	hosts := &store.Hosts{DB: conn}
	runs := &store.Runs{DB: conn}
	stages := &store.Stages{DB: conn}
	diffs := &store.SpecDiffs{DB: conn}
	arts := &store.Artifacts{DB: conn}
	hub := events.NewHub()
	runner := &orchestrator.Runner{Runs: runs, Hosts: hosts, Stages: stages, EventHub: hub}
	tiles := &api.TileEnricher{Runs: runs, Artifacts: arts, SpecDiffs: diffs}

	// The prod wiring: closures that render tile + pipeline fragments
	// for publishTileUpdate. Without these, publishTileUpdate is a no-op
	// except for the bare tile event.
	orchestrator.TileRenderer = func(_ context.Context, _ model.Host, _ *model.Run) string {
		return "<article>tile</article>"
	}
	orchestrator.PipelineRenderer = func(_ *model.Run, _ []model.Stage) string {
		return "<section>pipeline</section>"
	}

	ui := &api.UI{
		Hosts: hosts, Runs: runs, Stages: stages, SpecDiffs: diffs, Artifacts: arts,
		EventHub: hub, Runner: runner, Tiles: tiles,
	}
	agent := &api.Agent{
		Hosts: hosts, Runs: runs, Stages: stages, Artifacts: arts,
		SpecDiffs: diffs, Runner: runner, EventHub: hub,
	}

	hostID, err := hosts.Create(context.Background(), model.Host{
		Name: "sse-e2e", MAC: "aa:bb:cc:dd:ee:77",
		WoLBroadcastIP: "10.0.0.255", WoLPort: 9,
	})
	if err != nil {
		t.Fatalf("create host: %v", err)
	}

	router := NewRouter(Deps{UI: ui, Agent: agent})
	srv := httptest.NewServer(router)
	t.Cleanup(srv.Close)

	// Open the SSE stream.
	ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
	defer cancel()
	req, err := http.NewRequestWithContext(ctx, http.MethodGet, srv.URL+"/events", nil)
	if err != nil {
		t.Fatalf("new request: %v", err)
	}
	resp, err := http.DefaultClient.Do(req)
	if err != nil {
		t.Fatalf("GET /events: %v", err)
	}
	defer func() { _ = resp.Body.Close() }()

	if resp.StatusCode != http.StatusOK {
		t.Fatalf("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)
	}

	reader := bufio.NewReader(resp.Body)

	// Consume the hello preamble first.
	if err := waitForSSEEvent(reader, "hello", 1*time.Second); err != nil {
		t.Fatalf("hello preamble: %v", err)
	}

	// Give the Subscribe goroutine a moment to land in the map so our
	// publish doesn't beat it. This is the race the hub's select
	// default-drops on — if the subscriber isn't registered yet, the
	// publish is silently thrown away.
	time.Sleep(50 * time.Millisecond)

	// Publish a tile-update via Runner. This is what the real orchestrator
	// does on every state transition.
	runner.PublishTileUpdate(context.Background(), hostID)

	// The browser should see one SSE message with `event: tile-<hostID>`.
	wantName := "tile-1" // first host → ID 1
	_ = hostID
	if err := waitForSSEEvent(reader, wantName, 2*time.Second); err != nil {
		t.Fatalf("tile publish: %v", err)
	}
}

// waitForSSEEvent reads lines from the SSE stream until it sees
// `event: <name>` or the deadline elapses. It tolerates interleaved
// heartbeats because the hub's 15s heartbeat can fire between the
// hello and the event we're looking for.
func waitForSSEEvent(r *bufio.Reader, name string, timeout time.Duration) error {
	deadline := time.Now().Add(timeout)
	want := "event: " + name
	for time.Now().Before(deadline) {
		line, err := r.ReadString('\n')
		if err != nil {
			return err
		}
		if strings.TrimRight(line, "\r\n") == want {
			return nil
		}
	}
	return &timeoutErr{}
}

type timeoutErr struct{}

func (e *timeoutErr) Error() string { return "timeout waiting for sse event" }

// TestSSE_SubStepEvent confirms PublishSubStepUpdate lands on the wire
// with the exact "substep-{runID}-{stage}-{ordinal}" event name that
// detail-page swap targets key on. Without this, the template renders
// the right attribute but a middleware or renderer regression silently
// drops the payload.
func TestSSE_SubStepEvent(t *testing.T) {
	conn, err := db.Open(filepath.Join(t.TempDir(), "vetting.db"))
	if err != nil {
		t.Fatalf("open db: %v", err)
	}
	t.Cleanup(func() { _ = conn.Close() })

	hosts := &store.Hosts{DB: conn}
	runs := &store.Runs{DB: conn}
	stages := &store.Stages{DB: conn}
	diffs := &store.SpecDiffs{DB: conn}
	arts := &store.Artifacts{DB: conn}
	hub := events.NewHub()
	runner := &orchestrator.Runner{Runs: runs, Hosts: hosts, Stages: stages, EventHub: hub}
	tiles := &api.TileEnricher{Runs: runs, Artifacts: arts, SpecDiffs: diffs}

	orchestrator.SubStepRenderer = func(_ model.SubStep) string {
		return `<div class="substep">row</div>`
	}

	ui := &api.UI{
		Hosts: hosts, Runs: runs, Stages: stages, SpecDiffs: diffs, Artifacts: arts,
		EventHub: hub, Runner: runner, Tiles: tiles,
	}
	agent := &api.Agent{
		Hosts: hosts, Runs: runs, Stages: stages, Artifacts: arts,
		SpecDiffs: diffs, Runner: runner, EventHub: hub,
	}

	router := NewRouter(Deps{UI: ui, Agent: agent})
	srv := httptest.NewServer(router)
	t.Cleanup(srv.Close)

	ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
	defer cancel()
	req, _ := http.NewRequestWithContext(ctx, http.MethodGet, srv.URL+"/events", nil)
	resp, err := http.DefaultClient.Do(req)
	if err != nil {
		t.Fatalf("GET /events: %v", err)
	}
	defer func() { _ = resp.Body.Close() }()

	reader := bufio.NewReader(resp.Body)
	if err := waitForSSEEvent(reader, "hello", 1*time.Second); err != nil {
		t.Fatalf("hello preamble: %v", err)
	}
	time.Sleep(50 * time.Millisecond)

	runner.PublishSubStepUpdate(context.Background(), model.SubStep{
		RunID:     42,
		StageName: "CPUStress",
		Ordinal:   1,
		Name:      "Memory pass",
		State:     model.StagePassed,
	})

	if err := waitForSSEEvent(reader, "substep-42-CPUStress-1", 2*time.Second); err != nil {
		t.Fatalf("substep event: %v", err)
	}
}