package cmd

import (
	"os"

	"github.com/aws/aws-sdk-go/aws/session"
	"github.com/spf13/cobra"
	"go.uber.org/zap/zapcore"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/labels"
	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/log/zap"
	"sigs.k8s.io/controller-runtime/pkg/predicate"

	"github.com/tidbcloud/tiinf/transport/crossregion"

	"github.com/tidbcloud/eks-provider/controllers/cdc"
	cloudv1 "github.com/tidbcloud/infra-api/api/v1"
	compatk8s "github.com/tidbcloud/infra-provider/pkg/compat/k8s"
	pkgconfig "github.com/tidbcloud/infra-provider/pkg/config"
	"github.com/tidbcloud/infra-provider/pkg/o11y/sentry"
	"github.com/tidbcloud/infra-provider/pkg/procedure"
	//+kubebuilder:scaffold:imports
)

func buildCDCCacheConfig(selector labels.Selector) map[client.Object]cache.ByObject {
	return map[client.Object]cache.ByObject{
		// TODO: how to handle X-region cdc
		&cloudv1.TiCDC{}: {
			Label: selector,
		},
	}
}

func mustBuildCDCConfig() *pkgconfig.CDCConfig {
	// set logger first to avoid log lost
	opts := zap.Options{
		Development: true,
		TimeEncoder: zapcore.RFC3339TimeEncoder,
	}
	ctrl.SetLogger(zap.New(zap.UseFlagOptions(&opts)))

	cfg, err := pkgconfig.LoadCDCConfig(rootConfig)
	if err != nil {
		setupLog.Error(err, "unable to load root config")
		os.Exit(1)
	}

	overrideCDCConfigWithFlags(cfg)

	if cfg.Sentry.DSN != "" {
		sentryZapOpt, err := sentry.AddSentrytoLogger(cfg.Sentry.DSN, cfg.Sentry.Environment)
		if err != nil {
			setupLog.Error(err, "sentry init error")
			os.Exit(1)
		}
		opts.ZapOpts = append(opts.ZapOpts, sentryZapOpt)
		// reset logger with sentry integration
		ctrl.SetLogger(zap.New(zap.UseFlagOptions(&opts)))
	}
	if cfg.Region == "" {
		setupLog.Error(err, "region information is required")
		os.Exit(1)
	}
	predicates := cfg.Predicate
	var ok bool
	var region string

	if predicates.MatchLabels == nil {
		setupLog.Info("no predicate labels are configured")
		os.Exit(1)
	}
	if region, ok = predicates.MatchLabels[cloudv1.RegionLabel]; !ok {
		setupLog.Info("no region predicate are configured")
		os.Exit(1)
	}
	if _, ok = predicates.MatchLabels[cloudv1.ProviderTypeLabel]; !ok {
		setupLog.Info("no provider type predicate are configured")
		os.Exit(1)
	}
	provider := predicates.MatchLabels[cloudv1.ProviderTypeLabel]
	setupLog.Info("check cdc  config passed", "provider", provider, "region", region)
	return cfg
}

func overrideCDCConfigWithFlags(cdcConfig *pkgconfig.CDCConfig) {
	if region != "" {
		cdcConfig.Region = region
		if cdcConfig.Predicate.MatchLabels == nil {
			cdcConfig.Predicate.MatchLabels = make(map[string]string)
		}
		cdcConfig.Predicate.MatchLabels[cloudv1.RegionLabel] = region
	}
	if overrideRegistry != "" {
		cdcConfig.Override.OverrideRegistry = overrideRegistry
	}
}

func mustRegisterCDCControllers(cfg *pkgconfig.CDCConfig, mgr ctrl.Manager) {
	p, err := predicate.LabelSelectorPredicate(cfg.Predicate)
	if err != nil {
		setupLog.Error(err, "unable to create predicate")
		os.Exit(1)
	}
	// cdc
	awsSession := session.Must(session.NewSession())
	var infraCli client.Client
	var infraApiReader client.Reader
	if pkgconfig.FeatureEnabled(cfg.FeatureGates, pkgconfig.EnableDPCompatibilityMode, "", "") {
		infraCli = compatk8s.NewCompatK8sClient(mgr.GetClient(), crossregion.EnvType(cfg.DPCompatEnv), cfg.DPCompatAssumeRole, cfg.DPComaptDataPlaneRoleArn, cfg.DPCompatAssumeExternalId, cfg.Override, cfg.ClusterServiceEndpoint)
		infraApiReader = compatk8s.NewCompatK8sReader(mgr.GetAPIReader(), crossregion.EnvType(cfg.DPCompatEnv), cfg.DPCompatAssumeRole, cfg.DPComaptDataPlaneRoleArn, cfg.DPCompatAssumeExternalId, cfg.Override, cfg.ClusterServiceEndpoint)
	} else {
		infraCli = mgr.GetClient()
		infraApiReader = mgr.GetAPIReader()
	}
	cdcProvider, err := cdc.NewCdc(infraCli, infraApiReader, procedure.NewAWSService(awsSession), cfg)
	if err != nil {
		setupLog.Error(err, "build cdc provider failed")
		os.Exit(1)
	}
	if err := cdc.NewTiCDCReconcilerCompat(
		infraCli,
		infraApiReader,
		ctrl.Log.WithName("controllers").WithName("TiCDC"),
		nil,
		mgr.GetEventRecorderFor("cdc-controller"), cdcProvider).
		SetupWithManager(mgr, p); err != nil {
		setupLog.Error(err, "unable to create controller", "controller", "TiCDC")
		os.Exit(1)
	}
}

func runCDCControllers() {
	cfg := mustBuildCDCConfig()
	// parse cache selector from root config
	selector, err := metav1.LabelSelectorAsSelector(&cfg.Predicate)
	if err != nil {
		setupLog.Error(err, "invalid predicates")
		os.Exit(1)
	}
	mgr := mustBuildManager("cdc",
		cfg.Region,
		buildCDCCacheConfig(selector),
		[]client.Object{&cloudv1.Cluster{}, &cloudv1.Network{}}, false)
	mustRegisterCDCControllers(cfg, mgr)
	mustRunManager(ctrl.SetupSignalHandler(), mgr)
}

var cdcCMD = &cobra.Command{
	Use:   "run-cdc-controller",
	Short: "running cdc(ticdc) controllers",
	Run: func(cmd *cobra.Command, args []string) {
		runCDCControllers()
	},
}

func init() {
	rootCmd.AddCommand(cdcCMD)
}