textmachine/platform/internal/httpapi/stream_test.go

package httpapi

import (
	"encoding/json"
	"net/http"
	"net/http/httptest"
	"strings"
	"testing"

	"textmachine/platform/internal/pgstore"
)

// stream_test.go: the book's event stream.
//
// Everything pinned here is a rule a client's correctness stands on and none of it is observable
// from a single frame: that the sequence a resting book produces is FINITE, that a `note` is never
// dropped, and that a client which cannot be continued is told so instead of being silently started
// from now.

func streamOf(t *testing.T, lib *fakeLibrary, lastEventID string) *httptest.ResponseRecorder {
	t.Helper()
	h := readingServer(t, lib)
	r := httptest.NewRequest("GET", "/v0/books/bk_1/events", nil)
	r.Header.Set("Authorization", "Bearer token")
	if lastEventID != "" {
		r.Header.Set("Last-Event-ID", lastEventID)
	}
	w := httptest.NewRecorder()
	h.ServeHTTP(w, r)
	return w
}

// frames splits an SSE body into (event, id, data) triples.
func frames(t *testing.T, body string) []struct{ Event, ID, Data string } {
	t.Helper()
	var out []struct{ Event, ID, Data string }
	for _, block := range strings.Split(body, "\n\n") {
		var f struct{ Event, ID, Data string }
		for _, line := range strings.Split(block, "\n") {
			switch {
			case strings.HasPrefix(line, "event: "):
				f.Event = strings.TrimPrefix(line, "event: ")
			case strings.HasPrefix(line, "id: "):
				f.ID = strings.TrimPrefix(line, "id: ")
			case strings.HasPrefix(line, "data: "):
				f.Data = strings.TrimPrefix(line, "data: ")
			}
		}
		if f.Event != "" {
			out = append(out, f)
		}
	}
	return out
}

// A book at rest must not become an endless open-and-close. The sequence is bounded BY THE ID on
// `hello`: `0` on a book with no history is what the client has to present on the reconnect a
// browser makes by itself, and that reconnect is the one answered 204.
//
// Mutation caught: omitting the id from `hello`; sending `end` without one; answering 200 to the
// reconnect of a resting book.
func TestARestingBookProducesAFiniteSequence(t *testing.T) {
	lib := &fakeLibrary{stream: pgstore.StreamState{Position: 0, Revision: 5, StructureVersion: 3, AtRest: true}}
	w := streamOf(t, lib, "")
	if w.Code != http.StatusOK {
		t.Fatalf("status %d", w.Code)
	}
	got := frames(t, w.Body.String())
	if len(got) != 2 || got[0].Event != "hello" || got[1].Event != "end" {
		t.Fatalf("a resting book produced %+v", got)
	}
	for _, f := range got {
		if f.ID != "0" {
			t.Errorf("%s carries id %q, want 0 for a book with no history", f.Event, f.ID)
		}
	}
	var hello map[string]any
	if err := json.Unmarshal([]byte(got[0].Data), &hello); err != nil {
		t.Fatal(err)
	}
	if hello["contract"] != ContractVersion || hello["structure_version"] != float64(3) {
		t.Errorf("hello: %v", hello)
	}
	// And the reconnect the browser makes with that id is what stops it.
	if again := streamOf(t, lib, "0"); again.Code != http.StatusNoContent {
		t.Fatalf("the reconnect of a resting book answered %d, want 204", again.Code)
	}
}

// A state frame MAY be replaced by a later one of its kind; `note` is an ADDITION and MUST NOT be
// coalesced or dropped — a lost one is lost silently and forever.
//
// Mutation caught: coalescing by event name without exempting `note`; sending every frame.
func TestStateFramesCoalesceAndNotesNeverDo(t *testing.T) {
	lib := &fakeLibrary{
		stream: pgstore.StreamState{Position: 5, Revision: 9, StructureVersion: 1, AtRest: true},
		frames: []pgstore.Frame{
			{Position: 1, Event: pgstore.FrameProgress, Data: json.RawMessage(`{"progress":{"done":1}}`)},
			{Position: 2, Event: pgstore.FrameNote, Data: json.RawMessage(`{"note":{"id":"nt_1"}}`)},
			{Position: 3, Event: pgstore.FrameProgress, Data: json.RawMessage(`{"progress":{"done":2}}`)},
			{Position: 4, Event: pgstore.FrameNote, Data: json.RawMessage(`{"note":{"id":"nt_2"}}`)},
			{Position: 5, Event: pgstore.FrameProgress, Data: json.RawMessage(`{"progress":{"done":3}}`)},
		},
	}
	got := frames(t, streamOf(t, lib, "1").Body.String())
	var notes, progress int
	for _, f := range got {
		switch f.Event {
		case "note":
			notes++
		case "progress":
			progress++
		}
	}
	if notes != 2 {
		t.Errorf("%d note frames survived coalescing, want both", notes)
	}
	if progress != 1 {
		t.Errorf("%d progress frames, want the newest one only", progress)
	}
	// The gap that leaves is legal and declared: a client MUST NOT read a skipped number as a lost
	// frame. What must hold is that the ids are the BOOK's positions, not a per-connection count.
	for _, f := range got {
		if f.Event == "progress" && f.ID != "5" {
			t.Errorf("the surviving progress frame carries id %q, want the position it holds", f.ID)
		}
	}
}

// A client whose id predates the buffer cannot be continued. Replaying history beyond the buffer is
// forbidden — a one-off frame like a note would be lost silently — so it is told to re-read.
//
// Mutation caught: starting such a client from now without a resync frame.
func TestAClientTooFarBehindIsToldToResync(t *testing.T) {
	lib := &fakeLibrary{stream: pgstore.StreamState{Position: 900, Oldest: 800, Revision: 40, AtRest: true}}
	got := frames(t, streamOf(t, lib, "17").Body.String())
	if len(got) != 2 || got[1].Event != "resync_required" {
		t.Fatalf("frames: %+v", got)
	}
	if got[1].ID != "900" {
		t.Errorf("the connection frame carries id %q, want the last history frame", got[1].ID)
	}
}

// The BOUNDARY of that guard, which is where both of its wrong forms live. `last+1 < Oldest` is the
// question "is the frame this client needs next still here", and the two forms that read almost the
// same are `last < Oldest` — which re-syncs a client that is exactly caught up — and
// `last > 0 && last < Oldest`, which excuses id `0`, the id every connection to a fresh book carries.
//
// Mutation caught: either of those two. The far-behind case above answers all three identically.
func TestTheResyncBoundaryIsExact(t *testing.T) {
	for _, tc := range []struct {
		name       string
		state      pgstore.StreamState
		frames     []pgstore.Frame
		last       string
		wantResync bool
	}{
		{"one frame short of the buffer", pgstore.StreamState{Position: 801, Oldest: 800, AtRest: true}, nil, "798", true},
		{"exactly at the oldest frame", pgstore.StreamState{Position: 801, Oldest: 800, AtRest: true},
			[]pgstore.Frame{{Position: 800, Event: pgstore.FrameStatus, Data: json.RawMessage(`{}`)},
				{Position: 801, Event: pgstore.FrameStatus, Data: json.RawMessage(`{}`)}}, "799", false},
		{"inside the buffer", pgstore.StreamState{Position: 801, Oldest: 800, AtRest: true},
			[]pgstore.Frame{{Position: 801, Event: pgstore.FrameStatus, Data: json.RawMessage(`{}`)}}, "800", false},
		// `0` is a legitimate id and not "no id": it is what `hello` carries on a book with no history.
		{"the legitimate zero", pgstore.StreamState{Position: 2, Oldest: 1, AtRest: true},
			[]pgstore.Frame{{Position: 1, Event: pgstore.FrameStatus, Data: json.RawMessage(`{}`)},
				{Position: 2, Event: pgstore.FrameStatus, Data: json.RawMessage(`{}`)}}, "0", false},
	} {
		t.Run(tc.name, func(t *testing.T) {
			body := streamOf(t, &fakeLibrary{stream: tc.state, frames: tc.frames}, tc.last).Body.String()
			if got := strings.Contains(body, "event: resync_required"); got != tc.wantResync {
				t.Errorf("Last-Event-ID %s against Oldest %d re-synced=%v, want %v: %s",
					tc.last, tc.state.Oldest, got, tc.wantResync, body)
			}
		})
	}
}

// A prune that lands BEFORE the read shows as a hole at the HEAD of the batch rather than as an
// Oldest past the watermark — the read returns the frames after the hole and the watermark jumps
// over it, so the second test can never fire.
//
// Mutation caught: comparing only `state.Oldest > from+1` after advancing, which is the form that
// shipped: every frame in the hole, `note` included, was skipped in silence.
func TestAHoleAtTheHeadOfTheBatchAsksTheClientToResync(t *testing.T) {
	lib := &fakeLibrary{
		stream: pgstore.StreamState{Position: 100, Oldest: 1, Revision: 11},
		frames: []pgstore.Frame{{Position: 289, Event: pgstore.FrameNote, Data: json.RawMessage(`{"note":{"id":"nt_1"}}`)}},
	}
	// While the client was not reading, the writers minted 101..289 and the buffer pruned past them.
	lib.onSecondRead = func(s *pgstore.StreamState) { s.Oldest, s.Position, s.AtRest = 289, 289, true }
	body := streamOf(t, lib, "100").Body.String()
	if !strings.Contains(body, "event: resync_required") {
		t.Errorf("a hole at the head of the batch was skipped silently: %s", body)
	}
}

// HEAD reaches this handler too, and a HEAD has no body to stream: it answers the headers a GET
// would and ends, rather than running the pump for a request nobody can read (RFC 9110 §9.3.2).
//
// Mutation caught: dropping the short-circuit — the handler then streams and holds a goroutine.
func TestHeadOnTheStreamAnswersTheHeadersAndNothingElse(t *testing.T) {
	lib := &fakeLibrary{stream: pgstore.StreamState{Position: 4, Revision: 11, AtRest: true}}
	r := httptest.NewRequest("HEAD", "/v0/books/bk_1/events", nil)
	r.Header.Set("Authorization", "Bearer token")
	w := httptest.NewRecorder()
	readingServer(t, lib).ServeHTTP(w, r)
	if w.Code != http.StatusOK || w.Header().Get("Content-Type") != "text/event-stream" {
		t.Fatalf("HEAD answered %d %q", w.Code, w.Header().Get("Content-Type"))
	}
	if w.Body.Len() != 0 {
		t.Errorf("HEAD carried a body of %d bytes", w.Body.Len())
	}
}

// The book was cut again: every pair anchor and every cursor the client holds is invalid, and a
// delta cannot express what happened.
func TestAReCutEndsTheStreamWithAResync(t *testing.T) {
	lib := &fakeLibrary{stream: pgstore.StreamState{Position: 3, Revision: 10, StructureVersion: 3}}
	// The second read of the state — the one the pump makes after sending — sees the new version.
	lib.onSecondRead = func(s *pgstore.StreamState) { s.StructureVersion = 4 }
	got := frames(t, streamOf(t, lib, "").Body.String())
	if len(got) != 2 || got[1].Event != "resync_required" {
		t.Fatalf("frames: %+v", got)
	}
}

// A request WITHOUT Last-Event-ID always opens a new stream, even on a resting book: otherwise a
// client could not begin watching again after a run starts.
func TestAFreshRequestAlwaysOpensAStream(t *testing.T) {
	lib := &fakeLibrary{stream: pgstore.StreamState{Position: 42, Revision: 9, AtRest: true}}
	w := streamOf(t, lib, "")
	if w.Code != http.StatusOK {
		t.Fatalf("status %d, want a stream", w.Code)
	}
	if got := frames(t, w.Body.String()); got[0].Event != "hello" || got[0].ID != "42" {
		t.Fatalf("frames: %+v", got)
	}
}

// The pump's two gap tests, at the width a live connection actually produces. `emitFrame` mints one
// frame per write, so an ordinary hole is exactly ONE frame wide; the pin for PD-283 runs a hole of
// 189 frames, and both boundaries answer a hole that size identically — a shift of either by one is
// invisible to it.
//
// Mutation caught: `>=` for `>` in either check, or `from+2` for `from+1`.
func TestTheGapBoundariesAreExactAtOneFrame(t *testing.T) {
	frame := func(pos int64) pgstore.Frame {
		return pgstore.Frame{Position: pos, Event: pgstore.FrameNote,
			Data: json.RawMessage(`{"note":{"id":"nt_1"}}`)}
	}
	for _, tc := range []struct {
		name string
		// first is where the catch-up read after `Last-Event-ID: 100` begins; oldest is what the
		// buffer's floor becomes by the state read that follows it.
		first      int64
		oldest     int64
		wantResync bool
	}{
		{"the very next frame is there", 101, 1, false},
		{"exactly one frame missing at the head", 102, 1, true},
		{"the buffer still holds the next frame", 101, 102, false},
		{"the buffer pruned exactly that one frame", 101, 103, true},
	} {
		t.Run(tc.name, func(t *testing.T) {
			lib := &fakeLibrary{
				stream: pgstore.StreamState{Position: 200, Oldest: 1, Revision: 11},
				frames: []pgstore.Frame{frame(tc.first)},
			}
			// The state as it stands once the batch has been consumed: at rest, so a stream with no
			// gap ends instead of polling.
			lib.onSecondRead = func(s *pgstore.StreamState) {
				s.Oldest, s.Position, s.AtRest = tc.oldest, tc.first, true
			}
			body := streamOf(t, lib, "100").Body.String()
			if got := strings.Contains(body, "event: resync_required"); got != tc.wantResync {
				t.Errorf("re-synced=%v, want %v: %s", got, tc.wantResync, body)
			}
		})
	}
}