package backup

import (
	"context"
	"fmt"
	"time"

	"github.com/go-logr/logr"
	"github.com/google/uuid"
	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/tools/record"
	"k8s.io/utils/ptr"
	ctrl "sigs.k8s.io/controller-runtime"
	"sigs.k8s.io/controller-runtime/pkg/builder"
	"sigs.k8s.io/controller-runtime/pkg/client"
	"sigs.k8s.io/controller-runtime/pkg/controller"
	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
	"sigs.k8s.io/controller-runtime/pkg/log"
	"sigs.k8s.io/controller-runtime/pkg/predicate"

	"github.com/tidbcloud/eks-provider/controllers/backup/task"
	"github.com/tidbcloud/eks-provider/controllers/utils/brguardian"
	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/dataplane"
	"github.com/tidbcloud/infra-provider/pkg/deploy"
	"github.com/tidbcloud/infra-provider/pkg/k8s"
	"github.com/tidbcloud/infra-provider/pkg/nodegroup"
	"github.com/tidbcloud/infra-provider/pkg/util"
	providerconfigUtil "github.com/tidbcloud/infra-provider/pkg/util/providerconfig"
)

const (
	durationKey = "cloud.pingcap.com/duration"
)

// Reconciler is the backup reconciler
type Reconciler struct {
	Logger             logr.Logger
	EventRecorder      record.EventRecorder
	Client             client.Client
	Cache              client.Reader
	CachedRemoteClient dataplane.EnvProviderManager
	Config             config.BRConfig
	RootConfig         config.RootConfig
}

// Reconcile is part of the main kubernetes reconciliation loop which aims to
// move the current state of the cluster closer to the desired state.
func (r *Reconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
	logger := log.FromContext(ctx).
		WithValues(
			"reconcile-id", uuid.New().String(),
			"name", req.NamespacedName,
		)
	ctx = log.IntoContext(ctx, logger)
	logger.Info("Reconciling Backup")

	backup := &cloudv1.Backup{}
	if err := r.Cache.Get(ctx, req.NamespacedName, backup); err != nil {
		if apierrors.IsNotFound(err) {
			return ctrl.Result{}, nil
		}
		return ctrl.Result{}, fmt.Errorf("unable to get backup %s: %v", req.NamespacedName, err)
	}

	if ptr.Deref(backup.Spec.Paused, false) {
		logger.Info("Skipping paused backup", "backup", req.NamespacedName)
		return ctrl.Result{}, nil
	}

	cluster := &cloudv1.Cluster{
		ObjectMeta: metav1.ObjectMeta{
			Name:      backup.Spec.ClusterName,
			Namespace: backup.Namespace,
		},
	}
	if err := r.Cache.Get(ctx, client.ObjectKeyFromObject(cluster), cluster); err != nil {
		if !apierrors.IsNotFound(err) {
			return ctrl.Result{}, fmt.Errorf("unable to get cluster %s/%s: %v", cluster.Namespace, cluster.Name, err)
		}
		logger.Info("cluster not found, marking backup as failed or removing finalizer", "cluster", cluster.Name)
		return r.handleClusterGone(ctx, backup, logger)
	}

	remote, err := r.CachedRemoteClient.GetClient(ctx, cluster)
	if err != nil {
		return ctrl.Result{}, fmt.Errorf("failed to get dataplane k8s client: %v", err)
	}

	// Get EnvProvider for helm operations (CRRF daemon)
	envProvider, envProviderErr := r.CachedRemoteClient.GetEnvProvider(ctx, cluster)

	backupStorageURL := ptr.Deref(backup.Spec.Storage, "")
	if backupStorageURL == "" {
		return ctrl.Result{}, fmt.Errorf("backup storage is empty")
	}

	providerType := cluster.GetProviderType()
	nodeVersion, err := nodegroup.GetTiDBNodeVersion(ctx, cluster, providerType)
	if err != nil {
		return ctrl.Result{}, fmt.Errorf("failed to get node version: %w", err)
	}

	var cloud task.CloudProvider
	switch providerType {
	case consts.ProviderTypeEKS:
		region := backup.GetRegion()
		assumeRoleARN := util.GetRoleARN(cluster.Spec.Kubernetes.Account, r.RootConfig.FullManage.AssumeRoleName)
		externalID := r.RootConfig.FullManage.AssumeRoleExternalID
		cloud, err = task.NewAWSProvider(backupStorageURL, region, cluster.Spec.Kubernetes.Name, nodeVersion, cluster.Status.NodeIdentity.AWS.RoleARN, cluster.Spec.Kubernetes.Account, assumeRoleARN, externalID, providerconfigUtil.GetClusterProviderConfigNameFromCluster(cluster), cluster.GetTenantID(), cluster.GetProjectID(), cluster.Name, *cluster.Spec.Kubernetes.Cloud.AWS, &r.RootConfig)
		if err != nil {
			return ctrl.Result{}, fmt.Errorf("failed to build AWS provider: %v", err)
		}
	case consts.ProviderTypeGKE:
		region := backup.GetRegion()
		projectID := cluster.Spec.Kubernetes.Account
		nodeRole := cluster.Status.NodeIdentity.GCP.ServiceAccount
		cloud, err = task.NewGCPProvider(backupStorageURL, region, projectID, cluster.Spec.Kubernetes.Name,
			nodeVersion, nodeRole, providerconfigUtil.GetClusterProviderConfigNameFromCluster(cluster), cluster.GetTenantID(), cluster.GetProjectID())
		if err != nil {
			return ctrl.Result{}, fmt.Errorf("failed to build GCP provider: %v", err)
		}
	case consts.ProviderTypeAKS:
		tenantID := r.Config.Azure.CrossplaneAzureProviderTenantID
		subscriptionID := r.Config.Azure.CrossplaneAzureProviderSubscriptionID
		clientID := r.Config.Azure.CrossplaneAzureProviderClientID

		azure := cluster.Spec.Kubernetes.Cloud.Azure
		if azure == nil {
			return ctrl.Result{}, fmt.Errorf("no aks info found in cluster %s/%s", cluster.Namespace, cluster.Name)
		}
		cloud, err = task.NewAzureProvider(tenantID, subscriptionID, clientID,
			cluster.Spec.Kubernetes.Account, cluster.Spec.Kubernetes.Name,
			nodeVersion, *azure, backupStorageURL, providerconfigUtil.GetClusterProviderConfigNameFromCluster(cluster), cluster.GetTenantID(), cluster.GetProjectID())
		if err != nil {
			return ctrl.Result{}, fmt.Errorf("failed to build Azure provider: %v", err)
		}
	default:
		return ctrl.Result{}, fmt.Errorf("unsupported provider type %s", providerType)
	}
	c := task.NewContext(r.EventRecorder, r.Logger, r.Client, remote, backup, cluster, r.Config.Env, cloud)
	c.WithBRConfig(&r.Config)
	if envProviderErr == nil && envProvider != nil {
		c.WithEnvProvider(envProvider)
	}
	tasks := []deploy.Task{
		task.NewGuardian(c),
		task.NewNodePool(c),
		task.NewBackup(c),
	}
	// Add CRRF task for log backups with CRRF annotations
	if backup.Spec.Mode != nil && *backup.Spec.Mode == cloudv1.BackupModeLog {
		_, hasSpec := backup.Annotations[cloudv1.BackupAnnotationKeyCRRFSpec]
		_, hasStatus := backup.Annotations[cloudv1.BackupAnnotationKeyCRRFStatus]
		if hasSpec || hasStatus {
			tasks = append(tasks, task.NewCRRF(c))
		}
	}
	deployer := deploy.NewDeployer(tasks, nil, logger)
	deleted := !backup.DeletionTimestamp.IsZero()
	var res ctrl.Result
	var errs error
	var retriableErrs error
	if deleted {
		results := deployer.Destroy(ctx)
		res, errs, retriableErrs = brguardian.CheckpointDestroy(backup, results, time.Now())
		if errs == nil && retriableErrs == nil {
			if err := r.Client.Update(ctx, backup); err != nil {
				r.EventRecorder.Event(backup, corev1.EventTypeWarning, "ReconcileFailed", err.Error())
				return res, err
			}
		}
		// cloudv1.Backup does not have deletion status, so no need to update status
	} else {
		results := deployer.Deploy(ctx)
		res, errs, retriableErrs = deploy.MinRequeue(results)
		if errs == nil && retriableErrs == nil { // backup complete or failed, resources can be released once the status info is collected by dataflow-service
			duration, ok := backup.Annotations[durationKey]
			if !ok {
				duration = time.Since(backup.CreationTimestamp.Time).String()
				if backup.Annotations == nil {
					backup.Annotations = make(map[string]string)
				}
				backup.Annotations[durationKey] = duration
			}
			logger.Info("Backup finished, it can be deleted to release resources", "backup", req.NamespacedName, "cluster", backup.Spec.ClusterName, "duration", duration, "phase", backup.Status.Phase)
		}
		// Backup.Status field is a subresource
		// The Status field will lost when updating Backup.Spec
		// So we need to deepcopy Backup.Status before updating Backup.Spec
		// and re-assign it to Backup when updating Backup.Status
		backupStatus := backup.Status.DeepCopy()
		if err := r.Client.Update(ctx, backup); err != nil {
			r.EventRecorder.Event(backup, corev1.EventTypeWarning, "ReconcileFailed", err.Error())
			return res, err
		}
		backup.Status = *backupStatus
		if err := r.Client.Status().Update(ctx, backup); err != nil {
			r.EventRecorder.Event(backup, corev1.EventTypeWarning, "ReconcileFailed", err.Error())
			return res, err
		}
	}
	r.EventRecorder.Event(backup, corev1.EventTypeNormal, "ReconcileSuccess", "")
	if retriableErrs != nil {
		log.FromContext(ctx).Info("retriable error", "error", retriableErrs)
	}
	return res, errs
}

// handleClusterGone is called when the referenced cluster no longer exists.
// It releases a deleting Backup, otherwise it marks a nonterminal Backup as failed.
func (r *Reconciler) handleClusterGone(ctx context.Context, backup *cloudv1.Backup, logger logr.Logger) (ctrl.Result, error) {
	if !backup.DeletionTimestamp.IsZero() {
		controllerutil.RemoveFinalizer(backup, task.Finalizer)
		if err := r.Client.Update(ctx, backup); err != nil {
			logger.Error(err, "failed to remove finalizer after cluster gone")
			return ctrl.Result{}, fmt.Errorf("failed to remove finalizer: %v", err)
		}
		return ctrl.Result{}, nil
	}

	switch backup.Status.Phase {
	case cloudv1.BackupComplete, cloudv1.BackupFailed, cloudv1.BackupStopped, cloudv1.BackupTruncated:
		return ctrl.Result{}, nil
	default:
		backup.Status.Phase = cloudv1.BackupFailed
		if err := r.Client.Status().Update(ctx, backup); err != nil {
			logger.Error(err, "failed to update backup status after cluster gone")
			return ctrl.Result{}, fmt.Errorf("failed to update backup status: %v", err)
		}
		return ctrl.Result{}, nil
	}
}

func (r *Reconciler) SetupWithManager(mgr ctrl.Manager, preds ...predicate.Predicate) error {
	return ctrl.NewControllerManagedBy(mgr).
		For(&cloudv1.Backup{}, builder.WithPredicates(preds...)).
		WithOptions(controller.Options{RateLimiter: k8s.RetryLimiter, MaxConcurrentReconciles: 10}).
		Complete(r)
}