package cluster

import (
	"context"
	"errors"
	"fmt"
	"time"

	"github.com/pingcap/tidb-operator/pkg/apis/pingcap/v1alpha1"
	cloudv1 "github.com/tidbcloud/infra-api/api/v1"
	"github.com/tidbcloud/infra-api/pkg/pingcap/label"
	"github.com/tidbcloud/shared-provider/pkg/utils"
	"go.uber.org/multierr"
	corev1 "k8s.io/api/core/v1"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"sigs.k8s.io/controller-runtime/pkg/client"
)

const updateLastActiveTimeout = time.Second * 15

func (r *reconcileContext) observeCluster() error {
	var errs error
	if err := r.observeTidbClusterStatus(); err != nil {
		errs = multierr.Append(errs, err)
	}
	if err := r.observeMetricsStatus(); err != nil {
		errs = multierr.Append(errs, err)
	}
	if err := r.observeProxyStatus(); err != nil {
		errs = multierr.Append(errs, err)
	}
	if err := r.observeObjectStoreStatus(); err != nil {
		errs = multierr.Append(errs, err)
	}

	if r.tc != nil {
		nextCheck, err := r.bandwidthHandler.Sync(r.ctx, r.cluster, r.tc, r.querier)
		if err != nil {
			errs = multierr.Append(errs, err)
		} else {
			errs = multierr.Append(errs, &utils.RequeueAfterError{RequeueAfter: nextCheck})
		}
	}

	ctx, cancel := context.WithTimeout(r.ctx, updateLastActiveTimeout)
	defer cancel()
	nextCheck, err := r.updateLastActive(ctx)
	if err != nil {
		errs = multierr.Append(errs, err)
	} else {
		errs = multierr.Append(errs, &utils.RequeueAfterError{RequeueAfter: nextCheck})
	}

	if r.EvictDuration > 0 {
		if err := r.handleSafeToEvictAnno(); err != nil {
			errs = multierr.Append(errs, err)
		}
	}

	return errs
}

func (r *reconcileContext) observeObjectStoreStatus() error {
	return r.ensureStatus(func() error {
		r.cluster.Status.ObjectStore = &cloudv1.ObjectStoreStatus{
			S3: &cloudv1.Path{
				Path: r.backupBucket,
			},
		}
		return nil
	})
}

func (r *reconcileContext) observeProxyStatus() error {
	clusterProxyBaseUrl := fmt.Sprintf(
		"%s/proxy/namespace/%s/cluster/%s", r.BaseProxy, r.tcKey.Namespace, r.tcKey.Name)

	return r.ensureStatus(func() error {
		r.cluster.Status.Proxy = &cloudv1.ProxyStatus{
			Endpoint: clusterProxyBaseUrl,
		}
		return nil
	})
}

func (r *reconcileContext) observeMetricsStatus() error {
	if !r.haveTm() {
		return errors.New("TidbMonitor not exist")
	}

	metricsBaseUrl := fmt.Sprintf(
		"%s/proxy/namespace/%s/tidbmonitor/%s/prometheus", r.BaseProxy, r.tmKey.Namespace, r.tmKey.Name)

	return r.ensureStatus(func() error {
		r.cluster.Status.Metrics = &cloudv1.MetricsStatus{
			BaseURL: metricsBaseUrl,
		}
		return nil
	})
}

// nolint: gocognit,cyclop // nolint
func (r *reconcileContext) observeTidbClusterStatus() (errs error) {
	tidb, err := r.computeTiDBStatus()
	if err != nil {
		errs = multierr.Append(errs, err)
	}
	pd, err := r.computePDStatus()
	if err != nil {
		errs = multierr.Append(errs, err)
	}
	tikv, err := r.computeTiKVStatus()
	if err != nil {
		errs = multierr.Append(errs, err)
	}
	tiflash, err := r.computeTiFlashStatus()
	if err != nil {
		errs = multierr.Append(errs, err)
	}
	if errs != nil {
		return
	}

	var phase cloudv1.ClusterPhase
	secondaryStatus := cloudv1.DBSecondaryStatusHealthy
	if pd.Status == cloudv1.ComponentUnhealthy || tikv.Status == cloudv1.ComponentUnhealthy ||
		tidb.Status == cloudv1.ComponentUnhealthy || tiflash.Status == cloudv1.ComponentUnhealthy {
		secondaryStatus = cloudv1.DBSecondaryStatusUnhealthy
	}
	switch {
	// if both upgrading and scaling are ongoing (e.g. failover when upgrading), show Upgrading
	case tidb.Status == cloudv1.ComponentUpgrading || pd.Status == cloudv1.ComponentUpgrading ||
		tikv.Status == cloudv1.ComponentUpgrading || tiflash.Status == cloudv1.ComponentUpgrading:
		phase = cloudv1.ClusterPhaseUpgrading
	case tidb.Status == cloudv1.ComponentScaling || pd.Status == cloudv1.ComponentScaling ||
		tikv.Status == cloudv1.ComponentScaling || tiflash.Status == cloudv1.ComponentScaling:
		phase = cloudv1.ClusterPhaseScaling
	// cluster is considered as unavailable when one of TiKV, PD, TiDB is unavailable
	case tidb.Status == cloudv1.ComponentUnavailable || pd.Status == cloudv1.ComponentUnavailable ||
		tikv.Status == cloudv1.ComponentUnavailable:
		phase = cloudv1.ClusterPhaseUnavailable
		secondaryStatus = cloudv1.DBSecondaryStatusUnhealthy
	default:
		phase = cloudv1.ClusterPhaseRunning
	}
	reason := fmt.Sprintf("pd: %s, tidb: %s, tikv: %s, tiflash: %s", pd.Status, tidb.Status, tikv.Status, tiflash.Status)
	dbStatus := &cloudv1.DBStatus{
		Version:         r.tc.Spec.Version,
		TiDB:            tidb,
		TiKV:            tikv,
		PD:              pd,
		SecondaryStatus: secondaryStatus,
	}
	if r.cluster.Spec.Cluster.TiFlash.Replicas > 0 {
		dbStatus.TiFlash = tiflash
	}

	return r.ensureStatus(func() error {
		r.cluster.Status.Phase = phase
		r.cluster.Status.Reason = reason
		r.cluster.Status.DB = dbStatus
		return nil
	})
}

func (r *reconcileContext) computePDStatus() (*cloudv1.PDStatus, error) {
	var members []cloudv1.PDMember
	resourceRequests := r.tc.Spec.PD.ResourceRequirements.Requests
	for name, status := range r.tc.Status.PD.Members {
		memberStatus := cloudv1.MemberStatusNormal
		if !status.Health {
			memberStatus = cloudv1.MemberStatusUnavailable
		}
		members = append(members, cloudv1.PDMember{
			Name: name,
			Resources: cloudv1.Resources{
				CPU:     *resourceRequests.Cpu(),
				Memory:  *resourceRequests.Memory(),
				Storage: resourceRequests.Storage(),
			},
			Status: memberStatus,
		})
	}
	var status cloudv1.ComponentStatus
	switch {
	case !r.tc.PDIsAvailable():
		status = cloudv1.ComponentUnavailable
	case r.tc.PDUpgrading():
		status = cloudv1.ComponentUpgrading
	case r.tc.PDScaling():
		status = cloudv1.ComponentScaling
	case !r.tc.PDAllMembersReady():
		status = cloudv1.ComponentUnhealthy
	default:
		status = cloudv1.ComponentNormal
	}

	return &cloudv1.PDStatus{
		Members: members,
		Status:  status,
	}, nil
}

func (r *reconcileContext) computeTiDBStatus() (*cloudv1.TiDBStatus, error) {
	var members []cloudv1.TiDBMember
	resourceRequests := r.tc.Spec.TiDB.ResourceRequirements.Requests
	healthyCount := 0
	for name, status := range r.tc.Status.TiDB.Members {
		memberStatus := cloudv1.MemberStatusNormal
		if !status.Health {
			memberStatus = cloudv1.MemberStatusUnavailable
		} else {
			healthyCount++
		}

		var zone string
		if status.NodeName != "" {
			node := &corev1.Node{}
			if err := r.Cache.Get(r.ctx, client.ObjectKey{Name: status.NodeName}, node); err != nil {
				r.Error(err, "can't get node", "node", status.NodeName)
			} else {
				zone = node.GetLabels()[corev1.LabelTopologyZone]
			}
		}

		members = append(members, cloudv1.TiDBMember{
			Name: name,
			Labels: map[string]string{
				cloudv1.MemberLabelAvailableZone: zone,
			},
			Resources: cloudv1.Resources{
				CPU:    *resourceRequests.Cpu(),
				Memory: *resourceRequests.Memory(),
			},
			Status: memberStatus,
		})
	}
	var status cloudv1.ComponentStatus
	switch {
	case healthyCount == 0:
		status = cloudv1.ComponentUnavailable
	case r.tc.TiDBUpgrading():
		status = cloudv1.ComponentUpgrading
	case r.tc.TiDBScaling():
		status = cloudv1.ComponentScaling
	case !r.tc.TiDBAllMembersReady():
		status = cloudv1.ComponentUnhealthy
	default:
		status = cloudv1.ComponentNormal
	}

	return &cloudv1.TiDBStatus{
		Members: members,
		Status:  status,
	}, nil
}

func (r *reconcileContext) computeTiKVStatus() (*cloudv1.TiKVStatus, error) {
	tikvMembers, err := r.computeStoreMembers(&r.cluster.Spec.Cluster.TiKV)
	if err != nil {
		return nil, err
	}
	var status cloudv1.ComponentStatus
	switch {
	case !r.tc.TiKVIsAvailable():
		status = cloudv1.ComponentUnavailable
	case r.tc.TiKVUpgrading():
		status = cloudv1.ComponentUpgrading
	case r.tc.TiKVScaling():
		status = cloudv1.ComponentScaling
	case !r.tc.TiKVAllStoresReady():
		status = cloudv1.ComponentUnhealthy
	default:
		status = cloudv1.ComponentNormal
	}

	return &cloudv1.TiKVStatus{
		Members: tikvMembers,
		Status:  status,
	}, nil
}

func (r *reconcileContext) computeTiFlashStatus() (*cloudv1.TiFlashStatus, error) {
	if r.cluster.Spec.Cluster.TiFlash.Replicas == 0 {
		// dummy status
		return &cloudv1.TiFlashStatus{Status: cloudv1.ComponentNormal}, nil
	}
	members, err := r.computeStoreMembers(&r.cluster.Spec.Cluster.TiFlash)
	if err != nil {
		return nil, err
	}
	var status cloudv1.ComponentStatus
	readyStoreCount := 0
	for _, member := range members {
		if member.Status == cloudv1.MemberStatusNormal {
			readyStoreCount++
		}
	}
	switch {
	case readyStoreCount == 0:
		status = cloudv1.ComponentUnavailable
	case r.tc.TiFlashUpgrading():
		status = cloudv1.ComponentUpgrading
	case r.tc.Status.TiFlash.Phase == v1alpha1.ScalePhase:
		status = cloudv1.ComponentScaling
	case !r.tc.TiFlashAllStoresReady():
		status = cloudv1.ComponentUnhealthy
	default:
		status = cloudv1.ComponentNormal
	}

	return &cloudv1.TiFlashStatus{
		Members: members,
		Status:  status,
	}, nil
}

func (r *reconcileContext) computeStoreMembers(comp cloudv1.ClusterComponent) ([]cloudv1.StoreMember, error) {
	var resourceRequests corev1.ResourceList
	// nolint: exhaustive // nolint
	switch comp.GetComponentType() {
	case cloudv1.ComponentTiKV:
		resourceRequests = r.tc.Spec.TiKV.ResourceRequirements.Requests
	case cloudv1.ComponentTiFlash:
		resourceRequests = r.tc.Spec.TiFlash.ResourceRequirements.Requests
	default:
		return nil, fmt.Errorf("component %s is not store node", comp.GetComponentType())
	}

	if comp.GetReplicas() == 0 {
		return nil, nil
	}
	var members []cloudv1.StoreMember
	storePods, err := r.listComponentPods(comp.GetComponentType())
	if err != nil {
		return nil, err
	}
	for _, pod := range storePods {
		// 1. calculate member status
		var podToStoreMap map[string]v1alpha1.TiKVStore
		if comp.GetComponentType() == cloudv1.ComponentTiFlash {
			podToStoreMap = podToStore(r.tc.Status.TiFlash.Stores, r.tc.Status.TiFlash.TombstoneStores)
		} else {
			podToStoreMap = podToStore(r.tc.Status.TiKV.Stores, r.tc.Status.TiKV.TombstoneStores)
		}
		store, ok := podToStoreMap[pod.Name]
		var memberStatus cloudv1.MemberStatus
		switch {
		case ok && store.State == v1alpha1.TiKVStateUp:
			memberStatus = cloudv1.MemberStatusNormal
		case pod.DeletionTimestamp != nil:
			memberStatus = cloudv1.MemberStatusTerminating
		case ok && store.State == v1alpha1.TiKVStateTombstone:
			memberStatus = cloudv1.MemberStatusTerminating
		case ok && store.State == v1alpha1.TiKVStateOffline:
			memberStatus = cloudv1.MemberStatusLeaving
		case pod.Status.Phase == corev1.PodPending:
			memberStatus = cloudv1.MemberStatusCreating
		default:
			memberStatus = cloudv1.MemberStatusUnavailable
		}

		// 2. calculate member AZ
		var zone string
		if pod.Spec.NodeName != "" {
			node := &corev1.Node{}
			if err = r.Cache.Get(r.ctx, client.ObjectKey{Name: pod.Spec.NodeName}, node); err != nil {
				r.Error(err, "can't get node", "node", pod.Spec.NodeName)
			} else {
				zone = node.GetLabels()[corev1.LabelTopologyZone]
			}
		}

		// 3. compose
		members = append(members, cloudv1.StoreMember{
			Name: pod.Name,
			Labels: map[string]string{
				cloudv1.MemberLabelAvailableZone: zone,
			},
			Resources: cloudv1.Resources{
				CPU:     *resourceRequests.Cpu(),
				Memory:  *resourceRequests.Memory(),
				Storage: resourceRequests.Storage(),
			},
			Status: memberStatus,
		})
	}

	return members, nil
}

func (r *reconcileContext) listComponentPods(component cloudv1.ComponentType) ([]corev1.Pod, error) {
	podList := &corev1.PodList{}
	podLabels := map[string]string{
		label.ComponentLabelKey: string(component),
		label.InstanceLabelKey:  r.tc.Name,
	}
	if err := r.Cache.List(
		r.ctx, podList, client.InNamespace(r.tc.Namespace), client.MatchingLabels(podLabels)); err != nil {
		return nil, err
	}

	return podList.Items, nil
}

func (r *reconcileContext) updateLastActive(ctx context.Context) (recheckAfter time.Duration, err error) {
	now := metav1.Now()

	detectActiveDuration := utils.GetDetectActiveDuration(r.cluster)

	// just set it as now at the first time.
	if r.cluster.Status.LastActiveTimestamp == nil {
		err := r.ensureStatus(func() error {
			r.cluster.Status.LastActiveTimestamp = &now
			r.cluster.Status.LastDetectTimestamp = &now
			return nil
		})
		if err != nil {
			return 0, err
		}

		return detectActiveDuration, nil
	}

	if now.Sub(r.cluster.Status.LastDetectTimestamp.Time) < detectActiveDuration {
		return detectActiveDuration - now.Sub(r.cluster.Status.LastDetectTimestamp.Time), nil
	}

	active, err := Active(ctx, r.querier, r.cluster.Status.LastDetectTimestamp.Time, now.Time)
	if err != nil {
		return 0, err
	}

	err = r.ensureStatus(func() error {
		if active {
			r.cluster.Status.LastActiveTimestamp = &now
		}

		r.cluster.Status.LastDetectTimestamp = &now
		return nil
	})

	if err != nil {
		return 0, err
	}

	return detectActiveDuration, nil
}

func podToStore(normals, tombstones map[string]v1alpha1.TiKVStore) map[string]v1alpha1.TiKVStore {
	m := map[string]v1alpha1.TiKVStore{}
	for _, v := range tombstones {
		m[v.PodName] = v
	}
	for _, v := range normals {
		m[v.PodName] = v
	}

	return m
}