package core

import (
	"context"
	"errors"
	"fmt"
	"reflect"
	"sort"
	"strings"

	"github.com/pingcap/log"
	"github.com/tidbcloud/tiflash-manager/pkg/config"
	"github.com/tidbcloud/tiflash-manager/pkg/k8s"
	"go.uber.org/zap"
	appsv1 "k8s.io/api/apps/v1"
	autoscalingv2 "k8s.io/api/autoscaling/v2"
	corev1 "k8s.io/api/core/v1"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/client-go/kubernetes"
)

const (
	managedByLabelKey    = "app.kubernetes.io/managed-by"
	managedByLabelValue  = "tiflash-manager"
	managedPoolLabelKey  = "tiflash-manager.pingcap.com/pool"
	templatePoolLabelKey = "tiflash-manager.pingcap.com/template-pool"
)

type PoolReconciler struct {
	conf      *config.Config
	k8sClient kubernetes.Interface
}

func NewPoolReconciler(conf *config.Config) *PoolReconciler {
	return &PoolReconciler{
		conf:      conf,
		k8sClient: k8s.SingleKubernetesClient(),
	}
}

func (r *PoolReconciler) Reconcile(ctx context.Context, pools map[string]PoolSpec) error {
	names := make([]string, 0, len(pools))
	desiredPools := make(map[string]struct{}, len(pools))
	for poolName := range pools {
		if poolName == "" {
			continue
		}
		names = append(names, poolName)
		desiredPools[poolName] = struct{}{}
	}
	sort.Strings(names)

	var errs []error
	for _, poolName := range names {
		spec := pools[poolName]
		if !spec.Create {
			continue
		}

		templatePool := spec.EffectiveTemplatePool(r.conf.Tiflash.TemplatePool)
		if poolName == templatePool {
			continue
		}

		if err := r.ensurePool(ctx, poolName, templatePool); err != nil {
			errs = append(errs, fmt.Errorf("reconcile pool %s: %w", poolName, err))
		}
	}

	if err := r.pruneRemovedPools(ctx, desiredPools); err != nil {
		errs = append(errs, fmt.Errorf("prune removed pools: %w", err))
	}
	return errors.Join(errs...)
}

func (r *PoolReconciler) ensurePool(ctx context.Context, poolName, templatePool string) error {
	if err := r.ensurePeerService(ctx, poolName, templatePool); err != nil {
		return err
	}
	if err := r.ensureStatefulSet(ctx, poolName, templatePool); err != nil {
		return err
	}
	if err := r.ensureHPA(ctx, poolName, templatePool); err != nil {
		return err
	}
	return nil
}

func (r *PoolReconciler) ensurePeerService(ctx context.Context, poolName, templatePool string) error {
	name := r.conf.PeerServiceName(poolName)
	_, err := r.k8sClient.CoreV1().Services(r.conf.Tiflash.Namespace).Get(ctx, name, metav1.GetOptions{})
	if err == nil {
		return nil
	}
	if !apierrors.IsNotFound(err) {
		return err
	}

	templateName := r.conf.PeerServiceName(templatePool)
	template, err := r.k8sClient.CoreV1().Services(r.conf.Tiflash.Namespace).Get(ctx, templateName, metav1.GetOptions{})
	if err != nil {
		return fmt.Errorf("get template Service %s: %w", templateName, err)
	}

	service := cloneServiceForPool(template, poolName, templatePool, r.conf)
	if _, err := r.k8sClient.CoreV1().Services(r.conf.Tiflash.Namespace).Create(ctx, service, metav1.CreateOptions{}); err != nil {
		return err
	}

	log.Info("Created pool peer Service", zapPoolFields(poolName, templatePool)...)
	return nil
}

func (r *PoolReconciler) ensureStatefulSet(ctx context.Context, poolName, templatePool string) error {
	name := r.conf.StatefulSetName(poolName)
	_, err := r.k8sClient.AppsV1().StatefulSets(r.conf.Tiflash.Namespace).Get(ctx, name, metav1.GetOptions{})
	if err == nil {
		return nil
	}
	if !apierrors.IsNotFound(err) {
		return err
	}

	templateName := r.conf.StatefulSetName(templatePool)
	template, err := r.k8sClient.AppsV1().StatefulSets(r.conf.Tiflash.Namespace).Get(ctx, templateName, metav1.GetOptions{})
	if err != nil {
		return fmt.Errorf("get template StatefulSet %s: %w", templateName, err)
	}

	statefulSet := cloneStatefulSetForPool(template, poolName, templatePool, r.conf)
	if _, err := r.k8sClient.AppsV1().StatefulSets(r.conf.Tiflash.Namespace).Create(ctx, statefulSet, metav1.CreateOptions{}); err != nil {
		return err
	}

	log.Info("Created pool StatefulSet", zapPoolFields(poolName, templatePool)...)
	return nil
}

func (r *PoolReconciler) ensureHPA(ctx context.Context, poolName, templatePool string) error {
	name := r.conf.HPAName(poolName)
	_, err := r.k8sClient.AutoscalingV2().HorizontalPodAutoscalers(r.conf.Tiflash.Namespace).Get(ctx, name, metav1.GetOptions{})
	if err == nil {
		return nil
	}
	if !apierrors.IsNotFound(err) {
		return err
	}

	templateName := r.conf.HPAName(templatePool)
	template, err := r.k8sClient.AutoscalingV2().HorizontalPodAutoscalers(r.conf.Tiflash.Namespace).Get(ctx, templateName, metav1.GetOptions{})
	if err != nil {
		return fmt.Errorf("get template HPA %s: %w", templateName, err)
	}

	hpa := cloneHPAForPool(template, poolName, templatePool, r.conf)
	if _, err := r.k8sClient.AutoscalingV2().HorizontalPodAutoscalers(r.conf.Tiflash.Namespace).Create(ctx, hpa, metav1.CreateOptions{}); err != nil {
		return err
	}

	log.Info("Created pool HPA", zapPoolFields(poolName, templatePool)...)
	return nil
}

func (r *PoolReconciler) pruneRemovedPools(ctx context.Context, desiredPools map[string]struct{}) error {
	var errs []error

	if err := r.pruneRemovedHPAs(ctx, desiredPools); err != nil {
		errs = append(errs, err)
	}
	if err := r.pruneRemovedStatefulSets(ctx, desiredPools); err != nil {
		errs = append(errs, err)
	}
	if err := r.pruneRemovedServices(ctx, desiredPools); err != nil {
		errs = append(errs, err)
	}

	return errors.Join(errs...)
}

func (r *PoolReconciler) pruneRemovedHPAs(ctx context.Context, desiredPools map[string]struct{}) error {
	hpas, err := r.k8sClient.AutoscalingV2().HorizontalPodAutoscalers(r.conf.Tiflash.Namespace).List(ctx, metav1.ListOptions{
		LabelSelector: managedByLabelKey + "=" + managedByLabelValue,
	})
	if err != nil {
		return fmt.Errorf("list managed HPAs: %w", err)
	}

	var errs []error
	for i := range hpas.Items {
		hpa := hpas.Items[i]
		if !shouldDeleteManagedResource(hpa.Labels, desiredPools) {
			continue
		}
		if err := r.k8sClient.AutoscalingV2().HorizontalPodAutoscalers(r.conf.Tiflash.Namespace).Delete(ctx, hpa.Name, metav1.DeleteOptions{}); err != nil && !apierrors.IsNotFound(err) {
			errs = append(errs, fmt.Errorf("delete HPA %s: %w", hpa.Name, err))
			continue
		}
		log.Info("Deleted removed pool HPA", zap.String("poolName", hpa.Labels[managedPoolLabelKey]), zap.String("name", hpa.Name))
	}
	return errors.Join(errs...)
}

func (r *PoolReconciler) pruneRemovedStatefulSets(ctx context.Context, desiredPools map[string]struct{}) error {
	statefulSets, err := r.k8sClient.AppsV1().StatefulSets(r.conf.Tiflash.Namespace).List(ctx, metav1.ListOptions{
		LabelSelector: managedByLabelKey + "=" + managedByLabelValue,
	})
	if err != nil {
		return fmt.Errorf("list managed StatefulSets: %w", err)
	}

	var errs []error
	for i := range statefulSets.Items {
		statefulSet := statefulSets.Items[i]
		if !shouldDeleteManagedResource(statefulSet.Labels, desiredPools) {
			continue
		}
		if err := r.k8sClient.AppsV1().StatefulSets(r.conf.Tiflash.Namespace).Delete(ctx, statefulSet.Name, metav1.DeleteOptions{}); err != nil && !apierrors.IsNotFound(err) {
			errs = append(errs, fmt.Errorf("delete StatefulSet %s: %w", statefulSet.Name, err))
			continue
		}
		log.Info("Deleted removed pool StatefulSet", zap.String("poolName", statefulSet.Labels[managedPoolLabelKey]), zap.String("name", statefulSet.Name))
	}
	return errors.Join(errs...)
}

func (r *PoolReconciler) pruneRemovedServices(ctx context.Context, desiredPools map[string]struct{}) error {
	services, err := r.k8sClient.CoreV1().Services(r.conf.Tiflash.Namespace).List(ctx, metav1.ListOptions{
		LabelSelector: managedByLabelKey + "=" + managedByLabelValue,
	})
	if err != nil {
		return fmt.Errorf("list managed Services: %w", err)
	}

	var errs []error
	for i := range services.Items {
		service := services.Items[i]
		if !shouldDeleteManagedResource(service.Labels, desiredPools) {
			continue
		}
		if err := r.k8sClient.CoreV1().Services(r.conf.Tiflash.Namespace).Delete(ctx, service.Name, metav1.DeleteOptions{}); err != nil && !apierrors.IsNotFound(err) {
			errs = append(errs, fmt.Errorf("delete Service %s: %w", service.Name, err))
			continue
		}
		log.Info("Deleted removed pool Service", zap.String("poolName", service.Labels[managedPoolLabelKey]), zap.String("name", service.Name))
	}
	return errors.Join(errs...)
}

func shouldDeleteManagedResource(labels map[string]string, desiredPools map[string]struct{}) bool {
	if len(labels) == 0 {
		return false
	}

	poolName := labels[managedPoolLabelKey]
	if poolName == "" {
		return false
	}

	_, keep := desiredPools[poolName]
	return !keep
}

func cloneServiceForPool(template *corev1.Service, poolName, templatePool string, conf *config.Config) *corev1.Service {
	service := template.DeepCopy()
	rules := newReplacementRules(conf, poolName, templatePool)
	applyResourceTemplate(service, rules)
	rewriteStringMap(service.Labels, templatePool, poolName, rules)
	rewriteStringMap(service.Spec.Selector, templatePool, poolName, rules)
	resetObjectMeta(&service.ObjectMeta, conf.PeerServiceName(poolName), conf.Tiflash.Namespace, poolName, templatePool)
	service.Status = corev1.ServiceStatus{}
	if service.Spec.ClusterIP != corev1.ClusterIPNone {
		service.Spec.ClusterIP = ""
		service.Spec.ClusterIPs = nil
	}
	for i := range service.Spec.Ports {
		service.Spec.Ports[i].NodePort = 0
	}
	return service
}

func cloneStatefulSetForPool(template *appsv1.StatefulSet, poolName, templatePool string, conf *config.Config) *appsv1.StatefulSet {
	statefulSet := template.DeepCopy()
	rules := newReplacementRules(conf, poolName, templatePool)
	applyResourceTemplate(statefulSet, rules)
	rewriteStringMap(statefulSet.Labels, templatePool, poolName, rules)
	rewriteLabelSelector(statefulSet.Spec.Selector, templatePool, poolName, rules)
	rewriteStringMap(statefulSet.Spec.Template.Labels, templatePool, poolName, rules)
	for i := range statefulSet.Spec.VolumeClaimTemplates {
		rewriteStringMap(statefulSet.Spec.VolumeClaimTemplates[i].Labels, templatePool, poolName, rules)
	}
	resetObjectMeta(&statefulSet.ObjectMeta, conf.StatefulSetName(poolName), conf.Tiflash.Namespace, poolName, templatePool)
	statefulSet.Status = appsv1.StatefulSetStatus{}
	return statefulSet
}

func cloneHPAForPool(template *autoscalingv2.HorizontalPodAutoscaler, poolName, templatePool string, conf *config.Config) *autoscalingv2.HorizontalPodAutoscaler {
	hpa := template.DeepCopy()
	rules := newReplacementRules(conf, poolName, templatePool)
	applyResourceTemplate(hpa, rules)
	rewriteStringMap(hpa.Labels, templatePool, poolName, rules)
	resetObjectMeta(&hpa.ObjectMeta, conf.HPAName(poolName), conf.Tiflash.Namespace, poolName, templatePool)
	hpa.Status = autoscalingv2.HorizontalPodAutoscalerStatus{}
	return hpa
}

func resetObjectMeta(meta *metav1.ObjectMeta, name, namespace, poolName, templatePool string) {
	meta.Name = name
	meta.Namespace = namespace
	meta.UID = ""
	meta.ResourceVersion = ""
	meta.Generation = 0
	meta.CreationTimestamp = metav1.Time{}
	meta.DeletionTimestamp = nil
	meta.DeletionGracePeriodSeconds = nil
	meta.GenerateName = ""
	meta.ManagedFields = nil
	meta.OwnerReferences = nil
	meta.Finalizers = nil
	if meta.Labels == nil {
		meta.Labels = make(map[string]string)
	}
	meta.Labels[managedByLabelKey] = managedByLabelValue
	meta.Labels[managedPoolLabelKey] = poolName
	if meta.Annotations == nil {
		meta.Annotations = make(map[string]string)
	}
	meta.Annotations[templatePoolLabelKey] = templatePool
}

type stringReplacement struct {
	old string
	new string
}

type replacementRules struct {
	contains []stringReplacement
}

func newReplacementRules(conf *config.Config, poolName, templatePool string) replacementRules {
	return replacementRules{
		contains: []stringReplacement{
			{old: conf.PeerServiceName(templatePool), new: conf.PeerServiceName(poolName)},
			{old: conf.StatefulSetName(templatePool), new: conf.StatefulSetName(poolName)},
			{old: conf.HPAName(templatePool), new: conf.HPAName(poolName)},
		},
	}
}

// applyResourceTemplate rewrites explicit resource-name references inside the object graph.
func applyResourceTemplate(target any, rules replacementRules) {
	walkReplace(reflect.ValueOf(target), rules)
}

func walkReplace(value reflect.Value, rules replacementRules) {
	if !value.IsValid() {
		return
	}

	switch value.Kind() {
	case reflect.Pointer, reflect.Interface:
		if !value.IsNil() {
			walkReplace(value.Elem(), rules)
		}
	case reflect.Struct:
		for i := 0; i < value.NumField(); i++ {
			walkReplace(value.Field(i), rules)
		}
	case reflect.Slice, reflect.Array:
		for i := 0; i < value.Len(); i++ {
			walkReplace(value.Index(i), rules)
		}
	case reflect.String:
		if value.CanSet() {
			value.SetString(rewriteString(value.String(), rules))
		}
	}
}

func cloneReflectValue(value reflect.Value) reflect.Value {
	clone := reflect.New(value.Type()).Elem()
	clone.Set(value)
	return clone
}

func rewriteString(value string, rules replacementRules) string {
	for _, item := range rules.contains {
		value = strings.ReplaceAll(value, item.old, item.new)
	}
	return value
}

func rewriteStringMap(values map[string]string, templatePool, poolName string, rules replacementRules) {
	for key, value := range values {
		values[key] = rewritePoolScopedString(value, templatePool, poolName, rules)
	}
}

func rewriteLabelSelector(selector *metav1.LabelSelector, templatePool, poolName string, rules replacementRules) {
	if selector == nil {
		return
	}

	rewriteStringMap(selector.MatchLabels, templatePool, poolName, rules)
	for i := range selector.MatchExpressions {
		for j := range selector.MatchExpressions[i].Values {
			selector.MatchExpressions[i].Values[j] = rewritePoolScopedString(selector.MatchExpressions[i].Values[j], templatePool, poolName, rules)
		}
	}
}

func rewritePoolScopedString(value, templatePool, poolName string, rules replacementRules) string {
	if value == templatePool {
		return poolName
	}
	return rewriteString(value, rules)
}

func zapPoolFields(poolName, templatePool string) []zap.Field {
	return []zap.Field{
		zap.String("poolName", poolName),
		zap.String("templatePool", templatePool),
	}
}