package callstest

import (
	"encoding/json"
	"fmt"
	"net/http"
	"net/http/httptest"
	"net/url"
	"strings"
	"testing"
	"time"
)

// answer reads a refused stream: the status and the JSON body spec §4.5
// fixes the browser's behaviour on. A refusal never streams, so the
// in-process recorder is the right client for it.
func answer(t *testing.T, rec *httptest.ResponseRecorder) (peer, boot, reason string) {
	t.Helper()
	var body struct{ Peer, Boot, Reason string }
	if rec.Body.Len() > 0 {
		if err := json.Unmarshal(rec.Body.Bytes(), &body); err != nil {
			t.Fatalf("refusal body %q: %v", rec.Body.String(), err)
		}
	}
	return body.Peer, body.Boot, body.Reason
}

// Spec §4.1: the first event on a stream is hello, carrying the peer id
// the server issued and the boot id of this process. Spec §4.2: a
// session holds one peer per room, so a second fresh stream is refused
// with 409 naming the peer it already has.
func TestStreamSaysHelloWithPeerAndBootAndRefusesTheSecondFreshStream(t *testing.T) {
	x, sqid, _ := memberWithMeeting(t)
	c := x.browser(t)
	c.signIn("paul@example.com")
	s := c.stream("/m/" + sqid + "/stream")
	defer s.close()
	peer, boot := s.hello()
	if len(peer) != 32 || len(boot) != 32 {
		t.Fatalf("hello peer=%q boot=%q", peer, boot)
	}
	rec := c.getStream("/m/" + sqid + "/stream")
	if rec.Code != http.StatusConflict {
		t.Fatalf("second fresh stream: %d, want 409", rec.Code)
	}
	if named, _, _ := answer(t, rec); named != peer {
		t.Fatalf("409 names %q, want %q", named, peer)
	}
	// A second browser of the same person is a different session, so it
	// gets a peer of its own (§4.4 allows four streams per person).
	c2 := x.browser(t)
	c2.signIn("paul@example.com")
	s2 := c2.stream("/m/" + sqid + "/stream")
	defer s2.close()
	if peer2, boot2 := s2.hello(); peer2 == peer || boot2 != boot {
		t.Fatalf("second browser: peer %q boot %q", peer2, boot2)
	}
}

// Spec §2.5: a fresh stream is authorized by standing, a resume by the
// participation bound to this session. Neither ever answers 403.
func TestStreamNeedsStandingAndAResumeNeedsTheSession(t *testing.T) {
	x, sqid, tok := memberWithMeeting(t)
	stranger := x.browser(t)
	if rec := stranger.get("/m/" + sqid + "/stream"); rec.Code != http.StatusSeeOther {
		t.Fatalf("signed-out stream: %d, want the sign-in redirect", rec.Code)
	}
	c := x.browser(t)
	c.signIn("paul@example.com")
	// A second meeting, which the guest below is never given a link to.
	elsewhere, _ := createMeeting(t, c, "Not yours", "")
	s := c.stream("/m/" + sqid + "/stream")
	peer, boot := s.hello()
	s.close()

	guest := x.browser(t)
	guest.get("/j/" + tok)
	guest.signIn("guest@example.com")
	if rec := guest.getStream("/m/" + elsewhere + "/stream"); rec.Code != http.StatusNotFound {
		t.Fatalf("a meeting the guest has no standing for: %d, want 404", rec.Code)
	}
	rec := guest.getStream("/m/" + sqid + "/stream?peer=" + peer)
	if rec.Code != http.StatusNotFound {
		t.Fatalf("resume through another session: %d, want 404", rec.Code)
	}
	// The same answer an unknown peer gets, so nothing leaks (§4.2).
	if _, gotBoot, _ := answer(t, rec); gotBoot != boot {
		t.Fatalf("404 boot %q, want %q", gotBoot, boot)
	}
	unknown := guest.getStream("/m/" + sqid + "/stream?peer=0123456789abcdef0123456789abcdef")
	if unknown.Code != http.StatusNotFound || unknown.Body.String() != rec.Body.String() {
		t.Fatalf("unknown peer: %d %s", unknown.Code, unknown.Body.String())
	}
}

// Spec §4.5: a closed stream leaves the participation in the hub for
// thirty seconds, and a resume reattaches the same one — the same peer,
// the same boot. This participation never joined, so it is told only
// that it is waiting, which is all a peer outside the room may learn.
func TestResumeReattachesTheSameParticipation(t *testing.T) {
	x, sqid, _ := memberWithMeeting(t)
	c := x.browser(t)
	c.signIn("paul@example.com")
	s := c.stream("/m/" + sqid + "/stream")
	peer, boot := s.hello()
	s.close()

	s2 := c.stream("/m/" + sqid + "/stream?peer=" + peer)
	defer s2.close()
	p2, b2 := s2.hello()
	if p2 != peer || b2 != boot {
		t.Fatalf("resume hello %s %s, want %s %s", p2, b2, peer, boot)
	}
	ev, data, ok := s2.next(time.Second)
	if !ok || ev != "state" {
		t.Fatalf("after a resume: %s ok=%v", ev, ok)
	}
	var state struct{ Waiting bool }
	json.Unmarshal(data, &state)
	if !state.Waiting {
		t.Fatalf("state %s, want waiting", data)
	}
	// The resumed participation is still the session's one peer.
	if rec := c.getStream("/m/" + sqid + "/stream"); rec.Code != http.StatusConflict {
		t.Fatalf("fresh stream while resumed: %d, want 409", rec.Code)
	}
}

// Spec §4.1: the stream heartbeats, and dies at its TTL so hibernation
// is never held off by one. The browser resumes and loses nothing.
func TestTTLClosesTheStreamAndResumeWorksWithinTheWindow(t *testing.T) {
	x, sqid, _ := memberWithMeeting(t)
	x.ctl.SetStreamTimings(50*time.Millisecond, 300*time.Millisecond)
	c := x.browser(t)
	c.signIn("paul@example.com")
	s := c.stream("/m/" + sqid + "/stream")
	peer, _ := s.hello()
	if !s.waitHeartbeat(2 * time.Second) {
		t.Fatal("no heartbeat on an idle stream")
	}
	if !s.waitClosed(2 * time.Second) {
		t.Fatal("stream did not close at the TTL")
	}
	s2 := c.stream("/m/" + sqid + "/stream?peer=" + peer)
	defer s2.close()
	if p, _ := s2.hello(); p != peer {
		t.Fatalf("resume after the TTL: %s, want %s", p, peer)
	}
}

// Spec §4.5: a resume that misses its window is 404 with the same boot
// the browser last saw — the process did not restart, the window
// lapsed — and the browser's answer is a fresh stream, which gets a
// peer of its own because the old participation is gone.
func TestLapsedResumeIs404WithTheSameBootAndAFreshStreamGetsANewPeer(t *testing.T) {
	x, sqid, _ := memberWithMeeting(t)
	// The stream is left to die at its own TTL rather than closed from
	// this side: the handler detaches on its way out, before the
	// response is finished, so an EOF here means the resume window has
	// already been stamped — and the clock this test then moves is
	// ahead of it, not behind.
	x.ctl.SetStreamTimings(50*time.Millisecond, 150*time.Millisecond)
	c := x.browser(t)
	c.signIn("paul@example.com")
	s := c.stream("/m/" + sqid + "/stream")
	peer, boot := s.hello()
	if !s.waitClosed(2 * time.Second) {
		t.Fatal("stream did not close at the TTL")
	}
	x.advance(31 * time.Second)
	x.ctl.TickForTest()

	rec := c.getStream("/m/" + sqid + "/stream?peer=" + peer)
	_, gotBoot, _ := answer(t, rec)
	if rec.Code != http.StatusNotFound || gotBoot != boot {
		t.Fatalf("lapsed resume: %d boot %q, want 404 and %q", rec.Code, gotBoot, boot)
	}
	s3 := c.stream("/m/" + sqid + "/stream")
	defer s3.close()
	if p, b := s3.hello(); p == "" || p == peer || b != boot {
		t.Fatalf("fresh stream after the lapse: peer %q boot %q", p, b)
	}
}

// The same lapse, with no tick at all in it (Codex S2-2). §15.7 refuses
// the resume the moment the window passes; the browser's answer to a 404
// is a fresh stream, and until now that fresh stream was refused 409 by
// the very participation it had just been refused a resume of — so the
// client showed "This meeting is open in another tab" about a tab that
// did not exist, and the holding screen said the same. The tick is what
// used to clear it, up to fifteen seconds later.
func TestAFreshStreamAfterALapsedResumeSucceedsBeforeAnyTick(t *testing.T) {
	x, sqid, _ := memberWithMeeting(t)
	x.ctl.SetStreamTimings(50*time.Millisecond, 150*time.Millisecond)
	c := x.browser(t)
	c.signIn("paul@example.com")
	s := c.stream("/m/" + sqid + "/stream")
	peer, boot := s.hello()
	if !s.waitClosed(2 * time.Second) {
		t.Fatal("stream did not close at the TTL")
	}
	x.advance(31 * time.Second) // past the window, and no tick is run

	rec := c.getStream("/m/" + sqid + "/stream?peer=" + peer)
	if _, gotBoot, _ := answer(t, rec); rec.Code != http.StatusNotFound || gotBoot != boot {
		t.Fatalf("lapsed resume: %d boot %q, want 404 and %q", rec.Code, gotBoot, boot)
	}
	// The probe the client asks first says the same, and changes nothing.
	if probed := c.get("/m/" + sqid + "/stream?probe=1"); probed.Code != http.StatusNoContent {
		t.Fatalf("peerless probe after the lapse: %d, want 204", probed.Code)
	}
	if rec := c.get("/m/" + sqid + "/join"); rec.Code != http.StatusOK ||
		strings.Contains(rec.Body.String(), "already open in another tab") {
		t.Fatalf("holding screen after the lapse: %d, and it still says the meeting is open elsewhere", rec.Code)
	}
	s3 := c.stream("/m/" + sqid + "/stream")
	defer s3.close()
	if p, b := s3.hello(); p == "" || p == peer || b != boot {
		t.Fatalf("fresh stream after the lapse: peer %q boot %q", p, b)
	}
}

// The resume probe (§4.5, recorded for revision 9): EventSource hides
// the status of a refused stream from the page, so the browser asks
// first. The probe answers and does nothing else — it must not attach,
// must not clear the resume window and must not touch the ring.
func TestResumeProbeAnswersWithoutAttaching(t *testing.T) {
	x, sqid, _ := memberWithMeeting(t)
	c := x.browser(t)
	c.signIn("paul@example.com")

	// No Accept header: a probe attaches nothing, so ruling 15 exempts
	// it. Its answer is still never cached.
	rec0 := c.get("/m/" + sqid + "/stream?probe=1")
	if rec0.Code != http.StatusNoContent {
		t.Fatalf("probe with nothing open: %d, want 204", rec0.Code)
	}
	if cc := rec0.Header().Get("Cache-Control"); cc != "no-store" {
		t.Fatalf("probe Cache-Control %q, want no-store", cc)
	}
	s := c.stream("/m/" + sqid + "/stream")
	peer, boot := s.hello()
	rec := c.get("/m/" + sqid + "/stream?probe=1")
	named, _, _ := answer(t, rec)
	if rec.Code != http.StatusConflict || named != peer {
		t.Fatalf("probe for a fresh stream while one is open: %d %q", rec.Code, named)
	}
	s.close()

	probed := c.get("/m/" + sqid + "/stream?probe=1&peer=" + peer)
	if probed.Code != http.StatusNoContent {
		t.Fatalf("probe of a resumable peer: %d, want 204", probed.Code)
	}
	if cc := probed.Header().Get("Cache-Control"); cc != "no-store" {
		t.Fatalf("resume probe Cache-Control %q, want no-store", cc)
	}
	rec = c.get("/m/" + sqid + "/stream?probe=1&peer=0123456789abcdef0123456789abcdef")
	_, gotBoot, _ := answer(t, rec)
	if rec.Code != http.StatusNotFound || gotBoot != boot {
		t.Fatalf("probe of an unknown peer: %d boot %q", rec.Code, gotBoot)
	}
	// Probed twice and still resumable: the probe changed nothing.
	s2 := c.stream("/m/" + sqid + "/stream?peer=" + peer)
	defer s2.close()
	if p, _ := s2.hello(); p != peer {
		t.Fatalf("resume after two probes: %s, want %s", p, peer)
	}
}

// Spec §2.4 and §4.5: ending the meeting closes every stream with
// lost{ended}, and the peer id is remembered so a resume is told why.
// A meeting nobody streamed has no headstone in the hub at all, so
// ended_at is what refuses the stream (the Task 2 carry).
func TestEndingTheMeetingClosesTheStreamAndRefusesWhatComesAfter(t *testing.T) {
	x, sqid, _ := memberWithMeeting(t)
	c := x.browser(t)
	c.signIn("paul@example.com")
	s := c.stream("/m/" + sqid + "/stream")
	defer s.close()
	peer, boot := s.hello()

	if rec := c.post("/meetings/"+sqid+"/end", nil); rec.Code != http.StatusSeeOther {
		t.Fatalf("end meeting: %d", rec.Code)
	}
	ev, data, ok := s.next(2 * time.Second)
	if !ok || ev != "lost" {
		t.Fatalf("after the meeting ended: %s ok=%v", ev, ok)
	}
	var lost struct{ Reason string }
	json.Unmarshal(data, &lost)
	if lost.Reason != "ended" {
		t.Fatalf("lost %s, want reason ended", data)
	}
	if !s.waitClosed(2 * time.Second) {
		t.Fatal("the stream stayed open after the meeting ended")
	}
	rec := c.getStream("/m/" + sqid + "/stream?peer=" + peer)
	_, _, reason := answer(t, rec)
	if rec.Code != http.StatusGone || reason != "ended" {
		t.Fatalf("resume after the ending: %d %q, want 410 ended", rec.Code, reason)
	}
	if rec := c.getStream("/m/" + sqid + "/stream"); rec.Code != http.StatusNotFound {
		t.Fatalf("fresh stream on an ended meeting: %d, want 404", rec.Code)
	}

	// A meeting nobody opened a stream on: the hub has never heard of
	// it, so only the row can refuse it.
	quiet, _ := createMeeting(t, c, "Never opened", "")
	if rec := c.post("/meetings/"+quiet+"/end", nil); rec.Code != http.StatusSeeOther {
		t.Fatalf("end the quiet meeting: %d", rec.Code)
	}
	if rec := c.getStream("/m/" + quiet + "/stream"); rec.Code != http.StatusNotFound {
		t.Fatalf("fresh stream on a meeting nobody opened: %d, want 404", rec.Code)
	}
	// The peer the ending tombstoned above, offered to a room it was
	// never in: a tombstone answers only through its own room and
	// session, so this is the plain 404 an unknown peer gets.
	rec = c.getStream("/m/" + quiet + "/stream?peer=" + peer)
	if _, gotBoot, _ := answer(t, rec); rec.Code != http.StatusNotFound || gotBoot != boot {
		t.Fatalf("resume into a meeting nobody opened: %d boot %q", rec.Code, gotBoot)
	}
	if rec := c.getStream("/m/zzzzzzzzzz/stream"); rec.Code != http.StatusNotFound {
		t.Fatalf("stream for a meeting that does not exist: %d", rec.Code)
	}
}

// Ruling 15: the attaching stream is opened by EventSource, which always
// asks for text/event-stream. A cross-site top-level navigation asks for
// HTML, and it must not be able to take the session's one peer slot in
// the room and hold it for the resume window.
func TestAStreamWithoutTheEventStreamAcceptTakesNoPeer(t *testing.T) {
	x, sqid, _ := memberWithMeeting(t)
	c := x.browser(t)
	c.signIn("paul@example.com")

	rec := c.getNav("/m/" + sqid + "/stream") // a navigation, not EventSource
	if rec.Code != http.StatusNotAcceptable {
		t.Fatalf("stream without the Accept header: %d, want 406", rec.Code)
	}
	// The check runs before any lookup, so the answer is the same for
	// every meeting and every person, and says what is actually wrong.
	if got := strings.TrimSpace(rec.Body.String()); got != `{"error":"accept"}` {
		t.Fatalf("406 body %s", got)
	}
	if rec := c.getNav("/m/" + sqid + "/stream?peer=whatever"); rec.Code != http.StatusNotAcceptable {
		t.Fatalf("resume without the Accept header: %d, want 406", rec.Code)
	}
	s := c.stream("/m/" + sqid + "/stream")
	defer s.close()
	if peer, _ := s.hello(); peer == "" {
		t.Fatal("the stream that does ask for text/event-stream got no peer")
	}
}

// A resume's ?last= is refused above eight kibibytes, before it is
// parsed: one seq per peer and sixteen peers to a room, so nothing this
// size is a browser's.
func TestAnOversizeLastIsRefusedBeforeItIsParsed(t *testing.T) {
	x, sqid, _ := memberWithMeeting(t)
	c := x.browser(t)
	c.signIn("paul@example.com")
	s := c.stream("/m/" + sqid + "/stream")
	peer, _ := s.hello()
	s.close()

	// Valid JSON, and too big: a last that only the cap can refuse. Bad
	// JSON would be refused by the parse whether the cap is there or
	// not, and would prove nothing.
	var big strings.Builder
	big.WriteString("{")
	for i := 0; i < 700; i++ {
		if i > 0 {
			big.WriteString(",")
		}
		fmt.Fprintf(&big, "%q:1", fmt.Sprintf("%032x", i))
	}
	big.WriteString("}")
	if !json.Valid([]byte(big.String())) || big.Len() <= 8<<10 {
		t.Fatalf("the oversize last is %d bytes and valid=%v", big.Len(), json.Valid([]byte(big.String())))
	}
	rec := c.getStream("/m/" + sqid + "/stream?peer=" + peer + "&last=" + url.QueryEscape(big.String()))
	if rec.Code != http.StatusBadRequest {
		t.Fatalf("oversize last: %d, want 400", rec.Code)
	}
	// Still resumable: a refused query is not a refused participation.
	s2 := c.stream("/m/" + sqid + "/stream?peer=" + peer + "&last=" + url.QueryEscape(`{"`+peer+`":3}`))
	defer s2.close()
	if p, _ := s2.hello(); p != peer {
		t.Fatalf("resume after the refused last: %s, want %s", p, peer)
	}
}

// Spec §4.4: a subscriber that falls sixty-four events behind is
// declared lost and told so, and §4.5 makes lost{buffer} the one reason
// a browser answers with a resume instead of a screen. This is that path
// over real HTTP rather than against hub.Sub directly: a client that
// stops reading its socket backs the handler's write up, the hub's
// sixty-four-slot buffer fills behind the stalled handler, and the
// participation itself survives — so the resume that follows gets the
// same peer back.
func TestASlowClientIsLostOnTheBufferAndResumesTheSameParticipation(t *testing.T) {
	x, sqid, _ := memberWithMeeting(t)
	sender := x.browser(t)
	sender.signIn("paul@example.com")
	ss := sender.stream("/m/" + sqid + "/stream")
	defer ss.close()
	from, _ := ss.hello()
	if resp, _ := sender.signal(sqid, from, `{"type":"join"}`); resp.StatusCode != http.StatusOK {
		t.Fatalf("sender join: %d", resp.StatusCode)
	}

	// The victim reads its hello and then stops reading altogether.
	victim := newMemberBrowser(t, x, "paul@rslw.com")
	vs := victim.rawStream("/m/" + sqid + "/stream")
	defer vs.close()
	to := vs.hello()
	if resp, _ := victim.signal(sqid, to, `{"type":"join"}`); resp.StatusCode != http.StatusOK {
		t.Fatalf("victim join: %d", resp.StatusCode)
	}
	if ev, _, ok := ss.nextOf("peer", 2*time.Second); !ok {
		t.Fatalf("the sender never saw the victim admitted (got %q); nothing below would reach it", ev)
	}

	// The arithmetic, and it is the host's sysctls that it deliberately
	// does not depend on. Two things absorb events before the hub's own
	// buffer starts filling: the socket, and the handler's one in hand.
	// Both ends of the socket are pinned — the server's SO_SNDBUF by the
	// harness's own listener (about 16 KiB once Linux has doubled it),
	// the client's SO_RCVBUF by rawStream's dial (about 8 KiB) — so at
	// 60 KiB an event the socket holds at most one, whatever tcp_wmem's
	// ceiling happens to be on this machine. The hub's buffer is
	// sixty-four, and the handler holds one more. So the overflow is at
	// about sixty-six, and a hundred is comfortably past it.
	//
	// The limiter is thirty a second with a burst of sixty on the
	// harness's frozen clock, so the burst is spent in batches with a
	// refill between them rather than being throttled halfway through.
	const flood = 100
	payload := strings.Repeat("x", 60*1024)
	for i := 1; i <= flood; i++ {
		if i%50 == 1 && i > 1 {
			x.advance(2 * time.Second) // refills the whole burst
		}
		body := fmt.Sprintf(`{"type":"candidate","to":%q,"seq":%d,"gen":1,"payload":{"c":%q}}`, to, i, payload)
		if resp, b := sender.signal(sqid, from, body); resp.StatusCode != http.StatusOK {
			t.Fatalf("candidate %d: %d %s", i, resp.StatusCode, b)
		}
	}

	// Now the victim starts reading again. Whatever was buffered comes
	// through first; the stream ends on lost{buffer}.
	var seen int
	for {
		name, data, ok := vs.next(30 * time.Second)
		if !ok {
			t.Fatalf("the stream ended without lost{buffer} after %d of %d events; "+
				"if that is close to %d the socket absorbed the whole flood and the "+
				"harness's SO_SNDBUF did not take", seen, flood, flood)
		}
		seen++
		if name != "lost" {
			continue
		}
		var lost struct{ Reason string }
		json.Unmarshal(data, &lost)
		if lost.Reason != "buffer" {
			t.Fatalf("lost %s, want reason buffer", data)
		}
		break
	}
	// Events were actually lost, which is the whole claim: a stream that
	// delivered every one of them and then said lost{buffer} would be
	// something else entirely. The bound is the arithmetic above and not
	// merely "fewer than the flood": sixty-six is what the buffer, the
	// handler's one in hand and a socket that can hold at most one event
	// add up to, so a number well past it means the socket absorbed the
	// flood and the harness's pinned send buffer did not take.
	const ceiling = 72
	if seen > ceiling {
		t.Fatalf("the victim read %d of %d events before it was declared lost; "+
			"the buffer is sixty-four and the socket should hold at most one 60 KiB event, "+
			"so anything past %d means the send buffer was not pinned", seen, flood, ceiling)
	}

	// The participation is not the subscriber: it is still there, and a
	// resume gets the same peer and the roster back (§4.5).
	vs.close()
	vs2 := victim.stream("/m/" + sqid + "/stream?peer=" + to)
	defer vs2.close()
	if p, _ := vs2.hello(); p != to {
		t.Fatalf("resume after lost{buffer}: %s, want %s", p, to)
	}
	if ev, _, ok := vs2.nextOf("roster", 2*time.Second); !ok {
		t.Fatalf("the resumed peer is not admitted any more (got %q)", ev)
	}
}

// Spec §2.4's ruling 14, end to end: an admitted guest is never ended by
// their grant lapsing, because the participation is the authority once
// they are in. So signal keeps working, the stream keeps resuming — and
// fresh entry, which is standing's alone, is refused.
func TestAnAdmittedGuestCarriesOnPastGrantExpiryButGetsNoFreshEntry(t *testing.T) {
	x, sqid, tok := memberWithMeeting(t)
	m := x.browser(t)
	m.signIn("paul@example.com")
	sm := m.stream("/m/" + sqid + "/stream")
	defer sm.close()
	pm, _ := sm.hello()
	m.signal(sqid, pm, `{"type":"join"}`)

	g := x.browser(t)
	g.get("/j/" + tok)
	g.signIn("guest@example.com")
	sg := g.stream("/m/" + sqid + "/stream")
	defer sg.close()
	pg, _ := sg.hello()
	g.signal(sqid, pg, `{"type":"join"}`)
	m.signal(sqid, pm, `{"type":"admit","payload":{"peer":"`+pg+`"}}`)
	// Proof of admission before anything is claimed about it: a waiting
	// guest would also survive most of what follows, and would prove
	// nothing about ruling 14.
	if ev, _, ok := sg.nextOf("roster", 2*time.Second); !ok {
		t.Fatalf("the guest was never admitted (got %q)", ev)
	}

	x.advance(2*time.Hour + time.Minute) // the grant is over

	// signal: still theirs. join is idempotent for an admitted peer, and
	// a targeted relay is something only an admitted peer may send, so
	// both are worth asking.
	if resp, b := g.signal(sqid, pg, `{"type":"join"}`); resp.StatusCode != http.StatusOK {
		t.Fatalf("join past grant expiry: %d %s", resp.StatusCode, b)
	}
	if resp, b := g.signal(sqid, pg, `{"type":"offer","to":"`+pm+`","seq":1,"gen":1,"payload":{}}`); resp.StatusCode != http.StatusOK {
		t.Fatalf("offer past grant expiry: %d %s", resp.StatusCode, b)
	}
	if ev, _, ok := sm.nextOf("offer", 2*time.Second); !ok {
		t.Fatalf("the member never received the expired guest's offer (got %q)", ev)
	}

	// stream?peer=: still theirs, roster and all.
	sg.close()
	sg2 := g.stream("/m/" + sqid + "/stream?peer=" + pg)
	defer sg2.close()
	if p, _ := sg2.hello(); p != pg {
		t.Fatalf("resume past grant expiry: %s, want %s", p, pg)
	}
	if ev, _, ok := sg2.nextOf("roster", 2*time.Second); !ok {
		t.Fatalf("the resumed guest is not admitted any more (got %q)", ev)
	}

	// GET /m/{sqid}/ice is authorized by the guest's admitted
	// participation (hub.Authorize), never by standing — ruling 14
	// again, on the one endpoint §13 names "on every endpoint": a
	// lapsed grant does not cost this expired guest their ICE servers.
	x.ctl.SetRelay(relayContract(t))
	rec := g.get("/m/" + sqid + "/ice?peer=" + pg)
	if rec.Code != http.StatusOK {
		t.Fatalf("GET /ice past grant expiry: %d, want 200", rec.Code)
	}
	var ice struct {
		ICEServers []struct {
			URLs []string `json:"urls"`
		} `json:"iceServers"`
	}
	if err := json.Unmarshal(rec.Body.Bytes(), &ice); err != nil {
		t.Fatal(err)
	}
	if len(ice.ICEServers) == 0 {
		t.Fatal("GET /ice past grant expiry returned no servers")
	}

	// Fresh entry is not theirs: a stream with no peer is authorized by
	// standing, and standing is gone. 404, never 403 (§2.5).
	if rec := g.getStream("/m/" + sqid + "/stream"); rec.Code != http.StatusNotFound {
		t.Fatalf("fresh stream past grant expiry: %d, want 404", rec.Code)
	}
	// It is standing that refused that, not the one-peer-per-session
	// rule: this session's peer is refused with 409 while it holds one,
	// and 409 is what this would be if the standing check had passed.
	// (A second browser for the same guest cannot be used to make the
	// point: signing one in needs the join link again, and following the
	// link mints a fresh grant, which is the very thing that has to be
	// gone here.)
	//
	// The holding screen is *not* refused: roomAccess is standing OR a
	// live participation (§2.5), and ruling 14 keeps this one's
	// participation. TestRoomPageSurvivesGrantExpiryForAnAdmittedGuest
	// is the room page's half of the same rule.
	if rec := g.get("/m/" + sqid + "/join"); rec.Code != http.StatusOK {
		t.Fatalf("holding screen past grant expiry: %d, want 200", rec.Code)
	}
}

// Spec §2.4 and §13's "a member dropped on the tick", over HTTP: the
// team changes, the next tick re-derives standing for every joined
// participation, and the member who is no longer on it loses the room
// while the room is told they are gone.
func TestAMemberDroppedFromTheAllowlistIsLostOnTheTick(t *testing.T) {
	x, sqid, _ := memberWithMeeting(t)
	a := x.browser(t)
	a.signIn("paul@example.com")
	sa := a.stream("/m/" + sqid + "/stream")
	defer sa.close()
	pa, _ := sa.hello()
	a.signal(sqid, pa, `{"type":"join"}`)

	b := newMemberBrowser(t, x, "paul@rslw.com")
	sb := b.stream("/m/" + sqid + "/stream")
	defer sb.close()
	pb, _ := sb.hello()
	b.signal(sqid, pb, `{"type":"join"}`)
	if ev, _, ok := sa.nextOf("peer", 2*time.Second); !ok {
		t.Fatalf("the two members never met (got %q)", ev)
	}

	x.ctl.SetAllowlist("paul@example.com") // B is off the team
	x.ctl.TickForTest()

	ev, data, ok := sb.nextOf("lost", 2*time.Second)
	if !ok {
		t.Fatalf("the dropped member's stream stayed open (got %q)", ev)
	}
	var lost struct{ Reason string }
	json.Unmarshal(data, &lost)
	if lost.Reason != "standing" {
		t.Fatalf("dropped member got lost %s, want reason standing", data)
	}
	ev, data, ok = sa.nextOf("gone", 2*time.Second)
	if !ok {
		t.Fatalf("the room was never told the member left (got %q)", ev)
	}
	var gone struct{ From string }
	json.Unmarshal(data, &gone)
	if gone.From != pb {
		t.Fatalf("gone names %q, want %q", gone.From, pb)
	}

	// Their participation is gone, and the tombstone says why (§4.5).
	rec := b.getStream("/m/" + sqid + "/stream?peer=" + pb)
	if _, _, reason := answer(t, rec); rec.Code != http.StatusGone || reason != "standing" {
		t.Fatalf("resume after the drop: %d %q, want 410 standing", rec.Code, reason)
	}
	// And so is entry: no standing, no fresh stream, no room page.
	if rec := b.getStream("/m/" + sqid + "/stream"); rec.Code != http.StatusNotFound {
		t.Fatalf("fresh stream after the drop: %d, want 404", rec.Code)
	}
	if rec := b.get("/m/" + sqid); rec.Code != http.StatusNotFound {
		t.Fatalf("room page after the drop: %d, want 404", rec.Code)
	}
}

// A SIGTERM cancels the context the app was built on. rastrillo answers
// the same signal with a ten-second Shutdown, and Shutdown waits for
// handlers without cancelling their contexts, so a live stream that
// watched nothing would hold that whole deadline and the process would
// exit 1 instead of cleanly. The browser resumes, as at the TTL.
func TestShutdownEndsLiveStreams(t *testing.T) {
	x, sqid, _ := memberWithMeeting(t)
	c := x.browser(t)
	c.signIn("paul@example.com")
	s := c.stream("/m/" + sqid + "/stream")
	defer s.close()
	s.hello()

	x.shutdown()
	if !s.waitClosed(2 * time.Second) {
		t.Fatal("a live stream outlived the app's context")
	}
}

// SetStreamTimings is a test knob, and a zero from it must not panic a
// ticker inside a request goroutine: both timings have a floor.
func TestZeroStreamTimingsAreClampedNotFatal(t *testing.T) {
	x, sqid, _ := memberWithMeeting(t)
	// A zero heartbeat with a TTL that is not zero: the heartbeat is
	// what proves the clamp. hello is written before the ticker exists,
	// and net/http recovers a panic per connection, so a stream that
	// panicked on time.NewTicker(0) would still say hello and still
	// close — but it would never beat.
	x.ctl.SetStreamTimings(0, 500*time.Millisecond)
	c := x.browser(t)
	c.signIn("paul@example.com")
	s := c.stream("/m/" + sqid + "/stream")
	defer s.close()
	if peer, _ := s.hello(); peer == "" {
		t.Fatal("no peer")
	}
	if !s.waitHeartbeat(2 * time.Second) {
		t.Fatal("a floored heartbeat never beat")
	}
	if !s.waitClosed(2 * time.Second) {
		t.Fatal("the stream did not end at its TTL")
	}
	// The app is still serving: nothing panicked.
	if rec := c.get("/m/" + sqid + "/join"); rec.Code != http.StatusOK {
		t.Fatalf("holding screen after the floored stream: %d", rec.Code)
	}
}