package keyspacepoolconfig

import (
	"context"
	"fmt"
	"strconv"
	"time"

	serverlessv1alpha1 "github.com/tidbcloud/aws-shared-provider/apis/serverless/v1alpha1"
	"github.com/tidbcloud/aws-shared-provider/pkg/manager"
	corev1 "k8s.io/api/core/v1"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/util/wait"
	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/event"
	"sigs.k8s.io/controller-runtime/pkg/handler"
	"sigs.k8s.io/controller-runtime/pkg/log"
	"sigs.k8s.io/controller-runtime/pkg/predicate"
)

const (
	KeyspacePoolConfigName     = "keyspace-pool-config"
	KeyspacePoolSizeKey        = "size"
	KeyspacePoolMaxPendingKey  = "max-pending"
	defaultMaxPending          = 100
	defaultCreateRatePerSecond = 0

	KeyspacePoolCreateRatePerSecondKey = "create-rate-per-second"
	// Ignore tiny scheduler jitter; only log when we actually waited for a token.
	rateLimitedLogThreshold = 50 * time.Millisecond
)

type poolConfig struct {
	poolSize            int
	maxPending          int
	createRatePerSecond float64
}

// Reconciler reconciles a Keyspace pool config object.
type Reconciler struct {
	Client          client.Client
	Namespace       string
	KubeClusterName string
	Encrypt         bool

	CreateRateStore *CreateRateStore

	rateLimiter *createRateLimiter
}

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

	keyspacePoolCm := &corev1.ConfigMap{}
	if err := r.Client.Get(ctx, req.NamespacedName, keyspacePoolCm); err != nil {
		return ctrl.Result{}, err
	}
	createRatePerSecond, ok := r.getCreateRatePerSecond()
	if !ok {
		createRatePerSecond = defaultCreateRatePerSecond
	}
	poolCfg, err := parseKeyspacePoolConfig(keyspacePoolCm, createRatePerSecond)
	if err != nil {
		return ctrl.Result{}, err
	}

	keyspacesInPool := &serverlessv1alpha1.KeyspaceList{}
	if err := r.Client.List(ctx, keyspacesInPool,
		client.InNamespace(r.Namespace),
		client.MatchingFields{"status.inPool": "true"}); err != nil {
		return ctrl.Result{}, err
	}

	inPool := len(keyspacesInPool.Items)
	pending, creating, activating := 0, 0, 0
	for i := range keyspacesInPool.Items {
		switch keyspacesInPool.Items[i].Status.Phase {
		case serverlessv1alpha1.PendingPhase:
			pending++
		case serverlessv1alpha1.CreatingPhase:
			creating++
		case serverlessv1alpha1.ActivatingPhase:
			activating++
		}
	}

	inProgress := pending + creating + activating
	size := keyspacesToCreate(poolCfg.poolSize, inPool, inProgress, poolCfg.maxPending)
	logger.V(1).Info("keyspace pool",
		"size", poolCfg.poolSize,
		"in pool", inPool,
		"pending", pending,
		"creating", creating,
		"activating", activating,
		"max pending", poolCfg.maxPending,
		"create-rate-per-second", poolCfg.createRatePerSecond,
		"to create", size)

	var lastCreated *serverlessv1alpha1.Keyspace
	limiter := r.getRateLimiter()
	for i := 0; i < size; i++ {
		waitStart := time.Now()
		if err := limiter.wait(ctx, poolCfg.createRatePerSecond); err != nil {
			return ctrl.Result{}, err
		}
		waited := time.Since(waitStart)
		// Log only when the limiter actually delayed creation so operators can
		// distinguish real throttling from the unlimited path.
		if poolCfg.createRatePerSecond > 0 && waited >= rateLimitedLogThreshold {
			logger.Info("keyspace pool create rate limited",
				"namespace", r.Namespace,
				"create-rate-per-second", poolCfg.createRatePerSecond,
				"waited", waited)
		}
		keyspace := &serverlessv1alpha1.Keyspace{
			ObjectMeta: metav1.ObjectMeta{
				GenerateName: "kt-",
				Namespace:    r.Namespace,
			},
			Spec: serverlessv1alpha1.KeyspaceSpec{
				KeyspaceName: genKeyspaceToken(r.Namespace, r.KubeClusterName),
				Encrypt:      r.Encrypt,
			},
		}
		if err := r.Client.Create(ctx, keyspace); err != nil {
			return ctrl.Result{}, err
		}
		lastCreated = keyspace
	}
	if lastCreated != nil {
		// Wait until the informer cache observes the last create so a follow-up
		// reconcile does not refill the same pool shortfall.
		//nolint:gomnd // cache synchronization timeout
		if err := wait.PollUntilContextTimeout(ctx, time.Second, 8*time.Second, true,
			func(ctx context.Context) (bool, error) {
				err := r.Client.Get(ctx, client.ObjectKeyFromObject(lastCreated), &serverlessv1alpha1.Keyspace{})
				if apierrors.IsNotFound(err) {
					return false, nil
				}
				return err == nil, err
			},
		); err != nil {
			return ctrl.Result{}, err
		}
	}

	return ctrl.Result{}, nil
}

// size and max-pending remain ConfigMap-backed; create rate is supplied by the
// runtime store so reconciles do not need to re-parse that field.
func parseKeyspacePoolConfig(configMap *corev1.ConfigMap, createRatePerSecond float64) (poolConfig, error) {
	poolSize, err := strconv.Atoi(configMap.Data[KeyspacePoolSizeKey])
	if err != nil {
		return poolConfig{}, fmt.Errorf("parse %s: %w", KeyspacePoolSizeKey, err)
	}

	maxPending := defaultMaxPending
	value, ok := configMap.Data[KeyspacePoolMaxPendingKey]
	if !ok {
	} else {
		maxPending, err = strconv.Atoi(value)
		if err != nil {
			return poolConfig{}, fmt.Errorf("parse %s: %w", KeyspacePoolMaxPendingKey, err)
		}
		if maxPending <= 0 {
			return poolConfig{}, fmt.Errorf("%s must be greater than 0", KeyspacePoolMaxPendingKey)
		}
	}

	return poolConfig{
		poolSize:            poolSize,
		maxPending:          maxPending,
		createRatePerSecond: createRatePerSecond,
	}, nil
}

// parseCreateRatePerSecond is used only when hydrating or refreshing the in-memory store.
func parseCreateRatePerSecond(configMap *corev1.ConfigMap) (float64, error) {
	value, ok := configMap.Data[KeyspacePoolCreateRatePerSecondKey]
	if !ok || value == "" {
		return defaultCreateRatePerSecond, nil
	}

	createRatePerSecond, err := strconv.ParseFloat(value, 64)
	if err != nil {
		return 0, fmt.Errorf("parse %s: %w", KeyspacePoolCreateRatePerSecondKey, err)
	}
	if createRatePerSecond < 0 {
		return 0, fmt.Errorf("%s must be greater than or equal to 0", KeyspacePoolCreateRatePerSecondKey)
	}
	return createRatePerSecond, nil
}

func keyspacesToCreate(poolSize, inPool, inProgress, maxPending int) int {
	if inPool >= poolSize || inProgress >= maxPending {
		return 0
	}
	return min(poolSize-inPool, maxPending-inProgress)
}

func (r *Reconciler) getRateLimiter() *createRateLimiter {
	if r.rateLimiter == nil {
		r.rateLimiter = newCreateRateLimiter()
	}
	return r.rateLimiter
}

func (r *Reconciler) getCreateRatePerSecond() (float64, bool) {
	if r.CreateRateStore == nil {
		return defaultCreateRatePerSecond, false
	}
	return r.CreateRateStore.Get()
}

// ConfigMap updates are the persistence layer; this keeps the in-memory rate in sync.
func (r *Reconciler) syncCreateRateFromConfigMap(configMap *corev1.ConfigMap) error {
	createRatePerSecond, err := parseCreateRatePerSecond(configMap)
	if err != nil {
		return err
	}
	if r.CreateRateStore != nil {
		r.CreateRateStore.Set(createRatePerSecond)
	}
	return nil
}

func (r *Reconciler) isManagedConfigMap(object client.Object) bool {
	return object != nil &&
		object.GetNamespace() == r.Namespace &&
		object.GetName() == KeyspacePoolConfigName
}

func (r *Reconciler) syncCreateRateFromObject(object client.Object) error {
	configMap, ok := object.(*corev1.ConfigMap)
	if !ok {
		return nil
	}
	return r.syncCreateRateFromConfigMap(configMap)
}

func (r *Reconciler) configMapPredicate() predicate.Predicate {
	logger := ctrl.Log.WithName("controllers").WithName("KeyspacePoolConfig").WithName(r.Namespace)
	return predicate.Funcs{
		CreateFunc: func(e event.CreateEvent) bool {
			if !r.isManagedConfigMap(e.Object) {
				return false
			}
			// Sync the runtime rate before the reconcile request runs.
			if err := r.syncCreateRateFromObject(e.Object); err != nil {
				logger.Error(err, "failed to sync create rate from configmap create event")
			}
			return true
		},
		UpdateFunc: func(e event.UpdateEvent) bool {
			if !r.isManagedConfigMap(e.ObjectNew) {
				return false
			}
			// Sync the runtime rate before the reconcile request runs.
			if err := r.syncCreateRateFromObject(e.ObjectNew); err != nil {
				logger.Error(err, "failed to sync create rate from configmap update event")
			}
			return true
		},
		DeleteFunc: func(e event.DeleteEvent) bool {
			return r.isManagedConfigMap(e.Object)
		},
		GenericFunc: func(e event.GenericEvent) bool {
			if !r.isManagedConfigMap(e.Object) {
				return false
			}
			// Sync the runtime rate before the reconcile request runs.
			if err := r.syncCreateRateFromObject(e.Object); err != nil {
				logger.Error(err, "failed to sync create rate from configmap generic event")
			}
			return true
		},
	}
}

// SetupWithManager sets up the controller with the Manager.
func (r *Reconciler) SetupWithManager(mgr manager.Manager) error {
	byNamespaceFuncs := predicate.NewPredicateFuncs(func(object client.Object) bool {
		return object.GetNamespace() == r.Namespace
	})
	return ctrl.NewControllerManagedBy(mgr).Named("KeyspacePoolConfig-"+r.Namespace).
		For(&corev1.ConfigMap{}, builder.WithPredicates(r.configMapPredicate())).
		WithEventFilter(byNamespaceFuncs).
		Watches(&serverlessv1alpha1.Keyspace{}, &handler.Funcs{
			CreateFunc: r.handleCreateConfigEvent,
			UpdateFunc: r.handleUpdateConfigEvent,
			DeleteFunc: r.handleDeleteConfigEvent,
		}).
		Complete(r)
}