package fixtures

import (
	"context"
	"fmt"

	. "github.com/onsi/ginkgo/v2"
	. "github.com/onsi/gomega"
	"github.com/pingcap/errors"
	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/types"

	cloudv1 "github.com/tidbcloud/infra-api/api/v1"
	"github.com/tidbcloud/infra-api/test/e2ev2/framework"
)

var _ = Describe("cdc", func() {
	var (
		f   = framework.DefaultFramework
		p   = f.Provider
		ctx = context.Background()
		c   = f.InfraCli
		t   = f.TestContext
		ns  *corev1.Namespace
	)

	BeforeEach(func() {
		ns = &corev1.Namespace{
			ObjectMeta: metav1.ObjectMeta{
				GenerateName: fmt.Sprintf("e2e-%s-", p.Name()),
			},
		}
		f.ExpectNoError(c.Create(ctx, ns), "create namespace")
		f.Log("namespace: %s", ns.Name)
	})
	AfterEach(func() {
		f.ExpectNoError(c.Delete(ctx, ns), "delete namespace")
	})

	It("cdc capability", func() {
		network := p.NewNetwork(ctx)
		network.Namespace = ns.Name
		f.ExpectNoError(c.Create(ctx, network), "create network")
		f.Log("network: %s", network.Name)
		Eventually(func() error {
			err := c.Get(ctx, types.NamespacedName{Namespace: network.Namespace, Name: network.Name}, network)
			if err != nil {
				return err
			}
			if network.Status.Phase != cloudv1.NetworkPhaseNormal {
				return errors.New("network status is not normal")
			}

			return nil

		}, t.NetworkCreateTimeout, t.Interval).Should(Succeed(), fmt.Sprintf("network status to normal within %v", t.NetworkCreateTimeout))

		defer func() {
			f.ExpectNoError(c.Delete(ctx, network), "delete network1")
			Eventually(func() error {
				err := c.Get(ctx, types.NamespacedName{Namespace: network.Namespace, Name: network.Name}, network)
				if err != nil && !apierrors.IsNotFound(err) {
					return err
				}
				if apierrors.IsNotFound(err) {
					return nil
				}
				return errors.New("network is not deleted")
			}, t.NetworkDeleteTimeout, t.Interval).Should(Succeed(), fmt.Sprintf("network is deleted within %v", t.NetworkDeleteTimeout))
			f.Log("delete network1 done: %s", network.Name)
		}()

		cluster := p.NewCluster(ctx, network)
		cluster.GenerateName = "cdc-cluster-"
		f.ExpectNoError(c.Create(ctx, cluster), "create upstream cluster")
		f.Log("cluster: %s", cluster.Name)

		defer func() {
			f.ExpectNoError(c.Delete(ctx, cluster), "delete cluster")
			Eventually(func() error {
				err := c.Get(ctx, types.NamespacedName{Namespace: cluster.Namespace, Name: cluster.Name}, cluster)
				if err != nil && !apierrors.IsNotFound(err) {
					return err
				}
				if apierrors.IsNotFound(err) {
					return nil
				}
				return errors.New("cluster is not deleted")
			}, t.ClusterDeleteTimeout, t.Interval).Should(Succeed(), fmt.Sprintf("cluster is deleted within %v", t.NetworkDeleteTimeout))
			f.Log("delete cluster done: %s", cluster.Name)
		}()

		Eventually(func() error {
			err := c.Get(ctx, types.NamespacedName{Namespace: cluster.Namespace, Name: cluster.Name}, cluster)
			if err != nil {
				return err
			}
			if cluster.Status.Phase != cloudv1.ClusterPhaseRunning {
				return errors.New("cluster status is not running")
			}

			return nil
		}, t.ClusterCreateTimeout, t.Interval).Should(Succeed(), fmt.Sprintf("cluster status to normal within %v", t.ClusterCreateTimeout))
		f.Log("create cluster done: %s", cluster.Name)

		_, err := p.ConnectNetwork(ctx, c, network)
		f.ExpectNoError(err, "connect to network")

		cdc := &cloudv1.TiCDC{
			ObjectMeta: metav1.ObjectMeta{
				GenerateName: fmt.Sprintf("e2e-%s-", p.Name()),
				Namespace:    network.Namespace,
				Labels:       network.Labels,
			},
			Spec: cloudv1.TiCDCSpec{
				NetworkName:  network.Name,
				ClusterName:  cluster.Name,
				Replicas:     1,
				StorageSize:  resource.MustParse("10Gi"),
				InstanceType: framework.InstanceTypeT3Medium,
			},
		}
		deployAndDeleteCDC(ctx, f, cdc)

		if f.SupportCrossRegion {
			//create network in another region
			network2 := p.NewNetwork(ctx)
			network2.Spec.Region = "us-east-1"
			network2.Spec.CIDR = framework.DefaultAWSPeeringVPCCidr
			network2.Namespace = ns.Name
			f.ExpectNoError(c.Create(ctx, network2), "create network2")
			f.Log("network2: %s", network2.Name)
			Eventually(func() error {
				err := c.Get(ctx, types.NamespacedName{Namespace: network2.Namespace, Name: network2.Name}, network2)
				if err != nil {
					return err
				}
				if network2.Status.Phase != cloudv1.NetworkPhaseNormal {
					return errors.New("network status is not normal")
				}

				return nil
			}, t.NetworkCreateTimeout, t.Interval).Should(Succeed(), fmt.Sprintf("network status to normal within %v", t.NetworkCreateTimeout))
			f.Log("create network2 done: %s", network2.Name)

			defer func() {
				f.ExpectNoError(c.Delete(ctx, network2), "delete network2")
				Eventually(func() error {
					err := c.Get(ctx, types.NamespacedName{Namespace: network2.Namespace, Name: network2.Name}, network2)
					if err != nil && !apierrors.IsNotFound(err) {
						return err
					}
					if apierrors.IsNotFound(err) {
						return nil
					}
					return errors.New("network2 is not deleted")
				}, t.NetworkDeleteTimeout, t.Interval).Should(Succeed(), fmt.Sprintf("network2 is deleted within %v", t.NetworkDeleteTimeout))
				f.Log("delete network2 done: %s", network2.Name)
			}()

			np := &cloudv1.NetworkPeer{
				ObjectMeta: metav1.ObjectMeta{
					GenerateName: fmt.Sprintf("e2e-%s-", p.Name()),
					Namespace:    ns.Name,
					Labels:       network.Labels,
				},
				Spec: cloudv1.NetworkPeerSpec{
					Source: network.Name,
					Target: cloudv1.Target{
						TargetNetwork: &cloudv1.TargetNetwork{NetworkName: network2.Name},
					},
				},
			}
			f.ExpectNoError(c.Create(ctx, np), "create network peer")
			f.Log("network peer: %s", np.Name)
			defer func() {
				f.ExpectNoError(c.Delete(ctx, np), "delete network peer")
			}()

			Eventually(func() error {
				err := c.Get(ctx, types.NamespacedName{Namespace: np.Namespace, Name: np.Name}, np)
				if err != nil {
					return err
				}

				if np.Status.Phase != cloudv1.NetworkPeerPhaseActive {
					return errors.New("networkpeering status is not active")
				}

				return nil

			}, t.NetworkPeeringActiveTimeout, t.Interval).Should(Succeed(), fmt.Sprintf("networkpeering status to active within %v", t.NetworkPeeringActiveTimeout))
			f.Log("create network peer done: %s", np.Name)

			// cross region cdc without downstream cluster
			cdc := &cloudv1.TiCDC{
				ObjectMeta: metav1.ObjectMeta{
					GenerateName: fmt.Sprintf("e2e-%s-", p.Name()),
					Namespace:    network.Namespace,
					Labels:       network.Labels,
				},
				Spec: cloudv1.TiCDCSpec{
					NetworkName:  network.Name,
					Replicas:     1,
					StorageSize:  resource.MustParse("10Gi"),
					InstanceType: framework.InstanceTypeT3Medium,
				},
			}
			deployAndDeleteCDC(ctx, f, cdc)

			cluster2 := p.NewCluster(ctx, network2)
			cluster.GenerateName = "cdc-upstream-cluster-"

			f.ExpectNoError(c.Create(ctx, cluster2), "create upstream cluster")
			f.Log("cluster: %s", cluster2.Name)

			Eventually(func() error {
				err := c.Get(ctx, types.NamespacedName{Namespace: cluster2.Namespace, Name: cluster2.Name}, cluster2)
				if err != nil {
					return err
				}
				if cluster2.Status.Phase != cloudv1.ClusterPhaseRunning {
					return errors.New("cluster status is not running")
				}

				return nil
			}, t.ClusterCreateTimeout, t.Interval).Should(Succeed(), fmt.Sprintf("cluster status to normal within %v", t.ClusterCreateTimeout))

			defer func() {
				f.ExpectNoError(c.Delete(ctx, cluster2), "delete cluster2")
				Eventually(func() error {
					err := c.Get(ctx, types.NamespacedName{Namespace: cluster2.Namespace, Name: cluster2.Name}, cluster2)
					if err != nil && !apierrors.IsNotFound(err) {
						return err
					}
					if apierrors.IsNotFound(err) {
						return nil
					}
					return errors.New("cluster is not deleted")
				}, t.ClusterDeleteTimeout, t.Interval).Should(Succeed(), fmt.Sprintf("cluster2 is deleted within %v", t.NetworkDeleteTimeout))
				f.Log("delete cluster2 done: %s", cluster2.Name)
			}()

			// cross region cdc
			cdc = &cloudv1.TiCDC{
				ObjectMeta: metav1.ObjectMeta{
					GenerateName: fmt.Sprintf("e2e-%s-", p.Name()),
					Namespace:    network.Namespace,
					Labels: map[string]string{
						cloudv1.ProviderTypeLabel: cloudv1.ProviderAWS,
					},
				},
				Spec: cloudv1.TiCDCSpec{
					NetworkName:         network2.Name,
					ClusterName:         cluster2.Name,
					UpstreamClusterName: cluster.Name,
					Replicas:            1,
					StorageSize:         resource.MustParse("10Gi"),
					InstanceType:        framework.InstanceTypeT3Medium,
				},
			}
			deployAndDeleteCDC(ctx, f, cdc)
		}
	})
})

func deployAndDeleteCDC(
	ctx context.Context,
	f *framework.Framework,
	cdc *cloudv1.TiCDC) {
	t := f.TestContext
	c := f.InfraCli
	f.ExpectNoError(c.Create(ctx, cdc), "create cdc failed")
	f.Log("cdc: %s", cdc.Name)

	Eventually(func() error {
		err := c.Get(ctx, types.NamespacedName{Namespace: cdc.Namespace, Name: cdc.Name}, cdc)
		if err != nil {
			return err
		}
		if cdc.Status.Phase != cloudv1.TiCDCPhaseRunning {
			return errors.New("cdc status is not normal")
		}

		return nil

	}, t.CDCCreateTimeout, t.Interval).Should(Succeed(), fmt.Sprintf("cdc status to normal within %v", t.CDCCreateTimeout))
	f.Log("create cdc done: %s", cdc.Name)

	defer func() {
		f.ExpectNoError(c.Delete(ctx, cdc), "delete cdc")

		Eventually(func() error {
			err := c.Get(ctx, types.NamespacedName{Namespace: cdc.Namespace, Name: cdc.Name}, cdc)
			if err != nil && !apierrors.IsNotFound(err) {
				return err
			}
			if apierrors.IsNotFound(err) {
				return nil
			}
			return errors.New("cdc is not deleted")
		}, t.CDCDeleteTimeout, t.Interval).Should(Succeed(), fmt.Sprintf("cdc is deleted within %v", t.CDCDeleteTimeout))
		f.Log("delete cdc done: %s", cdc.Name)
	}()
}