package resourcemanager

import (
	"context"
	"encoding/json"
	"fmt"
	"io"
	"sync"
	"time"

	"github.com/patrickmn/go-cache"

	"github.com/ddliu/go-httpclient"
	"github.com/pingcap/log"
	v1 "github.com/tidbcloud/observability/api/regionalserver/resourceworker/v1"
	"github.com/tidbcloud/observability/pkg/common/configs"
	"github.com/tidbcloud/observability/pkg/common/constants"
	"github.com/tidbcloud/observability/pkg/common/errors"
	"github.com/tidbcloud/observability/pkg/common/message"
	"go.uber.org/zap"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"sigs.k8s.io/controller-runtime/pkg/client"
)

type clusterReaderWriter struct {
	cli     client.Client
	readCli client.Reader
}

var lock sync.RWMutex
var clusterUniqueKeyCache = cache.New(constants.DefaultCacheTimeout, constants.DefaultCacheTimeout*2)

func buildUniqueKey(data *v1.Cluster) string {
	bizType := data.Labels[constants.LabelBizType]
	tenantID := data.Labels[constants.LabelTenantID]
	projectID := data.Labels[constants.LabelProjectID]
	applicationID := data.Labels[constants.LabelApplicationID]

	return fmt.Sprintf("%s/%s/%s/%s", bizType, tenantID, projectID, applicationID)
}

func (p clusterReaderWriter) CreateCluster(ctx context.Context, data *v1.Cluster) error {
	lock.Lock()
	defer lock.Unlock()

	key := buildUniqueKey(data)

	requestID, ok := ctx.Value(constants.OpenAPIHeaderRequestId).(string)
	if !ok {
		log.Warn("fetch request id failed")
	} else if requestID == "" {
		log.Warn("request id is nil")
	} else {
		data.Annotations[string(constants.OpenAPIHeaderRequestId)] = requestID
	}

	if id, ok := clusterUniqueKeyCache.Get(key); ok {
		log.Warn("duplicated cluster unique key",
			zap.String("Key", key),
			zap.Any("ID", id))
		return errors.New(errors.DuplicatedCluster, "")
	} else {
		err := p.cli.Create(ctx, data)
		if err != nil {
			return WrapClientError(err)
		} else {
			clusterUniqueKeyCache.Set(key, data.Name, 60*time.Second)
			return nil
		}
	}
}

func (p clusterReaderWriter) GetCluster(ctx context.Context, id string) (*v1.Cluster, error) {
	clusters := new(v1.ClusterList)
	matchingLabels := client.MatchingLabels{
		constants.LabelClusterID: id,
	}

	err := p.cli.List(ctx, clusters,
		matchingLabels,
	)

	if err == nil && len(clusters.Items) == 0 {
		return nil, errors.NewErrorf(errors.ClusterNotFound, "cluster %s is not found", id)
	}

	if err == nil && len(clusters.Items) > 0 {
		return &clusters.Items[0], nil

	}

	return nil, WrapClientError(err)
}

func (p clusterReaderWriter) IsClusterIDExists(ctx context.Context, id string) (bool, error) {
	clusters := new(v1.ClusterList)
	matchingLabels := client.MatchingLabels{
		constants.LabelClusterID: id,
	}

	err := p.cli.List(ctx, clusters,
		matchingLabels,
	)

	if err == nil && len(clusters.Items) > 0 {
		return true, nil
	}

	return false, WrapClientError(err)
}

func (p clusterReaderWriter) IsClusterExists(ctx context.Context, tenantID, projectID, applicationID, bizType string) (bool, error) {
	list, err := p.QueryClusters(ctx, tenantID, projectID, applicationID, bizType, message.ContinueRequest{
		Limit: 1,
	})

	if err == nil && len(list.Items) > 0 {
		return true, nil
	}

	return false, WrapClientError(err)
}

func (p clusterReaderWriter) QueryClusters(ctx context.Context, tenantID, projectID, applicationID, bizType string, cr message.ContinueRequest) (*v1.ClusterList, error) {
	clusters := new(v1.ClusterList)
	matchingLabels := client.MatchingLabels{
		constants.LabelVendor: configs.GetServiceInfo().Vendor,
		constants.LabelRegion: configs.GetServiceInfo().Region,
	}

	if len(tenantID) > 0 {
		matchingLabels[constants.LabelTenantID] = tenantID
	}
	if len(projectID) > 0 {
		matchingLabels[constants.LabelProjectID] = projectID
	}
	if len(applicationID) > 0 {
		matchingLabels[constants.LabelApplicationID] = applicationID
	}
	if len(bizType) > 0 {
		matchingLabels[constants.LabelBizType] = bizType
	}

	err := p.cli.List(ctx, clusters,
		matchingLabels,
		client.Limit(cr.Limit),
		client.Continue(cr.ContinueToken),
	)

	return clusters, WrapClientError(err)
}

func (p clusterReaderWriter) ListClusters(ctx context.Context, opts ...client.ListOption) (*v1.ClusterList, error) {
	clusters := new(v1.ClusterList)
	if err := p.cli.List(ctx, clusters, opts...); err != nil {
		return nil, WrapClientError(err)
	}
	return clusters, nil
}

func (p clusterReaderWriter) QueryClustersByRuleTemplateID(ctx context.Context, ruleTemplateID string, cr message.ContinueRequest) (*v1.ClusterList, error) {
	clusters := new(v1.ClusterList)
	matchingLabels := client.MatchingLabels{
		constants.LabelVendor:     configs.GetServiceInfo().Vendor,
		constants.LabelRegion:     configs.GetServiceInfo().Region,
		constants.LabelTemplateID: ruleTemplateID,
	}

	if len(ruleTemplateID) <= 0 {
		return clusters, errors.New(errors.PrometheusRuleTemplateIDIsEmpty, "templateID not allow empty")
	}

	var err error
	if cr.ContinueToken == "" {
		err = p.readCli.List(ctx, clusters,
			matchingLabels,
			client.Limit(cr.Limit),
		)
	} else {
		err = p.readCli.List(ctx, clusters,
			matchingLabels,
			client.Limit(cr.Limit),
			client.Continue(cr.ContinueToken),
		)
	}
	return clusters, WrapClientError(err)
}

func (p clusterReaderWriter) DeleteCluster(ctx context.Context, id string) (err error) {
	cluster, err := p.GetCluster(ctx, id)
	if err != nil {
		return err
	}

	return WrapClientError(p.cli.Delete(ctx, &v1.Cluster{
		ObjectMeta: metav1.ObjectMeta{
			Name:      id,
			Namespace: cluster.Namespace,
		},
	}))
}

func (p clusterReaderWriter) SoftDeleteCluster(ctx context.Context, id string) (err error) {
	cluster, err := p.GetCluster(ctx, id)
	if err != nil {
		return err
	}

	if cluster.Labels[constants.LabelShareMode] == string(constants.EXCLUSIVE) || cluster.Labels[constants.LabelShareMode] == string(constants.ENHANCED) {
		err = createSnapshot(ctx, cluster)
		if err != nil {
			log.Error("SoftDeleteCluster failed, Because the call to createSnapshot failed", zap.Error(err))
			return WrapClientError(err)
		}
	}

	err = p.cli.Update(ctx, cluster)
	if err != nil {
		return WrapClientError(err)
	}

	cluster, err = p.GetCluster(ctx, id)
	if err != nil {
		return err
	}

	cluster.Status.Phase = constants.TO_BE_DELETED
	return WrapClientError(p.cli.Status().Update(ctx, cluster))

}

func (p clusterReaderWriter) UpdateCluster(ctx context.Context, data *v1.Cluster) error {
	return WrapClientError(p.cli.Update(ctx, data))
}

func (p clusterReaderWriter) SoftPauseCluster(ctx context.Context, cluster *v1.Cluster) (err error) {
	if cluster.Annotations[constants.AnnotationShareMode] == string(constants.EXCLUSIVE) || cluster.Annotations[constants.AnnotationShareMode] == string(constants.ENHANCED) {
		err = createSnapshot(ctx, cluster)
		if err != nil {
			log.Error("SoftPauseCluster failed, Because the call to createSnapshot failed", zap.Error(err))
			return WrapClientError(err)
		}
		err = p.cli.Update(ctx, cluster)
		if err != nil {
			return WrapClientError(err)
		}
	}

	cluster, err = p.GetCluster(ctx, cluster.Name)
	if err != nil {
		return err
	}

	cluster.Status.Phase = constants.TO_BE_PAUSED
	return WrapClientError(p.cli.Status().Update(ctx, cluster))
}

func (p clusterReaderWriter) PauseCluster(ctx context.Context, cluster *v1.Cluster) (err error) {
	err = p.cli.Update(ctx, cluster)
	if err != nil {
		return WrapClientError(err)
	}

	cluster.Status.Phase = constants.PAUSING
	return WrapClientError(p.cli.Status().Update(ctx, cluster))
}

func (p clusterReaderWriter) ResumeCluster(ctx context.Context, cluster *v1.Cluster) (err error) {
	err = p.cli.Update(ctx, cluster)
	if err != nil {
		return WrapClientError(err)
	}

	cluster.Status.Phase = constants.RESUMING
	cluster.Status.UpdateAt = time.Now().UTC().Format(time.RFC3339)
	return WrapClientError(p.cli.Status().Update(ctx, cluster))
}

var post = func(url string) ([]byte, error) {
	resp, err := httpclient.Do("POST", url, map[string]string{}, nil)
	if err == nil {
		return io.ReadAll(resp.Body)
	}
	return nil, err
}

func createSnapshot(ctx context.Context, cluster *v1.Cluster) error {
	url := fmt.Sprintf("http://svc-%s.%s.svc.cluster.local:%d/api/v1/admin/tsdb/snapshot?skip_head=false",
		cluster.Name, cluster.Namespace, 9090)

	data, err := post(url)
	if err != nil {
		err = errors.Wrap(errors.HttpClientResponseError, "", err)
		return err
	}

	respBody := new(SnapshotRespBody)

	err = json.Unmarshal(data, respBody)
	if err != nil {
		err = errors.NewErrorf(errors.UnmarshalError, "%s", err.Error())
		return err
	}

	if respBody.Status != "success" || len(respBody.Data.Name) == 0 {
		err = errors.NewErrorf(errors.UnrecognizedError, "create prometheus snapshot get resp : %s", string(data))
		return err
	}

	cluster.Annotations[constants.AnnotationNewDataSnapshot] = respBody.Data.Name
	return nil
}

type SnapshotRespBody struct {
	Status string `json:"status"`
	Data   struct {
		Name string `json:"name"`
	} `json:"data"`
}

type rdsclusterReaderWriter struct {
	rdscli client.Client
	clusterReaderWriter
}

func (p rdsclusterReaderWriter) CreateCluster(ctx context.Context, data *v1.Cluster) error {
	// unique index for <biz,tenant,project,application> will be applied at rds.
	err := p.rdscli.Create(ctx, data)
	if err != nil {
		return WrapClientError(err)
	}
	return nil
}
func (p rdsclusterReaderWriter) IsClusterIDExists(ctx context.Context, id string) (bool, error) {
	clusters := new(v1.ClusterList)
	matchingLabels := client.MatchingLabels{
		constants.LabelClusterID: id,
	}

	err := p.rdscli.List(ctx, clusters,
		matchingLabels,
	)

	if err == nil && len(clusters.Items) > 0 {
		return true, nil
	}
	// fall back to k8s cli
	return p.clusterReaderWriter.IsClusterIDExists(ctx, id)
}

func (p rdsclusterReaderWriter) ListClusters(ctx context.Context, opts ...client.ListOption) (*v1.ClusterList, error) {
	clusters := new(v1.ClusterList)
	err := p.rdscli.List(ctx, clusters, opts...)
	log.Warn("ListClusters via rdscli", zap.Error(err))
	if len(clusters.Items) > 0 {
		return clusters, nil
	}
	// fall back to k8s cli
	return p.clusterReaderWriter.ListClusters(ctx, opts...)
}

func (p rdsclusterReaderWriter) QueryClusters(ctx context.Context, tenantID, projectID, applicationID, bizType string, cr message.ContinueRequest) (*v1.ClusterList, error) {
	clusters := new(v1.ClusterList)
	matchingLabels := client.MatchingLabels{}

	if len(tenantID) > 0 {
		matchingLabels[constants.LabelTenantID] = tenantID
	}
	if len(projectID) > 0 {
		matchingLabels[constants.LabelProjectID] = projectID
	}
	if len(applicationID) > 0 {
		matchingLabels[constants.LabelApplicationID] = applicationID
	}
	if len(bizType) > 0 {
		matchingLabels[constants.LabelBizType] = bizType
	}

	err := p.rdscli.List(ctx, clusters,
		matchingLabels,
		client.Limit(cr.Limit),
		client.Continue(cr.ContinueToken),
	)
	if len(clusters.Items) > 0 {
		return clusters, WrapClientError(err)
	}
	// fall back to k8s cli
	return p.clusterReaderWriter.QueryClusters(ctx, tenantID, projectID, applicationID, bizType, cr)
}

func (p rdsclusterReaderWriter) GetCluster(ctx context.Context, id string) (*v1.Cluster, error) {
	clusters := new(v1.ClusterList)
	matchingLabels := client.MatchingLabels{
		constants.LabelClusterID: id,
	}

	err := p.rdscli.List(ctx, clusters,
		matchingLabels,
	)

	if err == nil && len(clusters.Items) > 0 {
		return &clusters.Items[0], nil
	}
	// fall back to k8s cli
	return p.clusterReaderWriter.GetCluster(ctx, id)
}

func (p rdsclusterReaderWriter) UpdateCluster(ctx context.Context, data *v1.Cluster) error {
	return WrapClientError(p.rdscli.Update(ctx, data))
}

func (p rdsclusterReaderWriter) DeleteCluster(ctx context.Context, id string) (err error) {
	clusters := new(v1.ClusterList)
	matchingLabels := client.MatchingLabels{
		constants.LabelClusterID: id,
	}

	err = p.rdscli.List(ctx, clusters,
		matchingLabels,
	)

	if err != nil {
		return WrapClientError(err)
	}
	if len(clusters.Items) == 0 {
		return errors.NewErrorf(errors.ClusterNotFound, "cluster %s is not found", id)
	}
	cluster := clusters.Items[0]
	return WrapClientError(p.rdscli.Delete(ctx, &v1.Cluster{
		ObjectMeta: metav1.ObjectMeta{
			Name:      id,
			Namespace: cluster.Namespace,
		},
	}))
}