package dwworker

import (
	"context"
	"encoding/json"
	"fmt"
	"strings"
	"time"

	"github.com/go-logr/logr"
	"github.com/pingcap/errors"
	appsv1 "k8s.io/api/apps/v1"
	corev1 "k8s.io/api/core/v1"
	"k8s.io/apimachinery/pkg/api/equality"
	"k8s.io/apimachinery/pkg/api/resource"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/client-go/kubernetes"
	"k8s.io/utils/pointer"
	ctrl "sigs.k8s.io/controller-runtime"
	"sigs.k8s.io/controller-runtime/pkg/client"
	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"

	cloudv1 "github.com/tidbcloud/infra-api/api/v1"
	"github.com/tidbcloud/infra-provider/pkg/config"
	"github.com/tidbcloud/infra-provider/pkg/consts"
	"github.com/tidbcloud/infra-provider/pkg/k8s"
)

const (
	finalizerName = "dwworkers.cloud.pingcap.com/finalizer"

	defaultWorkerImage       = "breezewish/dwworker:nightly" // Override in config
	workerCommand            = "/bin/dwworker"
	workerAPIPort      int32 = 8185
)

type ReconcileContext struct {
	Ctx context.Context

	RemoteClient     client.Client
	RemoteRestClient kubernetes.Interface

	LocalClient client.Client
	LocalCache  client.Reader

	Config config.RootConfig

	Obj     *cloudv1.DWWorker
	Cluster *cloudv1.Cluster // Maybe nil, if cluster is deleted.
	Network *cloudv1.Network

	ClusterProxyAuthDecorator RequestDecorator

	Logger logr.Logger
}

func (r *ReconcileContext) getInstanceName() string {
	return fmt.Sprintf("dwworker-%s", r.Obj.Name)
}

func (r *ReconcileContext) Reconcile() (res ctrl.Result, err error) {
	if r.Obj.DeletionTimestamp.IsZero() {
		// Object is not being deleted
		if !controllerutil.ContainsFinalizer(r.Obj, finalizerName) {
			controllerutil.AddFinalizer(r.Obj, finalizerName)
			if err := r.LocalClient.Update(r.Ctx, r.Obj); err != nil {
				return ctrl.Result{}, fmt.Errorf("update to add finalizer failed: %w", err)
			}
		}
	} else {
		// Object is being deleted
		if controllerutil.ContainsFinalizer(r.Obj, finalizerName) {
			if err := r.reconcileDelete(); err != nil {
				return ctrl.Result{}, err
			}
			controllerutil.RemoveFinalizer(r.Obj, finalizerName)
			if err := r.LocalClient.Update(r.Ctx, r.Obj); err != nil {
				return ctrl.Result{}, fmt.Errorf("update to remove finalizer failed: %w", err)
			}
		}

		// Stop reconciliation as the item is being deleted
		return ctrl.Result{}, nil
	}

	return r.reconcileCreateOrUpdate()
}

func (r *ReconcileContext) getDataPlaneNamespace() string {
	return cloudv1.ClusterPrefixId(r.Obj.Spec.ClusterName)
}

func (r *ReconcileContext) reconcileCreateOrUpdate() (res ctrl.Result, err error) {
	r.Logger.Info("DWWorker reconcileCreateOrUpdate", "obj", r.Obj)

	err = r.reconcileRemoteSpec()
	if err != nil {
		return ctrl.Result{}, err
	}

	return r.reconcileSelfStatus()
}

func (r *ReconcileContext) reconcileRemoteSpec() (err error) {
	// DWWorker should be auto-deleted when Cluster is deleted.
	if r.Cluster != nil {
		if !config.FeatureEnabled(r.Config.FeatureGates, config.EnableDPCompatibilityMode, "", "") {
			lastObj := r.Obj.DeepCopyObject()
			if err := controllerutil.SetOwnerReference(r.Cluster, r.Obj, r.LocalClient.Scheme()); err != nil {
				return fmt.Errorf("set DWWorker owner failed: %w", err)
			}
			if !equality.Semantic.DeepEqual(lastObj, r.Obj) {
				r.Logger.Info("DWWorker set ownerReference", "owner", r.Cluster.UID)
				if err := r.LocalClient.Update(r.Ctx, r.Obj); err != nil {
					return fmt.Errorf("update DWWorker ownerReference failed: %w", err)
				}
			}
		}
	}

	image := r.Config.ControllerConfig.DWWorker.Image
	if image == "" {
		image = defaultWorkerImage
	}

	// Always re-pull the image when it's nightly or latest
	imagePullPolicy := corev1.PullIfNotPresent
	if strings.HasSuffix(image, ":nightly") || strings.HasSuffix(image, ":latest") {
		imagePullPolicy = corev1.PullAlways
	}

	objDeploy := &appsv1.Deployment{
		ObjectMeta: metav1.ObjectMeta{
			Name:      r.getInstanceName(),
			Namespace: r.getDataPlaneNamespace(),
		},
		Spec: appsv1.DeploymentSpec{
			Replicas: pointer.Int32(1),
			Selector: &metav1.LabelSelector{
				MatchLabels: map[string]string{
					consts.NameLabelKey:     "dwworker",
					consts.InstanceLabelKey: r.getInstanceName(),
				},
			},
			Template: corev1.PodTemplateSpec{
				ObjectMeta: metav1.ObjectMeta{
					Labels: map[string]string{
						consts.NameLabelKey:     "dwworker",
						consts.InstanceLabelKey: r.getInstanceName(),
					},
				},
				Spec: corev1.PodSpec{
					Containers: []corev1.Container{
						{
							Name:            "dwworker",
							Image:           image,
							ImagePullPolicy: imagePullPolicy,
							Command: []string{
								workerCommand,
							},
							Args: r.Obj.Spec.Args,
							Ports: []corev1.ContainerPort{
								{
									ContainerPort: workerAPIPort,
								},
							},
							// TODO: Make these configurable when necessary.
							// For example, when new versions of DwWorker has heavy computations.
							Resources: corev1.ResourceRequirements{
								Limits: corev1.ResourceList{
									corev1.ResourceCPU: resource.MustParse("1"),
								},
							},
						},
					},
				},
			},
		},
	}

	applyResult, err := k8s.ApplyWithResult(r.Ctx, r.RemoteClient, objDeploy)
	if err != nil {
		return fmt.Errorf("apply dwworker deployment failed: %w", err)
	}
	if applyResult == k8s.ApplyResultCreated || applyResult == k8s.ApplyResultPatched {
		r.Logger.Info("[remote] Applyed dwworker deployment")
	}

	objService := &corev1.Service{
		ObjectMeta: metav1.ObjectMeta{
			Name:      r.getInstanceName(),
			Namespace: r.getDataPlaneNamespace(),
		},
		Spec: corev1.ServiceSpec{
			Selector: map[string]string{
				consts.NameLabelKey:     "dwworker",
				consts.InstanceLabelKey: r.getInstanceName(),
			},
			Ports: []corev1.ServicePort{
				{
					Name: "api",
					Port: workerAPIPort,
				},
			},
		},
	}
	// The service should be auto-deleted when deployment is deleted.
	if err := controllerutil.SetOwnerReference(objDeploy, objService, r.RemoteClient.Scheme()); err != nil {
		return fmt.Errorf("set dwworker service owner failed: %w", err)
	}

	applyResult, err = k8s.ApplyWithResult(r.Ctx, r.RemoteClient, objService)
	if err != nil {
		return fmt.Errorf("apply dwworker service failed: %w", err)
	}
	if applyResult == k8s.ApplyResultCreated || applyResult == k8s.ApplyResultPatched {
		r.Logger.Info("[remote] Applyed dwworker service")
	}

	return nil
}

func (r *ReconcileContext) reconcileSelfStatus() (res ctrl.Result, err error) {
	if r.Obj.Status.Status == "" {
		r.Obj.Status = cloudv1.DWWorkerStatus{
			Status:       cloudv1.DWWorkerServiceStatusRunning,
			ErrorMessage: "",
		}
		if err := r.LocalClient.Status().Update(r.Ctx, r.Obj); err != nil {
			return ctrl.Result{}, fmt.Errorf("update status to running failed: %w", err)
		}
	}

	// Skip updating status any more for failed tasks
	if r.Obj.Status.Status == cloudv1.DWWorkerServiceStatusFatalError ||
		r.Obj.Status.Status == cloudv1.DWWorkerServiceStatusIdle {
		return ctrl.Result{}, nil
	}

	var apiClient APIClient
	switch r.Network.Spec.Provider {
	case consts.ProviderTypeEKS:
		// service exposes self via DNS with the service name, so we use r.getInstanceName() as name.
		apiClient = newAPIClientWithKubeProxy(
			r.Ctx, r.RemoteRestClient, r.getDataPlaneNamespace(), r.getInstanceName(), int(workerAPIPort))
	case consts.ProviderTypeGKE:
		// Since there is a firewall between the master and worker nodes in GKE,
		// which by default only allows ports 443 (https) and 10250 (kubelet) for master-to-node traffic,
		// it is difficult to access dwwroker status information via kubeproxy.
		// therefore, we use cluster-proxy for this purpose in GKE.
		type withProxyBaseUrl struct {
			ProxyBaseUrl string `json:"proxyBaseUrl,omitempty"`
		}
		objWithProxyBaseUrl := withProxyBaseUrl{}
		if err := json.Unmarshal(r.Network.Status.FromProvider.Raw, &objWithProxyBaseUrl); err != nil {
			return ctrl.Result{}, errors.Annotate(err, "failed to unmarshal provider specific network status")
		}
		clusterName := r.Obj.Spec.ClusterName
		dwWorkerName := r.Obj.Name
		r.Logger.Info("GKE constructing cluster proxy http link", "cluster", clusterName, "dwWorker", dwWorkerName)
		apiClient = newAPIClientWithClusterProxy(
			r.Ctx, objWithProxyBaseUrl.ProxyBaseUrl, clusterName, dwWorkerName, r.ClusterProxyAuthDecorator)
	default:
		return ctrl.Result{}, fmt.Errorf("unsupported provider type %s", r.Network.Spec.Provider)
	}

	serviceInfo, err := apiClient.GetServiceInfo()
	if err != nil {
		return ctrl.Result{}, err
	}

	lastStatus := r.Obj.Status.DeepCopy()
	r.Obj.Status = *serviceInfo.DeepCopy()

	if !equality.Semantic.DeepEqual(&r.Obj.Status, lastStatus) {
		r.Logger.Info("Update status", "newStatus", r.Obj.Status, "oldStatus", lastStatus)
		if err := r.LocalClient.Status().Update(r.Ctx, r.Obj); err != nil {
			r.Logger.Error(err, "update status failed")
			return ctrl.Result{}, fmt.Errorf("update status failed: %w", err)
		}
	}

	// When serviceStatus is Error or Idle, no reconcile is needed anymore.
	// For other serviceStatus, we need reconcile continuously.
	if r.Obj.Status.Status == cloudv1.DWWorkerServiceStatusFatalError ||
		r.Obj.Status.Status == cloudv1.DWWorkerServiceStatusIdle {
		return ctrl.Result{}, nil
	}
	return ctrl.Result{RequeueAfter: 10 * time.Second}, nil
}

func (r *ReconcileContext) reconcileDelete() error {
	r.Logger.Info("DWWorker reconcileDelete", "obj", r.Obj)

	// No need to delete the service, because it will be deleted when deployment
	// is deleted.

	r.Logger.Info("[remote] Deleting dwworker deployment")
	if err := client.IgnoreNotFound(r.RemoteClient.Delete(
		r.Ctx,
		&appsv1.Deployment{
			ObjectMeta: metav1.ObjectMeta{
				Name:      r.getInstanceName(),
				Namespace: r.getDataPlaneNamespace(),
			},
		},
	)); err != nil {
		return fmt.Errorf("delete dwworker deployment failed: %w", err)
	}

	return nil
}