package service

import (
	"bufio"
	"context"
	"fmt"
	"io"
	"net/http"
	"strconv"
	"strings"

	opv2 "github.com/pingcap/tidb-operator/api/v2/core/v1alpha1"
	apiv1 "github.com/tidbcloud/cluster-service-ng/api/tidbcloud/cluster-agent/v1"
	"golang.org/x/sync/errgroup"
	corev1 "k8s.io/api/core/v1"
	"sigs.k8s.io/controller-runtime/pkg/client"
)

const tikvUsageFetchMaxConcurrency = 4

func (s *AgentService) GetClusterTopology(ctx context.Context, req *apiv1.GetClusterTopologyReq) (*apiv1.GetClusterTopologyResp, error) {
	return s.buildClusterTopologyV2(ctx, req)
}

func (s *AgentService) buildClusterTopologyV2(ctx context.Context, req *apiv1.GetClusterTopologyReq) (*apiv1.GetClusterTopologyResp, error) {
	resp := &apiv1.GetClusterTopologyResp{}
	ns := buildNamespace(req.ClusterId)

	var pdPodList corev1.PodList
	if err := s.dataplaneCli.List(ctx, &pdPodList, client.InNamespace(ns), getComponentPodLabels("pd", clusterName)); err != nil {
		return nil, fmt.Errorf("failed to list pd pods: %w", err)
	}
	for _, pd := range pdPodList.Items {
		resp.PdNodes = append(resp.PdNodes, &apiv1.GetClusterTopologyResp_PDNodeInfo{
			Name: pd.Name,
			Host: fmt.Sprintf("%s.%s.%s.svc", pd.Name, tikvHeadlessService, ns),
			Port: uint32(opv2.DefaultPDPortClient),
		})
	}

	var tikvPodList corev1.PodList
	if err := s.dataplaneCli.List(ctx, &tikvPodList, client.InNamespace(ns), getComponentPodLabels("tikv", clusterName)); err != nil {
		return nil, fmt.Errorf("failed to list tikv pods: %w", err)
	}
	for _, tikv := range tikvPodList.Items {
		resp.TikvNodes = append(resp.TikvNodes, &apiv1.GetClusterTopologyResp_TiKVNodeInfo{
			Name: tikv.Name,
			Host: fmt.Sprintf("%s.%s.%s.svc", tikv.Name, tikvHeadlessService, ns),
			Port: uint32(opv2.DefaultTiKVPortClient),
		})
	}
	if req.IncludeTikvUsage {
		if err := s.fetchTiKVUsage(ctx, resp); err != nil {
			return nil, fmt.Errorf("failed to fetch tikv usage: %w", err)
		}
	}
	return resp, nil
}

func getComponentPodLabels(compType, clusterName string) client.MatchingLabels {
	return map[string]string{
		opv2.LabelKeyCluster:   clusterName,
		opv2.LabelKeyComponent: compType,
	}
}

//nolint:unused // Reserved for future use
func getGroupName(pod corev1.Pod) string {
	return pod.GetLabels()[opv2.LabelKeyGroup]
}

func (s *AgentService) fetchTiKVUsage(ctx context.Context, resp *apiv1.GetClusterTopologyResp) error {
	if s.tikvHTTPClient == nil {
		return fmt.Errorf("tikv HTTP client unavailable")
	}
	if len(resp.TikvNodes) == 0 {
		return nil
	}

	concurrency := tikvUsageFetchMaxConcurrency
	if len(resp.TikvNodes) < concurrency {
		concurrency = len(resp.TikvNodes)
	}

	g, ctx := errgroup.WithContext(ctx)
	g.SetLimit(concurrency)

	for _, tikvNode := range resp.TikvNodes {
		node := tikvNode
		g.Go(func() error {
			capacity, available, err := queryTiKVStorageUsage(ctx, s.tikvHTTPClient, node.Host)
			if err != nil {
				return fmt.Errorf("failed to query metrics for TiKV node %s: %w", node.Name, err)
			}
			node.StorageCapacityBytes = capacity
			node.StorageAvailableBytes = available
			return nil
		})
	}
	if err := g.Wait(); err != nil {
		return err
	}

	return nil
}

func queryTiKVStorageUsage(ctx context.Context, httpClient *http.Client, host string) (capacity, available uint64, err error) {
	// Construct the Prometheus query URL
	queryURL, err := buildTiKVMetricsURL(host)
	if err != nil {
		return 0, 0, fmt.Errorf("failed to build metrics URL: %w", err)
	}

	// Create HTTP request
	req, err := http.NewRequestWithContext(ctx, http.MethodGet, queryURL, http.NoBody)
	if err != nil {
		return 0, 0, fmt.Errorf("failed to create request: %w", err)
	}

	// Execute request
	//nolint:gosec // G704: request URL is built by buildTiKVMetricsURL with internal host/path validation.
	resp, err := httpClient.Do(req)
	if err != nil {
		return 0, 0, fmt.Errorf("failed to execute request: %w", err)
	}
	defer func() {
		_ = resp.Body.Close()
	}()

	if resp.StatusCode != http.StatusOK {
		return 0, 0, fmt.Errorf("unexpected status code: %d", resp.StatusCode)
	}

	capacity, available, err = parseStorageUsage(resp.Body)
	if err != nil {
		return 0, 0, fmt.Errorf("failed to parse storage usage: %w", err)
	}
	return capacity, available, nil
}

func buildTiKVMetricsURL(host string) (string, error) {
	return buildHTTPSURL(host, int(opv2.DefaultTiKVPortStatus), "/metrics")
}

/*
example metrics:
# HELP tikv_store_size_bytes Size of storage.
# TYPE tikv_store_size_bytes gauge
tikv_store_size_bytes{keyspace_name="",region_id="",storage_class="",type="all_used"} 1166477691
tikv_store_size_bytes{keyspace_name="",region_id="",storage_class="",type="available"} 1583013987973
tikv_store_size_bytes{keyspace_name="",region_id="",storage_class="",type="capacity"} 1584180465664
tikv_store_size_bytes{keyspace_name="10430094504260193936",region_id="859",storage_class="standard",type="used"} 0
*/

func parseStorageUsage(respBody io.Reader) (capacity, available uint64, err error) {
	const maxMetricsLineSize = 1024 * 1024

	scanner := bufio.NewScanner(respBody)
	scanner.Buffer(make([]byte, 0, 64*1024), maxMetricsLineSize)

	var matches int
	var capacityFound bool
	var availableFound bool

	for scanner.Scan() {
		metricType, value, ok, parseErr := parseTiKVStoreSizeMetricLine(scanner.Text())
		if parseErr != nil {
			return 0, 0, parseErr
		}
		if !ok {
			continue
		}

		matches++
		switch metricType {
		case "capacity":
			if capacityFound {
				return 0, 0, fmt.Errorf("duplicate capacity metric")
			}
			capacityFound = true
			capacity = value
		case "available":
			if availableFound {
				return 0, 0, fmt.Errorf("duplicate available metric")
			}
			availableFound = true
			available = value
		}
	}
	if err := scanner.Err(); err != nil {
		return 0, 0, fmt.Errorf("failed to scan response body: %w", err)
	}
	if matches != 2 {
		return 0, 0, fmt.Errorf("should have 2 matches, got %d", matches)
	}
	if !capacityFound || !availableFound {
		return 0, 0, fmt.Errorf("not all required metrics found")
	}
	return capacity, available, nil
}

func parseTiKVStoreSizeMetricLine(line string) (metricType string, value uint64, ok bool, err error) {
	if !strings.HasPrefix(line, "tikv_store_size_bytes{") {
		return "", 0, false, nil
	}

	typeIdx := strings.Index(line, `type="`)
	if typeIdx == -1 {
		return "", 0, false, nil
	}
	typeStart := typeIdx + len(`type="`)
	typeEnd := strings.IndexByte(line[typeStart:], '"')
	if typeEnd == -1 {
		return "", 0, false, fmt.Errorf("invalid tikv_store_size_bytes metric line")
	}
	metricType = line[typeStart : typeStart+typeEnd]
	if metricType != "capacity" && metricType != "available" {
		return "", 0, false, nil
	}

	valueIdx := strings.LastIndexByte(line, ' ')
	if valueIdx == -1 || valueIdx == len(line)-1 {
		return "", 0, false, fmt.Errorf("invalid tikv_store_size_bytes metric value")
	}
	valueStr := strings.TrimSpace(line[valueIdx+1:])
	value, err = strconv.ParseUint(strings.TrimSpace(valueStr), 10, 64)
	if err != nil {
		return "", 0, false, fmt.Errorf("failed to parse %s: %w", metricType, err)
	}
	return metricType, value, true, nil
}