package hub

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

// rosterMedia finds one peer's media in a roster, and whether the peer
// is in it at all.
func rosterMedia(ro *Roster, peer PeerID) (*MediaState, bool) {
	if ro == nil {
		return nil, false
	}
	for _, rp := range ro.Peers {
		if rp.Peer == peer {
			return rp.Media, true
		}
	}
	return nil, false
}

// TestSetMediaFansOutAChangeAndNothingElse is §15.19's delivery: every
// other admitted peer hears a change, in an envelope the hub built, and
// nobody else hears anything — not the sender, not a waiting guest, and
// nothing in any ring. A repeat of the same state is accepted and silent.
func TestSetMediaFansOutAChangeAndNothingElse(t *testing.T) {
	now := time.Date(2026, 9, 23, 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)
	g, _ := h.Open("r1", "sg", 100, "Gus", KindGuest)
	h.Join(ctx, "r1", "sg", g.Peer) // waiting
	drain(a.Sub)
	drain(b.Sub)
	drain(g.Sub)

	if err := h.SetMedia(ctx, "r1", "sa", a.Peer, MediaState{Muted: true}); err != nil {
		t.Fatal(err)
	}
	got := drain(b.Sub)
	if len(got) != 1 {
		t.Fatalf("Bob got %d events, want 1: %v", len(got), types(got))
	}
	m := got[0]
	if m.Type != TypeMedia || m.From != a.Peer || m.Name != "Ann" {
		t.Fatalf("Bob got %+v, want media from Ann", m)
	}
	if string(m.Payload) != `{"muted":true,"cameraOff":false}` {
		t.Fatalf("payload %s, want the canonical two booleans", m.Payload)
	}
	if m.To != "" || m.Seq != 0 || m.Gen != 0 || m.Roster != nil {
		t.Fatalf("Bob got %+v, want nothing but type, from, name and payload", m)
	}
	if n := len(drain(a.Sub)); n != 0 {
		t.Fatalf("the sender must not hear its own report, got %d", n)
	}
	if n := len(drain(g.Sub)); n != 0 {
		t.Fatalf("a waiting guest must hear nothing, got %d", n)
	}
	if n := ringLen(h, "r1", "sb", b.Peer); n != 0 {
		t.Fatalf("media is state, never ringed; ring has %d", n)
	}

	// The same state again: accepted, and the room hears nothing.
	if err := h.SetMedia(ctx, "r1", "sa", a.Peer, MediaState{Muted: true}); err != nil {
		t.Fatal(err)
	}
	if n := len(drain(b.Sub)); n != 0 {
		t.Fatalf("a report that changes nothing must send nothing, got %d", n)
	}
	// A different state is a change again.
	if err := h.SetMedia(ctx, "r1", "sa", a.Peer, MediaState{Muted: true, CameraOff: true}); err != nil {
		t.Fatal(err)
	}
	got = drain(b.Sub)
	if len(got) != 1 || string(got[0].Payload) != `{"muted":true,"cameraOff":true}` {
		t.Fatalf("camera off should be one more media event, got %v", got)
	}
}

// TestRostersCarryMediaAndUnknownIsAbsent is the "state, not event"
// half: a peer admitted after the report learns it from its roster, the
// room learns a newcomer's from the peer event once it has one, and a
// peer that has never reported — every client older than §15.19 — has
// no media at all rather than a guessed false.
func TestRostersCarryMediaAndUnknownIsAbsent(t *testing.T) {
	now := time.Date(2026, 9, 23, 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) // never reports: an old client
	if err := h.SetMedia(ctx, "r1", "sa", a.Peer, MediaState{Muted: true}); err != nil {
		t.Fatal(err)
	}
	drain(a.Sub)
	drain(b.Sub)

	c, _ := h.Open("r1", "sc", 3, "Cy", KindMember)
	h.Join(ctx, "r1", "sc", c.Peer)
	cev := drain(c.Sub)
	var ro *Roster
	for _, m := range cev {
		if m.Type == TypeRoster {
			ro = m.Roster
		}
	}
	am, ok := rosterMedia(ro, a.Peer)
	if !ok || am == nil || !am.Muted || am.CameraOff {
		t.Fatalf("Cy's roster should say Ann is muted, got %+v (present %v)", am, ok)
	}
	bm, ok := rosterMedia(ro, b.Peer)
	if !ok || bm != nil {
		t.Fatalf("Bob never reported, so his roster entry must carry no media, got %+v (present %v)", bm, ok)
	}
	// And on the wire, absent rather than a zero object: that is what
	// an old client ignores and a new one reads as unknown.
	raw, _ := json.Marshal(RosterPeer{Peer: b.Peer, Name: "Bob", Kind: "member"})
	if string(raw) != `{"peer":"`+string(b.Peer)+`","name":"Bob","kind":"member"}` {
		t.Fatalf("an unreported peer marshals as %s, want no media key", raw)
	}

	// The newcomer's own first report reaches everyone already there,
	// the old client included: it ignores the type, and it is not the
	// hub's business to know that.
	drain(a.Sub)
	drain(b.Sub)
	if err := h.SetMedia(ctx, "r1", "sc", c.Peer, MediaState{CameraOff: true}); err != nil {
		t.Fatal(err)
	}
	for _, s := range []*Sub{a.Sub, b.Sub} {
		ev := drain(s)
		if len(ev) != 1 || ev[0].Type != TypeMedia || ev[0].From != c.Peer {
			t.Fatalf("Cy's report should reach both others once, got %v", types(ev))
		}
	}
}

// TestPeerEventCarriesMediaFromTheParticipation: the `peer` event is
// built from the participation, like the roster, so whatever media the
// participation holds at admission goes with it. Today a report needs
// admission first, so this is nil in practice — the test pins the
// contract so that loosening that rule later cannot forget the event.
func TestPeerEventCarriesMediaFromTheParticipation(t *testing.T) {
	now := time.Date(2026, 9, 23, 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)
	drain(a.Sub)
	b, _ := h.Open("r1", "sb", 2, "Bob", KindMember)
	// Bob's media is set directly under the lock, as if he had reported
	// before this admission; the hub's own paths only allow a report once
	// admitted, so this is the peer event's contract, not a flow.
	h.mu.Lock()
	p, _ := h.findLocked("r1", "sb", b.Peer)
	p.Media = &MediaState{Muted: true}
	h.mu.Unlock()
	h.Join(ctx, "r1", "sb", b.Peer)
	ev := drain(a.Sub)
	if len(ev) != 1 || ev[0].Type != TypePeer {
		t.Fatalf("Ann should get one peer event, got %v", types(ev))
	}
	bm, ok := rosterMedia(ev[0].Roster, b.Peer)
	if !ok || bm == nil || !bm.Muted {
		t.Fatalf("the peer event should carry Bob's media, got %+v", bm)
	}
}

// TestResumeRosterCarriesMediaMissedWhileDetached: a change made while a
// peer's stream is down is not ringed and not replayed, and it does not
// need to be — the roster the resume is answered with says it.
func TestResumeRosterCarriesMediaMissedWhileDetached(t *testing.T) {
	now := time.Date(2026, 9, 23, 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)
	if err := h.SetMedia(ctx, "r1", "sb", b.Peer, MediaState{Muted: true}); err != nil {
		t.Fatal(err)
	}
	drain(a.Sub)
	drain(b.Sub)

	h.Detach("r1", b.Peer, b.Sub)
	if err := h.SetMedia(ctx, "r1", "sa", a.Peer, MediaState{Muted: true}); err != nil {
		t.Fatal(err)
	}
	if n := ringLen(h, "r1", "sb", b.Peer); n != 0 {
		t.Fatalf("a media report to a detached peer must not be ringed, ring has %d", n)
	}
	res, err := h.Resume(ctx, "r1", "sb", b.Peer, map[PeerID]uint64{})
	if err != nil {
		t.Fatal(err)
	}
	if len(res.Replay) != 0 || len(res.Gap) != 0 {
		t.Fatalf("nothing to replay and no gap, got %d replayed and %d gaps", len(res.Replay), len(res.Gap))
	}
	am, _ := rosterMedia(res.Roster, a.Peer)
	if am == nil || !am.Muted {
		t.Fatalf("the resume roster should say Ann is muted, got %+v", am)
	}
	// Bob's own state survived his detach: the participation holds it.
	bm, _ := rosterMedia(res.Roster, b.Peer)
	if bm == nil || !bm.Muted {
		t.Fatalf("Bob's own report should survive his resume, got %+v", bm)
	}

	// A leave ends the participation, and with it the state: a fresh
	// join starts unknown.
	if err := h.Leave("r1", "sb", b.Peer); err != nil {
		t.Fatal(err)
	}
	b2, _ := h.Open("r1", "sb", 2, "Bob", KindMember)
	h.Join(ctx, "r1", "sb", b2.Peer)
	ev := drain(a.Sub)
	var fresh *Roster
	for _, m := range ev {
		if m.Type == TypePeer && m.Peer == b2.Peer {
			fresh = m.Roster
		}
	}
	if bm, ok := rosterMedia(fresh, b2.Peer); !ok || bm != nil {
		t.Fatalf("a rejoin must start with no media, got %+v (present %v)", bm, ok)
	}
}

// TestSetMediaAuthority is Fanout's authority, reused: bound to this
// room and session, admitted, and a member's standing re-derived.
func TestSetMediaAuthority(t *testing.T) {
	now := time.Date(2026, 9, 23, 9, 0, 0, 0, time.UTC)
	standing := map[int64]bool{1: true, 2: true, 100: true}
	h := New(Config{
		Now:    func() time.Time { return now },
		Logger: slog.New(slog.NewTextHandler(io.Discard, nil)),
		StandingFor: func(ctx context.Context, room string, id int64) (Kind, bool, error) {
			if id >= 100 {
				return KindGuest, standing[id], nil
			}
			return KindMember, standing[id], nil
		},
	})
	ctx := context.Background()
	m := MediaState{Muted: true}
	a, _ := h.Open("r1", "sa", 1, "Ann", KindMember)
	if err := h.SetMedia(ctx, "r1", "sa", a.Peer, m); !errors.Is(err, ErrNotAdmitted) {
		t.Fatalf("unjoined: %v", err)
	}
	h.Join(ctx, "r1", "sa", a.Peer)
	if err := h.SetMedia(ctx, "r1", "sb", a.Peer, m); !errors.Is(err, ErrUnknownPeer) {
		t.Fatalf("another session: %v", err)
	}
	if err := h.SetMedia(ctx, "r2", "sa", a.Peer, m); !errors.Is(err, ErrUnknownPeer) {
		t.Fatalf("another room: %v", err)
	}
	g, _ := h.Open("r1", "sg", 100, "Gus", KindGuest)
	h.Join(ctx, "r1", "sg", g.Peer)
	if err := h.SetMedia(ctx, "r1", "sg", g.Peer, m); !errors.Is(err, ErrNotAdmitted) {
		t.Fatalf("waiting guest: %v", err)
	}
	b, _ := h.Open("r1", "sb", 2, "Bob", KindMember)
	h.Join(ctx, "r1", "sb", b.Peer)
	drain(a.Sub)
	drain(b.Sub)

	standing[1] = false
	if err := h.SetMedia(ctx, "r1", "sa", a.Peer, m); !errors.Is(err, ErrNotYours) {
		t.Fatalf("member whose standing is gone: %v", err)
	}
	ev := drain(b.Sub)
	if len(ev) != 1 || ev[0].Type != TypeGone {
		t.Fatalf("Bob should see Ann gone and no media, got %v", types(ev))
	}
}

// TestSetMediaUnderConcurrency is for `make race`: reports, resumes,
// rosters and a tick all at once, over the same participations. Media
// is a pointer handed to rosters that are serialised on other
// goroutines, so a write through it anywhere would show up here.
func TestSetMediaUnderConcurrency(t *testing.T) {
	now := time.Date(2026, 9, 23, 9, 0, 0, 0, time.UTC)
	var clock sync.Mutex
	h := New(Config{
		Now: func() time.Time {
			clock.Lock()
			defer clock.Unlock()
			return now
		},
		Logger: slog.New(slog.NewTextHandler(io.Discard, nil)),
		StandingFor: func(ctx context.Context, room string, id int64) (Kind, bool, error) {
			return KindMember, true, nil
		},
	})
	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)

	var wg sync.WaitGroup
	stop := make(chan struct{})
	// Readers: marshal whatever arrives, the way the stream handler does.
	for _, s := range []*Sub{a.Sub, b.Sub} {
		wg.Add(1)
		go func(s *Sub) {
			defer wg.Done()
			for {
				select {
				case <-stop:
					return
				case m, ok := <-s.C():
					if !ok {
						return
					}
					_, _ = json.Marshal(m)
				}
			}
		}(s)
	}
	var writers sync.WaitGroup
	for i, who := range []struct {
		session string
		peer    PeerID
	}{{"sa", a.Peer}, {"sb", b.Peer}} {
		writers.Add(1)
		go func(i int, session string, peer PeerID) {
			defer writers.Done()
			for n := 0; n < 200; n++ {
				_ = h.SetMedia(ctx, "r1", session, peer, MediaState{Muted: n%2 == 0, CameraOff: (n+i)%3 == 0})
			}
		}(i, who.session, who.peer)
	}
	writers.Add(1)
	go func() {
		defer writers.Done()
		for n := 0; n < 50; n++ {
			h.Tick(ctx)
			c, err := h.Open("r1", "sc", 3, "Cy", KindMember)
			if err == nil {
				_ = h.Join(ctx, "r1", "sc", c.Peer)
				_, _ = json.Marshal(drain(c.Sub))
				_ = h.Leave("r1", "sc", c.Peer)
			}
		}
	}()
	writers.Wait()
	close(stop)
	wg.Wait()
}

// TestSetMediaCarriesTheReceivingHalf is §15.20 in the hub: the report's
// optional members are part of the state — a change to either is fanned
// out, the same list again is silent, and the hub keeps its own copy of
// the list so a caller that reuses its slice cannot rewrite a roster
// already sent.
func TestSetMediaCarriesTheReceivingHalf(t *testing.T) {
	now := time.Date(2026, 9, 25, 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)

	list := []PeerID{b.Peer}
	if err := h.SetMedia(ctx, "r1", "sa", a.Peer, MediaState{AudioBlocked: true, NoSoundFrom: list}); err != nil {
		t.Fatal(err)
	}
	got := drain(b.Sub)
	if len(got) != 1 {
		t.Fatalf("Bob got %d events, want 1", len(got))
	}
	want := `{"muted":false,"cameraOff":false,"audioBlocked":true,"noSoundFrom":["` + string(b.Peer) + `"]}`
	if string(got[0].Payload) != want {
		t.Fatalf("payload %s, want %s", got[0].Payload, want)
	}
	list[0] = "rewritten"
	c, _ := h.Open("r1", "sc", 3, "Cy", KindMember)
	h.Join(ctx, "r1", "sc", c.Peer)
	var ro *Roster
	for _, ev := range drain(c.Sub) {
		if ev.Type == TypeRoster {
			ro = ev.Roster
		}
	}
	drain(b.Sub)
	m, _ := rosterMedia(ro, a.Peer)
	if m == nil || len(m.NoSoundFrom) != 1 || m.NoSoundFrom[0] != b.Peer {
		t.Fatalf("roster media %+v, want the hub's own copy of the list", m)
	}

	if err := h.SetMedia(ctx, "r1", "sa", a.Peer, MediaState{AudioBlocked: true, NoSoundFrom: []PeerID{b.Peer}}); err != nil {
		t.Fatal(err)
	}
	if got := drain(b.Sub); len(got) != 0 {
		t.Fatalf("the same report again reached Bob: %v", types(got))
	}
	if err := h.SetMedia(ctx, "r1", "sa", a.Peer, MediaState{}); err != nil {
		t.Fatal(err)
	}
	got = drain(b.Sub)
	if len(got) != 1 || string(got[0].Payload) != `{"muted":false,"cameraOff":false}` {
		t.Fatalf("clearing the bad news sent %v", got)
	}
}

// §15.24: hearing is kept with its emptiness — [] is "nobody yet", nil is
// a client that does not say — carried after the other lists, copied so
// nothing the caller holds is shared with a roster, and a change between
// nil and [] is a change.
func TestSetMediaCarriesHearing(t *testing.T) {
	now := time.Date(2026, 10, 1, 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)

	if err := h.SetMedia(ctx, "r1", "sa", a.Peer, MediaState{Hearing: []PeerID{}}); err != nil {
		t.Fatal(err)
	}
	got := drain(b.Sub)
	if len(got) != 1 || string(got[0].Payload) != `{"muted":false,"cameraOff":false,"hearing":[]}` {
		t.Fatalf("empty hearing sent %v", got)
	}
	list := []PeerID{b.Peer}
	if err := h.SetMedia(ctx, "r1", "sa", a.Peer, MediaState{ScreenStalled: list, Hearing: list}); err != nil {
		t.Fatal(err)
	}
	got = drain(b.Sub)
	want := `{"muted":false,"cameraOff":false,"screenStalled":["` + string(b.Peer) + `"],"hearing":["` + string(b.Peer) + `"]}`
	if len(got) != 1 || string(got[0].Payload) != want {
		t.Fatalf("payload %v, want %s", got, want)
	}
	list[0] = "rewritten"
	c, _ := h.Open("r1", "sc", 3, "Cy", KindMember)
	h.Join(ctx, "r1", "sc", c.Peer)
	var ro *Roster
	for _, ev := range drain(c.Sub) {
		if ev.Type == TypeRoster {
			ro = ev.Roster
		}
	}
	drain(b.Sub)
	m, _ := rosterMedia(ro, a.Peer)
	if m == nil || len(m.Hearing) != 1 || m.Hearing[0] != b.Peer {
		t.Fatalf("roster media %+v, want the hub's own copy of the list", m)
	}
	// Back to []: a change. Then nil: a change again, and the two-key report.
	if err := h.SetMedia(ctx, "r1", "sa", a.Peer, MediaState{Hearing: []PeerID{}}); err != nil {
		t.Fatal(err)
	}
	if got := drain(b.Sub); len(got) != 1 {
		t.Fatalf("back to empty sent %d", len(got))
	}
	if err := h.SetMedia(ctx, "r1", "sa", a.Peer, MediaState{}); err != nil {
		t.Fatal(err)
	}
	got = drain(b.Sub)
	if len(got) != 1 || string(got[0].Payload) != `{"muted":false,"cameraOff":false}` {
		t.Fatalf("nil after [] sent %v", got)
	}
	if err := h.SetMedia(ctx, "r1", "sa", a.Peer, MediaState{}); err != nil {
		t.Fatal(err)
	}
	if got := drain(b.Sub); len(got) != 0 {
		t.Fatalf("the same report again reached Bob: %v", types(got))
	}
	var round MediaState
	if err := json.Unmarshal([]byte(`{"muted":false,"cameraOff":false,"hearing":[]}`), &round); err != nil || round.Hearing == nil {
		t.Fatalf("[] must decode as an empty, non-nil list: %+v %v", round, err)
	}
}