package repository_ddb

import (
	"context"
	"dataservices/commons/errs"
	"dataservices/ddb"
	"dataservices/internal/entity"
	"fmt"
	"github.com/aws/aws-sdk-go-v2/aws"
	"github.com/aws/aws-sdk-go-v2/service/dynamodb"
	"github.com/aws/aws-sdk-go-v2/service/dynamodb/types"
	"github.com/stretchr/testify/require"
	"github.com/tidbcloud/tiinf/monitor/tilog"
	"sort"
	"testing"
	"time"
)

func TestDynamoDB(t *testing.T) {
	ctx := context.Background()
	client, err := newDynamoDBClientForTest(ctx)
	require.NoError(t, err)

	listTablesOutput, err := client.ListTables(ctx, &dynamodb.ListTablesInput{})
	require.NoError(t, err)
	fmt.Printf("list tables: %#v ---\n", listTablesOutput.TableNames)

	err = CreateUserClusterInfoTableIfNotExist(ctx, client)
	require.NoError(t, err)
	defer func() {
		ddb.DeleteTable(ctx, client, entity.TableUserClusterInfo)
	}()

	clusterId := uint64(789019)
	info := entity.UserClusterInfo{
		ProjectId:       123,
		ClusterId:       clusterId,
		IsDedicate:      true,
		EndpointService: "endpoint service name",
		Host:            "?",
		Port:            4000,
	}
	err = PutUserClusterInfo(ctx, client, &info)
	require.NoError(t, err)

	result, err := QueryUserClusterInfo(ctx, client, 123, clusterId)
	require.NoError(t, err)
	fmt.Printf("query: %#v  ---\n\n", result)
	require.Equal(t, 1, len(result))
	require.Equal(t, info, result[0])

	getInfo, err := GetUserClusterInfo(ctx, client, 123, clusterId)
	require.NoError(t, err)
	fmt.Printf("get info: %#v  ---\n\n", getInfo)
	require.Equal(t, info, *getInfo)

	// Query not exist info.
	result, err = QueryUserClusterInfo(ctx, client, 123, clusterId+123)
	require.NoError(t, err)
	require.Equal(t, 0, len(result))

	// Get not exist info.
	getInfo, err = GetUserClusterInfo(ctx, client, 123, clusterId+123)
	require.Error(t, errs.ErrUserClusterNotFound, err)

	// update not exist info.
	err = UpdateUserClusterInfo(ctx, client, &entity.UserClusterInfo{
		ProjectId:       123,
		ClusterId:       clusterId + 123,
		IsDedicate:      true,
		EndpointService: "endpoint service addr",
		EndpointId:      "id??",
		PrivateDNS:      "dns??",
	})
	// won't return error even the record not exist, UpdateUserClusterInfo will auto create an record.
	require.NoError(t, err)

	info.PrivateDNS = "new private dns name"
	err = PutUserClusterInfo(ctx, client, &info)
	require.NoError(t, err)
	getInfo, err = GetUserClusterInfo(ctx, client, 123, clusterId)
	require.NoError(t, err)
	fmt.Printf("get info: %#v  ---\n\n", getInfo)
	require.Equal(t, info, *getInfo)

	clusterInfos, err := ScanUserClusterInfo(ctx, client)
	require.NoError(t, err)
	for _, info := range clusterInfos {
		fmt.Printf("scan info: %v \n", info)
	}
}

func TestQueryUserClusterInfoWithPendingPrivateLink(t *testing.T) {
	ctx := context.Background()
	cli, err := newDynamoDBClientForTest(ctx)
	require.NoError(t, err)

	_ = ddb.DeleteTable(ctx, cli, entity.TableUserClusterInfo)
	for {
		// wait table finish deleted.
		exist, err := ddb.IsTableExist(ctx, cli, entity.TableUserClusterInfo)
		require.NoError(t, err)
		if !exist {
			break
		}
	}
	var pendingInfos []entity.UserClusterInfo
	err = CreateUserClusterInfoTableIfNotExist(ctx, cli)
	require.NoError(t, err)

	info1 := entity.UserClusterInfo{
		ProjectId:       123,
		ClusterId:       456,
		IsDedicate:      true,
		EndpointService: "endpoint service name",
	}
	err = PutUserClusterInfo(ctx, cli, &info1)
	require.NoError(t, err)

	info2 := info1
	info2.ClusterId = 789
	info2.EndpointId = "endpoint-id-2"
	err = PutUserClusterInfo(ctx, cli, &info2)
	require.NoError(t, err)

	info3 := info1
	info3.ClusterId = 1350
	info3.EndpointId = "endpoint-id-3"
	err = PutUserClusterInfo(ctx, cli, &info3)
	require.NoError(t, err)

	info4 := info1
	info4.ClusterId = 5790
	info4.EndpointId = "endpoint-id-4"
	info4.PrivateDNS = "private-dns-4"
	err = PutUserClusterInfo(ctx, cli, &info4)
	require.NoError(t, err)

	infos, err := ScanUserClusterInfo(ctx, cli)
	require.NoError(t, err)
	require.Equal(t, 4, len(infos))

	pendingInfos, err = QueryUserClusterInfoWithPendingPrivateLink(ctx, cli)
	require.NoError(t, err)
	require.Equal(t, 2, len(pendingInfos))
	sort.Slice(pendingInfos, func(i, j int) bool {
		return pendingInfos[i].ClusterId < pendingInfos[j].ClusterId
	})
	require.Equal(t, info2, pendingInfos[0])
	require.Equal(t, info3, pendingInfos[1])

	info3.PrivateDNS = "private-dns-3"
	err = PutUserClusterInfo(ctx, cli, &info3)
	require.NoError(t, err)

	info5 := info1
	info5.ClusterId = 5791
	info5.EndpointId = "endpoint-id-5"
	err = PutUserClusterInfo(ctx, cli, &info5)

	info6 := info1
	info6.ClusterId = 5792
	info6.EndpointId = "endpoint-id-6"
	err = PutUserClusterInfo(ctx, cli, &info6)

	pendingInfos, err = QueryUserClusterInfoWithPendingPrivateLink(ctx, cli)
	require.NoError(t, err)
	require.Equal(t, 3, len(pendingInfos))
	sort.Slice(pendingInfos, func(i, j int) bool {
		return pendingInfos[i].ClusterId < pendingInfos[j].ClusterId
	})
	require.Equal(t, info2, pendingInfos[0])
	require.Equal(t, info5, pendingInfos[1])
	require.Equal(t, info6, pendingInfos[2])

	err = ddb.DeleteTable(ctx, cli, entity.TableUserClusterInfo)
	require.NoError(t, err)
}

func TestDescribeTable(t *testing.T) {
	ctx := context.Background()
	cli, err := newDynamoDBClientForTest(ctx)
	require.NoError(t, err)

	exist, err := ddb.IsTableExist(ctx, cli, "t-not-exist")
	require.NoError(t, err)
	require.Equal(t, false, exist)

	exist, err = ddb.IsTableExist(ctx, cli, "endpoint_publish")
	require.NoError(t, err)
	require.Equal(t, true, exist)

	resp, err := cli.DescribeTable(ctx, &dynamodb.DescribeTableInput{TableName: aws.String("endpoint_publish")})
	require.NoError(t, err)
	fmt.Printf("%#v ---\n\n", resp.Table)
	require.NotNil(t, resp)

	fmt.Printf("%v \n", *resp.Table.TableName)
	fmt.Printf("%v \n", *resp.Table.TableId)
	fmt.Printf("---def--\n")
	for _, v := range resp.Table.AttributeDefinitions {
		fmt.Printf("%v  %v\n", *v.AttributeName, v.AttributeType)
	}
	fmt.Printf("---key--\n")
	for _, v := range resp.Table.KeySchema {
		fmt.Printf("%v  %v\n", *v.AttributeName, v.KeyType)
	}
	fmt.Printf("---key finish--\n")
	if resp.Table.GlobalTableVersion == nil {
		fmt.Printf("global is nil: %v \n", resp.Table.GlobalTableVersion)
	} else {
		fmt.Printf("global: %v \n", *resp.Table.GlobalTableVersion)
	}
	fmt.Printf("ProvisionedThroughput \n")
	if resp.Table.ProvisionedThroughput != nil {
		fmt.Printf("r %v \n", *resp.Table.ProvisionedThroughput.ReadCapacityUnits)
		fmt.Printf("w %v \n", *resp.Table.ProvisionedThroughput.WriteCapacityUnits)
	}
	fmt.Printf("%v \n", resp.Table.TableStatus)
	if resp.Table.BillingModeSummary != nil {
		fmt.Printf("billmode: %v \n", resp.Table.BillingModeSummary.BillingMode)
	}
}

func TestGlobalTable(t *testing.T) {
	ctx := context.Background()
	region1 := "us-west-1"
	region2 := "us-west-2"
	cli1, err := newDynamoDBClientForTestByRegion(ctx, region1)
	require.NoError(t, err)
	cli2, err := newDynamoDBClientForTestByRegion(ctx, region2)
	require.NoError(t, err)

	allGlobalTables, err := cli1.ListGlobalTables(ctx, &dynamodb.ListGlobalTablesInput{})
	require.NoError(t, err)
	for _, gt := range allGlobalTables.GlobalTables {
		regions := []string{}
		for _, g := range gt.ReplicationGroup {
			regions = append(regions, *g.RegionName)
		}
		fmt.Printf("global tables: %v    %v --\n", *gt.GlobalTableName, regions)
	}

	defer func() {
		cli1.DeleteTable(ctx, &dynamodb.DeleteTableInput{
			TableName: aws.String(entity.TableUserClusterInfo),
		})
		cli2.DeleteTable(ctx, &dynamodb.DeleteTableInput{
			TableName: aws.String(entity.TableUserClusterInfo),
		})
	}()

	cli1.DeleteTable(ctx, &dynamodb.DeleteTableInput{
		TableName: aws.String(entity.TableUserClusterInfo),
	})
	cli2.DeleteTable(ctx, &dynamodb.DeleteTableInput{
		TableName: aws.String(entity.TableUserClusterInfo),
	})
	for {
		exist1, err := ddb.IsTableExist(ctx, cli1, entity.TableUserClusterInfo)
		require.NoError(t, err)
		exist2, err := ddb.IsTableExist(ctx, cli2, entity.TableUserClusterInfo)
		require.NoError(t, err)
		if !exist1 && !exist2 {
			break
		}
		time.Sleep(time.Millisecond * 500)
	}

	createTableInput := &dynamodb.CreateTableInput{
		AttributeDefinitions: []types.AttributeDefinition{{
			AttributeName: aws.String("project_id"),
			AttributeType: types.ScalarAttributeTypeN,
		}, {
			AttributeName: aws.String("cluster_id"),
			AttributeType: types.ScalarAttributeTypeN,
		}},
		KeySchema: []types.KeySchemaElement{{
			AttributeName: aws.String("project_id"),
			KeyType:       types.KeyTypeHash,
		}, {
			AttributeName: aws.String("cluster_id"),
			KeyType:       types.KeyTypeRange,
		}},
		TableName: aws.String(entity.TableUserClusterInfo),
		ProvisionedThroughput: &types.ProvisionedThroughput{
			ReadCapacityUnits:  aws.Int64(10),
			WriteCapacityUnits: aws.Int64(10),
		},
		StreamSpecification: &types.StreamSpecification{
			StreamEnabled:  aws.Bool(true),
			StreamViewType: types.StreamViewTypeNewAndOldImages,
		},
	}

	_, err = ddb.CreateTable(ctx, cli1, createTableInput)
	require.NoError(t, err)

	_, err = ddb.CreateTable(ctx, cli2, createTableInput)
	require.NoError(t, err)

	allGlobalTables, err = cli1.ListGlobalTables(ctx, &dynamodb.ListGlobalTablesInput{})
	require.NoError(t, err)
	fmt.Printf("global tables num: %v  --\n", len(allGlobalTables.GlobalTables))
	for _, gt := range allGlobalTables.GlobalTables {
		regions := []string{}
		for _, g := range gt.ReplicationGroup {
			regions = append(regions, *g.RegionName)
		}
		fmt.Printf("global tables: %v    %v --\n", *gt.GlobalTableName, regions)
	}

	_, err = cli1.CreateGlobalTable(ctx, &dynamodb.CreateGlobalTableInput{
		GlobalTableName: aws.String(entity.TableUserClusterInfo),
		ReplicationGroup: []types.Replica{
			{
				RegionName: aws.String(region1),
			},
			{
				RegionName: aws.String(region2),
			},
		},
	})
	require.NoError(t, err)
	//fmt.Printf("create global table output: %#v \n\n", *createGlobalTableOutput.GlobalTableDescription)

	info1 := entity.UserClusterInfo{
		ProjectId:       123,
		ClusterId:       456,
		IsDedicate:      true,
		EndpointService: "endpoint service name",
	}

	err = PutUserClusterInfo(ctx, cli1, &info1)
	require.NoError(t, err)

	getInfo1, err := GetUserClusterInfo(ctx, cli1, info1.ProjectId, info1.ClusterId)
	require.NoError(t, err)
	require.Equal(t, info1, *getInfo1)

	start := time.Now()
	for {
		getInfo1, err = GetUserClusterInfo(ctx, cli2, info1.ProjectId, info1.ClusterId)
		require.NoError(t, err)
		if getInfo1.EndpointService == info1.EndpointService {
			break
		}
		time.Sleep(time.Millisecond * 500)
	}
	fmt.Printf("read in region2 sync cost %v ---------\n\n", time.Since(start))
	require.Equal(t, info1, *getInfo1)

	info2 := entity.UserClusterInfo{
		ProjectId:       123,
		ClusterId:       456789,
		IsDedicate:      true,
		EndpointService: "endpoint service name",
	}
	err = PutUserClusterInfo(ctx, cli2, &info2)
	require.NoError(t, err)

	getInfo2, err := GetUserClusterInfo(ctx, cli2, info2.ProjectId, info2.ClusterId)
	require.NoError(t, err)
	require.Equal(t, info2, *getInfo2)

	start = time.Now()
	for {
		getInfo2, err = GetUserClusterInfo(ctx, cli1, info2.ProjectId, info2.ClusterId)
		require.NoError(t, err)
		if getInfo2.EndpointService == info2.EndpointService {
			break
		}
		time.Sleep(time.Millisecond * 500)
	}
	fmt.Printf("read in region1 sync cost %v ---------\n\n", time.Since(start))
	require.Equal(t, info2, *getInfo2)
}

// go test -run=TestLongTime -timeout 99999s
func TestLongTime(t *testing.T) {
	ctx := context.Background()
	client, err := newDynamoDBClientForTest(ctx)
	require.NoError(t, err)

	tilog.Info(ctx, "sleep now")
	time.Sleep(time.Second * 3700)
	tilog.Info(ctx, "finish sleep")

	listTablesOutput, err := client.ListTables(ctx, &dynamodb.ListTablesInput{})
	require.NoError(t, err)
	fmt.Printf("tables: %#v ---\n", listTablesOutput.TableNames)

	err = CreateUserClusterInfoTableIfNotExist(ctx, client)
	require.NoError(t, err)

	clusterId := uint64(789019)
	info := entity.UserClusterInfo{
		ProjectId:       123,
		ClusterId:       clusterId,
		IsDedicate:      true,
		EndpointService: "endpoint service name",
		Host:            "?",
		Port:            4000,
	}
	err = PutUserClusterInfo(ctx, client, &info)
	require.NoError(t, err)

	result, err := QueryUserClusterInfo(ctx, client, 123, clusterId)
	require.NoError(t, err)
	fmt.Printf("query: %#v  ---\n\n", result)
	require.Equal(t, 1, len(result))
	require.Equal(t, info, result[0])

	getInfo, err := GetUserClusterInfo(ctx, client, 123, clusterId)
	require.NoError(t, err)
	fmt.Printf("get info: %#v  ---\n\n", getInfo)
	require.Equal(t, info, *getInfo)

	info.PrivateDNS = "new private dns name"
	err = PutUserClusterInfo(ctx, client, &info)
	require.NoError(t, err)
	getInfo, err = GetUserClusterInfo(ctx, client, 123, clusterId)
	require.NoError(t, err)
	fmt.Printf("get info: %#v  ---\n\n", getInfo)
	require.Equal(t, info, *getInfo)

	clusterInfos, err := ScanUserClusterInfo(ctx, client)
	require.NoError(t, err)
	for _, info := range clusterInfos {
		fmt.Printf("scan info: %v \n", info)
	}

	err = ddb.DeleteTable(ctx, client, entity.TableUserClusterInfo)
	require.NoError(t, err)
}

func newDynamoDBClientForTest(ctx context.Context) (*dynamodb.Client, error) {
	region := "us-west-2"
	return newDynamoDBClientForTestByRegion(ctx, region)
}