package hub

import (
	"context"
	"encoding/json"
	"errors"
	"io"
	"log/slog"
	"testing"
	"time"
)

func twoAdmitted(t *testing.T) (*Hub, *time.Time, OpenResult, OpenResult) {
	t.Helper()
	now := time.Date(2026, 9, 10, 9, 0, 0, 0, time.UTC)
	h := testHub(&now)
	ctx := context.Background()
	a, _ := h.Open("r1", "sa", 1, "Ann", KindMember)
	h.Join(ctx, "r1", "sa", a.Peer)
	b, _ := h.Open("r1", "sb", 2, "Bob", KindMember)
	h.Join(ctx, "r1", "sb", b.Peer)
	drain(a.Sub)
	drain(b.Sub)
	return h, &now, a, b
}

func TestRelayGoesOnlyToTheNamedAdmittedPeerWithServerBuiltEnvelope(t *testing.T) {
	h, _, a, b := twoAdmitted(t)
	ctx := context.Background()
	g, _ := h.Open("r1", "sg", 100, "Gus", KindGuest)
	h.Join(ctx, "r1", "sg", g.Peer)
	drain(a.Sub)
	drain(b.Sub)
	err := h.Relay(testCtx, "r1", "sa", a.Peer, Message{Type: TypeOffer, To: b.Peer, Seq: 1, Gen: 1, From: "forged", Name: "Forged", Payload: json.RawMessage(`{"sdp":"x"}`)})
	if err != nil {
		t.Fatal(err)
	}
	bev := drain(b.Sub)
	if len(bev) != 1 || bev[0].From != a.Peer || bev[0].Name != "Ann" || string(bev[0].Payload) != `{"sdp":"x"}` {
		t.Fatalf("b got %+v", bev)
	}
	if ev := drain(a.Sub); len(ev) != 0 {
		t.Fatalf("sender echoed %v", types(ev))
	}
	if ev := drain(g.Sub); len(ev) != 0 {
		t.Fatalf("waiting guest received a relay: %v", types(ev))
	}
	if err := h.Relay(testCtx, "r1", "sa", a.Peer, Message{Type: TypeOffer, To: g.Peer, Seq: 2, Gen: 1}); err == nil {
		t.Fatal("relay to a waiting peer accepted")
	}
	if err := h.Relay(testCtx, "r1", "sg", g.Peer, Message{Type: TypeOffer, To: a.Peer, Seq: 1, Gen: 1}); err != ErrNotAdmitted {
		t.Fatalf("waiting guest relaying: %v", err)
	}
	if err := h.Relay(testCtx, "r1", "sb", a.Peer, Message{Type: TypeOffer, To: b.Peer, Seq: 3, Gen: 1}); err != ErrUnknownPeer {
		t.Fatalf("peer through another session: %v", err)
	}
	if err := h.Relay(testCtx, "r2", "sa", a.Peer, Message{Type: TypeOffer, To: b.Peer, Seq: 3, Gen: 1}); err != ErrUnknownPeer {
		t.Fatalf("peer through another room: %v", err)
	}
	if err := h.Relay(testCtx, "r1", "sa", a.Peer, Message{Type: TypeChat, To: b.Peer, Seq: 4, Gen: 1}); !errors.Is(err, ErrNotTargeted) {
		t.Fatalf("chat as a targeted relay: %v", err)
	}
}

// A peer is not its own target: a self-addressed relay or reset would
// otherwise ring a message the sender already has and open a pair with
// itself, which no roster ever names.
func TestAPeerIsNotItsOwnTarget(t *testing.T) {
	h, _, a, _ := twoAdmitted(t)
	if err := h.Relay(testCtx, "r1", "sa", a.Peer, Message{Type: TypeOffer, To: a.Peer, Seq: 1, Gen: 1}); err != ErrUnknownPeer {
		t.Fatalf("self-addressed relay: %v", err)
	}
	if err := h.Reset(testCtx, "r1", "sa", a.Peer, a.Peer); err != ErrUnknownPeer {
		t.Fatalf("self-addressed reset: %v", err)
	}
	if ev := drain(a.Sub); len(ev) != 0 {
		t.Fatalf("a self target reached the sender: %v", types(ev))
	}
	p, _ := h.Find("r1", "sa", a.Peer)
	if len(p.ring) != 0 {
		t.Fatalf("a self target filled the ring: %+v", p.ring)
	}
}

func TestStaleGenIsRefusedAndResetConverges(t *testing.T) {
	h, _, a, b := twoAdmitted(t)
	if err := h.Relay(testCtx, "r1", "sa", a.Peer, Message{Type: TypeOffer, To: b.Peer, Seq: 1, Gen: 2}); err != ErrStaleGen {
		t.Fatalf("gen 2 before any reset: %v", err)
	}
	if err := h.Reset(testCtx, "r1", "sa", a.Peer, b.Peer); err != nil {
		t.Fatal(err)
	}
	if err := h.Reset(testCtx, "r1", "sb", b.Peer, a.Peer); err != nil { // simultaneous resets converge on the higher gen
		t.Fatal(err)
	}
	aev, bev := drain(a.Sub), drain(b.Sub)
	if len(aev) != 2 || aev[1].Type != TypePair || aev[1].Gen != 3 || aev[1].Peer != b.Peer {
		t.Fatalf("a got %+v", aev)
	}
	if len(bev) != 2 || bev[1].Gen != 3 || bev[1].Peer != a.Peer {
		t.Fatalf("b got %+v", bev)
	}
	if err := h.Relay(testCtx, "r1", "sa", a.Peer, Message{Type: TypeOffer, To: b.Peer, Seq: 1, Gen: 1}); err != ErrStaleGen {
		t.Fatal("old gen accepted after reset")
	}
	if err := h.Relay(testCtx, "r1", "sa", a.Peer, Message{Type: TypeOffer, To: b.Peer, Seq: 1, Gen: 3}); err != nil {
		t.Fatal(err)
	}
}

func TestResumeReplaysAboveLastAndSendsRosterOnlyWhenAdmitted(t *testing.T) {
	h, now, a, b := twoAdmitted(t)
	ctx := context.Background()
	for i := uint64(1); i <= 5; i++ {
		h.Relay(testCtx, "r1", "sb", b.Peer, Message{Type: TypeCandidate, To: a.Peer, Seq: i, Gen: 1, Payload: json.RawMessage(`{}`)})
	}
	drain(a.Sub)
	h.Detach("r1", a.Peer, nil)
	*now = now.Add(5 * time.Second)
	// Two more from the same sender *while a is detached*: the point of
	// the ring is that a message sent into the window is kept, not just
	// one that was already delivered and re-read. Without these, the
	// replay below would be satisfied by the ring alone and would say
	// nothing about what happens during the thirty seconds a browser is
	// away (Task 3 carry-over).
	for i := uint64(6); i <= 7; i++ {
		if err := h.Relay(testCtx, "r1", "sb", b.Peer, Message{Type: TypeCandidate, To: a.Peer, Seq: i, Gen: 1, Payload: json.RawMessage(`{}`)}); err != nil {
			t.Fatalf("relay %d into the resume window: %v", i, err)
		}
	}
	res, err := h.Resume(testCtx, "r1", "sa", a.Peer, map[PeerID]uint64{b.Peer: 3})
	if err != nil {
		t.Fatal(err)
	}
	if !res.Admitted || res.Roster == nil {
		t.Fatalf("resume %+v", res)
	}
	var seqs []uint64
	for _, m := range res.Replay {
		seqs = append(seqs, m.Seq)
	}
	if len(seqs) != 4 || seqs[0] != 4 || seqs[1] != 5 || seqs[2] != 6 || seqs[3] != 7 {
		t.Fatalf("replay seqs %v, want 4 5 6 7 (everything above last, delivered and undelivered alike)", seqs)
	}
	if len(res.Gap) != 0 {
		t.Fatalf("a ring that held everything reported a gap: %v", res.Gap)
	}
	if res.Sub == nil || res.Sub.Lost() {
		t.Fatal("resume did not attach a fresh subscriber")
	}
	// The old window is over; the tick must not reap a participation
	// that has resumed.
	*now = now.Add(ResumeWindow + time.Second)
	h.Tick(ctx)
	if _, ok := h.Find("r1", "sa", a.Peer); !ok {
		t.Fatal("the tick reaped a participation that had resumed")
	}
	// A waiting guest resumes and learns nothing.
	g, _ := h.Open("r1", "sg", 100, "Gus", KindGuest)
	h.Join(ctx, "r1", "sg", g.Peer)
	h.Detach("r1", g.Peer, nil)
	gres, err := h.Resume(testCtx, "r1", "sg", g.Peer, nil)
	if err != nil {
		t.Fatal(err)
	}
	if gres.Admitted || gres.Roster != nil || len(gres.Replay) != 0 {
		t.Fatalf("waiting guest resume leaked %+v", gres)
	}
}

func TestResumeGapWhenTheRingRolledOver(t *testing.T) {
	h, now, a, b := twoAdmitted(t)
	for i := uint64(1); i <= 40; i++ {
		h.Relay(testCtx, "r1", "sb", b.Peer, Message{Type: TypeCandidate, To: a.Peer, Seq: i, Gen: 1})
	}
	drain(a.Sub)
	h.Detach("r1", a.Peer, nil)
	*now = now.Add(time.Second)
	res, _ := h.Resume(testCtx, "r1", "sa", a.Peer, map[PeerID]uint64{b.Peer: 2})
	if len(res.Gap) != 1 || res.Gap[0] != b.Peer || len(res.Replay) != 0 {
		t.Fatalf("expected a gap for b, got replay %d gap %v", len(res.Replay), res.Gap)
	}
}

// One ring is shared by every sender, so one peer's burst can evict
// another's messages wholesale. The gap must still be reported, which
// the ring's own oldest seq per sender cannot see.
func TestResumeGapWhenAnotherSendersBurstEvictedTheWholeSender(t *testing.T) {
	h, now, a, b := twoAdmitted(t)
	ctx := context.Background()
	c, _ := h.Open("r1", "sc", 3, "Cat", KindMember)
	h.Join(ctx, "r1", "sc", c.Peer)
	drain(a.Sub)
	for i := uint64(1); i <= 3; i++ {
		h.Relay(testCtx, "r1", "sb", b.Peer, Message{Type: TypeCandidate, To: a.Peer, Seq: i, Gen: 1})
	}
	for i := uint64(1); i <= RingSize; i++ {
		h.Relay(testCtx, "r1", "sc", c.Peer, Message{Type: TypeCandidate, To: a.Peer, Seq: i, Gen: 1})
	}
	drain(a.Sub)
	h.Detach("r1", a.Peer, nil)
	*now = now.Add(time.Second)
	res, err := h.Resume(testCtx, "r1", "sa", a.Peer, map[PeerID]uint64{b.Peer: 1, c.Peer: RingSize})
	if err != nil {
		t.Fatal(err)
	}
	if len(res.Gap) != 1 || res.Gap[0] != b.Peer {
		t.Fatalf("b's messages were evicted wholesale, gap %v", res.Gap)
	}
}

// The ring and the eviction marks outlive the peer that filled them.
// A resume must not name a peer that has left: the roster the browser
// reconciles against says it is gone, so a gap for it would ask for a
// reset with nobody, and its replayed signalling would rebuild the
// connection the roster has just told the browser to close.
func TestResumeNamesNoPeerThatHasLeft(t *testing.T) {
	h, now, a, b := twoAdmitted(t)
	for i := uint64(1); i <= 40; i++ {
		h.Relay(testCtx, "r1", "sb", b.Peer, Message{Type: TypeCandidate, To: a.Peer, Seq: i, Gen: 1})
	}
	p, _ := h.Find("r1", "sa", a.Peer)
	if p.evicted[b.Peer] == 0 {
		t.Fatal("the ring did not evict anything, so this proves nothing")
	}
	if err := h.Leave("r1", "sb", b.Peer); err != nil {
		t.Fatal(err)
	}
	drain(a.Sub)
	h.Detach("r1", a.Peer, nil)
	*now = now.Add(time.Second)
	res, err := h.Resume(testCtx, "r1", "sa", a.Peer, map[PeerID]uint64{b.Peer: 2})
	if err != nil {
		t.Fatal(err)
	}
	if len(res.Gap) != 0 {
		t.Fatalf("gap names a peer that left: %v", res.Gap)
	}
	if len(res.Replay) != 0 {
		t.Fatalf("replay carries a departed peer's signalling: %+v", res.Replay)
	}
}

// A second resume of an attached peer replaces the attachment: the old
// stream is told replaced, and the participation itself carries on.
func TestResumeReplacesAnExistingAttachment(t *testing.T) {
	h, _, a, b := twoAdmitted(t)
	res, err := h.Resume(testCtx, "r1", "sa", a.Peer, nil)
	if err != nil {
		t.Fatal(err)
	}
	old := drain(a.Sub)
	if len(old) != 1 || old[0].Type != TypeLost || old[0].Reason != ReasonReplaced {
		t.Fatalf("old stream got %+v", old)
	}
	if !a.Sub.Lost() {
		t.Fatal("the old subscriber was not closed")
	}
	if ev := drain(b.Sub); len(ev) != 0 {
		t.Fatalf("the room was told the peer left: %v", types(ev))
	}
	if !res.Admitted {
		t.Fatal("the replaced attachment lost its admission")
	}
	if err := h.Relay(testCtx, "r1", "sb", b.Peer, Message{Type: TypeOffer, To: a.Peer, Seq: 1, Gen: 1}); err != nil {
		t.Fatal(err)
	}
	if ev := drain(res.Sub); len(ev) != 1 || ev[0].Type != TypeOffer {
		t.Fatalf("the new attachment got %v", types(ev))
	}
	if _, err := h.Resume(testCtx, "r1", "sa", a.Peer, nil); err != nil {
		t.Fatalf("a replaced attachment left a tombstone: %v", err)
	}
}

func TestResumeAnswers(t *testing.T) {
	h, now, a, _ := twoAdmitted(t)
	if _, err := h.Resume(testCtx, "r1", "sb", a.Peer, nil); err == nil || err.(*ResumeError).Status != 404 || err.(*ResumeError).Boot != h.Boot() {
		t.Fatalf("resume through another session: %v", err)
	}
	if _, err := h.Resume(testCtx, "r9", "sa", a.Peer, nil); err == nil || err.(*ResumeError).Status != 404 {
		t.Fatalf("resume through another room: %v", err)
	}
	h.Takeover("r1", "sa")
	if _, err := h.Resume(testCtx, "r1", "sa", a.Peer, nil); err == nil || err.(*ResumeError).Status != 410 || err.(*ResumeError).Reason != ReasonReplaced {
		t.Fatalf("resume after takeover: %v", err)
	}
	*now = now.Add(TombstoneLife + time.Minute)
	h.Tick(context.Background())
	if _, err := h.Resume(testCtx, "r1", "sa", a.Peer, nil); err == nil || err.(*ResumeError).Status != 404 {
		t.Fatalf("resume after tombstone expiry: %v", err)
	}
	if _, err := h.Open("r1", "sa", 1, "Ann", KindMember); err != nil {
		t.Fatalf("fresh stream after tombstone: %v", err)
	}
}

// Why a participation ended is account state, and a peer id is in every
// roster the room-mates hold. A tombstone is therefore answered only
// through the room and session it was bound to; anyone else gets the
// same 404 an unknown peer gets (spec §4.2).
func TestATombstoneIsAnsweredOnlyThroughItsOwnRoomAndSession(t *testing.T) {
	h, _, a, b := twoAdmitted(t)
	if _, err := h.Open("r2", "sother", 9, "Other", KindMember); err != nil {
		t.Fatal(err)
	}
	if err := h.Remove(context.Background(), "r1", "sb", b.Peer, a.Peer); err != nil {
		t.Fatal(err)
	}

	// Ann's own session, in the room she was removed from.
	_, err := h.Resume(testCtx, "r1", "sa", a.Peer, nil)
	re, ok := err.(*ResumeError)
	if !ok || re.Status != 410 || re.Reason != ReasonRemoved {
		t.Fatalf("her own resume: %v", err)
	}
	if re := h.CanResume(testCtx, "r1", "sa", a.Peer); re == nil || re.Status != 410 || re.Reason != ReasonRemoved {
		t.Fatalf("her own probe: %v", re)
	}

	// Bob, who has her peer id from the roster, learns nothing.
	for _, probe := range []struct{ what, sqid, session string }{
		{"another session in the same room", "r1", "sb"},
		{"another room this session is in", "r2", "sother"},
		{"another room and another session", "r2", "sb"},
	} {
		_, err := h.Resume(testCtx, probe.sqid, probe.session, a.Peer, nil)
		re, ok := err.(*ResumeError)
		if !ok || re.Status != 404 || re.Boot != h.Boot() || re.Reason != "" {
			t.Fatalf("%s: %v", probe.what, err)
		}
		if re := h.CanResume(testCtx, probe.sqid, probe.session, a.Peer); re == nil || re.Status != 404 || re.Reason != "" {
			t.Fatalf("%s, probed: %+v", probe.what, re)
		}
	}
}

// Spec §15.3: a probe changes nothing. Not the participation, and not
// the kind either — recording a member's demotion to guest would send a
// fresh roster to every admitted peer in the room, which is a room-wide
// event caused by a request whose whole contract is that it has no
// effects (Codex S2-10). The transition is real, and is applied by the
// next thing this member actually does.
func TestAProbeNeverWritesAKindNorBroadcastsARoster(t *testing.T) {
	now := time.Date(2026, 9, 10, 9, 0, 0, 0, time.UTC)
	// Standing this test can move: a member taken off the team while
	// holding a live guest grant is still someone — ok stays true — and
	// their kind becomes guest. That is the branch a probe wrote through.
	kind := KindMember
	h := New(Config{
		Now:    func() time.Time { return now },
		Logger: slog.New(slog.NewTextHandler(io.Discard, nil)),
		StandingFor: func(ctx context.Context, room string, personID int64) (Kind, bool, error) {
			return kind, true, nil
		},
	})
	a, _ := h.Open("r1", "sa", 1, "Ann", KindMember)
	h.Join(testCtx, "r1", "sa", a.Peer)
	b, _ := h.Open("r1", "sb", 2, "Bob", KindMember)
	h.Join(testCtx, "r1", "sb", b.Peer)
	drain(a.Sub)
	drain(b.Sub)

	kind = KindGuest // off the team, grant still live

	if re := h.CanResume(testCtx, "r1", "sa", a.Peer); re != nil {
		t.Fatalf("the probe refused a participation that is still there: %+v", re)
	}
	p, ok := h.Find("r1", "sa", a.Peer)
	if !ok || p.Kind != KindMember {
		t.Fatalf("the probe wrote the kind: %v", p.Kind)
	}
	if ev := drain(a.Sub); len(ev) != 0 {
		t.Fatalf("the probe sent its own stream %v", types(ev))
	}
	if ev := drain(b.Sub); len(ev) != 0 {
		t.Fatalf("the probe sent the room %v", types(ev))
	}

	// The next mutating request applies it, and owes the room the roster
	// a kind change earns (§2.4).
	if err := h.Relay(testCtx, "r1", "sa", a.Peer, Message{Type: TypeOffer, To: b.Peer, Seq: 1, Gen: 1}); err != nil {
		t.Fatal(err)
	}
	if p, _ := h.Find("r1", "sa", a.Peer); p.Kind != KindGuest {
		t.Fatalf("kind after a mutating request: %v, want guest", p.Kind)
	}
	if ev := types(drain(a.Sub)); len(ev) != 1 || ev[0] != TypeRoster {
		t.Fatalf("the demoted member got %v, want one roster", ev)
	}
	if ev := types(drain(b.Sub)); len(ev) != 2 || ev[0] != TypeRoster || ev[1] != TypeOffer {
		t.Fatalf("the room got %v, want the roster then the offer", ev)
	}
}

func TestResumeAfterWindowIsA404WithTheSameBoot(t *testing.T) {
	h, now, a, _ := twoAdmitted(t)
	h.Detach("r1", a.Peer, nil)
	*now = now.Add(ResumeWindow + time.Second)
	h.Tick(context.Background())
	_, err := h.Resume(testCtx, "r1", "sa", a.Peer, nil)
	re, ok := err.(*ResumeError)
	if !ok || re.Status != 404 || re.Boot != h.Boot() {
		t.Fatalf("lapsed resume: %v", err)
	}
}

// Spec §4.5's window is thirty seconds, not thirty plus however long it
// is until the next tick: a resume that arrives after resume_until is
// refused on arrival (§15.7). The refusal itself still writes nothing —
// a probe changes nothing — but the fresh stream the browser opens next
// must not be refused 409 by the very participation it was just refused
// a resume of (Codex S2-2): a browser told "not here" and then "open in
// another tab" about the same peer has nowhere to go and shows a
// takeover screen for a tab that does not exist. Open ends it and takes
// its place.
func TestAResumePastItsWindowIsRefusedAndAFreshStreamTakesItsPlace(t *testing.T) {
	now := time.Date(2026, 9, 10, 9, 0, 0, 0, time.UTC)
	h := testHub(&now)
	a, _ := h.Open("r1", "sa", 1, "Ann", KindMember)
	h.Detach("r1", a.Peer, a.Sub)

	now = now.Add(ResumeWindow - time.Second)
	if re := h.CanResume(testCtx, "r1", "sa", a.Peer); re != nil {
		t.Fatalf("inside the window: %+v", re)
	}
	now = now.Add(2 * time.Second) // past resume_until, before any tick
	re := h.CanResume(testCtx, "r1", "sa", a.Peer)
	if re == nil || re.Status != 404 || re.Boot != h.Boot() {
		t.Fatalf("past the window: %+v", re)
	}
	if _, err := h.Resume(testCtx, "r1", "sa", a.Peer, nil); err == nil {
		t.Fatal("Resume attached a participation past its window")
	}
	// Refused, and the refusal itself removed nothing: a probe and a
	// refused resume are both defined as changing nothing.
	if _, ok := h.Find("r1", "sa", a.Peer); !ok {
		t.Fatal("the refusal removed the participation")
	}
	// Nor does the session still hold a peer here as far as anything
	// that asks is concerned — the holding screen's "open in another
	// tab" notice reads this.
	if _, live := h.LiveForSession("r1", "sa"); live {
		t.Fatal("a lapsed participation still counts as this session's live peer")
	}
	// The fresh stream, before any tick: it succeeds, with a new peer.
	fresh, err := h.Open("r1", "sa", 1, "Ann", KindMember)
	if err != nil {
		t.Fatalf("a fresh stream before the tick: %v", err)
	}
	if fresh.Peer == a.Peer || fresh.Peer == "" {
		t.Fatalf("fresh stream reused the lapsed peer: %q", fresh.Peer)
	}
	if _, ok := h.Find("r1", "sa", a.Peer); ok {
		t.Fatal("the lapsed participation survived the fresh stream")
	}
	if id, live := h.LiveForSession("r1", "sa"); !live || id != fresh.Peer {
		t.Fatalf("after the fresh stream the session holds %q (live=%v), want %q", id, live, fresh.Peer)
	}
	// And the tick that follows finds nothing left of it to remove.
	h.Tick(context.Background())
	if _, ok := h.Find("r1", "sa", fresh.Peer); !ok {
		t.Fatal("the tick removed the fresh participation")
	}
}

// Spec §4.6: a `pair` at or below the generation a browser already holds
// is ignored. The hub owns the half of that rule this package can gate —
// generations only ever rise, and only the newest is live — and the
// browser's half rests on it: two resets from one side yield 2 and 3, so
// the `pair` carrying 2 is superseded by the time it could be applied,
// and signalling that still carries 2 is refused. Ignoring a superseded
// `pair` is therefore safe rather than merely tidy.
func TestGenerationsOnlyRiseAndASupersededPairIsStale(t *testing.T) {
	h, _, a, b := twoAdmitted(t)
	for i := 0; i < 2; i++ {
		if err := h.Reset(testCtx, "r1", "sa", a.Peer, b.Peer); err != nil {
			t.Fatalf("reset %d: %v", i+1, err)
		}
	}
	for _, side := range []struct {
		name  string
		ms    []Message
		other PeerID
	}{
		{"a", drain(a.Sub), b.Peer},
		{"b", drain(b.Sub), a.Peer},
	} {
		if len(side.ms) != 2 {
			t.Fatalf("%s got %v, want one pair per reset", side.name, types(side.ms))
		}
		for i, want := range []uint64{2, 3} {
			m := side.ms[i]
			if m.Type != TypePair || m.Peer != side.other || m.Gen != want {
				t.Fatalf("%s pair %d: %+v, want pair with %s at gen %d", side.name, i, m, side.other, want)
			}
		}
	}

	// Gen 2 is exactly the superseded one: announced by the first reset,
	// replaced by the second before either side could finish with it.
	if err := h.Relay(testCtx, "r1", "sa", a.Peer, Message{Type: TypeOffer, To: b.Peer, Seq: 1, Gen: 2}); err != ErrStaleGen {
		t.Fatalf("signalling at the superseded generation: %v, want ErrStaleGen", err)
	}
	// A generation the hub never issued is refused too: only the current
	// one is live, from either direction.
	if err := h.Relay(testCtx, "r1", "sb", b.Peer, Message{Type: TypeOffer, To: a.Peer, Seq: 1, Gen: 4}); err != ErrStaleGen {
		t.Fatalf("signalling above the current generation: %v, want ErrStaleGen", err)
	}
	if err := h.Relay(testCtx, "r1", "sa", a.Peer, Message{Type: TypeOffer, To: b.Peer, Seq: 1, Gen: 3}); err != nil {
		t.Fatalf("signalling at the current generation: %v", err)
	}
}