package vercel

import (
	"context"
	"encoding/json"
	"fmt"
	"net/url"
	"strconv"
	"time"

	"github.com/gin-gonic/gin"
	"github.com/pingcap/errors"
	accountpb "github.com/tidbcloud/account/idl/pbgen/proto/account"
	dataapipb "github.com/tidbcloud/dataservices/idl/pbgen/proto/dataapi"
	"github.com/tidbcloud/ecosystem-service/config"
	"github.com/tidbcloud/ecosystem-service/internal/service/grpc/account"
	ecoAcl "github.com/tidbcloud/ecosystem-service/internal/service/grpc/acl"
	"github.com/tidbcloud/ecosystem-service/internal/service/grpc/dataapi"
	"github.com/tidbcloud/ecosystem-service/internal/service/grpc/mgmt"
	vercelProxy "github.com/tidbcloud/ecosystem-service/internal/service/proxy/vercel"
	vercelDB "github.com/tidbcloud/ecosystem-service/internal/storage/db/vercel"
	"github.com/tidbcloud/ecosystem-service/internal/utils"
	"github.com/tidbcloud/pkgv2/log"
	"github.com/tidbcloud/pkgv2/password"
	globalserverlessv1 "github.com/tidbcloud/tidb-management-service/api/spec/global/serverless/v1"
	mgmtapi "github.com/tidbcloud/tidb-management-service/api/spec/tidb_mgmt_service/v1"
	"go.uber.org/zap"
)

const (
	// core-svc requires username length less than 32, password length less than 64
	vercelUsernamePrefix = "vu-"
)

type ClusterConnLogic struct {
	clusterConnectionDAO vercelDB.ClusterConnectionDAO
	accountDAO           vercelDB.TiDBVercelAccountDAO
	configurationDAO     vercelDB.VercelIntegrationConfigurationDAO
	vercelAPIClient      vercelProxy.APIClient
	accountProviderSvc   account.AccountProviderSvc
	mgmtSvc              mgmt.MgmtSvc
	aclSvc               ecoAcl.AclSvc
	conf                 *config.ConfigWrapper
	dataAppConnectionDAO vercelDB.DataAppConnectionDAO
	dataApiSvc           dataapi.DataApiSvc
	composeEntityDAO     vercelDB.ComposeEntityDAO
}

func NewClusterConnLogic(connectionDAO vercelDB.ClusterConnectionDAO,
	accountDAO vercelDB.TiDBVercelAccountDAO,
	configurationDAO vercelDB.VercelIntegrationConfigurationDAO,
	vercelAPIClient vercelProxy.APIClient,
	accountService account.AccountProviderSvc,
	mgmtSvc mgmt.MgmtSvc,
	aclSvc ecoAcl.AclSvc,
	conf *config.ConfigWrapper,
	dataAppConnectionDAO vercelDB.DataAppConnectionDAO,
	dataApiSvc dataapi.DataApiSvc,
	composeEntityDAO vercelDB.ComposeEntityDAO) *ClusterConnLogic {
	return &ClusterConnLogic{
		clusterConnectionDAO: connectionDAO,
		accountDAO:           accountDAO,
		configurationDAO:     configurationDAO,
		vercelAPIClient:      vercelAPIClient,
		accountProviderSvc:   accountService,
		mgmtSvc:              mgmtSvc,
		aclSvc:               aclSvc,
		conf:                 conf,
		dataAppConnectionDAO: dataAppConnectionDAO,
		dataApiSvc:           dataApiSvc,
		composeEntityDAO:     composeEntityDAO,
	}
}

func (l *ClusterConnLogic) getAccessibleProjects(ctx context.Context, user *accountpb.GetAuthUserResponse) ([]*accountpb.Project, error) {
	orgs, err := l.accountProviderSvc.ListOrgsByUser(ctx, user.User.Id)
	if err != nil {
		return nil, errors.Trace(err)
	}

	orgIDs := make([]uint64, len(orgs))
	for i, org := range orgs {
		orgIDs[i] = org.Id
	}

	orgProjects, err := l.accountProviderSvc.ListProjectsByOrgsInUser(ctx, user.User.Id, orgIDs)
	if err != nil {
		return nil, errors.Trace(err)
	}

	var projects []*accountpb.Project
	for _, project := range orgProjects {
		projects = append(projects, project.Projects...)
	}
	return projects, nil
}

// fill response with Vercel project name
func (l *ClusterConnLogic) fillVercelDetailInformation(ctx context.Context, configurationID string, connectionVOs []TiDBVercelConnectionVO, logger *zap.Logger) error {
	if len(connectionVOs) == 0 {
		return nil
	}

	configuration, err := l.configurationDAO.GetIntegrationConfigurationByID(ctx, configurationID)
	if err != nil {
		return err
	}

	// check if the configuration is created for a team
	// if yes, when querying Vercel API, teamID is required to set.
	var teamID string
	if configuration.TeamID.Valid {
		teamID = configuration.TeamID.String
	} else {
		teamID = ""
	}

	projects, err := l.RetrieveAllProjects(ctx, vercelProxy.RetrieveProjectQueryParameters{
		TeamID: teamID,
	}, configuration.ApiAccessToken, logger)
	if err != nil {
		return err
	}
	projectMap := make(map[string]*vercelProxy.Project)
	for i, project := range projects {
		projectMap[project.ID] = &projects[i]
	}
	for i := range connectionVOs {
		if project, ok := projectMap[connectionVOs[i].VercelProjectID]; ok {
			connectionVOs[i].VercelProjectName = project.Name
		}
	}

	return nil
}

// fill response with TiDB org name, project name and cluster name
func (l *ClusterConnLogic) fillTiDBDetailInformation(ctx context.Context, connectionVOs []TiDBVercelConnectionVO, logger *zap.Logger) error {
	if len(connectionVOs) == 0 {
		return nil
	}

	// deduplicate cluster id
	clusterIDMap := map[uint64]bool{}
	projectIDMap := map[uint64]bool{}
	orgIDMap := map[uint64]bool{}
	dataAppIDMap := map[uint64]bool{}

	var clusterIDs []uint64
	var dataAppIDs []uint64
	var projectIDs []uint64
	var orgIDs []uint64
	for i := range connectionVOs {
		projectID, _ := strconv.ParseUint(connectionVOs[i].TiDBProjectID, 10, 64)
		orgID, _ := strconv.ParseUint(connectionVOs[i].TiDBOrgID, 10, 64)
		targetID, _ := strconv.ParseUint(connectionVOs[i].TargetID, 10, 64)

		if connectionVOs[i].TargetType == TargetTypeCluster {
			if _, ok := clusterIDMap[targetID]; !ok {
				clusterIDs = append(clusterIDs, targetID)
				clusterIDMap[targetID] = true
			}
		} else if connectionVOs[i].TargetType == TargetTypeDataApp {
			if _, ok := dataAppIDMap[targetID]; !ok {
				dataAppIDs = append(dataAppIDs, targetID)
				dataAppIDMap[targetID] = true
			}
		}

		if _, ok := projectIDMap[projectID]; !ok {
			projectIDs = append(projectIDs, projectID)
			projectIDMap[projectID] = true
		}
		if _, ok := orgIDMap[orgID]; !ok {
			orgIDs = append(orgIDs, orgID)
			orgIDMap[orgID] = true
		}
	}

	clustersMap := make(map[uint64]*mgmtapi.Cluster, len(clusterIDs))
	var err error
	if len(clusterIDs) > 0 {
		clustersMap, err = l.mgmtSvc.ListClusters(ctx, clusterIDs, mgmtapi.ClusterFilter_CLUSTER_VIEW_LOCAL)
		if err != nil {
			return errors.Trace(err)
		}
	}

	projects, err := l.accountProviderSvc.ListProjects(ctx, projectIDs)
	if err != nil {
		return errors.Trace(err)
	}
	projectsMap := map[uint64]*accountpb.Project{}
	for i := range projects {
		projectsMap[projects[i].Id] = projects[i]
	}

	orgs, err := l.accountProviderSvc.ListOrgs(ctx, orgIDs)
	if err != nil {
		return errors.Trace(err)
	}
	orgsMap := map[uint64]*accountpb.Org{}
	for i := range orgs {
		orgsMap[orgs[i].Id] = orgs[i]
	}

	dataAppMap := make(map[uint64]*dataapipb.AppInfo, len(dataAppIDs))
	if len(dataAppIDs) > 0 {
		for _, id := range dataAppIDs {
			app, err := l.dataApiSvc.ShowDataApp(ctx, id)
			if err != nil {
				logger.Error("Failed to ShowDataApp", zap.Error(err), zap.Uint64("id", id))
			}
			if app != nil {
				dataAppMap[id] = app
			}
		}
	}

	for i := range connectionVOs {
		targetID, _ := strconv.ParseUint(connectionVOs[i].TargetID, 10, 64)
		projectID, _ := strconv.ParseUint(connectionVOs[i].TiDBProjectID, 10, 64)
		orgID, _ := strconv.ParseUint(connectionVOs[i].TiDBOrgID, 10, 64)
		if connectionVOs[i].TargetType == TargetTypeCluster {
			if cluster, ok := clustersMap[targetID]; ok {
				connectionVOs[i].TargetName = cluster.ClusterName
			}
		} else if connectionVOs[i].TargetType == TargetTypeDataApp {
			if app, ok := dataAppMap[targetID]; ok {
				connectionVOs[i].TargetName = app.AppName
			}
		}

		if project, ok := projectsMap[projectID]; ok {
			connectionVOs[i].TiDBProjectName = project.Name
		}
		if org, ok := orgsMap[orgID]; ok {
			connectionVOs[i].TiDBOrgName = org.Name
		}
	}

	return nil
}

type PublicEndpointDisableError struct{}

func (*PublicEndpointDisableError) Error() string {
	return "Cannot connect to Serverless clusters with Public Endpoint disabled"
}

type PublicEndpointNotAllowAllError struct{}

func (*PublicEndpointNotAllowAllError) Error() string {
	return "Cannot connect to Serverless clusters with Public Endpoint not allow all public IPv4 addresses"
}

func (l *ClusterConnLogic) prepareEnvironments(ctx context.Context, cluster *mgmtapi.Cluster, account *vercelDB.TiDBVercelAccount, database string, logger *zap.Logger) (*ClusterEnvironments, error) {
	envs := new(ClusterEnvironments)

	pe := cluster.GetServerlessCluster().GetCluster().GetEndpoints().GetPublic()
	if pe.Disabled {
		return nil, &PublicEndpointDisableError{}
	}

	allowAll := false
	for _, network := range cluster.GetServerlessCluster().GetCluster().GetEndpoints().GetPublic().GetAuthorizedNetwork() {
		if network.StartIpAddress == "0.0.0.0" && network.EndIpAddress == "255.255.255.255" {
			allowAll = true
		}
	}
	if !allowAll {
		return nil, &PublicEndpointNotAllowAllError{}
	}

	prefix := cluster.GetServerlessCluster().GetCluster().GetUserPrefix()
	envs.DBHost = pe.Host
	envs.DBPort = int(pe.Port)

	var username string
	password := password.Decorate(randomString(32))
	for {
		username = prefix + "." + vercelUsernamePrefix + randomString(8)
		account, err := l.accountDAO.GetVercelAccount(ctx, account.TiDBClusterID, account.VercelProjectID, username)
		if err != nil {
			return nil, err
		}
		if account != nil {
			continue
		}
		break
	}
	envs.DBUser = username
	envs.DBPassword = password
	envs.DBDatabase = database
	account.TiDBClusterUsername = username

	return envs, nil
}

func (l *ClusterConnLogic) AddClusterConnection(ctx context.Context, vercelProjectID string, vercelConfigID string, frameworkType string, environments *ClusterEnvironments, connection *vercelDB.ClusterConnection) error {
	configuration, err := l.configurationDAO.GetIntegrationConfigurationByID(ctx, vercelConfigID)
	if err != nil {
		return err
	}

	var targetEnv []string
	if connection.BranchEnabled {
		targetEnv = []string{"production"}
	} else {
		targetEnv = []string{"production", "preview", "development"}
	}

	var descriptions []vercelProxy.EnvironmentDescription
	if frameworkType == "General" {
		descriptions = []vercelProxy.EnvironmentDescription{
			{
				Key:               "TIDB_HOST",
				TargetEnvironment: targetEnv,
				Type:              "encrypted",
				Value:             environments.DBHost,
			},
			{
				Key:               "TIDB_PORT",
				TargetEnvironment: targetEnv,
				Type:              "encrypted",
				Value:             strconv.Itoa(environments.DBPort),
			},
			{
				Key:               "TIDB_USER",
				TargetEnvironment: targetEnv,
				Type:              "encrypted",
				Value:             environments.DBUser,
			},
			{
				Key:               "TIDB_PASSWORD",
				TargetEnvironment: targetEnv,
				Type:              "encrypted",
				Value:             environments.DBPassword,
			},
			{
				Key:               "TIDB_DATABASE",
				TargetEnvironment: targetEnv,
				Type:              "encrypted",
				Value:             environments.DBDatabase,
			},
		}
		if environments.DBSslCa != nil {
			descriptions = append(descriptions, vercelProxy.EnvironmentDescription{
				Key:               "TIDB_SSL_CA",
				TargetEnvironment: targetEnv,
				Type:              "encrypted",
				Value:             string(environments.DBSslCa), // PEM file is text file in Base64 ASCII encoding
			})
		}
	} else if frameworkType == "Prisma" {
		descriptions = []vercelProxy.EnvironmentDescription{
			{
				Key:               "DATABASE_URL",
				TargetEnvironment: targetEnv,
				Type:              "encrypted",
				// TODO: when there is no more dev-tier cluster, we should set sslaccept=strict to serverless-tier.
				Value: fmt.Sprintf("mysql://%s:%s@%s:%d/%s?sslaccept=strict", environments.DBUser, url.QueryEscape(environments.DBPassword), environments.DBHost, environments.DBPort, url.QueryEscape(environments.DBDatabase)),
			},
		}
	} else if frameworkType == "TiDB Cloud Serverless Driver" {
		descriptions = []vercelProxy.EnvironmentDescription{
			{
				Key:               "DATABASE_URL",
				TargetEnvironment: targetEnv,
				Type:              "encrypted",
				Value:             fmt.Sprintf("mysql://%s:%s@%s/%s", environments.DBUser, url.QueryEscape(environments.DBPassword), environments.DBHost, url.QueryEscape(environments.DBDatabase)),
			},
		}
	} else {
		return errors.Errorf("unsupported framework type: %s", frameworkType)
	}

	result, err := l.vercelAPIClient.CreateEnvironmentVariables(ctx, vercelProjectID, utils.GetTeamIDString(configuration.TeamID), descriptions, configuration.ApiAccessToken)
	if err != nil {
		return err
	}

	identifier, err := transferToIdentifierEntity(result)
	if err != nil {
		return err
	}

	err = l.composeEntityDAO.AddClusterConnection(ctx, connection, identifier)
	if err != nil {
		return err
	}

	return nil
}

func (l *ClusterConnLogic) ConfigureTiDBCluster(ctx context.Context, user *accountpb.User, cluster *mgmtapi.Cluster, req AddConnectionRequest, logger *zap.Logger) (*ClusterEnvironments, error) {
	if log.GetLevel().CapitalString() == "DEBUG" {
		s, _ := json.MarshalIndent(cluster, "", "\t")
		logger.Debug(fmt.Sprintf("mgmtSvc.GetCluster %s", s))
	}

	account := vercelDB.TiDBVercelAccount{
		TiDBOrgID:       cluster.OrganizationId,
		TiDBProjectID:   cluster.ProjectId,
		TiDBClusterID:   cluster.ClusterId,
		VercelProjectID: req.VercelProjectID,
	}
	envs, err := l.prepareEnvironments(ctx, cluster, &account, req.Database, logger)
	logger.Debug(fmt.Sprintf("ClusterEnvironments %#v", envs))
	if err != nil {
		return nil, errors.Annotate(err, "failed to prepare environments")
	}
	if err := l.accountDAO.AddVercelAccount(ctx, account); err != nil {
		return nil, errors.Annotate(err, "failed to add vercel account to database")
	}
	operatorUser, jwt, err := l.aclSvc.GetJwtByUser(ctx, cluster.ClusterId, user.Id, user.Email)
	if err != nil {
		return nil, errors.Annotate(err, "failed to get operator from account")
	}
	if err := l.mgmtSvc.CreateSqlUser(ctx, cluster.ClusterId, operatorUser, jwt, envs.DBUser, envs.DBPassword); err != nil {
		return nil, errors.Annotate(err, "failed to create cluster account through core-svc")
	}
	logger.Info(fmt.Sprintf("Create account, username: %s", envs.DBUser))

	return envs, nil
}

func (l *ClusterConnLogic) CheckConnIfExists(ctx context.Context, clusterID uint64, vercelProjectID string) (bool, error) {
	c, err := l.clusterConnectionDAO.GetVercelConnection(ctx, clusterID, vercelProjectID)
	if err != nil {
		return false, errors.Annotate(err, "GetVercelConnection failed")
	}
	if c != nil {
		return true, nil
	}
	return false, nil
}

func (l *ClusterConnLogic) CheckVercelProjectIfConfigured(ctx context.Context, vercelProjectID string) (bool, error) {
	c, err := l.clusterConnectionDAO.GetVercelConnectionByVercelProjectID(ctx, vercelProjectID)
	if err != nil {
		return false, errors.Annotate(err, "GetVercelConnection failed")
	}
	if c != nil {
		return true, nil
	}
	return false, nil
}

func (l *ClusterConnLogic) DeleteVercelEnvs(ctx context.Context, clusterID uint64, req VercelProject, logger *zap.Logger) error {
	connection, err := l.clusterConnectionDAO.GetVercelConnection(ctx, clusterID, req.VercelProjectID)
	if err != nil {
		return err
	}

	if connection == nil {
		return errors.NotFoundf("connection")
	}

	if err := l.DeleteEnvsByEnvIdentifier(ctx, connection.VercelProjectID, utils.GetTeamIDString(connection.VercelIntegrationConfiguration.TeamID), connection.EnvironmentIdentifier, connection.VercelIntegrationConfiguration.ApiAccessToken); err != nil {
		return err
	}

	if connection.BranchEnvironmentIdentifiers != nil {
		for _, identifier := range connection.BranchEnvironmentIdentifiers {
			if err := l.DeleteBranchEnvsByEnvIdentifier(ctx, connection.VercelProjectID, utils.GetTeamIDString(connection.VercelIntegrationConfiguration.TeamID), identifier, connection.VercelIntegrationConfiguration.ApiAccessToken); err != nil {
				return err
			}
		}
	}

	logger.Info(fmt.Sprintf("DeleteVercelEnvs for %s", connection.VercelProjectID))
	return nil
}

func (l *ClusterConnLogic) getConnectionVOsByOrgID(ctx context.Context, clusterConnections []vercelDB.ClusterConnection, appConnections []vercelDB.DataAppConnection, logger *zap.Logger) ([]TiDBTargetVercelConnectionVO, error) {
	if len(clusterConnections) == 0 && len(appConnections) == 0 {
		return []TiDBTargetVercelConnectionVO{}, nil
	}
	// Connections may be configured by different Vercel integrations
	configurationMap := map[string]bool{}
	var confIDs []string

	connectionsResponse := make([]TiDBVercelConnectionVO, 0, len(clusterConnections)+len(appConnections))
	for _, connection := range clusterConnections {
		connectionsResponse = append(connectionsResponse, TiDBVercelConnectionVO{
			TiDBProject: TiDBProject{
				TiDBOrgID:     strconv.FormatUint(connection.TiDBOrgID, 10),
				TiDBProjectID: "", // Empty for org-level query
			},
			VercelProject: VercelProject{
				VercelProjectID: connection.VercelProjectID,
			},
			ConnectionTarget: ConnectionTarget{
				TargetType:    TargetTypeCluster,
				TargetID:      strconv.FormatUint(connection.TiDBClusterID, 10),
				BranchEnabled: connection.BranchEnabled,
			},
		})
		id := connection.VercelIntegrationConfigurationID
		if _, ok := configurationMap[id]; !ok {
			confIDs = append(confIDs, id)
			configurationMap[id] = true
		}
	}

	for _, connection := range appConnections {
		connectionsResponse = append(connectionsResponse, TiDBVercelConnectionVO{
			TiDBProject: TiDBProject{
				TiDBOrgID:     strconv.FormatUint(connection.TiDBOrgID, 10),
				TiDBProjectID: "", // Empty for org-level query
			},
			VercelProject: VercelProject{
				VercelProjectID: connection.VercelProjectID,
			},
			ConnectionTarget: ConnectionTarget{
				TargetType:    TargetTypeDataApp,
				TargetID:      strconv.FormatUint(connection.DataAppID, 10),
				BranchEnabled: false,
			},
		})
		id := connection.VercelIntegrationConfigurationID
		if _, ok := configurationMap[id]; !ok {
			confIDs = append(confIDs, id)
			configurationMap[id] = true
		}
	}

	for _, id := range confIDs {
		if err := l.fillVercelDetailInformation(ctx, id, connectionsResponse, logger); err != nil {
			return nil, err
		}
	}

	if err := l.fillTiDBDetailInformation(ctx, connectionsResponse, logger); err != nil {
		return nil, err
	}

	targetMap := map[string]TiDBTargetVercelConnectionVO{}
	for _, connection := range connectionsResponse {
		targetKey := fmt.Sprintf("%s-%s", connection.TargetType, connection.TargetID)
		if targetConn, ok := targetMap[targetKey]; !ok {
			targetMap[targetKey] = TiDBTargetVercelConnectionVO{
				ConnectionTarget: ConnectionTarget{
					TargetType:    connection.TargetType,
					TargetName:    connection.TargetName,
					TargetID:      connection.TargetID,
					BranchEnabled: connection.BranchEnabled,
				},
				Projects: []VercelProject{
					{
						VercelProjectID:   connection.VercelProjectID,
						VercelProjectName: connection.VercelProjectName,
					},
				},
				TiDBProject: TiDBProject{
					TiDBOrgID:     connection.TiDBOrgID,
					TiDBProjectID: "", // Empty for org-level query
				},
				TiDBOrgName: connection.TiDBOrgName,
			}
		} else {
			targetConn.Projects = append(targetConn.Projects, VercelProject{
				VercelProjectID:   connection.VercelProjectID,
				VercelProjectName: connection.VercelProjectName,
			})
			targetMap[targetKey] = targetConn
		}
	}

	vercelConnections := make([]TiDBTargetVercelConnectionVO, 0, len(targetMap))
	for _, connection := range targetMap {
		vercelConnections = append(vercelConnections, connection)
	}
	return vercelConnections, nil
}

func (l *ClusterConnLogic) GetConnectionVOsByTiDBClusterID(ctxReq context.Context, connections []vercelDB.ClusterConnection, logger *zap.Logger) ([]TiDBVercelConnectionVO, error) {
	if len(connections) == 0 {
		return []TiDBVercelConnectionVO{}, nil
	}

	connectionsResponse := make([]TiDBVercelConnectionVO, 0, len(connections))
	for _, connection := range connections {
		connectionsResponse = append(connectionsResponse, TiDBVercelConnectionVO{
			TiDBProject: TiDBProject{
				TiDBOrgID:     strconv.FormatUint(connection.TiDBOrgID, 10),
				TiDBProjectID: strconv.FormatUint(connection.TiDBProjectID, 10),
			},
			VercelProject: VercelProject{
				VercelProjectID: connection.VercelProjectID,
			},
			ConnectionTarget: ConnectionTarget{
				TargetType:    TargetTypeCluster,
				TargetID:      strconv.FormatUint(connection.TiDBClusterID, 10),
				BranchEnabled: connection.BranchEnabled,
			},
		})
	}

	// Connections may be configured by different Vercel integrations
	configurationMap := map[string]bool{}
	var confIDs []string
	for i := range connections {
		id := connections[i].VercelIntegrationConfigurationID
		if _, ok := configurationMap[id]; !ok {
			confIDs = append(confIDs, id)
			configurationMap[id] = true
		}
	}
	for _, id := range confIDs {
		if err := l.fillVercelDetailInformation(ctxReq, id, connectionsResponse, logger); err != nil {
			return nil, err
		}
	}

	if err := l.fillTiDBDetailInformation(ctxReq, connectionsResponse, logger); err != nil {
		return nil, err
	}
	return connectionsResponse, nil
}

func (l *ClusterConnLogic) getConnectionVOsByConfiguration(ctx context.Context, clusterConnections []vercelDB.ClusterConnection, appConnections []vercelDB.DataAppConnection, req VercelIntegrationConfiguration, logger *zap.Logger) ([]TiDBVercelConnectionVO, error) {
	if len(clusterConnections) == 0 && len(appConnections) == 0 {
		return []TiDBVercelConnectionVO{}, nil
	}

	connectionsResponse := make([]TiDBVercelConnectionVO, 0, len(clusterConnections)+len(appConnections))
	for _, connection := range clusterConnections {
		connectionsResponse = append(connectionsResponse, TiDBVercelConnectionVO{
			TiDBProject: TiDBProject{
				TiDBOrgID:     strconv.FormatUint(connection.TiDBOrgID, 10),
				TiDBProjectID: strconv.FormatUint(connection.TiDBProjectID, 10),
			},
			VercelProject: VercelProject{
				VercelProjectID: connection.VercelProjectID,
			},
			ConnectionTarget: ConnectionTarget{
				TargetType:    TargetTypeCluster,
				TargetID:      strconv.FormatUint(connection.TiDBClusterID, 10),
				BranchEnabled: connection.BranchEnabled,
			},
		})
	}

	for _, connection := range appConnections {
		connectionsResponse = append(connectionsResponse, TiDBVercelConnectionVO{
			TiDBProject: TiDBProject{
				TiDBOrgID:     strconv.FormatUint(connection.TiDBOrgID, 10),
				TiDBProjectID: strconv.FormatUint(connection.TiDBProjectID, 10),
			},
			VercelProject: VercelProject{
				VercelProjectID: connection.VercelProjectID,
			},
			ConnectionTarget: ConnectionTarget{
				TargetType:    TargetTypeDataApp,
				TargetID:      strconv.FormatUint(connection.DataAppID, 10),
				BranchEnabled: false,
			},
		})
	}

	if err := l.fillVercelDetailInformation(ctx, req.ConfigurationID, connectionsResponse, logger); err != nil {
		return nil, err
	}

	if err := l.fillTiDBDetailInformation(ctx, connectionsResponse, logger); err != nil {
		return nil, err
	}
	return connectionsResponse, nil
}

func (l *ClusterConnLogic) RetrieveAllProjects(ctx context.Context, parameters vercelProxy.RetrieveProjectQueryParameters, token string, logger *zap.Logger) ([]vercelProxy.Project, error) {
	// if the page size is not set, set page size to default value 100
	if len(parameters.Limit) == 0 {
		parameters.Limit = "100"
	}

	var result []vercelProxy.Project
	projects, err := l.vercelAPIClient.RetrieveProjects(ctx, parameters, token)
	if err != nil {
		return nil, err
	}
	result = append(result, projects.Projects...)

	for projects.Pagination.Next != nil {
		parameters.Until = strconv.FormatInt(*projects.Pagination.Next, 10)
		projects, err = l.vercelAPIClient.RetrieveProjects(ctx, parameters, token)
		if err != nil {
			return nil, err
		}
		logger.Debug("Get project from vercel", zap.Any("page", projects.Pagination))
		result = append(result, projects.Projects...)
	}

	logger.Debug("Retrieve projects from vercel", zap.Any("projects", result))
	return result, nil
}

func (l *ClusterConnLogic) DeleteEnvsByEnvIdentifier(ctx context.Context, vercelProjectID string, teamID string, identifier vercelDB.EnvironmentIdentifier, token string) error {
	var envsToDelete []string
	if identifier.TiDBHostID.Valid && identifier.TiDBHostID.String != "" {
		envsToDelete = append(envsToDelete, identifier.TiDBHostID.String)
	}
	if identifier.TiDBPortID.Valid && identifier.TiDBPortID.String != "" {
		envsToDelete = append(envsToDelete, identifier.TiDBPortID.String)
	}
	if identifier.TiDBUserID.Valid && identifier.TiDBUserID.String != "" {
		envsToDelete = append(envsToDelete, identifier.TiDBUserID.String)
	}
	if identifier.TiDBPasswordID.Valid && identifier.TiDBPasswordID.String != "" {
		envsToDelete = append(envsToDelete, identifier.TiDBPasswordID.String)
	}
	if identifier.TiDBDatabaseID.Valid && identifier.TiDBDatabaseID.String != "" {
		envsToDelete = append(envsToDelete, identifier.TiDBDatabaseID.String)
	}
	if identifier.TiDBSslCaID.Valid && identifier.TiDBSslCaID.String != "" {
		envsToDelete = append(envsToDelete, identifier.TiDBSslCaID.String)
	}
	if identifier.DatabaseUrlID.Valid && identifier.DatabaseUrlID.String != "" {
		envsToDelete = append(envsToDelete, identifier.DatabaseUrlID.String)
	}

	return l.vercelAPIClient.DeleteEnvironmentVariables(ctx, vercelProjectID, teamID, envsToDelete, token)
}

func (l *ClusterConnLogic) DeleteBranchEnvsByEnvIdentifier(ctx context.Context, vercelProjectID string, teamID string, identifier vercelDB.BranchEnvironmentIdentifier, token string) error {
	envsToDelete := utils.GetBranchEnvsToDelete(identifier)

	return l.vercelAPIClient.DeleteEnvironmentVariables(ctx, vercelProjectID, teamID, envsToDelete, token)
}

func (l *ClusterConnLogic) CreateServerlessCluster(ctx context.Context, orgID, projectID uint64, email, clusterName string, vercelProjectIDs []string, configurationID string, logger *zap.Logger) (uint64, error) {
	// If vercel projects are selected, use the first one's functionRegion as the functionRegion.
	// If no vercel project is selected, use "iad1" as the functionRegion.
	vercelRegion := "iad1"
	if len(vercelProjectIDs) > 0 {
		pID := vercelProjectIDs[0]
		conf, err := l.configurationDAO.GetIntegrationConfigurationByID(ctx, configurationID)
		if err != nil {
			return 0, err
		}

		teamID := utils.GetTeamIDString(conf.TeamID)
		project, err := l.vercelAPIClient.GetProject(ctx, pID, teamID, conf.ApiAccessToken)
		if err != nil {
			return 0, err
		}
		vercelRegion = project.FunctionRegion
	}

	logger.Debug(fmt.Sprintf("region config: %#v\n vercelRegion: %s", l.conf.VercelConfig.RegionConfig, vercelRegion))
	// Search for nearest region in TiDB cluster by looking up the map in config.
	// If the mapping does not exit, use the default region to create serverless cluster.
	r := l.conf.VercelConfig.RegionConfig.DefaultRegion
	if v, found := l.conf.VercelConfig.RegionConfig.RegionMap[vercelRegion]; found {
		r = v
	}

	cluster, err := l.mgmtSvc.CreateServerlessCluster(ctx, orgID, projectID, email, clusterName, r.Provider, r.Region)
	if err != nil {
		return 0, err
	}
	return cluster.ClusterId, nil
}

func (l *ClusterConnLogic) WaitingClusterReady(ctx *gin.Context, clusterID uint64) error {
	ticker := time.NewTicker(3 * time.Second)
	timer := time.After(1 * time.Minute)
	for {
		select {
		case <-timer:
			return errors.New("Timeout waiting for cluster to be ready, please check status on dashboard.")
		case <-ticker.C:
			cluster, err := l.mgmtSvc.GetCluster(ctx, clusterID, mgmtapi.ClusterFilter_CLUSTER_VIEW_BASIC)
			if err != nil {
				return err
			}
			if cluster.GetServerlessCluster().GetUserStatus() == globalserverlessv1.UserClusterStatus_USER_CLUSTER_STATUS_NORMAL {
				return nil
			}
		}
	}
}

func (l *ClusterConnLogic) validateCluster(cluster *mgmtapi.Cluster, orgID, projectID, clusterID uint64) error {
	if cluster == nil {
		return errors.New("cluster not found")
	}

	if cluster.OrganizationId != orgID {
		return errors.New("cluster does not belong to this organization")
	}

	if cluster.ProjectId != projectID {
		return errors.New("cluster does not belong to this project")
	}

	if cluster.ClusterId != clusterID {
		return errors.Errorf("cluster id does not match, id in query: %d, id in cluster: %d", clusterID, cluster.ClusterId)
	}

	if cluster.DeployType != mgmtapi.DeployType_DEPLOY_TYPE_SERVERLESS {
		return errors.New("cluster is not serverless")
	}
	if cluster.GetServerlessCluster().GetUserStatus() == globalserverlessv1.UserClusterStatus_USER_CLUSTER_STATUS_DELETED {
		return errors.New("cluster is deleted")
	}

	return nil
}