package component

import (
	"context"

	"github.com/aws/aws-sdk-go-v2/config"
	"github.com/pingcap/errors"
	"github.com/pingcap/log"
	"github.com/tidbcloud/infra-provider/regional-server/api/v1alpha1"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/cloud/aliyun"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/cloud/aws"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/cloud/azure"
	rsconfig "github.com/tidbcloud/infra-provider/regional-server/pkg/config"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/consts"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/controllers"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/controllers/appbox"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/controllers/bucket"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/controllers/dataplane"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/controllers/internet"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/controllers/key"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/controllers/loadbalancer"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/controllers/nodepool"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/controllers/privatelinkendpoint"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/controllers/privatelinkservice"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/controllers/resourcetagger"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/controllers/vpc"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/controllers/vpcpeering"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/metrics"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/metrics/objectstorage"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/providerconfig"
	"go.uber.org/zap"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/runtime"
	"k8s.io/client-go/dynamic"
	"k8s.io/client-go/rest"
	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/healthz"
	metricsserver "sigs.k8s.io/controller-runtime/pkg/metrics/server"
	"sigs.k8s.io/controller-runtime/pkg/predicate"
)

type Controller struct {
	cfg *ControllerOptions
}

func NewController(options *ControllerOptions) Component {
	return &Controller{
		cfg: options,
	}
}

type ControllerOptions struct {
	KubeConfig             *rest.Config
	Config                 rsconfig.RootConfig
	TmplConfig             rsconfig.TemplatesConfig
	Scheme                 *runtime.Scheme
	HealthProbeBindAddress string
	MetricsAddress         string
	LeaderElection         bool
	EnabledControllers     map[string]bool
}

func (h *Controller) Start(ctx context.Context) error {
	cs, err := buildCacheSelector(h.cfg.Config)
	if err != nil {
		log.Fatal("unable to build cache selector", zap.Error(err))
	}
	mgr, err := ctrl.NewManager(h.cfg.KubeConfig, ctrl.Options{
		Scheme:                 h.cfg.Scheme,
		HealthProbeBindAddress: h.cfg.HealthProbeBindAddress,
		Metrics:                metricsserver.Options{BindAddress: h.cfg.MetricsAddress},
		LeaderElection:         h.cfg.LeaderElection,
		LeaderElectionID:       "fc66da15.dedicated.regionalserver.tidbcloud.com",
		// Only load objects to local cache if it matches the label selector.
		Cache: cache.Options{ByObject: cs},
	})
	if err != nil {
		return errors.Trace(err)
	}

	rsmanager := controllers.NewManager(mgr.GetClient(), h.cfg.Scheme)
	go func() {
		err := <-rsmanager.Errors()
		log.Fatal("init mutli-k8s client manager error", zap.Error(err))
	}()

	h.startControllers(ctx, mgr, rsmanager)
	if err := h.startObjectStorageMetrics(ctx); err != nil {
		return errors.Trace(err)
	}
	if err := mgr.Start(ctx); err != nil {
		return errors.Trace(err)
	}
	return nil
}

func (h *Controller) startObjectStorageMetrics(ctx context.Context) error {
	dynamicClient, err := dynamic.NewForConfig(h.cfg.KubeConfig)
	if err != nil {
		return errors.Trace(err)
	}
	runner, err := objectstorage.NewRunner(ctx, h.cfg.Config.ObjectStorageMetrics, dynamicClient)
	if err != nil {
		return errors.Trace(err)
	}
	if runner == nil {
		return nil
	}
	metrics.MustRegister(runner.Collector())
	runner.Start(ctx)
	return nil
}

func (h *Controller) Name() string {
	return "controller"
}

func (h *Controller) startControllers(ctx context.Context, mgr ctrl.Manager, rsmanager *controllers.Manager) {
	rootConf := h.cfg.Config
	tmplConfig := h.cfg.TmplConfig
	provider := rootConf.Provider

	var awsCli aws.Client
	var aliyunCli aliyun.Client
	var azureCli azure.Client
	var pfInitializer providerconfig.Initializer
	var err error
	switch provider {
	case consts.ProviderTypeEKS:
		awsCli, err = initAWSClientSet()
		if err != nil {
			log.Fatal("failed to init aws client", zap.Error(err))
		}
		pfInitializer = providerconfig.NewAWSInitializer()
	case consts.ProviderTypeACK:
		aliyunCli = aliyun.NewClientSet()
		pfInitializer = providerconfig.NewAliCloudInitializer()
	case consts.ProviderTypeAKS:
		azureCli, err = azure.NewDefaultClientSet()
		if err != nil {
			log.Fatal("failed to init azure client", zap.Error(err))
		}
		pfInitializer = providerconfig.NewAzureInitializer()
	default:
		log.Fatal("unsupported provider", zap.String("provider", provider))
	}
	h.initProviderConfig(ctx, pfInitializer)

	p, err := predicate.LabelSelectorPredicate(rootConf.Predicate)
	if err != nil {
		log.Fatal("unable to create predicate", zap.Error(err))
	}

	if h.shouldStartController("dataplane") {
		reconciler := dataplane.NewDataPlaneReconciler(
			mgr.GetClient(),
			ctrl.Log.WithName("controllers").WithName("dataplane"),
			mgr.GetScheme(),
			mgr.GetEventRecorderFor("dataplane-controller"),
			rootConf,
			tmplConfig,
		)
		switch provider {
		case consts.ProviderTypeEKS:
			pleZoneResolver, err := dataplane.NewAWSPrivateLinkEndpointZoneResolver(awsCli)
			if err != nil {
				log.Fatal("failed to init private link endpoint zone resolver", zap.Error(err))
			}
			reconciler.WithPrivateLinkEndpointZoneResolver(pleZoneResolver)
		case consts.ProviderTypeACK:
			pleZoneResolver, err := dataplane.NewAliyunPrivateLinkEndpointZoneResolver(aliyunCli)
			if err != nil {
				log.Fatal("failed to init private link endpoint zone resolver", zap.Error(err))
			}
			reconciler.WithPrivateLinkEndpointZoneResolver(pleZoneResolver)
		}
		if err := reconciler.SetupWithManager(mgr, p); err != nil {
			log.Fatal("unable to create dataplane controller", zap.Error(err))
		}
	}

	if h.shouldStartController("bucket") {
		if err := bucket.NewReconciler(
			mgr.GetClient(),
			ctrl.Log.WithName("controllers").WithName("bucket"),
			mgr.GetEventRecorderFor("bucket-controller"),
			rootConf,
			azureCli,
		).SetupWithManager(mgr, p); err != nil {
			log.Fatal("unable to create bucket controller", zap.Error(err))
		}
	}

	if h.shouldStartController("key") {
		if err := key.NewReconciler(
			mgr.GetClient(),
			ctrl.Log.WithName("controllers").WithName("key"),
			mgr.GetScheme(),
			mgr.GetEventRecorderFor("key-controller"),
			rootConf,
		).SetupWithManager(mgr, p); err != nil {
			log.Fatal("unable to create key controller", zap.Error(err))
		}
	}

	if h.shouldStartController("vpc") {
		if err := vpc.NewVPCReconciler(
			mgr.GetClient(),
			ctrl.Log.WithName("controllers").WithName("vpc"),
			mgr.GetScheme(),
			mgr.GetEventRecorderFor("vpc-controller"),
			rootConf,
		).SetupWithManager(mgr, p); err != nil {
			log.Fatal("unable to create vpc controller", zap.Error(err))
		}
	}

	if h.shouldStartController("vpcpeering") {
		if err := vpcpeering.NewReconciler(
			mgr.GetClient(),
			ctrl.Log.WithName("controllers").WithName("vpcpeering"),
			mgr.GetEventRecorderFor("vpcpeering-controller"),
			rootConf,
		).WithAWSClient(awsCli).SetupWithManager(mgr, p); err != nil {
			log.Fatal("unable to create vpcpeering controller", zap.Error(err))
		}
	}

	if h.shouldStartController("loadbalancer") {
		if err := loadbalancer.NewLoadBalancerReconciler(
			mgr.GetClient(),
			ctrl.Log.WithName("controllers").WithName("loadbalancer"),
			mgr.GetScheme(),
			mgr.GetEventRecorderFor("loadbalancer-controller"),
			rsmanager,
			awsCli,
			aliyunCli,
			rootConf,
		).SetupWithManager(mgr, p); err != nil {
			log.Fatal("unable to create loadbalancer controller", zap.Error(err))
		}
	}

	if h.shouldStartController("appbox") {
		if err := appbox.NewAPPBoxReconciler(
			mgr.GetClient(),
			ctrl.Log.WithName("controllers").WithName("appbox"),
			mgr.GetScheme(),
			rootConf,
			tmplConfig,
			rsmanager,
			mgr.GetEventRecorderFor("appbox-controller"),
			provider,
			awsCli,
			aliyunCli,
			azureCli,
		).SetupWithManager(mgr, p); err != nil {
			log.Fatal("unable to create appbox controller", zap.Error(err))
		}
	}

	if h.shouldStartController("internet") {
		if err := internet.NewInternetReconciler(
			mgr.GetClient(),
			ctrl.Log.WithName("controllers").WithName("internet"),
			mgr.GetScheme(),
			mgr.GetEventRecorderFor("internet-controller"),
			rootConf,
			aliyunCli,
		).SetupWithManager(mgr, p); err != nil {
			log.Fatal("unable to create internet controller", zap.Error(err))
		}
	}

	if h.shouldStartController("privatelinkservice") {
		if err := privatelinkservice.NewPrivateLinkServiceReconciler(
			mgr.GetClient(),
			ctrl.Log.WithName("controllers").WithName("privatelinkservice"),
			mgr.GetScheme(),
			mgr.GetEventRecorderFor("privatelinkservice-controller"),
			rootConf,
			awsCli,
			aliyunCli,
			azureCli,
		).SetupWithManager(mgr, p); err != nil {
			log.Fatal("unable to create privatelinkservice controller", zap.Error(err))
		}
	}

	if h.shouldStartController("privatelinkendpoint") {
		if err := privatelinkendpoint.NewPrivateLinkEndpointReconciler(
			mgr.GetClient(),
			ctrl.Log.WithName("controllers").WithName("privatelinkendpoint"),
			mgr.GetScheme(),
			mgr.GetEventRecorderFor("privatelinkendpoint-controller"),
			rootConf,
			awsCli,
			aliyunCli,
			azureCli,
		).SetupWithManager(mgr, p); err != nil {
			log.Fatal("unable to create privatelinkendpoint controller", zap.Error(err))
		}
	}

	if h.shouldStartController("nodepool") {
		if err := nodepool.NewReconciler(
			mgr.GetClient(),
			ctrl.Log.WithName("controllers").WithName("nodepool"),
			mgr.GetScheme(),
			mgr.GetEventRecorderFor("nodepool-controller"),
			rootConf,
			awsCli,
		).SetupWithManager(mgr, p); err != nil {
			log.Fatal("unable to create nodepool controller", zap.Error(err))
		}
	}

	// Resource Tagger controller: Automatically tags cloud resources (EC2, EBS) for TiDB clusters
	if h.shouldStartController("resourcetagger") && rootConf.ResourceTaggerConfig.Enabled {
		reconciler, err := resourcetagger.NewReconciler(
			mgr.GetClient(),
			ctrl.Log.WithName("controllers").WithName("ResourceTagger"),
			&rootConf,
			rsmanager,
			awsCli,
			aliyunCli,
		)
		if err != nil {
			log.Fatal("unable to create ResourceTagger reconciler", zap.Error(err))
		}
		if err := reconciler.SetupWithManager(mgr); err != nil {
			log.Fatal("unable to setup ResourceTagger controller", zap.Error(err))
		}
	} else {
		log.Info("Resource tagger is disabled, skipping controller registration")
	}

	metrics.SetupCollectors(provider, mgr.GetClient(), ctrl.Log.WithName("metrics-collector"))
	if err := mgr.AddHealthzCheck("healthz", healthz.Ping); err != nil {
		log.Fatal("unable to set up health check", zap.Error(err))
	}
	if err := mgr.AddReadyzCheck("readyz", healthz.Ping); err != nil {
		log.Fatal("unable to set up ready check", zap.Error(err))
	}
}

func (h *Controller) initProviderConfig(ctx context.Context, pfInitializer providerconfig.Initializer) {
	accountPath := h.cfg.Config.AccountListPath
	if accountPath == "" {
		log.Info("no account list config found in root config, skipping ProviderConfig initialization")
		return
	}

	clientNoCache, err := client.New(h.cfg.KubeConfig, client.Options{
		Scheme: h.cfg.Scheme,
	})
	if err != nil {
		log.Fatal("failed to create non-cached client", zap.Error(err))
	}
	if err := pfInitializer.InitializeProviderConfig(
		ctx, clientNoCache, accountPath, &h.cfg.Config); err != nil {
		log.Fatal("failed to initialize provider config", zap.Error(err))
	}
	log.Info("successfully initialized provider configs from account list")
}

func initAWSClientSet() (aws.Client, error) {
	cfg, err := config.LoadDefaultConfig(context.TODO())
	if err != nil {
		return nil, err
	}
	return aws.NewClientSet(cfg), nil
}

func buildCacheSelector(cfg rsconfig.RootConfig) (map[client.Object]cache.ByObject, error) {
	predicates := cfg.Predicate
	// parse cache selector from root config
	selector, err := metav1.LabelSelectorAsSelector(&predicates)
	if err != nil {
		return nil, err
	}
	byObject := make(map[client.Object]cache.ByObject, len(buildCacheObjects()))
	for _, obj := range buildCacheObjects() {
		byObject[obj] = cache.ByObject{Label: selector}
	}
	return byObject, nil
}

func buildCacheObjects() []client.Object {
	return []client.Object{
		&v1alpha1.DataPlane{},
		&v1alpha1.APPBox{},
		&v1alpha1.NodePool{},
		&v1alpha1.VPC{},
		&v1alpha1.VPCPeering{},
		&v1alpha1.Internet{},
		&v1alpha1.LoadBalancer{},
	}
}

func (h *Controller) shouldStartController(name string) bool {
	if len(h.cfg.EnabledControllers) == 0 {
		return true
	}
	return h.cfg.EnabledControllers[name]
}