package main

import (
	"sync/atomic"
	"testing"
	"time"
)

// Batches land in the bucket of the epoch they were read in, summed, and
// newest follows the highest epoch committed.
func TestRxCommitBucketsByEpoch(t *testing.T) {
	var s rxStats
	s.commit(0, 1, 100)
	s.commit(0, 1, 200)
	s.commit(1, 1, 300)

	if f, b := s.bucket(0); f != 2 || b != 300 {
		t.Errorf("epoch 0 = %d frames, %d bytes; want 2, 300", f, b)
	}
	if f, b := s.bucket(1); f != 1 || b != 300 {
		t.Errorf("epoch 1 = %d frames, %d bytes; want 1, 300", f, b)
	}
	if got := s.newest.Load(); got != 1 {
		t.Errorf("newest = %d, want 1", got)
	}
}

// A slot coming round again belongs to its new epoch, and the epoch it replaced
// reports nothing rather than the stale counts.
func TestRxBucketWraps(t *testing.T) {
	var s rxStats
	s.commit(0, 1, 100)
	s.commit(rateBuckets, 1, 200)

	if f, b := s.bucket(0); f != 0 || b != 0 {
		t.Errorf("evicted epoch = %d frames, %d bytes; want 0, 0", f, b)
	}
	if f, b := s.bucket(rateBuckets); f != 1 || b != 200 {
		t.Errorf("new epoch = %d frames, %d bytes; want 1, 200", f, b)
	}
}

// Every stream's frontier is past the epoch being read, so the bucket is
// complete across the board and the rate comes from the window behind the
// lowest frontier.
func TestDirectionRateReadsOneBucketBack(t *testing.T) {
	d := &direction{masked: &atomic.Bool{}, rxStats: []*rxStats{{}, {}}}
	d.rxStats[0].commit(1, 1, 500)
	d.rxStats[1].commit(1, 1, 700)
	d.rxStats[0].commit(2, 1, 900)
	d.rxStats[1].commit(2, 1, 1100)

	d.readRateBucket(time.Now())
	if d.rateFrames != 2 || d.rateBytes != 1200 {
		t.Errorf("rate bucket = %d frames, %d bytes; want the completed epoch, 2 and 1200",
			d.rateFrames, d.rateBytes)
	}
}

// Reading a bucket is a claim that every stream has delivered through it, so a
// stream still short of the epoch holds the read back rather than having its
// half-filled bucket summed.
func TestDirectionRateWaitsForSlowestStream(t *testing.T) {
	d := &direction{masked: &atomic.Bool{}, rxStats: []*rxStats{{}, {}}}
	d.rxStats[0].commit(1, 1, 500)
	d.rxStats[0].commit(2, 1, 900)

	// One stream has never delivered at all, so there is no epoch every stream
	// has reached and nothing to read.
	d.readRateBucket(time.Now())
	if d.rateFrames != 0 || d.rateBytes != 0 {
		t.Errorf("rate = %d frames, %d bytes; want nothing while a stream has no frontier",
			d.rateFrames, d.rateBytes)
	}

	// The straggler is still filling the epoch the leader finished, so its
	// bucket must not be read yet.
	d.rxStats[1].commit(1, 1, 700)
	d.readRateBucket(time.Now())
	if d.rateFrames != 0 || d.rateBytes != 0 {
		t.Errorf("rate = %d frames, %d bytes; want nothing while a stream is still filling the epoch",
			d.rateFrames, d.rateBytes)
	}

	// Once it moves past, the bucket is complete for both streams and is read
	// whole.
	d.rxStats[1].commit(2, 1, 1100)
	d.readRateBucket(time.Now())
	if d.rateFrames != 2 || d.rateBytes != 1200 {
		t.Errorf("rate = %d frames, %d bytes; want the whole completed epoch, 2 and 1200",
			d.rateFrames, d.rateBytes)
	}
}

// An epoch only advances when a frame is read, so a frontier that stops moving
// is a quiet wire and must not keep reporting the last bucket.
func TestDirectionRateGoesStale(t *testing.T) {
	d := &direction{masked: &atomic.Bool{}, rxStats: []*rxStats{{}}}
	d.rxStats[0].commit(1, 1, 500)
	d.rxStats[0].commit(2, 1, 900)

	now := time.Now()
	d.readRateBucket(now)
	if d.rateFrames == 0 {
		t.Fatal("expected a rate while the epoch was still moving")
	}
	d.readRateBucket(now.Add(2 * rateStale))
	if d.rateFrames != 0 || d.rateBytes != 0 {
		t.Errorf("stale rate = %d frames, %d bytes; want 0, 0", d.rateFrames, d.rateBytes)
	}
}

// Each bucket entering the settled window donates once, as it arrives.
func settleAll(frames, bytes []uint64) {
	for i := 1; i <= len(frames); i++ {
		fillBack(frames[:i], bytes[:i])
	}
}

// A donated byte must never display twice: once a burst's excess has refilled
// an earlier deficit, the burst bucket itself reaches the display slot
// trimmed. The pre-settled-window implementation recomputed the smear from
// the raw ring per sample and displayed the excess again — 2C here, and >20G
// on the wire.
func TestDirectionRateNeverRedisplaysDonatedExcess(t *testing.T) {
	d := &direction{masked: &atomic.Bool{}, rxStats: []*rxStats{{}}}
	c := bucketWireCap
	seq := []uint64{c, 0, 2 * c, c, c, c, c, c}
	now := time.Now()
	for i, b := range seq {
		d.rxStats[0].commit(int64(i+1), 0, b)
		d.readRateBucket(now)
		if d.rateBytes > c {
			t.Errorf("after epoch %d: displayed %d bytes, over line rate %d", i+1, d.rateBytes, c)
		}
	}
	// Jitter settled: the display holds flat line rate, not just under it.
	if d.rateBytes != c {
		t.Errorf("settled display = %d bytes, want exactly %d", d.rateBytes, c)
	}
}

// A stall's deficit is refilled exactly by the burst that drains its backlog,
// so host jitter flattens to line rate.
func TestSmearRepairsStallBurst(t *testing.T) {
	c := bucketWireCap
	bytes := []uint64{c, 0, 0, 3 * c}
	frames := make([]uint64, 4)
	settleAll(frames, bytes)
	for i, want := range []uint64{c, c, c, c} {
		if bytes[i] != want {
			t.Errorf("bucket %d = %d bytes, want %d", i, bytes[i], want)
		}
	}
}

// Excess refills the nearest earlier deficit, so a genuine wire dip — a
// deficit with no matching excess — keeps its full size in its own bucket.
func TestSmearLeavesRealLossInPlace(t *testing.T) {
	c := bucketWireCap
	bytes := []uint64{0, c, 0, 2 * c}
	frames := make([]uint64, 4)
	settleAll(frames, bytes)
	for i, want := range []uint64{0, c, c, c} {
		if bytes[i] != want {
			t.Errorf("bucket %d = %d bytes, want %d", i, bytes[i], want)
		}
	}
}

// A deficit after the excess is the host not having read those frames yet, not
// jitter to repair: nothing ever moves forward.
func TestSmearNeverMovesForward(t *testing.T) {
	c := bucketWireCap
	bytes := []uint64{c, 2 * c, 0}
	frames := make([]uint64, 3)
	settleAll(frames, bytes)
	for i, want := range []uint64{c, 2 * c, 0} {
		if bytes[i] != want {
			t.Errorf("bucket %d = %d bytes, want %d", i, bytes[i], want)
		}
	}
}

// Truly below line rate has nothing to move and displays as-is.
func TestSmearLeavesBelowRateAlone(t *testing.T) {
	c := bucketWireCap
	bytes := []uint64{c / 2, c / 4, c / 2}
	frames := []uint64{100, 50, 100}
	want := []uint64{c / 2, c / 4, c / 2}
	settleAll(frames, bytes)
	for i := range bytes {
		if bytes[i] != want[i] {
			t.Errorf("bucket %d = %d bytes, want untouched %d", i, bytes[i], want[i])
		}
	}
}

// Frames ride along with the bytes in the donor's own proportion, and nothing
// is invented or lost in the move.
func TestSmearConservesFramesAndBytes(t *testing.T) {
	c := bucketWireCap
	donorFrames := uint64(1000)
	donorBytes := 2*c - donorFrames*wireOverhead
	frames := []uint64{0, donorFrames}
	bytes := []uint64{0, donorBytes}
	settleAll(frames, bytes)

	if tf := frames[0] + frames[1]; tf != donorFrames {
		t.Errorf("total frames = %d, want %d conserved", tf, donorFrames)
	}
	if tb := bytes[0] + bytes[1]; tb != donorBytes {
		t.Errorf("total bytes = %d, want %d conserved", tb, donorBytes)
	}
	if got := bytes[0] + frames[0]*wireOverhead; got != c {
		t.Errorf("refilled bucket = %d wire bytes, want exactly %d", got, c)
	}
	if frames[0] != donorFrames/2 {
		t.Errorf("moved frames = %d, want the donor's proportion %d", frames[0], donorFrames/2)
	}
	if got := bytes[1] + frames[1]*wireOverhead; got != c {
		t.Errorf("donor left = %d wire bytes, want trimmed to %d", got, c)
	}
}

// The window is a rolling sum of one span of deltas: what leaves the ring
// leaves the sum, so after a wrap only the last len slots are counted.
func TestErrsRingRollingSum(t *testing.T) {
	r := newErrsRing(3)
	for i := 1; i <= 5; i++ {
		r.push(errs{lost: uint64(i), link: 1})
	}
	want := errs{lost: 3 + 4 + 5, link: 3}
	if r.sum != want {
		t.Errorf("sum = %+v, want %+v", r.sum, want)
	}
}

// Which delta feeds which bucket is the whole taxonomy the panel reports, so
// it is pinned here rather than left to whoever reads classify next.
func TestClassifyBuckets(t *testing.T) {
	cur := sample{
		lost: 1, late: 7,
		crcErr: 2, badMagic: 3, badLen: 4, badHdr: 10,
		txErrs: 6, rxErrs: 5,
	}
	got := classify(cur, sample{}, 9, 8)
	want := errs{lost: 1, corrupt: 2 + 3 + 4 + 10, link: 8 + 5, internal: 9 + 7 + 6}
	if got != want {
		t.Errorf("classify = %+v, want %+v", got, want)
	}
}

// A masked tick's faults stop at the gate while traffic and the window clock
// keep moving; the tick after the mask lifts aggregates again.
func TestMaskedTickAggregatesNothing(t *testing.T) {
	var masked atomic.Bool
	d := &direction{masked: &masked, window: newErrsRing(8)}

	masked.Store(true)
	if e := d.aggregate(10, 640, errs{lost: 5, link: 2}); e != (errs{}) {
		t.Errorf("accepted %+v from a masked tick, want nothing", e)
	}
	v := d.displayView()
	if v.since != (errs{}) || v.window != (errs{}) {
		t.Errorf("since = %+v window = %+v after a masked tick, want both empty", v.since, v.window)
	}
	if v.rxFrames != 10 || v.rxBytes != 640 {
		t.Errorf("traffic = %d frames %d bytes, want it counted through the mask", v.rxFrames, v.rxBytes)
	}

	masked.Store(false)
	d.aggregate(0, 0, errs{lost: 1})
	v = d.displayView()
	if v.since.lost != 1 || v.window.lost != 1 {
		t.Errorf("since.lost = %d window.lost = %d after the mask lifted, want 1 and 1", v.since.lost, v.window.lost)
	}
}

// A reset zeroes the since-reset sums; the rolling window is about now rather
// than since the reset, so it keeps what it held.
func TestResetZeroesSinceAndKeepsWindow(t *testing.T) {
	var masked atomic.Bool
	d := &direction{masked: &masked, window: newErrsRing(8)}
	d.aggregate(100, 6400, errs{lost: 3})
	d.reset()

	v := d.displayView()
	if v.rxFrames != 0 || v.rxBytes != 0 {
		t.Errorf("traffic = %d frames %d bytes after reset, want 0 and 0", v.rxFrames, v.rxBytes)
	}
	if v.since != (errs{}) {
		t.Errorf("since = %+v after reset, want empty", v.since)
	}
	if v.window.lost != 3 {
		t.Errorf("window.lost = %d after reset, want the pre-reset window kept", v.window.lost)
	}
}