package customstatus

import (
	"context"
	"errors"
	"fmt"
	"strconv"

	"braces.dev/errtrace"
	"github.com/tidbcloud/pkgv2/db"
	"github.com/tidbcloud/tiinf/monitor/tilog"
	"go.opentelemetry.io/otel"
	"go.uber.org/zap"
	"gorm.io/gorm"

	globaldedicatedv1 "github.com/tidbcloud/tidb-management-service/api/spec/global/dedicated/v1"
	globalngv1 "github.com/tidbcloud/tidb-management-service/api/spec/global/nextgen/v1"
	globalserverlessv1 "github.com/tidbcloud/tidb-management-service/api/spec/global/serverless/v1"
	openapicommonv1beta1 "github.com/tidbcloud/tidb-management-service/api/spec/tidb_cloud_open_api/common/v1beta1"
	tidbmgmtservicev1 "github.com/tidbcloud/tidb-management-service/api/spec/tidb_mgmt_service/v1"
	"github.com/tidbcloud/tidb-management-service/tidb-mgmt-service/domain/adapter/dedicatedglobal"
	"github.com/tidbcloud/tidb-management-service/tidb-mgmt-service/domain/adapter/nextgenglobal"
	"github.com/tidbcloud/tidb-management-service/tidb-mgmt-service/domain/adapter/serverlessglobal"
	"github.com/tidbcloud/tidb-management-service/tidb-mgmt-service/domain/constant"
	"github.com/tidbcloud/tidb-management-service/tidb-mgmt-service/domain/entity"
	"github.com/tidbcloud/tidb-management-service/tidb-mgmt-service/domain/repository"
)

var tracer = otel.Tracer("Domain.CustomStatus")

type Service interface {
	SetClusterCustomStatus(
		ctx context.Context,
		status *entity.ClusterCustomStatus,
	) (*entity.ClusterCustomStatus, *entity.ClusterCustomStatus, openapicommonv1beta1.ClusterState, error)
	GetClusterCustomStatusByID(ctx context.Context, clusterId, statusId uint64) (*entity.ClusterCustomStatus, error)
	GetClusterCustomStatusByName(
		ctx context.Context,
		clusterId uint64,
		statusName string,
		statusType tidbmgmtservicev1.CustomStatusType,
	) (*entity.ClusterCustomStatus, error)
	ListClusterCustomStatus(
		ctx context.Context,
		clusterIds []uint64,
		statusTypes []tidbmgmtservicev1.CustomStatusType,
		includeDeleted bool,
	) (map[uint64][]*entity.ClusterCustomStatus, error)
	DeleteClusterCustomStatus(ctx context.Context, clusterId, statusId uint64) error
	ClearClusterCustomStatusRecordByClusterId(ctx context.Context, clusterId uint64) error
}

type service struct {
	clusterRepo      repository.ClusterRepository
	customStatusRepo repository.CustomStatusRepository

	dedicatedCli  dedicatedglobal.DedicatedGlobal
	serverlessCli serverlessglobal.ServerlessGlobal
	nextgenCli    nextgenglobal.NextGenGlobal
	transaction   db.TransProvider
}

func NewService(
	clusterRepo repository.ClusterRepository,
	customStatusRepo repository.CustomStatusRepository,
	dedicatedCli dedicatedglobal.DedicatedGlobal,
	serverlessCli serverlessglobal.ServerlessGlobal,
	nextgenCli nextgenglobal.NextGenGlobal,
	transaction db.TransProvider,
) Service {
	go func() {
		ctx := context.Background()
		err := customStatusRepo.ClearExpiredClusterCustomStatus(ctx)
		if err != nil {
			tilog.Error(ctx, "clear expired cluster custom status failed", zap.Error(err))
		}
	}()

	return &service{
		clusterRepo:      clusterRepo,
		customStatusRepo: customStatusRepo,
		dedicatedCli:     dedicatedCli,
		serverlessCli:    serverlessCli,
		nextgenCli:       nextgenCli,
		transaction:      transaction,
	}
}

func (svc *service) SetClusterCustomStatus(ctx context.Context, status *entity.ClusterCustomStatus) (
	clusterCustomStatus *entity.ClusterCustomStatus,
	mutexCustomStatus *entity.ClusterCustomStatus,
	mutexState openapicommonv1beta1.ClusterState,
	err error,
) {
	ctx, span := tracer.Start(ctx, "SetClusterCustomStatus")
	defer span.End()

	if status == nil {
		return nil, nil, mutexState, errtrace.Wrap(errors.New("status is invalid"))
	}
	status.StatusId = entity.GenStatusId()

	if err := svc.transaction.Transaction(ctx, func(ctx context.Context) error {
		// Serialize custom status registration with resource-pool pause admission.
		// The pause path locks the same cluster rows after NG enters PAUSE_PRECHECKING.
		if err := svc.customStatusRepo.LockClustersForUpdate(ctx, []uint64{status.ClusterId}); err != nil {
			if errors.Is(err, gorm.ErrRecordNotFound) {
				return errtrace.Wrap(constant.ErrClusterNotFound)
			}
			return fmt.Errorf("failed to lock cluster for custom status: %w", err)
		}
		// check cluster exist
		cluster, err := svc.clusterRepo.GetCluster(ctx, status.ClusterId, true)
		if err != nil {
			if errors.Is(err, gorm.ErrRecordNotFound) {
				return errtrace.Wrap(constant.ErrClusterNotFound)
			}
			return fmt.Errorf("failed to find cluster in database: %w", err)
		}

		// check mutex
		var isMutex bool
		isMutex, mutexCustomStatus, err = svc.customStatusRepo.IsMutexCustomStatus(ctx, status)
		if err != nil {
			return fmt.Errorf("failed to check mutex custom status: %w", err)
		}
		if isMutex {
			tilog.Error(ctx, "mutex custom status is exist", zap.Any("new custom status", status), zap.Any("mutex custom status", mutexCustomStatus))
			return fmt.Errorf("status type: %s, err: %w", mutexCustomStatus.StatusType.String(), constant.ErrClusterCustomStatusForbidden)
		}

		// check global cluster status
		// action -> cluster real state
		if status.CustomMutex != nil && len(status.CustomMutex.ClusterAction) != 0 {
			isMutex, mutexState, err = svc.isMutexClusterState(ctx, cluster, status)
			if err != nil {
				return fmt.Errorf("failed to check global cluster state: %w", err)
			}
			if isMutex {
				switch mutexState {
				case openapicommonv1beta1.ClusterState_CLUSTER_STATE_UNSPECIFIED:
					return fmt.Errorf("cluster is not realy, err: %w", constant.ErrClusterCustomStatusForbidden)
				default:
					return fmt.Errorf("cluster running action: %s, err: %w", mutexState.String(), constant.ErrClusterCustomStatusForbidden)
				}
			}
		}

		// TODO: remove this
		if status.StatusType == tidbmgmtservicev1.CustomStatusType_CUSTOM_STATUS_TYPE_IMPORT && cluster.DeployType == entity.DeployTypeDedicated {
			status.OverrideState = openapicommonv1beta1.ClusterState_IMPORTING.Enum()
		}

		id, err := svc.customStatusRepo.NewClusterCustomStatus(ctx, status)
		if err != nil {
			return fmt.Errorf("failed to insert cluster custom status record: %w", err)
		}

		// get statusRecord
		status, err = svc.customStatusRepo.GetClusterCustomStatusById(ctx, status.ClusterId, id)
		if err != nil {
			return fmt.Errorf("failed to get cluster custom status record: %w", err)
		}
		return nil
	}); err != nil {
		return status, mutexCustomStatus, mutexState, err
	}

	return status, mutexCustomStatus, mutexState, nil
}

// check the global cluster state with the status mutex action
func (svc *service) isMutexClusterState(
	ctx context.Context,
	cluster *entity.Cluster,
	status *entity.ClusterCustomStatus,
) (bool, openapicommonv1beta1.ClusterState, error) {
	if cluster == nil || status == nil {
		return false, openapicommonv1beta1.ClusterState_CLUSTER_STATE_UNSPECIFIED, errtrace.Wrap(
			errors.New("cluster or status is invalid"),
		)
	}

	switch cluster.DeployType {
	case entity.DeployTypeDedicated:
		globalCluster, err := svc.dedicatedCli.GetCluster(
			ctx,
			&globaldedicatedv1.GetClusterRequest{ClusterId: status.ClusterId},
		)
		if err != nil {
			return false, openapicommonv1beta1.ClusterState_CLUSTER_STATE_UNSPECIFIED, fmt.Errorf(
				"failed to get global cluster: %w",
				err,
			)
		}
		if globalCluster == nil {
			return false, openapicommonv1beta1.ClusterState_CLUSTER_STATE_UNSPECIFIED, errtrace.Wrap(
				errors.New("global cluster object should not be nil"),
			)
		}
		// check cluster status
		for _, mutexAction := range status.CustomMutex.ClusterAction {
			if mutex, state := ifDedicatedStateCausedByAction(mutexAction, globalCluster.State); mutex {
				tilog.Warn(
					ctx,
					"cluster global state is mutex",
					zap.Uint64("cluster_id", status.ClusterId),
					zap.String("cluster_state", globalCluster.State.String()),
					zap.Any("mutex action", mutexAction),
				)
				return true, state, nil
			}
		}
	case entity.DeployTypeNextGen:
		globalCluster, err := svc.nextgenCli.GetCluster(
			ctx,
			&globalngv1.GetClusterRequest{ClusterId: status.ClusterId},
		)
		if err != nil {
			return false, openapicommonv1beta1.ClusterState_CLUSTER_STATE_UNSPECIFIED, fmt.Errorf(
				"failed to get global cluster: %w",
				err,
			)
		}
		if globalCluster == nil {
			return false, openapicommonv1beta1.ClusterState_CLUSTER_STATE_UNSPECIFIED, errtrace.Wrap(
				errors.New("global cluster object should not be nil"),
			)
		}
		if status.IsActionMutexed(tidbmgmtservicev1.ClusterAction_CLUSTER_ACTION_PAUSE_CLUSTER) &&
			globalCluster.GetSharedPoolId() != "" {
			resourcePoolID, err := strconv.ParseUint(globalCluster.GetSharedPoolId(), 10, 64)
			if err != nil {
				return false, openapicommonv1beta1.ClusterState_CLUSTER_STATE_UNSPECIFIED, fmt.Errorf(
					"failed to parse nextgen resource pool id %q: %w",
					globalCluster.GetSharedPoolId(),
					err,
				)
			}
			resourcePoolProfile, err := svc.nextgenCli.GetResourcePool(
				ctx,
				&globalngv1.GetResourcePoolRequest{ResourcePoolId: resourcePoolID},
			)
			if err != nil {
				return false, openapicommonv1beta1.ClusterState_CLUSTER_STATE_UNSPECIFIED, fmt.Errorf(
					"failed to get nextgen resource pool: %w",
					err,
				)
			}
			if resourcePoolProfile == nil || resourcePoolProfile.GetResourcePool() == nil {
				return false, openapicommonv1beta1.ClusterState_CLUSTER_STATE_UNSPECIFIED, errtrace.Wrap(
					errors.New("nextgen resource pool object should not be nil"),
				)
			}
			if mutex, state := ifNextgenResourcePoolStateCausedByAction(
				tidbmgmtservicev1.ClusterAction_CLUSTER_ACTION_PAUSE_CLUSTER,
				resourcePoolProfile.GetResourcePool().GetLifecycleState(),
			); mutex {
				tilog.Warn(
					ctx,
					"resource pool global state is mutex",
					zap.Uint64("cluster_id", status.ClusterId),
					zap.Uint64("resource_pool_id", resourcePoolID),
					zap.String("resource_pool_state", resourcePoolProfile.GetResourcePool().GetLifecycleState().String()),
				)
				return true, state, nil
			}
		}
		// check cluster status
		for _, mutexAction := range status.CustomMutex.ClusterAction {
			if mutex, state := ifNextgenStateCausedByAction(mutexAction, globalCluster.State); mutex {
				tilog.Warn(
					ctx,
					"cluster global state is mutex",
					zap.Uint64("cluster_id", status.ClusterId),
					zap.String("cluster_state", globalCluster.State.String()),
					zap.Any("mutex action", mutexAction),
				)
				return true, state, nil
			}
		}
	case entity.DeployTypeServerless:
		globalClusters, err := svc.serverlessCli.ListClusters(ctx, &globalserverlessv1.ServerlessServiceListRequest{
			Filter: &globalserverlessv1.ClusterFilter{
				ClusterIds:  []string{strconv.FormatUint(status.ClusterId, 10)},
				ShowDetails: false,
			},
		})
		if err != nil {
			return false, openapicommonv1beta1.ClusterState_CLUSTER_STATE_UNSPECIFIED, fmt.Errorf(
				"failed to get global cluster: %w",
				err,
			)
		}
		if globalClusters == nil || len(globalClusters.Items) == 0 {
			return false, openapicommonv1beta1.ClusterState_CLUSTER_STATE_UNSPECIFIED, errtrace.Wrap(
				errors.New("global cluster object should not be nil"),
			)
		}

		if globalClusters.Items[0].UserStatus == globalserverlessv1.UserClusterStatus_USER_CLUSTER_STATUS_CREATING {
			tilog.Warn(
				ctx,
				"cluster is not ready",
				zap.Uint64("cluster_id", status.ClusterId),
				zap.String("cluster_state", globalClusters.Items[0].UserStatus.String()),
			)
			return true, openapicommonv1beta1.ClusterState_CREATING, nil
		}

	default:
		return false, 0, errtrace.Wrap(errors.New("cluster deploy type is invalid"))
	}

	return false, 0, nil
}

func (svc *service) GetClusterCustomStatusByID(
	ctx context.Context,
	clusterId, statusId uint64,
) (*entity.ClusterCustomStatus, error) {
	ctx, span := tracer.Start(ctx, "GetClusterCustomStatusByID")
	defer span.End()

	if clusterId == 0 {
		return nil, errtrace.Wrap(errors.New("cluster id is invalid"))
	}

	if statusId == 0 {
		return nil, errtrace.Wrap(errors.New("status id is invalid"))
	}

	var (
		status *entity.ClusterCustomStatus
		err    error
	)
	if err := svc.transaction.Transaction(ctx, func(ctx context.Context) error {
		// check cluster exist
		_, err = svc.clusterRepo.GetCluster(ctx, clusterId, true)
		if err != nil {
			if errors.Is(err, gorm.ErrRecordNotFound) {
				return errtrace.Wrap(constant.ErrClusterNotFound)
			}
			return fmt.Errorf("failed to find cluster in database: %w", err)
		}

		// get status record
		status, err = svc.customStatusRepo.GetClusterCustomStatusById(ctx, clusterId, statusId)
		if err != nil {
			return fmt.Errorf("failed to get cluster custom status record: %w", err)
		}
		return nil
	}); err != nil {
		return status, errtrace.Wrap(err)
	}

	return status, nil
}

func (svc *service) GetClusterCustomStatusByName(
	ctx context.Context,
	clusterId uint64,
	statusName string,
	statusType tidbmgmtservicev1.CustomStatusType,
) (*entity.ClusterCustomStatus, error) {
	ctx, span := tracer.Start(ctx, "GetClusterCustomStatusByName")
	defer span.End()

	if clusterId == 0 {
		return nil, errtrace.Wrap(errors.New("cluster id is invalid"))
	}

	if statusType == tidbmgmtservicev1.CustomStatusType_CUSTOM_STATUS_TYPE_UNSPECIFIED || statusName == "" {
		return nil, errtrace.Wrap(errors.New("status type and status name is invalid"))
	}

	var (
		status *entity.ClusterCustomStatus
		err    error
	)
	if err := svc.transaction.Transaction(ctx, func(ctx context.Context) error {
		// check cluster exist
		_, err = svc.clusterRepo.GetCluster(ctx, clusterId, true)
		if err != nil {
			if errors.Is(err, gorm.ErrRecordNotFound) {
				return errtrace.Wrap(constant.ErrClusterNotFound)
			}
			return fmt.Errorf("failed to find cluster in database: %w", err)
		}

		status, err = svc.customStatusRepo.GetClusterCustomStatusByName(ctx, clusterId, statusName, statusType)
		if err != nil {
			return fmt.Errorf("failed to get cluster custom status record: %w", err)
		}

		return nil
	}); err != nil {
		return status, errtrace.Wrap(err)
	}

	return status, nil
}

func (svc *service) ListClusterCustomStatus(
	ctx context.Context,
	clusterIds []uint64,
	statusTypes []tidbmgmtservicev1.CustomStatusType,
	includeDeleted bool,
) (map[uint64][]*entity.ClusterCustomStatus, error) {
	ctx, span := tracer.Start(ctx, "ListClusterCustomStatus")
	defer span.End()

	cluster2statuses, err := svc.customStatusRepo.ListClusterCustomStatus(ctx, clusterIds, statusTypes, includeDeleted)
	if err != nil {
		if errors.Is(err, gorm.ErrRecordNotFound) {
			return nil, errtrace.Wrap(err)
		}
		return nil, fmt.Errorf("failed to query cluster custom status record: %w", err)
	}

	return cluster2statuses, nil
}

func (svc *service) DeleteClusterCustomStatus(ctx context.Context, clusterId, statusId uint64) error {
	ctx, span := tracer.Start(ctx, "DeleteClusterCustomStatus")
	defer span.End()

	if clusterId == 0 {
		return errtrace.Wrap(errors.New("cluster id is invalid"))
	}

	if statusId == 0 {
		return errtrace.Wrap(errors.New("status id is invalid"))
	}

	if err := svc.transaction.Transaction(ctx, func(ctx context.Context) error {
		// check cluster exist
		status, err := svc.customStatusRepo.GetClusterCustomStatusById(ctx, clusterId, statusId)
		if err != nil {
			if errors.Is(err, gorm.ErrRecordNotFound) {
				return nil // already deleted
			}
			return fmt.Errorf("failed to find cluster in database: %w", err)
		}

		// delete status record
		err = svc.customStatusRepo.DeleteClusterCustomStatus(ctx, status.StatusId)
		if err != nil {
			return fmt.Errorf("failed to delete cluster custom status record: %w", err)
		}

		return nil
	}); err != nil {
		return errtrace.Wrap(err)
	}

	return nil
}

func (svc *service) ClearClusterCustomStatusRecordByClusterId(ctx context.Context, clusterId uint64) error {
	ctx, span := tracer.Start(ctx, "ClearClusterCustomStatusRecordByClusterId")
	defer span.End()

	if clusterId == 0 {
		return errtrace.Wrap(errors.New("cluster id is invalid"))
	}

	if err := svc.transaction.Transaction(ctx, func(ctx context.Context) error {
		// check cluster exist
		_, err := svc.clusterRepo.GetCluster(ctx, clusterId, true)
		if err != nil {
			if errors.Is(err, gorm.ErrRecordNotFound) {
				return errtrace.Wrap(constant.ErrClusterNotFound)
			}
			return fmt.Errorf("failed to find cluster in database: %w", err)
		}

		err = svc.customStatusRepo.ClearClusterCustomStatusByClusterId(ctx, clusterId)
		if err != nil {
			return fmt.Errorf("failed to clear cluster custom status record: %w", err)
		}
		return nil
	}); err != nil {
		return errtrace.Wrap(err)
	}

	return nil
}

func (svc *service) ClearExpiredClusterCustomStatusRecord(ctx context.Context) error {
	ctx, span := tracer.Start(ctx, "ClearExpiredClusterCustomStatusRecord")
	defer span.End()
	err := svc.customStatusRepo.ClearExpiredClusterCustomStatus(ctx)
	if err != nil {
		return fmt.Errorf("failed to clear cluster custom status record: %w", err)
	}

	return nil
}

// If the cluster state is caused by the action, it is considered mutex
func ifDedicatedStateCausedByAction(
	action tidbmgmtservicev1.ClusterAction,
	clusterState globaldedicatedv1.Cluster_State,
) (bool, openapicommonv1beta1.ClusterState) {
	switch action {
	case tidbmgmtservicev1.ClusterAction_CLUSTER_ACTION_MODIFY_CLUSTER:
		if clusterState == globaldedicatedv1.Cluster_MODIFYING {
			return true, openapicommonv1beta1.ClusterState_MODIFYING
		}
	case tidbmgmtservicev1.ClusterAction_CLUSTER_ACTION_RESUME_CLUSTER:
		if clusterState == globaldedicatedv1.Cluster_RESUMING {
			return true, openapicommonv1beta1.ClusterState_RESUMING
		}
	case tidbmgmtservicev1.ClusterAction_CLUSTER_ACTION_PAUSE_CLUSTER:
		switch clusterState {
		case globaldedicatedv1.Cluster_PAUSING:
			return true, openapicommonv1beta1.ClusterState_PAUSING
		case globaldedicatedv1.Cluster_PAUSED:
			return true, openapicommonv1beta1.ClusterState_PAUSED
		}
	case tidbmgmtservicev1.ClusterAction_CLUSTER_ACTION_SCALE_CLUSTER:
		if clusterState == globaldedicatedv1.Cluster_SCALING {
			return true, openapicommonv1beta1.ClusterState_MODIFYING
		}
	}

	return false, openapicommonv1beta1.ClusterState_CLUSTER_STATE_UNSPECIFIED
}

// If the cluster state is caused by the action, it is considered mutex
func ifNextgenStateCausedByAction(
	action tidbmgmtservicev1.ClusterAction,
	clusterState globalngv1.Cluster_State,
) (bool, openapicommonv1beta1.ClusterState) {
	switch action {
	case tidbmgmtservicev1.ClusterAction_CLUSTER_ACTION_MODIFY_CLUSTER:
		if clusterState == globalngv1.Cluster_MODIFYING {
			return true, openapicommonv1beta1.ClusterState_MODIFYING
		}
	case tidbmgmtservicev1.ClusterAction_CLUSTER_ACTION_PAUSE_CLUSTER:
		switch clusterState {
		case globalngv1.Cluster_PAUSING:
			return true, openapicommonv1beta1.ClusterState_PAUSING
		case globalngv1.Cluster_PAUSED:
			return true, openapicommonv1beta1.ClusterState_PAUSED
		}
	}
	return false, openapicommonv1beta1.ClusterState_CLUSTER_STATE_UNSPECIFIED
}

func ifNextgenResourcePoolStateCausedByAction(
	action tidbmgmtservicev1.ClusterAction,
	resourcePoolState globalngv1.ResourcePool_LifecycleState,
) (bool, openapicommonv1beta1.ClusterState) {
	if action != tidbmgmtservicev1.ClusterAction_CLUSTER_ACTION_PAUSE_CLUSTER {
		return false, openapicommonv1beta1.ClusterState_CLUSTER_STATE_UNSPECIFIED
	}

	switch resourcePoolState {
	case globalngv1.ResourcePool_PAUSE_PRECHECKING, globalngv1.ResourcePool_PAUSING:
		return true, openapicommonv1beta1.ClusterState_PAUSING
	case globalngv1.ResourcePool_PAUSED:
		return true, openapicommonv1beta1.ClusterState_PAUSED
	case globalngv1.ResourcePool_RESUMING:
		return true, openapicommonv1beta1.ClusterState_RESUMING
	default:
		return false, openapicommonv1beta1.ClusterState_CLUSTER_STATE_UNSPECIFIED
	}
}