package keyspace

import (
	"context"
	"errors"
	"os"
	"strconv"
	"time"

	serverlessv1alpha1 "github.com/tidbcloud/aws-shared-provider/apis/serverless/v1alpha1"
	"github.com/tidbcloud/aws-shared-provider/pkg/manager"
	"github.com/tidbcloud/aws-shared-provider/pkg/metrics"
	"github.com/tidbcloud/aws-shared-provider/pkg/utils"
	"github.com/tidbcloud/aws-shared-provider/pkg/utils/pd"
	cloudv1 "github.com/tidbcloud/infra-api/api/v1"
	infra "github.com/tidbcloud/tidb-management-service/api/serverless_infra"
	infraclient "github.com/tidbcloud/tidb-management-service/api/serverless_infra/client"
	"github.com/tidbcloud/tidb-management-service/api/serverless_infra/meta"
	corev1 "k8s.io/api/core/v1"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
	"k8s.io/client-go/tools/record"
	ctrl "sigs.k8s.io/controller-runtime"
	"sigs.k8s.io/controller-runtime/pkg/client"
	"sigs.k8s.io/controller-runtime/pkg/controller"
	"sigs.k8s.io/controller-runtime/pkg/log"
	ctrlmanager "sigs.k8s.io/controller-runtime/pkg/manager"
	"sigs.k8s.io/controller-runtime/pkg/predicate"
)

const (
	MaxConcurrentReconciles = 15

	defaultWaitInterval = 5 * time.Second
	metricsSyncInterval = 15 * time.Second
)

// Reconciler reconciles a Keyspace object.
type Reconciler struct {
	Client          client.Client
	InfraClusterCli infraclient.Client
	InfraClient     client.Client

	PdClient pd.HttpPdClient

	ActivateTidbImage string
	Namespace         string

	Recorder record.EventRecorder

	EnableTLS bool
}

// 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)
	logger.V(2).Info("start reconcile keyspace")
	start := time.Now()
	defer func() {
		logger.V(2).Info("finish reconcile keyspace", "time", time.Since(start))
	}()

	keyspace := &serverlessv1alpha1.Keyspace{}
	if err := r.Client.Get(ctx, req.NamespacedName, keyspace); err != nil {
		if apierrors.IsNotFound(err) {
			return ctrl.Result{}, nil
		}
		return utils.ReconcileErr(logger, "Keyspace", req.Name, err)
	}

	if keyspace.DeletionTimestamp != nil {
		if keyspace.Status.BindCluster != "" && keyspace.Status.ClusterNs != "" {
			if keyspace.Status.BindCRType == serverlessv1alpha1.TypeBranch {
				branch := &cloudv1.Branch{}
				if err := r.InfraClient.Get(ctx, client.ObjectKey{
					Name:      keyspace.Status.BindCluster,
					Namespace: keyspace.Status.ClusterNs,
				}, branch); err == nil {
					logger.Info("waiting infra branch to delete", "branch", keyspace.Status.BindCluster)
					return ctrl.Result{Requeue: true, RequeueAfter: defaultWaitInterval}, nil
				} else if !apierrors.IsNotFound(err) {
					return utils.ReconcileErr(logger, "Keyspace", keyspace.Name, err)
				}
			} else {
				cluster := &infra.Cluster{ObjectMeta: meta.ObjectMeta{
					Name:      keyspace.Status.BindCluster,
					Namespace: keyspace.Status.ClusterNs,
				}}
				if err := r.InfraClusterCli.Get(ctx, cluster); err == nil {
					logger.Info("waiting infra cluster to delete", "cluster", keyspace.Status.BindCluster)
					return ctrl.Result{Requeue: true, RequeueAfter: defaultWaitInterval}, nil
				} else if !errors.Is(err, infraclient.ErrNotFound) {
					return utils.ReconcileErr(logger, "Keyspace", keyspace.Name, err)
				}
			}
		}

		if err := r.removeKeyspace(ctx, keyspace.Spec.KeyspaceName); err != nil {
			r.Recorder.Event(keyspace, corev1.EventTypeWarning, "remove keyspace", err.Error())
			return utils.ReconcileErr(logger, "Keyspace", keyspace.Name, err)
		}
		if err := r.removeFinalizer(ctx, keyspace); err != nil {
			return utils.ReconcileErr(logger, "Keyspace", keyspace.Name, err)
		}

		return ctrl.Result{}, nil
	}

	if err := r.ensureFinalizer(ctx, keyspace); err != nil {
		return utils.ReconcileErr(logger, "Keyspace", keyspace.Name, err)
	}

	if keyspace.Status.Phase == serverlessv1alpha1.PendingPhase {
		keyspace.Status.Phase = serverlessv1alpha1.CreatingPhase
		if err := r.Client.Status().Update(ctx, keyspace); err != nil {
			return utils.ReconcileErr(logger, "Keyspace", keyspace.Name, err)
		}
	}
	if keyspace.Status.Phase == serverlessv1alpha1.CreatingPhase {
		keyspaceMeta, err := r.createKeyspace(ctx, keyspace)
		if err != nil {
			r.Recorder.Event(keyspace, corev1.EventTypeWarning, "create keyspace", err.Error())
			return utils.ReconcileErr(logger, "Keyspace", keyspace.Name, err)
		}
		if keyspaceMeta.State != keyspaceEnabled {
			logger.Info("keyspace not enabled, delete it!", "keyspace", keyspace.Spec.KeyspaceName)
			return utils.ReconcileErr(logger, "Keyspace", keyspace.Name, r.Client.Delete(ctx, keyspace))
		}
		r.Recorder.Event(keyspace, corev1.EventTypeNormal, "create keyspace", "create keyspace success")
		keyspace.Status.Phase = serverlessv1alpha1.ActivatingPhase
		keyspace.Status.KeyspaceId = keyspaceMeta.Id
		if err := r.Client.Status().Update(ctx, keyspace); err != nil {
			return utils.ReconcileErr(logger, "Keyspace", keyspace.Name, err)
		}
	}
	if keyspace.Status.Phase == serverlessv1alpha1.ActivatingPhase {
		if err := r.activateKeyspace(ctx, keyspace); err != nil {
			r.Recorder.Event(keyspace, corev1.EventTypeWarning, "activate keyspace", err.Error())
			return utils.ReconcileErr(logger, "Keyspace", keyspace.Name, err)
		}
		r.Recorder.Event(keyspace, corev1.EventTypeNormal, "activate keyspace", "activate keyspace success")
		keyspace.Status.Phase = serverlessv1alpha1.AvailablePhase
		if err := r.Client.Status().Update(ctx, keyspace); err != nil {
			return utils.ReconcileErr(logger, "Keyspace", keyspace.Name, err)
		}
	}

	return ctrl.Result{}, nil
}

func (r *Reconciler) runMetricsUpdater(ctx context.Context) error {
	logger := log.FromContext(ctx).WithName("keyspace-metrics").WithValues("namespace", r.Namespace)
	ticker := time.NewTicker(metricsSyncInterval)
	defer ticker.Stop()

	for {
		list := &serverlessv1alpha1.KeyspaceList{}
		if err := r.Client.List(ctx, list,
			client.InNamespace(r.Namespace),
			client.MatchingFields{"status.phase": string(serverlessv1alpha1.AvailablePhase)}); err != nil {
			logger.Error(err, "failed to sync keyspace available metrics")
		} else {
			metrics.KeyspaceAvailable.WithLabelValues(r.Namespace).Set(float64(len(list.Items)))
		}

		select {
		case <-ctx.Done():
			return nil
		case <-ticker.C:
		}
	}
}

// SetupWithManager sets up the controller with the Manager.
func (r *Reconciler) SetupWithManager(mgr manager.Manager) error {
	r.Recorder = mgr.GetEventRecorderFor("Keyspace")

	byNamespaceFilter := func(object client.Object) bool {
		return object.GetNamespace() == r.Namespace
	}
	if err := ctrl.NewControllerManagedBy(mgr).Named("Keyspace-" + r.Namespace).
		For(&serverlessv1alpha1.Keyspace{}).
		WithEventFilter(predicate.NewPredicateFuncs(byNamespaceFilter)).
		WithOptions(controller.Options{MaxConcurrentReconciles: r.maxConcurrentReconciles()}).
		Complete(r); err != nil {
		return err
	}

	return mgr.Add(ctrlmanager.RunnableFunc(r.runMetricsUpdater))
}

func (r *Reconciler) maxConcurrentReconciles() int {
	maxConcurrent := MaxConcurrentReconciles
	if v := os.Getenv("MAX_CONCURRENT_KEYSPACES"); v != "" {
		if n, err := strconv.Atoi(v); err == nil && n > 0 {
			maxConcurrent = n
		}
	}
	return maxConcurrent
}