package metricsprovider

import (
	"context"
	"fmt"
	"io"
	"log/slog"
	"net/http"
	"strings"
	"sync"
	"time"

	"github.com/pingcap/tidb-operator/api/v2/core/v1alpha1"
	dto "github.com/prometheus/client_model/go"
	"github.com/prometheus/common/expfmt"
	autoscalingv2 "k8s.io/api/autoscaling/v2"
	corev1 "k8s.io/api/core/v1"
	"k8s.io/apimachinery/pkg/labels"
	"k8s.io/apimachinery/pkg/types"
	"k8s.io/metrics/pkg/apis/custom_metrics"
	"sigs.k8s.io/controller-runtime/pkg/client"
	"sigs.k8s.io/custom-metrics-apiserver/pkg/provider"
)

const (
	nodeExporterNamespace     = "monitoring"
	nodeExporterInstanceKey   = "app.kubernetes.io/instance"
	nodeExporterInstanceValue = "node-exporter"
	nodeExporterPortName      = "metrics"

	nodeExporterScrapeTimeout = 5 * time.Second

	// counterStaleDuration: 2× the HPA default sync period (15 s) so a missed
	// scrape does not cause the next valid scrape to compute a wrong rate.
	counterStaleDuration = 60 * time.Second

	ifacePrimaryAWSETH0    = "eth0"
	ifacePrefixAlicloudETH = "eth"
)

var ifaceDenylistPrefixes = []string{
	"lo", "veth", "docker", "cni", "cali", "flannel", "br-", "virbr", "kube-ipvs",
}

// awsInstanceBandwidthBytes maps AWS EC2 instance type to its baseline network
// bandwidth in bytes/sec.
// Source: EC2 Instance Types Guide, BaselineBandwidthInGbps field, converted: Gbps * 1e9 / 8.
// node_network_speed_bytes on EKS reports the burst (or link) speed; for burstable
// instance sizes this inflates the denominator and yields falsely low utilisation.
// Use this map as a fallback (or primary) to always measure against the sustained baseline.
// Covers all instance types in dev/staging/prod AWS node pools.
var awsInstanceBandwidthBytes = map[string]float64{
	// m8g (general-purpose, Graviton4)
	"m8g.large":    117_125_000,   // 0.937 Gbit/s baseline
	"m8g.xlarge":   234_375_000,   // 1.875 Gbit/s baseline
	"m8g.2xlarge":  468_750_000,   // 3.75 Gbit/s baseline
	"m8g.4xlarge":  937_500_000,   // 7.5 Gbit/s baseline
	"m8g.8xlarge":  1_875_000_000, // 15 Gbit/s
	"m8g.16xlarge": 3_750_000_000, // 30 Gbit/s

	// m8gd (general-purpose with local NVMe, Graviton4)
	"m8gd.4xlarge": 937_500_000, // 7.5 Gbit/s baseline

	// c8g (compute-optimized, Graviton4)
	"c8g.xlarge":   234_375_000,   // 1.875 Gbit/s baseline
	"c8g.2xlarge":  468_750_000,   // 3.75 Gbit/s baseline
	"c8g.4xlarge":  937_500_000,   // 7.5 Gbit/s baseline
	"c8g.8xlarge":  1_875_000_000, // 15 Gbit/s
	"c8g.16xlarge": 3_750_000_000, // 30 Gbit/s

	// r8g (memory-optimized, Graviton4)
	"r8g.4xlarge": 937_500_000,   // 7.5 Gbit/s baseline
	"r8g.8xlarge": 1_875_000_000, // 15 Gbit/s

	// r6i (memory-optimized, Intel Ice Lake)
	"r6i.large": 97_625_000, // 0.781 Gbit/s baseline

	// i8g (storage-optimized, Graviton4)
	"i8g.large":    146_500_000,   // 1.172 Gbit/s baseline
	"i8g.xlarge":   293_000_000,   // 2.344 Gbit/s baseline
	"i8g.2xlarge":  586_000_000,   // 4.688 Gbit/s baseline
	"i8g.4xlarge":  1_171_875_000, // 9.375 Gbit/s baseline
	"i8g.8xlarge":  3_125_000_000, // 25 Gbit/s
	"i8g.16xlarge": 4_687_500_000, // 37.5 Gbit/s

	// i8ge (storage-optimized with EFA, Graviton4)
	"i8ge.3xlarge": 1_562_500_000, // 12.5 Gbit/s baseline
	"i8ge.6xlarge": 4_687_500_000, // 37.5 Gbit/s

	// i7i (storage-optimized, Intel Emerald Rapids)
	"i7i.4xlarge": 1_171_875_000, // 9.375 Gbit/s baseline
}

// alicloudInstanceBandwidthBytes maps Alibaba Cloud ECS instance type to its
// maximum network bandwidth in bytes/sec.
// Source: DescribeInstanceTypes API, InstanceBandwidthRx field (Kbit/s), converted: Kbps * 1000 / 8.
// Covers all instance types in dev/staging/prod alicloud-ap-southeast-1 node pools.
var alicloudInstanceBandwidthBytes = map[string]float64{
	// g7 (admin nodes)
	"ecs.g7.large": 256_000_000, // 2.048 Gbit/s

	// g9i (general-purpose)
	"ecs.g9i.large":    320_000_000,   // 2.560 Gbit/s
	"ecs.g9i.xlarge":   512_000_000,   // 4.096 Gbit/s
	"ecs.g9i.2xlarge":  768_000_000,   // 6.144 Gbit/s
	"ecs.g9i.4xlarge":  1_536_000_000, // 12.288 Gbit/s
	"ecs.g9i.8xlarge":  2_560_000_000, // 20.480 Gbit/s
	"ecs.g9i.16xlarge": 3_584_000_000, // 28.672 Gbit/s

	// r9i (memory-optimized)
	"ecs.r9i.2xlarge": 768_000_000,   // 6.144 Gbit/s
	"ecs.r9i.4xlarge": 1_536_000_000, // 12.288 Gbit/s
	"ecs.r9i.8xlarge": 2_560_000_000, // 20.480 Gbit/s

	// c9i (compute-optimized)
	"ecs.c9i.large":    320_000_000,   // 2.560 Gbit/s
	"ecs.c9i.xlarge":   512_000_000,   // 4.096 Gbit/s
	"ecs.c9i.2xlarge":  768_000_000,   // 6.144 Gbit/s
	"ecs.c9i.4xlarge":  1_536_000_000, // 12.288 Gbit/s
	"ecs.c9i.8xlarge":  2_560_000_000, // 20.480 Gbit/s
	"ecs.c9i.16xlarge": 3_584_000_000, // 28.672 Gbit/s

	// i3g / i4 (local-disk)
	"ecs.i3g.2xlarge": 384_000_000,   // 3.072 Gbit/s
	"ecs.i4.xlarge":   512_000_000,   // 4.096 Gbit/s
	"ecs.i4.2xlarge":  768_000_000,   // 6.144 Gbit/s
	"ecs.i4.4xlarge":  1_280_000_000, // 10.240 Gbit/s
	"ecs.i4.8xlarge":  3_200_000_000, // 25.600 Gbit/s
	"ecs.i4.16xlarge": 6_400_000_000, // 51.200 Gbit/s
}

// cloudConfig bundles the per-provider interface selection and speed resolution
// logic derived for a specific node before scraping it.
type cloudConfig struct {
	// deviceFilter returns true if the named interface should be measured.
	deviceFilter func(string) bool
	// resolveSpeed returns the link speed in bytes/sec for the given device.
	// Both AWS and Alicloud resolve speed from a hardcoded instance-type map so
	// that utilisation is always measured against the sustained baseline bandwidth
	// rather than the burst value reported by node_network_speed_bytes.
	resolveSpeed func(device string, speedByDevice map[string]float64) (float64, bool)
}

type networkCounterSample struct {
	timestamp time.Time
	rxBytes   float64
	txBytes   float64
}

type bandwidthCache struct {
	mu    sync.Mutex
	items map[string]networkCounterSample
}

func newBandwidthCache() *bandwidthCache {
	return &bandwidthCache{items: make(map[string]networkCounterSample)}
}

func (c *bandwidthCache) cacheKey(targetKey, nodeName, device string) string {
	return targetKey + "/" + nodeName + "/" + device
}

func (c *bandwidthCache) load(targetKey, nodeName, device string, now time.Time) (networkCounterSample, bool) {
	c.mu.Lock()
	defer c.mu.Unlock()
	s, ok := c.items[c.cacheKey(targetKey, nodeName, device)]
	if !ok {
		return networkCounterSample{}, false
	}
	if now.Sub(s.timestamp) > counterStaleDuration {
		delete(c.items, c.cacheKey(targetKey, nodeName, device))
		return networkCounterSample{}, false
	}
	return s, true
}

func (c *bandwidthCache) store(targetKey, nodeName, device string, s networkCounterSample) {
	c.mu.Lock()
	defer c.mu.Unlock()
	c.items[c.cacheKey(targetKey, nodeName, device)] = s
}

func (p *cmProvider) GetMetricExtAvgBandwidthUtilization(
	ctx context.Context,
	name types.NamespacedName,
	info provider.CustomMetricInfo,
	_ labels.Selector,
) (*custom_metrics.MetricValue, error) {
	_, res, err := info.Normalized(p.mapper)
	if err != nil {
		return nil, fmt.Errorf("normalize metric info: %w", err)
	}
	if res != "tiproxygroup" {
		return nil, fmt.Errorf("ext_avg_bandwidth_utilization: resource %q not supported (want tiproxygroup)", res)
	}

	scaleTargetRef := autoscalingv2.CrossVersionObjectReference{
		APIVersion: v1alpha1.SchemeGroupVersion.String(),
		Kind:       "TiProxyGroup",
		Name:       name.Name,
	}
	scaleObj, err := getScale(ctx, p.scaleClient, p.mapper, name.Namespace, scaleTargetRef)
	if err != nil {
		return nil, fmt.Errorf("get scale for tiproxygroup %s: %w", name, err)
	}
	podSelector, err := labels.Parse(scaleObj.Status.Selector)
	if err != nil {
		return nil, fmt.Errorf("parse pod selector %q: %w", scaleObj.Status.Selector, err)
	}

	podList := new(corev1.PodList)
	if err = p.kubeClient.List(ctx, podList,
		&client.ListOptions{LabelSelector: podSelector, Namespace: name.Namespace},
	); err != nil {
		return nil, fmt.Errorf("list tiproxy pods: %w", err)
	}

	drainingPods := drainingPodNames(podList.Items)
	var runningPods []corev1.Pod
	for i := range podList.Items {
		pod := &podList.Items[i]
		if pod.Status.Phase != corev1.PodRunning || pod.Spec.NodeName == "" {
			continue
		}
		if drainingPods.Has(pod.Name) {
			continue
		}
		runningPods = append(runningPods, *pod)
	}
	if len(runningPods) == 0 {
		return nil, fmt.Errorf("no running tiproxy pods for tiproxygroup %s", name)
	}

	nodeExporterByNode, err := p.buildNodeExporterMap(ctx)
	if err != nil {
		return nil, err
	}

	targetKey := name.String()
	now := time.Now()
	var rxUtilSum, txUtilSum float64
	usablePods := 0

	nodePodCounts := make(map[string]int, len(runningPods))
	nodeOrder := make([]string, 0, len(runningPods))
	for i := range runningPods {
		nodeName := runningPods[i].Spec.NodeName
		if nodePodCounts[nodeName] == 0 {
			nodeOrder = append(nodeOrder, nodeName)
		}
		nodePodCounts[nodeName]++
	}

	for _, nodeName := range nodeOrder {
		podCount := nodePodCounts[nodeName]
		nePod, ok := nodeExporterByNode[nodeName]
		if !ok {
			slog.Warn("no node-exporter pod on node, skipping",
				"node", nodeName, "affected_pods", podCount)
			continue
		}
		cfg, err := p.detectCloudConfig(ctx, nodeName)
		if err != nil {
			return nil, err
		}
		rxUtil, txUtil, scrapeErr := p.scrapeNodeUtilization(ctx, targetKey, nodeName, nePod, cfg, now)
		if scrapeErr != nil {
			slog.Warn("bandwidth scrape failed, skipping node",
				"node", nodeName, "affected_pods", podCount, "err", scrapeErr)
			continue
		}
		rxUtilSum += rxUtil * float64(podCount)
		txUtilSum += txUtil * float64(podCount)
		usablePods += podCount
	}

	if usablePods == 0 {
		return nil, fmt.Errorf("no usable bandwidth samples for tiproxygroup %s (warm-up or all nodes failed)", name)
	}
	// Quorum guard: if fewer than half the running pods produced a usable sample,
	// the average would be unrepresentative and could trigger false scale-out.
	// Return an error so HPA sees a stale metric and holds position.
	if usablePods < (len(runningPods)+1)/2 {
		return nil, fmt.Errorf("too few usable bandwidth samples for tiproxygroup %s: %d/%d pods (quorum requires >= 50%%)",
			name, usablePods, len(runningPods))
	}

	avgRx := rxUtilSum / float64(usablePods)
	avgTx := txUtilSum / float64(usablePods)
	avgUtil := avgRx
	if avgTx > avgRx {
		avgUtil = avgTx
	}
	if avgUtil < 0 {
		avgUtil = 0
	}
	if avgUtil > 1 {
		avgUtil = 1
	}
	metricValue := int64(avgUtil * 100)

	slog.Info("ext_avg_bandwidth_utilization",
		"tiproxygroup", name, "usable_pods", usablePods,
		"total_pods", len(runningPods), "avg_rx_pct", int64(avgRx*100), "avg_tx_pct", int64(avgTx*100), "pct", metricValue)

	return p.metricFor(metricValue, name, info)
}

func (p *cmProvider) detectCloudConfig(ctx context.Context, nodeName string) (*cloudConfig, error) {
	node := new(corev1.Node)
	if err := p.kubeClient.Get(ctx, types.NamespacedName{Name: nodeName}, node); err != nil {
		return nil, fmt.Errorf("get node %s: %w", nodeName, err)
	}
	switch {
	case node.Labels["eks.amazonaws.com/nodegroup"] != "":
		instanceType := node.Labels["node.kubernetes.io/instance-type"]
		bw, ok := awsInstanceBandwidthBytes[instanceType]
		if !ok {
			return nil, fmt.Errorf("ext_avg_bandwidth_utilization: unknown aws instance type %q for node %s", instanceType, nodeName)
		}
		return &cloudConfig{
			deviceFilter: func(device string) bool {
				return device == ifacePrimaryAWSETH0
			},
			resolveSpeed: func(_ string, _ map[string]float64) (float64, bool) {
				return bw, true
			},
		}, nil
	case node.Labels["ack.aliyun.com"] != "":
		instanceType := node.Labels["node.kubernetes.io/instance-type"]
		bw, ok := alicloudInstanceBandwidthBytes[instanceType]
		if !ok {
			return nil, fmt.Errorf("ext_avg_bandwidth_utilization: unknown alicloud instance type %q for node %s", instanceType, nodeName)
		}
		return &cloudConfig{
			deviceFilter: func(device string) bool {
				return !isIfaceDenylisted(device) && strings.HasPrefix(device, ifacePrefixAlicloudETH)
			},
			resolveSpeed: func(_ string, _ map[string]float64) (float64, bool) {
				return bw, true
			},
		}, nil
	default:
		return nil, fmt.Errorf("ext_avg_bandwidth_utilization: unsupported cloud provider for node %s", nodeName)
	}
}

func isIfaceDenylisted(device string) bool {
	for _, prefix := range ifaceDenylistPrefixes {
		if strings.HasPrefix(device, prefix) {
			return true
		}
	}
	return false
}

func (p *cmProvider) buildNodeExporterMap(ctx context.Context) (map[string]*corev1.Pod, error) {
	podList := new(corev1.PodList)
	if err := p.kubeClient.List(ctx, podList,
		client.InNamespace(nodeExporterNamespace),
		client.MatchingLabels{nodeExporterInstanceKey: nodeExporterInstanceValue},
	); err != nil {
		return nil, fmt.Errorf("list node-exporter pods: %w", err)
	}
	m := make(map[string]*corev1.Pod, len(podList.Items))
	for i := range podList.Items {
		pod := &podList.Items[i]
		if pod.Status.Phase == corev1.PodRunning && pod.Spec.NodeName != "" {
			m[pod.Spec.NodeName] = pod
		}
	}
	return m, nil
}

func resolveNodeExporterPort(pod *corev1.Pod) (int32, error) {
	for i := range pod.Spec.Containers {
		for j := range pod.Spec.Containers[i].Ports {
			if pod.Spec.Containers[i].Ports[j].Name == nodeExporterPortName {
				return pod.Spec.Containers[i].Ports[j].ContainerPort, nil
			}
		}
	}
	return 0, fmt.Errorf("node-exporter pod %s has no port %q", pod.Name, nodeExporterPortName)
}

func (p *cmProvider) scrapeNodeUtilization(
	ctx context.Context,
	hpaKey, nodeName string,
	nePod *corev1.Pod,
	cfg *cloudConfig,
	now time.Time,
) (rxUtil, txUtil float64, err error) {
	port, err := resolveNodeExporterPort(nePod)
	if err != nil {
		return 0, 0, err
	}

	// TiProxy pods run with hostNetwork:true, so pod.Status.HostIP is the node IP.
	// node-exporter also binds on the host network at the same address.
	metricsURL := fmt.Sprintf("http://%s:%d/metrics?collect[]=netdev&collect[]=netclass", nePod.Status.HostIP, port)

	body, err := p.fetchNodeExporterMetrics(ctx, metricsURL)
	if err != nil {
		return 0, 0, err
	}
	defer body.Close() //nolint:errcheck // response body Close error is not actionable here

	families, parseErr := (&expfmt.TextParser{}).TextToMetricFamilies(body)
	if parseErr != nil {
		if len(families) == 0 {
			return 0, 0, fmt.Errorf("parse node-exporter for node %s: %w", nodeName, parseErr)
		}
		slog.Warn("partial parse error from node-exporter, continuing with available families",
			"node", nodeName, "err", parseErr)
	}

	rxByDevice := dtoFamilyToDeviceValues(families["node_network_receive_bytes_total"])
	txByDevice := dtoFamilyToDeviceValues(families["node_network_transmit_bytes_total"])
	speedByDevice := dtoFamilyToDeviceValues(families["node_network_speed_bytes"])

	maxRxUtil := -1.0
	maxTxUtil := -1.0
	warmUp := false

	for device, rxBytes := range rxByDevice {
		if !cfg.deviceFilter(device) {
			continue
		}
		txBytes, ok := txByDevice[device]
		if !ok {
			continue
		}
		speed, ok := cfg.resolveSpeed(device, speedByDevice)
		if !ok {
			continue
		}

		current := networkCounterSample{timestamp: now, rxBytes: rxBytes, txBytes: txBytes}
		prev, hasPrev := p.bwCache.load(hpaKey, nodeName, device, now)
		p.bwCache.store(hpaKey, nodeName, device, current)

		if !hasPrev {
			warmUp = true
			continue
		}

		deltaSeconds := now.Sub(prev.timestamp).Seconds()
		if deltaSeconds <= 0 {
			continue
		}
		deltaRx := rxBytes - prev.rxBytes
		deltaTx := txBytes - prev.txBytes
		// Counter reset guard: negative delta means the counter wrapped or restarted.
		if deltaRx < 0 {
			deltaRx = 0
		}
		if deltaTx < 0 {
			deltaTx = 0
		}

		if u := (deltaRx / deltaSeconds) / speed; u > maxRxUtil {
			maxRxUtil = u
		}
		if u := (deltaTx / deltaSeconds) / speed; u > maxTxUtil {
			maxTxUtil = u
		}
	}

	if maxRxUtil < 0 {
		if warmUp {
			return 0, 0, fmt.Errorf("node %s: warm-up scrape stored, rate available next cycle", nodeName)
		}
		return 0, 0, fmt.Errorf("no eligible network interface on node %s", nodeName)
	}
	return maxRxUtil, maxTxUtil, nil
}

func (p *cmProvider) fetchNodeExporterMetrics(ctx context.Context, url string) (io.ReadCloser, error) {
	reqCtx, cancel := context.WithTimeout(ctx, nodeExporterScrapeTimeout)

	req, err := http.NewRequestWithContext(reqCtx, http.MethodGet, url, http.NoBody)
	if err != nil {
		cancel()
		return nil, fmt.Errorf("build request for %s: %w", url, err)
	}

	resp, err := p.nodeExporterHTTPClient.Do(req)
	if err != nil {
		cancel()
		return nil, fmt.Errorf("GET %s: %w", url, err)
	}
	if resp.StatusCode != http.StatusOK {
		_ = resp.Body.Close()
		cancel()
		return nil, fmt.Errorf("node-exporter status %d for %s", resp.StatusCode, url)
	}
	return &cancelOnClose{ReadCloser: resp.Body, cancel: cancel}, nil
}

type cancelOnClose struct {
	io.ReadCloser
	cancel context.CancelFunc
}

func (c *cancelOnClose) Close() error {
	c.cancel()
	return c.ReadCloser.Close()
}

// dtoFamilyToDeviceValues extracts a map of device-label → gauge/counter value
// from a Prometheus MetricFamily. Returns an empty map if mf is nil.
func dtoFamilyToDeviceValues(mf *dto.MetricFamily) map[string]float64 {
	out := make(map[string]float64)
	if mf == nil {
		return out
	}
	for _, m := range mf.GetMetric() {
		device := ""
		for _, lp := range m.GetLabel() {
			if lp.GetName() == "device" {
				device = lp.GetValue()
				break
			}
		}
		if device == "" {
			continue
		}
		switch {
		case m.GetGauge() != nil:
			out[device] = m.GetGauge().GetValue()
		case m.GetCounter() != nil:
			out[device] = m.GetCounter().GetValue()
		}
	}
	return out
}