package cdccluster

import (
	"context"
	"fmt"

	"github.com/pelletier/go-toml"
	opv2alpha1 "github.com/pingcap/tidb-operator/api/v2/core/v1alpha1"
	opv1alpha1 "github.com/pingcap/tidb-operator/pkg/apis/pingcap/v1alpha1"
	cloudv1 "github.com/tidbcloud/infra-api/api/v1"
	"github.com/tidbcloud/infra-provider/pkg/consts"
	"github.com/tidbcloud/infra-provider/pkg/k8s"
	corev1 "k8s.io/api/core/v1"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
	"k8s.io/apimachinery/pkg/api/meta"
	"k8s.io/apimachinery/pkg/api/resource"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/types"
	"k8s.io/utils/ptr"
	"sigs.k8s.io/controller-runtime/pkg/client"
	"sigs.k8s.io/controller-runtime/pkg/log"

	cdcv1 "github.com/tidbcloud/ticdc-service/api/v1"
)

type v2Cluster struct {
	desired      *opv2alpha1.TiCDCGroup
	current      opv2alpha1.TiCDCGroup
	preStopImage string
}

func NewV2ClusterProvider(preStopImage string) Provider {
	return &v2Cluster{
		preStopImage: preStopImage,
	}
}

func (p *v2Cluster) Init(namespace, name string) {
	p.desired = &opv2alpha1.TiCDCGroup{
		ObjectMeta: metav1.ObjectMeta{
			Name:      name,
			Namespace: namespace,
		},
		Spec: opv2alpha1.TiCDCGroupSpec{
			Template: opv2alpha1.TiCDCTemplate{
				Spec: opv2alpha1.TiCDCTemplateSpec{
					Overlay: &opv2alpha1.Overlay{
						Pod: &opv2alpha1.PodOverlay{
							ObjectMeta: opv2alpha1.ObjectMeta{
								// All app.kubernetes.io labels are for compatibility with v1
								// It may be used in prometheus rules and svc selectors in v1
								// Also the overprovision scaler uses the instance label
								Labels: map[string]string{
									"app.kubernetes.io/component":  "ticdc",
									"app.kubernetes.io/instance":   name,
									"app.kubernetes.io/managed-by": "tidb-operator",
									"app.kubernetes.io/name":       "ticdc-cluster",
								},
							},
							Spec: &corev1.PodSpec{
								Affinity: &corev1.Affinity{
									PodAntiAffinity: &corev1.PodAntiAffinity{
										RequiredDuringSchedulingIgnoredDuringExecution: []corev1.PodAffinityTerm{
											{
												LabelSelector: &metav1.LabelSelector{
													MatchLabels: map[string]string{
														consts.ComponentLabelKey: string(cloudv1.ComponentTiCDC),
													},
												},
												TopologyKey: "kubernetes.io/hostname",
											},
										},
									},
								},
								HostNetwork:     true,
								DNSPolicy:       corev1.DNSClusterFirstWithHostNet,
								SecurityContext: &corev1.PodSecurityContext{},
							},
						},
					},
					PreStop: &opv2alpha1.TiCDCPreStop{
						Image: ptr.To(p.preStopImage),
					},
				},
			},
		},
	}
}

func (p *v2Cluster) SetReplicas(replicas int32) {
	p.desired.Spec.Replicas = ptr.To(replicas)
}

func (p *v2Cluster) SetVersion(baseImage string, version string) {
	p.desired.Spec.Template.Spec.Image = ptr.To(baseImage)
	p.desired.Spec.Template.Spec.Version = version
}

func (p *v2Cluster) SetUpstreamCluster(namespace string, name string) {
	// NOTE: TiDB Operator v2 does not support cross namespace yet
	p.desired.Spec.Cluster.Name = name
}

func (p *v2Cluster) SetTomlConfig(conf string) error {
	p.desired.Spec.Template.Spec.Config = opv2alpha1.ConfigFile(conf)
	return nil
}

func (p *v2Cluster) SetConfigField(key string, value interface{}) error {
	// use the utils from TiDB Operator v1 to parse and set config
	cdcConf := opv1alpha1.NewCDCConfig()
	if err := toml.Unmarshal([]byte(p.desired.Spec.Template.Spec.Config), &cdcConf.GenericConfig.MP); err != nil {
		return err
	}
	cdcConf.Set(key, value)
	b, err := cdcConf.MarshalTOML()
	if err != nil {
		return err
	}
	p.desired.Spec.Template.Spec.Config = opv2alpha1.ConfigFile(b)
	return nil
}

func (p *v2Cluster) SetNodeSelector(labelSelector map[string]string) {
	p.desired.Spec.Template.Spec.Overlay.Pod.Spec.NodeSelector = labelSelector
}

func (p *v2Cluster) SetStorage(class string, size string) {
	p.desired.Spec.Template.Spec.Overlay.Pod.Spec.Volumes = []corev1.Volume{
		{
			Name: TiCDCDefaultDirName,
			VolumeSource: corev1.VolumeSource{
				Ephemeral: &corev1.EphemeralVolumeSource{
					VolumeClaimTemplate: &corev1.PersistentVolumeClaimTemplate{
						Spec: corev1.PersistentVolumeClaimSpec{
							AccessModes:      []corev1.PersistentVolumeAccessMode{corev1.ReadWriteOnce},
							StorageClassName: ptr.To(class),
							Resources: corev1.VolumeResourceRequirements{
								Requests: map[corev1.ResourceName]resource.Quantity{
									corev1.ResourceStorage: resource.MustParse(size),
								},
							},
						},
					},
				},
			},
		},
	}
	p.desired.Spec.Template.Spec.Overlay.Pod.Spec.Containers = []corev1.Container{
		{
			Name: opv2alpha1.ContainerNameTiCDC,
			VolumeMounts: []corev1.VolumeMount{
				{
					Name:      TiCDCDefaultDirName,
					MountPath: TiCDCDefaultDirPath,
				},
			},
		},
	}
}

func (p *v2Cluster) GetClusterStatus(ctx context.Context, cli client.Client) (*cdcv1.TiCDCStatus, error) {
	if err := cli.Get(ctx, client.ObjectKeyFromObject(p.desired), &p.current); err != nil {
		return nil, err
	}
	phase, reason := convertPhaseReasonV2(&p.current)
	var instanceType string
	if p.current.Spec.Template.Spec.Overlay != nil &&
		p.current.Spec.Template.Spec.Overlay.Pod != nil &&
		p.current.Spec.Template.Spec.Overlay.Pod.Spec != nil {
		instanceType = p.current.Spec.Template.Spec.Overlay.Pod.Spec.NodeSelector[consts.InstanceTypeLabelKey]
	}
	return &cdcv1.TiCDCStatus{
		Phase:  phase,
		Reason: reason,
		NodeInfo: &cdcv1.NodeInfo{
			InstanceType: instanceType,
			Replicas:     int(*p.current.Spec.Replicas),
		},
	}, nil
}

func (p *v2Cluster) NeedReschedule(ctx context.Context, cli client.Client, desiredNode *Node) (bool, error) {
	err := cli.Get(ctx, client.ObjectKeyFromObject(p.desired), &p.current)
	if err != nil {
		if apierrors.IsNotFound(err) {
			return false, nil
		}
		return false, err
	}

	nodeSelector := p.current.Spec.Template.Spec.Overlay.Pod.Spec.NodeSelector
	instanceType, ok := nodeSelector[consts.InstanceTypeLabelKey]
	if !ok || instanceType != desiredNode.InstanceType {
		return true, nil
	}
	k8sVersion, ok := nodeSelector[consts.K8sVersionLabelKey]
	if !ok || k8sVersion != desiredNode.K8sVersion {
		return true, nil
	}
	tolerations := p.current.Spec.Template.Spec.Overlay.Pod.Spec.Tolerations
	if !containTaintValue(tolerations, desiredNode.TaintValue) {
		return true, nil
	}
	return false, nil
}

func (p *v2Cluster) Rescheduled(ctx context.Context, cli client.Client) (bool, error) {
	podSelector := client.MatchingLabels{
		"pingcap.com/group":     p.desired.Name,
		"pingcap.com/component": string(cloudv1.ComponentTiCDC),
	}
	pods := &corev1.PodList{}
	// list pods for the TiCDCGroup
	if err := cli.List(ctx, pods, client.InNamespace(p.desired.Namespace), podSelector); err != nil {
		return false, err
	}
	desiredSelector := p.desired.Spec.Template.Spec.Overlay.Pod.Spec.NodeSelector
	desiredTolerations := p.desired.Spec.Template.Spec.Overlay.Pod.Spec.Tolerations
	scheduled := 0
	for _, pod := range pods.Items {
		if pod.Spec.NodeName == "" { // pod not scheduled
			continue
		}
		selector := pod.Spec.NodeSelector
		if selector[consts.InstanceTypeLabelKey] != desiredSelector[consts.InstanceTypeLabelKey] {
			continue
		}
		if selector[consts.K8sVersionLabelKey] != desiredSelector[consts.K8sVersionLabelKey] {
			continue
		}
		tolerationsMatched := true
		for _, desiredToleration := range desiredTolerations {
			if !containTaintValue(pod.Spec.Tolerations, desiredToleration.Value) {
				tolerationsMatched = false
				break
			}
		}
		if !tolerationsMatched {
			continue
		}
		scheduled++
	}

	return scheduled == int(*p.desired.Spec.Replicas), nil
}

func (p *v2Cluster) ApplyCR(ctx context.Context, cli client.Client) error {
	return k8s.Apply(ctx, cli, p.desired)
}

func (p *v2Cluster) DeleteCR(ctx context.Context, cli client.Client, name types.NamespacedName) error {
	cdcGroup := opv2alpha1.TiCDCGroup{
		ObjectMeta: metav1.ObjectMeta{
			Name:      name.Name,
			Namespace: name.Namespace,
		},
	}
	return k8s.DeleteObject(ctx, cli, &cdcGroup)
}

func convertPhaseReasonV2(cdcGroup *opv2alpha1.TiCDCGroup) (cdcv1.TiCDCPhase, string) {
	suspended := meta.IsStatusConditionTrue(cdcGroup.Status.Conditions, opv2alpha1.CondSuspended)
	ready := meta.IsStatusConditionTrue(cdcGroup.Status.Conditions, opv2alpha1.CondReady)
	synced := cdcGroup.Generation == cdcGroup.Status.ObservedGeneration &&
		meta.IsStatusConditionTrue(cdcGroup.Status.Conditions, opv2alpha1.CondSynced)
	switch {
	case suspended:
		return cdcv1.TiCDCPhaseSuspend, "Suspend"
	case synced && ready:
		return cdcv1.TiCDCPhaseRunning, "Running"
	case !synced:
		if cdcGroup.Status.Replicas != *cdcGroup.Spec.Replicas {
			return cdcv1.TiCDCPhaseScaling, "Scaling"
		}
		return cdcv1.TiCDCPhaseUpgrading, "Upgrading"
	default:
		// if some instances are ready, should we mark it as running?
		return cdcv1.TiCDCPhaseUnavailable, "Unavailable"
	}
}

func (p *v2Cluster) SetRestart(restartedAt string) {
	// TODO: implement it
}

func (p *v2Cluster) SetTolerations(tolerations []corev1.Toleration) {
	p.desired.Spec.Template.Spec.Overlay.Pod.Spec.Tolerations = tolerations
}

func (p *v2Cluster) SetServiceAccount(ctx context.Context, cli client.Client) error {
	sa := &corev1.ServiceAccount{
		ObjectMeta: metav1.ObjectMeta{
			Name:      p.desired.Name,
			Namespace: p.desired.Namespace,
		},
	}
	if err := k8s.Apply(ctx, cli, sa); err != nil {
		return fmt.Errorf("create service account error: %w", err)
	}

	p.desired.Spec.Template.Spec.Overlay.Pod.Spec.ServiceAccountName = sa.Name
	return nil
}

func (p *v2Cluster) EnableMountCertificate(ctx context.Context, sidecarImage string, cli client.Client) error {
	log.FromContext(ctx).Info("EnableMountCertificate is not implemented for operator v2")
	return nil
}