package export

import (
	"context"
	"errors"
	"fmt"
	"net/netip"
	"os"
	"strings"
	"testing"
	"time"

	"github.com/prometheus/client_golang/prometheus"
	dto "github.com/prometheus/client_model/go"
	"ysun.co/rfm/collector"
	"ysun.co/rfm/config"
)

// --- mock types ---

type mockIfaceStats struct {
	entries []IfaceStatsEntry
	err     error
	rate    uint32
	rateErr error
}

func (m *mockIfaceStats) IfaceStats() ([]IfaceStatsEntry, error) {
	return m.entries, m.err
}

func (m *mockIfaceStats) SampleRate() (uint32, error) {
	if m.rateErr != nil {
		return 0, m.rateErr
	}
	if m.rate == 0 {
		return 1, nil
	}
	return m.rate, nil
}

type staticEnricher struct {
	srcASN  uint32
	srcCity string
	dstASN  uint32
	dstCity string
}

func (e *staticEnricher) Enrich(src, dst netip.Addr) (collector.Labels, collector.Labels) {
	return collector.Labels{ASN: e.srcASN, City: e.srcCity},
		collector.Labels{ASN: e.dstASN, City: e.dstCity}
}

type exportMockReader struct {
	events  [][]byte
	idx     int
	dropErr error
}

func (m *exportMockReader) ReadRawEvent() ([]byte, error) {
	if m.idx >= len(m.events) {
		return nil, os.ErrDeadlineExceeded
	}
	raw := m.events[m.idx]
	m.idx++
	return raw, nil
}

func (m *exportMockReader) SetDeadline(t time.Time) {}

func (m *exportMockReader) DroppedEvents() (uint64, error) {
	return 0, m.dropErr
}

func (m *exportMockReader) Close() error { return nil }

// --- test helpers ---

// collectAll calls Collect and returns a map of metric name -> summed value
func collectAll(t *testing.T, mc *MetricsCollector) map[string]float64 {
	t.Helper()
	metrics := collectMetrics(t, mc)
	result := make(map[string]float64)
	for _, m := range metrics {
		d := new(dto.Metric)
		if err := m.Write(d); err != nil {
			t.Fatalf("write metric: %v", err)
		}
		name := extractName(m.Desc())
		var val float64
		switch {
		case d.Counter != nil:
			val = d.Counter.GetValue()
		case d.Gauge != nil:
			val = d.Gauge.GetValue()
		}
		result[name] += val
	}
	return result
}

// collectMetrics calls Collect and returns all emitted metrics
func collectMetrics(t *testing.T, mc *MetricsCollector) []prometheus.Metric {
	t.Helper()
	ch := make(chan prometheus.Metric, 100)
	mc.Collect(ch)
	close(ch)
	var out []prometheus.Metric
	for m := range ch {
		out = append(out, m)
	}
	return out
}

// extractName parses fqName from Desc.String() format: Desc{fqName: "rfm_...", ...}
func extractName(desc *prometheus.Desc) string {
	s := desc.String()
	const prefix = "fqName: \""
	i := strings.Index(s, prefix)
	if i < 0 {
		return ""
	}
	s = s[i+len(prefix):]
	before, _, ok := strings.Cut(s, "\"")
	if !ok {
		return ""
	}
	return before
}

// assertCounter checks that the named metric has the expected value
func assertCounter(t *testing.T, vals map[string]float64, name string, want float64) {
	t.Helper()
	got, ok := vals[name]
	if !ok {
		t.Errorf("metric %q not found", name)
		return
	}
	if got != want {
		t.Errorf("%s = %g, want %g", name, got, want)
	}
}

// assertGauge checks that the named metric has the expected value
func assertGauge(t *testing.T, vals map[string]float64, name string, want float64) {
	t.Helper()
	got, ok := vals[name]
	if !ok {
		t.Errorf("metric %q not found", name)
		return
	}
	if got != want {
		t.Errorf("%s = %g, want %g", name, got, want)
	}
}

// metricLabels returns the label name->value map for a metric
func metricLabels(t *testing.T, m prometheus.Metric) map[string]string {
	t.Helper()
	d := new(dto.Metric)
	if err := m.Write(d); err != nil {
		t.Fatalf("write metric: %v", err)
	}
	labels := make(map[string]string)
	for _, lp := range d.Label {
		labels[lp.GetName()] = lp.GetValue()
	}
	return labels
}

// metricValue returns the numeric value of a metric (counter or gauge)
func metricValue(t *testing.T, m prometheus.Metric) float64 {
	t.Helper()
	d := new(dto.Metric)
	if err := m.Write(d); err != nil {
		t.Fatalf("write metric: %v", err)
	}
	switch {
	case d.Counter != nil:
		return d.Counter.GetValue()
	case d.Gauge != nil:
		return d.Gauge.GetValue()
	}
	return 0
}

// --- tests ---

func TestMetricsCollectorImplementsInterface(t *testing.T) {
	var _ prometheus.Collector = (*MetricsCollector)(nil)
}

func TestDescribe(t *testing.T) {
	mc := New(nil, nil)
	ch := make(chan *prometheus.Desc, 32)
	mc.Describe(ch)
	close(ch)

	var descs []*prometheus.Desc
	for d := range ch {
		descs = append(descs, d)
	}

	if got, want := len(descs), len(allDescs); got != want {
		t.Fatalf("got %d descriptors, want %d", got, want)
	}

	names := make(map[string]bool)
	for _, d := range descs {
		names[extractName(d)] = true
	}

	want := []string{
		"rfm_interface_rx_bytes_total",
		"rfm_interface_tx_bytes_total",
		"rfm_interface_rx_packets_total",
		"rfm_interface_tx_packets_total",
		"rfm_flow_bytes",
		"rfm_flow_packets",
		"rfm_flow_sampled_bytes",
		"rfm_flow_sampled_packets",
		"rfm_flow_bytes_total",
		"rfm_flow_packets_total",
		"rfm_flow_sampled_bytes_total",
		"rfm_flow_sampled_packets_total",
		"rfm_collector_active_flows",
		"rfm_collector_dropped_events_total",
		"rfm_collector_forced_evictions_total",
		"rfm_collector_folded_flows_total",
		"rfm_errors_total",
		"rfm_bpf_sample_rate",
		"rfm_ipfix_connected",
		"rfm_ipfix_dials_total",
		"rfm_ipfix_dial_errors_total",
		"rfm_ipfix_messages_total",
		"rfm_ipfix_records_total",
		"rfm_ipfix_dropped_records_total",
		"rfm_ipfix_send_errors_total",
	}
	for _, w := range want {
		if !names[w] {
			t.Errorf("missing descriptor %q", w)
		}
	}
}

func TestCollectHealth(t *testing.T) {
	c := collector.New(time.Minute, nil, config.DefaultMaxFlows)
	c.Record(collector.FlowEvent{
		Ifindex: 1,
		Dir:     0,
		Proto:   6,
		SrcAddr: netip.MustParseAddr("10.0.0.1"),
		DstAddr: netip.MustParseAddr("10.0.0.2"),
		SrcPort: 12345,
		DstPort: 80,
		Segs:    1,
		Len:     100,
	}, time.Now())

	mc := New(nil, c)
	vals := collectAll(t, mc)

	assertGauge(t, vals, "rfm_collector_active_flows", 1)
	assertCounter(t, vals, "rfm_collector_dropped_events_total", 0)
	assertCounter(t, vals, "rfm_collector_forced_evictions_total", 0)
}

func TestCollectIfaceStats(t *testing.T) {
	src := &mockIfaceStats{
		entries: []IfaceStatsEntry{
			{Ifindex: 1, Dir: 0, Proto: 4, Packets: 10, Bytes: 1000}, // rx ipv4
			{Ifindex: 1, Dir: 1, Proto: 4, Packets: 5, Bytes: 500},   // tx ipv4
			{Ifindex: 1, Dir: 0, Proto: 6, Packets: 3, Bytes: 300},   // rx ipv6
		},
	}

	mc := New(src, nil)
	vals := collectAll(t, mc)

	assertCounter(t, vals, "rfm_interface_rx_bytes_total", 1300) // 1000 + 300
	assertCounter(t, vals, "rfm_interface_tx_bytes_total", 500)
	assertCounter(t, vals, "rfm_interface_rx_packets_total", 13) // 10 + 3
	assertCounter(t, vals, "rfm_interface_tx_packets_total", 5)
}

func TestCollectIfaceStatsSkipsKeysSeenTwice(t *testing.T) {
	// deleting the key under the cursor restarts a hash map walk, so a scrape
	// that overlaps a detach can read a key twice
	src := &mockIfaceStats{
		entries: []IfaceStatsEntry{
			{Ifindex: 1, Dir: 0, Proto: 4, Packets: 10, Bytes: 1000},
			{Ifindex: 1, Dir: 1, Proto: 4, Packets: 5, Bytes: 500},
			{Ifindex: 1, Dir: 0, Proto: 4, Packets: 11, Bytes: 1100},
		},
	}
	mc := New(src, nil)
	mc.resolveIfname = func(uint32) string { return "eth0" }
	reg := prometheus.NewRegistry()
	reg.MustRegister(mc)

	mfs, err := reg.Gather()
	if err != nil {
		t.Fatalf("gather: %v", err)
	}
	for _, mf := range mfs {
		if mf.GetName() != "rfm_interface_rx_bytes_total" {
			continue
		}
		if got := len(mf.GetMetric()); got != 1 {
			t.Fatalf("rx series = %d, want 1", got)
		}
		if got := mf.GetMetric()[0].GetCounter().GetValue(); got != 1100 {
			t.Fatalf("rx bytes = %v, want the later read 1100", got)
		}
		return
	}
	t.Fatal("rfm_interface_rx_bytes_total missing")
}

func TestScrapeSurvivesAnInterfaceNameThatIsNotUTF8(t *testing.T) {
	// a link name is any bytes to the kernel, the registry refuses the whole
	// scrape over one label that is not utf-8
	src := &mockIfaceStats{
		entries: []IfaceStatsEntry{{Ifindex: 7, Dir: 0, Proto: 4, Packets: 1, Bytes: 60}},
	}
	c := collector.New(time.Minute, nil, 16)
	c.Record(collector.FlowEvent{
		Ifindex: 7, Proto: 17, SrcPort: 5000, DstPort: 53,
		SrcAddr: netip.MustParseAddr("10.0.0.1"),
		DstAddr: netip.MustParseAddr("10.0.0.2"),
		Segs:    1,
		Len:     60,
	}, time.Now())
	mc := New(src, c)
	mc.resolveIfname = func(uint32) string { return "d\xff" }
	reg := prometheus.NewRegistry()
	reg.MustRegister(mc)

	mfs, err := reg.Gather()
	if err != nil {
		t.Fatalf("gather: %v", err)
	}
	// the interface counters and the flow series name the link alike
	seen := map[string]bool{}
	for _, mf := range mfs {
		for _, m := range mf.GetMetric() {
			for _, l := range m.GetLabel() {
				if l.GetName() != "ifname" {
					continue
				}
				if l.GetValue() != "d\uFFFD" {
					t.Fatalf("%s ifname = %q, want the invalid byte replaced", mf.GetName(), l.GetValue())
				}
				seen[mf.GetName()] = true
			}
		}
	}
	for _, name := range []string{"rfm_interface_rx_packets_total", "rfm_flow_packets_total"} {
		if !seen[name] {
			t.Fatalf("%s missing", name)
		}
	}
}

func TestCollectCachesIfnames(t *testing.T) {
	src := &mockIfaceStats{
		entries: []IfaceStatsEntry{
			{Ifindex: 7, Dir: 0, Proto: 4, Packets: 10, Bytes: 1000},
			{Ifindex: 7, Dir: 1, Proto: 6, Packets: 5, Bytes: 500},
		},
	}

	c := collector.New(time.Minute, nil, config.DefaultMaxFlows)
	now := time.Now()
	c.Record(collector.FlowEvent{
		Ifindex: 7,
		Dir:     0,
		Proto:   6,
		SrcAddr: netip.MustParseAddr("10.0.0.1"),
		DstAddr: netip.MustParseAddr("10.0.0.2"),
		SrcPort: 1000,
		DstPort: 80,
		Segs:    1,
		Len:     100,
	}, now)
	c.Record(collector.FlowEvent{
		Ifindex: 7,
		Dir:     1,
		Proto:   17,
		SrcAddr: netip.MustParseAddr("10.0.0.3"),
		DstAddr: netip.MustParseAddr("10.0.0.4"),
		SrcPort: 2000,
		DstPort: 53,
		Segs:    1,
		Len:     200,
	}, now)

	mc := New(src, c)
	calls := make(map[uint32]int)
	mc.resolveIfname = func(ifindex uint32) string {
		calls[ifindex]++
		return fmt.Sprintf("if-%d", ifindex)
	}

	// one resolve per scrape however many series mention the interface,
	// a second scrape resolves again so a renamed interface shows up
	collectMetrics(t, mc)
	if got := calls[7]; got != 1 {
		t.Fatalf("ifname resolver calls for ifindex 7 = %d, want 1", got)
	}
	collectMetrics(t, mc)
	if got := calls[7]; got != 2 {
		t.Fatalf("ifname resolver calls after two scrapes = %d, want 2", got)
	}
}

func TestCollectFlowsNoEnricher(t *testing.T) {
	c := collector.New(time.Minute, nil, config.DefaultMaxFlows)
	c.Record(collector.FlowEvent{
		Ifindex: 2,
		Dir:     0,
		Proto:   6,
		SrcAddr: netip.MustParseAddr("192.168.1.1"),
		DstAddr: netip.MustParseAddr("192.168.1.2"),
		SrcPort: 1234,
		DstPort: 443,
		Segs:    1,
		Len:     200,
	}, time.Now())

	mc := New(nil, c)
	vals := collectAll(t, mc)

	assertGauge(t, vals, "rfm_flow_bytes", 200)
	assertGauge(t, vals, "rfm_flow_packets", 1)
	assertGauge(t, vals, "rfm_flow_sampled_bytes", 200)
	assertGauge(t, vals, "rfm_flow_sampled_packets", 1)

	// check labels, enrichment should be empty strings
	metrics := collectMetrics(t, mc)
	for _, m := range metrics {
		name := extractName(m.Desc())
		if name != "rfm_flow_bytes" {
			continue
		}
		labels := metricLabels(t, m)
		if labels["src_asn"] != "" {
			t.Errorf("src_asn = %q, want empty", labels["src_asn"])
		}
		if labels["dst_asn"] != "" {
			t.Errorf("dst_asn = %q, want empty", labels["dst_asn"])
		}
		if labels["direction"] != "ingress" {
			t.Errorf("direction = %q, want ingress", labels["direction"])
		}
	}
}

func TestCollectFlowsWithEnricher(t *testing.T) {
	e := &staticEnricher{
		srcASN: 64512, srcCity: "Berlin",
		dstASN: 13335, dstCity: "London",
	}
	c := collector.New(time.Minute, e, config.DefaultMaxFlows)
	c.Record(collector.FlowEvent{
		Ifindex: 3,
		Dir:     1,
		Proto:   17,
		SrcAddr: netip.MustParseAddr("10.1.0.1"),
		DstAddr: netip.MustParseAddr("10.2.0.1"),
		SrcPort: 5000,
		DstPort: 53,
		Segs:    1,
		Len:     64,
	}, time.Now())

	mc := New(nil, c)
	metrics := collectMetrics(t, mc)

	var found bool
	for _, m := range metrics {
		name := extractName(m.Desc())
		if name != "rfm_flow_bytes" {
			continue
		}
		found = true
		labels := metricLabels(t, m)

		if labels["src_asn"] != "64512" {
			t.Errorf("src_asn = %q, want 64512", labels["src_asn"])
		}
		if labels["dst_asn"] != "13335" {
			t.Errorf("dst_asn = %q, want 13335", labels["dst_asn"])
		}
		if labels["src_city"] != "Berlin" {
			t.Errorf("src_city = %q, want Berlin", labels["src_city"])
		}
		if labels["dst_city"] != "London" {
			t.Errorf("dst_city = %q, want London", labels["dst_city"])
		}
		if labels["direction"] != "egress" {
			t.Errorf("direction = %q, want egress", labels["direction"])
		}

		val := metricValue(t, m)
		if val != 64 {
			t.Errorf("flow_bytes = %g, want 64", val)
		}
	}
	if !found {
		t.Error("rfm_flow_bytes metric not found")
	}
}

func TestCollectFlowsScalesBySampleRate(t *testing.T) {
	src := &mockIfaceStats{rate: 10}

	c := collector.New(time.Minute, nil, config.DefaultMaxFlows)
	c.SetSampleRate(10, 0)
	c.Record(collector.FlowEvent{
		Ifindex: 2,
		Dir:     0,
		Proto:   6,
		SrcAddr: netip.MustParseAddr("192.168.1.1"),
		DstAddr: netip.MustParseAddr("192.168.1.2"),
		SrcPort: 1234,
		DstPort: 443,
		Segs:    1,
		Len:     200,
	}, time.Now())

	mc := New(src, c)
	vals := collectAll(t, mc)

	assertGauge(t, vals, "rfm_flow_sampled_bytes", 200)
	assertGauge(t, vals, "rfm_flow_sampled_packets", 1)
	assertGauge(t, vals, "rfm_flow_bytes", 2000)
	assertGauge(t, vals, "rfm_flow_packets", 10)
	assertGauge(t, vals, "rfm_bpf_sample_rate", 10)
}

func TestCollectNilSources(t *testing.T) {
	mc := New(nil, nil)
	// must not panic
	vals := collectAll(t, mc)
	// should have no iface or flow metrics, but also no panic
	_ = vals
}

func TestCollectFlowsAggregatesDuplicateLabels(t *testing.T) {
	// two flows with different ports but same exported labels
	// must produce one aggregated metric, not duplicate series
	c := collector.New(time.Minute, nil, config.DefaultMaxFlows)
	now := time.Now()
	c.Record(collector.FlowEvent{
		Ifindex: 1, Dir: 0, Proto: 6,
		SrcAddr: netip.MustParseAddr("10.0.0.1"),
		DstAddr: netip.MustParseAddr("10.0.0.2"),
		SrcPort: 1000, DstPort: 80, Segs: 1, Len: 100,
	}, now)
	c.Record(collector.FlowEvent{
		Ifindex: 1, Dir: 0, Proto: 6,
		SrcAddr: netip.MustParseAddr("10.0.0.1"),
		DstAddr: netip.MustParseAddr("10.0.0.2"),
		SrcPort: 2000, DstPort: 80, Segs: 1, Len: 200,
	}, now)

	mc := New(nil, c)

	// use a real registry to verify no duplicate series
	reg := prometheus.NewRegistry()
	reg.MustRegister(mc)
	mfs, err := reg.Gather()
	if err != nil {
		t.Fatalf("gather: %v", err)
	}

	// find rfm_flow_bytes and check it has exactly one series
	for _, mf := range mfs {
		if mf.GetName() != "rfm_flow_bytes" &&
			mf.GetName() != "rfm_flow_sampled_bytes" {
			continue
		}
		if got := len(mf.GetMetric()); got != 1 {
			t.Fatalf("%s series count = %d, want 1 (aggregated)", mf.GetName(), got)
		}
		val := mf.GetMetric()[0].GetGauge().GetValue()
		if val != 300 {
			t.Fatalf("%s = %g, want 300 (100+200)", mf.GetName(), val)
		}
	}
}

func TestCollectIfaceStatsError(t *testing.T) {
	src := &mockIfaceStats{
		entries: []IfaceStatsEntry{
			{Ifindex: 1, Dir: 0, Proto: 4, Packets: 10, Bytes: 1000},
		},
		err: fmt.Errorf("map iteration failed"),
	}

	mc := New(src, nil)
	metrics := collectMetrics(t, mc)

	// on error, no iface metrics should be emitted
	for _, m := range metrics {
		name := extractName(m.Desc())
		if strings.HasPrefix(name, "rfm_interface_") {
			t.Errorf("unexpected iface metric %q on error", name)
		}
	}
}

func TestCollectIfaceStatsFamily(t *testing.T) {
	// iface stats proto=4 is IPv4, proto=6 is IPv6
	// the label should be "family" with values "ipv4" / "ipv6"
	src := &mockIfaceStats{
		entries: []IfaceStatsEntry{
			{Ifindex: 1, Dir: 0, Proto: 4, Packets: 10, Bytes: 1000},
			{Ifindex: 1, Dir: 0, Proto: 6, Packets: 5, Bytes: 500},
		},
	}

	mc := New(src, nil)
	metrics := collectMetrics(t, mc)

	families := make(map[string]bool)
	for _, m := range metrics {
		name := extractName(m.Desc())
		if name != "rfm_interface_rx_bytes_total" {
			continue
		}
		labels := metricLabels(t, m)
		families[labels["family"]] = true
	}

	if !families["ipv4"] {
		t.Error("expected family=ipv4 label")
	}
	if !families["ipv6"] {
		t.Error("expected family=ipv6 label")
	}
}

func TestCollectErrorsTotal(t *testing.T) {
	src := &mockIfaceStats{
		entries: nil,
		err:     fmt.Errorf("map broken"),
	}

	mc := New(src, nil)
	metrics := collectMetrics(t, mc)

	var found bool
	for _, m := range metrics {
		name := extractName(m.Desc())
		if name != "rfm_errors_total" {
			continue
		}
		found = true
		labels := metricLabels(t, m)
		if labels["subsystem"] != "bpf_map" {
			t.Errorf("subsystem = %q, want bpf_map", labels["subsystem"])
		}
		val := metricValue(t, m)
		if val != 1 {
			t.Errorf("errors_total = %g, want 1", val)
		}
	}
	if !found {
		t.Error("rfm_errors_total not found")
	}
}

func TestCollectErrorsSubsystemLabels(t *testing.T) {
	// exercise both subsystem labels via a collector that has
	// accumulated ring_buffer and bpf_map errors through Run
	// the drop counter is polled by the eviction ticker, so keep it short
	c := collector.New(200*time.Millisecond, nil, config.DefaultMaxFlows)

	mr := &exportMockReader{
		events:  [][]byte{{0xde, 0xad}}, // garbage -> decode error -> ring_buffer
		dropErr: fmt.Errorf("map read failed"),
	}

	ctx, cancel := context.WithCancel(context.Background())
	errCh := make(chan error, 1)
	go func() { errCh <- c.Run(ctx, mr) }()

	// wait for errors to be counted
	deadline := time.Now().Add(time.Second)
	for {
		s := c.Stats()
		if s.RingBufErrors > 0 && s.BPFMapErrors > 0 {
			break
		}
		if time.Now().After(deadline) {
			t.Fatal("timed out waiting for error counters")
		}
		time.Sleep(5 * time.Millisecond)
	}
	cancel()
	<-errCh

	mc := New(nil, c)
	metrics := collectMetrics(t, mc)

	subsystems := make(map[string]float64)
	for _, m := range metrics {
		if extractName(m.Desc()) != "rfm_errors_total" {
			continue
		}
		labels := metricLabels(t, m)
		subsystems[labels["subsystem"]] = metricValue(t, m)
	}

	if _, ok := subsystems["ring_buffer"]; !ok {
		t.Error("missing rfm_errors_total{subsystem=\"ring_buffer\"}")
	} else if subsystems["ring_buffer"] < 1 {
		t.Errorf("ring_buffer errors = %g, want >= 1", subsystems["ring_buffer"])
	}

	if _, ok := subsystems["bpf_map"]; !ok {
		t.Error("missing rfm_errors_total{subsystem=\"bpf_map\"}")
	} else if subsystems["bpf_map"] < 1 {
		t.Errorf("bpf_map errors = %g, want >= 1", subsystems["bpf_map"])
	}
}

func TestCollectErrorsTotalZeroOnSuccess(t *testing.T) {
	src := &mockIfaceStats{
		entries: []IfaceStatsEntry{
			{Ifindex: 1, Dir: 0, Proto: 4, Packets: 1, Bytes: 100},
		},
	}

	mc := New(src, nil)
	vals := collectAll(t, mc)
	assertCounter(t, vals, "rfm_errors_total", 0)
}

func TestCollectErrorsTotalOnSampleRateLookupError(t *testing.T) {
	src := &mockIfaceStats{
		rateErr: fmt.Errorf("config map broken"),
	}

	c := collector.New(time.Minute, nil, config.DefaultMaxFlows)
	c.Record(collector.FlowEvent{
		Ifindex: 2,
		Dir:     0,
		Proto:   6,
		SrcAddr: netip.MustParseAddr("192.168.1.1"),
		DstAddr: netip.MustParseAddr("192.168.1.2"),
		SrcPort: 1234,
		DstPort: 443,
		Segs:    1,
		Len:     200,
	}, time.Now())

	mc := New(src, c)
	metrics := collectMetrics(t, mc)

	var found bool
	for _, m := range metrics {
		// a rate the scrape could not read is left out, a stand in of 1
		// would claim the programs sample every packet
		if extractName(m.Desc()) == "rfm_bpf_sample_rate" {
			t.Errorf("rfm_bpf_sample_rate = %g after a failed read, want it left out", metricValue(t, m))
		}
		if extractName(m.Desc()) != "rfm_errors_total" {
			continue
		}
		labels := metricLabels(t, m)
		if labels["subsystem"] != "bpf_map" {
			continue
		}
		found = true
		if val := metricValue(t, m); val != 1 {
			t.Errorf("bpf_map errors = %g, want 1", val)
		}
	}
	if !found {
		t.Error("missing rfm_errors_total{subsystem=\"bpf_map\"}")
	}
}

// refusingIfaceStats is a probe whose interface counter map refused updates
type refusingIfaceStats struct {
	mockIfaceStats
	refused uint64
	err     error
}

func (m *refusingIfaceStats) IfaceStatsErrors() (uint64, error) {
	return m.refused, m.err
}

// errorsBySubsystem returns the rfm_errors_total values of one scrape
func errorsBySubsystem(t *testing.T, mc *MetricsCollector) map[string]float64 {
	t.Helper()
	got := map[string]float64{}
	for _, m := range collectMetrics(t, mc) {
		if extractName(m.Desc()) == "rfm_errors_total" {
			got[metricLabels(t, m)["subsystem"]] += metricValue(t, m)
		}
	}
	return got
}

func TestCollectErrorsCountRefusedIfaceStatsUpdates(t *testing.T) {
	src := &refusingIfaceStats{refused: 7}
	mc := New(src, nil)
	if got := errorsBySubsystem(t, mc)["bpf_map"]; got != 7 {
		t.Fatalf("bpf_map errors = %v, want the 7 refused counter updates", got)
	}

	// a failed read counts as one more error and the refused updates keep
	// their last value, so the counter never goes back
	src.err = errors.New("lookup failed")
	if got := errorsBySubsystem(t, mc)["bpf_map"]; got != 8 {
		t.Fatalf("bpf_map errors = %v after a failed read, want 8", got)
	}
}

// the probe source reports both counts of errors the programs keep
var (
	_ IfaceStatsErrorSource = (*ProbeSource)(nil)
	_ GSOHeaderErrorSource  = (*ProbeSource)(nil)
)

// unparsedGSO is a probe whose programs counted gso skbs without parsing
// their headers
type unparsedGSO struct {
	mockIfaceStats
	unparsed uint64
	err      error
}

func (m *unparsedGSO) GSOHeaderErrors() (uint64, error) {
	return m.unparsed, m.err
}

func TestCollectErrorsCountUnparsedGSOHeaders(t *testing.T) {
	src := &unparsedGSO{unparsed: 4}
	mc := New(src, nil)
	if got := errorsBySubsystem(t, mc); got["gso_header"] != 4 || got["bpf_map"] != 0 {
		t.Fatalf("errors = %v, want the 4 unparsed gso skbs as gso_header errors", got)
	}

	// a failed read is a bpf_map error and the gso_header errors keep their
	// last value, so neither goes back
	src.err = errors.New("lookup failed")
	if got := errorsBySubsystem(t, mc); got["gso_header"] != 4 || got["bpf_map"] != 1 {
		t.Fatalf("errors = %v after a failed read, want 4 gso_header and 1 bpf_map", got)
	}

	// a source that keeps no such count has no series for it
	if got := errorsBySubsystem(t, New(&mockIfaceStats{}, nil)); len(got) != 1 {
		t.Fatalf("errors = %v from a source without gso header errors, want bpf_map alone", got)
	}
}

func TestCollectErrorsOfOtherSubsystems(t *testing.T) {
	mc := New(nil, collector.New(time.Minute, nil, config.DefaultMaxFlows))
	var parse, dropped uint64 = 3, 2
	mc.AddErrors("bmp", func() uint64 { return parse })
	mc.AddErrors("netlink", func() uint64 { return dropped })
	mc.AddErrors("bpf_map", func() uint64 { return 5 })

	got := errorsBySubsystem(t, mc)
	want := map[string]float64{"bmp": 3, "netlink": 2, "bpf_map": 5, "ring_buffer": 0, "ipfix": 0}
	if len(got) != len(want) {
		t.Fatalf("subsystems = %v, want %v", got, want)
	}
	for s, v := range want {
		if got[s] != v {
			t.Fatalf("%s errors = %v, want %v", s, got[s], v)
		}
	}

	// every subsystem is one series, a registry refuses a repeated one
	reg := prometheus.NewRegistry()
	reg.MustRegister(mc)
	if _, err := reg.Gather(); err != nil {
		t.Fatalf("gather: %v", err)
	}
}

func TestCollectIPFIXStats(t *testing.T) {
	c := collector.New(30*time.Second, nil, config.DefaultMaxFlows)
	mc := New(nil, c)
	mc.SetIPFIX(func() IPFIXStats {
		return IPFIXStats{
			Connected:    true,
			Dials:        2,
			DialErrors:   1,
			Messages:     10,
			Records:      200,
			QueueDropped: 3,
			Unsent:       4,
			EncodeErrors: 0,
			SendFailed:   40,
			SendErrors:   map[string]uint64{"EPERM": 5, "ECONNREFUSED": 1},
		}
	})

	metrics := collectMetrics(t, mc)
	got := map[string]float64{}
	for _, m := range metrics {
		d := new(dto.Metric)
		if err := m.Write(d); err != nil {
			t.Fatal(err)
		}
		var name strings.Builder
		name.WriteString(extractName(m.Desc()))
		for _, l := range d.Label {
			name.WriteString("{" + l.GetName() + "=" + l.GetValue() + "}")
		}
		switch {
		case d.Counter != nil:
			got[name.String()] = d.Counter.GetValue()
		case d.Gauge != nil:
			got[name.String()] = d.Gauge.GetValue()
		}
	}

	want := map[string]float64{
		"rfm_ipfix_connected":                                 1,
		"rfm_ipfix_dials_total":                               2,
		"rfm_ipfix_dial_errors_total":                         1,
		"rfm_ipfix_messages_total":                            10,
		"rfm_ipfix_records_total":                             200,
		"rfm_ipfix_dropped_records_total{reason=queue_full}":  3,
		"rfm_ipfix_dropped_records_total{reason=unconnected}": 4,
		"rfm_ipfix_dropped_records_total{reason=encode}":      0,
		"rfm_ipfix_dropped_records_total{reason=send}":        40,
		"rfm_ipfix_send_errors_total{errno=EPERM}":            5,
		"rfm_ipfix_send_errors_total{errno=ECONNREFUSED}":     1,
		// every lost record counts once by its reason, the failed messages
		// and dials lose nothing beyond the records counted
		"rfm_errors_total{subsystem=ipfix}": 47,
	}
	for name, val := range want {
		if got[name] != val {
			t.Errorf("%s = %v, want %v", name, got[name], val)
		}
	}
}

func TestCollectRollupCountersSurviveEviction(t *testing.T) {
	e := &staticEnricher{srcASN: 64500, dstASN: 64501, srcCity: "Oslo", dstCity: "Lima"}
	c := collector.New(time.Second, e, config.DefaultMaxFlows)
	c.SetSampleRate(10, 0)

	t0 := time.Now()
	ev := collector.FlowEvent{
		Ifindex: 2, Dir: 1, Proto: 17,
		SrcAddr: netip.MustParseAddr("192.168.1.1"),
		DstAddr: netip.MustParseAddr("192.168.1.2"),
		SrcPort: 1234, DstPort: 53, Segs: 1, Len: 100,
	}
	c.Record(ev, t0)
	other := ev
	other.SrcPort = 4321
	c.Record(other, t0)

	// the first scrape shows the new tuple at zero
	mc := New(nil, c)
	if got := collectAll(t, mc)["rfm_flow_bytes_total"]; got != 0 {
		t.Fatalf("rfm_flow_bytes_total on the first scrape = %v, want 0", got)
	}

	// evicting the flows resets the gauges but not the counters
	c.Evict(t0.Add(5 * time.Second))

	vals := collectAll(t, mc)
	if got := vals["rfm_flow_bytes"]; got != 0 {
		t.Fatalf("rfm_flow_bytes = %v after eviction, want no live flows", got)
	}
	if got := vals["rfm_flow_bytes_total"]; got != 2000 {
		t.Fatalf("rfm_flow_bytes_total = %v, want 2000", got)
	}
	if got := vals["rfm_flow_packets_total"]; got != 20 {
		t.Fatalf("rfm_flow_packets_total = %v, want 20", got)
	}
	if got := vals["rfm_flow_sampled_bytes_total"]; got != 200 {
		t.Fatalf("rfm_flow_sampled_bytes_total = %v, want 200", got)
	}
	if got := vals["rfm_flow_sampled_packets_total"]; got != 2 {
		t.Fatalf("rfm_flow_sampled_packets_total = %v, want 2", got)
	}

	for _, m := range collectMetrics(t, mc) {
		if extractName(m.Desc()) != "rfm_flow_bytes_total" {
			continue
		}
		labels := metricLabels(t, m)
		if labels["src_asn"] != "64500" || labels["dst_city"] != "Lima" || labels["direction"] != "egress" {
			t.Fatalf("unexpected labels %v", labels)
		}
	}

	// the counters outlive their flows for the retention, then leave the scrape
	c.Evict(t0.Add(collector.RollupRetention*time.Second - time.Second))
	if got := collectAll(t, mc)["rfm_flow_bytes_total"]; got != 2000 {
		t.Fatalf("rfm_flow_bytes_total inside the retention = %v, want 2000", got)
	}
	c.Evict(t0.Add(collector.RollupRetention*time.Second + time.Second))
	vals = collectAll(t, mc)
	if _, ok := vals["rfm_flow_bytes_total"]; ok {
		t.Fatal("idle rollup still exported after the retention window")
	}
}

func TestCollectRollupSeriesStartAtZero(t *testing.T) {
	c := collector.New(time.Minute, nil, config.DefaultMaxFlows)
	c.SetSampleRate(10, 0)
	ev := collector.FlowEvent{
		Ifindex: 2, Dir: 1, Proto: 6,
		SrcAddr: netip.MustParseAddr("192.168.1.1"),
		DstAddr: netip.MustParseAddr("192.168.1.2"),
		SrcPort: 1234, DstPort: 443, Segs: 1, Len: 100,
	}
	c.Record(ev, time.Now())

	// a series whose first sample already holds its first packets loses
	// them to increase(), which takes the first sample for history
	mc := New(nil, c)
	want := []map[string]float64{
		{"rfm_flow_packets_total": 0, "rfm_flow_bytes_total": 0, "rfm_flow_sampled_packets_total": 0, "rfm_flow_sampled_bytes_total": 0},
		{"rfm_flow_packets_total": 10, "rfm_flow_bytes_total": 1000, "rfm_flow_sampled_packets_total": 1, "rfm_flow_sampled_bytes_total": 100},
	}
	for i, w := range want {
		vals := collectAll(t, mc)
		for name, v := range w {
			got, ok := vals[name]
			if !ok || got != v {
				t.Fatalf("scrape %d: %s = %v (present %v), want %v", i+1, name, got, ok, v)
			}
		}
	}
}

// asnPerSource gives every source address an asn of its own, so every flow
// from a new source opens a new label tuple
type asnPerSource struct{}

func (asnPerSource) Enrich(src, dst netip.Addr) (collector.Labels, collector.Labels) {
	b := src.As16()
	return collector.Labels{ASN: 64512 + uint32(b[15])}, collector.Labels{}
}

// seriesID names a series by its labels
func seriesID(m *dto.Metric) string {
	var b strings.Builder
	for _, l := range m.GetLabel() {
		b.WriteString(l.GetName() + "=" + l.GetValue() + ",")
	}
	return b.String()
}

func TestRollupSumsStayExactAcrossTheCap(t *testing.T) {
	// flows and label tuples are both capped at four
	c := collector.New(time.Minute, asnPerSource{}, 4)
	c.SetSampleRate(10, 0)
	mc := New(nil, c)
	reg := prometheus.NewRegistry()
	reg.MustRegister(mc)

	// what a tsdb keeps, the last value of every series it scraped, a series
	// that leaves the scrape keeps its last value, so the sum is what
	// sum(increase()) adds up over all series that start at zero
	last := map[string]float64{}
	var folded float64
	scrape := func() int {
		t.Helper()
		mfs, err := reg.Gather()
		if err != nil {
			t.Fatalf("gather: %v", err)
		}
		var series int
		for _, mf := range mfs {
			switch mf.GetName() {
			case "rfm_flow_packets_total":
				for _, m := range mf.GetMetric() {
					last[seriesID(m)] = m.GetCounter().GetValue()
					series++
				}
			case "rfm_collector_folded_flows_total":
				folded = mf.GetMetric()[0].GetCounter().GetValue()
			}
		}
		return series
	}
	sum := func() float64 {
		var s float64
		for _, v := range last {
			s += v
		}
		return s
	}

	t0 := time.Now()
	var total float64
	send := func(src byte, at time.Time) {
		c.Record(collector.FlowEvent{
			Ifindex: 2, Dir: 0, Proto: 17, SrcPort: 5000, DstPort: 53,
			SrcAddr: netip.AddrFrom4([4]byte{10, 1, 0, src}),
			DstAddr: netip.MustParseAddr("10.0.0.1"),
			Segs:    1,
			Len:     100,
		}, at)
		total += 10
	}

	// four tuples fill the cap, the fifth finds no tuple a scrape has fully
	// shown to give up and its flow counts under empty labels
	for src := byte(1); src <= 5; src++ {
		send(src, t0)
	}
	scrape()
	scrape()
	if got := sum(); got != total {
		t.Fatalf("sum over all series = %v, want %v", got, total)
	}
	if folded != 1 {
		t.Fatalf("folded flows = %v, want 1", folded)
	}

	// the first tuple lost its flow to the fifth and two scrapes showed all
	// of it, a new tuple takes its place, its series keeps what it counted
	// and none of it moves to the overflow series
	send(6, t0.Add(time.Second))
	send(3, t0.Add(time.Second))
	send(5, t0.Add(time.Second))
	for range 2 {
		if n := scrape(); n > 5 {
			t.Fatalf("series = %d, want at most four tuples and the overflow", n)
		}
	}
	if got := sum(); got != total {
		t.Fatalf("sum over all series = %v, want %v", got, total)
	}
	if folded != 1 {
		t.Fatalf("folded flows = %v, want 1", folded)
	}
}

func TestCollectIPFIXErrorsComeFromTheExporter(t *testing.T) {
	// the exporter refuses a record when its queue is full and counts the
	// loss itself, the subsystem counter shows its count and nothing else
	exp := &recordingExporter{err: errors.New("ipfix queue full")}
	c := collector.New(30*time.Second, nil, config.DefaultMaxFlows)
	c.SetFlowExporter(exp)
	t0 := time.Now()
	c.Record(collector.FlowEvent{
		Proto: 6, SrcPort: 1, DstPort: 80,
		SrcAddr: netip.MustParseAddr("10.0.0.1"),
		DstAddr: netip.MustParseAddr("10.0.0.2"),
		Segs:    1,
		Len:     100,
	}, t0)
	c.Evict(t0.Add(time.Minute))

	mc := New(nil, c)
	if got := errorsBySubsystem(t, mc)["ipfix"]; got != 0 {
		t.Fatalf("ipfix errors = %v without exporter stats, want 0", got)
	}
	mc.SetIPFIX(func() IPFIXStats { return IPFIXStats{QueueDropped: 1, SendErrors: map[string]uint64{}} })
	vals := collectAll(t, mc)
	if got := vals["rfm_errors_total"]; got != 1 {
		t.Fatalf("rfm_errors_total summed over subsystems = %v, want 1 (one lost record)", got)
	}
}

type recordingExporter struct {
	err error
}

func (r *recordingExporter) ExportFlow(collector.ExportedFlow) error { return r.err }