// Package service add by blacknc <blacknc@163.com> at 2023/04/04 14:46:35
package service

import (
	"context"
	"dataservices/commons/config"
	"dataservices/commons/grpc"
	"dataservices/ddb"
	dto "dataservices/internal/dto/dataapi"
	"dataservices/internal/entity"
	"dataservices/internal/repository"
	"dataservices/internal/repository_ddb"
	"time"

	"github.com/aws/aws-sdk-go-v2/service/dynamodb"
	"github.com/tidbcloud/core-svc/idl/pbgen/proto/core"
	"github.com/tidbcloud/tiinf/errs"
	"github.com/tidbcloud/tiinf/monitor/tilog"
	"go.buf.build/protocolbuffers/go/linuxgit/tidbcloud/base"
	"go.uber.org/zap"
)

func CreateDynamoDBGlobalTable(ctx context.Context, region string, replicaRegions []string, tolerant bool) error {
	var (
		err    error
		client *dynamodb.Client
	)
	if len(replicaRegions) == 0 {
		return errs.NewBadRequest("At least two regions are required for global table", nil)
	}

	client, err = ddb.NewDynamodbClient(ctx, region)
	if err != nil {
		return errs.NewInternalDB("Failed to connect to the ddb", err)
	}

	if err := handleDDBError(ctx, tolerant, repository_ddb.CreateCustomDomainTable(ctx, client, replicaRegions)); err != nil {
		return err
	}
	if err := handleDDBError(ctx, tolerant, repository_ddb.CreateDataAppTable(ctx, client, replicaRegions)); err != nil {
		return err
	}
	if err := handleDDBError(ctx, tolerant, repository_ddb.CreateEndpointTable(ctx, client, replicaRegions)); err != nil {
		return err
	}
	if err := handleDDBError(ctx, tolerant, repository_ddb.CreateEndpointPublishTable(ctx, client, replicaRegions)); err != nil {
		return err
	}
	if err := handleDDBError(ctx, tolerant, repository_ddb.CreateAppClusterTable(ctx, client, replicaRegions)); err != nil {
		return err
	}
	if err := handleDDBError(ctx, tolerant, repository_ddb.CreateClusterTable(ctx, client, replicaRegions)); err != nil {
		return err
	}
	return nil
}

func DeleteDynamoDBGlobalTable(ctx context.Context, region string, replicaRegions []string, tolerant bool) error {
	var (
		err    error
		client *dynamodb.Client
	)
	if len(replicaRegions) == 0 {
		return errs.NewBadRequest("At least two regions are required for global table", nil)
	}

	client, err = ddb.NewDynamodbClient(ctx, region)
	if err != nil {
		return errs.NewInternalDB("Failed to connect to the ddb", err)
	}

	if err := handleDDBError(ctx, tolerant, repository_ddb.DeleteDataAppTable(ctx, client, replicaRegions)); err != nil {
		return err
	}
	if err := handleDDBError(ctx, tolerant, repository_ddb.DeleteEndpointTable(ctx, client, replicaRegions)); err != nil {
		return err
	}
	if err := handleDDBError(ctx, tolerant, repository_ddb.DeleteEndpointPublishTable(ctx, client, replicaRegions)); err != nil {
		return err
	}
	if err := handleDDBError(ctx, tolerant, repository_ddb.DeleteAppClusterTable(ctx, client, replicaRegions)); err != nil {
		return err
	}
	if err := handleDDBError(ctx, tolerant, repository_ddb.DeleteClusterTable(ctx, client, replicaRegions)); err != nil {
		return err
	}
	return nil
}

func handleDDBError(ctx context.Context, tolerant bool, err error) error {
	if err == nil {
		return nil
	}
	tilog.Error(ctx, "ddb error", zap.Error(err))
	if !tolerant {
		return err
	}
	return nil
}

func SyncDataToDDB(ctx context.Context, in dto.SyncDataToDDBReq) (*dto.SyncDataToDDBResp, error) {

	cli, err := ddb.NewDynamodbClient(ctx, config.Global().AwsConf.Region)
	if err != nil {
		return nil, err
	}
	result := make([]dto.SyncTableRes, 0, len(in.Tables))
	for _, table := range in.Tables {
		var (
			succWritten int
			err         error
		)
		switch table.TableName {
		case entity.TableDataApps:
			succWritten, err = syncDataApps(ctx, cli, table.IDs)
		case entity.TableEndpoints:
			succWritten, err = syncEndpoints(ctx, cli, table.IDs)
		case entity.TableEndpointPublish:
			succWritten, err = syncEndpointPublish(ctx, cli, table.IDs)
		case entity.TableDataAppClusters:
			succWritten, err = syncAppCluster(ctx, cli, table.IDs)
		case entity.TableUserClusterInfo:
			succWritten, err = syncCluster(ctx, cli, table.IDs)
		default:
			tilog.Error(ctx, "unsupported table", zap.String("table", table.TableName))
		}
		if err != nil {
			tilog.Error(ctx, "failed to sync table data", zap.String("table", table.TableName), zap.Error(err))
		} else {
			tilog.Info(ctx, "succeed to sync table data", zap.String("table", table.TableName), zap.Int("succeed_count", succWritten))
		}
		result = append(result, dto.SyncTableRes{
			TableName: table.TableName,
			Count:     succWritten,
			Error:     err,
		})
	}
	return &dto.SyncDataToDDBResp{Tables: result}, nil
}

func syncDataApps(ctx context.Context, cli *dynamodb.Client, ids []uint64) (int, error) {

	var (
		offset      = 0
		limit       = 50
		succWritten int
		err         error
	)

	for {
		var infos []*entity.DataApp
		var written int
		infos, err = repository.PageDataApps(ctx, ids, offset, limit)
		if err != nil || len(infos) == 0 {
			break
		}
		written, err = repository_ddb.BatchPutDataApp(ctx, cli, infos)
		succWritten += written
		tilog.Info(ctx, "sync dataapps progress",
			zap.Int("offset", offset),
			zap.Int("limit", limit),
			zap.Int("written", succWritten),
		)
		if err != nil {
			break
		}
		offset += limit
	}
	return succWritten, err
}

func syncEndpoints(ctx context.Context, cli *dynamodb.Client, ids []uint64) (int, error) {

	var (
		offset      = 0
		limit       = 50
		succWritten int
		err         error
	)

	for {
		var infos []*entity.Endpoint
		var written int
		infos, err = repository.PageEndpoints(ctx, ids, offset, limit)
		if err != nil || len(infos) == 0 {
			break
		}
		written, err = repository_ddb.BatchPutEndpoint(ctx, cli, infos)
		succWritten += written
		tilog.Info(ctx, "sync endpoints progress",
			zap.Int("offset", offset),
			zap.Int("limit", limit),
			zap.Int("written", succWritten),
		)
		if err != nil {
			break
		}
		offset += limit
	}
	return succWritten, err
}

func syncEndpointPublish(ctx context.Context, cli *dynamodb.Client, ids []uint64) (int, error) {

	var (
		offset      = 0
		limit       = 50
		succWritten int
		err         error
	)

	for {
		var infos []*entity.EndpointPublish
		var written int
		infos, err = repository.PageEndpointPublish(ctx, ids, offset, limit)
		if err != nil || len(infos) == 0 {
			break
		}
		written, err = repository_ddb.BatchPutPublishedEndpoint(ctx, cli, infos)
		succWritten += written
		tilog.Info(ctx, "sync published endpoint progress",
			zap.Int("offset", offset),
			zap.Int("limit", limit),
			zap.Int("written", succWritten),
		)
		if err != nil {
			break
		}
		offset += limit
	}
	return succWritten, err
}

func syncAppCluster(ctx context.Context, cli *dynamodb.Client, ids []uint64) (int, error) {

	var (
		offset      = 0
		limit       = 50
		succWritten int
		err         error
	)

	for {
		var infos []*entity.AppCluster
		var written int
		infos, err = repository.PageAppCluster(ctx, ids, offset, limit)
		if err != nil || len(infos) == 0 {
			break
		}
		written, err = repository_ddb.BatchPutAppCluster(ctx, cli, infos)
		succWritten += written
		tilog.Info(ctx, "sync app cluster progress",
			zap.Int("offset", offset),
			zap.Int("limit", limit),
			zap.Int("written", succWritten),
		)
		if err != nil {
			break
		}
		offset += limit
	}
	return succWritten, err
}

func syncCluster(ctx context.Context, cli *dynamodb.Client, ids []uint64) (int, error) {

	var (
		err   error
		step  = 50
		start = 0
		end   = start + step
	)

	if len(ids) == 0 {
		ids, err = getAllServerlessClusterIDs(ctx)
		if err != nil {
			return 0, err
		}
	}

	// list cluster info from cm
	clusterMap := make(map[uint64]*core.Cluster, len(ids))
	for start < len(ids) {
		if end > len(ids) {
			end = len(ids)
		}
		clusterList, berr := grpc.ListClusters(ctx, &core.ListClustersReq{
			Filter: &core.ClusterFilter{
				ClusterIds: ids[start:end],
			},
		})
		if berr != nil {
			return 0, err
		}
		for clusterID, clusterInfo := range clusterList.Clusters {
			clusterMap[clusterID] = clusterInfo
		}
		start, end = end, start+step
	}

	userClusterList := make([]*entity.UserClusterInfo, 0, len(ids))
	for _, clusterID := range ids {

		clusterInfo, ok := clusterMap[clusterID]
		if !ok || clusterInfo == nil {
			tilog.Error(ctx, "cluster not exists", zap.Uint64("cluster_id", clusterID))
			continue
		}
		isDedicated := isDedicatedCluster(clusterInfo)
		if isDedicated {
			continue
		}

		var host string
		var port uint64

		// privatelink has been used on prod env
		if !config.IsProd() {
			for _, connection := range clusterInfo.GetConnections() {
				if connection.Type == core.ClusterConnectionType_CLUSTER_CONNECTION_TYPE_PUBLIC {
					host = connection.Host
					port = uint64(connection.Port)
				}
			}
			if host == "" || port == 0 {
				tilog.Error(ctx, "not found user public endpoint connection for dev/staging env")
				continue
			}
		} else {
			port = uint64(4000)
		}

		userClusterList = append(userClusterList, &entity.UserClusterInfo{
			OrgId:     clusterInfo.TenantId,
			ProjectId: clusterInfo.ProjectId,
			ClusterId: clusterInfo.Id,
			Cloud:     clusterInfo.CloudProviderName,
			Region:    clusterInfo.RegionName,
			Host:      host,
			Port:      port,
		})
	}

	return repository_ddb.BatchPutUserClusterInfo(ctx, cli, userClusterList)
}

func getAllClusterIDs(ctx context.Context) ([]uint64, error) {

	var (
		offset        = 0
		limit         = 50
		clusterIDs    []uint64
		projectExists = make(map[uint64]bool)
	)

	for {
		infos, err := repository.PageDataApps(ctx, nil, offset, limit)
		if err != nil {
			return nil, err
		} else if len(infos) == 0 {
			break
		}
		for _, info := range infos {
			orgID := info.OrgId
			projectID := info.ProjectId
			if projectExists[projectID] {
				continue
			}
			projectExists[projectID] = true
			projectClusterIDs, err := getProjectAllClusterIDs(ctx, orgID, projectID)
			if err != nil {
				return nil, err
			}
			clusterIDs = append(clusterIDs, projectClusterIDs...)
		}
		offset += limit
	}
	return clusterIDs, nil
}

func getAllServerlessClusterIDs(ctx context.Context) ([]uint64, error) {
	var (
		clusterIDs      []uint64
		isDevTier       = true
		listClustersReq = &core.ListClustersReq{
			Pager: &base.Pager{
				Page:     1,
				PageSize: 20,
			},
			Filter: &core.ClusterFilter{
				IsDevTier: &isDevTier,
			},
			Sorters: []*base.Sorter{{OrderBy: "id", Sort: base.Sort_SORT_ASC}},
		}
	)

	start := time.Now()
	tilog.Info(ctx, "load all serverless clusters begin")
	for {
		res, err := grpc.ListClusters(ctx, listClustersReq)
		if err != nil {
			return nil, err
		}
		if len(res.OrderedClusterIds) == 0 {
			break
		}
		clusterIDs = append(clusterIDs, res.OrderedClusterIds...)
		tilog.Info(ctx, "load all serverless clusters progress",
			zap.Int64("page", listClustersReq.Pager.Page),
			zap.Int("count", len(clusterIDs)),
			zap.String("cost", time.Since(start).String()),
		)
		listClustersReq.Pager.Page++
		// Control request frequency
		time.Sleep(time.Millisecond * 200)
	}
	tilog.Info(ctx, "load all serverless clusters end",
		zap.Int("count", len(clusterIDs)),
		zap.String("cost", time.Since(start).String()),
	)
	return clusterIDs, nil
}