package client

import (
	"errors"
	"fmt"
	"log"
	"net"
	"sync"
	"sync/atomic"
	"time"

	"github.com/iceBear67/redapricot/client/wire"
)

// errPoolClosed is returned by Dial after Close: the set is shutting down
// and must not start new dials, so a caller (handleControlRequest,
// allocateForResume) gives up rather than waiting on a hub that will never
// be used.
var errPoolClosed = errors.New("worker set closed")

// errTooManyTunnels is returned when live+in-flight worker conns already
// equal maxTunnels. The ControlRequest is dropped; the hub closes the player
// when pendingTimeoutMs fires.
var errTooManyTunnels = errors.New("maxTunnels reached")

// dialAttempts bounds how many times a caller retries Dial when the conn it
// was handed dies before the stream could be attached to it. The race is
// narrow; an unbounded loop would spin against a hub that is refusing every
// connection.
const dialAttempts = 3

// WorkerPool tracks live 1:1 worker connections up to maxTunnels
// (PROTOCOL.md §7.1). There is no least-loaded placement and no sharing:
// every player gets its own TCP connection.
type WorkerPool struct {
	client     *Client
	maxTunnels int

	connSeq atomic.Int64 // conn ids, for log correlation

	mu      sync.Mutex
	conns   map[*WorkerConn]struct{}
	dialing int  // dials currently in flight
	closed  bool // closeAll ran; no new conns may join
}

func newWorkerPool(c *Client, maxTunnels int) *WorkerPool {
	return &WorkerPool{
		client:     c,
		maxTunnels: maxTunnels,
		conns:      make(map[*WorkerConn]struct{}),
	}
}

// Dial opens a dedicated worker conn for one player.
//
// A dial is never performed while holding p.mu: session establishment involves
// network I/O, and holding the lock across it would park every other player
// behind one unresponsive hub. Each caller dials independently; unlike the
// old mux pool there is no shared conn to wait for.
func (p *WorkerPool) Dial() (*WorkerConn, error) {
	p.mu.Lock()
	if p.closed {
		p.mu.Unlock()
		return nil, errPoolClosed
	}
	if len(p.conns)+p.dialing >= p.maxTunnels {
		p.mu.Unlock()
		return nil, errTooManyTunnels
	}
	p.dialing++
	p.mu.Unlock()

	wc, err := p.dialWorker()

	p.mu.Lock()
	p.dialing--
	if err != nil {
		p.mu.Unlock()
		return nil, err
	}
	if p.closed {
		// Close raced this dial: the conn must not enter the set.
		p.mu.Unlock()
		_ = wc.fc.Close()
		return nil, errPoolClosed
	}
	if len(p.conns) >= p.maxTunnels {
		p.mu.Unlock()
		_ = wc.fc.Close()
		return nil, errTooManyTunnels
	}
	p.conns[wc] = struct{}{}
	p.mu.Unlock()
	return wc, nil
}

// dialWorker establishes one worker conn. It must be called without p.mu held.
// The dial runs on the client's context so Close aborts it mid-handshake.
func (p *WorkerPool) dialWorker() (*WorkerConn, error) {
	sess, err := p.client.dialSession(p.client.ctx, MagicWorker)
	if err != nil {
		return nil, err
	}
	wc := &WorkerConn{
		pool:        p,
		fc:          sess.fc,
		sendWndInit: sess.peerWnd,
		resume:      sess.resume,
		grace:       p.client.resumeGrace(sess.hubGrace),
		id:          int(p.connSeq.Add(1)),
		done:        make(chan struct{}),
	}
	if p.client.statsOn() {
		wc.stats = &connStats{opened: time.Now()}
	}
	wc.lastPong.Store(time.Now().UnixMilli())
	go wc.readLoop()
	if sess.heartbeat {
		go wc.heartbeatLoop(p.client.cfg.pingInterval(), p.client.cfg.heartbeatTimeout())
	} else {
		log.Printf("worker conn: hub does not support the worker heartbeat; " +
			"a silently dropped path will only be caught by TCP keepalive")
	}
	log.Printf("opened worker conn (send window %d, recv window %d, heartbeat %v, resume %v)",
		wc.sendWndInit, p.client.streamWnd, sess.heartbeat, wc.resume)
	return wc, nil
}

func (p *WorkerPool) count() int {
	p.mu.Lock()
	defer p.mu.Unlock()
	return len(p.conns)
}

func (p *WorkerPool) remove(wc *WorkerConn) {
	p.mu.Lock()
	defer p.mu.Unlock()
	delete(p.conns, wc)
}

func (p *WorkerPool) closeAll() {
	p.mu.Lock()
	p.closed = true
	conns := make([]*WorkerConn, 0, len(p.conns))
	for wc := range p.conns {
		conns = append(conns, wc)
	}
	p.mu.Unlock()
	for _, wc := range conns {
		_ = wc.fc.Close()
	}
}

// WorkerConn is one 1:1 worker connection to the hub: it carries exactly one
// player. Only genuinely per-connection state lives here. Client-wide values
// (the shaper, the advertised receive window, the DATA chunk cap) belong to
// Client: reading them through a connection pointer would make every such
// read a re-parenting hazard once a stream can migrate between conns.
type WorkerConn struct {
	pool *WorkerPool
	fc   *wire.FramedConn

	sendWndInit int // hub's advertised receive window (our send budget)

	// resume is whether this conn negotiated stream resumption, and grace how
	// long a stream parked from it may keep trying to reattach. Both are
	// per-conn: a reattach may land on a different (or restarted) hub, so the
	// flag must be re-checked on the conn that will carry the RESUME.
	resume bool
	grace  time.Duration

	done     chan struct{} // closed when readLoop exits
	lastPong atomic.Int64  // unix ms of the most recent PONG

	id    int        // for log correlation only
	stats *connStats // nil unless diagnostics are enabled

	mu     sync.Mutex
	stream *Stream
	closed bool // readLoop has exited; attach must refuse
}

// heartbeatLoop proves the worker conn is still carrying frames end to end. TCP
// alone cannot tell us: a middlebox that drops an established flow (conntrack
// expiry, firewall state loss) sends no FIN or RST, so the read loop would park
// forever and the player on this conn would silently fail until the process
// restarted.
func (wc *WorkerConn) heartbeatLoop(interval, timeout time.Duration) {
	ticker := time.NewTicker(interval)
	defer ticker.Stop()
	for {
		select {
		case <-wc.done:
			return
		case <-ticker.C:
			if silent := time.Since(time.UnixMilli(wc.lastPong.Load())); silent > timeout {
				log.Printf("worker conn silent for %s; dropping it", silent.Round(time.Second))
				_ = wc.fc.Close() // readLoop unblocks and tears everything down
				return
			}
			msg := wire.NewWriter().U8(MuxPing).I64(time.Now().UnixMilli()).Out()
			if err := wc.fc.WriteFrame(msg); err != nil {
				return
			}
		}
	}
}

// attach publishes a stream on this conn, or reports false if the conn has
// already died (or is already bound — which would be a caller bug).
//
// The check is not advisory. Dial hands out a conn, and the conn's readLoop
// can exit before the caller gets here. A blind store would land on a conn
// nothing iterates and the stream would never be torn down.
func (wc *WorkerConn) attach(st *Stream) bool {
	wc.mu.Lock()
	defer wc.mu.Unlock()
	if wc.closed || wc.stream != nil {
		return false
	}
	wc.stream = st
	return true
}

func (wc *WorkerConn) getStream() *Stream {
	wc.mu.Lock()
	defer wc.mu.Unlock()
	return wc.stream
}

func (wc *WorkerConn) detach() *Stream {
	wc.mu.Lock()
	defer wc.mu.Unlock()
	st := wc.stream
	wc.stream = nil
	return st
}

// readLoop dispatches inbound tunnel frames. DATA is only enqueued (the
// stream's writeLoop does the actual destination writes), so a stalled
// destination cannot stall liveness / WND / FIN dispatch on this conn.
func (wc *WorkerConn) readLoop() {
	for {
		payload, err := wc.fc.ReadFrame()
		if err != nil {
			break
		}
		if wc.stats != nil {
			wc.stats.framesIn.Add(1)
		}
		r := wire.NewReader(payload)
		ftype, err := r.U8()
		if err != nil {
			continue
		}
		switch ftype {
		case MuxData:
			if st := wc.getStream(); st != nil {
				st.deliverFromHub(r.Remaining())
			}
		case MuxWnd:
			if delta, err := r.VarInt(); err == nil && delta > 0 {
				if st := wc.getStream(); st != nil {
					st.grantSendWnd(delta)
				}
			}
		case MuxFin:
			// Graceful: drain what is already queued to the destination first.
			if st := wc.detach(); st != nil {
				st.gracefulFin()
			}
		case MuxRst:
			// The reason is an optional trailing byte; older peers send none.
			reason := RstUnspecified
			if b, err := r.U8(); err == nil {
				reason = int(b)
			}
			if st := wc.detach(); st != nil {
				st.onRst(reason)
			}
		case MuxResumeAck:
			accepted, aerr := r.I64()
			delivered, derr := r.I64()
			cid, cerr := r.Bytes(CIDLen)
			if aerr != nil || derr != nil || cerr != nil {
				continue
			}
			if st := wc.getStream(); st != nil {
				st.deliverResume(resumeResult{accepted: accepted, delivered: delivered, cid: cid})
			}
		case MuxPing:
			nonce, _ := r.I64()
			_ = wc.fc.WriteFrame(wire.NewWriter().U8(MuxPong).I64(nonce).Out())
		case MuxPong:
			now := time.Now()
			wc.lastPong.Store(now.UnixMilli())
			// The probe's nonce is the timestamp we sent, echoed back, so the
			// round trip is free to measure and nobody was reading it.
			if wc.stats != nil {
				if sent, err := r.I64(); err == nil {
					wc.stats.observeRTT(now.Sub(time.UnixMilli(sent)))
				}
			}
		default:
			log.Printf("worker: unknown frame type %d", ftype)
		}
	}
	// Connection lost: tear down the bound stream and drop from the set.
	close(wc.done)
	wc.pool.remove(wc)
	wc.mu.Lock()
	// Marked before the pointer is cleared, under the same lock, so a concurrent
	// attach either lands in the field we are about to drain or is refused.
	wc.closed = true
	st := wc.stream
	wc.stream = nil
	wc.mu.Unlock()
	if st == nil {
		return
	}
	// Only the tunnel leg died. Where the session negotiated resumption the
	// destination socket is kept open and the stream reattaches over a fresh
	// conn (§7.5); otherwise this is the old, unconditional teardown.
	// During Close there is no reattach to come: a stream that parked now would
	// hold its destination socket open past shutdown, so teardown instead.
	if wc.pool.client.closing.Load() {
		st.teardown(false)
		return
	}
	if !st.park(wc.grace) {
		st.teardown(false)
	}
}

func (wc *WorkerConn) sendSyn(cid []byte) error {
	return wc.fc.WriteFrame(wire.NewWriter().U8(MuxSyn).Bytes(cid).Out())
}

func (wc *WorkerConn) sendData(data []byte) error {
	err := wc.fc.WriteFrame(wire.NewWriter().U8(MuxData).Bytes(data).Out())
	if wc.stats != nil {
		wc.stats.framesOut.Add(1)
		if err != nil {
			wc.stats.writeErrs.Add(1)
		}
	}
	return err
}

func (wc *WorkerConn) sendFin() {
	_ = wc.fc.WriteFrame(wire.NewWriter().U8(MuxFin).Out())
}

func (wc *WorkerConn) sendRst() {
	_ = wc.fc.WriteFrame(wire.NewWriter().U8(MuxRst).Out())
}

func (wc *WorkerConn) sendWndUpdate(delta int) {
	_ = wc.fc.WriteFrame(wire.NewWriter().U8(MuxWnd).VarInt(delta).Out())
}

// Stream bridges one player (via the hub) to one destination connection.
//
// Data from the hub is queued and written to the destination by a dedicated
// writeLoop goroutine. The queue is bounded by the advertised receive window —
// the hub never sends more un-credited bytes, so overflow is a protocol
// violation and resets the tunnel.
type Stream struct {
	client *Client
	wc     atomic.Pointer[WorkerConn]
	mapping Mapping
	srcIP   string
	srcPort int
	vel     *velocityForwarder // non-nil when the mapping sets velocitySecret

	share shaperShare   // this stream's position in the egress fair queue
	done  chan struct{} // closed on teardown; unparks a shaper wait

	// resumable is fixed at creation from the conn's negotiated flag. With it
	// false none of the bookkeeping below runs and no buffer is ever allocated,
	// so a client with resumption disabled pays exactly what it used to.
	resumable bool

	stats *streamStats // nil unless diagnostics are enabled

	// ackedOffset is the running sum of WND deltas received. Credit is granted
	// only as bytes reach the peer's terminal socket, so it is a sound lower
	// bound on what has been delivered. Atomic because the worker readLoop
	// advances it and must never block behind the send path.
	ackedOffset atomic.Int64

	// sendMu serializes the send path — buffer the chunk, advance the offset,
	// write the frame — against a reattach's retransmit, so replayed bytes can
	// never interleave with live ones.
	//
	// Deliberately not s.mu: deliverFromHub takes s.mu from the worker readLoop,
	// and holding s.mu across a WriteFrame would stall frame dispatch (WND/FIN
	// /heartbeat replies) on this conn.
	sendMu sync.Mutex
	un     unackedBuf // guarded by sendMu

	mu         sync.Mutex
	cond       *sync.Cond
	cid        []byte // takeover capability; re-minted by the hub on each resume
	dest       net.Conn
	connected  bool
	closed     bool
	parked     bool      // worker conn died; awaiting reattach on a fresh one
	parkedAt   time.Time // when the current hang began; diagnostics only
	finPending bool      // hub sent FIN; close the destination once the queue drains
	finToHub   bool      // destination closed while parked; FIN owed once reattached
	q          []qentry  // hub/local -> destination, waiting for writeLoop
	qBytes     int       // hub bytes only: bounds the peer against its window
	// acceptedOffset counts hub bytes enqueued toward the destination. This, not
	// "bytes written", is what a reattach reports: acceptance is synchronous and
	// stable at park time, whereas delivery is signalled asynchronously and goes
	// silent exactly when the connection dies — which would under-report and make
	// the hub retransmit bytes the player already has.
	acceptedOffset int64
	// deliveredOffset counts hub bytes actually written to the destination.
	// Distinct from acceptedOffset and needed for a different job: a reattach
	// replays from what the peer *accepted*, but sizes the window from what it
	// *delivered*, because the window is a promise about undelivered bytes.
	deliveredOffset int64
	sendWnd         int // flow control: budget for destination -> hub DATA
	consumed        int // flow control: drained bytes not yet credited back to the hub
	resumeWait      chan resumeResult
}

// resumeResult is the hub's answer to a RESUME: how far it got in both senses,
// plus a fresh CID — or the reason the reattach was refused.
type resumeResult struct {
	accepted  int64
	delivered int64
	cid       []byte
	err       error
}

// qentry is one queued write towards the destination. Only hub-originated
// entries take part in flow control; locally injected bytes (the velocity
// login response) are neither counted against the hub's window nor credited
// back when drained.
type qentry struct {
	data    []byte
	fromHub bool
}

func newStream(c *Client, wc *WorkerConn, cid []byte, m Mapping, ip string, port int) *Stream {
	s := &Stream{client: c, cid: cid, mapping: m, srcIP: ip, srcPort: port,
		resumable: wc.resume, sendWnd: wc.sendWndInit, done: make(chan struct{})}
	s.wc.Store(wc)
	if c.statsOn() {
		s.stats = &streamStats{opened: time.Now()}
	}
	if m.VelocitySecret != "" {
		s.vel = newVelocityForwarder(m.VelocitySecret, ip)
	}
	s.cond = sync.NewCond(&s.mu)
	return s
}

// conn returns the stream's current worker conn. May be the original or a
// reattach; callers that send must take one snapshot and use it for the
// whole operation so a concurrent rebind cannot split a write across conns.
func (s *Stream) conn() *WorkerConn { return s.wc.Load() }

func (s *Stream) name() string {
	if wc := s.conn(); wc != nil {
		return fmt.Sprintf("conn%d", wc.id)
	}
	return "conn?"
}

// run dials the destination, optionally writes the PROXY v2 header, then pumps
// destination -> hub (respecting the send window when negotiated).
func (s *Stream) run() {
	dest, err := net.DialTimeout("tcp", s.mapping.Destination, 10*time.Second)
	if err != nil {
		log.Printf("stream %s: dial %s failed: %v", s.name(), s.mapping.Destination, err)
		if wc := s.conn(); wc != nil {
			wc.detach()
			wc.sendRst()
		}
		s.teardown(false)
		return
	}
	if tcp, ok := dest.(*net.TCPConn); ok {
		_ = tcp.SetNoDelay(true)
	}

	if s.mapping.ProxyProtocol {
		if hdr := s.buildProxyHeader(dest); hdr != nil {
			if _, err := dest.Write(hdr); err != nil {
				log.Printf("stream %s: proxy header write: %v", s.name(), err)
			}
		}
	}

	s.mu.Lock()
	if s.closed {
		s.mu.Unlock()
		_ = dest.Close()
		return
	}
	s.dest = dest
	s.connected = true
	if s.finPending {
		// The hub FIN'd while we were still dialing, so gracefulFin could not
		// arm the drain deadline (there was no destination yet). Arm it now,
		// otherwise writeLoop can block on an unresponsive destination forever.
		_ = dest.SetWriteDeadline(time.Now().Add(finDrainTimeout))
	}
	s.cond.Broadcast() // wake writeLoop: queued hub bytes can flow now
	s.mu.Unlock()

	// destination -> hub
	buf := make([]byte, DataChunkSize)
	for {
		n, err := dest.Read(buf)
		if n > 0 {
			chunk := buf[:n]
			if s.vel != nil && !s.vel.Passthrough() {
				fwd, inject := s.vel.ProcessS2C(chunk)
				if len(inject) > 0 {
					s.injectToDest(inject)
				}
				chunk = fwd
			}
			if !s.sendToHub(chunk) {
				break
			}
		}
		if err != nil {
			break
		}
	}
	s.teardown(true)
}

// sendToHub forwards destination bytes to the hub in bounded DATA frames,
// honoring both the send window and the client-wide bandwidth cap.
// Returns false once the stream closed or the worker conn failed.
func (s *Stream) sendToHub(data []byte) bool {
	for len(data) > 0 {
		n := len(data)
		if n > s.client.chunk {
			n = s.client.chunk
		}
		// Credit first, bandwidth second. The reverse order would charge the
		// budget for bytes still parked on an exhausted window, so the client
		// would throttle itself below the configured rate. Holding credit while
		// waiting for tokens is free — credit is per-tunnel, and the hub returns
		// it as it drains data to the player, independent of our pacing.
		if !s.acquireSendWnd(n) {
			return false
		}
		var shaperStall stallClock
		shaperStall.begin(s.stats != nil)
		if !s.client.shaper.Acquire(&s.share, n, s.done) {
			return false
		}
		// Sampled here, before the socket write: emit's write to the hub can
		// block when the hub is not reading, and charging that time to the
		// bandwidth cap would blame maxBandwidth for a hub that is not draining.
		waited := shaperStall.elapsed()
		if !s.emit(data[:n]) {
			return false
		}
		if s.stats != nil {
			s.mu.Lock()
			// Separated from the window stall on purpose: this one says the
			// configured cap is the binding constraint, and raising maxBandwidth
			// is the fix. The window stall says the opposite.
			s.stats.shaperStall += waited
			s.stats.bytesUp += int64(n)
			s.mu.Unlock()
		}
		data = data[n:]
	}
	return true
}

// emit records a chunk for possible retransmission and writes it to the current
// worker conn. Returns false once the stream is finished with.
//
// The record is taken first and unconditionally. A write that fails on a dying
// connection has already spent window and may have put part of the frame on the
// wire, so the only trustworthy account of what the peer still owes us is the
// one taken before the attempt. That also makes a failure survivable: while the
// stream can still be resumed the bytes are already safe, and the reattach
// replays them from wherever the hub says it got to.
func (s *Stream) emit(chunk []byte) bool {
	s.sendMu.Lock()
	if s.resumable {
		// Reclaim what the hub has credited before growing the buffer, so the
		// outstanding region stays bounded by one window.
		s.un.advance(s.ackedOffset.Load())
		s.un.append(chunk)
	}
	wc := s.conn()
	var err error
	if wc != nil {
		err = wc.sendData(chunk)
	} else {
		err = errPoolClosed
	}
	s.sendMu.Unlock()

	if err == nil {
		return true
	}
	if !s.resumable {
		return false
	}
	// The conn is gone but the stream is not: readLoop parks it and a reattach
	// replays the buffer. Keep pumping the destination — acquireSendWnd stops us
	// once a full window is outstanding, so nothing is lost and nothing grows
	// without bound.
	return !s.isClosed()
}

func (s *Stream) isClosed() bool {
	s.mu.Lock()
	defer s.mu.Unlock()
	return s.closed
}

// writeLoop is the only writer to the destination. It drains the receive queue,
// credits the hub as bytes land on the destination socket, and performs the
// deferred graceful close when a FIN arrived with data still queued.
func (s *Stream) writeLoop() {
	for {
		s.mu.Lock()
		for !s.closed && (!s.connected || (len(s.q) == 0 && !s.finPending)) {
			s.cond.Wait()
		}
		if s.closed {
			s.mu.Unlock()
			return
		}
		if len(s.q) == 0 { // finPending and fully drained
			s.mu.Unlock()
			s.teardown(false)
			return
		}
		e := s.q[0]
		s.q = s.q[1:]
		if e.fromHub {
			s.qBytes -= len(e.data)
		}
		dest := s.dest
		s.mu.Unlock()

		if _, err := dest.Write(e.data); err != nil {
			s.teardown(true)
			return
		}
		if e.fromHub {
			s.mu.Lock()
			s.deliveredOffset += int64(len(e.data))
			if s.stats != nil {
				s.stats.bytesDown += int64(len(e.data))
			}
			s.mu.Unlock()
			s.credit(len(e.data))
		}
	}
}

// deliverFromHub enqueues hub bytes for the destination. Called from the worker
// readLoop; it never blocks — a peer that exceeds the advertised window is a
// protocol violator and gets the tunnel reset.
func (s *Stream) deliverFromHub(data []byte) {
	if s.vel != nil {
		s.vel.ObserveC2S(data) // observation only; bytes still forwarded verbatim
	}
	s.mu.Lock()
	if s.closed || s.finPending {
		s.mu.Unlock()
		return
	}
	if s.qBytes+len(data) > s.client.streamWnd {
		s.mu.Unlock()
		log.Printf("stream %s: peer exceeded flow-control window; resetting", s.name())
		if wc := s.conn(); wc != nil {
			wc.detach()
			wc.sendRst()
		}
		s.teardown(false)
		return
	}
	s.q = append(s.q, qentry{data: data, fromHub: true})
	s.qBytes += len(data)
	// Accepted, not delivered: from here the bytes are ours to write, and the
	// only way we fail to is by destroying the stream — which also ends any
	// prospect of resuming it. That makes this a sound reattach coordinate.
	s.acceptedOffset += int64(len(data))
	if s.stats != nil && s.qBytes > s.stats.qPeak {
		// How close the receive queue came to the advertised window: near it
		// means the destination is the slow party.
		s.stats.qPeak = s.qBytes
	}
	s.cond.Broadcast()
	s.mu.Unlock()
}

// injectToDest queues locally generated bytes (the velocity login response)
// for the destination, outside flow-control accounting.
func (s *Stream) injectToDest(data []byte) {
	s.mu.Lock()
	if s.closed || s.finPending {
		s.mu.Unlock()
		return
	}
	s.q = append(s.q, qentry{data: data})
	s.cond.Broadcast()
	s.mu.Unlock()
}

// acquireSendWnd blocks until the stream may send n more bytes to the hub.
// Returns false if the stream closed while waiting.
func (s *Stream) acquireSendWnd(n int) bool {
	s.mu.Lock()
	defer s.mu.Unlock()
	// Timed only when it actually blocks, so a stream that never runs out of
	// credit never reads the clock. A large windowStall is the signal that the
	// peer is not draining to its terminal socket — the bottleneck is past the
	// tunnel, not in it.
	var stall stallClock
	for !s.closed && s.sendWnd < n {
		stall.begin(s.stats != nil)
		s.cond.Wait()
	}
	if s.stats != nil {
		s.stats.windowStall += stall.elapsed()
	}
	if s.closed {
		return false
	}
	s.sendWnd -= n
	return true
}

func (s *Stream) grantSendWnd(delta int) {
	// The running sum doubles as the acked offset: the hub grants credit exactly
	// as bytes reach the player socket, so a byte that has been credited can
	// never need retransmitting. Advanced without a lock so the worker readLoop
	// never blocks behind a send in progress.
	s.ackedOffset.Add(int64(delta))
	s.mu.Lock()
	s.sendWnd += delta
	s.cond.Broadcast()
	s.mu.Unlock()
}

// credit accounts bytes drained to the destination and grants the hub more
// window once half of our receive window has been consumed.
// While parked the grant is only withheld, never dropped: consumed keeps
// accumulating and a reattach flushes it on the new conn. Resetting it would
// destroy up to half a window of credit per outage, and after a few flaps the
// stream would throttle to a crawl.
func (s *Stream) credit(n int) {
	s.mu.Lock()
	s.consumed += n
	if s.closed || s.parked || s.consumed*2 < s.client.streamWnd {
		s.mu.Unlock()
		return
	}
	delta := s.consumed
	s.consumed = 0
	s.mu.Unlock()
	if wc := s.conn(); wc != nil {
		wc.sendWndUpdate(delta)
	}
}

func (s *Stream) buildProxyHeader(dest net.Conn) []byte {
	srcIP := net.ParseIP(s.srcIP)
	if srcIP == nil {
		return nil
	}
	dstTCP, ok := dest.RemoteAddr().(*net.TCPAddr)
	if !ok {
		return nil
	}
	return BuildProxyV2(srcIP, s.srcPort, dstTCP.IP, dstTCP.Port)
}

// finDrainTimeout bounds how long a FIN'd stream may keep draining its queue
// into the destination, so a dead destination cannot hold the stream forever.
const finDrainTimeout = 10 * time.Second

// gracefulFin marks the hub-side close; writeLoop finishes the queue and then
// tears the stream down.
func (s *Stream) gracefulFin() {
	s.mu.Lock()
	if s.closed || s.finPending {
		s.mu.Unlock()
		return
	}
	s.finPending = true
	if s.dest != nil {
		_ = s.dest.SetWriteDeadline(time.Now().Add(finDrainTimeout))
	}
	s.cond.Broadcast()
	s.mu.Unlock()
}

// teardown closes the stream immediately; notifyHub sends a FIN when true.
// Idempotent; wakes every goroutine parked on the stream.
func (s *Stream) teardown(notifyHub bool) {
	// A parked stream owes the hub a FIN it cannot send: the only conn it has is
	// the one that just failed. Keep it alive so the reattach can deliver it and
	// the player gets a clean disconnect, rather than hanging until the hub's
	// grace expires.
	if notifyHub && s.noteFinWhileParked() {
		return
	}
	s.mu.Lock()
	if s.closed {
		s.mu.Unlock()
		return
	}
	s.closed = true
	dest := s.dest
	close(s.done) // guarded by the idempotence check above, so exactly once
	s.cond.Broadcast()
	s.mu.Unlock()

	// The stream can no longer be resumed, so the retransmit buffer is dead
	// weight — up to a full window of it per stream.
	s.sendMu.Lock()
	s.un.reset()
	s.sendMu.Unlock()

	if dest != nil {
		_ = dest.Close()
	}
	if wc := s.conn(); wc != nil {
		wc.detach()
		if notifyHub {
			wc.sendFin()
		}
	}
	s.logSummary()
}