package cdccluster

import (
	"context"
	"fmt"

	"github.com/pelletier/go-toml"
	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"
	cdcv1 "github.com/tidbcloud/ticdc-service/api/v1"
	ticdc_consts "github.com/tidbcloud/ticdc-service/pkg/consts"
	corev1 "k8s.io/api/core/v1"
	rbacv1 "k8s.io/api/rbac/v1"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
	"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"
)

type v1Cluster struct {
	desired *opv1alpha1.TidbCluster
	current opv1alpha1.TidbCluster
}

func NewV1ClusterProvider() Provider {
	return &v1Cluster{}
}

func (p *v1Cluster) Init(namespace, name string) {
	p.desired = &opv1alpha1.TidbCluster{
		ObjectMeta: metav1.ObjectMeta{
			// Note: operator v1 uses the same CRD as tidb cluster, we can't pass the same name as tidb cluster
			Name:      name,
			Namespace: namespace,
		},
		Spec: opv1alpha1.TidbClusterSpec{
			PVReclaimPolicy:      ptr.To(corev1.PersistentVolumeReclaimDelete),
			ConfigUpdateStrategy: opv1alpha1.ConfigUpdateStrategyRollingUpdate,
			TLSCluster: &opv1alpha1.TLSCluster{
				Enabled: true,
			},
			TiCDC: &opv1alpha1.TiCDCSpec{
				ComponentSpec: opv1alpha1.ComponentSpec{
					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:        ptr.To(true),
					PodSecurityContext: &corev1.PodSecurityContext{},
				},
			},
		},
	}
}

func (p *v1Cluster) SetReplicas(replicas int32) {
	p.desired.Spec.TiCDC.Replicas = replicas
}

func (p *v1Cluster) SetTomlConfig(conf string) error {
	cdcConf := opv1alpha1.NewCDCConfig()
	if err := toml.Unmarshal([]byte(conf), &cdcConf.GenericConfig.MP); err != nil {
		return err
	}
	p.desired.Spec.TiCDC.Config = cdcConf
	return nil
}

func (p *v1Cluster) SetConfigField(key string, value interface{}) error {
	p.desired.Spec.TiCDC.Config.Set(key, value)
	return nil
}

func (p *v1Cluster) SetUpstreamCluster(namespace, name string) {
	p.desired.Spec.Cluster = &opv1alpha1.TidbClusterRef{
		Namespace: namespace,
		Name:      name,
	}
}

func (p *v1Cluster) SetVersion(baseImage string, version string) {
	p.desired.Spec.TiCDC.Version = ptr.To(version)
	p.desired.Spec.TiCDC.BaseImage = baseImage
}

func (p *v1Cluster) SetNodeSelector(labelSelector map[string]string) {
	p.desired.Spec.TiCDC.NodeSelector = labelSelector
}

func (p *v1Cluster) SetStorage(class string, size string) {
	p.desired.Spec.TiCDC.AdditionalVolumes = []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.TiCDC.AdditionalVolumeMounts = []corev1.VolumeMount{
		{
			Name:      TiCDCDefaultDirName,
			MountPath: TiCDCDefaultDirPath,
		},
	}
}

func (p *v1Cluster) 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.TiCDC.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
	}
	if !containTaintValue(p.current.Spec.TiCDC.Tolerations, desiredNode.TaintValue) {
		return true, nil
	}
	return false, nil
}

func (p *v1Cluster) Rescheduled(ctx context.Context, cli client.Client) (bool, error) {
	podSelector := client.MatchingLabels{
		"app.kubernetes.io/instance":  p.desired.Name,
		"app.kubernetes.io/component": string(cloudv1.ComponentTiCDC),
	}
	pods := &corev1.PodList{}
	// list pods for the TiCDC TidbCluster
	if err := cli.List(ctx, pods, client.InNamespace(p.desired.Namespace), podSelector); err != nil {
		return false, err
	}
	desiredSelector := p.desired.Spec.TiCDC.NodeSelector
	desiredTolerations := p.desired.Spec.TiCDC.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.TiCDC.Replicas), nil
}

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

func (p *v1Cluster) DeleteCR(ctx context.Context, cli client.Client, name types.NamespacedName) error {
	tc := opv1alpha1.TidbCluster{
		ObjectMeta: metav1.ObjectMeta{
			Name:      name.Name,
			Namespace: name.Namespace,
		},
	}
	return k8s.DeleteObject(ctx, cli, &tc)
}

func (p *v1Cluster) 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
	}
	var reason string
	for _, cond := range p.current.Status.Conditions {
		if cond.Type == opv1alpha1.TidbClusterReady {
			reason = cond.Message
		}
	}
	status := p.current.Status.TiCDC
	return &cdcv1.TiCDCStatus{
		Phase: convertPhase(status),
		NodeInfo: &cdcv1.NodeInfo{
			InstanceType: p.current.Spec.TiCDC.NodeSelector[consts.InstanceTypeLabelKey],
			Replicas:     int(p.current.Spec.TiCDC.Replicas),
		},
		Reason: reason,
	}, nil
}

func (p *v1Cluster) SetRestart(restartedAt string) {
	if len(p.desired.Spec.TiCDC.Annotations) == 0 {
		p.desired.Spec.TiCDC.Annotations = make(map[string]string)
	}
	p.desired.Spec.TiCDC.Annotations[ticdc_consts.RestartTiCDCOperatorV1Annotation] = restartedAt
}

func (p *v1Cluster) SetTolerations(tolerations []corev1.Toleration) {
	p.desired.Spec.TiCDC.Tolerations = tolerations
}

func (p *v1Cluster) 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.TiCDC.ServiceAccount = sa.Name
	return nil
}

func (p *v1Cluster) EnableMountCertificate(ctx context.Context, sidecarImage string, cli client.Client) error {
	if err := p.createCertRBAC(ctx, cli); err != nil {
		return err
	}

	p.desired.Spec.TiCDC.AdditionalVolumes = append(p.desired.Spec.TiCDC.AdditionalVolumes, corev1.Volume{
		Name: "cert-volume",
		VolumeSource: corev1.VolumeSource{
			EmptyDir: &corev1.EmptyDirVolumeSource{},
		},
	})
	p.desired.Spec.TiCDC.AdditionalVolumeMounts = append(p.desired.Spec.TiCDC.AdditionalVolumeMounts, corev1.VolumeMount{
		Name:      "cert-volume",
		MountPath: ticdc_consts.TiCDCCertMountPath,
	})
	p.desired.Spec.TiCDC.AdditionalContainers = append(p.desired.Spec.TiCDC.AdditionalContainers, corev1.Container{
		Name:  "cert-sidecar",
		Image: sidecarImage,
		Env: []corev1.EnvVar{
			{
				Name:  "FOLDER",
				Value: ticdc_consts.TiCDCCertMountPath,
			},
			{
				Name:  "LABEL",
				Value: ticdc_consts.TiCDCCertLabel,
			},
		},
		VolumeMounts: []corev1.VolumeMount{
			{
				Name:      "cert-volume",
				MountPath: ticdc_consts.TiCDCCertMountPath,
			},
		},
	})
	return nil
}

func (p *v1Cluster) createCertRBAC(ctx context.Context, cli client.Client) error {
	role := &rbacv1.Role{
		ObjectMeta: metav1.ObjectMeta{
			Name:      p.desired.Name,
			Namespace: p.desired.Namespace,
		},
		Rules: []rbacv1.PolicyRule{
			{
				APIGroups: []string{""},
				Resources: []string{"secrets"},
				Verbs:     []string{"get", "list", "watch"},
			},
		},
	}
	if err := k8s.Apply(ctx, cli, role); err != nil {
		return fmt.Errorf("create role for cdc cert error: %w", err)
	}

	roleBinding := &rbacv1.RoleBinding{
		ObjectMeta: metav1.ObjectMeta{
			Name:      p.desired.Name,
			Namespace: p.desired.Namespace,
		},
		RoleRef: rbacv1.RoleRef{
			APIGroup: rbacv1.GroupName,
			Kind:     "Role",
			Name:     role.Name,
		},
		Subjects: []rbacv1.Subject{
			{
				Kind:      "ServiceAccount",
				Name:      p.desired.Spec.TiCDC.ServiceAccount,
				Namespace: p.desired.Namespace,
			},
		},
	}

	if err := k8s.Apply(ctx, cli, roleBinding); err != nil {
		return fmt.Errorf("create role binding for cdc cert error: %w", err)
	}
	return nil
}

// tidb-operator v1 seems only set 2 phase for ticdc
// https://github.com/pingcap/tidb-operator/blob/master/pkg/manager/member/ticdc_member_manager.go#L245-L249
func convertPhase(status opv1alpha1.TiCDCStatus) cdcv1.TiCDCPhase {
	switch status.Phase {
	case opv1alpha1.NormalPhase:
		if status.StatefulSet.ReadyReplicas == 0 {
			return cdcv1.TiCDCPhaseUnavailable
		}
		if status.Synced && status.StatefulSet.ReadyReplicas == status.StatefulSet.Replicas {
			return cdcv1.TiCDCPhaseRunning
		}
		return cdcv1.TiCDCPhaseScaling
	case opv1alpha1.UpgradePhase:
		return cdcv1.TiCDCPhaseUpgrading
	case opv1alpha1.ScalePhase:
		return cdcv1.TiCDCPhaseScaling
	default:
		return cdcv1.TiCDCPhaseCreating
	}
}

func containTaintValue(tolerations []corev1.Toleration, taintValue string) bool {
	for _, toleration := range tolerations {
		if toleration.Value == taintValue {
			return true
		}
	}
	return false
}