package input_o11y_metrics_v1

import (
	"bytes"
	"context"
	"encoding/json"
	"fmt"
	"io"
	"net/http"
	"net/url"
	"strings"
	"sync"
	"text/template"
	"time"

	"github.com/tidbcloud/cluster-service-ng/pkg/autoscale/framework"
	"gopkg.in/yaml.v2"
)

const InputO11yMetricsV1IdStr framework.InputProviderType = "InputO11yMetricsV1"

var InputO11yMetricsV1Meta = framework.InputProviderMeta{
	Type:             InputO11yMetricsV1IdStr,
	InterfaceVersion: framework.InputInterfaceV1IdStr,
}

type InputO11yMetricsV1Config struct {
	O11yConfig                *O11yConfig `yaml:"o11yConfig"`
	ComponentType             string      `yaml:"componentType"`
	DownsamplingIntervalInMin int         `yaml:"downsamplingIntervalInMin"`
}

func (c *InputO11yMetricsV1Config) FillBlankValuesFrom(another framework.ConfigData) framework.ConfigData {
	if another == nil {
		return c
	}
	other, ok := another.(*InputO11yMetricsV1Config)
	if !ok {
		return c
	}
	merged := *c
	if merged.O11yConfig == nil && other.O11yConfig != nil {
		merged.O11yConfig = other.O11yConfig
	}
	if merged.ComponentType == "" && other.ComponentType != "" {
		merged.ComponentType = other.ComponentType
	}
	if merged.DownsamplingIntervalInMin == 0 && other.DownsamplingIntervalInMin != 0 {
		merged.DownsamplingIntervalInMin = other.DownsamplingIntervalInMin
	}
	return &merged
}

func (c *InputO11yMetricsV1Config) UniqStr() string {
	return c.ComponentType
}

func ParseConfig(cfg map[string]any) (framework.ConfigData, error) {
	if cfg == nil {
		return &InputO11yMetricsV1Config{}, nil
	}
	content, err := yaml.Marshal(cfg)
	if err != nil {
		return nil, err
	}
	var c InputO11yMetricsV1Config
	if err := yaml.UnmarshalStrict(content, &c); err != nil {
		return nil, err
	}
	return &c, nil
}

type InputO11yMetricsV1 struct {
	eventReceiver             framework.EventReceiver
	o11yConfig                *O11yConfig
	componentType             string
	downsamplingIntervalInMin int
	httpClient                *http.Client
	tokenMu                   sync.RWMutex
	cachedToken               string
	tokenExpiry               time.Time
	clusterInfoMu             sync.RWMutex
	cachedClusterInfoMap      map[string]*cachedClusterInfo
}

type cachedClusterInfo struct {
	clusterUUID   string
	projectId     string
	tenantId      string
	region        string
	cloudProvider string
}

func NewInputO11yMetricsV1(eventReceiver framework.EventReceiver, configData framework.ConfigData) (any, error) {
	cfg, ok := configData.(*InputO11yMetricsV1Config)
	if !ok || cfg == nil {
		return nil, fmt.Errorf("[NewInputO11yMetricsV1] config type mismatch or nil")
	}
	if cfg.O11yConfig == nil {
		return nil, fmt.Errorf("[NewInputO11yMetricsV1] o11yConfig is required")
	}

	downsamplingIntervalInMin := cfg.DownsamplingIntervalInMin
	if downsamplingIntervalInMin == 0 {
		downsamplingIntervalInMin = 2
	}

	return &InputO11yMetricsV1{
		eventReceiver:             eventReceiver,
		o11yConfig:                cfg.O11yConfig,
		componentType:             cfg.ComponentType,
		downsamplingIntervalInMin: downsamplingIntervalInMin,
		httpClient: &http.Client{
			Timeout: 30 * time.Second,
		},
		cachedClusterInfoMap: make(map[string]*cachedClusterInfo),
	}, nil
}

func (i *InputO11yMetricsV1) GetRangeMetrics(
	clusterId framework.ClusterId,
	name framework.MetricName,
	fromTo framework.TimeRange,
) (framework.MetricTimeValueInt64s, error) {
	ctx := context.Background()

	i.eventReceiver.OnEvent(
		EventInputO11yMetricsV1GetRangeMetrics,
		framework.EventDataWithName("ClusterId", framework.CloneableString(clusterId)),
		framework.EventDataWithName("MetricName", framework.CloneableString(string(name))),
	)

	if clusterId == "" {
		return nil, fmt.Errorf("[InputO11yMetricsV1.GetRangeMetrics] clusterId parameter is required")
	}

	_, err := i.ensureClusterInfo(ctx, string(clusterId))
	if err != nil {
		return nil, fmt.Errorf("[InputO11yMetricsV1.GetRangeMetrics] failed to ensure cluster info: %w", err)
	}

	req := &ComponentMetricReq{
		ClusterId:     string(clusterId),
		MetricName:    string(name),
		ComponentType: i.componentType,
		StartTime:     fromTo.From.UTC().Format(time.RFC3339),
		EndTime:       fromTo.To.UTC().Format(time.RFC3339),
		Step:          i.downsamplingIntervalInMin,
	}

	series, err := i.queryComponentMetrics(ctx, req)
	if err != nil {
		i.eventReceiver.OnEvent(
			EventInputO11yMetricsV1GetMetricsError,
			framework.EventDataWithName("Error", framework.CloneableString(err.Error())),
		)
		return nil, fmt.Errorf("[InputO11yMetricsV1.GetRangeMetrics] failed to query metrics: %w", err)
	}

	result := make(framework.MetricTimeValueInt64s, len(series.Points))
	for idx, point := range series.Points {
		result[idx] = framework.MetricTimeValueInt64{
			Time:  time.Unix(point.Timestamp, 0),
			Value: framework.MetricValueInt64(int64(point.Value)),
		}
	}

	i.eventReceiver.OnEvent(
		EventInputO11yMetricsV1GetMetricsSuccess,
		framework.EventDataWithName("PointsCount", framework.CloneableInt(len(result))),
	)

	return result, nil
}

func (i *InputO11yMetricsV1) GetClusterTopology(clusterId framework.ClusterId) (framework.ClusterTopologyV1, error) {
	i.eventReceiver.OnEvent(
		EventInputO11yMetricsV1GetClusterTopology,
		framework.EventDataWithName("ClusterId", framework.CloneableString(clusterId)),
	)

	if clusterId == "" {
		return nil, fmt.Errorf("[InputO11yMetricsV1.GetClusterTopology] clusterId parameter is required")
	}

	replicas, err := i.getComponentReplicas(clusterId)
	if err != nil {
		i.eventReceiver.OnEvent(
			EventInputO11yMetricsV1GetClusterTopologyError,
			framework.EventDataWithName("Error", framework.CloneableString(err.Error())),
		)
		return nil, fmt.Errorf("[InputO11yMetricsV1.GetClusterTopology] failed to get replicas: %w", err)
	}

	topology := framework.ClusterTopologyV1{
		framework.InstanceName(i.componentType): {
			Name:     framework.InstanceName(i.componentType),
			Type:     framework.InstanceType(i.componentType),
			Replicas: uint(replicas),
		},
	}

	i.eventReceiver.OnEvent(
		EventInputO11yMetricsV1GetClusterTopologySuccess,
		framework.EventDataWithName("Replicas", framework.CloneableInt(int(replicas))),
	)

	return topology, nil
}

func (i *InputO11yMetricsV1) getComponentReplicas(clusterId framework.ClusterId) (int64, error) {
	var metricName string
	switch i.componentType {
	case TIDB:
		metricName = TIDB_NUMBER
	case TIKV:
		metricName = TIKV_NUMBER
	default:
		return 0, fmt.Errorf("unsupported component type: %s", i.componentType)
	}

	timeRange := framework.NewTimeRangeLatest(5 * time.Minute)
	metrics, err := i.GetRangeMetrics(clusterId, framework.MetricName(metricName), timeRange)
	if err != nil {
		return 0, fmt.Errorf("failed to get replica metrics: %w", err)
	}

	if len(metrics) == 0 {
		return 0, fmt.Errorf("no replica metrics found")
	}

	return int64(metrics[len(metrics)-1].Value), nil
}

// GetQPS returns the latest QPS value for the cluster.
// Note: This method is reserved for future use and is not currently used.
func (i *InputO11yMetricsV1) GetQPS(clusterId framework.ClusterId) (float64, error) {
	return i.getLatestMetricValue(clusterId, "qps")
}

// GetLatencyMs returns the latest P99 latency value for the cluster.
// Note: This method is reserved for future use and is not currently used.
func (i *InputO11yMetricsV1) GetLatencyMs(clusterId framework.ClusterId) (float64, error) {
	return i.getLatestMetricValue(clusterId, "latency")
}

func (i *InputO11yMetricsV1) getLatestMetricValue(clusterId framework.ClusterId, metricType string) (float64, error) {
	metricName := i.getMetricNameForType(metricType)
	if metricName == "" {
		return 0, fmt.Errorf("unknown metric type: %s", metricType)
	}

	timeRange := framework.NewTimeRangeLatest(5 * time.Minute)
	metrics, err := i.GetRangeMetrics(clusterId, framework.MetricName(metricName), timeRange)
	if err != nil {
		return 0, err
	}

	if len(metrics) == 0 {
		return 0, fmt.Errorf("no metrics found for %s", metricName)
	}

	return float64(metrics[len(metrics)-1].Value), nil
}

func (i *InputO11yMetricsV1) getMetricNameForType(metricType string) string {
	switch i.componentType {
	case TIDB:
		switch metricType {
		case "qps":
			return TIDB_TOTAL_QPS
		case "latency":
			return TIDB_P99_DURATION
		}
	case TIKV:
		if metricType == "qps" {
			return TIKV_WRITE_RPC_QPS
		}
	}
	return ""
}

func (i *InputO11yMetricsV1) queryComponentMetrics(ctx context.Context, req *ComponentMetricReq) (*LabelledMetricSeries, error) {
	cachedInfo, err := i.ensureClusterInfo(ctx, req.ClusterId)
	if err != nil {
		return nil, fmt.Errorf("failed to ensure cluster info: %w", err)
	}

	templateSet, ok := METRICS_TEMPLATE_SET[req.ComponentType]
	if !ok {
		return nil, fmt.Errorf("unsupported component type: %s", req.ComponentType)
	}
	promQLTemplate, ok := templateSet[req.MetricName]
	if !ok {
		return nil, fmt.Errorf("unsupported metric name: %s", req.MetricName)
	}

	promQL, err := i.renderPromQL(promQLTemplate, req.ClusterId, req.Step)
	if err != nil {
		return nil, fmt.Errorf("failed to render promql: %w", err)
	}

	return i.queryO11yPrometheus(ctx, cachedInfo.clusterUUID, req, promQL)
}

func (i *InputO11yMetricsV1) ensureClusterInfo(ctx context.Context, clusterId string) (*cachedClusterInfo, error) {
	i.clusterInfoMu.RLock()
	if cached, ok := i.cachedClusterInfoMap[clusterId]; ok {
		i.clusterInfoMu.RUnlock()
		return cached, nil
	}
	i.clusterInfoMu.RUnlock()

	return i.fetchAndCacheClusterInfo(ctx, clusterId)
}

func (i *InputO11yMetricsV1) fetchAndCacheClusterInfo(ctx context.Context, clusterId string) (*cachedClusterInfo, error) {
	i.clusterInfoMu.Lock()
	defer i.clusterInfoMu.Unlock()

	if cached, ok := i.cachedClusterInfoMap[clusterId]; ok {
		return cached, nil
	}

	token, err := i.getAuthToken(ctx)
	if err != nil {
		return nil, fmt.Errorf("failed to get auth token: %w", err)
	}

	params := url.Values{}
	params.Set("applicationID", clusterId)

	o11yBaseURL := i.o11yConfig.O11yBaseURL
	if o11yBaseURL == "" {
		o11yBaseURL = "https://" + defaultO11yGlobalServerDomain
	}
	apiURL := fmt.Sprintf("%s/api/v1/clusters?%s", o11yBaseURL, params.Encode())

	httpReq, err := http.NewRequestWithContext(ctx, http.MethodGet, apiURL, http.NoBody)
	if err != nil {
		return nil, err
	}
	httpReq.Header.Set("Authorization", "Bearer "+token)

	resp, err := i.httpClient.Do(httpReq)
	if err != nil {
		return nil, err
	}
	defer func() { _ = resp.Body.Close() }()

	if resp.StatusCode == http.StatusUnauthorized {
		i.invalidateToken()
		return nil, fmt.Errorf("unauthorized: token expired")
	}

	if resp.StatusCode != http.StatusOK {
		body, _ := io.ReadAll(resp.Body)
		return nil, fmt.Errorf("failed to query clusters: HTTP %d: %s", resp.StatusCode, string(body))
	}

	var result o11yClusterResponse
	if err := json.NewDecoder(resp.Body).Decode(&result); err != nil {
		return nil, err
	}

	if len(result.Data.Clusters) == 0 {
		return nil, fmt.Errorf("cluster not found for clusterId %s", clusterId)
	}
	if len(result.Data.Clusters) > 1 {
		return nil, fmt.Errorf("expected exactly one cluster for applicationID %s, but got %d", clusterId, len(result.Data.Clusters))
	}

	cluster := result.Data.Clusters[0]
	cached := &cachedClusterInfo{
		clusterUUID:   cluster.ID,
		projectId:     cluster.ProjectID,
		tenantId:      cluster.TenantID,
		region:        cluster.Region,
		cloudProvider: cluster.Vendor,
	}
	i.cachedClusterInfoMap[clusterId] = cached
	return cached, nil
}

func (i *InputO11yMetricsV1) queryO11yPrometheus(ctx context.Context, clusterUUID string, req *ComponentMetricReq, promQL string) (*LabelledMetricSeries, error) {
	token, err := i.getAuthToken(ctx)
	if err != nil {
		return nil, fmt.Errorf("failed to get auth token: %w", err)
	}

	startTime, err := time.Parse(time.RFC3339, req.StartTime)
	if err != nil {
		return nil, fmt.Errorf("invalid start time: %w", err)
	}
	endTime, err := time.Parse(time.RFC3339, req.EndTime)
	if err != nil {
		return nil, fmt.Errorf("invalid end time: %w", err)
	}

	params := url.Values{}
	params.Set("query", promQL)
	params.Set("start", fmt.Sprintf("%d", startTime.Unix()))
	params.Set("end", fmt.Sprintf("%d", endTime.Unix()))
	params.Set("step", fmt.Sprintf("%ds", req.Step*60))

	o11yBaseURL := i.o11yConfig.O11yBaseURL
	if o11yBaseURL == "" {
		o11yBaseURL = "https://" + defaultO11yGlobalServerDomain
	}
	apiURL := fmt.Sprintf("%s/prometheus/%s/api/v1/query_range?%s",
		o11yBaseURL, clusterUUID, params.Encode())

	httpReq, err := http.NewRequestWithContext(ctx, http.MethodGet, apiURL, http.NoBody)
	if err != nil {
		return nil, err
	}
	httpReq.Header.Set("Authorization", "Bearer "+token)

	resp, err := i.httpClient.Do(httpReq)
	if err != nil {
		return nil, err
	}
	defer func() { _ = resp.Body.Close() }()

	if resp.StatusCode == http.StatusUnauthorized {
		i.invalidateToken()
		return nil, fmt.Errorf("unauthorized: token expired")
	}

	if resp.StatusCode != http.StatusOK {
		body, _ := io.ReadAll(resp.Body)
		return nil, fmt.Errorf("failed to query prometheus: HTTP %d: %s", resp.StatusCode, string(body))
	}

	var result prometheusResponse
	if err := json.NewDecoder(resp.Body).Decode(&result); err != nil {
		return nil, err
	}

	if result.Status != "success" {
		return nil, fmt.Errorf("prometheus query failed: %s", result.Status)
	}

	var points []DataPoint
	for _, item := range result.Data.Result {
		points = append(points, item.Values...)
	}

	return &LabelledMetricSeries{
		Name: req.MetricName,
		Labels: MetricLabels{
			ClusterId: req.ClusterId,
			Step:      req.Step,
			StartTime: startTime.Unix(),
			EndTime:   endTime.Unix(),
		},
		Points: points,
	}, nil
}

func (i *InputO11yMetricsV1) getAuthToken(ctx context.Context) (string, error) {
	i.tokenMu.RLock()
	if i.cachedToken != "" && time.Now().Before(i.tokenExpiry.Add(-60*time.Second)) {
		token := i.cachedToken
		i.tokenMu.RUnlock()
		return token, nil
	}
	i.tokenMu.RUnlock()

	return i.refreshToken(ctx)
}

func (i *InputO11yMetricsV1) refreshToken(ctx context.Context) (string, error) {
	i.tokenMu.Lock()
	defer i.tokenMu.Unlock()

	if i.cachedToken != "" && time.Now().Before(i.tokenExpiry.Add(-60*time.Second)) {
		return i.cachedToken, nil
	}

	auth0TokenURL := i.o11yConfig.Auth0TokenURL
	if auth0TokenURL == "" {
		auth0TokenURL = defaultAuth0TokenURL
	}

	auth0Audience := i.o11yConfig.Auth0Audience
	if auth0Audience == "" {
		auth0Audience = defaultAuth0Audience
	}

	payload := fmt.Sprintf(`{"client_id":%q,"client_secret":%q,"audience":%q,"grant_type":"client_credentials"}`,
		i.o11yConfig.Auth0ClientID,
		i.o11yConfig.Auth0ClientSecret,
		auth0Audience,
	)

	httpReq, err := http.NewRequestWithContext(ctx, http.MethodPost, auth0TokenURL, strings.NewReader(payload))
	if err != nil {
		return "", err
	}
	httpReq.Header.Set("Content-Type", "application/json")

	resp, err := i.httpClient.Do(httpReq)
	if err != nil {
		return "", err
	}
	defer func() { _ = resp.Body.Close() }()

	if resp.StatusCode != http.StatusOK {
		body, _ := io.ReadAll(resp.Body)
		return "", fmt.Errorf("auth0 returned HTTP %d: %s", resp.StatusCode, string(body))
	}

	var result struct {
		AccessToken string `json:"access_token"`
		ExpiresIn   int    `json:"expires_in"`
	}
	if err := json.NewDecoder(resp.Body).Decode(&result); err != nil {
		return "", err
	}

	i.cachedToken = result.AccessToken
	i.tokenExpiry = time.Now().Add(time.Duration(result.ExpiresIn) * time.Second)

	return result.AccessToken, nil
}

func (i *InputO11yMetricsV1) invalidateToken() {
	i.tokenMu.Lock()
	defer i.tokenMu.Unlock()
	i.cachedToken = ""
	i.tokenExpiry = time.Time{}
}

func (i *InputO11yMetricsV1) renderPromQL(templateStr, clusterId string, step int) (string, error) {
	tmpl, err := template.New("promql").Parse(templateStr)
	if err != nil {
		return "", fmt.Errorf("failed to parse promql template: %w", err)
	}

	var buf bytes.Buffer
	data := struct {
		ClusterId string
		Step      int
	}{
		ClusterId: clusterId,
		Step:      step,
	}

	if err := tmpl.Execute(&buf, data); err != nil {
		return "", fmt.Errorf("failed to execute promql template: %w", err)
	}

	return buf.String(), nil
}

const (
	EventInputO11yMetricsV1GetRangeMetrics           framework.EventName = "InputO11yMetricsV1.GetRangeMetrics"
	EventInputO11yMetricsV1GetMetricsError           framework.EventName = "InputO11yMetricsV1.GetMetrics.Error"
	EventInputO11yMetricsV1GetMetricsSuccess         framework.EventName = "InputO11yMetricsV1.GetMetrics.Success"
	EventInputO11yMetricsV1GetClusterTopology        framework.EventName = "InputO11yMetricsV1.GetClusterTopology"
	EventInputO11yMetricsV1GetClusterTopologyError   framework.EventName = "InputO11yMetricsV1.GetClusterTopology.Error"
	EventInputO11yMetricsV1GetClusterTopologySuccess framework.EventName = "InputO11yMetricsV1.GetClusterTopology.Success"
)