package openapidedicated

import (
	"fmt"
	"strings"
	"testing"
	"time"

	"github.com/samber/lo"
	"github.com/spf13/viper"

	dedicatedapi "github.com/tidbcloud/tidb-management-service/functional-test/pkg/dedicatedapi/api"
	"github.com/tidbcloud/tidb-management-service/functional-test/pkg/dedicatedapi/api/client"
	"github.com/tidbcloud/tidb-management-service/functional-test/pkg/sql"
	"github.com/tidbcloud/tidb-management-service/utilities/randstr"
)

// TestDedicatedAwsStoreAuditLogToPingCAP verifies end-to-end behavior of tidb cluster setting storing audit log to PingCAP use case
// and get audit log download link.
func TestDedicatedAwsStoreAuditLogToPingCAP(t *testing.T) {
	tester := &dedicatedAwsStoreAuditLogToPingCAPTest{
		T:                 t,
		host:              viper.GetString("host"),
		apiKey:            viper.GetString("api_key"),
		prjID:             viper.GetString("prj_id"),
		clusterNamePrefix: viper.GetString("cluster_name_prefix"),
		provider:          "aws",
		regionCode:        "us-west-2",
		password:          randstr.GenWithCharset(16, randstr.AlphanumericCharset),
		port:              4000,
		tidbNodeSpec:      &client.V1beta1NodeSpec{NodeSpecKey: lo.ToPtr("2C8G")},
		tikvNodeSpec: &client.V1beta1NodeSpec{
			NodeSpecKey:          lo.ToPtr("2C8G"),
			DefaultStorageSizeGi: lo.ToPtr(int32(200)),
		},
	}

	tester.clusterName = tester.clusterNamePrefix + randstr.GenWithCharset(4, randstr.AlphabetCharset)
	tester.cli = dedicatedapi.NewClientT(tester.T, tester.host, tester.apiKey)

	tests := []struct {
		name string
		fn   func(t *testing.T)
	}{
		{"00_EnsureNetworkContainer", tester.Test_00_EnsureNetworkContainer},
		{"01_CreateCluster", tester.Test_01_CreateCluster},
		{"02_PollClusterUntilActive", tester.Test_02_PollClusterUntilActive},
		{"03_EnableAuditLogToPingCAP", tester.Test_03_EnableAuditLogToPingCAP},
		{"04_SetPublicNetworkAccess", tester.Test_04_SetPublicNetworkAccess},
		{"05_ExecuteSql", tester.Test_05_ExecuteSql},
		{"06_PauseAndResumeCluster", tester.Test_06_PauseAndResumeCluster},
		{"07_ListAuditLogsAndGetDownloadAddress", tester.Test_07_ListAuditLogsAndGetDownloadAddress},
	}

	for _, tt := range tests {
		if t.Failed() {
			t.Logf("Skipping test %s due to previous failure.", tt.name)
			continue
		}
		t.Run(tt.name, tt.fn)
	}
}

type dedicatedAwsStoreAuditLogToPingCAPTest struct {
	*testing.T

	host              string
	apiKey            string
	prjID             string
	clusterNamePrefix string
	provider          string
	regionCode        string
	password          string
	clusterName       string
	port              int32
	cli               *dedicatedapi.Cli
	tidbNodeSpec      *client.V1beta1NodeSpec
	tikvNodeSpec      *client.V1beta1NodeSpec
	clusterId         string
}

func (su *dedicatedAwsStoreAuditLogToPingCAPTest) Test_00_EnsureNetworkContainer(t *testing.T) {
	req := su.cli.NetworkContainerServiceAPI.NetworkContainerServiceListNetworkContainers(t.Context()).
		ProjectId(su.prjID).
		CloudProvider(su.provider).
		PageSize(pageSize)
	resp, _, err := req.Execute()
	if err != nil {
		t.Fatalf("failed to list network containers: %v", err)
	}
	isProjectAndCloudProviderValidForAll := lo.Reduce(
		resp.GetNetworkContainers(),
		func(agg bool, item client.V1beta1NetworkContainer, _ int) bool {
			pid, ok := item.GetLabels()["tidb.cloud/project"]
			return agg &&
				ok && pid == su.prjID && // Check result's project ID.
				string(item.GetCloudProvider()) == su.provider // Check result's cloud provider.
		},
		true,
	)
	if !isProjectAndCloudProviderValidForAll {
		t.Fatal("network container check failed")
	}
}

func (su *dedicatedAwsStoreAuditLogToPingCAPTest) Test_01_CreateCluster(t *testing.T) {
	// Create a cluster.
	req := su.cli.ClusterServiceAPI.ClusterServiceCreateCluster(t.Context()).
		Cluster(client.TidbCloudOpenApidedicatedv1beta1Cluster{
			DisplayName: su.clusterName,
			RegionId:    fmt.Sprintf("%s-%s", su.provider, su.regionCode),
			Labels:      lo.ToPtr(map[string]string{"tidb.cloud/project": su.prjID}),
			TidbNodeSetting: client.V1beta1ClusterTidbNodeSetting{
				NodeSpecKey: lo.ToPtr(su.tidbNodeSpec.GetNodeSpecKey()),
				TidbNodeGroups: []client.Dedicatedv1beta1TidbNodeGroup{
					{
						NodeCount: 1,
					},
				},
			},
			TikvNodeSetting: client.V1beta1ClusterStorageNodeSetting{
				NodeCount:     3,
				NodeSpecKey:   su.tikvNodeSpec.GetNodeSpecKey(),
				StorageSizeGi: su.tikvNodeSpec.GetDefaultStorageSizeGi(),
				StorageType:   lo.ToPtr(client.CLUSTERSTORAGENODESETTINGSTORAGETYPE_STANDARD),
			},
			Port:         su.port,
			RootPassword: lo.ToPtr(su.password),
		})

	resp, _, err := req.Execute()
	if err != nil {
		t.Fatalf("failed to create cluster: %v", err)
	}
	if resp.GetDisplayName() != su.clusterName {
		t.Fatalf("expected display name %s, got %s", su.clusterName, resp.GetDisplayName())
	}
	if resp.GetRegionId() != fmt.Sprintf("%s-%s", su.provider, su.regionCode) {
		t.Fatalf("expected region id %s, got %s", fmt.Sprintf("%s-%s", su.provider, su.regionCode), resp.GetRegionId())
	}
	if resp.GetState() != client.COMMONV1BETA1CLUSTERSTATE_CREATING {
		t.Fatalf("expected cluster state CREATING, got %s", resp.GetState())
	}

	su.clusterId = resp.GetClusterId()
}

func (su *dedicatedAwsStoreAuditLogToPingCAPTest) Test_02_PollClusterUntilActive(t *testing.T) {
	// List clusters to find the cluster.
	req := su.cli.ClusterServiceAPI.ClusterServiceListClusters(t.Context()).
		ProjectId(su.prjID).
		PageSize(pageSize)
	resp, _, err := req.Execute()
	if err != nil {
		t.Fatalf("failed to list clusters: %v", err)
	}
	if !lo.ContainsBy(resp.Clusters, func(item client.TidbCloudOpenApidedicatedv1beta1Cluster) bool {
		return item.GetClusterId() == su.clusterId
	}) {
		t.Fatalf("cluster %s not found in list", su.clusterId)
	}

	// Poll the cluster using GET method until it becomes active.
	// Timeout: 30 minutes (1800 seconds).
	for range 180 {
		req := su.cli.ClusterServiceAPI.ClusterServiceGetCluster(t.Context(), su.clusterId)
		resp, _, err := req.Execute()
		if err != nil {
			t.Fatalf("failed to get cluster: %v", err)
		}

		switch resp.GetState() {
		case client.COMMONV1BETA1CLUSTERSTATE_ACTIVE:
			return
		case client.COMMONV1BETA1CLUSTERSTATE_CREATING:
			t.Logf("Cluster state is CREATING")
		default:
			t.Fatalf("Cluster state is not ACTIVE or CREATING, but %s", resp.GetState())
		}
		time.Sleep(10 * time.Second)
	}

	t.Fatal("Cluster did not become active within the timeout period")
}

func (su *dedicatedAwsStoreAuditLogToPingCAPTest) Test_03_EnableAuditLogToPingCAP(t *testing.T) {
	req := su.cli.DatabaseAuditLogServiceAPI.DatabaseAuditLogServiceUpdateAuditLogConfig(t.Context(), su.clusterId).
		AuditLogConfig(client.DatabaseAuditLogServiceUpdateAuditLogConfigRequest{
			Enabled:       *client.NewNullableBool(lo.ToPtr(true)),
			BucketManager: lo.ToPtr(client.AUDITLOGCONFIGBUCKETMANAGER_PINGCAP),
		})

	resp, _, err := req.Execute()
	if err != nil {
		t.Fatalf("failed to update audit log config: %v", err)
	}
	if !resp.GetEnabled() {
		t.Fatal("expected audit log to be enabled")
	}
	if resp.GetBucketManager() != client.AUDITLOGCONFIGBUCKETMANAGER_PINGCAP {
		t.Fatalf("expected bucket manager to be PINGCAP, got %s", resp.GetBucketManager())
	}

	// Wait for a while to make sure the audit log is enabled.
	enabled := false
	for range 30 {
		req := su.cli.DatabaseAuditLogServiceAPI.DatabaseAuditLogServiceGetAuditLogConfig(t.Context(), su.clusterId)
		resp, _, err := req.Execute()
		if err != nil {
			t.Fatalf("failed to get audit log config: %v", err)
		}
		if resp.GetEnabled() && resp.GetBucketManager() == client.AUDITLOGCONFIGBUCKETMANAGER_PINGCAP {
			enabled = true
			break
		}
		time.Sleep(10 * time.Second)
	}
	if !enabled {
		t.Fatal("store audit log to PingCAP failed")
	}

	ruleReq := su.cli.DatabaseAuditLogServiceAPI.DatabaseAuditLogServiceCreateAuditLogFilterRule(t.Context(), su.clusterId).
		AuditLogFilterRule(client.DatabaseAuditLogServiceCreateAuditLogFilterRuleRequest{
			UserExpr: lo.ToPtr("root@%"),
		})
	ruleResp, _, err := ruleReq.Execute()
	if err != nil {
		t.Fatalf("failed to create audit log filter rule: %v", err)
	}
	if ruleResp.GetAuditLogFilterRuleId() == "" {
		t.Fatal("expected audit log filter rule id to be not empty")
	}
	time.Sleep(20 * time.Second)
}

func (su *dedicatedAwsStoreAuditLogToPingCAPTest) Test_04_SetPublicNetworkAccess(t *testing.T) {
	// Get the cluster first.
	getClusterReq := su.cli.ClusterServiceAPI.ClusterServiceGetCluster(t.Context(), su.clusterId)
	cluster, _, err := getClusterReq.Execute()
	if err != nil {
		t.Fatalf("failed to get cluster: %v", err)
	}
	if len(cluster.TidbNodeSetting.GetTidbNodeGroups()) != 1 {
		t.Fatalf("expected 1 tidb node group, got %d", len(cluster.TidbNodeSetting.GetTidbNodeGroups()))
	}

	group := cluster.TidbNodeSetting.GetTidbNodeGroups()[0]

	nullableTrue := client.NullableBool{}
	nullableTrue.Set(lo.ToPtr(true))

	updatePublicAccessReq := su.cli.TidbNodeGroupServiceAPI.TidbNodeGroupServiceUpdatePublicEndpointSetting(t.Context(), su.clusterId, group.GetTidbNodeGroupId()).
		PublicEndpointSetting(client.TidbNodeGroupServiceUpdatePublicEndpointSettingRequest{
			Enabled: nullableTrue,
			IpAccessList: []client.V1beta1PublicEndpointSettingIpAccessList{
				{
					CidrNotation: "0.0.0.0/0",
					Description:  lo.ToPtr("allow access from anywhere."),
				},
			},
		})
	_, _, err = updatePublicAccessReq.Execute()
	if err != nil {
		t.Fatalf("failed to update public endpoint setting: %v", err)
	}
}

func (su *dedicatedAwsStoreAuditLogToPingCAPTest) Test_05_ExecuteSql(t *testing.T) {
	for range 30 {
		time.Sleep(10 * time.Second)

		// List TiDB groups first.
		listGroupsReq := su.cli.TidbNodeGroupServiceAPI.TidbNodeGroupServiceListTidbNodeGroups(t.Context(), su.clusterId).
			PageSize(pageSize)
		listGroupsResp, _, err := listGroupsReq.Execute()
		if err != nil {
			t.Fatalf("failed to list tidb node groups: %v", err)
		}
		if len(listGroupsResp.GetTidbNodeGroups()) != 1 {
			t.Fatalf("expected 1 tidb node group, got %d", len(listGroupsResp.GetTidbNodeGroups()))
		}

		pubEndpoint, ok := lo.Find(
			listGroupsResp.GetTidbNodeGroups()[0].GetEndpoints(),
			func(item client.Dedicatedv1beta1TidbNodeGroupEndpoint) bool {
				return item.GetConnectionType() == client.V1BETA1TIDBNODEGROUPENDPOINTCONNECTIONTYPE_PUBLIC
			},
		)
		if !ok {
			t.Fatal("public endpoint not found")
		}

		// Skip if the public endpoint is not ready.
		if pubEndpoint.GetHost() == "" {
			continue
		}

		err = sql.ExecuteSqlWithUrl(t.Context(), pubEndpoint.GetHost(), "", su.password, "skip-verify", su.port)

		switch {
		case err == nil:
			// Success.
			return
		case strings.Contains(err.Error(), "connection refused"),
			strings.Contains(err.Error(), "no such host"),
			strings.Contains(err.Error(), "i/o timeout"):
			// Retry if the connection is refused.
			// Dns propagation may take some time.
		default:
			t.Fatalf("failed to execute SQL: %v", err)
		}
	}

	t.Fatal("Failed to execute SQL after multiple attempts")
}

func (su *dedicatedAwsStoreAuditLogToPingCAPTest) Test_06_PauseAndResumeCluster(t *testing.T) {
	// Pause the cluster.
	req := su.cli.ClusterServiceAPI.ClusterServicePauseCluster(t.Context(), su.clusterId)
	_, _, err := req.Execute()
	if err != nil {
		t.Fatalf("failed to pause cluster: %v", err)
	}

	// Wait for a while to make sure the cluster is paused.
	paused := false
	for range 100 {
		req := su.cli.ClusterServiceAPI.ClusterServiceGetCluster(t.Context(), su.clusterId)
		resp, _, err := req.Execute()
		if err != nil {
			t.Fatalf("failed to get cluster: %v", err)
		}
		if resp.GetState() == client.COMMONV1BETA1CLUSTERSTATE_PAUSED {
			paused = true
			break
		}
		time.Sleep(10 * time.Second)
	}
	if !paused {
		t.Fatal("Pause cluster failed")
	}

	// Resume the cluster.
	resumeReq := su.cli.ClusterServiceAPI.ClusterServiceResumeCluster(t.Context(), su.clusterId)
	_, _, err = resumeReq.Execute()
	if err != nil {
		t.Fatalf("failed to resume cluster: %v", err)
	}
	for range 100 {
		req := su.cli.ClusterServiceAPI.ClusterServiceGetCluster(t.Context(), su.clusterId)
		resp, _, err := req.Execute()
		if err != nil {
			t.Fatalf("failed to get cluster: %v", err)
		}
		if resp.GetState() == client.COMMONV1BETA1CLUSTERSTATE_ACTIVE {
			return
		}
		time.Sleep(10 * time.Second)
	}
	t.Fatal("Resume cluster failed")
}

func (su *dedicatedAwsStoreAuditLogToPingCAPTest) Test_07_ListAuditLogsAndGetDownloadAddress(t *testing.T) {
	existing := false
	for range 30 {
		req := su.cli.DatabaseAuditLogServiceAPI.DatabaseAuditLogServiceQueryAuditLogFiles(t.Context(), su.clusterId).
			StartTime(time.Now().Add(-24 * time.Hour)).
			EndTime(time.Now()).
			PageSize(pageSize)
		resp, _, err := req.Execute()
		if err != nil {
			t.Fatalf("failed to query audit log files: %v", err)
		}
		if len(resp.GetAuditLogFiles()) > 0 {
			for _, log := range resp.GetAuditLogFiles() {
				if log.GetFilename() == "" {
					t.Fatal("expected filename to be not empty")
				}
				if log.GetUri() == "" {
					t.Fatal("expected uri to be not empty")
				}
				if log.GetCreateTime().IsZero() {
					t.Fatal("expected create time to be not empty")
				}
				if log.GetSizeByte() == 0 {
					t.Fatal("expected size byte to be not zero")
				}
				if log.GetBucket() == "" {
					t.Fatal("expected bucket to be not empty")
				}

				req := su.cli.DatabaseAuditLogServiceAPI.DatabaseAuditLogServiceGenerateAuditLogFileDownloadAddress(t.Context(), su.clusterId).
					AuditLogFileUri(log.GetUri())
				resp, _, err := req.Execute()
				if err != nil {
					t.Fatalf("failed to generate download address: %v", err)
				}
				if resp.GetDownloadAddress() == "" {
					t.Fatal("expected download address to be not empty")
				}
			}
			existing = true
			break
		}
		time.Sleep(10 * time.Second)
	}
	if !existing {
		t.Fatal("No audit logs found")
	}
}