package cluster

import (
	"bytes"
	"context"
	"crypto/tls"
	"encoding/json"
	"fmt"
	"net/http"
	"reflect"
	"strconv"
	"strings"
	"sync"
	"time"

	"github.com/go-logr/logr"
	"github.com/google/go-cmp/cmp"
	"github.com/pingcap/log"
	"github.com/pingcap/tidb-operator/pkg/apis/pingcap/v1alpha1"
	operatorversioned "github.com/pingcap/tidb-operator/pkg/client/clientset/versioned"
	"github.com/pkg/errors"
	"github.com/prometheus/client_golang/prometheus"
	configv1alpha1 "github.com/tidbcloud/config-service/api/v1alpha1"
	"github.com/tidbcloud/config-service/gardener"
	"github.com/tidbcloud/config-service/pkg/auth"
	"github.com/tidbcloud/config-service/pkg/central"
	"github.com/tidbcloud/config-service/pkg/clinic_event"
	"github.com/tidbcloud/config-service/pkg/metric"
	"github.com/tidbcloud/config-service/pkg/pdapi"
	"github.com/tidbcloud/config-service/pkg/utils"
	"github.com/tidbcloud/config-service/sdk/api"
	"go.uber.org/zap"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/client-go/kubernetes"
)

const (
	namespacePrefix     = "tidb"
	dbName              = "db"
	reasonImageUpdated  = "image updated"
	reasonConfigUpdated = "configuration updated"

	// config type
	PDConfigType      = "pd-config"
	ClusterConfigType = "cluster-config"
)

// Lister list TiDB Cloud clusters from meta store
type Lister interface {
	List(orgID string) ([]Cluster, error)
	ListAliveClusters() ([]Cluster, error)
}

type Cluster interface {
	// GetResource get the desired state and current state of the cluster
	GetResource() (*v1alpha1.TidbCluster, error)

	// GetMeta get the cluster metadata
	GetMeta() *ClusterMeta

	// Get provider type
	GetProviderType() string

	// Get instance type
	GetInstanceType(api.ComponentName) string

	// ID returns cluster id
	ID() string

	// UpdateResourceConfig update the desired state of the cluster config according to mutateFn
	// Note: if anything else the cluster config is altered, an error will be raised
	UpdateResourceConfig(tc *v1alpha1.TidbCluster, cc *configv1alpha1.ClusterConfig, mutateFn func() error) (*v1alpha1.TidbCluster, bool, error)

	// GetPDClusterConfig get PD cluster config
	GetPDClusterConfig() (*configv1alpha1.PDClusterConfig, error)

	// UpdatePDClusterConfig update the PD cluster config
	UpdatePDClusterConfig(configv1alpha1.PDClusterConfig, bool) (bool, error)

	// Send event
	SendEvent(oldTc, tc *v1alpha1.TidbCluster, configType string) error
}

// ClusterMeta is the metadata of the cluster
// NOTE: this schema is temporary and ought to be changed
type ClusterMeta struct {
	TenantID   string
	ClusterID  string
	Provider   string
	Region     string
	Components map[api.ComponentName]api.ComponentInput
	Status     string
}

func (c *ClusterMeta) MachineTypeOf(comp api.ComponentName) string {
	if v, ok := c.Components[comp]; ok {
		return v.InstanceType
	}
	return ""
}

type centralClusterLister struct {
	host          string
	auth          *auth.Auth0ClientAuth
	gardenCli     kubernetes.Interface
	gardenProject string
	log           logr.Logger
	pdAuth        pdapi.PDAuth

	centralSvc *central.Service

	shootCache sync.Map
}

type shootClientSet struct {
	operatorCli operatorversioned.Interface
	kubeCli     kubernetes.Interface
	pdProxy     *PDProxy
}

func NewLister(host string, auth *auth.Auth0ClientAuth, pdAuth pdapi.PDAuth, gardenCli kubernetes.Interface, gardenProject string, log logr.Logger) Lister {
	centralSvc := central.NewService(fmt.Sprintf("http://%s", host), auth.Domain, auth.ClientID, auth.ClientSecret, auth.Audience)
	return &centralClusterLister{
		host:          host,
		auth:          auth,
		gardenCli:     gardenCli,
		gardenProject: gardenProject,
		pdAuth:        pdAuth,
		log:           log,
		centralSvc:    centralSvc,
	}
}

var _ Lister = &centralClusterLister{}

func (c *centralClusterLister) List(orgID string) ([]Cluster, error) {
	cResp, err := c.centralSvc.ListTenantClusters(orgID)
	if err != nil {
		return nil, errors.Wrap(err, "error list clusters")
	}
	var cList []Cluster

	for _, cluster := range cResp.Items {
		prjID := fmt.Sprintf("%d", cluster.ProjectID)
		// special for dev environment
		if c.gardenProject == "nightly" {
			prjID = fmt.Sprintf("nightly-ms-%s", prjID)
		}
		shoot := gardener.ShootName(prjID, gardener.Cloud(cluster.Provider), cluster.Region)

		scs, err := c.loadShootClientSet(shoot, cluster.Provider)
		if err != nil {
			c.log.Error(err, "error loading client set for shoot",
				"shoot", shoot, "org", orgID, "cluster", cluster.ID)
			continue
		}
		cList = append(cList, &dbaasCluster{
			tenantID:      orgID,
			tenantCluster: cluster,
			operatorCli:   scs.operatorCli,
			pdCli:         scs.pdProxy.GetPDClient(cluster.ID),
		})
	}
	return cList, nil
}

func (c *centralClusterLister) ListAliveClusters() ([]Cluster, error) {
	cResp, err := c.centralSvc.ListAliveCluster()
	if err != nil {
		c.log.Error(err, "list alive cluster failed")
		return nil, err
	}
	var cList []Cluster

	for _, cluster := range cResp.Items {
		prjID := fmt.Sprintf("%d", cluster.ProjectID)
		// special for dev environment
		if c.gardenProject == "nightly" {
			prjID = fmt.Sprintf("nightly-ms-%s", prjID)
		}
		shoot := gardener.ShootName(prjID, gardener.Cloud(cluster.Provider), cluster.Region)

		scs, err := c.loadShootClientSet(shoot, cluster.Provider)
		if err != nil {
			c.log.Error(err, "error loading client set for shoot",
				"shoot", shoot, "cluster", cluster.ID)
			continue
		}
		cList = append(cList, &dbaasCluster{
			tenantID:      strconv.FormatInt(cluster.TenantID, 10),
			tenantCluster: *cluster.TenantCluster(),
			operatorCli:   scs.operatorCli,
			pdCli:         scs.pdProxy.GetPDClient(cluster.ID),
		})
	}
	return cList, nil
}

func (c *centralClusterLister) loadShootClientSet(shoot string, provider string) (*shootClientSet, error) {
	// TODO(zhengjiajin): check shoot apiserver ready
	v, ok := c.shootCache.Load(shoot)
	if !ok {
		scs, err := c.newShootClientSet(c.gardenCli, fmt.Sprintf("garden-%s", c.gardenProject), shoot, provider)
		if err != nil {
			return nil, errors.Wrapf(err, "new operator clientset, shoot %s", shoot)
		}
		c.shootCache.Store(shoot, scs)
		return scs, nil
	}
	return v.(*shootClientSet), nil
}

var _ Cluster = &dbaasCluster{}

type dbaasCluster struct {
	tenantID      string
	tenantCluster central.TenantCluster
	operatorCli   operatorversioned.Interface
	pdCli         pdapi.DynamicPDClient
}

func NewDBaaSCluster(tenantID string, tenantCluster central.TenantCluster, operatorCli operatorversioned.Interface, pdCli pdapi.DynamicPDClient) Cluster {
	return &dbaasCluster{
		tenantID:      tenantID,
		tenantCluster: tenantCluster,
		operatorCli:   operatorCli,
		pdCli:         pdCli,
	}
}

func (c *dbaasCluster) GetResource() (*v1alpha1.TidbCluster, error) {
	namespace := fmt.Sprintf("%s%d", namespacePrefix, c.tenantCluster.ID)
	return c.operatorCli.PingcapV1alpha1().TidbClusters(namespace).Get(context.Background(), dbName, metav1.GetOptions{})
}

func (c *dbaasCluster) GetProviderType() string {
	return c.tenantCluster.Provider
}

func (c *dbaasCluster) GetInstanceType(compName api.ComponentName) string {
	switch compName {
	case "pd":
		return c.tenantCluster.TenantComponents.PD.MachineType
	case "tidb":
		return c.tenantCluster.TenantComponents.TiDB.MachineType
	case "tiproxy":
		return c.tenantCluster.TenantComponents.TiProxy.MachineType
	case "tikv":
		return c.tenantCluster.TenantComponents.TiKV.MachineType
	case "tiflash":
		return c.tenantCluster.TenantComponents.TiFlash.MachineType
	default:
		return ""
	}
}

func (c *dbaasCluster) GetMeta() *ClusterMeta {
	topo := c.tenantCluster.TenantComponents
	comps := map[api.ComponentName]api.ComponentInput{
		api.ComponentPD: {
			Component:    api.ComponentPD,
			InstanceType: fmt.Sprintf("%s.%s", c.tenantCluster.Provider, topo.PD.MachineType),
		},
		api.ComponentTiDB: {
			Component:    api.ComponentTiDB,
			InstanceType: fmt.Sprintf("%s.%s", c.tenantCluster.Provider, topo.TiDB.MachineType),
		},
		api.ComponentTiKV: {
			Component:    api.ComponentTiKV,
			InstanceType: fmt.Sprintf("%s.%s", c.tenantCluster.Provider, topo.TiKV.MachineType),
		},
	}
	if topo.TiProxy.Replicas > 0 {
		comps[api.ComponentTiProxy] = api.ComponentInput{
			Component:    api.ComponentTiProxy,
			InstanceType: fmt.Sprintf("%s.%s", c.tenantCluster.Provider, topo.TiProxy.MachineType),
		}
	}
	if topo.TiFlash.Replicas > 0 {
		comps[api.ComponentTiFlash] = api.ComponentInput{
			Component:    api.ComponentTiFlash,
			InstanceType: fmt.Sprintf("%s.%s", c.tenantCluster.Provider, topo.TiFlash.MachineType),
		}
		comps[api.ComponentTiFlashLearner] = api.ComponentInput{
			Component:    api.ComponentTiFlashLearner,
			InstanceType: fmt.Sprintf("%s.%s", c.tenantCluster.Provider, topo.TiFlash.MachineType),
		}
	}
	return &ClusterMeta{
		TenantID:   c.tenantID,
		ClusterID:  strconv.FormatInt(c.tenantCluster.ID, 10),
		Provider:   c.tenantCluster.Provider,
		Region:     c.tenantCluster.Region,
		Components: comps,
		Status:     c.tenantCluster.Status,
	}
}

func (c *dbaasCluster) ID() string {
	return strconv.FormatInt(c.tenantCluster.ID, 10)
}

func (c *dbaasCluster) UpdateResourceConfig(tc *v1alpha1.TidbCluster, cc *configv1alpha1.ClusterConfig, mutateFn func() error) (*v1alpha1.TidbCluster, bool, error) {
	log.Info("UpdateResourceConfig", zap.String("cluster", tc.Name), zap.String("id", c.ID()), zap.Bool("dryrun", cc.IsDryrun()), zap.Any("meta", c.GetMeta()))
	old := tc.DeepCopy()
	err := mutateFn()
	if err != nil {
		return nil, false, err
	}
	// no update
	if reflect.DeepEqual(old, tc) {
		return tc, false, nil
	}

	logPlan(old.Spec, tc.Spec, cc.IsDryrun(), fmt.Sprintf("TidbCluster resource of cluster %s", c.ID()))
	analysisPlanAndRecordMetrics(old, tc, cc.IsDryrun(), c.ID())
	namespace := fmt.Sprintf("%s%d", namespacePrefix, c.tenantCluster.ID)
	if !cc.IsDryrun() {

		// Modify the ops.tidbcloud.com label only when we need to update tc
		if tc.Annotations == nil {
			tc.Annotations = map[string]string{}
		}
		tc.Annotations[api.ConfigIDKey] = cc.Name
		tc.Annotations[api.ConfigVersionKey] = strconv.FormatUint(cc.Spec.ConfigMeta.ConfigVersion, 10)
		tc.Annotations[api.ConfigCommitKey] = cc.Spec.ConfigMeta.GitCommitID
		// since we perform comprehensive pre-check, simple back-off to next reconciliation chance on any conflicts
		tc, err = c.operatorCli.PingcapV1alpha1().TidbClusters(namespace).Update(context.Background(), tc, metav1.UpdateOptions{})
		return tc, true, err
	}
	return tc, false, nil
}

func (c *dbaasCluster) GetPDClusterConfig() (*configv1alpha1.PDClusterConfig, error) {
	conf, err := c.pdCli.GetConfig()
	if err != nil {
		return nil, err
	}
	return configv1alpha1.ExtractPDClusterConfig(conf), nil
}

func (c *dbaasCluster) UpdatePDClusterConfig(conf configv1alpha1.PDClusterConfig, dryrun bool) (bool, error) {
	pcc, err := c.GetPDClusterConfig()
	if err != nil {
		// cannot get pd config through https proxy, ignore now
		log.Warn("error get pd config", zap.Error(err))
		return false, nil
	}
	current := pcc.MP
	updated := utils.MergeMaps(current, conf.MP)
	// no update
	if reflect.DeepEqual(current, updated) {
		return false, nil
	}
	logPlan(current, updated, dryrun, fmt.Sprintf("PD cluster config of cluster %s", c.ID()))
	if !dryrun {
		err = c.pdCli.PutConfig(conf.MP)
		return true, err
	}
	return false, nil
}

func (c *dbaasCluster) SendEvent(oldTc, tc *v1alpha1.TidbCluster, configType string) error {
	eventRequest := clinic_event.EventRequest{
		ApplicationID:   c.tenantID,
		TargetID:        c.ID(),
		Payload:         make(map[string]interface{}),
		Context:         []string{},
		CalibrationTime: uint64(time.Now().Unix()),
		Catalog:         clinic_event.GitopsCatalog,
		//ContextID:     l.genClinicEventID(),
	}

	// // gen payload
	payloadByte, err := json.Marshal(tc.Spec)
	if err != nil {
		return err
	}
	payloadDecoder := json.NewDecoder(bytes.NewBuffer(payloadByte))
	payloadDecoder.UseNumber()
	_ = payloadDecoder.Decode(&eventRequest.Payload)

	// set contextID
	// Divide every 20 minutes into a time period, with the same event context id in the same time period
	nowTs := time.Now().Unix()
	eventRequest.ContextID = fmt.Sprintf("%s-%s-%s-%d-v1", clinic_event.GitopsCatalog, c.tenantID, c.ID(), nowTs-nowTs%(20*60))

	// add base/tenants contextid
	eventRequest.Context = genEventContexts(c.tenantID, nowTs-nowTs%(20*60))

	// diff
	eventRequest.Payload["diff"] = cmp.Diff(oldTc, tc)

	switch configType {
	case PDConfigType:
		eventRequest.SeverityLevel = int(clinic_event.Debug)
		eventRequest.Name = fmt.Sprintf("%s:%s", clinic_event.ConfigChangeCatalog, PDConfigType)
		eventRequest.Payload["summary"] = "update pd config"
	case ClusterConfigType:
		eventRequest.SeverityLevel = int(clinic_event.Info)
		if isUpgrade(oldTc, tc) {
			eventRequest.Name = fmt.Sprintf("%s:%s", clinic_event.UpgradeCatalog, clinic_event.ApplyStatus)
			eventRequest.Payload["summary"] = fmt.Sprintf("upgrade tidb version to %s", tc.Spec.Version)
		} else if components, ok := isPatch(oldTc, tc); ok {
			eventRequest.Name = fmt.Sprintf("%s:%s", clinic_event.PatchCatalog, clinic_event.ApplyStatus)
			eventRequest.Payload["summary"] = fmt.Sprintf("patch components [%s] ", strings.Join(components, ", "))
		} else if components, ok := isConfigChange(oldTc, tc); ok {
			eventRequest.Name = fmt.Sprintf("%s:%s", clinic_event.ConfigChangeCatalog, clinic_event.ApplyStatus)
			eventRequest.Payload["summary"] = fmt.Sprintf("update [%s] configuration", strings.Join(components, ", "))
		} else {
			return nil
		}
	}
	log.Info("Sent Event", zap.String("event name", eventRequest.Name), zap.String("cluster-id", c.ID()), zap.String("event-type", configType))
	return clinic_event.SendEvent(eventRequest)
}

func isUpgrade(oldTc, tc *v1alpha1.TidbCluster) bool {
	return tc.Spec.Version != oldTc.Spec.Version
}

func isPatch(oldTc, tc *v1alpha1.TidbCluster) ([]string, bool) {

	pacth := false
	components := []string{}

	// upgrade
	if tc.Spec.Version != oldTc.Spec.Version {
		return components, pacth
	}

	// tidb
	if (oldTc.Spec.TiDB != nil && tc.Spec.TiDB != nil) && (*tc.Spec.TiDB.Version != *oldTc.Spec.TiDB.Version) {
		pacth = true
		components = append(components, string(api.ComponentTiDB))
	}

	// tikv
	if (oldTc.Spec.TiKV != nil && tc.Spec.TiKV != nil) && (*tc.Spec.TiKV.Version != *oldTc.Spec.TiKV.Version) {
		pacth = true
		components = append(components, string(api.ComponentTiKV))
	}

	// pd
	if (oldTc.Spec.PD != nil && tc.Spec.PD != nil) && (*tc.Spec.PD.Version != *oldTc.Spec.PD.Version) {
		pacth = true
		components = append(components, string(api.ComponentPD))
	}

	// tiflash
	if (oldTc.Spec.TiFlash != nil && tc.Spec.TiFlash != nil) && (*tc.Spec.TiFlash.Version != *oldTc.Spec.TiFlash.Version) {
		pacth = true
		components = append(components, string(api.ComponentTiFlash))
	}

	return components, pacth

}

func isConfigChange(oldTc, tc *v1alpha1.TidbCluster) ([]string, bool) {

	configChange := false
	components := []string{}

	if oldTc.Spec.PD != nil && tc.Spec.PD != nil && !reflect.DeepEqual(oldTc.Spec.PD.Config, tc.Spec.PD.Config) {
		configChange = true
		components = append(components, string(api.ComponentPD))
	}

	if oldTc.Spec.TiDB != nil && tc.Spec.TiDB != nil && !reflect.DeepEqual(oldTc.Spec.TiDB.Config, tc.Spec.TiDB.Config) {
		configChange = true
		components = append(components, string(api.ComponentTiDB))
	}

	if oldTc.Spec.TiKV != nil && tc.Spec.TiKV != nil && !reflect.DeepEqual(oldTc.Spec.TiKV.Config, tc.Spec.TiKV.Config) {
		configChange = true
		components = append(components, string(api.ComponentTiKV))
	}

	if oldTc.Spec.TiFlash != nil && tc.Spec.TiFlash != nil && !reflect.DeepEqual(oldTc.Spec.TiFlash.Config, tc.Spec.TiFlash.Config) {
		configChange = true
		components = append(components, string(api.ComponentTiFlash))
	}

	return components, configChange
}

func genEventContexts(tenantid string, ts int64) []string {
	contexts := []string{
		fmt.Sprintf("%s-0-0-%d-v1", clinic_event.GitopsCatalog, ts),
		fmt.Sprintf("%s-%s-0-%d-v1", clinic_event.GitopsCatalog, tenantid, ts),
	}
	return contexts
}

func logPlan(before, after interface{}, dryrun bool, target string) {
	mode := "execute"
	if dryrun {
		mode = "dryrun"
	}
	d := cmp.Diff(before, after)
	if d != "" {
		msg := fmt.Sprintf("[%s] update %s, diff: %s", mode, target, strings.ReplaceAll(d, "\n", "\r"))
		log.Info(msg)
	}

}

// analysisPlanAndRecordMetrics will record the affected components for dryrun clusters via metric.
// only consider image and config changes here,
// other changes that might result in component restart shall not be introduced by gitops.
func analysisPlanAndRecordMetrics(old, new *v1alpha1.TidbCluster, dryrun bool, clusterId string) {
	if !dryrun {
		// clear all the metrics related to this clusterId
		metric.ConfigDryrunAffected.DeletePartialMatch(prometheus.Labels{"cluster_id": clusterId})
		return
	}

	record := func(clusterId string, comp string, reason string, val float64) {
		// avoid multiple reason present at the same time.
		metric.ConfigDryrunAffected.DeletePartialMatch(prometheus.Labels{"cluster_id": clusterId, "component": comp})
		metric.ConfigDryrunAffected.WithLabelValues(clusterId, comp, reason).Set(val)
	}

	if old.PDImage() != new.PDImage() {
		record(clusterId, v1alpha1.PDMemberType.String(), reasonImageUpdated, 1)
	} else if old.Spec.PD != nil && new.Spec.PD != nil && !reflect.DeepEqual(old.Spec.PD.Config, new.Spec.PD.Config) {
		record(clusterId, v1alpha1.PDMemberType.String(), reasonConfigUpdated, 1)
	} else {
		record(clusterId, v1alpha1.PDMemberType.String(), "", 0)
	}

	if old.TiDBImage() != new.TiDBImage() {
		record(clusterId, v1alpha1.TiDBMemberType.String(), reasonImageUpdated, 1)
	} else if old.Spec.TiDB != nil && new.Spec.TiFlash != nil && !reflect.DeepEqual(old.Spec.TiDB.Config, new.Spec.TiDB.Config) {
		record(clusterId, v1alpha1.TiDBMemberType.String(), reasonConfigUpdated, 1)
	} else {
		record(clusterId, v1alpha1.TiDBMemberType.String(), "", 0)
	}

	if old.TiKVImage() != new.TiKVImage() {
		record(clusterId, v1alpha1.TiKVMemberType.String(), reasonImageUpdated, 1)
	} else if old.Spec.TiKV != nil && new.Spec.TiKV != nil && !reflect.DeepEqual(old.Spec.TiKV.Config, new.Spec.TiKV.Config) {
		record(clusterId, v1alpha1.TiKVMemberType.String(), reasonConfigUpdated, 1)
	} else {
		record(clusterId, v1alpha1.TiKVMemberType.String(), "", 0)
	}

	if old.TiFlashImage() != new.TiFlashImage() {
		record(clusterId, v1alpha1.TiFlashMemberType.String(), reasonImageUpdated, 1)
	} else if old.Spec.TiFlash != nil && new.Spec.TiFlash != nil && !reflect.DeepEqual(old.Spec.TiFlash.Config, new.Spec.TiFlash.Config) {
		record(clusterId, v1alpha1.TiFlashMemberType.String(), reasonConfigUpdated, 1)
	} else {
		record(clusterId, v1alpha1.TiFlashMemberType.String(), "", 0)
	}
}

// newShootClientSet new operator client-set in shoot cluster
func (c *centralClusterLister) newShootClientSet(gardenCli kubernetes.Interface, namespace string, shootName string, provider string) (*shootClientSet, error) {
	secretName := fmt.Sprintf("%s.kubeconfig", shootName)
	secret, err := gardenCli.CoreV1().Secrets(namespace).Get(context.Background(), secretName, metav1.GetOptions{})
	if err != nil {
		return nil, err
	}

	restConfig, err := gardener.RestConfigFromSecret(secret)
	if err != nil {
		return nil, err
	}

	kubeCli, err := kubernetes.NewForConfig(restConfig)
	if err != nil {
		return nil, err
	}
	opCli, err := operatorversioned.NewForConfig(restConfig)
	if err != nil {
		return nil, err
	}
	svc, err := kubeCli.CoreV1().Services("tidb-admin").Get(context.Background(), "dbaas-proxy", metav1.GetOptions{})
	if err != nil {
		return nil, err
	}
	ings := svc.Status.LoadBalancer.Ingress
	if len(ings) < 1 {
		return nil, errors.Errorf("Proxy endpoint of shoot is not ready")
	}
	var proxyEp string
	switch provider {
	case "gcp":
		proxyEp = ings[0].IP
	case "aws":
		proxyEp = ings[0].Hostname
	default:
		return nil, fmt.Errorf("unsupport provider %v", provider)
	}
	pdProxy := &PDProxy{
		host: proxyEp,
		client: &http.Client{
			Timeout: 10 * time.Second,
			Transport: &http.Transport{
				TLSClientConfig: &tls.Config{
					// TODO(lorqor): Configure the TLS config once we've fully migrated to infra API
					InsecureSkipVerify: true,
				},
			},
		},
		auth: c.pdAuth,
	}
	return &shootClientSet{
		operatorCli: opCli,
		kubeCli:     kubeCli,
		pdProxy:     pdProxy,
	}, nil
}