/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package e2e

import (
	"context"
	"database/sql"
	"fmt"
	"reflect"
	"regexp"
	"time"

	"github.com/apache/shardingsphere-on-cloud/shardingsphere-operator/api/v1alpha1"
	"github.com/apache/shardingsphere-on-cloud/shardingsphere-operator/pkg/controllers"
	"github.com/apache/shardingsphere-on-cloud/shardingsphere-operator/pkg/reconcile/storagenode/aws"

	"bou.ke/monkey"
	"github.com/DATA-DOG/go-sqlmock"
	dbmeshawsrds "github.com/database-mesh/golang-sdk/aws/client/rds"
	. "github.com/onsi/ginkgo/v2"
	. "github.com/onsi/gomega"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"sigs.k8s.io/controller-runtime/pkg/client"
)

var _ = Describe("StorageNode Controller Suite Test For AWS RDS Instance", func() {
	storageProviderName := "test-storage-provider"
	instanceIdentifier := "test-instance-identifier"

	BeforeEach(func() {
		StorageProvider := &v1alpha1.StorageProvider{
			ObjectMeta: metav1.ObjectMeta{
				Name: storageProviderName,
			},
			Spec: v1alpha1.StorageProviderSpec{
				Provisioner: v1alpha1.ProvisionerAWSRDSInstance,
				Parameters: map[string]string{
					"engine":             "mysql",
					"engineVersion":      "5.7",
					"instanceClass":      "db.t3.micro",
					"allocatedStorage":   "20",
					"masterUsername":     "root",
					"masterUserPassword": "root123456",
				},
			},
		}

		Expect(k8sClient.Create(ctx, StorageProvider)).Should(Succeed())
	})

	AfterEach(func() {
		StorageProvider := &v1alpha1.StorageProvider{}
		Expect(k8sClient.Get(ctx, client.ObjectKey{Name: storageProviderName}, StorageProvider)).Should(Succeed())
		Expect(k8sClient.Delete(ctx, StorageProvider)).Should(Succeed())
	})

	Context("reconcile storageNode", func() {
		AfterEach(func() {
			monkey.UnpatchAll()
		})

		It("should create success", func() {
			// mock get instance func returns success
			monkey.PatchInstanceMethod(reflect.TypeOf(&aws.RdsClient{}), "GetInstance", func(_ *aws.RdsClient, _ context.Context, _ *v1alpha1.StorageNode) (*dbmeshawsrds.DescInstance, error) {
				return &dbmeshawsrds.DescInstance{
					DBInstanceStatus: dbmeshawsrds.DBInstanceStatusAvailable,
					Endpoint: dbmeshawsrds.Endpoint{
						Address: "127.0.0.1",
						Port:    3306,
					},
				}, nil
			})
			// mock delete instance func returns success
			monkey.PatchInstanceMethod(reflect.TypeOf(&aws.RdsClient{}), "DeleteInstance", func(_ *aws.RdsClient, _ context.Context, _ *v1alpha1.StorageNode, _ *v1alpha1.StorageProvider) error {
				return nil
			})

			nodeName := "test-storage-node-ready"
			node := &v1alpha1.StorageNode{
				ObjectMeta: metav1.ObjectMeta{
					Name:      nodeName,
					Namespace: "default",
					Annotations: map[string]string{
						v1alpha1.AnnotationsInstanceIdentifier: instanceIdentifier,
					},
				},
				Spec: v1alpha1.StorageNodeSpec{
					StorageProviderName: storageProviderName,
				},
			}

			// create resource
			Expect(k8sClient.Create(ctx, node)).Should(Succeed())

			// check storage node status
			Eventually(func() v1alpha1.StorageNodePhaseStatus {
				newSN := &v1alpha1.StorageNode{}
				Expect(k8sClient.Get(ctx, client.ObjectKey{Name: nodeName, Namespace: "default"}, newSN)).Should(Succeed())
				return newSN.Status.Phase
			}, 10*time.Second, 1*time.Second).Should(Equal(v1alpha1.StorageNodePhaseReady))

			// delete resource
			Expect(k8sClient.Delete(ctx, node)).Should(Succeed())
		})

		It("should delete success", func() {
			nodeName := "test-storage-node-delete"
			node := &v1alpha1.StorageNode{
				ObjectMeta: metav1.ObjectMeta{
					Name:      nodeName,
					Namespace: "default",
					Annotations: map[string]string{
						v1alpha1.AnnotationsInstanceIdentifier: instanceIdentifier,
					},
				},
				Spec: v1alpha1.StorageNodeSpec{
					StorageProviderName: storageProviderName,
				},
			}
			Expect(k8sClient.Create(ctx, node)).Should(Succeed())

			getNode := &v1alpha1.StorageNode{}
			Expect(k8sClient.Get(ctx, client.ObjectKey{Name: nodeName, Namespace: "default"}, getNode)).Should(Succeed())

			// delete storage node
			Expect(k8sClient.Delete(ctx, getNode)).Should(Succeed())
			Eventually(func() bool {
				newSN := &v1alpha1.StorageNode{}
				err := k8sClient.Get(ctx, client.ObjectKey{Name: nodeName, Namespace: "default"}, newSN)
				return err != nil
			}, 10*time.Second, 1*time.Second).Should(BeTrue())
		})

		It("should register and unregister storage unit success", func() {
			// mock mysql
			db, dbmock, err := sqlmock.New()
			Expect(err).Should(Succeed())
			Expect(dbmock).ShouldNot(BeNil())
			// mock rds DescribeDBInstances func returns success
			g := monkey.PatchInstanceMethod(reflect.TypeOf(&aws.RdsClient{}), "GetInstance", func(_ *aws.RdsClient, _ context.Context, _ *v1alpha1.StorageNode) (*dbmeshawsrds.DescInstance, error) {
				return &dbmeshawsrds.DescInstance{
					DBInstanceStatus: dbmeshawsrds.DBInstanceStatusAvailable,
					Endpoint: dbmeshawsrds.Endpoint{
						Address: "127.0.0.1",
						Port:    3306,
					},
				}, nil
			})
			monkey.PatchInstanceMethod(reflect.TypeOf(&aws.RdsClient{}), "DeleteInstance", func(_ *aws.RdsClient, _ context.Context, _ *v1alpha1.StorageNode, _ *v1alpha1.StorageProvider) error {
				return nil
			})
			monkey.Patch(sql.Open, func(_ string, _ string) (*sql.DB, error) {
				return db, nil
			})
			monkey.PatchInstanceMethod(reflect.TypeOf(db), "Close", func(_ *sql.DB) error {
				return nil
			})

			cn := &v1alpha1.ComputeNode{
				ObjectMeta: metav1.ObjectMeta{
					Name:      "test-compute-node",
					Namespace: "default",
					Labels: map[string]string{
						"app": "test-app",
					},
				},
				Spec: v1alpha1.ComputeNodeSpec{
					Selector: &metav1.LabelSelector{
						MatchLabels: map[string]string{
							"app": "test-app",
						},
					},
					PortBindings: []v1alpha1.PortBinding{
						{
							Name:          "http",
							ContainerPort: 3307,
							Protocol:      "TCP",
							ServicePort:   3307,
						},
					},
					Bootstrap: v1alpha1.BootstrapConfig{
						ServerConfig: v1alpha1.ServerConfig{
							Authority: v1alpha1.ComputeNodeAuthority{
								Users: []v1alpha1.ComputeNodeUser{
									{
										User:     "test-user",
										Password: "test-password",
									},
								},
								Privilege: v1alpha1.ComputeNodePrivilege{
									Type: v1alpha1.AllPermitted,
								},
							},
							Props: map[string]string{
								"proxy-frontend-database-protocol-type": "MySQL",
							},
							Mode: v1alpha1.ComputeNodeServerMode{
								Repository: v1alpha1.Repository{
									Type:  "ZooKeeper",
									Props: nil,
								},
								Type: "Zookeeper",
							},
						},
					},
				},
			}

			nodeName := "test-storage-node-register"
			node := &v1alpha1.StorageNode{
				ObjectMeta: metav1.ObjectMeta{
					Name:      nodeName,
					Namespace: "default",
					Annotations: map[string]string{
						v1alpha1.AnnotationsInstanceIdentifier:              instanceIdentifier,
						controllers.AnnotationKeyRegisterStorageUnitEnabled: "true",
						v1alpha1.AnnotationsInstanceDBName:                  "test-db-name",
						controllers.AnnotationKeyComputeNodeName:            "test-compute-node",
						controllers.AnnotationKeyLogicDatabaseName:          "test-logic-db-name",
					},
				},
				Spec: v1alpha1.StorageNodeSpec{
					StorageProviderName: storageProviderName,
				},
			}

			Expect(k8sClient.Create(ctx, cn)).Should(Succeed())
			Expect(k8sClient.Create(ctx, node)).Should(Succeed())

			dbmock.ExpectExec(regexp.QuoteMeta("CREATE DATABASE IF NOT EXISTS")).WillReturnResult(sqlmock.NewResult(1, 1))
			dbmock.ExpectExec(regexp.QuoteMeta("USE")).WillReturnResult(sqlmock.NewResult(1, 1))
			dbmock.ExpectExec(regexp.QuoteMeta("REGISTER STORAGE UNIT IF NOT EXISTS")).WillReturnResult(sqlmock.NewResult(0, 0))

			Eventually(func() v1alpha1.StorageNodePhaseStatus {
				newSN := &v1alpha1.StorageNode{}
				Expect(k8sClient.Get(ctx, client.ObjectKey{Name: nodeName, Namespace: "default"}, newSN)).Should(Succeed())
				return newSN.Status.Phase
			}, time.Second*10, time.Millisecond*250).Should(Equal(v1alpha1.StorageNodePhaseReady))

			Eventually(func() bool {
				newSN := &v1alpha1.StorageNode{}
				Expect(k8sClient.Get(ctx, client.ObjectKey{Name: nodeName, Namespace: "default"}, newSN)).Should(Succeed())
				return newSN.Status.Registered
			}, 20, 2).Should(BeTrue())

			// delete storage node
			Expect(k8sClient.Delete(ctx, node)).Should(Succeed())

			dbmock.ExpectExec(regexp.QuoteMeta("USE")).WillReturnResult(sqlmock.NewResult(1, 1))
			dbmock.ExpectQuery(regexp.QuoteMeta("SHOW RULES USED STORAGE UNIT")).WillReturnRows(sqlmock.NewRows([]string{"type", "name"}).AddRow("sharding", "t_order"))
			dbmock.ExpectExec("DROP SHARDING TABLE RULE").WillReturnResult(sqlmock.NewResult(1, 1))
			dbmock.ExpectExec(regexp.QuoteMeta("UNREGISTER STORAGE UNIT")).WillReturnResult(sqlmock.NewResult(0, 0))
			Eventually(func() v1alpha1.StorageNodePhaseStatus {
				newSN := &v1alpha1.StorageNode{}
				Expect(k8sClient.Get(ctx, client.ObjectKey{Name: nodeName, Namespace: "default"}, newSN)).Should(Succeed())
				return newSN.Status.Phase
			}, 20, 2).Should(Equal(v1alpha1.StorageNodePhaseDeleting))

			g.Unpatch()
			monkey.PatchInstanceMethod(reflect.TypeOf(&aws.RdsClient{}), "GetInstance", func(_ *aws.RdsClient, _ context.Context, _ *v1alpha1.StorageNode) (*dbmeshawsrds.DescInstance, error) {
				return nil, nil
			})

			Eventually(func() bool {
				newSN := &v1alpha1.StorageNode{}
				err := k8sClient.Get(ctx, client.ObjectKey{Name: nodeName, Namespace: "default"}, newSN)
				return apierrors.IsNotFound(err)
			}, 20, 2).Should(BeTrue())
		})
	})
})

var _ = Describe("StorageNode Controller Suite Test For AWS Aurora Cluster", func() {
	storageProviderName := "test-storage-provider"
	clusterIdentifier := "test-aurora-cluster-identifier"

	BeforeEach(func() {
		provider := &v1alpha1.StorageProvider{
			ObjectMeta: metav1.ObjectMeta{
				Name: storageProviderName,
			},
			Spec: v1alpha1.StorageProviderSpec{
				Provisioner: v1alpha1.ProvisionerAWSAurora,
				Parameters: map[string]string{
					"engine":             "aurora-mysql",
					"engineVersion":      "5.7",
					"instanceClass":      "db.t2.small",
					"masterUsername":     "test-user",
					"masterUserPassword": "test-password",
				},
			},
		}
		Expect(k8sClient.Create(ctx, provider)).Should(Succeed())
	})

	AfterEach(func() {
		monkey.UnpatchAll()

		StorageProvider := &v1alpha1.StorageProvider{}
		Expect(k8sClient.Get(ctx, client.ObjectKey{Name: storageProviderName}, StorageProvider)).Should(Succeed())
		Expect(k8sClient.Delete(ctx, StorageProvider)).Should(Succeed())
	})

	Context("When Creat StorageNode with Aurora Cluster Not Exist", func() {
		It("Should Success", func() {
			snName := "test-storage-node-creating"
			// monkey patch
			monkey.PatchInstanceMethod(reflect.TypeOf(&aws.RdsClient{}), "GetAuroraCluster", func(_ *aws.RdsClient, _ context.Context, _ *v1alpha1.StorageNode) (*dbmeshawsrds.DescCluster, error) {
				return &dbmeshawsrds.DescCluster{
					DBClusterIdentifier: clusterIdentifier,
					Status:              string(dbmeshawsrds.DBClusterStatusCreating),
					PrimaryEndpoint:     "test-primary-endpoint",
					ReaderEndpoint:      "test-reader-endpoint",
					Port:                3306,
				}, nil
			})
			monkey.PatchInstanceMethod(reflect.TypeOf(&aws.RdsClient{}), "GetInstancesByFilters", func(_ *aws.RdsClient, _ context.Context, _ map[string][]string) ([]*dbmeshawsrds.DescInstance, error) {
				return []*dbmeshawsrds.DescInstance{
					{
						DBInstanceIdentifier: fmt.Sprintf("%s-insatnce-0", clusterIdentifier),
						DBInstanceStatus:     dbmeshawsrds.DBInstanceStatusCreating,
						Endpoint:             dbmeshawsrds.Endpoint{Address: "test-instance-0-endpoint", Port: 3306},
					},
				}, nil
			})

			sn := &v1alpha1.StorageNode{
				ObjectMeta: metav1.ObjectMeta{
					Name:      snName,
					Namespace: "default",
					Annotations: map[string]string{
						v1alpha1.AnnotationsClusterIdentifier: clusterIdentifier,
					},
				},
				Spec: v1alpha1.StorageNodeSpec{
					StorageProviderName: storageProviderName,
					Replicas:            2,
				},
			}
			Expect(k8sClient.Create(ctx, sn)).Should(Succeed())

			Eventually(func() string {
				newSN := &v1alpha1.StorageNode{}
				Expect(k8sClient.Get(ctx, client.ObjectKey{Name: snName, Namespace: "default"}, newSN)).Should(Succeed())
				return newSN.Status.Cluster.Status
			}, time.Second*10, time.Millisecond*250).Should(Equal(string(dbmeshawsrds.DBClusterStatusCreating)))
		})

		It("should success when cluster is available", func() {
			snName := "test-storage-node-available"
			monkey.PatchInstanceMethod(reflect.TypeOf(&aws.RdsClient{}), "GetAuroraCluster", func(_ *aws.RdsClient, _ context.Context, _ *v1alpha1.StorageNode) (*dbmeshawsrds.DescCluster, error) {
				return &dbmeshawsrds.DescCluster{
					DBClusterIdentifier: clusterIdentifier,
					Status:              string(dbmeshawsrds.DBClusterStatusAvailable),
					PrimaryEndpoint:     "test-primary-endpoint",
					ReaderEndpoint:      "test-reader-endpoint",
					Port:                3306,
				}, nil
			})
			monkey.PatchInstanceMethod(reflect.TypeOf(&aws.RdsClient{}), "GetInstancesByFilters", func(_ *aws.RdsClient, _ context.Context, _ map[string][]string) ([]*dbmeshawsrds.DescInstance, error) {
				return []*dbmeshawsrds.DescInstance{
					{
						DBInstanceIdentifier: fmt.Sprintf("%s-insatnce-0", clusterIdentifier),
						DBInstanceStatus:     dbmeshawsrds.DBInstanceStatusAvailable,
						Endpoint:             dbmeshawsrds.Endpoint{Address: "test-instance-0-endpoint", Port: 3306},
					},
					{
						DBInstanceIdentifier: fmt.Sprintf("%s-insatnce-1", clusterIdentifier),
						DBInstanceStatus:     dbmeshawsrds.DBInstanceStatusAvailable,
						Endpoint:             dbmeshawsrds.Endpoint{Address: "test-instance-1-endpoint", Port: 3306},
					},
				}, nil
			})
			monkey.PatchInstanceMethod(reflect.TypeOf(&aws.RdsClient{}), "DeleteAuroraCluster", func(_ *aws.RdsClient, _ context.Context, _ *v1alpha1.StorageNode, _ *v1alpha1.StorageProvider) error {
				return nil
			})

			sn := &v1alpha1.StorageNode{
				ObjectMeta: metav1.ObjectMeta{
					Name:      snName,
					Namespace: "default",
					Annotations: map[string]string{
						v1alpha1.AnnotationsClusterIdentifier: clusterIdentifier,
					},
				},
				Spec: v1alpha1.StorageNodeSpec{
					StorageProviderName: storageProviderName,
					Replicas:            2,
				},
			}
			Expect(k8sClient.Create(ctx, sn)).Should(Succeed())

			newSN := &v1alpha1.StorageNode{}
			Eventually(func() v1alpha1.StorageNodePhaseStatus {
				Expect(k8sClient.Get(ctx, client.ObjectKey{Name: snName, Namespace: "default"}, newSN)).Should(Succeed())
				return newSN.Status.Phase
			}, time.Second*10, time.Millisecond*250).Should(Equal(v1alpha1.StorageNodePhaseReady))

			Expect(newSN.Status.Instances).Should(HaveLen(2))
		})
	})
})
