package datasource

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

	"github.com/pingcap/log"
	"github.com/pkg/errors"
	"github.com/tidbcloud/metrics-integration/pipeline/central/config"
	"github.com/tidbcloud/metrics-integration/pipeline/central/entity"
	"github.com/tidbcloud/metrics-integration/pipeline/central/metrics"
	"github.com/tidbcloud/pkgv2/auth/auth0"
	"go.uber.org/zap"
)

type MetricDataAddress struct {
	IntegrationType  string
	TidbClusterInfo  TidbClusterInfo
	QueryUrl         string
	ProjectQueryUrl  string
	QueryHost        string
	ProjectQueryHost string
	IsThanos         bool
}

type metricDataAddresses struct {
	addresses map[entity.ID]MetricDataAddress
	lock      sync.RWMutex
}

func (addrs *metricDataAddresses) set(tidbClusterID entity.ID, addr MetricDataAddress) {
	addrs.lock.Lock()
	defer addrs.lock.Unlock()

	addrs.addresses[tidbClusterID] = addr
}

func (addrs *metricDataAddresses) get(tidbClusterID entity.ID) (MetricDataAddress, bool) {
	addrs.lock.RLock()
	defer addrs.lock.RUnlock()

	addr, ok := addrs.addresses[tidbClusterID]
	return addr, ok
}

type integrationMetricDataAddresses struct {
	integrationType string
	addresses       metricDataAddresses
}

const (
	datadogIntegrationType   string = "datadog"
	thirdPromIntegrationType string = "thirdProm"
)

var (
	datadogIntegrationTidbClusterMetricDataAddresses = integrationMetricDataAddresses{
		integrationType: datadogIntegrationType,
		addresses: metricDataAddresses{
			addresses: map[entity.ID]MetricDataAddress{},
			lock:      sync.RWMutex{},
		},
	}

	thirdPromIntegrationTidbClusterMetricDataAddresses = integrationMetricDataAddresses{
		integrationType: thirdPromIntegrationType,
		addresses: metricDataAddresses{
			addresses: map[entity.ID]MetricDataAddress{},
			lock:      sync.RWMutex{},
		},
	}
)

func (iaddrs *integrationMetricDataAddresses) setMetricDataAddress(tidbClusterID entity.ID, addr MetricDataAddress) {
	defer func() {
		log.Debug("setMetricDataAddress result",
			zap.String("integrationType", iaddrs.integrationType),
			zap.Uint64("tidb_cluster_id", tidbClusterID.Uint64()),
			zap.String("QueryUrl", addr.QueryUrl),
			zap.Bool("IsThanos", addr.IsThanos))
	}()

	iaddrs.addresses.set(tidbClusterID, addr)
}

func (iaddrs *integrationMetricDataAddresses) getMetricDataAddress(tidbClusterID entity.ID) (addr MetricDataAddress, exists bool) {
	defer func() {
		log.Debug("getMetricDataAddress result",
			zap.String("integrationType", iaddrs.integrationType),
			zap.Uint64("tidb_cluster_id", tidbClusterID.Uint64()),
			zap.String("QueryUrl", addr.QueryUrl),
			zap.Bool("IsThanos", addr.IsThanos),
			zap.Bool("addr_exists", exists))
	}()

	return iaddrs.addresses.get(tidbClusterID)
}

type TidbClusterInfo struct {
	Vendor      string
	Region      string
	TenantID    entity.ID
	ProjectID   entity.ID
	ClusterID   entity.ID
	ServicePlan string
}

func shouldSkipProjectDatasourceQuery(info TidbClusterInfo) bool {
	if isDedicatedServicePlan(info.ServicePlan) {
		return info.ProjectID.Uint64() == 0
	}

	if isNextGenServicePlan(info.ServicePlan) {
		return true
	}

	return info.ProjectID.Uint64() == 0
}

func isDedicatedServicePlan(servicePlan string) bool {
	sp := strings.TrimSpace(servicePlan)
	// Empty ServicePlan (from old configs without this field) is treated as Dedicated.
	return sp == "" || strings.EqualFold(sp, "Dedicated")
}

func isNextGenServicePlan(servicePlan string) bool {
	return entity.ServicePlan(servicePlan).IsNextGen()
}

func (iaddrs *integrationMetricDataAddresses) getAndSetMetricDataAddressWithTidbClusterInfo(ctx context.Context, info TidbClusterInfo) (addr MetricDataAddress, err error) {
	defer func() {
		if err == nil {
			iaddrs.setMetricDataAddress(info.ClusterID, addr)
		}

		log.Debug("getAndSetMetricDataAddressWithTidbClusterInfo result",
			zap.String("integrationType", iaddrs.integrationType),
			zap.String("tidb_cluster_vendor", info.Vendor),
			zap.String("tidb_cluster_region", info.Region),
			zap.Uint64("tidb_cluster_tenant_id", info.TenantID.Uint64()),
			zap.Uint64("tidb_cluster_project_id", info.ProjectID.Uint64()),
			zap.Uint64("tidb_cluster_id", info.ClusterID.Uint64()),
			zap.String("service_plan", info.ServicePlan),
			zap.String("IntegrationType", addr.IntegrationType),
			zap.String("QueryUrl", addr.QueryUrl),
			zap.String("ProjectQueryUrl", addr.ProjectQueryUrl),
			zap.Bool("IsThanos", addr.IsThanos),
			zap.Error(err))
	}()

	addr, exists := iaddrs.getMetricDataAddress(info.ClusterID)
	if !exists || addr.IsThanos {
		queryUrl, queryHost, projectQueryUrl, projectQueryHost, err := getMetricDataQueryUrlWithRetry(ctx, info)
		if err != nil {
			log.Error("getAndSetMetricDataAddressWithTidbClusterInfo call getMetricDataQueryUrlWithRetry fail",
				zap.String("integrationType", iaddrs.integrationType),
				zap.String("tidb_cluster_vendor", info.Vendor),
				zap.String("tidb_cluster_region", info.Region),
				zap.Uint64("tidb_cluster_tenant_id", info.TenantID.Uint64()),
				zap.Uint64("tidb_cluster_project_id", info.ProjectID.Uint64()),
				zap.Uint64("tidb_cluster_id", info.ClusterID.Uint64()),
				zap.String("queryUrl", queryUrl),
				zap.String("projectQueryUrl", projectQueryUrl),
				zap.Error(err))
			return MetricDataAddress{}, err
		}

		addr = MetricDataAddress{
			IntegrationType:  iaddrs.integrationType,
			TidbClusterInfo:  info,
			QueryUrl:         queryUrl,
			ProjectQueryUrl:  projectQueryUrl,
			QueryHost:        queryHost,
			ProjectQueryHost: projectQueryHost,
			IsThanos:         false,
		}
	}

	return addr, nil
}

type queryClustersResult struct {
	Code    int               `json:"code"`    //http error code
	Message string            `json:"message"` //http error msg
	Data    queryClustersResp `json:"data"`    //The returned content, storage in JSON format
}

type queryClustersResp struct {
	Clusters           []clusterBaseInfo `json:"clusters"`
	ContinueToken      string            `json:"continueToken" form:"continueToken"`
	RemainingItemCount int64             `json:"remainingItemCount" form:"remainingItemCount"`
}

type clusterBaseInfo struct {
	ID              string `json:"id" form:"id"`
	Vendor          string `json:"vendor" form:"vendor"`
	Region          string `json:"region" form:"region"`
	TenantID        string `json:"tenantID" form:"tenantID"`
	ProjectID       string `json:"projectID" form:"projectID"`
	ApplicationID   string `json:"applicationID" form:"applicationID"`
	BizType         string `json:"bizType" form:"bizType"`
	InternalReadUri string `json:"internalReadUri" form:"internalReadUri"`

	Name      string   `json:"name" form:"name"`
	Labels    []string `json:"labels" form:"labels"`
	Version   string   `json:"version" form:"version"`
	ShareMode string   `json:"shareMode" form:"shareMode"`
	Note      string   `json:"note" form:"note"`
}

var errNoMetricDataClusterFound = errors.New("get no metric clusters for tidb cluster from o11y global server")

func getMetricDataQueryUrlWithRetry(ctx context.Context, info TidbClusterInfo) (queryUrl string, queryHost string, projectQueryUrl string, projectQueryHost string, err error) {
	defer func() {
		if err != nil {
			log.Error("getMetricDataQueryUrlWithRetry process fail",
				zap.String("tidb_cluster_vendor", info.Vendor),
				zap.String("tidb_cluster_region", info.Region),
				zap.Uint64("tidb_cluster_tenant_id", info.TenantID.Uint64()),
				zap.Uint64("tidb_cluster_project_id", info.ProjectID.Uint64()),
				zap.Uint64("tidb_cluster_id", info.ClusterID.Uint64()),
				zap.String("service_plan", info.ServicePlan),
				zap.Error(err),
			)
		}
	}()

	reqCnt := 3
	for i := 0; i < reqCnt; i++ {
		queryUrl, queryHost, projectQueryUrl, projectQueryHost, canRetry, err := getMetricDataQueryUrl(ctx, info)
		if err == nil {
			return queryUrl, queryHost, projectQueryUrl, projectQueryHost, nil
		}

		if i == reqCnt-1 {
			return "", "", "", "", err
		}

		if canRetry {
			time.Sleep(100 * time.Millisecond * time.Duration(i+1))
			continue
		}

		return "", "", "", "", err
	}

	return "", "", "", "", errors.New("unknown error for get metric data query url with retry")
}

func getMetricDataQueryUrl(ctx context.Context, info TidbClusterInfo) (queryUrl string, queryHost string, projectQueryUrl string, projectQueryHost string, canRetry bool, err error) {
	o11yGlobalServerUrl := getO11yGlobalServerURL()
	o11yBizType := config.Global().O11yGlobalServerConfig.GetBizTypeByServicePlan(info.ServicePlan)
	if o11yBizType == "" {
		return "", "", "", "", false, fmt.Errorf("bizType is not configured for service plan %q", info.ServicePlan)
	}
	path := "/api/v1/clusters/"

	// For NextGen clusters, projectID should not be included in the query.
	useNextGenRouting := isNextGenServicePlan(info.ServicePlan)
	var urlQuery, projectUrlQuery string
	if useNextGenRouting {
		// NextGen cluster: no projectID in query parameters
		urlQuery = fmt.Sprintf("vendor=%s&region=%s&tenantID=%d&applicationID=%d&bizType=%s",
			info.Vendor, info.Region, info.TenantID.Uint64(), info.ClusterID.Uint64(), o11yBizType)
		projectUrlQuery = fmt.Sprintf("vendor=%s&region=%s&tenantID=%d&applicationID=%s&bizType=%s",
			info.Vendor, info.Region, info.TenantID.Uint64(), "k8s-infra", o11yBizType)
	} else {
		// Traditional cluster (Dedicated): include projectID
		urlQuery = fmt.Sprintf("vendor=%s&region=%s&tenantID=%d&projectID=%d&applicationID=%d&bizType=%s",
			info.Vendor, info.Region, info.TenantID.Uint64(), info.ProjectID.Uint64(), info.ClusterID.Uint64(), o11yBizType)
		projectUrlQuery = fmt.Sprintf("vendor=%s&region=%s&tenantID=%d&projectID=%d&applicationID=%s&bizType=%s",
			info.Vendor, info.Region, info.TenantID.Uint64(), info.ProjectID.Uint64(), "k8s-infra", o11yBizType)
	}
	urlString := fmt.Sprintf("%s%s?%s", o11yGlobalServerUrl, path, urlQuery)
	projectUrlString := fmt.Sprintf("%s%s?%s", o11yGlobalServerUrl, path, projectUrlQuery)

	startTime := time.Now()
	method := "GET"
	defer func() {
		duration := float64(time.Since(startTime)) / float64(time.Millisecond)
		metrics.O11yServerRequestDuration.WithLabelValues(o11yGlobalServerUrl, path, method).Observe(duration)
		if err != nil {
			metrics.O11yServerRequestErrCnt.WithLabelValues(o11yGlobalServerUrl, path, method).Inc()
		}

		log.Info("getMetricDataQueryUrl result",
			zap.String("service_plan", info.ServicePlan),
			zap.String("biztype", o11yBizType),
			zap.String("urlString", urlString),
			zap.String("queryUrl", queryUrl),
			zap.String("queryHost", queryHost),
			zap.String("projectUrlString", projectUrlString),
			zap.String("projectUrlQuery", projectUrlQuery),
			zap.String("projectQueryHost", projectQueryHost),
			zap.Error(err),
		)
	}()

	token, err := auth0.GetClientToken(
		config.Global().O11yGlobalServerConfig.Auth0Domain,
		config.Global().O11yGlobalServerConfig.Auth0ClientID,
		config.Global().O11yGlobalServerConfig.Auth0ClientSecret,
		config.Global().O11yGlobalServerConfig.Auth0Audience)
	if err != nil {
		return "", "", "", "", true, err
	}

	orgId := fmt.Sprintf("%d", info.TenantID.Uint64())

	o11yCluster, err := getO11YClusterInfo(ctx, method, urlString, token, o11yBizType, orgId)
	if err != nil {
		return "", "", "", "", false, err
	}

	queryUrl = rewriteRegionalServerURL(o11yCluster.Data.Clusters[0].InternalReadUri)
	queryHost = getURLHost(o11yCluster.Data.Clusters[0].InternalReadUri)
	if len(queryUrl) == 0 {
		return "", "", "", "", false, errors.New("internalReadUri is empty or not exists for tidb cluster from o11y global server")
	}

	if shouldSkipProjectDatasourceQuery(info) {
		return queryUrl, queryHost, "", "", false, nil
	}

	projectO11yCluster, err := getO11YClusterInfo(ctx, method, projectUrlString, token, o11yBizType, orgId)
	if err != nil {
		return "", "", "", "", false, err
	}

	projectQueryUrl = rewriteRegionalServerURL(projectO11yCluster.Data.Clusters[0].InternalReadUri)
	projectQueryHost = getURLHost(projectO11yCluster.Data.Clusters[0].InternalReadUri)
	if len(projectQueryUrl) == 0 {
		return "", "", "", "", false, errors.New("internalReadUri is empty or not exists for tidb cluster from o11y global server")
	}

	return queryUrl, queryHost, projectQueryUrl, projectQueryHost, false, nil
}

func getO11YClusterInfo(ctx context.Context, method string, url string, token string, bizType string, orgId string) (queryClustersResult, error) {
	result := queryClustersResult{}
	req, err := http.NewRequestWithContext(ctx, method, url, nil)
	if err != nil {
		return result, err
	}
	if host := getOriginalHostForLocalURL(url, config.Global().O11yGlobalServerConfig.Url); host != "" {
		req.Host = host
	}
	req.Header.Add("Authorization", fmt.Sprintf("Bearer %s", token))
	req.Header.Add("X-Biz-Type", bizType)
	req.Header.Add("X-Org-Id", orgId)

	resp, err := http.DefaultClient.Do(req)
	if err != nil {
		return result, err
	}
	defer resp.Body.Close() //nolint

	if resp.StatusCode != http.StatusOK {
		body, _ := io.ReadAll(resp.Body)
		return result, fmt.Errorf("http response status %s from o11y global server, body: %s", resp.Status, body)
	}

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

	if result.Code != 0 {
		return result, fmt.Errorf("o11y global server process fail, result code: %d, message: %s", result.Code, result.Message)
	}

	metricDataClustersLen := len(result.Data.Clusters)
	if metricDataClustersLen == 0 {
		return result, errNoMetricDataClusterFound
	}

	if metricDataClustersLen > 1 {
		return result, fmt.Errorf("get %d metric clusters for tidb cluster from o11y global server", metricDataClustersLen)
	}

	return result, nil
}

func getO11yGlobalServerURL() string {
	cfg := config.Global().O11yGlobalServerConfig
	if cfg.LocalTestEnabled && cfg.LocalTestGlobalServerURL != "" {
		return cfg.LocalTestGlobalServerURL
	}

	return cfg.Url
}

func rewriteRegionalServerURL(rawURL string) string {
	cfg := config.Global().O11yGlobalServerConfig
	if !cfg.LocalTestEnabled || rawURL == "" {
		return rawURL
	}

	originalURL, err := url.Parse(rawURL)
	if err != nil {
		return rawURL
	}

	// Check per-hostname slice first (NextGen plans may have dedicated regional servers).
	for _, m := range cfg.LocalTestRegionalServerURLMappings {
		if originalURL.Host == m.Host {
			localURL, err := url.Parse(m.URL)
			if err == nil && localURL.Scheme != "" && localURL.Host != "" {
				originalURL.Scheme = localURL.Scheme
				originalURL.Host = localURL.Host
				return originalURL.String()
			}
		}
	}

	if cfg.LocalTestRegionalServerURL == "" {
		return rawURL
	}

	localURL, err := url.Parse(cfg.LocalTestRegionalServerURL)
	if err != nil || localURL.Scheme == "" || localURL.Host == "" {
		return rawURL
	}

	originalURL.Scheme = localURL.Scheme
	originalURL.Host = localURL.Host
	return originalURL.String()
}

func getURLHost(rawURL string) string {
	parsedURL, err := url.Parse(rawURL)
	if err != nil {
		return ""
	}

	return parsedURL.Host
}

func getOriginalHostForLocalURL(requestURL string, originalURL string) string {
	requestHost := getURLHost(requestURL)
	originalHost := getURLHost(originalURL)
	if requestHost == "" || originalHost == "" {
		return ""
	}

	cfg := config.Global().O11yGlobalServerConfig
	localGlobalHost := getURLHost(cfg.LocalTestGlobalServerURL)
	localRegionalHost := getURLHost(cfg.LocalTestRegionalServerURL)
	if requestHost == localGlobalHost || requestHost == localRegionalHost {
		return originalHost
	}

	// Also recognise any host from the per-hostname mappings.
	for _, m := range cfg.LocalTestRegionalServerURLMappings {
		if requestHost == getURLHost(m.URL) {
			return originalHost
		}
	}

	return ""
}

type metricDataAddress struct{}

var (
	metricDataAddressKey metricDataAddress
)

func SetMetricDataAddressToContextWithTidbClusterInfo(ctx context.Context, info TidbClusterInfo) (context.Context, error) {
	addr, err := datadogIntegrationTidbClusterMetricDataAddresses.getAndSetMetricDataAddressWithTidbClusterInfo(ctx, info)
	if err != nil {
		return ctx, err
	}

	return context.WithValue(ctx, metricDataAddressKey, addr), nil
}

func SetMetricDataAddressToContextWithCache(ctx context.Context, tidbClusterID entity.ID) (context.Context, error) {
	addr, ok := datadogIntegrationTidbClusterMetricDataAddresses.getMetricDataAddress(tidbClusterID)
	if !ok {
		return ctx, errors.New("addr not exists in cache")
	}

	if len(addr.QueryUrl) == 0 {
		return ctx, errors.New("invalid metric_data_addr in cache")
	}

	return context.WithValue(ctx, metricDataAddressKey, addr), nil
}

func SetMetricDataAddressToContextWithTidbClusterInfoV2(ctx context.Context, info TidbClusterInfo) (context.Context, error) {
	addr, err := thirdPromIntegrationTidbClusterMetricDataAddresses.getAndSetMetricDataAddressWithTidbClusterInfo(ctx, info)
	if err != nil {
		return ctx, err
	}

	return context.WithValue(ctx, metricDataAddressKey, addr), nil
}

func GetMetricDataAddressFromContext(ctx context.Context) (addr MetricDataAddress, err error) {
	if ctx == nil {
		return MetricDataAddress{}, errors.New("ctx is nil")
	}

	value := ctx.Value(metricDataAddressKey)
	if value == nil {
		return MetricDataAddress{}, errors.New("no metric_data_addr in context")
	}

	addr, ok := value.(MetricDataAddress)
	if !ok || len(addr.QueryUrl) == 0 {
		return MetricDataAddress{}, errors.New("invalid metric_data_addr in context")
	}

	return addr, nil
}