package action

import (
	"context"
	"fmt"
	"path/filepath"
	"strconv"

	"github.com/tidbcloud/fastflow/pkg/entity/run"
	regionaldedicatedv1 "github.com/tidbcloud/infra-provider/api/tidbcloud/regional-server/v1"
	"github.com/tidbcloud/pkgv2/db"
	"github.com/tidbcloud/tiinf/monitor/tilog"
	"go.uber.org/zap"

	globalcommonv1 "github.com/tidbcloud/tidb-management-service/api/spec/global/common/v1"
	globaldedicatedv1 "github.com/tidbcloud/tidb-management-service/api/spec/global/dedicated/v1"
	accountserver "github.com/tidbcloud/tidb-management-service/dedicated-global-service/domain/adapter/account"
	"github.com/tidbcloud/tidb-management-service/dedicated-global-service/domain/adapter/clusterconfig"
	"github.com/tidbcloud/tidb-management-service/dedicated-global-service/domain/adapter/eventbus"
	"github.com/tidbcloud/tidb-management-service/dedicated-global-service/domain/adapter/regionserver"
	"github.com/tidbcloud/tidb-management-service/dedicated-global-service/domain/entity"
	cidrrepo "github.com/tidbcloud/tidb-management-service/dedicated-global-service/domain/repository/cidr"
	clusterrepo "github.com/tidbcloud/tidb-management-service/dedicated-global-service/domain/repository/cluster"
	"github.com/tidbcloud/tidb-management-service/dedicated-global-service/domain/repository/cmek"
	"github.com/tidbcloud/tidb-management-service/dedicated-global-service/domain/repository/region"
	"github.com/tidbcloud/tidb-management-service/dedicated-global-service/domain/repository/tidbgroup"
	"github.com/tidbcloud/tidb-management-service/dedicated-global-service/domain/service/workflow/common"
	"github.com/tidbcloud/tidb-management-service/utilities/bizerr"
	"github.com/tidbcloud/tidb-management-service/utilities/clusterevent"
	"github.com/tidbcloud/tidb-management-service/utilities/workflow"
)

const (
	CreateCluster = "create_cluster"
)

type CreateClusterAction struct {
	clusterRepo   clusterrepo.Repo
	tidbGroupRepo tidbgroup.Repo
	regionRepo    region.Repo
	cidrRepo      cidrrepo.Repo
	cmekRepo      cmek.Repo

	clusterConfigCli clusterconfig.Client
	regionalCli      regionserver.Client
	accountCli       accountserver.Client

	// Caution: If you publish event in the action, you should put code inside a transaction.
	eventCommitter eventbus.WALCommitter
	transaction    db.Transactioner
}

func NewCreateClusterAction(
	clusterRepo clusterrepo.Repo,
	regionRepo region.Repo,
	tidbGroupRepo tidbgroup.Repo,
	cidrRepo cidrrepo.Repo,
	cmekRepo cmek.Repo,
	clusterConfigCli clusterconfig.Client,
	regionalCli regionserver.Client,
	accountCli accountserver.Client,
	committer eventbus.WALCommitter,
	transaction db.Transactioner,
) *CreateClusterAction {
	return &CreateClusterAction{
		clusterRepo:   clusterRepo,
		regionRepo:    regionRepo,
		tidbGroupRepo: tidbGroupRepo,
		cidrRepo:      cidrRepo,
		cmekRepo:      cmekRepo,
		// auditLogSvc: auditLogSvc,
		clusterConfigCli: clusterConfigCli,
		regionalCli:      regionalCli,
		accountCli:       accountCli,
		eventCommitter:   committer,
		transaction:      transaction,
	}
}

func (a *CreateClusterAction) Name() string {
	return CreateCluster
}

func (a *CreateClusterAction) ParameterNew() any {
	return &common.ClusterIdParameter{}
}

func (a *CreateClusterAction) Run(ctx run.ExecuteContext, params any) error {
	cctx, span := workflow.GenerateContext(ctx)

	clusterId, err := common.GetClusterId(params)
	if err != nil {
		tilog.Error(cctx, "failed to get cluster id", zap.Error(err))
		return err
	}

	vpcId, ok := ctx.ShareData().Get(common.KeyVpcId)
	if !ok {
		tilog.Error(cctx, "failed to get vpc_id")
		return bizerr.New(bizerr.ErrCodeUnknown, "failed to get vpc_id")
	}
	if vpcId == "" {
		tilog.Error(cctx, "vpc_id is empty")
		return bizerr.New(bizerr.ErrCodeUnknown, "vpc_id is empty")
	}

	tilog.Info(cctx, "start to create cluster", zap.Uint64("cluster_id", clusterId))
	defer func() {
		span.End()
		tilog.Info(cctx, "end to create cluster", zap.Uint64("cluster_id", clusterId))
	}()

	if err := a.transaction.Transaction(cctx, func(_ context.Context) error {
		cluster, err := a.clusterRepo.GetAndLoadComponents(cctx, clusterId)
		if err != nil {
			tilog.Error(cctx, "failed to get cluster", zap.Error(err), zap.Uint64("cluster_id", clusterId))
			return err
		}
		if cluster.GetDeleteTime() != nil {
			tilog.Warn(cctx, "cluster is deleted", zap.Uint64("cluster", cluster.ClusterId))
			return nil
		}

		// initializd event
		initializedEv := entity.NewClusterEventBuilder(cctx, cluster, clusterevent.EventTypeInitialized).Build()

		if err := a.callRegionServer(cctx, cluster, vpcId); err != nil {
			tilog.Error(cctx, "failed to call region server", zap.Error(err), zap.Uint64("cluster_id", clusterId))
			return err
		}

		// preparing event
		preparingEv := entity.NewClusterEventBuilder(cctx, cluster, clusterevent.EventTypePreparing).Build()
		if err := a.eventCommitter.BatchWrite(cctx, clusterevent.TopicClusterEvent, []clusterevent.Event{initializedEv, preparingEv}); err != nil {
			tilog.Error(cctx, "failed to publish cluster event", zap.Error(err))
			return err
		}

		return nil
	}); err != nil {
		tilog.Error(cctx, "failed to create cluster", zap.Error(err), zap.Uint64("cluster_id", clusterId))
		return err
	}

	return nil
}

func (a *CreateClusterAction) buildVpcSettings(
	ctx context.Context,
	cluster *entity.Cluster,
	vpcId string,
) (*regionaldedicatedv1.Cluster_VpcSettings, bizerr.BizError) {
	vpcSettings := &regionaldedicatedv1.Cluster_VpcSettings{
		Vpc: nil,
	}

	switch cluster.CloudProviderId {
	case globalcommonv1.CloudProvider_AWS:
		vpcSettings.Vpc = &regionaldedicatedv1.Cluster_VpcSettings_ExistingAwsVpcId{
			ExistingAwsVpcId: vpcId,
		}
	case globalcommonv1.CloudProvider_GCP:
		vpcSettings.Vpc = &regionaldedicatedv1.Cluster_VpcSettings_ExistingGcpNetworkName{
			ExistingGcpNetworkName: vpcId,
		}
	case globalcommonv1.CloudProvider_AZURE:
		vpcSettings.Vpc = &regionaldedicatedv1.Cluster_VpcSettings_ExistingAzureVnetId{
			ExistingAzureVnetId: vpcId,
		}
	default:
		tilog.Error(ctx, "unsupported cloud provider", zap.String("cloud_provider", cluster.CloudProviderId.String()))
		return nil, bizerr.New(bizerr.ErrCodeUnknown, "unsupported cloud provider %s", cluster.CloudProviderId.String())
	}

	return vpcSettings, nil
}

func (a *CreateClusterAction) callRegionServer(ctx context.Context, cluster *entity.Cluster, vpcId string) error {
	regionEntity, err := a.regionRepo.SelectRegionById(ctx, cluster.RegionId, cluster.GetOrganizationId())
	if err != nil {
		tilog.Error(ctx, "failed to find region", zap.Error(err), zap.String("region_id", cluster.RegionId.String()))
		return err
	}

	vpcSettings, err := a.buildVpcSettings(ctx, cluster, vpcId)
	if err != nil {
		tilog.Error(ctx, "failed to build vpc settings", zap.Error(err))
		return err
	}

	tidbGroupEn, err := a.tidbGroupRepo.LoadDefault(ctx, cluster.ClusterId)
	if err != nil {
		tilog.Error(ctx, "failed to load default work pool", zap.Error(err))
		return err
	}

	// check default node group
	for i := 0; i < len(cluster.ClusterComponents); i++ {
		if cluster.ClusterComponents[i].GetComponentType() == globaldedicatedv1.ComponentType_TIDB ||
			cluster.ClusterComponents[i].GetComponentType() == globaldedicatedv1.ComponentType_TIPROXY {
			if tidbGroupEn.Id != cluster.ClusterComponents[i].GetTidbGroupId() {
				tilog.Error(
					ctx,
					"tidbGroup id of tidb component is not match default tidb group",
					zap.Uint64("tidb_component_id", cluster.ClusterComponents[i].Id),
					zap.Uint64("default_tidb_group_id", tidbGroupEn.Id),
				)
				return bizerr.New(
					bizerr.ErrCodeParams,
					"tidbGroup id of tidb component %d is not match default tidb group",
					cluster.ClusterComponents[i].Id,
				)
			}
		}
	}

	var cmekEn *entity.CustomerManagedEncryptionKey
	if cluster.CustomerManagedEncryptionKeyId != "" {
		cmekId, perr := strconv.ParseUint(cluster.CustomerManagedEncryptionKeyId, 10, 64)
		if perr != nil {
			tilog.Error(ctx, "failed to parse cmek id", zap.Error(perr))
			return perr
		}
		cmekEn, err = a.cmekRepo.Get(ctx, cmekId)
		if err != nil {
			tilog.Error(ctx, "failed to get cmek", zap.Error(err))
			return err
		}
	}

	// build regional cluster
	regionCluster, err := cvtClusterEntityToRegionalRequest(ctx, cluster, vpcSettings, tidbGroupEn.Id, cmekEn)
	if err != nil {
		tilog.Error(ctx, "failed to convert cluster entity to regional request", zap.Error(err))
		return err
	}

	// set cluster tidb jwks
	jwks, err := a.accountCli.GetClusterAuthPubKey(ctx, cluster.OrganizationId, cluster.ProjectId, cluster.ClusterId)
	if err != nil {
		tilog.Error(ctx, "failed to get cluster auth public key", zap.Error(err))
		return err
	}
	regionCluster.Cluster.SecurityConfigurations.TidbAuthTokenJwks = jwks

	// get cluster component configurations from config center
	regionCluster.Cluster.ComponentConfigurations, err = a.clusterConfigCli.ListComponentConfigurations(ctx, cluster)
	if err != nil {
		tilog.Error(ctx, "failed to get component configurations", zap.Error(err))
		return err
	}

	// get image tag from config center
	regionCluster.Cluster.ClusterVersion, err = a.clusterConfigCli.GetClusterComponentsVersion(ctx, cluster)
	if err != nil {
		tilog.Error(ctx, "failed to get cluster version components", zap.Error(err))
		return err
	}

	// get cluster init sql statements from config center
	regionCluster.Cluster.InitSqlStatements, err = a.clusterConfigCli.GetClusterInitSqlStmts(ctx, cluster)
	if err != nil {
		tilog.Error(ctx, "failed to get cluster init sql statements", zap.Error(err))
		return err
	}

	if _, err := a.regionalCli.CreateCluster(ctx, regionEntity, regionCluster); err != nil {
		tilog.Error(ctx, "failed to create cluster", zap.Error(err))
		return err
	}

	return nil
}

func cvtClusterEntityToRegionalRequest(
	ctx context.Context,
	cluster *entity.Cluster,
	vpcSettings *regionaldedicatedv1.Cluster_VpcSettings,
	tidbGroupId uint64,
	cmekEn *entity.CustomerManagedEncryptionKey,
) (*regionaldedicatedv1.CreateClusterRequest, bizerr.BizError) {
	nodeSpec, err := buildRegionalNodeClaim(ctx, cluster.ClusterComponents)
	if err != nil {
		tilog.Error(ctx, "failed to build regional node claim", zap.Error(err))
		return nil, err
	}

	res := &regionaldedicatedv1.CreateClusterRequest{
		Cluster: &regionaldedicatedv1.Cluster{
			Id:                 strconv.FormatUint(cluster.ClusterId, 10),
			TenantId:           strconv.FormatUint(cluster.OrganizationId, 10),
			ProjectId:          strconv.FormatUint(cluster.ProjectId, 10),
			ComponentNodeSpecs: nodeSpec,
			// ComponentConfigurations:  get from cluster center not need set here.
			// ClusterVersion:         get from cluster center not need set here.
			VpcSettings:            vpcSettings,
			SecurityConfigurations: &regionaldedicatedv1.Cluster_SecurityConfigurations{},
			ClusterIdentifierHash:  cluster.ClusterIdentifierHash,
			Port:                   cluster.Port,
			ClusterVersion:         &regionaldedicatedv1.ClusterVersion{},
			DefaultTidbGroupId:     fmt.Sprintf("%d", tidbGroupId),
		},
	}

	if cmekEn != nil {
		switch cluster.CloudProviderId {
		case globalcommonv1.CloudProvider_AWS:
			res.Cluster.SecurityConfigurations.Cmek = &regionaldedicatedv1.Cluster_SecurityConfigurations_CmekAwsMasterKeyArn{
				CmekAwsMasterKeyArn: cmekEn.GetAwsKeyArn(),
			}
		case globalcommonv1.CloudProvider_AZURE:
			res.Cluster.SecurityConfigurations.Cmek = &regionaldedicatedv1.Cluster_SecurityConfigurations_AzureCmek{
				AzureCmek: &regionaldedicatedv1.Cluster_AzureCmek{
					ApplicationId: cmekEn.GetAzureKeyVault().
						GetAzureAppRegistration().
						GetAppRegistrationId(),
					ManagedIdentityResourceId: cmekEn.GetAzureKeyVault().
						GetAzureAppRegistration().
						GetAppRegistrationUmi(),
					KeyName:      cmekEn.GetAzureKeyVault().GetKeyVaultKey(),
					KeyVaultName: cmekEn.GetAzureKeyVault().GetKeyVaultName(),
				},
			}
		default:
			tilog.Error(
				ctx,
				"unsupported cmek cloud provider",
				zap.String("cloud_provider", cluster.CloudProviderId.String()),
			)
			return nil, bizerr.New(
				bizerr.ErrCodeUnknown,
				"unsupported cmek cloud provider %s",
				cluster.CloudProviderId.String(),
			)
		}
	}

	return res, nil
}

func buildRegionalNodeClaim(
	ctx context.Context,
	components []*entity.ClusterComponent,
) ([]*regionaldedicatedv1.Cluster_NodeSpec, bizerr.BizError) {
	if len(components) == 0 {
		tilog.Error(ctx, "empty cluster components")
		return nil, bizerr.New(bizerr.ErrCodeParams, "empty cluster components")
	}

	claim := make([]*regionaldedicatedv1.Cluster_NodeSpec, 0, len(components))

	for _, comp := range components {
		tilog.Info(ctx, "build regional node claim", zap.Any("component", comp))
		switch comp.GetComponentType() {
		case globaldedicatedv1.ComponentType_PD:
			dataStorage := comp.GetStorages()[globaldedicatedv1.Sku_DATA.String()]
			claim = append(claim, &regionaldedicatedv1.Cluster_NodeSpec{
				ComponentType: regionaldedicatedv1.ComponentType_PD,
				InstanceType:  comp.GetInstanceType(),
				Count:         comp.GetCount(),
				Storage: &regionaldedicatedv1.Cluster_Storage{
					Type:          dataStorage.GetVolumeType().String(),
					SizeGi:        dataStorage.GetSizeGi(),
					Iops:          dataStorage.GetIops(),
					ThroughputMbs: dataStorage.GetThroughputMibps(),
				},
			})

		case globaldedicatedv1.ComponentType_TIKV:
			dataStorage := comp.GetStorages()[globaldedicatedv1.Sku_DATA.String()]
			nodeSpec := &regionaldedicatedv1.Cluster_NodeSpec{
				ComponentType: regionaldedicatedv1.ComponentType_TIKV,
				InstanceType:  comp.GetInstanceType(),
				Count:         comp.GetCount(),
				Storage: &regionaldedicatedv1.Cluster_Storage{
					Type:          dataStorage.GetVolumeType().String(),
					SizeGi:        dataStorage.GetSizeGi(),
					Iops:          dataStorage.GetIops(),
					ThroughputMbs: dataStorage.GetThroughputMibps(),
				},
			}
			// need set raft log storage
			if raftStorage, ok := comp.GetStorages()[globaldedicatedv1.Sku_RAFT_LOG.String()]; ok {
				nodeSpec.AdditionalStorage = []*regionaldedicatedv1.Cluster_AdditionalStorage{
					{
						Name:          globaldedicatedv1.Sku_RAFT_LOG.String(),
						Type:          raftStorage.GetVolumeType().String(),
						SizeGi:        raftStorage.GetSizeGi(),
						Iops:          raftStorage.GetIops(),
						ThroughputMbs: raftStorage.GetThroughputMibps(),
						MountPath: filepath.Dir(
							entity.DefaultRaftLogDirPath,
						), // raft log storage mount path is the parent directory of raft log path
					},
				}
			}
			claim = append(claim, nodeSpec)

		case globaldedicatedv1.ComponentType_TIDB:
			claim = append(claim, &regionaldedicatedv1.Cluster_NodeSpec{
				ComponentType: regionaldedicatedv1.ComponentType_TIDB,
				InstanceType:  comp.GetInstanceType(),
				Count:         comp.GetCount(),
			})
		case globaldedicatedv1.ComponentType_TIPROXY:
			claim = append(claim, &regionaldedicatedv1.Cluster_NodeSpec{
				ComponentType: regionaldedicatedv1.ComponentType_TIPROXY,
				InstanceType:  comp.GetInstanceType(),
				Count:         comp.GetCount(),
			})
		case globaldedicatedv1.ComponentType_TIFLASH:
			dataStorage := comp.GetStorages()[globaldedicatedv1.Sku_DATA.String()]
			nodeSpec := &regionaldedicatedv1.Cluster_NodeSpec{
				ComponentType: regionaldedicatedv1.ComponentType_TIFLASH,
				InstanceType:  comp.GetInstanceType(),
				Count:         comp.GetCount(),
				Storage: &regionaldedicatedv1.Cluster_Storage{
					Type:          dataStorage.GetVolumeType().String(),
					SizeGi:        dataStorage.GetSizeGi(),
					Iops:          dataStorage.GetIops(),
					ThroughputMbs: dataStorage.GetThroughputMibps(),
				},
			}
			claim = append(claim, nodeSpec)

		default:
			tilog.Error(ctx, "invalid component type", zap.String("component_type", comp.GetComponentType().String()))
			return nil, bizerr.New(bizerr.ErrCodeParams, "invalid component type %s", comp.GetComponentType())
		}
	}

	return claim, nil
}