package main

// Per-connection transport evidence for the session log line.
//
// The frame writer can only see that a stream write took long; it cannot
// tell WHY. Three different things gate a QUIC write and they need three
// different fixes: the congestion controller's window (BBR's recovery
// window, see bbr/PROVENANCE.md), the controller's pacing rate (an
// app-limited sender cannot raise BBR's bandwidth estimate above what it
// offers), and the peer's flow-control credit (Chrome's MAX_DATA /
// MAX_STREAM_DATA, which only advance as its reader drains streams). This
// file gathers all three next to each other: the sender's own model from
// the vendored bbr package, and from quic-go's qlog stream the packets it
// declared lost and the DATA_BLOCKED / STREAM_DATA_BLOCKED frames it sent -
// a sender emits those exactly when flow control, not congestion control,
// stopped it.

import (
	"context"
	"fmt"
	"net"
	"sync"
	"sync/atomic"
	"time"

	"github.com/apernet/quic-go"
	"github.com/apernet/quic-go/congestion"
	"github.com/apernet/quic-go/qlog"
	"github.com/apernet/quic-go/qlogwriter"
	"war3replays/w3cs-wt-bridge/bbr"
)

type linkStatsKey struct{}

// bbrObserver is what the bridge needs from a congestion controller it
// installed; the stock controller offers nothing and logs as such.
type bbrObserver interface {
	Snapshot() bbr.Snapshot
	TakeMaxAckStall() time.Duration
}

type linkStats struct {
	remote net.Addr

	mu  sync.Mutex
	bbr bbrObserver

	packetsSent   atomic.Uint64
	packetsLost   atomic.Uint64
	dataBlocked   atomic.Uint64
	streamBlocked atomic.Uint64
	// quic-go's own view of the window and in-flight at the last metrics
	// update, as a cross-check on the controller's snapshot.
	cwnd     atomic.Int64
	inFlight atomic.Int64
	minRTT   atomic.Int64 // nanoseconds
	srtt     atomic.Int64 // nanoseconds
	// Bits per second the connection delivered lately (acked bytes over
	// the last few one-second samples), for the frame writer's backlog
	// budget.
	deliveredBps atomic.Int64
}

// delivered is the lately delivered rate in bits per second, 0 before the
// first sample or without a holder.
func (s *linkStats) delivered() int64 {
	if s == nil {
		return 0
	}
	return s.deliveredBps.Load()
}

// newLinkStats attaches a fresh holder to the connection context. Installed
// as quic.Transport.ConnContext, so the holder is visible to the qlog
// tracer, to http3's ConnContext (useBBR) and to every request context.
func newLinkStats(ctx context.Context, info *quic.ClientInfo) (
	context.Context, error) {
	return context.WithValue(ctx, linkStatsKey{},
		&linkStats{remote: info.RemoteAddr}), nil
}

func linkStatsFrom(ctx context.Context) *linkStats {
	stats, _ := ctx.Value(linkStatsKey{}).(*linkStats)
	return stats
}

func (s *linkStats) setBBR(observer bbrObserver) {
	s.mu.Lock()
	defer s.mu.Unlock()
	s.bbr = observer
}

// takeMaxAckStall is the longest ack silence beyond a round trip, with
// data in flight, since the previous call; 0 without a controller to ask.
func (s *linkStats) takeMaxAckStall() time.Duration {
	if s == nil {
		return 0
	}
	s.mu.Lock()
	observer := s.bbr
	s.mu.Unlock()
	if observer == nil {
		return 0
	}
	return observer.TakeMaxAckStall()
}

// deltas is the counters since the previous call, for the periodic line.
type linkDeltas struct {
	lost, dataBlocked, streamBlocked uint64
}

type linkReporter struct {
	stats             *linkStats
	lastLost          uint64
	lastDataBlocked   uint64
	lastStreamBlocked uint64
	lastAcked         congestion.ByteCount
	lastAckedAt       time.Time
}

func newLinkReporter(stats *linkStats) *linkReporter {
	return &linkReporter{stats: stats}
}

// sampleDelivered folds the controller's acked-bytes total since the
// previous call into the delivered rate (a ~3-sample average: a stall
// delays bytes without losing them, so its flush lands in the next sample
// and the average is unmoved). Called once a second by the session.
func (r *linkReporter) sampleDelivered() {
	if r == nil || r.stats == nil {
		return
	}
	s := r.stats
	s.mu.Lock()
	observer := s.bbr
	s.mu.Unlock()
	if observer == nil {
		return
	}
	acked := observer.Snapshot().BytesAcked
	now := time.Now()
	if !r.lastAckedAt.IsZero() && acked >= r.lastAcked {
		if elapsed := now.Sub(r.lastAckedAt).Seconds(); elapsed >= 0.2 {
			sample := float64(acked-r.lastAcked) * 8 / elapsed
			previous := float64(s.deliveredBps.Load())
			if previous > 0 {
				sample = previous*0.65 + sample*0.35
			}
			s.deliveredBps.Store(int64(sample))
		}
	}
	r.lastAcked, r.lastAckedAt = acked, now
}

// line renders the transport evidence for one log interval, or "" when the
// connection has no holder (a listener not set up through newLinkStats).
func (r *linkReporter) line() string {
	if r == nil || r.stats == nil {
		return ""
	}
	s := r.stats
	lost, dataBlocked, streamBlocked := s.packetsLost.Load(),
		s.dataBlocked.Load(), s.streamBlocked.Load()
	d := linkDeltas{lost - r.lastLost, dataBlocked - r.lastDataBlocked,
		streamBlocked - r.lastStreamBlocked}
	r.lastLost, r.lastDataBlocked, r.lastStreamBlocked =
		lost, dataBlocked, streamBlocked

	out := fmt.Sprintf(" link lost=%d dataBlocked=%d streamBlocked=%d "+
		"cwnd=%dKiB inflight=%dKiB minRtt=%dms srtt=%dms deliv=%.2fMbps",
		d.lost, d.dataBlocked, d.streamBlocked,
		s.cwnd.Load()/1024, s.inFlight.Load()/1024,
		time.Duration(s.minRTT.Load()).Milliseconds(),
		time.Duration(s.srtt.Load()).Milliseconds(),
		float64(s.deliveredBps.Load())/1e6)

	s.mu.Lock()
	observer := s.bbr
	s.mu.Unlock()
	if observer == nil {
		return out + " bbr=off"
	}
	snap := observer.Snapshot()
	return out + fmt.Sprintf(" bbr mode=%s rec=%t full=%t appLim=%t "+
		"bw=%.2fMbps pace=%.2fMbps cwnd=%dKiB rwnd=%dKiB inflight=%dKiB "+
		"minRtt=%dms",
		snap.Mode, snap.InRecovery, snap.AtFullBW, snap.AppLimited,
		float64(snap.Bandwidth)/1e6, float64(snap.PacingRate)/1e6,
		snap.Cwnd/1024, snap.RecoveryWindow/1024, snap.BytesInFlight/1024,
		snap.MinRTT.Milliseconds())
}

// How stale a controller snapshot may be and still describe the link. BBR
// publishes one per congestion event (every ack), so a gap means the
// connection went quiet, not that the model aged.
const linkReportMaxAge = 3 * time.Second

// report renders the link estimate the relay's governor consumes, or
// ("", false) while the controller has no bandwidth sample yet.
//
// The governor used to learn capacity from the browser's receive rate, which
// only ever measures what the relay chose to send: after one cut the
// "capacity" is the cut cadence, the next cut lowers it again, and the
// session walks to the floor (measured 2026-08-22: 70 cuts in four minutes
// on a path BBR was delivering 20+ Mbps over).
//
// Two numbers travel, because one is not enough. bwBps is BBR's estimate: a
// max filter over ten round trips, which an app-limited sender can only
// raise by offering more - the right thing to climb toward, and on a path
// that delivers in flushes (a Wi-Fi client's ~0.5 s holds) an optimistic
// one: it read 27-31 Mbps while frames blocked 300-550 ms and half were
// dropped. ackedBytes is the lifetime delivered total; its delta over the
// report interval is what the path actually carried, so it is the right
// thing to cut toward.
//
// qdrop is the writer's lifetime count of dependent frames dropped
// latest-wins. Every one of them breaks the encoder's assumption that the
// browser holds what recent frames defined, so the relay asks the game for
// a geometry anchor when it sees the count move (w3cs_webrtc_relay.cpp
// note_bridge_link) instead of leaving the next frames unresolvable until
// the GOP's own anchor, up to eight frames later.
func (r *linkReporter) report(qdrop uint64) (string, bool) {
	if r == nil || r.stats == nil {
		return "", false
	}
	s := r.stats
	s.mu.Lock()
	observer := s.bbr
	s.mu.Unlock()
	if observer == nil {
		return "", false
	}
	snap := observer.Snapshot()
	if snap.Bandwidth <= 0 {
		return "", false
	}
	return fmt.Sprintf(`{"t":"link","bwBps":%d,"minRttMs":%d,"srttMs":%d,`+
		`"sentBytes":%d,"ackedBytes":%d,"lostBytes":%d,`+
		`"sent":%d,"lost":%d,"qdrop":%d,"appLimited":%t}`,
		uint64(snap.Bandwidth), snap.MinRTT.Milliseconds(),
		time.Duration(s.srtt.Load()).Milliseconds(),
		snap.BytesSent, snap.BytesAcked, snap.BytesLost,
		s.packetsSent.Load(), s.packetsLost.Load(), qdrop,
		snap.AppLimited), true
}

// linkTrace is the qlog sink quic-go writes this connection's events to.
// It keeps counters instead of a file: the events of interest are the
// handful that explain a blocked stream write.
type linkTrace struct{ stats *linkStats }

func newLinkTrace(ctx context.Context, _ bool, _ quic.ConnectionID) qlogwriter.Trace {
	stats := linkStatsFrom(ctx)
	if stats == nil {
		return nil
	}
	return &linkTrace{stats: stats}
}

func (t *linkTrace) AddProducer() qlogwriter.Recorder { return t }
func (t *linkTrace) SupportsSchemas(string) bool      { return true }
func (t *linkTrace) Close() error                     { return nil }

func (t *linkTrace) RecordEvent(event qlogwriter.Event) {
	switch e := event.(type) {
	case qlog.PacketSent:
		t.stats.packetsSent.Add(1)
		t.countBlocked(e.Frames)
	case *qlog.PacketSent:
		t.stats.packetsSent.Add(1)
		t.countBlocked(e.Frames)
	case qlog.PacketLost, *qlog.PacketLost:
		t.stats.packetsLost.Add(1)
	case qlog.MetricsUpdated:
		t.metrics(e)
	case *qlog.MetricsUpdated:
		t.metrics(*e)
	}
}

func (t *linkTrace) countBlocked(frames []qlog.Frame) {
	for _, frame := range frames {
		switch frame.Frame.(type) {
		case *qlog.DataBlockedFrame, qlog.DataBlockedFrame:
			t.stats.dataBlocked.Add(1)
		case *qlog.StreamDataBlockedFrame, qlog.StreamDataBlockedFrame:
			t.stats.streamBlocked.Add(1)
		}
	}
}

func (t *linkTrace) metrics(e qlog.MetricsUpdated) {
	if e.CongestionWindow > 0 {
		t.stats.cwnd.Store(int64(e.CongestionWindow))
	}
	t.stats.inFlight.Store(int64(e.BytesInFlight))
	if e.MinRTT > 0 {
		t.stats.minRTT.Store(int64(e.MinRTT))
	}
	if e.SmoothedRTT > 0 {
		t.stats.srtt.Store(int64(e.SmoothedRTT))
	}
}