package gateway

import (
	"sort"
	"testing"
	"time"
)

type syntheticPacerDelivery struct {
	at    time.Time
	flow  string
	bytes int64
}

func TestFairPacerEightFlowSharesAndCapacitySteps(t *testing.T) {
	start := time.Date(2026, time.January, 1, 0, 0, 0, 0, time.UTC)
	flows := []string{"one", "two", "three", "four", "five", "six", "seven", "eight"}
	pacer := newFairPacer(8000)
	next := make(map[string]time.Time, len(flows))
	baseline := runSyntheticPacer(pacer, start, start.Add(60*time.Second), flows, next)
	assertSyntheticFairness(t, baseline, flows)
	assertSyntheticCap(t, baseline, 1_000_000)

	pacer.setKbps(6000)
	quarter := runSyntheticPacer(pacer, start.Add(60*time.Second), start.Add(70*time.Second), flows, next)
	assertSyntheticFairness(t, quarter, flows)
	assertSyntheticCap(t, quarter, 750_000)

	pacer.setKbps(4000)
	half := runSyntheticPacer(pacer, start.Add(70*time.Second), start.Add(80*time.Second), flows, next)
	assertSyntheticFairness(t, half, flows)
	assertSyntheticCap(t, half, 500_000)
}

func TestFairPacerBoundsCatchupAfterHostStall(t *testing.T) {
	start := time.Date(2026, time.January, 1, 0, 0, 0, 0, time.UTC)
	pacer := newFairPacer(8000)
	_ = pacer.reserveAt(start, "one", 1000)
	resumed := start.Add(100 * time.Millisecond)
	next := pacer.reserveAt(resumed, "one", 1000)
	if next.Before(resumed.Add(-fairPacerMaximumCatchup)) || next.After(resumed.Add(10*time.Millisecond)) {
		t.Fatalf("post-stall reservation = %s, want bounded catchup near %s", next, resumed)
	}
	pacer.mu.Lock()
	debt := pacer.flows["one"].debt
	pacer.mu.Unlock()
	if debt <= 0 || debt > nativeApolloVideoQueueLatency-fairPacerMaximumCatchup {
		t.Fatalf("post-stall debt = %s, want bounded valid schedule debt", debt)
	}
}

func TestFairPacerRepaysBoundedDebtAfterHostStall(t *testing.T) {
	start := time.Date(2026, time.January, 1, 0, 0, 0, 0, time.UTC)
	pacer := newFairPacer(8000)
	next := make(map[string]time.Time)
	deliveries := runSyntheticPacerWithStall(
		pacer, start, start.Add(6*time.Second), []string{"one"}, next,
		start.Add(time.Second), 100*time.Millisecond,
	)
	if total := syntheticDeliveryBytes(deliveries); total < 5_990_000 || total > 6_010_000 {
		t.Fatalf("post-stall delivery bytes = %d, want nominal throughput after bounded debt repayment", total)
	}
	pacer.mu.Lock()
	remaining := pacer.flows["one"].debt
	pacer.mu.Unlock()
	if remaining != 0 {
		t.Fatalf("post-stall debt = %s after repayment, want zero", remaining)
	}
	assertSyntheticCap(t, deliveries, 1_000_000)
	t.Logf("single-flow debt repaid: bytes=%d remaining=%s", syntheticDeliveryBytes(deliveries), remaining)
}

func TestFairPacerRepaysSimultaneousEightFlowDebtAcrossCapacitySteps(t *testing.T) {
	start := time.Date(2026, time.January, 1, 0, 0, 0, 0, time.UTC)
	flows := []string{"one", "two", "three", "four", "five", "six", "seven", "eight"}

	tests := []struct {
		name           string
		kbps           int64
		bytesPerSecond int64
		minimumBytes   int64
	}{
		{name: "baseline", kbps: 8000, bytesPerSecond: 1_000_000, minimumBytes: 9_980_000},
		{name: "quarter", kbps: 6000, bytesPerSecond: 750_000, minimumBytes: 7_480_000},
		{name: "half", kbps: 4000, bytesPerSecond: 500_000, minimumBytes: 4_980_000},
	}
	for _, test := range tests {
		pacer := newFairPacer(8000)
		next := make(map[string]time.Time, len(flows))
		_ = runSyntheticPacer(pacer, start, start.Add(time.Second), flows, next)
		resumed := start.Add(1100 * time.Millisecond)
		for _, flow := range flows {
			next[flow] = pacer.reserveAt(resumed, flow, 1000)
		}
		assertSyntheticDebt(t, pacer, flows, true)
		pacer.setKbps(test.kbps)
		deliveries := runSyntheticPacer(pacer, resumed, resumed.Add(10*time.Second), flows, next)
		if total := syntheticDeliveryBytes(deliveries); total < test.minimumBytes {
			t.Fatalf("%s post-stall delivery bytes = %d, want at least %d", test.name, total, test.minimumBytes)
		}
		assertSyntheticFairness(t, deliveries, flows)
		assertSyntheticCap(t, deliveries, test.bytesPerSecond)
		assertSyntheticDebt(t, pacer, flows, false)
		t.Logf("%s eight-flow debt repaid: bytes=%d cap=%d", test.name, syntheticDeliveryBytes(deliveries), test.bytesPerSecond*5*105/100)
	}
}

func assertSyntheticDebt(t *testing.T, pacer *fairPacer, flows []string, wantDebt bool) {
	t.Helper()
	pacer.mu.Lock()
	defer pacer.mu.Unlock()
	for _, flow := range flows {
		debt := pacer.flows[flow].debt
		if wantDebt && (debt <= 0 || debt > nativeApolloVideoQueueLatency-fairPacerMaximumCatchup) {
			t.Fatalf("flow %s active debt = %s, want bounded nonzero debt", flow, debt)
		}
		if !wantDebt && debt != 0 {
			t.Fatalf("flow %s debt = %s after repayment, want zero", flow, debt)
		}
	}
}

func runSyntheticPacer(pacer *fairPacer, start, end time.Time, flows []string, next map[string]time.Time) []syntheticPacerDelivery {
	const packetBytes = 1000
	for _, flow := range flows {
		if next[flow].IsZero() {
			next[flow] = pacer.reserveAt(start, flow, packetBytes)
		}
	}
	var deliveries []syntheticPacerDelivery
	for {
		flow := ""
		at := end.Add(time.Nanosecond)
		for _, candidate := range flows {
			if next[candidate].Before(at) {
				flow, at = candidate, next[candidate]
			}
		}
		if at.After(end) {
			return deliveries
		}
		deliveries = append(deliveries, syntheticPacerDelivery{at: at, flow: flow, bytes: packetBytes})
		next[flow] = pacer.reserveAt(at, flow, packetBytes)
	}
}

func runSyntheticPacerWithStall(pacer *fairPacer, start, end time.Time, flows []string, next map[string]time.Time, stallAt time.Time, stall time.Duration) []syntheticPacerDelivery {
	const packetBytes = 1000
	for _, flow := range flows {
		if next[flow].IsZero() {
			next[flow] = pacer.reserveAt(start, flow, packetBytes)
		}
	}
	now := start
	stalled := false
	var deliveries []syntheticPacerDelivery
	for {
		flow := ""
		target := end.Add(time.Nanosecond)
		for _, candidate := range flows {
			if next[candidate].Before(target) {
				flow, target = candidate, next[candidate]
			}
		}
		if target.After(end) {
			return deliveries
		}
		if !stalled && !target.Before(stallAt) {
			now = stallAt.Add(stall)
			stalled = true
		}
		if now.Before(target) {
			now = target
		}
		if now.After(end) {
			return deliveries
		}
		deliveries = append(deliveries, syntheticPacerDelivery{at: now, flow: flow, bytes: packetBytes})
		next[flow] = pacer.reserveAt(now, flow, packetBytes)
	}
}

func syntheticDeliveryBytes(deliveries []syntheticPacerDelivery) int64 {
	var total int64
	for _, delivery := range deliveries {
		total += delivery.bytes
	}
	return total
}

func assertSyntheticFairness(t *testing.T, deliveries []syntheticPacerDelivery, flows []string) {
	t.Helper()
	counts := make(map[string]int64, len(flows))
	for _, delivery := range deliveries {
		counts[delivery.flow] += delivery.bytes
	}
	total := int64(0)
	for _, flow := range flows {
		total += counts[flow]
	}
	target := total / int64(len(flows))
	for _, flow := range flows {
		delta := counts[flow] - target
		if delta < 0 {
			delta = -delta
		}
		if target == 0 || float64(delta)/float64(target) > 0.10 {
			t.Fatalf("flow %s share=%d target=%d", flow, counts[flow], target)
		}
	}
}

func assertSyntheticCap(t *testing.T, deliveries []syntheticPacerDelivery, bytesPerSecond int64) {
	t.Helper()
	sort.Slice(deliveries, func(first, second int) bool { return deliveries[first].at.Before(deliveries[second].at) })
	for first, total, last := 0, int64(0), 0; first < len(deliveries); first++ {
		for last < len(deliveries) && deliveries[last].at.Sub(deliveries[first].at) <= 5*time.Second {
			total += deliveries[last].bytes
			last++
		}
		if total > bytesPerSecond*5*105/100 {
			t.Fatalf("five-second egress=%d exceeds cap=%d", total, bytesPerSecond*5)
		}
		total -= deliveries[first].bytes
	}
}