package cluster

import (
	"context"

	"github.com/tidbcloud/pkgv2/db"

	globalserverlessv1 "github.com/tidbcloud/tidb-management-service/api/spec/global/serverless/v1"
	regionalserverlessv1 "github.com/tidbcloud/tidb-management-service/api/spec/regional/serverless/v1"
	openapicommonv1beta1 "github.com/tidbcloud/tidb-management-service/api/spec/tidb_cloud_open_api/common/v1beta1"
	"github.com/tidbcloud/tidb-management-service/tidb-mgmt-service/domain/adapter/account"
	"github.com/tidbcloud/tidb-management-service/tidb-mgmt-service/domain/adapter/billing"
	"github.com/tidbcloud/tidb-management-service/tidb-mgmt-service/domain/adapter/eventbus"
	"github.com/tidbcloud/tidb-management-service/tidb-mgmt-service/domain/adapter/remoteencryptor"
	"github.com/tidbcloud/tidb-management-service/tidb-mgmt-service/domain/adapter/serverlessglobal"
	"github.com/tidbcloud/tidb-management-service/tidb-mgmt-service/domain/entity"
	"github.com/tidbcloud/tidb-management-service/tidb-mgmt-service/domain/repository"
	"github.com/tidbcloud/tidb-management-service/tidb-mgmt-service/pkg/common"
	"github.com/tidbcloud/tidb-management-service/tidb-mgmt-service/pkg/config"
	"github.com/tidbcloud/tidb-management-service/utilities/enum"
	workflowimpl "github.com/tidbcloud/tidb-management-service/utilities/workflow"
)

type Service interface {
	CreateCluster(ctx context.Context, cluster *entity.Cluster) (*entity.Cluster, error)
	GetCluster(ctx context.Context, clusterId uint64, view entity.ClusterView) (*entity.Cluster, error)
	CountClusters(ctx context.Context, filter *repository.ClusterFilter) (int64, error)
	ListClusters(ctx context.Context, filter *repository.ClusterFilter, page *common.Page) ([]*entity.Cluster, error)
	UpdateCluster(ctx context.Context, cluster *entity.Cluster) (*entity.Cluster, error)
	DeleteCluster(ctx context.Context, clusterId uint64) (*entity.Cluster, *entity.ClusterCustomStatus, error)
	CleanClusters(ctx context.Context, organizationId uint64) error
	BatchCreateClusters(ctx context.Context, clusters []*entity.Cluster) ([]*entity.Cluster, error)
	RentCluster(ctx context.Context,
		shadowPoolID, targetProjectID uint64,
		namePrefix, passwd, sourceIP, tag string) (string, *entity.Cluster, error)
	ClaimClusterFromZero(ctx context.Context, cluster *entity.Cluster, zeroId string) (*entity.Cluster, error)
	// GetClusterByZeroID retrieves a cluster by its zero ID, validating the embedded secret.
	GetClusterByZeroID(ctx context.Context, zeroId string) (*entity.Cluster, error)
	// MoveCluster moves a single Starter or Essential cluster to a different project (synchronous).
	// displayName is optional: if non-empty it becomes the cluster's new display name after the
	// move; if empty the existing display name is kept unchanged.
	MoveCluster(ctx context.Context, orgId, clusterId, targetProjectId uint64, displayName string) (*entity.Cluster, error)
	// BatchMoveClusters moves multiple Starter/Essential clusters to a different project concurrently.
	// Failures are collected and returned; they do not prevent other clusters from being processed.
	BatchMoveClusters(ctx context.Context,
		orgId, sourceProjectId, targetProjectId uint64,
		servicePlans []string,
	) (total int32, successIds, failedIds []string, err error)
	// TakeoverCluster picks a prepared shadow cluster from the given pool, resets its password,
	// and transitions it to TakenOver state. The entire operation is synchronous and transactional.
	// orgID must match the pool's OrganizationID, otherwise a 403 error is returned.
	TakeoverCluster(ctx context.Context, shadowPoolID uint64, orgID uint64, passwd string) (*entity.Cluster, error)
	BindTags(ctx context.Context, organizationID, clusterID uint64, canonicalTags []string) ([]entity.Tag, error)
	UnbindTags(ctx context.Context, organizationID, clusterID uint64, canonicalTags []string) ([]entity.Tag, error)
}

type service struct {
	clusterRepo        repository.ClusterRepository
	resourceTagRepo    repository.ResourceTagRepository
	bindingRepo        repository.ClusterTagBindingRepository
	zeroRentRecordRepo repository.ZeroRentRecordRepository
	serverlessCli      serverlessglobal.ServerlessGlobal
	customStatusRepo   repository.CustomStatusRepository
	lockRepo           repository.LockRepo
	shadowPoolRepo     repository.ShadowPoolRepository
	accountCli         account.Client
	billingCli         billing.Client
	config             *config.BizConfig
	transaction        db.TransProvider
	encryptor          remoteencryptor.Client
	workflow           workflowimpl.Workflow
	orgRuConfigRepo    repository.OrganizationRuConfigRepository
	eventCommitter     eventbus.WALCommitter
}

func NewService(
	clusterRepo repository.ClusterRepository,
	resourceTagRepo repository.ResourceTagRepository,
	bindingRepo repository.ClusterTagBindingRepository,
	zeroRentRecordRepo repository.ZeroRentRecordRepository,
	customStatusRepo repository.CustomStatusRepository,
	lockRepo repository.LockRepo,
	shadowPoolRepo repository.ShadowPoolRepository,
	serverlessCli serverlessglobal.ServerlessGlobal,
	accountCli account.Client,
	billingCli billing.Client,
	cfg *config.BizConfig,
	transaction db.TransProvider,
	encryptor remoteencryptor.Client,
	workflow workflowimpl.Workflow,
	orgRuConfigRepo repository.OrganizationRuConfigRepository,
	eventCommitter eventbus.WALCommitter,
) (Service, error) {
	svc := &service{
		clusterRepo:        clusterRepo,
		resourceTagRepo:    resourceTagRepo,
		bindingRepo:        bindingRepo,
		zeroRentRecordRepo: zeroRentRecordRepo,
		customStatusRepo:   customStatusRepo,
		lockRepo:           lockRepo,
		shadowPoolRepo:     shadowPoolRepo,
		serverlessCli:      serverlessCli,
		accountCli:         accountCli,
		billingCli:         billingCli,
		config:             cfg,
		transaction:        transaction,
		encryptor:          encryptor,
		workflow:           workflow,
		orgRuConfigRepo:    orgRuConfigRepo,
		eventCommitter:     eventCommitter,
	}

	return svc, nil
}

func cvtServerlessCluster(
	base *entity.Cluster,
	serverlessCluster *globalserverlessv1.ServerlessServiceCluster,
) *entity.Cluster {
	if serverlessCluster == nil {
		return base
	}
	base.GlobalCluster = serverlessCluster

	// In the past, there was no region and cloud provider columns in mgmt db.
	// The columns were added in the later db schema migrations.
	// For compatibility, we need to update the region and cloud provider field in cluster entity.
	if base.Region == "" {
		if globalRegion := serverlessCluster.GetRegion(); globalRegion != "" {
			base.Region = globalRegion
		}
	}

	// not set cloud provider if it is already set
	if base.CloudProvider == openapicommonv1beta1.Region_CLOUD_PROVIDER_UNSPECIFIED {
		if globalProvider := serverlessCluster.GetCloudProvider(); globalProvider != "" {
			if p, err := enum.FromString[openapicommonv1beta1.Region_CloudProvider](globalProvider); err == nil {
				base.CloudProvider = openapicommonv1beta1.Region_CloudProvider(p)
			}
		}
	}

	if base.EngineType == openapicommonv1beta1.EngineType_ENGINE_TYPE_UNSPECIFIED {
		engineType := serverlessCluster.GetEngineType()
		if engineType != "" {
			if et, err := enum.FromString[openapicommonv1beta1.EngineType](engineType); err == nil {
				base.EngineType = openapicommonv1beta1.EngineType(et)
			}
		}
	}

	if base.EngineType == openapicommonv1beta1.EngineType_ENGINE_TYPE_UNSPECIFIED {
		base.EngineType = openapicommonv1beta1.EngineType_MYSQL_COMPATIBLE
	}

	if serverlessCluster.EngineType == "" {
		serverlessCluster.EngineType = base.EngineType.String()
	}

	// For the reason that deletion is performed asynchronously
	// clusters that have been deleted in mgmt may be normal in global.
	if base.DeletedAt != nil {
		serverlessCluster.UserStatus = globalserverlessv1.UserClusterStatus_USER_CLUSTER_STATUS_DELETED
		if serverlessCluster.Cluster != nil {
			serverlessCluster.Cluster.State = regionalserverlessv1.ClusterState_CLUSTER_STATE_DELETED
		}
	}

	return base
}

// cvt mgmt state to regional state
func cvtServerlessState(in openapicommonv1beta1.ClusterState) regionalserverlessv1.ClusterState {
	switch in {
	case openapicommonv1beta1.ClusterState_CREATING:
		return regionalserverlessv1.ClusterState_CLUSTER_STATE_CREATING
	case openapicommonv1beta1.ClusterState_DELETING:
		return regionalserverlessv1.ClusterState_CLUSTER_STATE_DELETING
	case openapicommonv1beta1.ClusterState_ACTIVE,
		openapicommonv1beta1.ClusterState_RESTORING,
		openapicommonv1beta1.ClusterState_MAINTENANCE,
		openapicommonv1beta1.ClusterState_IMPORTING:
		return regionalserverlessv1.ClusterState_CLUSTER_STATE_READY
	case openapicommonv1beta1.ClusterState_DELETED:
		return regionalserverlessv1.ClusterState_CLUSTER_STATE_DELETED
	}

	return regionalserverlessv1.ClusterState_CLUSTER_STATE_UNSPECIFIED
}