package scrape

import (
	"encoding/json"
	"fmt"
	"strconv"
	"time"

	"github.com/pingcap/dbaas/pkg/constants"
	"github.com/pingcap/kvproto/pkg/metapb"
	"github.com/pingcap/log"
	tidbClusterV1alpha1 "github.com/pingcap/tidb-operator/pkg/apis/pingcap/v1alpha1"
	tidbClusterInformers "github.com/pingcap/tidb-operator/pkg/client/informers/externalversions"
	tidbClusterV1alpha1listers "github.com/pingcap/tidb-operator/pkg/client/listers/pingcap/v1alpha1"
	"github.com/pingcap/tidb-operator/pkg/label"
	"github.com/pingcap/tidb-operator/pkg/pdapi"
	"go.uber.org/multierr"
	"go.uber.org/zap"
	corev1 "k8s.io/api/core/v1"
	"k8s.io/apimachinery/pkg/labels"
	kubeinformers "k8s.io/client-go/informers"
	"k8s.io/client-go/kubernetes"
	corev1list "k8s.io/client-go/listers/core/v1"

	"github.com/tidbcloud/billing-reporter/metrics"
	"github.com/tidbcloud/billing-reporter/report"
	"github.com/tidbcloud/billing-reporter/utils"
)

type ScrapeManager struct {
	kubeCli    kubernetes.Interface
	podLister  corev1list.PodLister
	nodeLister corev1list.NodeLister
	tcLister   tidbClusterV1alpha1listers.TidbClusterLister
	pdControl  pdapi.PDControlInterface
	sqsManager *report.SqsManager
}

var (
	errorBasicInfo = BasicInfo{Heartbeat: false}
)

func NewScrapManager(
	kubeCli kubernetes.Interface,
	kubeFactory kubeinformers.SharedInformerFactory,
	informerFactory tidbClusterInformers.SharedInformerFactory,
	sqsManager *report.SqsManager) *ScrapeManager {
	return &ScrapeManager{
		kubeCli:    kubeCli,
		sqsManager: sqsManager,
		podLister:  kubeFactory.Core().V1().Pods().Lister(),
		nodeLister: kubeFactory.Core().V1().Nodes().Lister(),
		pdControl:  pdapi.NewDefaultPDControl(kubeCli),
		tcLister:   informerFactory.Pingcap().V1alpha1().TidbClusters().Lister(),
	}
}

func (sm *ScrapeManager) List() (ret []*tidbClusterV1alpha1.TidbCluster, err error) {
	return sm.tcLister.List(labels.Everything())
}

func (sm *ScrapeManager) SendSqsMessage(sqsMsg report.SqsMessage) error {
	return sm.sqsManager.SendSqsMessage(sqsMsg)
}

func (sm *ScrapeManager) ScrapeTidbClusterListInfo() error {
	metrics.ScrapeStatus.WithLabelValues("total").Inc()
	tclist, err := sm.List()
	if err != nil {
		metrics.ScrapeStatus.WithLabelValues("list_failed").Inc()
		return err
	}
	var infos []*TidbClusterHeartbeatInfo
	var errs []error
	metrics.ClusterStatusCounter.WithLabelValues("total").Add(float64(len(tclist)))
	for _, tc := range tclist {
		info, e := sm.ScrapeTidbClusterInfo(tc)
		if e != nil {
			errs = append(errs, e)
			log.Error("failed to scrape tidbcluster", zap.String("namespace", tc.Namespace), zap.String("name", tc.Name), zap.Error(e))
			metrics.ScrapeStatus.WithLabelValues("scrape_failed").Inc()
			info.Msg = e.Error()
		} else {
			metrics.ClusterStatusCounter.WithLabelValues(info.Status).Inc()
		}
		if info != nil {
			log.Info("scraped cluster heartbeat", zap.Uint64("tenant", info.TenantId), zap.Uint64("cluster", info.ClusterId), zap.Uint32("profile", info.ClusterProfileId), zap.Int("tidb", len(info.Tidbs)), zap.Int("tikv", len(info.Tikvs)), zap.Int("tiflash", len(info.Tiflashes)), zap.String("status", info.Status))
			infos = append(infos, info)
		}
	}

	// There is no tidbcluster in kubernetes here
	if infos == nil || len(infos) < 1 {
		return nil
	}
	msg, err := BuildSqsMessage(infos)
	if err != nil {
		return err
	}
	if err := sm.SendSqsMessage(msg); err != nil {
		errs = append(errs, err)
	}
	if len(errs) < 1 {
		return nil
	}
	return multierr.Combine(errs...)
}

func (sm *ScrapeManager) ScrapeTidbClusterInfo(tc *tidbClusterV1alpha1.TidbCluster) (*TidbClusterHeartbeatInfo, error) {
	info, err := initTidbClusterHeartbeatInfo(tc)
	if err != nil {
		return info, err
	}

	isCreating, err := utils.IsClusterInCreatingStatus(sm.kubeCli, tc)
	if err != nil {
		log.Error("Get cluster creating status failed", zap.Error(err), zap.String("namespace", tc.Namespace))
	}
	if isCreating {
		info.Status = string(constants.TidbClusterCreating)
		return info, nil
	}

	// scape tikv info, if error happens during scraping tikvList, we would set the status as unknown and return the info
	tikvInfoList, err := sm.scrapeTikvListInfo(tc)
	if err != nil {
		info.Status = string(constants.TidbClusterUnknown)
		return info, err
	}
	info.Tikvs = tikvInfoList

	tiflashInfoList, err := sm.scrapeTiflashListInfo(tc)
	if err != nil {
		info.Status = string(constants.TidbClusterUnknown)
		return info, err
	}
	info.Tiflashes = tiflashInfoList

	// scrape tidb info, if error happens during scraping tidbList, we would set the status as unknown and return the info
	tidbInfoList, err := sm.scrapTidbListInfo(tc)
	if err != nil {
		info.Status = string(constants.TidbClusterUnknown)
		return info, err
	}
	info.Tidbs = tidbInfoList
	info.Status = checkTidbClusterStatus(info, tc)
	info.Msg = ""
	return info, nil
}

func initTidbClusterHeartbeatInfo(tc *tidbClusterV1alpha1.TidbCluster) (*TidbClusterHeartbeatInfo, error) {
	// todo: use tc labels
	tidbReplica, tikvReplica, tiflashReplica := getComponentReplicas(tc)

	info := &TidbClusterHeartbeatInfo{
		Name:      tc.Name,
		Namespace: tc.Namespace,
		Timestamp: time.Now().Unix(),
		Tidbs:     make([]TidbHeartbeatInfo, tidbReplica),
		Tikvs:     make([]TikvHeartbeatInfo, tikvReplica),
		Tiflashes: make([]TiflashHeartbeatInfo, tiflashReplica),
	}
	// add tenantID
	tenantID, err := strconv.ParseUint(tc.Labels[constants.TiDBClusterTenantIDLabel], 10, 64)
	if err != nil {
		info.Status = string(constants.TidbClusterUnknown)
		return info, err
	}
	info.TenantId = tenantID
	// add clusterID
	clusterID, err := strconv.ParseUint(tc.Labels[constants.TiDBClusterClusterIDLabel], 10, 64)
	if err != nil {
		info.Status = string(constants.TidbClusterUnknown)
		return info, err
	}
	info.ClusterId = clusterID

	if config, found := tc.Annotations[TiDBClusterConfig]; found {
		var h HomogenousClusterConfig
		err = json.Unmarshal([]byte(config), &h)
		if err != nil {
			info.Status = string(constants.TidbClusterUnknown)
			return info, err
		}
		info.ClusterConfig = HeterogeneousClusterConfig{
			TiDB:    buildHeterogeneousComponents(tidbReplica, h.TiDB),
			TiKV:    buildHeterogeneousComponents(tikvReplica, h.TiKV),
			TiFlash: buildHeterogeneousComponents(tiflashReplica, h.TiFlash),
		}
	} else {
		// add clusterProfileID
		clusterProfileID, err := convertUInt32(tc.Labels[constants.TiDBClusterClusterProfileIDLabel])
		if err != nil {
			info.Status = string(constants.TidbClusterUnknown)
			return info, err
		}
		info.ClusterProfileId = clusterProfileID
	}
	return info, nil
}

func buildHeterogeneousComponents(realReplicas int32, hc HomogenousComponent) []Component {
	var cpms []Component
	for i := 0; i < int(realReplicas); i++ {
		cpms = append(cpms, Component{
			Provider:     hc.Provider,
			Region:       hc.Region,
			InstanceType: hc.InstanceType,
			Disks:        hc.Disks,
		})
	}
	return cpms
}

// If the majority of the tikv's store state is up and at least 1 tidb is running,
// we would set the tidbcluster as available, otherwise it would be unavailable
func checkTidbClusterStatus(tcHeartbeatInfo *TidbClusterHeartbeatInfo, tc *tidbClusterV1alpha1.TidbCluster) string {
	tidbReplica, tikvReplica, _ := getComponentReplicas(tc)
	tikvHealthCount := 0
	tidbHealthCount := 0
	tikvHealth := false
	for _, tikv := range tcHeartbeatInfo.Tikvs {
		if tikv.Heartbeat && tikv.Store.State == metapb.StoreState_Up.String() {
			tikvHealthCount++
		}
	}
	if int32(tikvHealthCount) > (tikvReplica / 2) {
		tikvHealth = true
	}

	for _, tidb := range tcHeartbeatInfo.Tidbs {
		if tidb.Heartbeat && tidb.Podphase == string(corev1.PodRunning) {
			tidbHealthCount++
		}
	}
	// If the tikv store's count with Up State is smaller than the majority of the tikv desired replicas
	// or the none of tidb is health, we would think this tidbcluster is unavailable
	if !tikvHealth || tidbHealthCount < 1 {
		return string(constants.TidbClusterUnavailable)
	}

	// If there exists at least one tidb or tikv instances is unhealthy while the tidbcluster still could work,
	// we think this tidbcluster is unHealthy
	if int32(tikvHealthCount) < tikvReplica || int32(tidbHealthCount) < tidbReplica {
		return string(constants.TidbClusterUnhealthy)
	}

	// If the pd/tikv/tidb 's revision is not equal to the latest revision, we think this tidbcluster is upgrading
	if tc.Status.PD.StatefulSet.CurrentRevision != tc.Status.PD.StatefulSet.UpdateRevision ||
		tc.Status.TiKV.StatefulSet.CurrentRevision != tc.Status.TiKV.StatefulSet.UpdateRevision ||
		tc.Status.TiDB.StatefulSet.CurrentRevision != tc.Status.TiDB.StatefulSet.UpdateRevision {
		return string(constants.TidbClusterUpgrading)
	}

	// If the replicas info between tikv/tidb statefulset spec and status is not the same,
	// we think this tidbcluster is scaling
	if tc.Status.TiKV.StatefulSet.Replicas != tikvReplica ||
		tc.Status.TiDB.StatefulSet.Replicas != tidbReplica {
		return string(constants.TidbClusterScaling)
	}

	return string(constants.TidbClusterNormal)
}

func (sm *ScrapeManager) scrapeTikvListInfo(tc *tidbClusterV1alpha1.TidbCluster) ([]TikvHeartbeatInfo, error) {
	l := label.New().TiKV().Instance(tc.Name)
	selector, err := l.Selector()
	if err != nil {
		return nil, err
	}
	pods, err := sm.podLister.Pods(tc.Namespace).List(selector)
	if err != nil {
		return nil, err
	}

	pdClient := sm.pdControl.GetPDClient(pdapi.Namespace(tc.Namespace), tc.Name, tc.IsTLSClusterEnabled())
	stores, err := pdClient.GetStores()
	if err != nil {
		return nil, err
	}
	podStoreMap := map[string]*pdapi.StoreInfo{}
	for _, store := range stores.Stores {
		podName := getPodNameFromStore(store)
		podStoreMap[podName] = store
	}

	var infoList []TikvHeartbeatInfo
	var errs []error
	for _, pod := range pods {
		info, err := sm.scrapeTikvInfo(pod, podStoreMap)
		if err != nil {
			errs = append(errs, err)
		}
		infoList = append(infoList, info)
	}
	if len(errs) < 1 {
		return infoList, nil
	}
	return infoList, multierr.Combine(errs...)
}

func (sm *ScrapeManager) scrapeTiflashListInfo(tc *tidbClusterV1alpha1.TidbCluster) ([]TiflashHeartbeatInfo, error) {
	l := label.New().TiFlash().Instance(tc.Name)
	selector, err := l.Selector()
	if err != nil {
		return nil, err
	}
	pods, err := sm.podLister.Pods(tc.Namespace).List(selector)
	if err != nil {
		return nil, err
	}

	var infoList []TiflashHeartbeatInfo
	var errs []error
	for _, pod := range pods {
		info, err := sm.buildBasicInfo(pod)
		if err != nil {
			errs = append(errs, err)
		}
		infoList = append(infoList, TiflashHeartbeatInfo{
			BasicInfo: info,
			Tiflash:   TiflashInfo{Health: true},
		})
	}
	if len(errs) < 1 {
		return infoList, nil
	}
	return infoList, multierr.Combine(errs...)
}

func (sm *ScrapeManager) scrapeTikvInfo(pod *corev1.Pod, podStoreMap map[string]*pdapi.StoreInfo) (TikvHeartbeatInfo, error) {
	info := TikvHeartbeatInfo{
		BasicInfo: BasicInfo{
			Name:      pod.Name,
			Heartbeat: false,
		},
	}
	podInfo, err := sm.buildBasicInfo(pod)
	if err != nil {
		return info, err
	}
	info.BasicInfo = podInfo
	store, existed := podStoreMap[pod.Name]
	if !existed {
		return info, err
	}
	info.Heartbeat = true
	info.Store = StoreInfo{
		StoreId: store.Store.Id,
		State:   store.Store.State.String(),
		Labels:  getLabelsFromStore(store),
	}
	return info, nil
}

func (sm *ScrapeManager) scrapTidbListInfo(tc *tidbClusterV1alpha1.TidbCluster) ([]TidbHeartbeatInfo, error) {
	l := label.New().TiDB().Instance(tc.Name)
	selector, err := l.Selector()
	if err != nil {
		return nil, err
	}
	pods, err := sm.podLister.Pods(tc.Namespace).List(selector)
	if err != nil {
		return nil, err
	}

	var infoList []TidbHeartbeatInfo
	var errs []error
	for _, pod := range pods {
		info, err := sm.buildTidbInfo(tc, pod)
		if err != nil {
			errs = append(errs, err)
		}
		infoList = append(infoList, info)
	}
	if len(errs) < 1 {
		return infoList, nil
	}
	return infoList, multierr.Combine(errs...)
}

func (sm *ScrapeManager) buildTidbInfo(tc *tidbClusterV1alpha1.TidbCluster, pod *corev1.Pod) (TidbHeartbeatInfo, error) {
	info := TidbHeartbeatInfo{
		BasicInfo: BasicInfo{
			Name:      pod.Name,
			Heartbeat: false,
		},
	}
	podInfo, err := sm.buildBasicInfo(pod)
	if err != nil {
		return info, err
	}
	info.BasicInfo = podInfo
	// TODO: use direct HTTP/HTTPS visit after fix the tls error in multi tidbclusters env
	member, ok := tc.Status.TiDB.Members[pod.Name]
	if !ok {
		info.Tidb.Health = false
	} else {
		info.Tidb.Health = member.Health
	}
	return info, nil
}

func (sm *ScrapeManager) buildBasicInfo(pod *corev1.Pod) (BasicInfo, error) {
	nodeInfo, err := sm.scrapeNodeInfo(pod.Spec.NodeName)
	if err != nil {
		return errorBasicInfo, err
	}
	return BasicInfo{
		Name:      pod.Name,
		Heartbeat: true,
		Podphase:  string(pod.Status.Phase),
		NodeInfo:  nodeInfo,
	}, nil
}

func (sm *ScrapeManager) scrapeNodeInfo(nodeName string) (NodeInfo, error) {
	node, err := sm.nodeLister.Get(nodeName)
	if err != nil {
		return NodeInfo{}, err
	}
	return NodeInfo{
		Name:   nodeName,
		Labels: node.Labels,
	}, nil
}

// ConvertUInt32 turn a string into a uint32
func convertUInt32(str string) (uint32, error) {
	if str != "" {
		i, err := strconv.ParseUint(str, 10, 64)
		if err != nil {
			return 0, err
		}
		return uint32(i), nil
	}
	return 0, fmt.Errorf("parmater is empty")
}