// Licensed to Apache Software Foundation (ASF) under one or more contributor
// license agreements. See the NOTICE file distributed with
// this work for additional information regarding copyright
// ownership. Apache Software Foundation (ASF) licenses this file to you under
// the Apache License, Version 2.0 (the "License"); you may
// not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing,
// software distributed under the License is distributed on an
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
// KIND, either express or implied.  See the License for the
// specific language governing permissions and limitations
// under the License.

// Package metrics implements parsing of Prometheus text format metrics.
package metrics

import (
	"fmt"
	"regexp"
	"sort"
	"strconv"
	"strings"

	dto "github.com/prometheus/client_model/go"
	"github.com/prometheus/common/expfmt"
	"github.com/prometheus/common/model"
)

// Label represents a metric label as a key-value pair.
type Label struct {
	Name  string
	Value string
}

// RawMetric represents a parsed Prometheus metric.
type RawMetric struct {
	Name   string
	Desc   string
	Type   string // lowercase prometheus type: "counter","gauge","histogram","summary","untyped",""
	Labels []Label
	Value  float64
}

// MetricKey is used for uniquely identifying metrics.
type MetricKey struct {
	Name   string
	Labels []Label
}

// String generates a canonical representation of the MetricKey.
func (mk MetricKey) String() string {
	if len(mk.Labels) == 0 {
		return mk.Name
	}

	sortedLabels := make([]Label, len(mk.Labels))
	copy(sortedLabels, mk.Labels)
	sort.Slice(sortedLabels, func(i, j int) bool {
		if sortedLabels[i].Name != sortedLabels[j].Name {
			return sortedLabels[i].Name < sortedLabels[j].Name
		}
		return sortedLabels[i].Value < sortedLabels[j].Value
	})

	var labelParts []string
	for _, label := range sortedLabels {
		labelParts = append(labelParts, fmt.Sprintf(`%s="%s"`, label.Name, label.Value))
	}

	return fmt.Sprintf("%s{%s}", mk.Name, strings.Join(labelParts, ","))
}

const (
	labelNodeRole      = "node_role"
	labelPodName       = "pod_name"
	labelContainerName = "container_name"
)

var (
	helpLineRegex = regexp.MustCompile(`^#\s+HELP\s+(\S+)\s+(.+)$`)
	// metricLineRegex matches metric lines: metric_name{label1="value1",label2="value2"} value.
	metricLineRegex = regexp.MustCompile(`^(\w+)(?:\{([^}]+)\})?\s+(.+)$`)
	labelRegex      = regexp.MustCompile(`(\w+)="([^"]+)"`)
)

// metricTypeToString converts a dto.MetricType to its lowercase Prometheus text representation.
func metricTypeToString(mt dto.MetricType) string {
	switch mt {
	case dto.MetricType_COUNTER:
		return "counter"
	case dto.MetricType_GAUGE:
		return "gauge"
	case dto.MetricType_SUMMARY:
		return "summary"
	case dto.MetricType_HISTOGRAM:
		return "histogram"
	case dto.MetricType_UNTYPED:
		return "untyped"
	default:
		return ""
	}
}

// ResolveMetricType resolves the metric type for a given series name using the typeMap.
// For histogram/summary component series (ending in _bucket/_sum/_count), the base family
// name is used to look up the type. Returns "" if unknown.
func ResolveMetricType(typeMap map[string]string, name string) string {
	if t, ok := typeMap[name]; ok {
		return t
	}
	// Trim histogram/summary suffixes to find the base family name.
	for _, suffix := range []string{"_bucket", "_sum", "_count"} {
		if strings.HasSuffix(name, suffix) {
			base := strings.TrimSuffix(name, suffix)
			if base != "" {
				if t, ok := typeMap[base]; ok {
					return t
				}
			}
			break
		}
	}
	return ""
}

// ParseWithAgentLabels parses Prometheus text format metrics and returns structured RawMetric objects.
func ParseWithAgentLabels(text string, nodeRole, podName, containerName string) ([]RawMetric, error) {
	return ParseWithNodeLabels(text, nodeRole, podName, containerName, nil)
}

// ParseWithNodeLabels is ParseWithAgentLabels plus the node's own labels (e.g. the data-node
// tier "type"). Each node label is stamped under a "node_" prefix (so "type" becomes
// "node_type"), which keeps it from ever colliding with a metric-intrinsic label of the same
// name (e.g. the merge "type"). pod_name and container_name are skipped because they are
// already stamped as first-class labels.
func ParseWithNodeLabels(text string, nodeRole, podName, containerName string, nodeLabels map[string]string) ([]RawMetric, error) {
	// Build authoritative type map using expfmt.
	// TextToMetricFamilies may return a non-nil error AND a non-empty families map
	// simultaneously (e.g. for "unexpected end of input stream" on trailing newline).
	// We always use whatever families were successfully parsed, ignoring parse errors.
	typeMap := make(map[string]string)
	parser := expfmt.NewTextParser(model.LegacyValidation)
	families, _ := parser.TextToMetricFamilies(strings.NewReader(text))
	for familyName, mf := range families {
		typeMap[familyName] = metricTypeToString(mf.GetType())
	}

	lines := strings.Split(text, "\n")
	var metrics []RawMetric
	helpMap := make(map[string]string)

	for _, line := range lines {
		line = strings.TrimSpace(line)
		if line == "" || strings.HasPrefix(line, "#") {
			if matches := helpLineRegex.FindStringSubmatch(line); matches != nil {
				metricName := matches[1]
				helpText := matches[2]
				helpMap[metricName] = helpText
			}
			continue
		}

		matches := metricLineRegex.FindStringSubmatch(line)
		if matches == nil {
			continue
		}

		metricName := matches[1]
		labelStr := matches[2]
		valueStr := matches[3]

		var labels []Label
		if labelStr != "" {
			labelMatches := labelRegex.FindAllStringSubmatch(labelStr, -1)
			for _, labelMatch := range labelMatches {
				labels = append(labels, Label{
					Name:  labelMatch[1],
					Value: labelMatch[2],
				})
			}
		}

		// Add agent identity labels if provided
		if nodeRole != "" {
			labels = append(labels, Label{
				Name:  labelNodeRole,
				Value: nodeRole,
			})
		}
		if podName != "" {
			labels = append(labels, Label{
				Name:  labelPodName,
				Value: podName,
			})
		}
		if containerName != "" {
			labels = append(labels, Label{
				Name:  labelContainerName,
				Value: containerName,
			})
		}
		for nodeLabelKey, nodeLabelValue := range nodeLabels {
			if nodeLabelValue == "" || nodeLabelKey == labelPodName || nodeLabelKey == labelContainerName {
				continue
			}
			labels = append(labels, Label{
				Name:  "node_" + nodeLabelKey,
				Value: nodeLabelValue,
			})
		}

		value, parseErr := strconv.ParseFloat(valueStr, 64)
		if parseErr != nil {
			return nil, fmt.Errorf("failed to parse metric value for %s: %w", metricName, parseErr)
		}

		desc := helpMap[metricName]

		metrics = append(metrics, RawMetric{
			Name:   metricName,
			Labels: labels,
			Value:  value,
			Desc:   desc,
			Type:   ResolveMetricType(typeMap, metricName),
		})
	}

	return metrics, nil
}
