package integration

import (
	"context"
	"crypto/tls"
	"fmt"
	"math/rand"
	"net"
	"os"
	"path/filepath"
	"reflect"
	"strings"
	"testing"
	"time"

	pcav1beta1 "github.com/cert-manager/aws-privateca-issuer/pkg/api/v1beta1"
	. "github.com/onsi/ginkgo/v2"
	. "github.com/onsi/gomega"
	asappsv1 "github.com/pingcap/advanced-statefulset/client/apis/apps/v1"
	operatorv1alpha1 "github.com/pingcap/tidb-operator/pkg/apis/pingcap/v1alpha1"
	corev1 "k8s.io/api/core/v1"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
	"k8s.io/apimachinery/pkg/api/resource"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
	"k8s.io/apimachinery/pkg/runtime"
	apiruntime "k8s.io/apimachinery/pkg/runtime"
	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
	"k8s.io/apimachinery/pkg/util/yaml"
	clientgoscheme "k8s.io/client-go/kubernetes/scheme"
	"k8s.io/client-go/rest"
	"k8s.io/client-go/tools/clientcmd"
	clientcmdapi "k8s.io/client-go/tools/clientcmd/api"
	"k8s.io/client-go/util/workqueue"
	"k8s.io/utils/ptr"
	ctrl "sigs.k8s.io/controller-runtime"
	"sigs.k8s.io/controller-runtime/pkg/client"
	"sigs.k8s.io/controller-runtime/pkg/envtest"
	logf "sigs.k8s.io/controller-runtime/pkg/log"
	"sigs.k8s.io/controller-runtime/pkg/log/zap"
	metricsserver "sigs.k8s.io/controller-runtime/pkg/metrics/server"
	"sigs.k8s.io/controller-runtime/pkg/predicate"
	"sigs.k8s.io/controller-runtime/pkg/webhook"

	"github.com/tidbcloud/eks-provider/controllers/cluster"
	"github.com/tidbcloud/eks-provider/controllers/configchange"
	configmanager "github.com/tidbcloud/eks-provider/controllers/utils/configmanager"
	workflowctrl "github.com/tidbcloud/eks-provider/controllers/workflow"
	cloudv1 "github.com/tidbcloud/infra-api/api/v1"
	"github.com/tidbcloud/infra-provider/pkg/account"
	config2 "github.com/tidbcloud/infra-provider/pkg/config"
	"github.com/tidbcloud/infra-provider/pkg/consts"
	"github.com/tidbcloud/infra-provider/pkg/dataplane"
	"github.com/tidbcloud/infra-provider/pkg/k8s"
	"github.com/tidbcloud/infra-provider/pkg/scheme"
	utiltime "github.com/tidbcloud/infra-provider/pkg/util/time"
)

const (
	kubeApiQPS       = 65536
	CustomReconciler = "my-reconciler"
	timeout          = time.Second * 120
	interval         = time.Second * 2
	azureRegion      = "earth"
)

// only on node 1 (synchronized)
var testEnv *envtest.Environment

// on all nodes
var (
	k8sClient client.Client
	k8sConfig *rest.Config
)
var testNamespacePrefix = "test-"

const (
	configChangeReconcilerLabel = "integration-test"
	clusterReconcilerLabel      = "cluster-reconciler"
	workflowReconcilerLabel     = "workflow-reconciler"
)

var (
	// we use separate reconciler label to isolate test for different controllers
	configChangePredicate, _ = predicate.LabelSelectorPredicate(metav1.LabelSelector{
		MatchLabels: map[string]string{
			cloudv1.ReconcilerLabelKey: configChangeReconcilerLabel,
		},
		MatchExpressions: nil,
	})
	clusterPredicate, _ = predicate.LabelSelectorPredicate(metav1.LabelSelector{
		MatchLabels: map[string]string{
			cloudv1.ReconcilerLabelKey: clusterReconcilerLabel,
		},
		MatchExpressions: nil,
	})
	workflowPredicate, _ = predicate.LabelSelectorPredicate(metav1.LabelSelector{
		MatchLabels: map[string]string{
			cloudv1.ReconcilerLabelKey: workflowReconcilerLabel,
		},
		MatchExpressions: nil,
	})
	configRevisionGCPeriod = time.Second * 2
	configRevisionTTL      = time.Second * 5
	azureZones             = []string{"moon-1", "moon-2", "moon-3"}
)

var (
	cachedK8sProvider dataplane.EnvProviderManager
	dataplaneScheme   *runtime.Scheme
)

func init() {
	dataplaneScheme = runtime.NewScheme()
	utilruntime.Must(asappsv1.AddToScheme(dataplaneScheme))
	utilruntime.Must(operatorv1alpha1.AddToScheme(dataplaneScheme))
	utilruntime.Must(clientgoscheme.AddToScheme(dataplaneScheme))
	utilruntime.Must(operatorv1alpha1.SchemeBuilder.AddToScheme(dataplaneScheme))
	utilruntime.Must(pcav1beta1.AddToScheme(dataplaneScheme))
}

// These tests use Ginkgo (BDD-style Go testing framework). Refer to
// http://onsi.github.io/ginkgo/ to learn more about Ginkgo.
func TestAPIs(t *testing.T) {
	RegisterFailHandler(Fail)
	rand.Seed(time.Now().UnixNano())
	RunSpecs(t, "Controllers Integration Suite")
}

// SynchronizedBeforeSuite accepts two functions as arguments
// The first function runs in main process, pass []byte to the second function
// The second function runs in parallel
// We start the testEnv (etcd+apiserver+controllers) in the first function
// Then pass kubeconfig to the second function, the second function creates a new kube client
// for a test suite
var _ = SynchronizedBeforeSuite(func() []byte {
	// runs in node 1 only (synchronized)
	// setup a singleton api-server and all necessary business controllers and mock controllers
	logf.SetLogger(zap.New(zap.WriteTo(GinkgoWriter), zap.UseDevMode(true)))

	By("Modify variables for Integration test")
	var (
		controllerInitDelay   = 1 * time.Second
		controllerMaxInterval = 2 * time.Second
	)
	k8s.RetryLimiter = workqueue.NewTypedMaxOfRateLimiter(
		workqueue.NewTypedItemExponentialFailureRateLimiter[ctrl.Request](controllerInitDelay, controllerMaxInterval),
	)

	By("bootstrapping test environment")

	testEnv = &envtest.Environment{
		CRDDirectoryPaths: []string{
			filepath.Join("../../../infra-api", "config/crd/bases"),
			filepath.Join("../", "external-crds"),
		},
		ErrorIfCRDPathMissing: true,
		WebhookInstallOptions: envtest.WebhookInstallOptions{
			Paths: []string{
				filepath.Join("../../../infra-api", "config/webhook"),
			},
		},
		ControlPlaneStopTimeout:  time.Minute * 5,
		ControlPlaneStartTimeout: time.Minute * 5,
	}

	cfg, err := testEnv.Start()
	Expect(err).NotTo(HaveOccurred())
	Expect(cfg).NotTo(BeNil())
	cfg.QPS = kubeApiQPS
	cfg.Burst = kubeApiQPS

	data, err := GenerateKubeConfiguration("", cfg)
	Expect(err).NotTo(HaveOccurred())
	fname, err := filepath.Abs("./kubeconfig.tmp")
	Expect(err).NotTo(HaveOccurred())
	err = os.WriteFile(fname, data, 0o644) // #nosec G306
	Expect(err).NotTo(HaveOccurred())
	logf.Log.Info("write test kubeconfig", "filename", fname)

	k8sClient = newClient(cfg)
	Expect(k8sClient).NotTo(BeNil())
	k8sConfig = cfg

	webhookInstallOptions := &testEnv.WebhookInstallOptions
	k8sManager, err := ctrl.NewManager(cfg, ctrl.Options{
		Scheme: scheme.Scheme,
		WebhookServer: webhook.NewServer(webhook.Options{
			Host:    webhookInstallOptions.LocalServingHost,
			Port:    webhookInstallOptions.LocalServingPort,
			CertDir: webhookInstallOptions.LocalServingCertDir,
		}),
		LeaderElection: false,
		Metrics: metricsserver.Options{
			BindAddress: "0",
		},
	})
	Expect(err).ToNot(HaveOccurred())

	rootConfig := config2.RootConfig{
		ClusterController: config2.ClusterControllerConfig{
			ConfigChange: config2.ConfigChangeConfig{},
		},
	}
	cachedK8sProvider = dataplane.NewFakeEnvProviderManager(dataplaneScheme, nil)
	// setup ConfigChange controller
	err = configchange.NewConfigChangeReconciler(k8sManager, cachedK8sProvider,
		ctrl.Log.WithName("controllers").WithName("ConfigChange"),
		rootConfig,
	).SetupWithManager(k8sManager, configChangePredicate)
	Expect(err).ToNot(HaveOccurred())

	// setup cluster controller
	azureConfig := config2.RootConfig{
		ControllerConfig: config2.ControllerConfig{
			Provider: consts.ProviderTypeAKS,
			Network: config2.NetworkConfig{
				Region: azureRegion,
			},
		},
		Predicate: metav1.LabelSelector{
			MatchLabels: map[string]string{
				cloudv1.ProviderTypeLabel: consts.ProviderTypeAKS,
				cloudv1.RegionLabel:       azureRegion,
			},
			MatchExpressions: []metav1.LabelSelectorRequirement{
				{
					Key:      cloudv1.ReconcilerLabelKey,
					Operator: metav1.LabelSelectorOpIn,
					Values:   []string{clusterReconcilerLabel},
				},
			},
		},
		ClusterController: config2.ClusterControllerConfig{
			Azure: config2.AzureConfig{
				CrossplaneAzureProviderSubscriptionID: "123",
				CrossplaneAzureProviderTenantID:       "1356",
				CrossplaneAzureProviderClientID:       "xxxx",
			},
		},
	}
	acctInfoGetter := &account.Info{
		ClusterInfo: account.ClusterInfo{
			ComponentPCAARN: "this:is:aws:pca:arn",
		},
	}
	configManager := configmanager.NewConfigManager(k8sManager.GetClient(), config2.PreConfigKeys, &azureConfig)
	clusterReconciler := &cluster.ClusterReconciler{
		Client:                k8sManager.GetClient(),
		CachedProvider:        cachedK8sProvider,
		AcctInfoGetter:        acctInfoGetter,
		CloudConfigProvider:   newFakeAzureCloudConfig(),
		ConfigManager:         configManager,
		Config:                azureConfig,
		DataplaneHTTPEndpoint: &fakeDataplaneHTTPEndpoint{},
		Cache:                 k8sManager.GetCache(),
		Logger:                ctrl.Log.WithName("controllers").WithName("Cluster"),
		EventRecorder:         k8sManager.GetEventRecorderFor("cluster-controller"),
	}
	err = clusterReconciler.SetupWithManager(k8sManager, clusterPredicate)
	Expect(err).ToNot(HaveOccurred())

	// setup Workflow controller
	wfReconciler := workflowctrl.NewWorkflowReconciler(
		k8sManager,
		cachedK8sProvider,
		ctrl.Log.WithName("controllers").WithName("Workflow"),
		utiltime.RealClock{},
		nil, // gcpCli not needed for tests
		workflowctrl.WithPDAuth(noopProxyPDAuth),
	)
	err = wfReconciler.SetupWithManager(k8sManager, workflowPredicate)
	Expect(err).ToNot(HaveOccurred())

	// register webhooks
	k8sManager.GetWebhookServer().Register("/mutate-cloud-pingcap-com-v1-cluster-revision", &webhook.Admission{Handler: newFakeClusterMutator(k8sManager)})
	err = cloudv1.RegisterWebhooks(k8sManager)
	Expect(err).ToNot(HaveOccurred())

	ctx := ctrl.SetupSignalHandler()
	go func() {
		defer GinkgoRecover()
		err = k8sManager.Start(ctx)
		Expect(err).ToNot(HaveOccurred())
	}()

	// wait for the webhook server to get ready
	dialer := &net.Dialer{Timeout: time.Second}
	addrPort := fmt.Sprintf("%s:%d", webhookInstallOptions.LocalServingHost, webhookInstallOptions.LocalServingPort)
	Eventually(func() error {
		conn, err := tls.DialWithDialer(dialer, "tcp", addrPort, &tls.Config{InsecureSkipVerify: true}) // #nosec G402
		if err != nil {
			return err
		}
		conn.Close()
		return nil
	}).Should(Succeed())

	err = initProvider(k8sClient)
	Expect(err).NotTo(HaveOccurred())

	return data
}, func(kubeconfigData []byte) {
	// runs on all node
	// setup the k8s and gardener clients for each node
	clientCfg, err := clientcmd.NewClientConfigFromBytes(kubeconfigData)
	Expect(err).To(Succeed())
	cfg, err := clientCfg.ClientConfig()
	Expect(err).To(Succeed())
	cfg.QPS = kubeApiQPS
	cfg.Burst = kubeApiQPS
	if k8sConfig == nil {
		k8sConfig = cfg
	}
	if k8sClient == nil {
		k8sClient = newClient(k8sConfig)
	}
})

func newClient(cfg *rest.Config) client.Client {
	var err error
	k8sClient, err = client.New(cfg, client.Options{Scheme: scheme.Scheme})
	Expect(err).NotTo(HaveOccurred())
	return k8sClient
}

var _ = SynchronizedAfterSuite(func() {
	// runs on all nodes
	fmt.Printf("ginkgo node %d exit", GinkgoParallelProcess())
}, func() {
	// runs in node 1 only (synchronized)
	By("tearing down the test environment")
	err := testEnv.Stop()
	Expect(err).NotTo(HaveOccurred())
})

var providerTypeTPL = &cloudv1.ProviderType{
	ObjectMeta: metav1.ObjectMeta{
		Name: "name",
	},
	Spec: cloudv1.ProviderTypeSpec{
		Capabilities: nil,
	},
}

var (
	storage     = resource.MustParse("1000Gi")
	providerTPL = &cloudv1.Provider{
		ObjectMeta: metav1.ObjectMeta{
			Name: "name",
		},
		Spec: cloudv1.ProviderSpec{
			Type: "name",
			Regions: []cloudv1.Region{
				{
					Name:  "test-region",
					Zones: []string{"z-a", "z-b", "z-c"},
				},
			},
			InstanceTypes: []cloudv1.InstanceType{
				{
					Name:   "compute",
					Arch:   consts.AMD64ArchName,
					CPU:    resource.MustParse("8"),
					Memory: resource.MustParse("16Gi"),
				},
				{
					Name:       "storage",
					Arch:       consts.AMD64ArchName,
					CPU:        resource.MustParse("8"),
					Memory:     resource.MustParse("32Gi"),
					MinStorage: &storage,
					MaxStorage: &storage,
				},
				{
					Name:   "test",
					Arch:   consts.AMD64ArchName,
					CPU:    resource.MustParse("1"),
					Memory: resource.MustParse("2Gi"),
				},
				{
					Name:   "i3.2xlarge",
					Arch:   consts.AMD64ArchName,
					CPU:    resource.MustParse("1"),
					Memory: resource.MustParse("2Gi"),
				},
				{
					Name:   "r5.2xlarge",
					Arch:   consts.AMD64ArchName,
					CPU:    resource.MustParse("8"),
					Memory: resource.MustParse("16Gi"),
				},
				{
					Name:   "c5.9xlarge",
					Arch:   consts.AMD64ArchName,
					CPU:    resource.MustParse("36"),
					Memory: resource.MustParse("72Gi"),
				},
			},
		},
	}
)

func initProvider(client client.Client) error {
	ctx := context.Background()

	regions := []cloudv1.Region{
		{
			Name:  "us-east-2",
			Zones: []string{"z1", "z2", "z3"},
		},
	}

	tp := providerTypeTPL.DeepCopy()
	tp.Name = consts.ProviderTypeEKS
	err := client.Create(ctx, tp)
	if err != nil {
		return err
	}

	tp = providerTypeTPL.DeepCopy()
	tp.Name = consts.ProviderTypeGKE
	err = client.Create(ctx, tp)
	if err != nil {
		return err
	}

	p := providerTPL.DeepCopy()
	p.Name = consts.ProviderTypeEKS
	p.Spec.Type = consts.ProviderTypeEKS
	p.Spec.Regions = append(p.Spec.Regions, regions...)
	spec := &cloudv1.GardenerProviderSpec{
		TypeMeta: metav1.TypeMeta{
			Kind:       reflect.TypeOf(cloudv1.GardenerProviderSpec{}).Name(),
			APIVersion: cloudv1.GroupVersion.String(),
		},
		EnableGracefulShutdownWebhook: ptr.To(false),
		AWS: cloudv1.AwsConfig{
			AccountID:              "9527",
			PublicHostedZoneDomain: "aws.pingcap.com",
			CrossplaneProvider:     "provider-aws",
		},
		DataFlowAgent: cloudv1.DataFlowAgentConfig{Image: "test image"},
	}
	p.Spec.ForProvider = &apiruntime.RawExtension{Object: spec}

	err = client.Create(ctx, p)
	if err != nil {
		return err
	}

	p = providerTPL.DeepCopy()
	p.Name = consts.ProviderTypeGKE
	p.Spec.Type = consts.ProviderTypeGKE
	p.Spec.Regions = regions
	spec = &cloudv1.GardenerProviderSpec{
		TypeMeta: metav1.TypeMeta{
			Kind:       reflect.TypeOf(cloudv1.GardenerProviderSpec{}).Name(),
			APIVersion: cloudv1.GroupVersion.String(),
		},
		EnableGracefulShutdownWebhook: ptr.To(true),
		AWS: cloudv1.AwsConfig{
			AccountID:          "9527",
			CrossplaneProvider: "provider-aws",
		},
		GCP: cloudv1.GcpConfig{
			ProjectID:              "test-project",
			CertManagerRole:        "aws:dfdfdf/certrole",
			PublicHostedZoneDomain: "gcp.pingcap.com",
			CrossplaneProvider:     "provider-gcp",
		},
		DataFlowAgent: cloudv1.DataFlowAgentConfig{Image: "test image"},
	}
	p.Spec.ForProvider = &apiruntime.RawExtension{Object: spec}
	err = client.Create(ctx, p)
	if err != nil {
		return err
	}

	return nil
}

// change from https://github.com/kubernetes/client-go/issues/711#issuecomment-730112049
func GenerateKubeConfiguration(namespace string, cfg *rest.Config) ([]byte, error) {
	if namespace == "" {
		namespace = "kube-system"
	}

	clusters := make(map[string]*clientcmdapi.Cluster)
	clusters["default-cluster"] = &clientcmdapi.Cluster{
		Server:                   cfg.Host,
		CertificateAuthorityData: cfg.CAData,
	}

	contexts := make(map[string]*clientcmdapi.Context)
	contexts["default-context"] = &clientcmdapi.Context{
		Cluster:   "default-cluster",
		Namespace: namespace,
		AuthInfo:  "default-context",
	}

	authinfos := make(map[string]*clientcmdapi.AuthInfo)
	authinfos["default-context"] = &clientcmdapi.AuthInfo{
		// try to cover all auth strategies that aren't plugins
		ClientCertificateData: cfg.CertData,
		ClientKeyData:         cfg.KeyData,
		Token:                 cfg.BearerToken,
		Username:              cfg.Username,
		Password:              cfg.Password,
	}

	clientConfig := clientcmdapi.Config{
		Kind:           "Config",
		APIVersion:     "v1",
		Clusters:       clusters,
		Contexts:       contexts,
		CurrentContext: "default-context",
		AuthInfos:      authinfos,
	}

	data, err := clientcmd.Write(clientConfig)
	return data, err
}

func ApplyObjectFromDir(client client.Client, dir string) error {
	infos, err := os.ReadDir(dir)
	if err != nil {
		return err
	}
	for _, f := range infos {
		data, err := os.ReadFile(filepath.Join(dir, f.Name()))
		if err != nil {
			return err
		}
		obj := &unstructured.Unstructured{}
		err = yaml.Unmarshal(data, obj)
		if err != nil {
			return err
		}
		err = client.Create(context.TODO(), obj)
		if err != nil {
			return err
		}
	}
	return nil
}

func prepareNetwork(namespace string, ctx context.Context) (*cloudv1.Network, client.Client, *cloudv1.GardenerNetworkStatus) {
	var (
		shootClient  client.Client
		gardenStatus *cloudv1.GardenerNetworkStatus
		err          error
	)

	ns := &corev1.Namespace{
		ObjectMeta: metav1.ObjectMeta{
			Name: namespace,
		},
	}
	err = k8sClient.Create(ctx, ns)
	if err != nil && !apierrors.IsAlreadyExists(err) {
		Expect(err).To(Succeed())
	}

	nt := &cloudv1.Network{
		ObjectMeta: metav1.ObjectMeta{
			Name:      "network-" + cloudv1.RandomString(5),
			Namespace: ns.Name,
			Labels: map[string]string{
				"cloud.pingcap.com/project": cloudv1.RandomString(5),
			},
		},
		Spec: cloudv1.NetworkSpec{
			Provider: "eks",
			Region:   "us-east-2",
			CIDR:     "10.251.0.0/16",
		},
	}
	Expect(k8sClient.Create(ctx, nt)).To(Succeed())

	nt.Status.Phase = cloudv1.NetworkPhaseNormal
	Expect(k8sClient.Status().Update(ctx, nt)).To(Succeed())
	return nt, shootClient, gardenStatus
}

// newLegacyLegacyCluster build a Cluster CR for cloud-native arch
func newLegacyCluster(network *cloudv1.Network, tenantID, clusterID, instanceType string) *cloudv1.Cluster {
	c := &cloudv1.Cluster{
		ObjectMeta: metav1.ObjectMeta{
			Name:      fmt.Sprintf("%d", rand.Int()),
			Namespace: network.Namespace,
			Labels: map[string]string{
				cloudv1.ReconcilerLabelKey: "enable-gitops",
				cloudv1.ProjectTagKey:      strings.SplitN(network.Namespace, "-", 2)[1],
			},
			Annotations: map[string]string{},
		},
		Spec: cloudv1.ClusterSpec{
			NetworkName: network.Name,
			Kubernetes: cloudv1.Kubernetes{
				Provider: consts.ProviderTypeKinD,
				Name:     "dummy",
				Endpoint: "https://dummy",
				Account:  "dummy",
				Network:  "dummy",
			},
			Cluster: cloudv1.ClusterTopology{
				TiDB: cloudv1.TiDB{
					Replicas:     2,
					InstanceType: instanceType,
				},
				TiKV: cloudv1.TiKV{
					Replicas:     3,
					InstanceType: instanceType,
					StorageSize:  resource.MustParse("1000Gi"),
				},
				PD: cloudv1.PD{
					Replicas:     3,
					InstanceType: instanceType,
					StorageSize:  resource.MustParse("10Gi"),
				},
				TiFlash: cloudv1.TiFlash{
					Replicas:     2,
					InstanceType: instanceType,
					StorageSize:  resource.MustParse("1000Gi"),
				},
			},
			SqlEndpoints: cloudv1.SqlEndpoints{
				Public: cloudv1.Service{
					Enabled: ptr.To(true),
				},
			},
		},
	}
	if tenantID != "" {
		c.Labels[cloudv1.TenantTagKey] = tenantID
	}
	if clusterID != "" {
		c.Labels[cloudv1.ClusterTagKey] = clusterID
	}

	cr, err := createInitialConfigRevision(context.Background(), k8sClient, c)
	if err != nil {
		panic(fmt.Errorf("create configRevision: %w", err))
	}
	c.Spec.ConfigRevisionRef = &cloudv1.Reference{Name: cr.Name}

	return c
}

func genNamespace() string {
	return testNamespacePrefix + cloudv1.RandomString(5)
}