package throttler

import (
	"context"
	"fmt"
	"time"

	"github.com/go-logr/logr"
	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/errors"
	"k8s.io/client-go/tools/record"
	ctrl "sigs.k8s.io/controller-runtime"
	"sigs.k8s.io/controller-runtime/pkg/cache"
	"sigs.k8s.io/controller-runtime/pkg/client"
	"sigs.k8s.io/controller-runtime/pkg/controller"
	"sigs.k8s.io/controller-runtime/pkg/handler"
	"sigs.k8s.io/controller-runtime/pkg/manager"
	"sigs.k8s.io/controller-runtime/pkg/predicate"
	"sigs.k8s.io/controller-runtime/pkg/source"

	"github.com/pingcap/tidb-operator/pkg/apis/pingcap/v1alpha1"
	"github.com/tidbcloud/shared-provider/pkg/utils/netns"
)

const (
	// TODO: use tidb-operator api.
	labelKeyComponent = "app.kubernetes.io/component"
	labelKeyManagedBy = "app.kubernetes.io/managed-by"
	labelKeyInstance  = "app.kubernetes.io/instance"
	labelKeyName      = "app.kubernetes.io/name"

	labelValTiDBOperator = "tidb-operator"
	labelValTiDB         = "tidb"
	labelValTiDBCluster  = "tidb-cluster"

	defaultInterface = "eth0"
)

// Reconciler reconciles a Network object.
type Reconciler struct {
	Logger logr.Logger
	Client client.Client

	Cache cache.Cache

	Recorder record.EventRecorder

	NodeName string

	t Throttler
}

//+kubebuilder:rbac:groups=pingcap.com,resources=tidbclusters,verbs=get;list;watch
//+kubebuilder:rbac:groups=core,resources=pods,verbs=get;list;watch

// 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 := r.Logger.WithValues("namespace", req.Namespace, "name", req.Name)
	logger.Info("Start reconcile")

	startTime := time.Now()
	// TODO(liubo02): combine two defer log
	defer func() {
		logger.Info("Finish reconcile",
			"time", time.Since(startTime),
		)
	}()

	tc := v1alpha1.TidbCluster{}
	if err := r.Cache.Get(ctx, req.NamespacedName, &tc); err != nil {
		if apierrors.IsNotFound(err) {
			return ctrl.Result{}, nil
		}

		return ctrl.Result{}, fmt.Errorf("can't get tidbcluster %s: %w", req.NamespacedName, err)
	}

	pl := corev1.PodList{}

	if err := r.Cache.List(context.Background(), &pl,
		client.InNamespace(tc.Namespace),
		client.MatchingLabels(map[string]string{
			labelKeyManagedBy: labelValTiDBOperator,
			labelKeyComponent: labelValTiDB,
			labelKeyName:      labelValTiDBCluster,
			labelKeyInstance:  tc.Name,
		}),
		// client.MatchingFields(map[string]string{
		// 	"spec.nodeName": r.NodeName,
		// }),
	); err != nil {
		return ctrl.Result{}, fmt.Errorf("can't list tidb pods managed by tidbcluster %s: %w", req.NamespacedName, err)
	}

	logger.WithValues("podNum", len(pl.Items)).Info("wait for throttle config syncing")

	cfg, err := NewThrottleConfig(tc.Annotations)
	if err != nil {
		r.Recorder.Eventf(&tc,
			corev1.EventTypeWarning,
			"SyncBandwidthThrottle",
			"invalid throttle config for %s/%s: %v",
			tc.Namespace, tc.Name, err,
		)

		// can't sync until fix the throttle config manually
		return ctrl.Result{}, nil
	}

	errList := []error{}
	for i := range pl.Items {
		p := &pl.Items[i]

		logger.WithValues("pod", p.Name).Info("sync throttle config")

		r.Recorder.Eventf(&tc,
			corev1.EventTypeNormal,
			"SyncBandwidthThrottle",
			"sync pod %s bandwidth throttle config",
			p.Name,
		)

		if err := r.t.Throttle(p, cfg); err != nil {
			errList = append(errList, err)
		}
	}

	if len(errList) != 0 {
		return ctrl.Result{}, errors.NewAggregate(errList)
	}

	return ctrl.Result{}, nil
}

// SetupWithManager sets up the controller with the Manager.
func (r *Reconciler) SetupWithManager(mgr manager.Manager) error {
	r.Client = mgr.GetClient()
	r.Cache = mgr.GetCache()
	r.Recorder = mgr.GetEventRecorderFor("Throttler")

	c, err := controller.New("Throttler", mgr, controller.Options{
		Reconciler: r,
	})
	if err != nil {
		return fmt.Errorf("can't new a controller: %w", err)
	}

	r.Logger = c.GetLogger()
	cr, err := netns.NewDockerRuntime()
	if err != nil {
		return fmt.Errorf("can't new docker runtime: %w", err)
	}
	r.t = NewThrottler(cr, defaultInterface)

	return r.watches(c)
}

func (r *Reconciler) watches(c controller.Controller) error {
	if err := c.Watch(
		source.NewKindWithCache(&v1alpha1.TidbCluster{}, r.Cache),
		&handler.Funcs{
			CreateFunc: r.handleCreateTidbClusterEvent,
			UpdateFunc: r.handleUpdateTidbClusterEvent,
		},
	); err != nil {
		return fmt.Errorf("can't watch tidbcluster: %w", err)
	}

	p, err := predicate.LabelSelectorPredicate(metav1.LabelSelector{
		MatchLabels: map[string]string{
			labelKeyManagedBy: labelValTiDBOperator,
			labelKeyName:      labelValTiDBCluster,
			labelKeyComponent: labelValTiDB,
		},
	})
	if err != nil {
		return fmt.Errorf("invalid label selector for tidb pod: %w", err)
	}

	if err := c.Watch(
		source.NewKindWithCache(&corev1.Pod{}, r.Cache),
		&handler.Funcs{
			CreateFunc: r.handleCreatePodEvent,
			UpdateFunc: r.handleUpdatePodEvent,
		},
		p,
	); err != nil {
		return fmt.Errorf("can't watch pods: %w", err)
	}

	return nil
}