package callstest

import (
	"bytes"
	"encoding/json"
	"fmt"
	"log/slog"
	"net/http"
	"net/url"
	"strings"
	"testing"
	"time"

	"amadan.net/oficina/calls/internal/calls"
)

// Spec §4.2: every POST /signal must carry a peer the hub finds in this
// room, bound to this session; anything else is refused with 404 —
// never 403, and the same 404 whether the peer never existed or exists
// through another session (§2.5's "a meeting that is not yours does not
// exist").
func TestSignalNeedsAPeerBoundToThisSessionAndRoom(t *testing.T) {
	x, sqid, tok := memberWithMeeting(t)
	c := x.browser(t)
	c.signIn("paul@example.com")
	s := c.stream("/m/" + sqid + "/stream")
	defer s.close()
	peer, boot := s.hello()

	resp, unknownBody := c.signal(sqid, "0123456789abcdef0123456789abcdef", `{"type":"join"}`)
	if resp.StatusCode != http.StatusNotFound {
		t.Fatalf("foreign peer: %d", resp.StatusCode)
	}
	// Minor 6: the 404 body carries boot, identical for ErrUnknownPeer
	// and ErrNotYours, the same shape stream.go's own 404s carry.
	var unknownAnswer struct {
		Error string
		Boot  string
	}
	if err := json.Unmarshal(unknownBody, &unknownAnswer); err != nil {
		t.Fatalf("404 body %s: %v", unknownBody, err)
	}
	if unknownAnswer.Error != "not-found" || unknownAnswer.Boot != boot {
		t.Fatalf("404 body %+v, want error=not-found boot=%q", unknownAnswer, boot)
	}

	other := x.browser(t)
	other.get("/j/" + tok)
	other.signIn("guest@example.com")
	resp, otherSessionBody := other.signal(sqid, peer, `{"type":"join"}`)
	if resp.StatusCode != http.StatusNotFound {
		t.Fatalf("peer through another session: %d", resp.StatusCode)
	}
	if !bytes.Equal(unknownBody, otherSessionBody) {
		t.Fatalf("404 bodies differ: unknown %q, other session %q", unknownBody, otherSessionBody)
	}

	resp, _ = c.signal(sqid, peer, `{"type":"join"}`)
	if resp.StatusCode != http.StatusOK {
		t.Fatalf("own join: %d", resp.StatusCode)
	}
}

// A peer that is perfectly real, and bound to this very session, but in
// a different room, is the same 404 — and from Task 5's fix round 2 it
// is refused earlier than it used to be: the LiveForSession gate answers
// it before the limiter, the body read or any hub dispatch, because this
// session holds no participation in the room named in the URL. Either
// way the rule is §4.2's "cross-room delivery is impossible by
// construction": every lookup starts from the room in the URL, so a peer
// id from elsewhere names nothing here.
func TestSignalRefusesAPeerFromAnotherMeeting(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, _ := s.hello()

	elsewhere, _ := createMeeting(t, c, "Different room", "")
	if resp, _ := c.signal(elsewhere, peer, `{"type":"join"}`); resp.StatusCode != http.StatusNotFound {
		t.Fatalf("peer from another meeting: %d, want 404", resp.StatusCode)
	}
}

// Task 5 review Important 1: the limiter is keyed on session+room, never
// on anything the request body claims, so a second session flooding the
// endpoint spends only its own bucket.
func TestSignalRateLimitIsPerSessionNotPerClaimedPeer(t *testing.T) {
	x, sqid, _ := memberWithMeeting(t)
	victim := x.browser(t)
	victim.signIn("paul@example.com")
	vs := victim.stream("/m/" + sqid + "/stream")
	defer vs.close()
	vpeer, _ := vs.hello()
	if resp, _ := victim.signal(sqid, vpeer, `{"type":"join"}`); resp.StatusCode != http.StatusOK {
		t.Fatalf("victim join: %d", resp.StatusCode)
	}

	// A second session in the same room, flooding while *claiming the
	// victim's own peer id* — peer ids are published in rosters, so
	// this is exactly what a limiter keyed on the body's claimed peer
	// cannot tell from the victim's own traffic (review Important 1).
	attacker := newMemberBrowser(t, x, "paul@rslw.com")
	as := attacker.stream("/m/" + sqid + "/stream")
	defer as.close()
	as.hello() // the attacker's own peer, unused below on purpose

	for i := 0; i < 80; i++ {
		attacker.signal(sqid, vpeer, `{"type":"leave"}`)
	}

	// The victim's own session, still using its own real peer, must be
	// unaffected by traffic that only ever came from another session.
	if resp, _ := victim.signal(sqid, vpeer, `{"type":"leave"}`); resp.StatusCode == http.StatusTooManyRequests {
		t.Fatal("another session impersonating this peer id in its own requests drained its rate-limit bucket")
	}
}

// Task 5 review Important 1: rotating the claimed peer id within one
// session must not evade the limit — every claimed id, real or not,
// spends the same session+room bucket. (The room half of the key is
// gated by LiveForSession before the limiter is ever reached — fix
// round 2 item 1 — so this test is only about the peer half: the
// session here already holds a live participation in sqid throughout.)
func TestSignalRateLimitSharesOneBucketAcrossClaimedPeerIDs(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, _ := s.hello()
	c.signal(sqid, peer, `{"type":"join"}`)

	var got429 bool
	for i := 0; i < 100; i++ {
		// A distinct, format-valid, but otherwise made-up peer id on
		// every call: none of them is bound to anything, so if the
		// limiter were still keyed on the claimed peer, rotating like
		// this would never be throttled.
		fake := fmt.Sprintf("%032x", i)
		resp, _ := c.signal(sqid, fake, `{"type":"join"}`)
		if resp.StatusCode == http.StatusTooManyRequests {
			got429 = true
			break
		}
	}
	if !got429 {
		t.Fatal("rotating the claimed peer id within one session evaded the rate limit")
	}
}

// Task 5 review Important 2: an oversized peer field is refused, and
// never becomes a log argument — the log's "peer" is only ever a value
// already known to be exactly 32 lowercase hex.
func TestSignalRefusesAnOversizedPeerAndNeverLogsIt(t *testing.T) {
	var buf bytes.Buffer
	x := newHarnessWithLogger(t, calls.EnvDevelopment, slog.New(slog.NewTextHandler(&buf, &slog.HandlerOptions{Level: slog.LevelDebug})), "paul@example.com")
	c := x.browser(t)
	c.signIn("paul@example.com")
	sqid, _ := createMeeting(t, c, "x", "")
	s := c.stream("/m/" + sqid + "/stream")
	defer s.close()
	s.hello()

	huge := strings.Repeat("a", 60*1024)
	resp, _ := c.signal(sqid, huge, `{"type":"join"}`)
	if resp.StatusCode != http.StatusBadRequest {
		t.Fatalf("60 KB peer: %d, want 400", resp.StatusCode)
	}
	if strings.Contains(buf.String(), huge) {
		t.Fatal("the oversized peer string reached the log")
	}
}

// fakeButValidSqid builds a room id in the exact shape ids.go's
// validSqid accepts (ten characters of sqidAlphabet) without touching
// crypto/rand — deterministic and cheap for a test that needs many
// distinct, well-formed ids no meeting will ever have.
func fakeButValidSqid(i int) string {
	const digits = "23456789" // a safe subset of sqidAlphabet
	b := make([]byte, 10)
	for j := 9; j >= 0; j-- {
		b[j] = digits[i%len(digits)]
		i /= len(digits)
	}
	return string(b)
}

// Task 5 review fix round 2 item 1: rotating the room id in the URL
// must not mint a rate-limit bucket per request. A session with no live
// participation in a room is refused before the limiter is ever
// touched, so fifty rotated (but well-formed) room ids leave the
// limiter's map exactly as it was after the one real signal.
func TestSignalSqidRotationNeverMintsARateLimiterBucket(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, _ := s.hello()
	if resp, _ := c.signal(sqid, peer, `{"type":"join"}`); resp.StatusCode != http.StatusOK {
		t.Fatalf("join: %d", resp.StatusCode)
	}
	if n := x.ctl.SignalLimiterSize(); n != 1 {
		t.Fatalf("limiter buckets after one signal: %d, want 1", n)
	}

	for i := 0; i < 50; i++ {
		fake := fakeButValidSqid(i)
		if resp, _ := c.signal(fake, peer, `{"type":"join"}`); resp.StatusCode != http.StatusNotFound {
			t.Fatalf("rotated room %q: %d, want 404", fake, resp.StatusCode)
		}
	}
	if n := x.ctl.SignalLimiterSize(); n != 1 {
		t.Fatalf("limiter buckets after 50 rotated rooms: %d, want 1 — rotation must not mint a bucket per request", n)
	}
}

// Task 5 review fix round 2 item 1: an oversized room id is refused
// before it can reach the limiter's key, mint a bucket, or a log line.
func TestSignalRefusesAnOversizedRoomIDAndNeverLogsIt(t *testing.T) {
	var buf bytes.Buffer
	x := newHarnessWithLogger(t, calls.EnvDevelopment, slog.New(slog.NewTextHandler(&buf, &slog.HandlerOptions{Level: slog.LevelDebug})), "paul@example.com")
	c := x.browser(t)
	c.signIn("paul@example.com")

	huge := strings.Repeat("a", 1024)
	resp, badRoomBody := c.signalRaw(huge, `{"peer":"0123456789abcdef0123456789abcdef","type":"join"}`)
	if resp.StatusCode != http.StatusNotFound {
		t.Fatalf("1 KB room id: %d, want 404", resp.StatusCode)
	}
	if n := x.ctl.SignalLimiterSize(); n != 0 {
		t.Fatalf("limiter buckets after an oversized room id: %d, want 0", n)
	}
	if strings.Contains(buf.String(), huge) {
		t.Fatal("the oversized room id reached the log")
	}

	// Byte for byte the not-found body every other refusal here carries.
	// A malformed room id is refused earlier and for a different reason
	// than a peer that is not yours, and neither of those is anybody's
	// business from outside: one answer, or the difference between them
	// is a way to ask which meetings exist (§2.5).
	sqid, _ := createMeeting(t, c, "A room this session has never opened a stream in", "")
	resp, unboundBody := c.signal(sqid, "0123456789abcdef0123456789abcdef", `{"type":"join"}`)
	if resp.StatusCode != http.StatusNotFound {
		t.Fatalf("a room with no live participation: %d, want 404", resp.StatusCode)
	}
	if !bytes.Equal(badRoomBody, unboundBody) {
		t.Fatalf("404 bodies differ: bad room %q, no participation %q", badRoomBody, unboundBody)
	}
}

// Task 5's fix round 2 ruling, and §4.5's window: the LiveForSession
// gate in front of the limiter is about a session that holds nothing in
// this room, not about a browser whose stream happens to be down. A
// detached participation inside its resume window is still this
// session's peer, so its signalling is still accepted — which is what
// lets a browser post `leave`, or answer an offer, in the gap between
// two streams. Once the window lapses and the tick clears it, the same
// request is the same 404 a stranger gets.
func TestSignalFromADetachedSessionInsideItsWindowIsAccepted(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()
	if resp, _ := c.signal(sqid, peer, `{"type":"join"}`); resp.StatusCode != http.StatusOK {
		t.Fatalf("join: %d", resp.StatusCode)
	}
	if ev, _, ok := s.nextOf("roster", 2*time.Second); !ok {
		t.Fatalf("never admitted (got %q)", ev)
	}

	s.close() // the stream is gone; the participation has thirty seconds
	if resp, b := c.signal(sqid, peer, `{"type":"join"}`); resp.StatusCode != http.StatusOK {
		t.Fatalf("signal from a detached session inside its window: %d %s", resp.StatusCode, b)
	}
	if n := x.ctl.SignalLimiterSize(); n != 1 {
		t.Fatalf("limiter buckets: %d, want 1 — the detached session reached the limiter", n)
	}

	x.advance(31 * time.Second)
	x.ctl.TickForTest()
	if resp, _ := c.signal(sqid, peer, `{"type":"join"}`); resp.StatusCode != http.StatusNotFound {
		t.Fatalf("signal after the window lapsed: %d, want 404", resp.StatusCode)
	}
}

// Task 5's fix round 2 item 3: the 429 line names the peer the hub says
// this session holds, never the one the request claimed. A flood that
// rotates its claimed peer id must leave one log line's worth of truth
// behind — otherwise the rate-limit log is written by the attacker.
func TestTheRateLimitLogNamesTheSessionsRealPeer(t *testing.T) {
	var buf bytes.Buffer
	x := newHarnessWithLogger(t, calls.EnvDevelopment, slog.New(slog.NewTextHandler(&buf, &slog.HandlerOptions{Level: slog.LevelDebug})), "paul@example.com")
	c := x.browser(t)
	c.signIn("paul@example.com")
	sqid, _ := createMeeting(t, c, "x", "")
	s := c.stream("/m/" + sqid + "/stream")
	defer s.close()
	peer, _ := s.hello()
	c.signal(sqid, peer, `{"type":"join"}`)

	var claimed string
	var got429 bool
	for i := 0; i < 100; i++ {
		claimed = fmt.Sprintf("%032x", i)
		resp, _ := c.signal(sqid, claimed, `{"type":"join"}`)
		if resp.StatusCode == http.StatusTooManyRequests {
			got429 = true
			break
		}
	}
	if !got429 {
		t.Fatal("the burst never hit the rate limit")
	}

	var rateLine string
	for _, line := range strings.Split(buf.String(), "\n") {
		if strings.Contains(line, "outcome=rate") {
			rateLine = line
			break
		}
	}
	if rateLine == "" {
		t.Fatalf("no rate-limit line in the log:\n%s", buf.String())
	}
	if !strings.Contains(rateLine, "peer="+peer) {
		t.Fatalf("the 429 line does not name this session's own peer: %s", rateLine)
	}
	if strings.Contains(rateLine, "peer="+claimed) {
		t.Fatalf("the 429 line names the peer the request claimed: %s", rateLine)
	}
	// And nothing else: type is unknown, because the body was never
	// read (the limiter runs first), and the room is the one in the URL.
	if !strings.Contains(rateLine, "type=unknown") || !strings.Contains(rateLine, "room="+sqid) {
		t.Fatalf("the 429 line: %s", rateLine)
	}
}

// Task 5 review minor 8: the body shapes signal.go must classify
// correctly, including the ones client.signal cannot express (a
// malformed document, a body with no peer at all).
func TestSignalBodyShapeTable(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, _ := s.hello()
	if resp, _ := c.signal(sqid, peer, `{"type":"join"}`); resp.StatusCode != http.StatusOK {
		t.Fatalf("setup join: %d", resp.StatusCode)
	}
	absent := fmt.Sprintf("%032x", 0) // well-formed, bound to nobody

	table := []struct {
		name string
		body string
		want int
	}{
		{"chat is slice 4", `{"peer":"` + string(peer) + `","type":"chat"}`, http.StatusBadRequest},
		{"share is slice 4", `{"peer":"` + string(peer) + `","type":"share"}`, http.StatusBadRequest},
		{"bogus type", `{"peer":"` + string(peer) + `","type":"bogus"}`, http.StatusBadRequest},
		{"malformed JSON", `{`, http.StatusBadRequest},
		{"missing peer", `{"type":"join"}`, http.StatusBadRequest},
		{"malformed admit payload.peer", `{"peer":"` + string(peer) + `","type":"admit","payload":{"peer":"not-hex"}}`, http.StatusBadRequest},
		{"well-formed absent admit target", `{"peer":"` + string(peer) + `","type":"admit","payload":{"peer":"` + absent + `"}}`, http.StatusConflict},
	}
	for _, tc := range table {
		t.Run(tc.name, func(t *testing.T) {
			resp, _ := c.signalRaw(sqid, tc.body)
			if resp.StatusCode != tc.want {
				t.Fatalf("%s: %d, want %d", tc.name, resp.StatusCode, tc.want)
			}
		})
	}
}

func TestSignalLimits(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, _ := s.hello()
	c.signal(sqid, peer, `{"type":"join"}`)

	big := `{"type":"candidate","to":"` + peer + `","seq":1,"gen":1,"payload":{"x":"` + strings.Repeat("a", 70*1024) + `"}}`
	if resp, _ := c.signal(sqid, peer, big); resp.StatusCode != http.StatusRequestEntityTooLarge {
		t.Fatalf("64 KB: %d", resp.StatusCode)
	}

	// join, not leave: a successful leave ends the participation, and
	// from fix round 2 on a session with no live participation in this
	// room is refused at the LiveForSession gate before it ever reaches
	// the limiter (item 1) — a burst that ended its own room membership
	// on the first call would never reach the limiter again to prove
	// anything. join on an already-admitted peer is idempotent (spec
	// §4.3), so the participation, and the gate, stay open throughout.
	var got429 bool
	for i := 0; i < 100; i++ {
		resp, _ := c.signal(sqid, peer, `{"type":"join"}`)
		if resp.StatusCode == http.StatusTooManyRequests {
			if resp.Header.Get("Retry-After") == "" {
				t.Fatal("429 without Retry-After")
			}
			got429 = true
			break
		}
	}
	if !got429 {
		t.Fatal("100 messages in a burst never hit the rate limit")
	}
}

// Renamed from TestSignalStaleGenAnswers409WithTheCurrentGen: the
// ledger records the decision that no gen goes inline in the response
// (see task-5-report.md) — the roster already carries it (spec §4.6).
func TestSignalStaleGenIsRefusedWith409(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"}`)

	resp, _ := a.signal(sqid, pa, `{"type":"offer","to":"`+pb+`","seq":1,"gen":7,"payload":{}}`)
	if resp.StatusCode != http.StatusConflict {
		t.Fatalf("stale gen: %d", resp.StatusCode)
	}
}

// Spec §4.3: a guest's join marks it waiting and tells admitted members
// with knock; the guest itself learns nothing at all from a fresh join
// (ruling 9, and hub's own TestMemberJoinsAndGuestWaits, which asserts
// zero events on a waiting guest's sub) — it is the *resume* of a
// waiting or unjoined participation that answers state{waiting:true}
// (stream.go, Task 4), not the join itself. This is the guest half of
// standingFor's hub.KindGuest mapping proven end-to-end over HTTP: a
// guest waits, a member in the same room sees knock.
func TestSignalGuestJoinWaitsAndKnocksTheRoom(t *testing.T) {
	x, sqid, tok := memberWithMeeting(t)
	enableWaitingRoom(t, x, sqid)
	member := x.browser(t)
	member.signIn("paul@example.com")
	ms := member.stream("/m/" + sqid + "/stream")
	defer ms.close()
	mpeer, _ := ms.hello()
	if resp, _ := member.signal(sqid, mpeer, `{"type":"join"}`); resp.StatusCode != http.StatusOK {
		t.Fatalf("member join: %d", resp.StatusCode)
	}
	if ev, _, ok := ms.next(2 * time.Second); !ok || ev != "roster" {
		t.Fatalf("member after its own join: %s ok=%v, want roster", ev, ok)
	}

	guest := x.browser(t)
	guest.get("/j/" + tok)
	guest.signIn("guest@example.com")
	gs := guest.stream("/m/" + sqid + "/stream")
	defer gs.close()
	gpeer, _ := gs.hello()

	if resp, _ := guest.signal(sqid, gpeer, `{"type":"join"}`); resp.StatusCode != http.StatusOK {
		t.Fatalf("guest join: %d", resp.StatusCode)
	}
	if ev, _, ok := ms.next(2 * time.Second); !ok || ev != "knock" {
		t.Fatalf("member after the guest's join: %s ok=%v, want knock", ev, ok)
	}
	if ev, _, ok := gs.next(200 * time.Millisecond); ok {
		t.Fatalf("waiting guest received %q, want nothing", ev)
	}

	// The guest really is only waiting, not admitted: resuming its
	// stream answers state{waiting:true}, not a roster.
	gs.close()
	gs2 := guest.stream("/m/" + sqid + "/stream?peer=" + gpeer)
	defer gs2.close()
	if p, _ := gs2.hello(); p != gpeer {
		t.Fatalf("resume peer %q, want %q", p, gpeer)
	}
	ev, data, ok := gs2.next(2 * time.Second)
	if !ok || ev != "state" {
		t.Fatalf("waiting guest's resume: %s ok=%v, want state", ev, ok)
	}
	var state struct{ Waiting bool }
	json.Unmarshal(data, &state)
	if !state.Waiting {
		t.Fatalf("state %s, want waiting", data)
	}
}

// Spec §4.5 and §4.6: two admitted peers, one drops its stream and
// resumes with ?peer= and last, and gets hello, roster, then the
// replay — in that order, and only the messages above last.
func TestSignalAdmittedResumeReplaysAfterRoster(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()
	if resp, _ := a.signal(sqid, pa, `{"type":"join"}`); resp.StatusCode != http.StatusOK {
		t.Fatalf("a's join: %d", resp.StatusCode)
	}
	if ev, _, ok := sa.next(2 * time.Second); !ok || ev != "roster" {
		t.Fatalf("a after its own join: %s ok=%v, want roster", ev, ok)
	}

	b := newMemberBrowser(t, x, "paul@rslw.com")
	sb := b.stream("/m/" + sqid + "/stream")
	pb, _ := sb.hello()
	if resp, _ := b.signal(sqid, pb, `{"type":"join"}`); resp.StatusCode != http.StatusOK {
		t.Fatalf("b's join: %d", resp.StatusCode)
	}
	if ev, _, ok := sb.next(2 * time.Second); !ok || ev != "roster" {
		t.Fatalf("b after its own join: %s ok=%v, want roster", ev, ok)
	}
	if ev, _, ok := sa.next(2 * time.Second); !ok || ev != "peer" {
		t.Fatalf("a on b's admission: %s ok=%v, want peer", ev, ok)
	}

	// One offer delivered live, so it is out of the replay's range.
	if resp, _ := a.signal(sqid, pa, `{"type":"offer","to":"`+pb+`","seq":1,"gen":1,"payload":{}}`); resp.StatusCode != http.StatusOK {
		t.Fatalf("a's first offer: %d", resp.StatusCode)
	}
	if ev, _, ok := sb.next(2 * time.Second); !ok || ev != "offer" {
		t.Fatalf("b receiving the first offer live: %s ok=%v", ev, ok)
	}

	sb.close()

	// A second offer, sent while b is detached, lands only in the ring.
	if resp, _ := a.signal(sqid, pa, `{"type":"offer","to":"`+pb+`","seq":2,"gen":1,"payload":{}}`); resp.StatusCode != http.StatusOK {
		t.Fatalf("a's second offer: %d", resp.StatusCode)
	}

	last, err := json.Marshal(map[string]uint64{pa: 1})
	if err != nil {
		t.Fatal(err)
	}
	sb2 := b.stream("/m/" + sqid + "/stream?peer=" + pb + "&last=" + url.QueryEscape(string(last)))
	defer sb2.close()
	if p, _ := sb2.hello(); p != pb {
		t.Fatalf("resume hello peer %q, want %q", p, pb)
	}
	if ev, _, ok := sb2.next(2 * time.Second); !ok || ev != "roster" {
		t.Fatalf("first after resume hello: %s ok=%v, want roster", ev, ok)
	}
	ev, data, ok := sb2.next(2 * time.Second)
	if !ok || ev != "offer" {
		t.Fatalf("second after resume: %s ok=%v, want the replayed offer", ev, ok)
	}
	var replay struct{ Seq uint64 }
	json.Unmarshal(data, &replay)
	if replay.Seq != 2 {
		t.Fatalf("replayed offer seq %d, want 2 (the one sent while b was detached)", replay.Seq)
	}
}

// The §0.1 and §4.4 promise: no payload, address, name, title, token or
// session key in any log line, on accepted and refused messages and on
// an internal-error path. Task 5 review item 4: the earlier version of
// this test passed vacuously — it never exercised a guest's knock, an
// admit, or the 500 arm, and its address ("paul@rslw.com") happened not
// to leave a distinctive trace ("paul" and "example.com" both appear
// harmlessly elsewhere). zaphod-xyzzy@example.com is chosen so its
// derived display name (store.ensurePerson's local-part rule,
// "zaphod-xyzzy") is its own needle.
func TestSignallingLogsCarryNoContent(t *testing.T) {
	var buf bytes.Buffer
	x := newHarnessWithLogger(t, calls.EnvDevelopment, slog.New(slog.NewTextHandler(&buf, &slog.HandlerOptions{Level: slog.LevelDebug})), "paul@example.com")
	c := x.browser(t)
	c.signIn("paul@example.com")
	sqid, tok := createMeeting(t, c, "Secret title", "")
	enableWaitingRoom(t, x, sqid)
	s := c.stream("/m/" + sqid + "/stream")
	defer s.close()
	peer, _ := s.hello()
	if resp, _ := c.signal(sqid, peer, `{"type":"join"}`); resp.StatusCode != http.StatusOK {
		t.Fatalf("member join: %d", resp.StatusCode)
	}

	// A real guest, via the join link, not another member: its join is
	// the knock, and it stays out of the log entirely until admitted.
	guest := x.browser(t)
	guest.get("/j/" + tok)
	guest.signIn("zaphod-xyzzy@example.com")
	gs := guest.stream("/m/" + sqid + "/stream")
	defer gs.close()
	gpeer, _ := gs.hello()
	if resp, _ := guest.signal(sqid, gpeer, `{"type":"join"}`); resp.StatusCode != http.StatusOK {
		t.Fatalf("guest join (the knock): %d", resp.StatusCode)
	}

	// The admit round: previously untested by this file at all.
	if resp, _ := c.signal(sqid, peer, `{"type":"admit","payload":{"peer":"`+gpeer+`"}}`); resp.StatusCode != http.StatusOK {
		t.Fatalf("admit: %d", resp.StatusCode)
	}

	c.signal(sqid, peer, `{"type":"offer","to":"`+gpeer+`","seq":1,"gen":1,"payload":{"sdp":"v=0 SDPSECRET a=candidate 203.0.113.9"}}`)
	c.signal(sqid, peer, `{"type":"candidate","to":"`+gpeer+`","seq":2,"gen":1,"payload":{"candidate":"candidate:1 1 udp 2 203.0.113.9 5000 typ host"}}`)
	c.signal(sqid, peer, `{"type":"offer","to":"`+gpeer+`","seq":3,"gen":9,"payload":{"sdp":"STALESECRET"}}`) // outcome=stale-gen
	c.signal(sqid, peer, `{"type":"bogus"}`)                                                                  // outcome=type

	// An accepted chat appends nothing to the log at all — no line, let
	// alone one carrying its text — and an oversize one is refused with
	// the usual outcome-only line, never the text that overflowed it.
	beforeChat := buf.Len()
	if resp, _ := c.signal(sqid, peer, `{"type":"chat","payload":{"text":"xyzzy-plugh-chat"}}`); resp.StatusCode != http.StatusOK {
		t.Fatalf("accepted chat: %d", resp.StatusCode)
	}
	if buf.Len() != beforeChat {
		t.Fatalf("an accepted chat appended %d bytes to the log: %q", buf.Len()-beforeChat, buf.String()[beforeChat:])
	}
	// "xyzzy-oversize-" is its own needle: the earlier assertion only
	// proved the accepted chat's text never grew the log, not that a
	// *refused* chat's text is equally absent (review fix round 1,
	// item 2).
	oversizeChat := `{"type":"chat","payload":{"text":"xyzzy-oversize-` + strings.Repeat("a", 2049) + `"}}`
	if resp, _ := c.signal(sqid, peer, oversizeChat); resp.StatusCode != http.StatusRequestEntityTooLarge {
		t.Fatalf("oversize chat: %d", resp.StatusCode)
	} // outcome=size

	x.advance(time.Minute)
	x.ctl.TickForTest()

	// Force the internal-error arm without breaking sign-in itself:
	// closing the whole database also breaks rastrillo's own
	// per-request session lookup, which refuses the request before it
	// ever reaches the signal handler. Dropping only the meetings table
	// leaves sessions and persons answering normally, so the request
	// reaches hub.Join, whose standing re-derivation reads meetings and
	// surfaces a genuine driver error — a sentinel this switch does not
	// already name — exercising the one log line no other case in this
	// test reaches (review minor 9).
	//
	// In its own trailing t.Run, named for what it does and that it
	// must be last: the table it drops is gone for the rest of the
	// process, so nothing after this subtest may touch it (review fix
	// round 2 item 7 — a structural marker, not just a comment, for a
	// constraint a future edit could otherwise violate silently).
	t.Run("forced internal error (must run last: drops the meetings table)", func(t *testing.T) {
		if err := x.db.G.Exec("DROP TABLE meetings").Error; err != nil {
			t.Fatal(err)
		}
		resp, _ := c.signal(sqid, peer, `{"type":"join"}`)
		if resp.StatusCode != http.StatusInternalServerError {
			t.Fatalf("forced internal error: %d, want 500", resp.StatusCode)
		}
	})

	logs := buf.String()
	for _, want := range []string{"outcome=stale-gen", "outcome=type", "outcome=size", "outcome=internal"} {
		if !strings.Contains(logs, want) {
			t.Errorf("log missing %q:\n%s", want, logs)
		}
	}

	forbidden := []string{
		"Secret title", "SDPSECRET", "203.0.113.9", "STALESECRET", tok,
		"paul@example.com", "zaphod-xyzzy@example.com", "zaphod-xyzzy",
		"xyzzy-plugh-chat", "xyzzy-oversize-",
	}
	for _, ck := range c.cookies {
		forbidden = append(forbidden, ck.Value)
	}
	for _, ck := range guest.cookies {
		forbidden = append(forbidden, ck.Value)
	}
	for _, needle := range forbidden {
		if needle != "" && strings.Contains(logs, needle) {
			t.Errorf("log carries %q", needle)
		}
	}
}