// Spectator fan-out (WHOS-WATCHING-NOW-SPEC.md, sections 5.3 and 13.4).
//
// A host's seat relay exposes one loopback "tap": a copy of every envelope
// the host receives (frame, resource, recovery planes), the Opus audio
// packets, and a few text markers. This file dials that tap once per seat
// and fans it out to any number of read-only WebTransport sessions:
//
//	binary  [u8 tag][envelope]   tag 1 frame, 2 resource, 3 recovery,
//	                             4 audio ([u32 seq][u32 pts_ms][opus])
//	text    {"t":"spectate","token":..} the token spectators join with
//	        {"t":"epoch","id":n} a complete state description starts here
//	        {"t":"reset"}        the host stream restarted its numbering
//	        {"t":"parked",...}   the host parked/resumed the engine
//	        {"t":"sessionEnded","reason":...} / {"t":"engineStalled"}
//
// The bridge taps every host session it serves from the session's first
// byte and feeds a seatMirror (mirror.go) with it. A late join is then
// served from memory - snapshot, delta, dictionary keyframe - and the
// spectator follows the live tap from the next geometry anchor on. The
// host's planes, governors and engine are never involved. Only a tap that
// never saw an epoch (a host the bridge does not serve, a bridge restarted
// mid-session) falls back to the relay's in-band epoch: the bridge asks
// with {"t":"join"} and attaches the waiting spectators at the marker.
//
// Nothing a spectator does reaches the relay's governors: tap acks mean
// "consumed from loopback", every spectator has its own frame writer
// (latest-wins, stuck-stream reset) and its own byte-capped reliable
// queues, and a spectator that cannot keep up is dropped, never waited for.
package main

import (
	"context"
	"encoding/json"
	"fmt"
	"log"
	"net"
	"os"
	"strconv"
	"strings"
	"sync"
	"time"

	"github.com/coder/websocket"
	"war3replays/w3cs-wt-bridge/webtransport"
)

const (
	tagAudio = 4

	// Token-dialed taps (no host session of our own) close this long
	// after the last spectator leaves; host-session taps live as long as
	// the host session and only drop their snapshot cache when idle.
	tapLinger   = 30 * time.Second
	cacheLinger = 60 * time.Second
	// Epoch-fallback requests are coalesced per seat.
	spectatorJoinInterval = 10 * time.Second
	// Frames (and audio) start once the join bundle has mostly left the
	// reliable queue, so they never compete with it for the link.
	frameGateBacklog = 1024 * 1024
	tapDialRetry     = 2 * time.Second
	tapDialAttempts  = 30
)

func envInt(name string, fallback int) int {
	if raw := os.Getenv(name); raw != "" {
		if value, err := strconv.Atoi(raw); err == nil && value >= 0 {
			return value
		}
	}
	return fallback
}

// 0 = unlimited (the default: spectators cost bandwidth, not game CPU).
var (
	spectatorsPerSeat   = envInt("W3CS_SPECTATORS_PER_SEAT", 0)
	spectatorsPerBox    = envInt("W3CS_SPECTATORS_PER_BOX", 0)
	spectatorsPerIP     = envInt("W3CS_SPECTATORS_PER_IP", 0)
	spectatorQueueBytes = envInt("W3CS_SPECTATOR_QUEUE_MB", 256) * 1024 * 1024
)

// spectatorSink is what the tap dispatches into; the QUIC-backed
// spectator implements it and the tests use a recording fake.
type spectatorSink interface {
	enqueueFrame(msg frameMsg)
	enqueueReliable(tag byte, data []byte) bool
	reliableBacklog() int
	sendAudio(data []byte)
	control(object any)
	end(reason string)
}

type gateMode int

const (
	// gatePending: waiting to be served (from the mirror, or at the next
	// epoch marker).
	gatePending gateMode = iota
	// gateEpoch: legacy in-band join - attached at the epoch marker, frames
	// from that epoch's first keyframe on.
	gateEpoch
	// gateMirror: served from the mirror - reliable planes flow at once,
	// frames from the first anchor after the bundle has left the queue.
	gateMirror
)

// gate is the per-spectator join state machine.
type gate struct {
	mode         gateMode
	attached     bool
	framesOpen   bool
	keyframeSeen bool
	joinedAt     time.Time
}

// admit reports whether one tap message should reach this spectator and
// advances the gate. backlog is the spectator's reliable queue occupancy.
// Pure, so the tests can drive it directly.
func (g *gate) admit(tag byte, msg frameMsg, backlog int) bool {
	if !g.attached {
		return false
	}
	switch tag {
	case tagResource:
		return true
	case tagRecovery:
		if g.mode == gateEpoch {
			if !g.keyframeSeen && msg.newFrame && msg.keyframe {
				g.keyframeSeen = true
			}
			return g.keyframeSeen
		}
		return true
	case tagFrame, tagAudio:
		if g.mode == gateEpoch {
			return g.keyframeSeen
		}
		if !g.framesOpen && tag == tagFrame && msg.newFrame &&
			(msg.keyframe || msg.anchor) && backlog < frameGateBacklog {
			g.framesOpen = true
		}
		return g.framesOpen
	}
	return false
}

type seatTap struct {
	seat      int
	token     string
	sessionID string // the bridge's own host session, "" for a token dial
	mirror    *seatMirror

	mu         sync.Mutex
	sinks      map[spectatorSink]*gate
	plane      *relayPlane
	cancel     context.CancelFunc
	closeTimer *time.Timer
	lastJoin   time.Time
	closed     bool
	wake       chan struct{}

	frames, reliable, dropped, joinsMirror, joinsEpoch uint64
	// Frames the TAP carried, before any gate: seen counts new frames,
	// anchors counts those a waiting spectator could open its gate on.
	// Without these, "tap-frames=0" cannot distinguish a silent host from
	// a gate that never opened.
	framesSeen, anchorsSeen uint64
}

var (
	tapsMu sync.Mutex
	taps   = make(map[int]*seatTap)
)

func spectatorTotal() int {
	tapsMu.Lock()
	defer tapsMu.Unlock()
	total := 0
	for _, tap := range taps {
		tap.mu.Lock()
		total += len(tap.sinks)
		tap.mu.Unlock()
	}
	return total
}

func spectatorsFromIP(remote string) int {
	tapsMu.Lock()
	defer tapsMu.Unlock()
	count := 0
	for _, tap := range taps {
		tap.mu.Lock()
		for sink := range tap.sinks {
			if spec, ok := sink.(*spectator); ok && spec.remoteIP == remote {
				count++
			}
		}
		tap.mu.Unlock()
	}
	return count
}

func newSeatTap(seat int, token, sessionID string, plane *relayPlane,
	cancel context.CancelFunc) *seatTap {
	return &seatTap{seat: seat, token: token, sessionID: sessionID,
		mirror: newSeatMirror(seat, relayBundleFetcher(seat)),
		sinks:  make(map[spectatorSink]*gate), plane: plane, cancel: cancel,
		wake: make(chan struct{}, 1)}
}

func (t *seatTap) start(ctx context.Context) {
	go t.run(ctx)
	go t.serve(ctx)
	go t.report(ctx)
}

// hostTap keeps a tap on the bridge's own host session for as long as
// that session lives, so the mirror sees the stream from its bootstrap.
// Started by serveSession right after the planes attach.
func hostTap(ctx context.Context, seat int, sessionID string) {
	if sessionID == "" {
		return
	}
	base := fmt.Sprintf("ws://%s:%d", *relayHost, *relayBase+seat)
	failures := 0
	for ctx.Err() == nil && failures < tapDialAttempts {
		tapCtx, cancel := context.WithCancel(ctx)
		plane, err := dialRelay(tapCtx, fmt.Sprintf("%s/tap?session=%s", base,
			sessionID))
		if err != nil {
			cancel()
			failures++
			select {
			case <-ctx.Done():
				return
			case <-time.After(tapDialRetry):
			}
			continue
		}
		failures = 0
		tap := newSeatTap(seat, "", sessionID, plane, cancel)
		tapsMu.Lock()
		if old := taps[seat]; old != nil {
			old.shutdown("stream-ended")
		}
		taps[seat] = tap
		tapsMu.Unlock()
		log.Printf("seat %d host tap attached (session %s)", seat, sessionID)
		tap.start(tapCtx)
		<-tapCtx.Done()
		// The relay closed the tap (host left, tap replaced, relay gone).
		// While our host session still lives, come back for it.
		select {
		case <-ctx.Done():
			return
		case <-time.After(tapDialRetry):
		}
	}
}

// tapFor returns the seat's tap for a spectator's token, dialing one by
// token when the bridge serves no host session on that seat. A host tap
// whose token differs means the token is stale (a newer host session).
func tapFor(seat int, token string) (*seatTap, error) {
	tapsMu.Lock()
	defer tapsMu.Unlock()
	if tap := taps[seat]; tap != nil {
		tap.mu.Lock()
		same := tap.token == token && !tap.closed
		empty := len(tap.sinks) == 0
		host := tap.sessionID != ""
		tap.mu.Unlock()
		if same {
			return tap, nil
		}
		if host || !empty {
			return nil, fmt.Errorf("stream-ended")
		}
		tap.shutdown("stream-ended")
		delete(taps, seat)
	}
	ctx, cancel := context.WithCancel(context.Background())
	base := fmt.Sprintf("ws://%s:%d", *relayHost, *relayBase+seat)
	plane, err := dialRelay(ctx, fmt.Sprintf("%s/tap?spectate=%s", base, token))
	if err != nil {
		cancel()
		return nil, fmt.Errorf("stream-ended")
	}
	tap := newSeatTap(seat, token, "", plane, cancel)
	taps[seat] = tap
	tap.start(ctx)
	return tap, nil
}

func (t *seatTap) add(sink spectatorSink) {
	t.mu.Lock()
	t.sinks[sink] = &gate{mode: gatePending}
	if t.closeTimer != nil {
		t.closeTimer.Stop()
		t.closeTimer = nil
	}
	count := len(t.sinks)
	t.mu.Unlock()
	t.announce(count)
	t.kick()
}

func (t *seatTap) remove(sink spectatorSink) {
	t.mu.Lock()
	delete(t.sinks, sink)
	count := len(t.sinks)
	if count == 0 && t.closeTimer == nil && !t.closed {
		linger := tapLinger
		if t.sessionID != "" {
			linger = cacheLinger
		}
		t.closeTimer = time.AfterFunc(linger, func() {
			t.mu.Lock()
			idle := len(t.sinks) == 0
			t.closeTimer = nil
			t.mu.Unlock()
			if !idle {
				return
			}
			if t.sessionID != "" {
				// The host tap stays (the mirror must keep following);
				// only the serialized snapshot is released.
				t.mirror.dropCache()
				return
			}
			tapsMu.Lock()
			if taps[t.seat] == t {
				delete(taps, t.seat)
			}
			tapsMu.Unlock()
			t.shutdown("idle")
		})
	}
	t.mu.Unlock()
	t.announce(count)
}

func (t *seatTap) announce(count int) {
	t.sendText(map[string]any{"t": "spectators", "count": count})
}

// kick wakes the serving goroutine (coalesced).
func (t *seatTap) kick() {
	select {
	case t.wake <- struct{}{}:
	default:
	}
}

// join asks the relay for an in-band epoch - the fallback when the mirror
// has nothing to serve from. Coalesced per seat.
func (t *seatTap) join() {
	t.mu.Lock()
	if time.Since(t.lastJoin) < spectatorJoinInterval {
		t.mu.Unlock()
		return
	}
	t.lastJoin = time.Now()
	t.mu.Unlock()
	t.sendText(map[string]any{"t": "join"})
}

func (t *seatTap) sendText(object any) {
	payload, err := json.Marshal(object)
	if err != nil || t.plane == nil {
		return
	}
	ctx, done := context.WithTimeout(context.Background(), 5*time.Second)
	defer done()
	_ = t.plane.conn.Write(ctx, websocket.MessageText, payload)
}

// endReason names why a finished tap stopped. Pure, so the tests can drive
// it directly.
func (t *seatTap) endReason(ctx context.Context) string {
	if t.sessionID != "" && ctx.Err() != nil {
		return "host-left"
	}
	return "stream-ended"
}

func (t *seatTap) run(ctx context.Context) {
	// Why the tap ended decides what every spectator sees. "stream-ended"
	// means "the fan-out lost its source, rejoin" and the page reloads;
	// "host-left" is terminal and shows the ended card. A host tap whose
	// own context is done ended WITH the bridge's host session - the host
	// is gone, and a rejoin would chase a seat nobody holds. The relay's
	// {"t":"sessionEnded"} marker says the same thing, but the host's
	// signaling socket and its QUIC session die together when a tab
	// closes, so the marker loses that race about half the time (seen in
	// the spectate canary: spectators reloaded instead of ending).
	defer func() { t.shutdown(t.endReason(ctx)) }()
	for {
		kind, data, err := t.plane.conn.Read(ctx)
		if err != nil {
			return
		}
		switch kind {
		case websocket.MessageBinary:
			t.handleBinary(data)
			t.plane.ack(len(data))
		case websocket.MessageText:
			t.handleText(data)
		}
	}
}

// serve hands pending spectators their join bundle. Runs outside the tap
// lock because a bundle may need a snapshot build first.
func (t *seatTap) serve(ctx context.Context) {
	ticker := time.NewTicker(time.Second)
	defer ticker.Stop()
	for {
		select {
		case <-ctx.Done():
			return
		case <-t.wake:
		case <-ticker.C:
		}
		t.servePending()
	}
}

func (t *seatTap) pendingSinks() []spectatorSink {
	t.mu.Lock()
	defer t.mu.Unlock()
	var pending []spectatorSink
	for sink, g := range t.sinks {
		if g.mode == gatePending {
			pending = append(pending, sink)
		}
	}
	return pending
}

func (t *seatTap) servePending() {
	pending := t.pendingSinks()
	if len(pending) == 0 {
		return
	}
	if t.mirror.ready() {
		if t.mirror.keyframePending() {
			return // the keyframe's last fragment wakes us again
		}
		if err := t.mirror.ensureSnapshot(); err == nil {
			t.attachFromMirror()
			return
		}
	}
	// Fallback: an in-band epoch. A state description already streaming
	// makes the mirror ready shortly; otherwise ask for one and attach the
	// waiting spectators live at its marker.
	if t.mirror.epochOpen() {
		return
	}
	t.mu.Lock()
	for _, g := range t.sinks {
		if g.mode == gatePending {
			g.mode = gateEpoch
		}
	}
	t.mu.Unlock()
	t.join()
}

// attachFromMirror serves every pending spectator from the prepared
// snapshot. Under the tap lock, so no tap message can fall between the
// delta copied into the bundle and the live stream that follows it.
func (t *seatTap) attachFromMirror() {
	t.mu.Lock()
	defer t.mu.Unlock()
	for sink, g := range t.sinks {
		if g.mode != gatePending {
			continue
		}
		bundle, err := t.mirror.takeBundle()
		if err != nil {
			return // stale cache or keyframe mid-flight: next wake retries
		}
		if spec, ok := sink.(*spectator); ok {
			spec.raiseLimits(bundle.bytes)
		}
		ok := true
		for _, fragment := range bundle.resource {
			if !sink.enqueueReliable(tagResource, fragment) {
				ok = false
				break
			}
		}
		for _, fragment := range bundle.recovery {
			if !ok || !sink.enqueueReliable(tagRecovery, fragment) {
				ok = false
				break
			}
		}
		if !ok {
			delete(t.sinks, sink)
			go sink.end("too-slow")
			continue
		}
		g.mode = gateMirror
		g.attached = true
		g.framesOpen = false
		g.keyframeSeen = false
		g.joinedAt = time.Now()
		t.joinsMirror++
		sink.control(map[string]any{"t": "joined", "mode": "mirror",
			"bytes": bundle.bytes, "keyframe": bundle.keyframe})
		log.Printf("seat %d spectator joined from mirror: %.1f MiB "+
			"(delta %d msgs) keyframe=%d s0=%d", t.seat,
			float64(bundle.bytes)/1048576.0, bundle.delta, bundle.keyframe,
			bundle.s0)
	}
}

// handleBinary feeds the mirror and fans one tagged tap message out
// through every gate.
func (t *seatTap) handleBinary(data []byte) {
	if len(data) < 2 {
		return
	}
	tag, payload := data[0], data[1:]
	var msg frameMsg
	if tag == tagFrame || tag == tagRecovery {
		msg = parseFrameMsg(payload)
	}
	t.mu.Lock()
	defer t.mu.Unlock()
	if tag == tagFrame && msg.newFrame {
		t.framesSeen++
		if msg.keyframe || msg.anchor {
			t.anchorsSeen++
		}
	}
	switch tag {
	case tagResource:
		t.mirror.feedResource(payload)
	case tagRecovery:
		t.mirror.feedRecovery(payload)
		if msg.last {
			t.kick() // a completed keyframe may make the mirror ready
		}
	}
	for sink, g := range t.sinks {
		if !g.admit(tag, msg, sink.reliableBacklog()) {
			continue
		}
		switch tag {
		case tagFrame:
			t.frames++
			sink.enqueueFrame(msg)
		case tagResource, tagRecovery:
			t.reliable++
			if !sink.enqueueReliable(tag, payload) {
				t.dropped++
				delete(t.sinks, sink)
				go sink.end("too-slow")
			}
		case tagAudio:
			sink.sendAudio(payload)
		}
	}
}

func (t *seatTap) handleText(data []byte) {
	var marker struct {
		T      string `json:"t"`
		ID     uint32 `json:"id"`
		Hidden bool   `json:"hidden"`
		Reason string `json:"reason"`
		Token  string `json:"token"`
	}
	if json.Unmarshal(data, &marker) != nil {
		return
	}
	var raw map[string]any
	_ = json.Unmarshal(data, &raw)
	switch marker.T {
	case "spectate":
		t.mu.Lock()
		t.token = marker.Token
		t.mu.Unlock()
	case "epoch":
		t.mirror.markEpoch()
		t.mu.Lock()
		for _, g := range t.sinks {
			if g.mode == gateEpoch && !g.attached {
				g.attached = true
				g.keyframeSeen = false
				g.joinedAt = time.Now()
				t.joinsEpoch++
			}
		}
		t.mu.Unlock()
	case "reset":
		t.mirror.markReset()
		t.mu.Lock()
		sinks := make([]spectatorSink, 0, len(t.sinks))
		for sink, g := range t.sinks {
			g.mode = gatePending
			g.attached = false
			g.framesOpen = false
			g.keyframeSeen = false
			sinks = append(sinks, sink)
		}
		t.mu.Unlock()
		for _, sink := range sinks {
			sink.control(map[string]any{"t": "epochReset"})
		}
	case "sessionEnded":
		reason := marker.Reason
		if reason == "" {
			reason = "host-left"
		}
		t.shutdown(reason)
	case "parked", "engineStalled":
		t.mu.Lock()
		sinks := make([]spectatorSink, 0, len(t.sinks))
		for sink := range t.sinks {
			sinks = append(sinks, sink)
		}
		t.mu.Unlock()
		for _, sink := range sinks {
			sink.control(raw)
		}
	}
}

// shutdown ends every spectator with the reason and closes the tap.
func (t *seatTap) shutdown(reason string) {
	t.mu.Lock()
	if t.closed {
		t.mu.Unlock()
		return
	}
	t.closed = true
	sinks := make([]spectatorSink, 0, len(t.sinks))
	for sink := range t.sinks {
		sinks = append(sinks, sink)
	}
	t.sinks = make(map[spectatorSink]*gate)
	if t.closeTimer != nil {
		t.closeTimer.Stop()
	}
	t.mu.Unlock()
	for _, sink := range sinks {
		sink.end(reason)
	}
	tapsMu.Lock()
	if taps[t.seat] == t {
		delete(taps, t.seat)
	}
	tapsMu.Unlock()
	if t.cancel != nil {
		t.cancel()
	}
	if t.plane != nil {
		t.plane.conn.CloseNow()
	}
	t.mirror.close()
}

func (t *seatTap) report(ctx context.Context) {
	ticker := time.NewTicker(5 * time.Second)
	defer ticker.Stop()
	ticks := 0
	for {
		select {
		case <-ctx.Done():
			return
		case <-ticker.C:
			ticks++
			t.mu.Lock()
			count := len(t.sinks)
			// Why a spectator is not drawing yet: its gate mode, whether
			// frames were let through, and the reliable backlog the frame
			// gate waits on (it opens under frameGateBacklog).
			gates := make([]string, 0, count)
			for sink, g := range t.sinks {
				mode := "pending"
				switch g.mode {
				case gateEpoch:
					mode = "epoch"
				case gateMirror:
					mode = "mirror"
				}
				open := "wait"
				if (g.mode == gateEpoch && g.keyframeSeen) || g.framesOpen {
					open = "open"
				}
				gates = append(gates, fmt.Sprintf("%s/%s/%.1fMiB", mode, open,
					float64(sink.reliableBacklog())/1048576.0))
			}
			line := fmt.Sprintf("seat %d spectators=%d tap-frames=%d "+
				"tap-reliable=%d dropped=%d joins=%d/%d seen=%d anchors=%d "+
				"gates=%s", t.seat, count,
				t.frames, t.reliable, t.dropped, t.joinsMirror, t.joinsEpoch,
				t.framesSeen, t.anchorsSeen, strings.Join(gates, ","))
			t.mu.Unlock()
			if count > 0 {
				log.Print(line)
			}
			if ticks%6 == 0 {
				s := t.mirror.stats()
				state := "ready"
				if s.Broken != "" {
					state = "unavailable (" + s.Broken + ")"
				} else if !t.mirror.ready() {
					state = "not-ready"
				}
				log.Printf("seat %d mirror %s: textures=%d levels=%d %.1f MiB "+
					"blobs=%d %.1f MiB keyframe=%d gen=%d snapshot=%.1f MiB/%d "+
					"builds=%d lastBuild=%dms xorMiss=%d skipped=%d refs=%d",
					t.seat, state, s.Textures, s.Levels,
					float64(s.TextureBytes)/1048576.0, s.Blobs,
					float64(s.BlobBytes)/1048576.0, s.Keyframe, s.Generation,
					float64(s.LastSnapshotBytes)/1048576.0, s.LastSnapshotMessages,
					s.Builds, s.LastBuildMs, s.XorMismatches, s.Skipped+s.Unknown,
					s.References)
			}
		}
	}
}

// byteQueue is one spectator's ordered, byte-capped reliable plane.
type byteQueue struct {
	mu     sync.Mutex
	cond   *sync.Cond
	items  [][]byte
	total  int
	limit  int
	closed bool
}

func newByteQueue(limit int) *byteQueue {
	q := &byteQueue{limit: limit}
	q.cond = sync.NewCond(&q.mu)
	return q
}

func (q *byteQueue) push(data []byte) bool {
	q.mu.Lock()
	defer q.mu.Unlock()
	if q.closed {
		return true
	}
	if q.total+len(data) > q.limit {
		return false
	}
	q.items = append(q.items, data)
	q.total += len(data)
	q.cond.Signal()
	return true
}

// pop blocks until an item is available or the queue is closed.
func (q *byteQueue) pop() ([]byte, bool) {
	q.mu.Lock()
	defer q.mu.Unlock()
	for len(q.items) == 0 && !q.closed {
		q.cond.Wait()
	}
	if len(q.items) == 0 {
		return nil, false
	}
	data := q.items[0]
	q.items[0] = nil
	q.items = q.items[1:]
	q.total -= len(data)
	return data, true
}

func (q *byteQueue) pending() int {
	q.mu.Lock()
	defer q.mu.Unlock()
	return q.total
}

func (q *byteQueue) raise(limit int) {
	q.mu.Lock()
	defer q.mu.Unlock()
	if limit > q.limit {
		q.limit = limit
	}
}

func (q *byteQueue) close() {
	q.mu.Lock()
	defer q.mu.Unlock()
	q.closed = true
	q.items = nil
	q.total = 0
	q.cond.Broadcast()
}

// spectator is one read-only WebTransport session on a seat tap.
type spectator struct {
	seat     int
	session  *webtransport.Session
	cancel   context.CancelFunc
	remoteIP string
	control_ func(any)
	writer   *frameWriter

	resourceQ *byteQueue
	recoveryQ *byteQueue
	endOnce   sync.Once
}

func (s *spectator) enqueueFrame(msg frameMsg) { s.writer.enqueue(msg) }

func (s *spectator) enqueueReliable(tag byte, data []byte) bool {
	if tag == tagRecovery {
		return s.recoveryQ.push(data)
	}
	return s.resourceQ.push(data)
}

func (s *spectator) reliableBacklog() int { return s.resourceQ.pending() }

// raiseLimits makes room for a join bundle larger than the default queue.
func (s *spectator) raiseLimits(bundleBytes int) {
	s.resourceQ.raise(3*bundleBytes + 16*1024*1024)
	s.recoveryQ.raise(3*bundleBytes + 16*1024*1024)
}

func (s *spectator) sendAudio(data []byte) {
	_ = s.session.SendDatagram(data)
}

func (s *spectator) control(object any) { s.control_(object) }

func (s *spectator) end(reason string) {
	s.endOnce.Do(func() {
		s.control_(map[string]any{"t": "streamEnded", "reason": reason})
		// Give the control message a moment to flush before the session
		// closes underneath it.
		time.AfterFunc(200*time.Millisecond, func() {
			s.session.CloseWithError(0, reason)
			s.cancel()
			s.resourceQ.close()
			s.recoveryQ.close()
		})
	})
}

// pumpQueue writes one spectator's reliable plane onto one long-lived uni
// stream. A write error ends the spectator.
func (s *spectator) pumpQueue(ctx context.Context, tag byte, q *byteQueue) {
	stream, err := s.session.OpenUniStreamSync(ctx)
	if err != nil {
		s.end("stream-ended")
		return
	}
	if _, err := stream.Write([]byte{tag}); err != nil {
		s.end("stream-ended")
		return
	}
	for {
		data, ok := q.pop()
		if !ok {
			return
		}
		if _, err := stream.Write(lengthPrefixed(data)); err != nil {
			s.end("stream-ended")
			return
		}
	}
}

func remoteIP(addr string) string {
	if host, _, err := net.SplitHostPort(addr); err == nil {
		return host
	}
	return addr
}

// Reasons a spectator's client may escalate to a rejoin; everything else
// (blob-cache pressure, soft drops) is its own business.
var rejoinReasons = map[string]bool{
	"anchor-stall": true, "resource-apply": true, "dependency": true,
}

// retireLegacyRecovery protects older pages that reset sequencing on their
// existing resource stream before requesting a mirror rejoin. The queue may
// still contain the old epoch, so appending a back-stamped snapshot is never
// safe. A new page reconnects this one spectator on a fresh WebTransport
// session; the host and every other spectator are untouched.
func (t *seatTap) retireLegacyRecovery(sink spectatorSink, reason string) bool {
	if !rejoinReasons[reason] {
		return false
	}
	log.Printf("seat %d spectator legacy recovery (%s): fresh session",
		t.seat, reason)
	sink.end("stream-ended")
	return true
}

// serveSpectator runs after the hello named a spectate token. The seat
// claim, the nginx per-IP gate and the catalog lease are all untouched:
// a spectator is a soft seat.
func serveSpectator(ctx context.Context, cancel context.CancelFunc,
	seat int, session *webtransport.Session, token string,
	remote string, sendControl func(any), readControl func() ([]byte, error)) {
	ip := remoteIP(remote)
	if spectatorsPerBox > 0 && spectatorTotal() >= spectatorsPerBox ||
		spectatorsPerIP > 0 && spectatorsFromIP(ip) >= spectatorsPerIP {
		sendControl(map[string]any{"t": "wtError", "message": "stream-full"})
		return
	}
	tap, err := tapFor(seat, token)
	if err != nil {
		sendControl(map[string]any{"t": "wtError", "message": err.Error()})
		return
	}
	if spectatorsPerSeat > 0 {
		tap.mu.Lock()
		full := len(tap.sinks) >= spectatorsPerSeat
		tap.mu.Unlock()
		if full {
			sendControl(map[string]any{"t": "wtError", "message": "stream-full"})
			return
		}
	}
	spec := &spectator{seat: seat, session: session, cancel: cancel,
		remoteIP: ip, control_: sendControl,
		resourceQ: newByteQueue(spectatorQueueBytes),
		recoveryQ: newByteQueue(spectatorQueueBytes)}
	// Spectator acks are no-ops: the relay's in-flight accounting belongs
	// to the host, and the tap is acked as it is consumed from loopback.
	spec.writer = newFrameWriter(session, func(int) {})
	defer spec.writer.close()
	defer spec.resourceQ.close()
	defer spec.recoveryQ.close()
	go spec.pumpQueue(ctx, tagResource, spec.resourceQ)
	go spec.pumpQueue(ctx, tagRecovery, spec.recoveryQ)

	log.Printf("seat %d spectator from %s attached", seat, remote)
	sendControl(map[string]any{"t": "wtReady", "seat": seat,
		"spectator": true, "input": false, "audio": true})
	tap.add(spec)
	defer tap.remove(spec)

	// Datagrams from a spectator carry nothing the relay may see.
	go func() {
		for {
			if _, err := session.ReceiveDatagram(ctx); err != nil {
				return
			}
		}
	}()

	// Control stream: input, visibility, session and reset messages are
	// dropped. Older pages can still send requestRecovery after resetting
	// their client on this same command stream. Re-serving the mirror there
	// is unsafe: already queued/in-flight bytes precede the back-stamped
	// snapshot. End only this spectator so its page reconnects on fresh QUIC
	// streams. Current pages replace their WebTransport directly and never
	// send this legacy request.
	for {
		event, err := readControl()
		if err != nil {
			return
		}
		var message struct {
			T      string `json:"t"`
			Reason string `json:"reason"`
		}
		if json.Unmarshal(event, &message) != nil {
			continue
		}
		if message.T == "requestRecovery" &&
			tap.retireLegacyRecovery(spec, message.Reason) {
			return
		}
	}
}