// w3cs WebTransport bridge: terminates browser QUIC sessions and bridges
// them onto a seat relay's local WebSocket planes.
//
// Why a sidecar: the relay already serves every command plane over local
// WebSockets (its TCP fallback), and loopback TCP never drops or reorders.
// The lossy leg is the internet path, and that is exactly the part QUIC
// replaces. The bridge re-planes the byte-identical relay messages:
//
//	frame plane     one QUIC unidirectional stream PER FRAME. A dependent
//	                frame superseded while its stream is still blocked is
//	                cancelled (RESET_STREAM), so one loss or stall can
//	                never head-of-line block later frames — the SCTP/TCP
//	                failure mode this bridge exists to remove. Keyframes
//	                and geometry anchors are never cancelled.
//	resource plane  one long-lived stream (ordered + lossless; texture XOR
//	                deltas are stateful and need exactly this).
//	recovery plane  one reliable stream per recovery frame.
//	control (bidi)  hello/ready handshake, then reliable input events.
//	datagrams       pointer moves from the browser (latest-wins; a lost
//	                move is superseded by the next one ~16 ms later).
//
// Stream framing to the browser: first byte tags the plane (1 frame,
// 2 resource, 3 recovery), then length-prefixed relay messages exactly as
// the DataChannel would have delivered them. The proven client transport
// runs unchanged on top.
//
// Backpressure: each relay message is acked (text byte count, the same ack
// protocol the browser's WebSocket fallback speaks) only after its QUIC
// stream write returns, so relay-side in_flight reflects what QUIC flow
// control has not yet accepted and the relay governor keeps working.
//
// That accounting alone is NOT enough. A frame this writer drops before it
// ever starts (latest-wins on the unstarted queue) is acked too - dropping
// it costs no wire, and leaving it unacked would pin in_flight forever. So
// on a path that cannot carry the current cadence the relay sees a plane
// that is never full: its only congestion signal is the buffered amount of
// a LOOPBACK socket. Measured 2026-08-22: a 0.6% loss path capped QUIC near
// 5 Mbps while the relay kept pacing 20 fps x 53 KiB, this writer dropped
// more than half of it unstarted, and the relay's governor never learned a
// capacity (`capacity=0.00Mbps`) or cut the rate - the viewer got 5 fps of
// broken chains instead of 10 solid ones. `framePressure` reports below
// give the relay the send-side truth it cannot observe itself.
package main

import (
	"context"
	"crypto/tls"
	"encoding/binary"
	"encoding/json"
	"errors"
	"flag"
	"fmt"
	"io"
	"log"
	"net"
	"net/http"
	"net/url"
	"os"
	"strconv"
	"strings"
	"sync"
	"time"

	"github.com/apernet/quic-go"
	"github.com/apernet/quic-go/congestion"
	"github.com/apernet/quic-go/http3"
	"github.com/coder/websocket"
	"war3replays/w3cs-wt-bridge/bbr"
	"war3replays/w3cs-wt-bridge/webtransport"
)

const (
	tagFrame    = 1
	tagResource = 2
	tagRecovery = 3

	envelopeMagic   = 0x53433357 // "W3CS"
	envelopeVersion = 1
	kindFrame       = 3
	flagKeyframe    = 0x2
	flagAnchor      = 0x8

	// Cancel code for superseded dependent frames; the browser treats any
	// reset as "this frame never completed" and the Reassembler ages the
	// partial fragments out.
	cancelSuperseded = webtransport.StreamErrorCode(1)
)

var (
	// UDP 443 first: it is what routers, VPNs, security products and ISP
	// gear let through; the page dials it first and falls back to the
	// legacy 4443 with a fresh object. Every address must bind, or the
	// bridge does not start: a silently missing primary would cost every
	// viewer a full first try. nginx owns TCP 443 only; UDP is free.
	listenAddrs = flag.String("listen", ":443,:4443",
		"comma-separated HTTP/3 (UDP) listen addresses; all must bind")
	certFile          = flag.String("cert", "/etc/ssl/w3cs/w3cs.crt", "TLS certificate")
	keyFile           = flag.String("key", "/etc/ssl/w3cs/w3cs.key", "TLS key")
	relayHost         = flag.String("relay-host", "127.0.0.1", "relay host")
	relayBase         = flag.Int("relay-base", 8144, "relay port base (seat N = base+N)")
	seatCount         = flag.Int("seats", 9, "highest seat number served")
	reforgedRelayBase = flag.Int("reforged-relay-base", 0,
		"Reforged relay port base (slot N = base+N); zero disables the route")
	reforgedSlotCount = flag.Int("reforged-slots", 0,
		"highest isolated Reforged slot served; zero disables the route")
)

const reforgedProfile = "reforged-d3d11-live-v1"

// certLoader re-reads the certificate when the file changes, so Let's
// Encrypt renewals apply without a restart.
type certLoader struct {
	mu       sync.Mutex
	cert     *tls.Certificate
	loadedAt time.Time
}

func (c *certLoader) get() (*tls.Certificate, error) {
	c.mu.Lock()
	defer c.mu.Unlock()
	if c.cert != nil && time.Since(c.loadedAt) < time.Minute {
		return c.cert, nil
	}
	cert, err := tls.LoadX509KeyPair(*certFile, *keyFile)
	if err != nil {
		if c.cert != nil {
			return c.cert, nil
		}
		return nil, err
	}
	c.cert = &cert
	c.loadedAt = time.Now()
	return c.cert, nil
}

type frameMsg struct {
	data      []byte
	frame     uint32
	fragment  uint16
	fragments uint16
	valid     bool
	reliable  bool
	keyframe  bool
	anchor    bool
	last      bool
	newFrame  bool
}

// parseFrameMsg classifies one relay frame-plane message. Anything that is
// not a well-formed disposable frame fragment is treated as reliable: the
// bridge must never cancel what it does not understand.
func parseFrameMsg(data []byte) frameMsg {
	msg := frameMsg{data: data, reliable: true, newFrame: true, last: true}
	if len(data) < 32 ||
		binary.LittleEndian.Uint32(data) != envelopeMagic ||
		data[4] != envelopeVersion || data[5] != kindFrame {
		return msg
	}
	flags := binary.LittleEndian.Uint16(data[6:])
	msg.frame = binary.LittleEndian.Uint32(data[16:])
	fragIdx := binary.LittleEndian.Uint16(data[20:])
	fragCount := binary.LittleEndian.Uint16(data[22:])
	if fragCount == 0 || fragIdx >= fragCount {
		return msg
	}
	msg.fragment = fragIdx
	msg.fragments = fragCount
	msg.valid = true
	msg.reliable = flags&(flagKeyframe|flagAnchor) != 0
	msg.keyframe = flags&flagKeyframe != 0
	msg.anchor = flags&flagAnchor != 0
	msg.newFrame = fragIdx == 0
	msg.last = fragIdx == fragCount-1
	return msg
}

// How long a superseded dependent frame may keep writing before its
// stream is reset. Cancelling instantly on supersede looked right on
// paper but measured terribly: a transmitted-then-cancelled frame is
// pure wire waste, and under congestion (small cwnd, frame time close
// to the frame interval) it became a cancel spiral — the relay sent
// 27-29 fps while 3-17 fps completed arrival. Transmitted bytes are
// sunk cost: let the in-flight frame finish and apply latest-wins to
// the UNSTARTED queue instead (dropping those wastes nothing). The
// reset stays only as an emergency valve for a frame that is truly
// wedged while newer data waits.
const staleWriteCancel = 300 * time.Millisecond

// A blocked write is only evidence of congestion when frames are waiting
// behind it. On its own it measures transmission time: a 59 KiB frame on a
// 3 Mbps path takes ~160 ms to hand to QUIC no matter how healthy the link
// is. Reporting that as pressure ratcheted a settled 4 fps stream's learned
// capacity down every interval (3.46 -> 2.90 Mbps) with nothing queued and
// nothing dropped, which is how a governor talks itself to the floor.
const framePressureBlock = 100 * time.Millisecond

// Fraction of offered frames this writer may discard before the cadence is
// judged not to fit. Latest-wins dropping is NORMAL: the relay paces a
// little above the path on purpose and the newest frame wins, so a healthy
// 30 fps session steadily drops a few percent. Only a large fraction means
// the relay is pacing for a path that is not there - measured 2026-08-22, a
// fitting session dropped ~1 frame in 7 while a broken one dropped 3 in 4.
const framePressureDropShare = 3 // report when more than 1/3 is dropped

// ...but never on a handful of frames. At a cadence the governor has
// already cut, a couple of dropped frames trip the share test on their own
// (4 of 9 offered is "44%"), and the resulting cut lowers the cadence
// again, which makes the next couple of drops an even larger share. Real
// overload is not subtle: pacing 20 fps into a 5 fps link dropped 10-15
// frames a second.
const framePressureMinDrops = 6

// framePressureReport renders one send-side pressure report for the relay,
// or ("", false) when the interval carried no evidence of pressure.
//
// `dropped` counts frames discarded before their stream ever started and
// `opened` those that got one, so their sum is what the relay offered this
// interval. Both clear themselves the moment the cadence fits, which is what
// makes them safe control inputs.
func framePressureReport(dropped, opened uint64, blocked time.Duration,
	queueLen int) (string, bool) {
	offered := dropped + opened
	overloaded := offered > 0 && dropped >= framePressureMinDrops &&
		dropped*framePressureDropShare > offered
	// A write stuck past a frame interval WITH frames waiting behind it is
	// the same overload one interval earlier. Alone it is only transmission
	// time (a 59 KiB frame on a 3 Mbps path takes ~160 ms to hand to QUIC),
	// and reporting that ratcheted a settled stream's learned capacity down
	// with nothing queued and nothing dropped.
	wedged := queueLen > 0 && blocked >= framePressureBlock &&
		dropped >= framePressureMinDrops
	if !overloaded && !wedged {
		return "", false
	}
	return fmt.Sprintf(
		`{"t":"framePressure","dropped":%d,"offered":%d,"blockedMs":%d,`+
			`"queue":%d}`,
		dropped, offered, blocked.Milliseconds(), queueLen), true
}

// A report must hold for this many consecutive one-second samples. One
// sample of drops is what a link HOLD looks like - a Wi-Fi client going
// off-channel for a scan stops everything for ~0.5 s and then flushes it all
// (measured 2026-08-22 on a viewer's path: RTT 596 -> 391 -> 190 ms on
// consecutive pings every ~10 s, a clean 49 ms between). Every frame offered
// during the hold is dropped latest-wins, which trips the share test, and
// the governor cut a cadence that had nothing to do with it; at 13 fps a
// hold drops ~6 frames, which is why sessions converged there. A cadence
// that really does not fit drops frames EVERY second.
const framePressureStreak = 2

// An ack stall: data in flight and no ack for this long beyond a round
// trip (bbr/snapshot.go). Counted in the session line as `stalls` for
// diagnosis only. It was tried as an exemption from the pressure report
// (a Wi-Fi hold drops one second of frames that no cadence could have
// delivered) and measured wrong on the very path it was built for: under
// load that path stalls every second - retries and a ~15 Mbps bottleneck
// - so the exemption hid the overload and the governor never cut (40 fps
// into it, half the frames dropped, 2-11 drawn). Stalls there ARE the
// congestion signal; the two-sample streak below is what absorbs a lone
// hold.
const linkStallGap = 200 * time.Millisecond

type pressureGate struct{ streak int }

func (g *pressureGate) report(dropped, opened uint64,
	blocked time.Duration, queueLen int) (string, bool) {
	report, ok := framePressureReport(dropped, opened, blocked, queueLen)
	if !ok {
		g.streak = 0
		return "", false
	}
	if g.streak++; g.streak < framePressureStreak {
		return "", false
	}
	return report, true
}

// frameWriter owns the per-frame streams and the supersede policy. One
// goroutine writes; the pump goroutine enqueues, drops unstarted stale
// frames, and only resets a blocked stream past staleWriteCancel.
type frameWriter struct {
	session *webtransport.Session
	ack     func(int)
	// onDrop, when set, is told the lifetime drop count the moment
	// latest-wins discards frames: the relay asks the game for a geometry
	// anchor on it, and every frame between the drop and that anchor is
	// unresolvable for the viewer, so a once-a-second report is ten
	// frames too slow at the cadences this path runs (measured
	// 2026-08-22: 96 anchors requested, 19% of arrivals still
	// unresolved). Called under the writer's lock, like ack.
	onDrop func(total uint64)
	// deliveredBps, when set, is the rate the connection has lately been
	// delivering (linkstats.go). It turns the supersede policy from
	// "always latest-wins" into "latest-wins once the backlog would take
	// longer than frameBacklogBudget to deliver at that rate". A link that
	// STALLS (no ack for ~300 ms, then full rate again - measured every
	// second on a viewer's Wi-Fi, 2026-08-22) queues two or three frames
	// per stall; dropping all but the newest threw away 20% of the frames
	// at 7-13 fps, and every drop cost 1-2 more frames that were already
	// encoded against the dropped one (25% of arrivals unresolved however
	// fast an anchor followed). Those frames deliver fine a moment later;
	// the browser absorbs the late ones (applies, does not draw) and the
	// chains they carry stay whole. A link that is SATURATED builds a
	// backlog that never clears, which the time budget still catches.
	deliveredBps func() int64

	mu          sync.Mutex
	cond        *sync.Cond
	queue       []frameMsg
	queuedBytes int
	closed      bool
	cur         *webtransport.SendStream
	curFrame    uint32
	curRel      bool
	writing     bool
	writeStart  time.Time
	aborted     map[uint32]bool

	// Session-lifetime counters plus a per-report peak, logged by the
	// session's stats reporter.
	opened       uint64
	finished     uint64
	cancelled    uint64
	queueDropped uint64
	written      uint64
	maxBlock     time.Duration
}

func newFrameWriter(session *webtransport.Session, ack func(int)) *frameWriter {
	writer := &frameWriter{session: session, ack: ack,
		aborted: make(map[uint32]bool)}
	writer.cond = sync.NewCond(&writer.mu)
	go writer.run()
	go writer.staleWatch()
	return writer
}

// How much queued frame data, measured as delivery time at the rate the
// connection has lately been delivering, this writer carries before it
// supersedes. Well above a stall (~300 ms) and the frames a stall queues,
// well below what a cadence that does not fit accumulates in seconds.
const frameBacklogBudget = 600 * time.Millisecond

// backlogWithinBudgetLocked reports whether the queue plus `extra` bytes
// would still deliver within frameBacklogBudget at the delivered rate.
// Without a rate (no controller, or nothing delivered yet) the answer is
// no, which is the original always-latest-wins policy. Requires w.mu.
func (w *frameWriter) backlogWithinBudgetLocked(extra int) bool {
	if w.deliveredBps == nil {
		return false
	}
	bps := w.deliveredBps()
	if bps <= 0 {
		return false
	}
	seconds := float64(w.queuedBytes+extra) * 8 / float64(bps)
	return seconds < frameBacklogBudget.Seconds()
}

// maybeCancelStaleLocked resets the in-flight dependent frame only when a
// newer frame is waiting AND the write has been stuck past the stale
// window AND the backlog is past its time budget - a stalled link holds
// a write for 300 ms with two frames behind it, and cancelling then
// wastes the bytes already on the wire for nothing. Requires w.mu.
func (w *frameWriter) maybeCancelStaleLocked(newerFrame uint32) {
	if w.writing && !w.curRel && w.cur != nil && w.curFrame < newerFrame &&
		time.Since(w.writeStart) > staleWriteCancel &&
		!w.backlogWithinBudgetLocked(0) {
		w.aborted[w.curFrame] = true
		w.cancelled++
		w.cur.CancelWrite(cancelSuperseded)
	}
}

// staleWatch backs up the enqueue-time check: once the relay pauses under
// backpressure, no new enqueue would ever fire the stale cancel for the
// frame that is blocking everything queued behind it.
func (w *frameWriter) staleWatch() {
	ticker := time.NewTicker(100 * time.Millisecond)
	defer ticker.Stop()
	for range ticker.C {
		w.mu.Lock()
		if w.closed {
			w.mu.Unlock()
			return
		}
		newest := uint32(0)
		for _, queued := range w.queue {
			if queued.frame > newest {
				newest = queued.frame
			}
		}
		if newest > 0 {
			w.maybeCancelStaleLocked(newest)
		}
		w.mu.Unlock()
	}
}

func (w *frameWriter) enqueue(msg frameMsg) {
	w.mu.Lock()
	defer w.mu.Unlock()
	if w.closed {
		w.ack(len(msg.data))
		return
	}
	if msg.newFrame && !msg.reliable && !w.backlogWithinBudgetLocked(len(msg.data)) {
		// Latest-wins on the unstarted queue: dropping a frame that never
		// started costs no wire. Fragments of the frame currently being
		// written stay — dropping them would leave its stream dangling.
		kept := w.queue[:0]
		droppedBefore := w.queueDropped
		for _, queued := range w.queue {
			if !queued.reliable && queued.frame < msg.frame &&
				queued.frame != w.curFrame {
				w.aborted[queued.frame] = true
				// Count FRAMES, not queue entries: the relay splits a frame
				// into 16 KiB fragments (kWireFragment), so one dropped
				// frame is several entries. Counted per entry, two dropped
				// frames out of thirteen read as "8 dropped of 21 offered"
				// against the opened-FRAMES side of the pressure test and
				// reported overload every second on a path delivering 85%
				// of frames - 70 governor cuts in four minutes, measured
				// 2026-08-22. Fragments arrive in order, so a frame that
				// is not the current one is whole in the queue and its
				// first fragment counts it exactly once.
				if queued.newFrame {
					w.queueDropped++
				}
				w.queuedBytes -= len(queued.data)
				w.ack(len(queued.data))
				continue
			}
			kept = append(kept, queued)
		}
		w.queue = kept
		if w.queueDropped != droppedBefore && w.onDrop != nil {
			w.onDrop(w.queueDropped)
		}
	}
	if msg.newFrame && !msg.reliable {
		w.maybeCancelStaleLocked(msg.frame)
	}
	w.queue = append(w.queue, msg)
	w.queuedBytes += len(msg.data)
	w.cond.Signal()
}

func (w *frameWriter) close() {
	w.mu.Lock()
	w.closed = true
	if w.cur != nil {
		w.cur.CancelWrite(cancelSuperseded)
	}
	w.cond.Broadcast()
	w.mu.Unlock()
}

func (w *frameWriter) run() {
	for {
		w.mu.Lock()
		for len(w.queue) == 0 && !w.closed {
			w.cond.Wait()
		}
		if w.closed {
			for _, queued := range w.queue {
				w.ack(len(queued.data))
			}
			w.queue = nil
			w.queuedBytes = 0
			w.mu.Unlock()
			return
		}
		msg := w.queue[0]
		w.queue = w.queue[1:]
		w.queuedBytes -= len(msg.data)
		if w.aborted[msg.frame] && !msg.reliable {
			if msg.last {
				delete(w.aborted, msg.frame)
			}
			w.ack(len(msg.data))
			w.mu.Unlock()
			continue
		}
		needStream := w.cur == nil || w.curFrame != msg.frame
		if needStream && w.cur != nil {
			// The relay moved on without a last fragment (it can drop a
			// frame mid-send under its own backpressure); finish the old
			// stream so the browser sees a clean end, not a stall.
			w.cur.Close()
			w.cur = nil
		}
		w.mu.Unlock()

		if needStream {
			stream, err := w.session.OpenUniStreamSync(w.session.Context())
			if err != nil {
				w.ack(len(msg.data))
				continue
			}
			if _, err := stream.Write([]byte{tagFrame}); err != nil {
				stream.CancelWrite(cancelSuperseded)
				w.ack(len(msg.data))
				continue
			}
			w.mu.Lock()
			w.cur = stream
			w.curFrame = msg.frame
			w.curRel = msg.reliable
			w.writeStart = time.Now()
			w.opened++
			w.mu.Unlock()
		}

		w.mu.Lock()
		stream := w.cur
		w.writing = true
		w.mu.Unlock()
		writeBegan := time.Now()
		ok := stream != nil
		if ok {
			buffer := make([]byte, 4+len(msg.data))
			binary.LittleEndian.PutUint32(buffer, uint32(len(msg.data)))
			copy(buffer[4:], msg.data)
			if _, err := stream.Write(buffer); err != nil {
				ok = false
			}
		}
		w.mu.Lock()
		w.writing = false
		if blocked := time.Since(writeBegan); blocked > w.maxBlock {
			w.maxBlock = blocked
		}
		if !ok {
			// Cancelled (stale) or session-dead: drop the rest of this
			// frame as it pops.
			w.aborted[msg.frame] = true
			if w.cur == stream {
				w.cur = nil
			}
		} else {
			w.written++
			if msg.last {
				stream.Close()
				if w.cur == stream {
					w.cur = nil
				}
				w.finished++
				delete(w.aborted, msg.frame)
			}
		}
		// The aborted set only needs to cover frames still draining out of
		// the queue; keep it from growing over a long session.
		if len(w.aborted) > 64 {
			w.aborted = make(map[uint32]bool)
		}
		w.mu.Unlock()
		w.ack(len(msg.data))
	}
}

// snapshotStats returns the lifetime counters plus the max single-write
// block since the previous snapshot.
func (w *frameWriter) snapshotStats() (opened, finished, cancelled,
	queueDropped uint64, queueLen int, maxBlock time.Duration) {
	w.mu.Lock()
	defer w.mu.Unlock()
	opened, finished = w.opened, w.finished
	cancelled, queueDropped = w.cancelled, w.queueDropped
	queueLen = len(w.queue)
	maxBlock = w.maxBlock
	w.maxBlock = 0
	return
}

/* Congestion control for every QUIC connection this bridge serves.
 *
 * quic-go implements the RFC 9002 controller, which reads every lost packet
 * as congestion. On a path with a little ambient loss its ceiling is about
 * MSS/(RTT*sqrt(loss)) no matter how much bandwidth is really there: a
 * viewer measured 2026-08-22 at 0.6% loss and 50 ms RTT was held to 3-5
 * Mbps while plain paced UDP over the same path carried 22. A busy replay
 * needs ~9 Mbps, so the governor had to cut to ~5 fps on a link that could
 * have carried full cadence.
 *
 * BBR estimates delivery rate and RTT instead, so ambient loss does not
 * shrink its bandwidth model. Its LOSS RECOVERY still does, though, and
 * for this sender it did so badly: BBR v1 seeds the recovery window from
 * what is in flight when the loss is detected, and between frames that is
 * almost nothing here. Measured 2026-08-22 on the same 0.6%-loss path: a
 * 55 KiB frame that pacing hands to QUIC in ~45 ms took 200-500 ms to
 * write, ~5 loss events a second kept the sender in recovery, the writer
 * dropped 20% of frames unstarted, and the relay governor parked at 15
 * fps while another viewer on a clean path ran 32. The "realtime" profile
 * (a local addition, see bbr/PROVENANCE.md) floors the recovery window at
 * one bandwidth-delay product, so a loss holds the sender AT the delivery
 * rate the path already proved rather than below it.
 *
 * Set W3CS_BBR=0 to fall back to the stock controller, or W3CS_BBR_PROFILE
 * to one of the bbr profiles (standard = upstream recovery), which is how
 * the boxes are A/B compared. */
func useBBR(ctx context.Context, conn *quic.Conn) context.Context {
	if os.Getenv("W3CS_BBR") == "0" {
		return ctx
	}
	sender := bbr.NewBbrSender(
		bbr.DefaultClock{}, congestion.ByteCount(protocolInitialPacketSize),
		bbrProfile)
	conn.SetCongestionControl(sender)
	if stats := linkStatsFrom(ctx); stats != nil {
		stats.setBBR(sender)
	}
	return ctx
}

// bbrProfile is resolved once at startup so a typo fails the service
// loudly instead of silently picking a profile per connection.
var bbrProfile = mustBBRProfile(os.Getenv("W3CS_BBR_PROFILE"))

func mustBBRProfile(name string) bbr.Profile {
	if name == "" {
		return bbr.ProfileRealtime
	}
	profile, err := bbr.ParseProfile(name)
	if err != nil {
		log.Fatalf("W3CS_BBR_PROFILE: %v", err)
	}
	return profile
}

// Matches quic-go's own initial datagram size; the sender re-learns the real
// one from the connection through SetMaxDatagramSize.
const protocolInitialPacketSize = 1280

// relayPlane is one WebSocket to the relay plus its serialized text sender
// (byte-count acks and, on the frame plane, framePressure reports).
type relayPlane struct {
	conn  *websocket.Conn
	ackCh chan string
}

func dialPlane(ctx context.Context, base, path, session string) (*relayPlane, error) {
	return dialRelay(ctx, fmt.Sprintf("%s/%s?session=%s", base, path, session))
}

func dialReforgedPlane(ctx context.Context, base, path,
	ticket string) (*relayPlane, error) {
	target, ok := reforgedRelayTarget(base, path)
	if !ok {
		return nil, errors.New("invalid Reforged relay plane")
	}
	auth, ok := reforgedRelayAuthMessage(ticket)
	if !ok {
		return nil, errors.New("invalid Reforged transport ticket")
	}
	plane, err := dialRelay(ctx, target)
	if err != nil {
		return nil, err
	}
	writeCtx, done := context.WithTimeout(ctx, 5*time.Second)
	err = plane.conn.Write(writeCtx, websocket.MessageText, auth)
	done()
	if err != nil {
		plane.conn.CloseNow()
		return nil, errors.New("Reforged relay authentication failed")
	}
	readCtx, readDone := context.WithTimeout(ctx, 5*time.Second)
	kind, acknowledgement, err := plane.conn.Read(readCtx)
	readDone()
	if err != nil || kind != websocket.MessageText ||
		!validReforgedRelayAuthAck(acknowledgement, path) {
		plane.conn.CloseNow()
		return nil, errors.New("Reforged relay authentication failed")
	}
	return plane, nil
}

func dialRelay(ctx context.Context, target string) (*relayPlane, error) {
	dialCtx, cancel := context.WithTimeout(ctx, 5*time.Second)
	defer cancel()
	conn, _, err := websocket.Dial(dialCtx, target, nil)
	if err != nil {
		return nil, err
	}
	conn.SetReadLimit(2 * 1024 * 1024)
	plane := &relayPlane{conn: conn, ackCh: make(chan string, 1024)}
	go func() {
		for text := range plane.ackCh {
			writeCtx, done := context.WithTimeout(ctx, 10*time.Second)
			err := conn.Write(writeCtx, websocket.MessageText, []byte(text))
			done()
			if err != nil {
				return
			}
		}
	}()
	return plane, nil
}

func (p *relayPlane) ack(bytes int) {
	p.send(strconv.Itoa(bytes))
}

// send queues one text message to the relay. Acks and pressure reports share
// the ack goroutine so they stay in order on the wire; the relay tells them
// apart by their first byte (a digit is an ack, '{' is a report).
func (p *relayPlane) send(text string) {
	select {
	case p.ackCh <- text:
	default:
		// The channel only backs up if the relay stopped reading; the
		// in-flight counter recovers at the next plane connect.
	}
}

type seatSession struct {
	seat    int
	session *webtransport.Session
	cancel  context.CancelFunc
}

var (
	activeMu       sync.Mutex
	active         = make(map[int]*seatSession)
	reforgedActive = make(map[int]*seatSession)
)

func lengthPrefixed(data []byte) []byte {
	out := make([]byte, 4+len(data))
	binary.LittleEndian.PutUint32(out, uint32(len(data)))
	copy(out[4:], data)
	return out
}

// pumpReliable copies one relay plane onto one long-lived uni stream.
func pumpReliable(ctx context.Context, plane *relayPlane,
	session *webtransport.Session, tag byte, fail func(error)) {
	stream, err := session.OpenUniStreamSync(ctx)
	if err != nil {
		fail(fmt.Errorf("open stream %d: %w", tag, err))
		return
	}
	if _, err := stream.Write([]byte{tag}); err != nil {
		fail(fmt.Errorf("stream %d tag: %w", tag, err))
		return
	}
	for {
		kind, data, err := plane.conn.Read(ctx)
		if err != nil {
			fail(fmt.Errorf("relay plane %d read: %w", tag, err))
			return
		}
		if kind != websocket.MessageBinary {
			continue
		}
		if _, err := stream.Write(lengthPrefixed(data)); err != nil {
			fail(fmt.Errorf("stream %d write: %w", tag, err))
			return
		}
		plane.ack(len(data))
	}
}

// pumpRecovery gives every recovery frame its own reliable QUIC stream.
// Recovery frames are rare, self-contained epoch boundaries. Keeping all of
// them on one session-long stream made one bad reader/parser state permanent:
// the resource and disposable frame streams kept moving, but every later
// recovery queued behind the wedged stream and the relay blocked normal frames
// forever waiting for an end-to-end recoveryReady. A fresh ordered stream per
// recovery preserves reliability without carrying that head-of-line state into
// the next epoch.
func pumpRecovery(ctx context.Context, plane *relayPlane,
	session *webtransport.Session, fail func(error)) {
	var stream *webtransport.SendStream
	var streamFrame uint32
	var streamFragments uint16
	var nextFragment uint16
	cancelActive := func() {
		if stream != nil {
			stream.CancelWrite(cancelSuperseded)
			stream = nil
		}
	}
	defer cancelActive()
	for {
		kind, data, err := plane.conn.Read(ctx)
		if err != nil {
			fail(fmt.Errorf("relay recovery read: %w", err))
			return
		}
		if kind != websocket.MessageBinary {
			continue
		}
		msg := parseFrameMsg(data)
		if !msg.valid || !msg.keyframe {
			fail(fmt.Errorf("invalid recovery frame fragment"))
			return
		}
		if msg.newFrame {
			if stream != nil {
				fail(fmt.Errorf("recovery frame %d ended before frame %d",
					streamFrame, msg.frame))
				return
			}
			stream, err = session.OpenUniStreamSync(ctx)
			if err != nil {
				fail(fmt.Errorf("open recovery stream: %w", err))
				return
			}
			streamFrame = msg.frame
			streamFragments = msg.fragments
			nextFragment = 0
			n, writeErr := stream.Write([]byte{tagRecovery})
			if writeErr != nil || n != 1 {
				if writeErr == nil {
					writeErr = io.ErrShortWrite
				}
				fail(fmt.Errorf("recovery stream tag: %w", writeErr))
				return
			}
		} else if stream == nil || msg.frame != streamFrame {
			fail(fmt.Errorf("recovery fragment for frame %d without active frame",
				msg.frame))
			return
		}
		if msg.fragments != streamFragments || msg.fragment != nextFragment {
			fail(fmt.Errorf(
				"recovery frame %d fragment order %d/%d, expected %d/%d",
				streamFrame, msg.fragment, msg.fragments, nextFragment,
				streamFragments))
			return
		}
		framed := lengthPrefixed(data)
		n, writeErr := stream.Write(framed)
		if writeErr != nil || n != len(framed) {
			if writeErr == nil {
				writeErr = io.ErrShortWrite
			}
			fail(fmt.Errorf("recovery stream %d write: %w",
				streamFrame, writeErr))
			return
		}
		plane.ack(len(data))
		nextFragment++
		if msg.last {
			if nextFragment != streamFragments {
				fail(fmt.Errorf("recovery frame %d closed after %d/%d fragments",
					streamFrame, nextFragment, streamFragments))
				return
			}
			if err := stream.Close(); err != nil {
				fail(fmt.Errorf("recovery stream %d close: %w",
					streamFrame, err))
				return
			}
			stream = nil
			streamFrame = 0
			streamFragments = 0
			nextFragment = 0
		}
	}
}

func pumpFrames(ctx context.Context, plane *relayPlane,
	writer *frameWriter, fail func(error)) {
	for {
		kind, data, err := plane.conn.Read(ctx)
		if err != nil {
			fail(fmt.Errorf("relay frame read: %w", err))
			return
		}
		if kind != websocket.MessageBinary {
			continue
		}
		msg := parseFrameMsg(data)
		writer.enqueue(msg)
	}
}

func serveSession(seat int, session *webtransport.Session, sessionID string,
	remote string, link *linkStats, relayPortBase int, profile string) {
	ctx, cancel := context.WithCancel(session.Context())
	defer cancel()
	linkReport := newLinkReporter(link)
	isReforged := profile != ""
	sessionLabel := transportSessionLogLabel(sessionID, profile)
	relayAuthority := sessionID

	// The control stream opens first: its hello triggers the plane dials
	// and its ready reply tells the page it can start the game session.
	control, err := session.AcceptStream(ctx)
	if err != nil {
		return
	}
	controlWriteMu := sync.Mutex{}
	sendControl := func(object any) {
		payload, err := json.Marshal(object)
		if err != nil {
			return
		}
		controlWriteMu.Lock()
		defer controlWriteMu.Unlock()
		control.Write(lengthPrefixed(payload))
	}

	readControl := func() ([]byte, error) {
		header := make([]byte, 4)
		if _, err := ioReadFull(control, header); err != nil {
			return nil, err
		}
		size := binary.LittleEndian.Uint32(header)
		if size == 0 || size > 1024*1024 {
			return nil, fmt.Errorf("invalid control length %d", size)
		}
		payload := make([]byte, size)
		if _, err := ioReadFull(control, payload); err != nil {
			return nil, err
		}
		return payload, nil
	}

	hello, err := readControl()
	if err != nil {
		return
	}
	var helloMsg struct {
		T        string `json:"t"`
		Session  string `json:"session"`
		Spectate string `json:"spectate"`
		Profile  string `json:"profile"`
	}
	if isReforged {
		ticket, ok := reforgedTransportTicketFromHello(hello, profile)
		if !ok {
			sendControl(map[string]any{"t": "wtError",
				"message": "invalid Reforged session identity"})
			return
		}
		relayAuthority = ticket
		helloMsg.T = "hello"
		helloMsg.Profile = profile
	} else {
		if json.Unmarshal(hello, &helloMsg) != nil || helloMsg.T != "hello" {
			sendControl(map[string]any{"t": "wtError", "message": "expected hello"})
			return
		}
		if sessionID == "" {
			sessionID = helloMsg.Session
			sessionLabel = sessionID
			relayAuthority = sessionID
		}
	}
	if helloMsg.Spectate != "" {
		// A read-only spectator of this seat's host: no seat claim, no
		// plane dials of its own, nothing forwarded to the relay.
		serveSpectator(ctx, cancel, seat, session, helloMsg.Spectate, remote,
			sendControl, readControl)
		return
	}

	// One host session per seat, newest wins — the relay applies the same
	// rule to its planes, so a stale bridge must not hold them.
	activeTable := active
	if isReforged {
		activeTable = reforgedActive
	}
	current := &seatSession{seat: seat, session: session, cancel: cancel}
	claimed := false
	claim := func() {
		activeMu.Lock()
		defer activeMu.Unlock()
		if old := activeTable[seat]; old != nil {
			old.session.CloseWithError(0, "seat session replaced")
			old.cancel()
		}
		activeTable[seat] = current
		claimed = true
	}
	// Preserve the existing Classic replacement timing. Reforged first proves
	// its exact session against every relay plane so a bearer from another
	// slot cannot evict the active browser for this slot.
	if !isReforged {
		claim()
	}
	defer func() {
		if !claimed {
			return
		}
		activeMu.Lock()
		if activeTable[seat] == current {
			delete(activeTable, seat)
		}
		activeMu.Unlock()
	}()

	base := fmt.Sprintf("ws://%s:%d", *relayHost, relayPortBase+seat)
	dial := dialPlane
	if isReforged {
		dial = dialReforgedPlane
	}
	resource, err := dial(ctx, base, "resource", relayAuthority)
	if err != nil {
		sendControl(map[string]any{"t": "wtError",
			"message": "relay resource plane unavailable"})
		return
	}
	defer resource.conn.CloseNow()
	recovery, err := dial(ctx, base, "recovery", relayAuthority)
	if err != nil {
		sendControl(map[string]any{"t": "wtError",
			"message": "relay recovery plane unavailable"})
		return
	}
	defer recovery.conn.CloseNow()
	frame, err := dial(ctx, base, "frame", relayAuthority)
	if err != nil {
		sendControl(map[string]any{"t": "wtError",
			"message": "relay frame plane unavailable"})
		return
	}
	defer frame.conn.CloseNow()
	// The input lane is additive (the page keeps its signaling copy), so an
	// older relay without /input degrades gracefully.
	input, inputErr := dial(ctx, base, "input", relayAuthority)
	if input != nil {
		defer input.conn.CloseNow()
	}
	if isReforged {
		if inputErr != nil {
			sendControl(map[string]any{"t": "wtError",
				"message": "relay input plane unavailable"})
			return
		}
		claim()
	}

	log.Printf("seat %d session %s: planes attached (input=%v)",
		seat, sessionLabel, inputErr == nil)
	ready := map[string]any{"t": "wtReady", "seat": seat,
		"input": inputErr == nil}
	if isReforged {
		ready["profile"] = profile
	}
	sendControl(ready)
	// Tap our own host session from its first byte: the state mirror
	// behind it serves later spectator joins without touching the host
	// (spectate.go, mirror.go).
	if !isReforged {
		go hostTap(ctx, seat, sessionID)
	}

	var failOnce sync.Once
	fail := func(err error) {
		failOnce.Do(func() {
			log.Printf("seat %d session %s: %v", seat, sessionLabel, err)
			session.CloseWithError(0, "bridge plane failed")
			cancel()
		})
	}

	writer := newFrameWriter(session, frame.ack)
	defer writer.close()
	writer.onDrop = func(total uint64) {
		frame.send(fmt.Sprintf(`{"t":"drop","qdrop":%d}`, total))
	}
	writer.deliveredBps = link.delivered

	// Frame-plane health, the counters that caught the cancel spiral:
	// finished must track opened, cancels/queue-drops must stay rare, and
	// maxBlockMs is how hard QUIC flow control pushed back this interval.
	// Sampled every second so the relay's governor can react within its own
	// two-second adjustment window; logged every fifth sample as before.
	go func() {
		ticker := time.NewTicker(time.Second)
		defer ticker.Stop()
		var lastDropped, lastOpened uint64
		var loggedBlock time.Duration
		var pressure pressureGate
		var stalls uint64
		sample := 0
		for {
			select {
			case <-ctx.Done():
				return
			case <-ticker.C:
				opened, finished, cancelled, queueDropped, queueLen,
					maxBlock := writer.snapshotStats()
				if maxBlock > loggedBlock {
					loggedBlock = maxBlock
				}
				dropped := queueDropped - lastDropped
				started := opened - lastOpened
				lastDropped, lastOpened = queueDropped, opened
				if link.takeMaxAckStall() >= linkStallGap {
					stalls++
				}
				linkReport.sampleDelivered()
				if report, ok := pressure.report(
					dropped, started, maxBlock, queueLen); ok {
					frame.send(report)
				}
				// The link estimate rides the same socket: the governor
				// floors its capacity model with it (linkstats.go).
				if report, ok := linkReport.report(queueDropped); ok {
					frame.send(report)
				}
				if sample++; sample == 5 {
					sample = 0
					log.Printf("seat %d wt-frames opened=%d finished=%d "+
						"cancelled=%d qdrop=%d queue=%d maxBlockMs=%d "+
						"stalls=%d%s",
						seat, opened, finished, cancelled, queueDropped,
						queueLen, loggedBlock.Milliseconds(), stalls,
						linkReport.line())
					loggedBlock = 0
				}
			}
		}
	}()

	go pumpReliable(ctx, resource, session, tagResource, fail)
	go pumpRecovery(ctx, recovery, session, fail)
	go pumpFrames(ctx, frame, writer, fail)

	forwardInput := func(event []byte) {
		if input == nil {
			return
		}
		writeCtx, done := context.WithTimeout(ctx, 5*time.Second)
		defer done()
		if err := input.conn.Write(writeCtx, websocket.MessageText,
			event); err != nil {
			input = nil
		}
	}

	// Datagrams: pointer moves. Loss needs no handling — the page races
	// every input over its reliable lanes too and the relay dedupes.
	go func() {
		for {
			event, err := session.ReceiveDatagram(ctx)
			if err != nil {
				return
			}
			forwardInput(event)
		}
	}()

	// Drain the input plane so relay-side pings never back up (the relay
	// does not send on it today).
	if input != nil {
		go func(plane *relayPlane) {
			for {
				if _, _, err := plane.conn.Read(ctx); err != nil {
					return
				}
			}
		}(input)
	}

	// Control stream: reliable input events until the session ends.
	for {
		event, err := readControl()
		if err != nil {
			return
		}
		forwardInput(event)
	}
}

// ioReadFull avoids importing io just for ReadFull semantics on the
// webtransport stream interface.
func ioReadFull(reader interface{ Read([]byte) (int, error) },
	buffer []byte) (int, error) {
	total := 0
	for total < len(buffer) {
		n, err := reader.Read(buffer[total:])
		total += n
		if err != nil {
			return total, err
		}
	}
	return total, nil
}

func main() {
	flag.Parse()
	log.SetFlags(log.LstdFlags | log.LUTC)

	loader := &certLoader{}
	if _, err := loader.get(); err != nil {
		log.Fatalf("load certificate: %v", err)
	}
	tlsConf := &tls.Config{
		// quic-go does not add the h3 ALPN to a custom TLSConfig; without
		// it every handshake dies with "no ALPN protocol selected".
		NextProtos: []string{http3.NextProtoH3},
		GetCertificate: func(*tls.ClientHelloInfo) (*tls.Certificate, error) {
			return loader.get()
		},
	}

	addrs, err := splitListenAddrs(*listenAddrs)
	if err != nil {
		log.Fatalf("listen: %v", err)
	}
	reforgedEnabled, err := validateReforgedRouteConfig(
		*reforgedRelayBase, *reforgedSlotCount)
	if err != nil {
		log.Fatalf("Reforged route: %v", err)
	}
	if reforgedEnabled {
		if err := validateReforgedRelayHost(*relayHost); err != nil {
			log.Fatalf("Reforged route: %v", err)
		}
	}
	wtServer := &webtransport.Server{
		// The page lives on TCP :443 and this bridge on UDP :443/:4443; the
		// default same-authority origin check fails on the legacy port, so
		// same hostname is the trust boundary here; localhost stays open
		// for lab pages.
		CheckOrigin: func(r *http.Request) bool {
			origin := r.Header.Get("Origin")
			if origin == "" {
				return true
			}
			parsed, err := url.Parse(origin)
			if err != nil {
				return false
			}
			originHost := parsed.Hostname()
			requestHost := r.Host
			if host, _, err := net.SplitHostPort(requestHost); err == nil {
				requestHost = host
			}
			return strings.EqualFold(originHost, requestHost) ||
				originHost == "localhost" || originHost == "127.0.0.1"
		},
		H3: &http3.Server{
			// Addr is informational only: the sockets are bound by
			// listenAll, one per -listen address.
			Addr:            addrs[0],
			TLSConfig:       tlsConf,
			EnableDatagrams: true,
			QUICConfig: &quic.Config{
				EnableDatagrams: true,
				MaxIdleTimeout:  45 * time.Second,
			},
			ConnContext: useBBR,
		},
	}

	mux := http.NewServeMux()
	mux.HandleFunc("/seat/", func(w http.ResponseWriter, r *http.Request) {
		rest := strings.TrimPrefix(r.URL.Path, "/seat/")
		seat, err := strconv.Atoi(rest)
		if err != nil || seat < 1 || seat > *seatCount {
			http.Error(w, "unknown seat", http.StatusNotFound)
			return
		}
		sessionID := r.URL.Query().Get("session")
		session, err := wtServer.Upgrade(w, r)
		if err != nil {
			log.Printf("seat %d upgrade failed: %v", seat, err)
			http.Error(w, "upgrade failed", http.StatusBadRequest)
			return
		}
		// "via" is the UDP endpoint the session arrived on: with the page
		// dialling 443 first and 4443 second, the split between the two
		// is the pilot's measure.
		log.Printf("seat %d session %s connected from %s via %s (origin %q)",
			seat, sessionID, r.RemoteAddr, listenAddrFrom(r.Context()),
			r.Header.Get("Origin"))
		go serveSession(seat, session, sessionID, r.RemoteAddr,
			linkStatsFrom(r.Context()), *relayBase, "")
	})
	if reforgedEnabled {
		mux.HandleFunc("/reforged/seat/", func(w http.ResponseWriter,
			r *http.Request) {
			rest := strings.TrimPrefix(r.URL.Path, "/reforged/seat/")
			slot, parseErr := strconv.Atoi(rest)
			if parseErr != nil || slot < 1 || slot > *reforgedSlotCount {
				http.Error(w, "unknown Reforged slot", http.StatusNotFound)
				return
			}
			if !validReforgedRouteQuery(r.URL.RawQuery) {
				http.Error(w, "invalid Reforged session", http.StatusForbidden)
				return
			}
			session, upgradeErr := wtServer.Upgrade(w, r)
			if upgradeErr != nil {
				log.Printf("Reforged slot %d upgrade failed: %v", slot,
					upgradeErr)
				http.Error(w, "upgrade failed", http.StatusBadRequest)
				return
			}
			// No lifecycle bearer or transport ticket enters this URL or log.
			log.Printf("Reforged slot %d connected from %s via %s (origin %q)",
				slot, r.RemoteAddr, listenAddrFrom(r.Context()),
				r.Header.Get("Origin"))
			go serveSession(slot, session, "", r.RemoteAddr,
				linkStatsFrom(r.Context()), *reforgedRelayBase, reforgedProfile)
		})
	}
	wtServer.H3.Handler = mux

	conns, err := listenAll(addrs)
	if err != nil {
		log.Fatalf("listen: %v", err)
	}
	log.Printf("w3cs wt-bridge listening on %s (relay %s:%d+N)",
		strings.Join(addrs, ","), *relayHost, *relayBase)
	if err := serveAll(wtServer, conns); err != nil {
		log.Fatalf("listen: %v", err)
		os.Exit(1)
	}
}

// splitListenAddrs turns the -listen flag into one address per UDP
// socket: comma-separated, blanks ignored, at least one required.
func splitListenAddrs(flagValue string) ([]string, error) {
	var addrs []string
	for _, part := range strings.Split(flagValue, ",") {
		part = strings.TrimSpace(part)
		if part == "" {
			continue
		}
		addrs = append(addrs, part)
	}
	if len(addrs) == 0 {
		return nil, errors.New("no listen address")
	}
	return addrs, nil
}

// listenAll binds every address before anything serves, so one port
// that cannot bind (taken, or not permitted) fails the start as a whole
// instead of leaving a half-reachable bridge. On error, nothing stays
// bound.
func listenAll(addrs []string) ([]*net.UDPConn, error) {
	conns := make([]*net.UDPConn, 0, len(addrs))
	for _, addr := range addrs {
		udpAddr, err := net.ResolveUDPAddr("udp", addr)
		if err == nil {
			var conn *net.UDPConn
			conn, err = net.ListenUDP("udp", udpAddr)
			if err == nil {
				conns = append(conns, conn)
				continue
			}
		}
		for _, conn := range conns {
			conn.Close()
		}
		return nil, fmt.Errorf("%s: %w", addr, err)
	}
	return conns, nil
}

// serveAll serves one QUIC listener per bound socket on the same
// webtransport.Server (ServeQUICConn is per connection, so the server
// does not care which socket a connection came in on). The first
// listener to fail ends the whole bridge, which systemd restarts.
func serveAll(wtServer *webtransport.Server, conns []*net.UDPConn) error {
	errs := make(chan error, len(conns))
	for _, conn := range conns {
		go func(conn *net.UDPConn) {
			errs <- serveWithLinkStats(wtServer, conn)
		}(conn)
	}
	return <-errs
}

type listenAddrKey struct{}

// listenAddrFrom is the local UDP endpoint the connection behind a request
// arrived on ("" outside a served connection).
func listenAddrFrom(ctx context.Context) string {
	addr, _ := ctx.Value(listenAddrKey{}).(string)
	return addr
}

// serveWithLinkStats is webtransport.Server.ListenAndServe with our own
// quic.Transport underneath, because the per-connection evidence in
// linkstats.go has to be attached in Transport.ConnContext: that is the
// one hook whose context both quic-go's tracer and http3's ConnContext
// (useBBR) derive from. The QUIC config mirrors what the vendored
// server's serve() sets (datagrams and partial-delivery resets are
// requirements ServeQUICConn checks). The socket is closed on return.
func serveWithLinkStats(wtServer *webtransport.Server, udpConn *net.UDPConn) error {
	defer udpConn.Close()
	endpoint := udpConn.LocalAddr().String()

	quicConf := wtServer.H3.QUICConfig.Clone()
	quicConf.EnableDatagrams = true
	quicConf.EnableStreamResetPartialDelivery = true
	quicConf.Tracer = newLinkTrace
	transport := &quic.Transport{Conn: udpConn,
		ConnContext: func(ctx context.Context, info *quic.ClientInfo) (
			context.Context, error) {
			ctx, err := newLinkStats(ctx, info)
			if err != nil {
				return nil, err
			}
			return context.WithValue(ctx, listenAddrKey{}, endpoint), nil
		}}
	defer transport.Close()
	listener, err := transport.ListenEarly(wtServer.H3.TLSConfig, quicConf)
	if err != nil {
		return err
	}
	defer listener.Close()

	for {
		conn, err := listener.Accept(context.Background())
		if err != nil {
			return err
		}
		go func() {
			if err := wtServer.ServeQUICConn(conn); err != nil &&
				!errors.Is(err, http.ErrServerClosed) {
				log.Printf("http3: error serving QUIC connection: %v", err)
			}
		}()
	}
}