// Per-seat state mirror: instant spectator joins that never touch the host
// (WHOS-WATCHING-NOW-SPEC.md, section 13.4).
//
// The relay's tap is a byte-identical copy of everything the host receives.
// Fed with it from the host session's first byte, this mirror tracks the
// state a fresh client needs to start decoding the host's command stream:
//
//	textures   create record + the current bytes of every mip level
//	           (full updates, sub-rect updates and CRC-gated XOR deltas
//	           applied exactly as the browser applies them)
//	buffers    create records (their content travels as frame geometry)
//	blobs      reliable-plane geometry/texture bases of the last two
//	           keyframe generations (the browser's own retention rule)
//	cursor     the last cursor bitmap and visibility
//	keyframe   the host's current recovery frame, verbatim, as received on
//	           the recovery plane
//
// A join is served from memory: a synthesized resource snapshot (stamped
// with sequence numbers that end exactly at the mirror's consistency point
// S0), the tap's reliable messages recorded since S0 ("delta"), and the
// keyframe with W3CS_DICTIONARY_ONLY set (the browser keeps it as the
// anchors' zstd dictionary and absorbs its blob definitions, draws
// nothing). From the next live geometry anchor on, the spectator decodes
// the host's stream as-is. Nothing here asks the relay, the proxy or the
// host's browser for anything.
//
// Resource memory is the price: roughly the host's live texture set plus
// two generations of reliable blobs per mirrored seat, bounded by
// W3CS_MIRROR_MAX_MB (the mirror declares itself unavailable above it and
// joins fall back to the relay's in-band epoch).
package main

import (
	"bytes"
	"encoding/binary"
	"errors"
	"fmt"
	"hash/crc32"
	"io"
	"log"
	"net/http"
	"sort"
	"sync"
	"time"

	"github.com/klauspost/compress/zstd"
)

const (
	envelopeSize = 32

	kindHello     = 1
	kindResource  = 2
	kindRepair    = 4
	kindError     = 6
	kindBundle    = 7
	kindReference = 8

	flagCompressed     = 0x1
	flagLast           = 0x4
	flagDictionaryOnly = 0x10

	codecZstd = 2

	opCreateBuffer            = 2
	opCreateTexture           = 4
	opUpdateTexture           = 5
	opDestroyResource         = 6
	opDefineBlob              = 7
	opDefineBlobXor           = 8
	opDefineBlobXorMask       = 9
	opResetBlobCache          = 10
	opDefineBlobFloat16       = 11
	opUpdateTextureBlob       = 12
	opSnapshotBegin           = 13
	opSnapshotEnd             = 14
	opSetCursor               = 30
	opShowCursor              = 31
	opDefineBlobQuant         = 35
	opUpdateTextureXor        = 36
	opDefineBlobRef           = 37
	opUpdateTextureImage      = 40
	opUpdateTextureImageBlob  = 44
	recordHeaderSize          = 8
	createTextureSize         = 32
	createBufferSize          = 25
	updateTextureHeader       = 40
	updateTextureBlobLen      = 44
	updateTextureImageHeader  = 48
	updateTextureImageBlobLen = 48
	defineBlobHeader          = 24

	wireFragment     = 60 * 1024
	maxResourceBytes = 64 * 1024 * 1024
	// Snapshot message sizing: the browser decodes each reliable message
	// in one piece and parses at most 16,384 records per message.
	snapshotMessageBytes   = 2 * 1024 * 1024
	snapshotMessageRecords = 4096
	// A cached snapshot serves joins for this long; the recorded delta
	// bridges it to the present. Past either bound the next join rebuilds.
	snapshotMaxAge   = 30 * time.Second
	snapshotMaxDelta = 16 * 1024 * 1024
	// After a host reset its own epoch follows within moments; no join
	// request is needed (or wanted) in that window.
	resetEpochGrace = 15 * time.Second
)

var mirrorMaxBytes = int64(envInt("W3CS_MIRROR_MAX_MB", 256)) * 1024 * 1024

var errMirrorUnavailable = errors.New("mirror-unavailable")

type envelope struct {
	kind     byte
	flags    uint16
	session  uint32
	sequence uint32
	frame    uint32
	index    uint16
	count    uint16
	size     uint32
}

func parseEnvelope(data []byte) (envelope, []byte, error) {
	if len(data) < envelopeSize ||
		binary.LittleEndian.Uint32(data) != envelopeMagic ||
		data[4] != envelopeVersion {
		return envelope{}, nil, errors.New("bad envelope")
	}
	e := envelope{
		kind:     data[5],
		flags:    binary.LittleEndian.Uint16(data[6:]),
		session:  binary.LittleEndian.Uint32(data[8:]),
		sequence: binary.LittleEndian.Uint32(data[12:]),
		frame:    binary.LittleEndian.Uint32(data[16:]),
		index:    binary.LittleEndian.Uint16(data[20:]),
		count:    binary.LittleEndian.Uint16(data[22:]),
		size:     binary.LittleEndian.Uint32(data[24:]),
	}
	payload := data[envelopeSize:]
	if uint32(len(payload)) != e.size || e.count == 0 || e.index >= e.count {
		return envelope{}, nil, errors.New("bad envelope geometry")
	}
	return e, payload, nil
}

// message is one reassembled reliable-plane message.
type message struct {
	kind    byte
	flags   uint16
	session uint32
	endSeq  uint32
	payload []byte
}

// assembler reassembles the contiguous fragments of one reliable message.
// The plane is ordered, so a fragment that does not continue the current
// message is corruption, never reordering.
type assembler struct {
	active bool
	head   envelope
	raw    [][]byte
	pieces [][]byte
	flags  uint16
	have   uint16
}

func (a *assembler) reset() {
	a.active = false
	a.raw = nil
	a.pieces = nil
}

func (a *assembler) push(data []byte) (*message, error) {
	e, payload, err := parseEnvelope(data)
	if err != nil {
		return nil, err
	}
	if e.index == 0 {
		a.active = true
		a.head = e
		a.raw = a.raw[:0]
		a.pieces = a.pieces[:0]
		a.flags = 0
		a.have = 0
	} else if !a.active || e.index != a.have || e.count != a.head.count ||
		e.kind != a.head.kind || e.session != a.head.session ||
		e.sequence-uint32(e.index) != a.head.sequence {
		a.active = false
		return nil, errors.New("fragment out of order")
	}
	a.raw = append(a.raw, data)
	a.pieces = append(a.pieces, payload)
	a.flags |= e.flags
	a.have++
	if a.have < e.count {
		return nil, nil
	}
	a.active = false
	total := 0
	for _, piece := range a.pieces {
		total += len(piece)
	}
	joined := make([]byte, 0, total)
	for _, piece := range a.pieces {
		joined = append(joined, piece...)
	}
	return &message{kind: a.head.kind, flags: a.flags, session: a.head.session,
		endSeq: a.head.sequence + uint32(e.count) - 1, payload: joined}, nil
}

type mipLayout struct {
	pitch, rows, blockBytes, bytesPerPixel uint32
}

const (
	formatDXT1 = 827611204
	formatDXT2 = 844388420
	formatDXT3 = 861165636
	formatDXT4 = 877942852
	formatDXT5 = 894720068
)

// textureLayout mirrors the browser's command-state layout table exactly.
func textureLayout(format, width, height uint32) (mipLayout, bool) {
	blocks := func(n uint32) uint32 { return (n + 3) / 4 }
	switch format {
	case formatDXT1:
		return mipLayout{pitch: blocks(width) * 8, rows: blocks(height),
			blockBytes: 8}, true
	case formatDXT2, formatDXT3, formatDXT4, formatDXT5:
		return mipLayout{pitch: blocks(width) * 16, rows: blocks(height),
			blockBytes: 16}, true
	case 21, 22:
		return mipLayout{pitch: width * 4, rows: height, bytesPerPixel: 4}, true
	case 23, 24, 25, 26:
		return mipLayout{pitch: width * 2, rows: height, bytesPerPixel: 2}, true
	}
	return mipLayout{}, false
}

type mirrorMip struct {
	width, height uint32
	layout        mipLayout
	bytes         []byte
	// shared: bytes alias a retained blob's payload (copy-on-write).
	shared bool
	// blobID names the retained blob whose bytes are exactly this level's
	// current content; zero once the level changed or the blob is gone.
	blobID uint32
	// imagePayload is an owned opcode-40 inline image update for the current
	// contents. imageBlobID/meta remember its original opcode-44 form so a
	// snapshot can use the smaller bind while that definition is still active.
	// Once the browser-equivalent blob cache reclaims it, the inline copy keeps
	// late joins independent of blob lifetime and id reuse.
	imagePayload []byte
	imageBlobID  uint32
	imageMeta    [4]byte
}

type mirrorTexture struct {
	id, generation, format, levels uint32
	create                         []byte
	mips                           []*mirrorMip
}

type mirrorBlob struct {
	id      uint32
	payload []byte // 24-byte define header + bytes, re-emitted verbatim
	gen     uint32
	seq     uint64
}

type keyframeCopy struct {
	frame     uint32
	fragments [][]byte // verbatim recovery-plane fragments
	dict      [][]byte // the same with W3CS_DICTIONARY_ONLY set
}

type snapshotCache struct {
	s0         uint32
	messages   [][]byte // stamped fragments, in order
	bytes      int
	createdAt  time.Time
	ready      bool
	err        error
	done       chan struct{}
	delta      [][]byte // tap reliable fragments after s0, verbatim
	deltaBytes int
}

type mirrorStats struct {
	Textures, Levels, Blobs int
	TextureBytes, BlobBytes int64
	Keyframe                uint32
	Generation              uint32
	XorMismatches, Unknown  int
	Skipped                 int
	References, Builds      int
	LastBuildMs             int64
	LastSnapshotBytes       int
	LastSnapshotMessages    int
	Broken                  string
}

// seatMirror is the state model of one seat's host stream.
type seatMirror struct {
	mu sync.Mutex

	seat        int
	fetchBundle func(hexDigest string) ([]byte, error)

	session      uint32
	epochSeen    bool
	resetAt      time.Time
	broken       string
	textures     map[uint32]*mirrorTexture
	buffers      map[uint32][]byte
	blobs        map[uint32]*mirrorBlob
	cursor       []byte
	cursorShow   []byte
	keyframe     *keyframeCopy
	keyframeGen  uint32
	pendingKey   *keyframeCopy
	pendingFrame uint32
	completedSeq uint32
	defineSeq    uint64
	bytes        int64

	reliable assembler
	cache    *snapshotCache

	dec *zstd.Decoder
	enc *zstd.Encoder

	xorMismatches, unknownRecords, skipped, references, builds int
	lastBuild                                                  time.Duration
}

func newSeatMirror(seat int, fetch func(string) ([]byte, error)) *seatMirror {
	dec, _ := zstd.NewReader(nil, zstd.WithDecoderConcurrency(1))
	enc, _ := zstd.NewWriter(nil, zstd.WithEncoderLevel(zstd.SpeedFastest),
		zstd.WithEncoderConcurrency(1))
	m := &seatMirror{seat: seat, fetchBundle: fetch, dec: dec, enc: enc}
	m.clearLocked()
	return m
}

// relayBundleFetcher resolves a RESOURCE_REFERENCE digest through the
// relay's own bundle endpoint (memory LRU, then disk spool).
func relayBundleFetcher(seat int) func(string) ([]byte, error) {
	client := &http.Client{Timeout: 10 * time.Second}
	return func(hexDigest string) ([]byte, error) {
		url := fmt.Sprintf("http://%s:%d/resource-bundle/%s", *relayHost,
			*relayBase+seat, hexDigest)
		response, err := client.Get(url)
		if err != nil {
			return nil, err
		}
		defer response.Body.Close()
		if response.StatusCode != http.StatusOK {
			return nil, fmt.Errorf("bundle %s: status %d", hexDigest,
				response.StatusCode)
		}
		return io.ReadAll(io.LimitReader(response.Body, maxResourceBytes+1))
	}
}

func (m *seatMirror) clearLocked() {
	m.textures = make(map[uint32]*mirrorTexture)
	m.buffers = make(map[uint32][]byte)
	m.blobs = make(map[uint32]*mirrorBlob)
	m.cursor = nil
	m.cursorShow = nil
	m.keyframe = nil
	m.keyframeGen = 0
	m.pendingKey = nil
	m.bytes = 0
	m.reliable.reset()
	m.dropCacheLocked()
}

// markEpoch: the relay's {"t":"epoch"} marker precedes a complete state
// description on the reliable plane; rebuild from here.
func (m *seatMirror) markEpoch() {
	m.mu.Lock()
	defer m.mu.Unlock()
	m.clearLocked()
	m.epochSeen = true
	m.broken = ""
}

// markReset: the host's codec restarted its numbering; the host's own
// epoch follows on its own (the relay bumps both nonces with the reset).
func (m *seatMirror) markReset() {
	m.mu.Lock()
	defer m.mu.Unlock()
	m.clearLocked()
	m.epochSeen = false
	m.resetAt = time.Now()
	m.broken = ""
}

func (m *seatMirror) failLocked(reason string) {
	if m.broken != "" {
		return
	}
	log.Printf("seat %d mirror unavailable: %s", m.seat, reason)
	m.broken = reason
	// Free the resources; the assembler keeps parsing so the next epoch
	// marker can restart from a clean state.
	m.textures = make(map[uint32]*mirrorTexture)
	m.buffers = make(map[uint32][]byte)
	m.blobs = make(map[uint32]*mirrorBlob)
	m.bytes = 0
	m.dropCacheLocked()
}

// ready reports whether a join can be served from memory: an epoch was
// seen from its start and the host has a recovery keyframe for it.
func (m *seatMirror) ready() bool {
	m.mu.Lock()
	defer m.mu.Unlock()
	return m.readyLocked()
}

func (m *seatMirror) readyLocked() bool {
	return m.epochSeen && m.broken == "" && m.keyframe != nil
}

// epochOpen: a state description is streaming (marker seen, no keyframe
// yet) or is about to (the host just reset), so a waiting spectator will
// be served without asking the relay for another epoch.
func (m *seatMirror) epochOpen() bool {
	m.mu.Lock()
	defer m.mu.Unlock()
	if m.broken != "" {
		return false
	}
	if m.epochSeen {
		return m.keyframe == nil
	}
	return !m.resetAt.IsZero() && time.Since(m.resetAt) < resetEpochGrace
}

func (m *seatMirror) stats() mirrorStats {
	m.mu.Lock()
	defer m.mu.Unlock()
	s := mirrorStats{Textures: len(m.textures), Blobs: len(m.blobs),
		Generation: m.keyframeGen, XorMismatches: m.xorMismatches,
		Unknown: m.unknownRecords, Skipped: m.skipped, References: m.references,
		Builds: m.builds, LastBuildMs: m.lastBuild.Milliseconds(),
		Broken: m.broken}
	for _, texture := range m.textures {
		s.Levels += len(texture.mips)
		for _, mip := range texture.mips {
			if mip.imagePayload != nil {
				s.TextureBytes += int64(len(mip.imagePayload))
			} else if !mip.shared {
				s.TextureBytes += int64(len(mip.bytes))
			}
		}
	}
	for _, blob := range m.blobs {
		s.BlobBytes += int64(len(blob.payload))
	}
	if m.keyframe != nil {
		s.Keyframe = m.keyframe.frame
	}
	if m.cache != nil && m.cache.ready {
		s.LastSnapshotBytes = m.cache.bytes
		s.LastSnapshotMessages = len(m.cache.messages)
	}
	return s
}

// feedResource takes one tap resource-plane fragment (tag 2), verbatim.
func (m *seatMirror) feedResource(data []byte) {
	m.mu.Lock()
	defer m.mu.Unlock()
	if m.cache != nil {
		if e, _, err := parseEnvelope(data); err == nil && e.sequence > m.cache.s0 {
			m.cache.delta = append(m.cache.delta, data)
			m.cache.deltaBytes += len(data)
			if m.cache.deltaBytes > snapshotMaxDelta {
				m.dropCacheLocked()
			}
		}
	}
	msg, err := m.reliable.push(data)
	if err != nil {
		m.failLocked("reliable plane: " + err.Error())
		return
	}
	if msg == nil {
		return
	}
	m.completedSeq = msg.endSeq
	if m.session == 0 {
		m.session = msg.session
	}
	if m.broken != "" || !m.epochSeen {
		return
	}
	if err := m.applyMessage(msg); err != nil {
		m.failLocked(err.Error())
	}
}

// feedRecovery takes one tap recovery-plane fragment (tag 3): a keyframe.
func (m *seatMirror) feedRecovery(data []byte) {
	m.mu.Lock()
	defer m.mu.Unlock()
	e, _, err := parseEnvelope(data)
	if err != nil || e.kind != kindFrame || e.flags&flagKeyframe == 0 {
		return
	}
	if e.index == 0 {
		m.pendingKey = &keyframeCopy{frame: e.frame}
		m.pendingFrame = e.frame
	}
	if m.pendingKey == nil || m.pendingFrame != e.frame ||
		uint16(len(m.pendingKey.fragments)) != e.index {
		m.pendingKey = nil
		return
	}
	m.pendingKey.fragments = append(m.pendingKey.fragments, data)
	dict := make([]byte, len(data))
	copy(dict, data)
	binary.LittleEndian.PutUint16(dict[6:], e.flags|flagDictionaryOnly)
	m.pendingKey.dict = append(m.pendingKey.dict, dict)
	if e.index+1 < e.count {
		return
	}
	m.keyframe = m.pendingKey
	m.pendingKey = nil
	m.keyframeGen++
	// The browser reclaims epoch bases older than the previous keyframe
	// when a new one lands; two generations stay resident. Current lossy
	// textures own an inline image copy, so their state does not pin a browser
	// blob beyond this ordinary lifetime.
	for id, blob := range m.blobs {
		if m.keyframeGen >= 2 && blob.gen <= m.keyframeGen-2 {
			m.bytes -= int64(len(blob.payload))
			delete(m.blobs, id)
		}
	}
	for _, texture := range m.textures {
		for _, mip := range texture.mips {
			if mip.blobID != 0 && m.blobs[mip.blobID] == nil {
				m.unshareMip(mip)
				mip.blobID = 0
			}
		}
	}
	if m.bytes > mirrorMaxBytes {
		m.failLocked(fmt.Sprintf("oversize: %d MiB mirrored", m.bytes>>20))
	}
}

// keyframePending: a recovery frame is mid-flight on the tap; a join right
// now would hand out the previous one while anchors already name this one.
func (m *seatMirror) keyframePending() bool {
	m.mu.Lock()
	defer m.mu.Unlock()
	return m.pendingKey != nil
}

func (m *seatMirror) decode(msg *message) ([]byte, error) {
	if msg.flags&flagCompressed == 0 {
		return msg.payload, nil
	}
	if len(msg.payload) < 8 {
		return nil, errors.New("truncated compressed payload")
	}
	size := binary.LittleEndian.Uint32(msg.payload)
	if size > maxResourceBytes {
		return nil, errors.New("compressed payload too large")
	}
	if msg.payload[4] != codecZstd {
		return nil, fmt.Errorf("unsupported codec %d", msg.payload[4])
	}
	out, err := m.dec.DecodeAll(msg.payload[8:], make([]byte, 0, size))
	if err != nil {
		return nil, fmt.Errorf("zstd: %w", err)
	}
	if uint32(len(out)) != size {
		return nil, errors.New("decompressed size mismatch")
	}
	return out, nil
}

func (m *seatMirror) applyMessage(msg *message) error {
	switch msg.kind {
	case kindHello, kindError:
		return nil
	case kindResource, kindRepair, kindBundle, kindReference:
	default:
		return fmt.Errorf("unexpected reliable kind %d", msg.kind)
	}
	records, err := m.decode(msg)
	if err != nil {
		return err
	}
	switch msg.kind {
	case kindBundle:
		if len(records) < 32 {
			return errors.New("truncated bundle")
		}
		records = records[32:]
	case kindReference:
		if len(records) != 36 {
			return errors.New("bad bundle reference")
		}
		expected := binary.LittleEndian.Uint32(records[32:])
		if m.fetchBundle == nil {
			return errors.New("bundle reference without a fetcher")
		}
		m.references++
		fetched, err := m.fetchBundle(fmt.Sprintf("%x", records[:32]))
		if err != nil {
			return fmt.Errorf("bundle fetch: %w", err)
		}
		if uint32(len(fetched)) != expected {
			return errors.New("fetched bundle size differs")
		}
		records = fetched
	}
	return m.applyRecords(records)
}

func (m *seatMirror) applyRecords(records []byte) error {
	offset := 0
	for offset+recordHeaderSize <= len(records) {
		opcode := records[offset]
		size := int(binary.LittleEndian.Uint32(records[offset+4:]))
		offset += recordHeaderSize
		if size > len(records)-offset {
			return errors.New("truncated record")
		}
		payload := records[offset : offset+size]
		offset += size
		if err := m.applyRecord(opcode, payload); err != nil {
			return err
		}
	}
	if offset != len(records) {
		return errors.New("trailing record bytes")
	}
	return nil
}

func (m *seatMirror) applyRecord(opcode byte, payload []byte) error {
	if err := m.applyRecordInner(opcode, payload); err != nil {
		return err
	}
	if m.bytes > mirrorMaxBytes {
		return fmt.Errorf("oversize: %d MiB mirrored", m.bytes>>20)
	}
	return nil
}

func (m *seatMirror) applyRecordInner(opcode byte, payload []byte) error {
	switch opcode {
	case opCreateTexture:
		return m.createTexture(payload)
	case opCreateBuffer:
		if len(payload) != createBufferSize {
			return errors.New("bad create-buffer record")
		}
		id := binary.LittleEndian.Uint32(payload)
		m.buffers[id] = append([]byte(nil), payload...)
	case opUpdateTexture:
		if len(payload) < updateTextureHeader {
			return errors.New("truncated texture update")
		}
		return m.updateTexture(payload[:updateTextureHeader],
			payload[updateTextureHeader:], 0)
	case opUpdateTextureBlob:
		if len(payload) != updateTextureBlobLen {
			return errors.New("bad texture-blob update")
		}
		blobID := binary.LittleEndian.Uint32(payload[40:])
		blob := m.blobs[blobID]
		if blob == nil {
			// The browser raises a repairable state error here; the host's
			// own stream is what is off. Skip alike.
			m.skipped++
			return nil
		}
		return m.updateTexture(payload[:updateTextureHeader],
			blob.payload[defineBlobHeader:], blobID)
	case opUpdateTextureImage, opUpdateTextureImageBlob:
		return m.updateTextureImage(opcode, payload)
	case opUpdateTextureXor:
		return m.updateTextureXor(payload)
	case opDestroyResource:
		if len(payload) != 4 {
			return errors.New("bad destroy record")
		}
		id := binary.LittleEndian.Uint32(payload)
		if texture := m.textures[id]; texture != nil {
			for _, mip := range texture.mips {
				m.releaseMip(mip)
			}
			delete(m.textures, id)
		}
		delete(m.buffers, id)
	case opDefineBlob:
		if len(payload) < defineBlobHeader {
			return errors.New("truncated blob define")
		}
		id := binary.LittleEndian.Uint32(payload)
		size := binary.LittleEndian.Uint32(payload[4:])
		if id == 0 || uint32(len(payload)-defineBlobHeader) != size {
			return errors.New("bad blob define")
		}
		if old := m.blobs[id]; old != nil {
			for _, texture := range m.textures {
				for _, mip := range texture.mips {
					if mip.blobID == id {
						m.unshareMip(mip)
						mip.blobID = 0
					}
				}
			}
			m.bytes -= int64(len(old.payload))
		}
		m.defineSeq++
		m.blobs[id] = &mirrorBlob{id: id, payload: payload, gen: m.keyframeGen,
			seq: m.defineSeq}
		m.bytes += int64(len(payload))
	case opResetBlobCache:
		if len(payload) != 0 {
			return errors.New("bad blob-cache reset")
		}
		for _, texture := range m.textures {
			for _, mip := range texture.mips {
				if mip.blobID != 0 {
					m.unshareMip(mip)
					mip.blobID = 0
				}
			}
		}
		for id, blob := range m.blobs {
			m.bytes -= int64(len(blob.payload))
			delete(m.blobs, id)
		}
	case opSetCursor:
		m.cursor = append([]byte(nil), payload...)
	case opShowCursor:
		m.cursorShow = append([]byte(nil), payload...)
	case opSnapshotBegin, opSnapshotEnd:
	case opDefineBlobXor, opDefineBlobXorMask, opDefineBlobFloat16,
		opDefineBlobQuant, opDefineBlobRef:
		// Geometry deltas live on the frame plane; one on the reliable
		// plane would need a base this mirror does not model.
		return fmt.Errorf("reliable blob delta opcode %d", opcode)
	default:
		// Pipeline state records are frame-plane material; the reliable
		// plane only carries resources, so anything else is just noted.
		m.unknownRecords++
	}
	return nil
}

func (m *seatMirror) createTexture(payload []byte) error {
	if len(payload) != createTextureSize {
		return errors.New("bad create-texture record")
	}
	words := make([]uint32, 8)
	for i := range words {
		words[i] = binary.LittleEndian.Uint32(payload[i*4:])
	}
	id, generation, width, height, levels, format :=
		words[0], words[1], words[2], words[3], words[4], words[6]
	if id == 0 || width == 0 || height == 0 || width > 8192 || height > 8192 ||
		levels == 0 || levels > 16 {
		return errors.New("bad texture bounds")
	}
	if old := m.textures[id]; old != nil {
		for _, mip := range old.mips {
			m.releaseMip(mip)
		}
		delete(m.textures, id)
	}
	texture := &mirrorTexture{id: id, generation: generation, format: format,
		levels: levels, create: append([]byte(nil), payload...)}
	for level := uint32(0); level < levels; level++ {
		mipWidth, mipHeight := width>>level, height>>level
		if mipWidth == 0 {
			mipWidth = 1
		}
		if mipHeight == 0 {
			mipHeight = 1
		}
		layout, ok := textureLayout(format, mipWidth, mipHeight)
		if !ok {
			// The browser fails this texture too; the host's own session
			// is what suffers. Keep the mirror alive without it.
			m.skipped++
			return nil
		}
		mip := &mirrorMip{width: mipWidth, height: mipHeight, layout: layout,
			bytes: make([]byte, layout.pitch*layout.rows)}
		texture.mips = append(texture.mips, mip)
		m.bytes += int64(len(mip.bytes))
	}
	m.textures[id] = texture
	return nil
}

type textureUpdate struct {
	id, generation, level, x, y, width, height, pitch, format, size uint32
}

func parseTextureUpdate(header []byte) textureUpdate {
	w := func(i int) uint32 { return binary.LittleEndian.Uint32(header[i*4:]) }
	return textureUpdate{id: w(0), generation: w(1), level: w(2), x: w(3),
		y: w(4), width: w(5), height: w(6), pitch: w(7), format: w(8), size: w(9)}
}

func (m *seatMirror) clearMipImage(mip *mirrorMip) {
	if mip.imagePayload != nil {
		m.bytes -= int64(len(mip.imagePayload))
	}
	mip.imagePayload = nil
	mip.imageBlobID = 0
	mip.imageMeta = [4]byte{}
}

func (m *seatMirror) releaseMip(mip *mirrorMip) {
	if !mip.shared {
		m.bytes -= int64(len(mip.bytes))
	}
	m.clearMipImage(mip)
	mip.bytes = nil
	mip.shared = false
	mip.blobID = 0
}

// updateTextureImage canonicalizes either wire form into an owned inline
// image update. A synthesized spectator snapshot can send it through the
// production browser ImageDecoder without decoding WebP on the bridge. For
// opcode 44, the original blob id and metadata allow a smaller bind while its
// definition remains in the active browser-equivalent blob window.
func (m *seatMirror) updateTextureImage(opcode byte, payload []byte) error {
	if opcode == opUpdateTextureImageBlob {
		if len(payload) != updateTextureImageBlobLen {
			return errors.New("bad texture-image-blob update")
		}
	} else if len(payload) < updateTextureImageHeader {
		return errors.New("truncated texture-image update")
	}
	u := parseTextureUpdate(payload[:updateTextureHeader])
	texture := m.textures[u.id]
	if texture == nil {
		m.skipped++
		return nil
	}
	if texture.format != u.format || u.level >= uint32(len(texture.mips)) {
		m.skipped++
		return nil
	}
	mip := texture.mips[u.level]
	layout, ok := textureLayout(u.format, u.width, u.height)
	if !ok || u.format < 21 || u.format > 26 || u.x != 0 || u.y != 0 ||
		u.width > mip.width || u.height > mip.height || u.pitch != layout.pitch ||
		u.size != layout.pitch*layout.rows {
		m.skipped++
		return nil
	}
	if u.width != mip.width || u.height != mip.height {
		return errors.New("partial texture-image update cannot be mirrored")
	}
	var imagePayload []byte
	var imageBlobID uint32
	var imageMeta [4]byte
	if opcode == opUpdateTextureImageBlob {
		imageBlobID = binary.LittleEndian.Uint32(payload[40:])
		blob := m.blobs[imageBlobID]
		if imageBlobID == 0 || blob == nil || len(blob.payload) <= defineBlobHeader ||
			payload[44] != 1 || payload[45] != 1 || payload[46] > 100 ||
			payload[47] > 6 {
			return errors.New("invalid texture-image-blob update")
		}
		encoded := blob.payload[defineBlobHeader:]
		imagePayload = make([]byte, updateTextureImageHeader+len(encoded))
		copy(imagePayload, payload[:updateTextureHeader])
		imagePayload[40] = payload[44]
		binary.LittleEndian.PutUint32(imagePayload[44:], uint32(len(encoded)))
		copy(imagePayload[updateTextureImageHeader:], encoded)
		copy(imageMeta[:], payload[44:48])
	} else {
		encoded := binary.LittleEndian.Uint32(payload[44:])
		if payload[40] != 1 || encoded == 0 ||
			uint64(encoded) != uint64(len(payload)-updateTextureImageHeader) {
			return errors.New("invalid texture-image update")
		}
		imagePayload = append([]byte(nil), payload...)
	}
	if !mip.shared {
		m.bytes -= int64(len(mip.bytes))
	}
	mip.bytes = nil
	mip.shared = false
	mip.blobID = 0
	m.clearMipImage(mip)
	mip.imagePayload = imagePayload
	mip.imageBlobID = imageBlobID
	mip.imageMeta = imageMeta
	m.bytes += int64(len(mip.imagePayload))
	texture.generation = u.generation
	return nil
}

// updateTexture applies a (sub-rect) level update the way the browser
// does. blobID names the retained blob the source came from; a full-level
// blob update aliases the blob's bytes instead of copying them.
func (m *seatMirror) updateTexture(header, source []byte, blobID uint32) error {
	u := parseTextureUpdate(header)
	texture := m.textures[u.id]
	if texture == nil {
		// A texture the mirror skipped (unsupported format) or a host-side
		// ordering fault the browser also rejects; skip alike.
		m.skipped++
		return nil
	}
	if texture.format != u.format || u.level >= uint32(len(texture.mips)) ||
		uint32(len(source)) != u.size {
		m.skipped++
		return nil
	}
	mip := texture.mips[u.level]
	sourceLayout, ok := textureLayout(u.format, u.width, u.height)
	if !ok || u.pitch != sourceLayout.pitch || u.size != u.pitch*sourceLayout.rows {
		m.skipped++
		return nil
	}
	full := u.x == 0 && u.y == 0 && u.width == mip.width && u.height == mip.height
	texture.generation = u.generation
	if mip.imagePayload != nil && !full {
		return errors.New("partial texture update without an exact image rebase")
	}
	if full {
		m.clearMipImage(mip)
	}
	if full && blobID != 0 {
		if !mip.shared {
			m.bytes -= int64(len(mip.bytes))
		}
		mip.bytes = source
		mip.shared = true
		mip.blobID = blobID
		return nil
	}
	if full {
		if !mip.shared {
			m.bytes -= int64(len(mip.bytes))
		}
		mip.bytes = append([]byte(nil), source...)
		mip.shared = false
		mip.blobID = 0
		m.bytes += int64(len(mip.bytes))
		return nil
	}
	var destX, destY uint32
	if mip.layout.blockBytes != 0 {
		destX = (u.x / 4) * mip.layout.blockBytes
		destY = u.y / 4
	} else {
		destX = u.x * mip.layout.bytesPerPixel
		destY = u.y
	}
	if destY+sourceLayout.rows > mip.layout.rows || destX+u.pitch > mip.layout.pitch {
		m.skipped++
		return nil
	}
	m.unshareMip(mip)
	for row := uint32(0); row < sourceLayout.rows; row++ {
		copy(mip.bytes[(destY+row)*mip.layout.pitch+destX:],
			source[row*u.pitch:(row+1)*u.pitch])
	}
	mip.blobID = 0
	return nil
}

// unshareMip gives a level private bytes before an in-place change: the
// previous bytes may belong to a retained blob or to a snapshot being built.
func (m *seatMirror) unshareMip(mip *mirrorMip) {
	private := make([]byte, len(mip.bytes))
	copy(private, mip.bytes)
	if !mip.shared {
		m.bytes -= int64(len(mip.bytes))
	}
	mip.bytes = private
	mip.shared = false
	m.bytes += int64(len(private))
}

func (m *seatMirror) updateTextureXor(payload []byte) error {
	if len(payload) < 48 {
		return errors.New("truncated texture xor")
	}
	u := parseTextureUpdate(payload[:updateTextureHeader])
	baseCrc := binary.LittleEndian.Uint32(payload[40:])
	resultCrc := binary.LittleEndian.Uint32(payload[44:])
	texture := m.textures[u.id]
	if texture == nil {
		m.skipped++
		return nil
	}
	if texture.format != u.format || u.level >= uint32(len(texture.mips)) ||
		u.x != 0 || u.y != 0 {
		m.skipped++
		return nil
	}
	mip := texture.mips[u.level]
	if mip.imagePayload != nil {
		return errors.New("texture xor without an exact image rebase")
	}
	if uint32(len(mip.bytes)) != u.size {
		m.skipped++
		return nil
	}
	if crc32.ChecksumIEEE(mip.bytes) != baseCrc {
		// Same as the browser: skip; the encoder's periodic full update
		// restores the level.
		m.xorMismatches++
		return nil
	}
	wordCount := (u.size + 3) / 4
	maskSize := (wordCount + 1) / 2
	if uint32(len(payload)) < 48+maskSize {
		return errors.New("truncated texture xor mask")
	}
	mask := payload[48 : 48+maskSize]
	values := payload[48+maskSize:]
	out := make([]byte, len(mip.bytes))
	copy(out, mip.bytes)
	valueOffset := 0
	for index := range out {
		word := index >> 2
		bits := mask[word>>1]
		if word&1 != 0 {
			bits >>= 4
		}
		if bits&(1<<(index&3)) != 0 {
			if valueOffset >= len(values) {
				return errors.New("texture xor has too few values")
			}
			out[index] ^= values[valueOffset]
			valueOffset++
		}
	}
	if valueOffset != len(values) {
		return errors.New("texture xor has extra values")
	}
	if crc32.ChecksumIEEE(out) != resultCrc {
		m.xorMismatches++
		return nil
	}
	if !mip.shared {
		m.bytes -= int64(len(mip.bytes))
	}
	mip.bytes = out
	mip.shared = false
	mip.blobID = 0
	m.bytes += int64(len(out))
	texture.generation = u.generation
	return nil
}

// ---- snapshot ---------------------------------------------------------

// recordBatch accumulates records into snapshot messages.
type recordBatch struct {
	messages [][]byte
	current  []byte
	count    int
}

func (b *recordBatch) add(opcode byte, parts ...[]byte) {
	size := 0
	for _, part := range parts {
		size += len(part)
	}
	if len(b.current) > 0 && (len(b.current)+recordHeaderSize+size >
		snapshotMessageBytes || b.count >= snapshotMessageRecords) {
		b.flush()
	}
	header := make([]byte, recordHeaderSize)
	header[0] = opcode
	binary.LittleEndian.PutUint32(header[4:], uint32(size))
	b.current = append(b.current, header...)
	for _, part := range parts {
		b.current = append(b.current, part...)
	}
	b.count++
}

func (b *recordBatch) flush() {
	if len(b.current) == 0 {
		return
	}
	b.messages = append(b.messages, b.current)
	b.current = nil
	b.count = 0
}

// collectLocked serializes the mirrored state into plain record messages.
// Order matters for the browser: blobs before the texture updates that
// reference them, creates before updates.
func (m *seatMirror) collectLocked() [][]byte {
	batch := &recordBatch{}
	blobs := make([]*mirrorBlob, 0, len(m.blobs))
	for _, blob := range m.blobs {
		blobs = append(blobs, blob)
	}
	sort.Slice(blobs, func(i, j int) bool { return blobs[i].seq < blobs[j].seq })
	for _, blob := range blobs {
		batch.add(opDefineBlob, blob.payload)
	}
	// Image expansion resolves blob ids before the browser applies any other
	// record in the same reliable message. Definitions must therefore finish
	// in an earlier message than every opcode-44 bind.
	batch.flush()
	bufferIDs := make([]uint32, 0, len(m.buffers))
	for id := range m.buffers {
		bufferIDs = append(bufferIDs, id)
	}
	sort.Slice(bufferIDs, func(i, j int) bool { return bufferIDs[i] < bufferIDs[j] })
	for _, id := range bufferIDs {
		batch.add(opCreateBuffer, m.buffers[id])
	}
	textureIDs := make([]uint32, 0, len(m.textures))
	for id := range m.textures {
		textureIDs = append(textureIDs, id)
	}
	sort.Slice(textureIDs, func(i, j int) bool { return textureIDs[i] < textureIDs[j] })
	for _, id := range textureIDs {
		texture := m.textures[id]
		create := append([]byte(nil), texture.create...)
		binary.LittleEndian.PutUint32(create[4:], texture.generation)
		batch.add(opCreateTexture, create)
		for level, mip := range texture.mips {
			if mip.imagePayload != nil {
				payload := append([]byte(nil), mip.imagePayload...)
				binary.LittleEndian.PutUint32(payload[4:], texture.generation)
				blob := m.blobs[mip.imageBlobID]
				encoded := payload[updateTextureImageHeader:]
				if mip.imageBlobID != 0 && blob != nil &&
					bytes.Equal(blob.payload[defineBlobHeader:], encoded) {
					bind := make([]byte, updateTextureImageBlobLen)
					copy(bind, payload[:updateTextureHeader])
					binary.LittleEndian.PutUint32(bind[40:], mip.imageBlobID)
					copy(bind[44:], mip.imageMeta[:])
					batch.add(opUpdateTextureImageBlob, bind)
				} else {
					batch.add(opUpdateTextureImage, payload)
				}
				continue
			}
			header := make([]byte, updateTextureHeader)
			for i, word := range []uint32{texture.id, texture.generation,
				uint32(level), 0, 0, mip.width, mip.height, mip.layout.pitch,
				texture.format, uint32(len(mip.bytes))} {
				binary.LittleEndian.PutUint32(header[i*4:], word)
			}
			if mip.blobID != 0 && m.blobs[mip.blobID] != nil {
				ref := make([]byte, 4)
				binary.LittleEndian.PutUint32(ref, mip.blobID)
				batch.add(opUpdateTextureBlob, header, ref)
			} else {
				batch.add(opUpdateTexture, header, mip.bytes)
			}
		}
	}
	if m.cursor != nil {
		batch.add(opSetCursor, m.cursor)
	}
	if m.cursorShow != nil {
		batch.add(opShowCursor, m.cursorShow)
	}
	batch.flush()
	return batch.messages
}

// encodeMessages compresses and fragments plain record messages into
// stamped wire fragments whose sequences end exactly at s0.
func (m *seatMirror) encodeMessages(plain [][]byte, session, s0 uint32) ([][]byte, int, error) {
	type wire struct {
		payload []byte
		flags   uint16
	}
	wires := make([]wire, 0, len(plain))
	fragmentsTotal := uint64(0)
	for _, records := range plain {
		compressed := m.enc.EncodeAll(records, make([]byte, 8, 8+len(records)/2))
		binary.LittleEndian.PutUint32(compressed, uint32(len(records)))
		compressed[4] = codecZstd
		item := wire{payload: records}
		if len(compressed) < len(records) {
			item = wire{payload: compressed, flags: flagCompressed}
		}
		wires = append(wires, item)
		fragmentsTotal += uint64((len(item.payload) + wireFragment - 1) / wireFragment)
		if len(item.payload) == 0 {
			fragmentsTotal++
		}
	}
	if fragmentsTotal+1 > uint64(s0) {
		return nil, 0, fmt.Errorf("snapshot of %d fragments does not fit below sequence %d",
			fragmentsTotal, s0)
	}
	sequence := s0 - uint32(fragmentsTotal) + 1
	fragments := make([][]byte, 0, fragmentsTotal)
	bytes := 0
	for _, item := range wires {
		count := (len(item.payload) + wireFragment - 1) / wireFragment
		if count == 0 {
			count = 1
		}
		for index := 0; index < count; index++ {
			start := index * wireFragment
			end := start + wireFragment
			if end > len(item.payload) {
				end = len(item.payload)
			}
			piece := item.payload[start:end]
			flags := item.flags
			if index+1 == count {
				flags |= flagLast
			}
			fragment := make([]byte, envelopeSize+len(piece))
			binary.LittleEndian.PutUint32(fragment, envelopeMagic)
			fragment[4] = envelopeVersion
			fragment[5] = kindResource
			binary.LittleEndian.PutUint16(fragment[6:], flags)
			binary.LittleEndian.PutUint32(fragment[8:], session)
			binary.LittleEndian.PutUint32(fragment[12:], sequence)
			binary.LittleEndian.PutUint32(fragment[16:], 0)
			binary.LittleEndian.PutUint16(fragment[20:], uint16(index))
			binary.LittleEndian.PutUint16(fragment[22:], uint16(count))
			binary.LittleEndian.PutUint32(fragment[24:], uint32(len(piece)))
			binary.LittleEndian.PutUint32(fragment[28:], crc32.ChecksumIEEE(piece))
			copy(fragment[envelopeSize:], piece)
			fragments = append(fragments, fragment)
			bytes += len(fragment)
			sequence++
		}
	}
	return fragments, bytes, nil
}

// ensureSnapshot returns once a fresh snapshot cache exists (building one
// when needed). Callers must not hold any tap lock: a build compresses the
// whole texture set and takes a few hundred milliseconds at most.
func (m *seatMirror) ensureSnapshot() error {
	for {
		m.mu.Lock()
		if !m.readyLocked() {
			m.mu.Unlock()
			return errMirrorUnavailable
		}
		cache := m.cache
		if cache != nil && !cache.ready && cache.err == nil {
			done := cache.done
			m.mu.Unlock()
			<-done
			continue
		}
		if cache != nil && cache.ready &&
			time.Since(cache.createdAt) < snapshotMaxAge {
			m.mu.Unlock()
			return nil
		}
		// Build: the consistency point is the last complete reliable
		// message; fragments of a message still arriving are seeded into
		// the delta so the spectator sees it whole.
		started := time.Now()
		plain := m.collectLocked()
		session, s0 := m.session, m.completedSeq
		cache = &snapshotCache{s0: s0, createdAt: started, done: make(chan struct{})}
		if m.reliable.active {
			for _, raw := range m.reliable.raw {
				cache.delta = append(cache.delta, raw)
				cache.deltaBytes += len(raw)
			}
		}
		m.cache = cache
		m.builds++
		m.mu.Unlock()

		fragments, bytes, err := m.encodeMessages(plain, session, s0)

		m.mu.Lock()
		if m.cache == cache {
			cache.messages, cache.bytes, cache.err = fragments, bytes, err
			cache.ready = err == nil
			m.lastBuild = time.Since(started)
			if err != nil {
				log.Printf("seat %d mirror snapshot failed: %v", m.seat, err)
				m.cache = nil
			} else {
				log.Printf("seat %d mirror snapshot built: %d messages %d fragments "+
					"%.1f MiB s0=%d in %d ms", m.seat, len(plain), len(fragments),
					float64(bytes)/1048576.0, s0, m.lastBuild.Milliseconds())
			}
		} else {
			// Dropped under us (epoch/reset): whoever waits retries.
			err = errMirrorUnavailable
		}
		close(cache.done)
		m.mu.Unlock()
		if err != nil {
			return err
		}
		return nil
	}
}

// joinBundle is the entire state a joining spectator needs, taken from the
// cache ensureSnapshot prepared. Called under the tap lock, so no tap
// message can fall between the delta copied here and the live stream the
// caller attaches right after.
type joinBundle struct {
	resource [][]byte
	recovery [][]byte
	bytes    int
	s0       uint32
	keyframe uint32
	delta    int
}

func (m *seatMirror) takeBundle() (*joinBundle, error) {
	m.mu.Lock()
	defer m.mu.Unlock()
	if !m.readyLocked() || m.cache == nil || !m.cache.ready {
		return nil, errMirrorUnavailable
	}
	if m.pendingKey != nil {
		return nil, errors.New("keyframe-pending")
	}
	bundle := &joinBundle{s0: m.cache.s0, keyframe: m.keyframe.frame,
		delta: len(m.cache.delta)}
	bundle.resource = make([][]byte, 0, len(m.cache.messages)+len(m.cache.delta))
	bundle.resource = append(bundle.resource, m.cache.messages...)
	bundle.resource = append(bundle.resource, m.cache.delta...)
	bundle.recovery = append([][]byte(nil), m.keyframe.dict...)
	bundle.bytes = m.cache.bytes + m.cache.deltaBytes
	for _, fragment := range bundle.recovery {
		bundle.bytes += len(fragment)
	}
	return bundle, nil
}

func (m *seatMirror) dropCache() {
	m.mu.Lock()
	defer m.mu.Unlock()
	m.dropCacheLocked()
}

// dropCacheLocked forgets the serialized snapshot; a build in flight
// notices the swap when it finishes and reports unavailable.
func (m *seatMirror) dropCacheLocked() {
	m.cache = nil
}

func (m *seatMirror) close() {
	m.mu.Lock()
	defer m.mu.Unlock()
	m.clearLocked()
	m.dec.Close()
	_ = m.enc.Close()
}