package impl

import (
	"context"
	"encoding/json"
	"fmt"
	"net/http"
	"strconv"
	"time"

	"github.com/go-resty/resty/v2"
	"github.com/pingcap/errors"
	"github.com/tidbcloud/dataflow-service-ng/app/internal/infra_metrics"
	"github.com/tidbcloud/dataflow-service-ng/app/utils"
	"github.com/tidbcloud/dataflow-service-ng/internal/config"
	"github.com/tidbcloud/pkgv2/auth/auth0"
	"github.com/tidbcloud/pkgv2/bizerr"
	"github.com/tidbcloud/pkgv2/log"
	"go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp"
	"go.uber.org/zap"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/client-go/kubernetes"
	restclient "k8s.io/client-go/rest"
)

var HTTPClientTimeout = 8 * time.Second

type infraMetricsImpl struct {
	conf   *config.O11yConfig
	client *resty.Client
}

func NewInfraMetrics() infra_metrics.InfraMetrics {
	imp := &infraMetricsImpl{
		conf:   &config.Global().O11yConfig,
		client: resty.New().SetTimeout(HTTPClientTimeout),
	}

	imp.client.SetTransport(otelhttp.NewTransport(imp.client.GetClient().Transport))
	return imp
}

// getClientToken get client token of o11y global server from auth0
func (api *infraMetricsImpl) getClientToken() (string, error) {
	o11yConfig := api.conf
	auth0ClientSecret, err := api.getAuth0SecretFromBase()
	if err != nil {
		return "", err
	}

	token, err := auth0.GetClientToken(o11yConfig.Auth0Domain, o11yConfig.Auth0ClientID,
		auth0ClientSecret, o11yConfig.Auth0Audience)
	if err != nil {
		return "", err
	}

	return token, nil
}

func (api *infraMetricsImpl) getAuth0SecretFromBase() (string, error) {
	baseCfg, err := restclient.InClusterConfig()
	if err != nil {
		return "", errors.Annotate(err, "initialize the kubernetes config failed")
	}
	baseCli, err := kubernetes.NewForConfig(baseCfg)
	if err != nil {
		return "", errors.Annotate(err, "initialize the base kube client failed")
	}

	secretNamespace := fmt.Sprintf("%s-ms", config.Global().Env)
	secretName := config.O11yAuth0Secret
	secret, err := baseCli.CoreV1().Secrets(secretNamespace).Get(context.TODO(), secretName, metav1.GetOptions{})
	if err != nil {
		return "", err
	}
	return string(secret.Data[config.O11yAuth0SecretKey]), nil
}

func (api *infraMetricsImpl) queryO11yClusterUUID(
	ctx context.Context,
	cluster *utils.Cluster,
) (string, bizerr.BizError) {
	token, err := api.getClientToken()
	if err != nil {
		return "", bizerr.New(bizerr.ErrCodeSvcO11y, "failed to get auth0 token, %s", err.Error())
	}

	o11yConfig := api.conf
	queryURL := fmt.Sprintf("https://%s/api/v1/clusters/", o11yConfig.GlobalServerDomain)
	client := api.client.R().EnableTrace().SetContext(ctx).SetAuthToken(token)
	client.SetQueryParams(map[string]string{
		"vendor":        cluster.GetCloudProvider(),
		"region":        cluster.GetRegionName(),
		"tenantID":      strconv.FormatUint(cluster.GetTenantID(), 10),
		"bizType":       o11yConfig.BizType,
		"applicationID": strconv.FormatUint(cluster.GetClusterID(), 10),
	})
	resp, err := client.Get(queryURL)
	if err != nil {
		return "", bizerr.New(bizerr.ErrCodeSvcO11y, "failed to query o11y cluster uuid, %s", err.Error())
	}
	if resp.StatusCode() != http.StatusOK {
		return "", bizerr.New(bizerr.ErrCodeSvcO11y, "failed to query o11y cluster uuid, body: %s", resp.String())
	}
	result := &infra_metrics.O11yClusterResponse{}
	if err = json.Unmarshal(resp.Body(), result); err != nil {
		return "", bizerr.New(bizerr.ErrCodeSvcO11y, "failed to unmarshal response body: %s", err)
	}

	if len(result.Data.Clusters) == 0 {
		return "", bizerr.New(bizerr.ErrCodeSvcO11y, "failed to query o11y cluster uuid, no cluster found")
	}

	return result.Data.Clusters[0].ID, nil
}

func (api *infraMetricsImpl) queryClusterMetrics(
	ctx context.Context,
	cluster *utils.Cluster,
	expr string,
	o11yClusterUUID string,
	params map[string]string,
) ([]byte, bizerr.BizError) {
	logger := log.WithContextS(ctx).With(zap.Uint64("clusterID", cluster.GetClusterID())).With(zap.String("expr", expr))

	queryURL := fmt.Sprintf("https://www.ds.%s.%s.observability.tidbcloud.com/internal/metrics/%s/api/v1/query",
		cluster.GetRegionName(), cluster.GetCloudProvider(), o11yClusterUUID)

	client := api.client.R().EnableTrace().SetContext(ctx)
	client.SetQueryParams(params)
	resp, rawErr := client.Get(queryURL)
	if rawErr != nil {
		logger.Error("failed to query metrics", zap.Error(rawErr), zap.Any("url", queryURL),
			zap.Any("params", params))
		return nil, bizerr.New(bizerr.ErrCodeSvcO11y, "failed to query metrics, %s", rawErr.Error())
	}
	if resp.StatusCode() != http.StatusOK {
		logger.Error("failed to query metrics", zap.Error(rawErr), zap.Any("url", queryURL),
			zap.Any("params", params))
		return nil, bizerr.New(bizerr.ErrCodeSvcO11y, "failed to query metrics, body: %s", resp.String())
	}

	logger.Infof("query metrics response: %s", resp.String())
	return resp.Body(), nil
}

func (api *infraMetricsImpl) QueryClusterMetrics(
	ctx context.Context,
	cluster *utils.Cluster,
	expr string,
	queryTime time.Time,
) ([]byte, bizerr.BizError) {
	logger := log.WithContextS(ctx).With(zap.Uint64("clusterID", cluster.GetClusterID())).With(zap.String("expr", expr))

	o11yClusterUUID, err := api.queryO11yClusterUUID(ctx, cluster)
	if err != nil {
		logger.Error("failed to query o11y cluster uuid", zap.Error(err))
		return nil, err
	}
	logger.Infof("o11y cluster uuid: %s", o11yClusterUUID)
	timeString := queryTime.Format(time.RFC3339)
	params := map[string]string{
		string(infra_metrics.QueryPromSQLType): expr,
		"time":                                 timeString,
	}
	result, err := api.queryClusterMetrics(ctx, cluster, expr, o11yClusterUUID, params)
	if err != nil {
		logger.Error("failed to query cluster metrics", zap.Error(err))
		return nil, err
	}
	return result, nil
}

func (api *infraMetricsImpl) QueryClusterMetricsWithParam(
	ctx context.Context,
	cluster *utils.Cluster,
	expr string,
	params map[string]string,
) ([]byte, bizerr.BizError) {
	logger := log.WithContextS(ctx).With(zap.Uint64("clusterID", cluster.GetClusterID())).With(zap.String("expr", expr))

	o11yClusterUUID, err := api.queryO11yClusterUUID(ctx, cluster)
	if err != nil {
		logger.Error("failed to query o11y cluster uuid", zap.Error(err))
		return nil, err
	}
	logger.Infof("o11y cluster uuid: %s", o11yClusterUUID)
	result, err := api.queryClusterMetrics(ctx, cluster, expr, o11yClusterUUID, params)
	if err != nil {
		logger.Error("failed to query cluster metrics", zap.Error(err))
		return nil, err
	}
	return result, nil
}

func (api *infraMetricsImpl) QueryRangeClusterMetrics(
	ctx context.Context,
	cluster *utils.Cluster,
	expr string,
) ([]byte, bizerr.BizError) {
	logger := log.WithContextS(ctx).With(zap.Uint64("clusterID", cluster.GetClusterID())).With(zap.String("expr", expr))

	o11yClusterUUID, err := api.queryO11yClusterUUID(ctx, cluster)
	if err != nil {
		logger.Error("failed to query o11y cluster uuid", zap.Error(err))
		return nil, err
	}
	logger.Infof("o11y cluster uuid: %s", o11yClusterUUID)

	params := map[string]string{
		string(infra_metrics.QueryRangePromSQLType): expr,
	}
	result, err := api.queryClusterMetrics(ctx, cluster, expr, o11yClusterUUID, params)
	if err != nil {
		logger.Error("failed to query cluster metrics", zap.Error(err))
		return nil, err
	}
	return result, nil
}