package impl

import (
	"context"
	"strconv"

	"github.com/AlekSi/pointer"
	"github.com/elliotchance/pie/v2"
	workflowEntity "github.com/shiningrush/fastflow/pkg/entity"
	"github.com/tidbcloud/core-svc/idl/pbgen/proto/core"
	appconfig "github.com/tidbcloud/dataflow-service/app/config"
	"github.com/tidbcloud/dataflow-service/app/internal/status_matrix"
	"github.com/tidbcloud/dataflow-service/app/models/bucket"
	"github.com/tidbcloud/dataflow-service/app/models/domain"
	"github.com/tidbcloud/dataflow-service/app/repo"
	"github.com/tidbcloud/dataflow-service/app/repo/entity"
	"github.com/tidbcloud/dataflow-service/app/utils"
	"github.com/tidbcloud/dataflow-service/app/workflow"
	bizerr_wrapper "github.com/tidbcloud/dataflow-service/pkg/bizerr-wrapper"
	"github.com/tidbcloud/dataflow-service/pkg/grpc/grpcli"
	"github.com/tidbcloud/dataflow-service/pkg/log"
	"github.com/tidbcloud/pkgv2/bizerr"
	"github.com/tidbcloud/pkgv2/db"
	"go.uber.org/zap"
)

type eksBucketModel struct {
	baseBucketModel
	dbProvider              db.TransProvider
	bucketRepo              repo.BucketRepo
	bucketConfigurationRepo repo.AWSBucketConfigurationRepo
	bucketReplicationRepo   repo.AWSBucketReplicationRepo

	workflowClient         workflow.Workflow
	tidbClusterInfraClient grpcli.TiDBClusterInfraClient
	coreCli                grpcli.CoreGrpcClient
	awsBucketManager       bucket.AWSBucketManagerModel

	statusMatrix status_matrix.StatusMatrix
}

func NewEKSBucketModel(
	dbProvider db.TransProvider,
	bucketRepo repo.BucketRepo,
	bucketConfigurationRepo repo.AWSBucketConfigurationRepo,
	workflowClient workflow.Workflow,
	tidbClusterInfraClient grpcli.TiDBClusterInfraClient,
	coreCli grpcli.CoreGrpcClient,
	awsBucketManager bucket.AWSBucketManagerModel,
	bucketReplicationRepo repo.AWSBucketReplicationRepo,
) *eksBucketModel {
	return &eksBucketModel{
		baseBucketModel: baseBucketModel{
			bucketRepo: bucketRepo,
		},

		dbProvider:              dbProvider,
		bucketRepo:              bucketRepo,
		bucketConfigurationRepo: bucketConfigurationRepo,
		workflowClient:          workflowClient,
		tidbClusterInfraClient:  tidbClusterInfraClient,
		coreCli:                 coreCli,
		statusMatrix:            NewBucketStatusMatrix(),
		awsBucketManager:        awsBucketManager,
		bucketReplicationRepo:   bucketReplicationRepo,
	}
}

func (m *eksBucketModel) CreateBucket(
	ctx context.Context, req *domain.CreateBucketReqV2,
) (*domain.CreateBucketRespV2, bizerr.BizError) {
	ctx = log.WithLogID(ctx)
	logger := log.GetLogger(ctx)
	err := m.checkCreateBucketParams(logger, req)
	if err != nil {
		return nil, err
	}

	bucket, err := m.queryBucketIsAlreadyExisted(ctx, logger, req)
	if err != nil {
		return nil, err
	}
	if bucket != nil {
		logger.Info("bucket already exists",
			zap.Any("name", bucket.Name),
			zap.Any("id", bucket.ID),
			zap.Any("project_id", bucket.ProjectID),
			zap.Any("status", bucket.Status),
		)
		return &domain.CreateBucketRespV2{
			BucketID: uint64(bucket.ID),
			Status:   bucket.Status,
		}, nil
	}

	bucket, dagIns, err := m.createBucketEntity(ctx, logger, req)
	if err != nil {
		return nil, err
	}
	return &domain.CreateBucketRespV2{
		BucketID:   uint64(bucket.ID),
		Status:     bucket.Status,
		WorkflowID: &dagIns.ID,
	}, nil
}

func (m *eksBucketModel) createBucketEntity(
	ctx context.Context, logger *zap.Logger, req *domain.CreateBucketReqV2,
) (*entity.Bucket, *workflowEntity.DagInstance, bizerr.BizError) {
	bucket, err := m.genCreateBucketEntity(ctx, logger, req)
	if err != nil {
		return nil, nil, err
	}
	bucketConfiguration := m.genCreateBucketConfigurationEntity(req)

	if req.BusinessPurpose != nil {
		bucket.Extension = map[string]string{"type": string(*req.BusinessPurpose)}
	}
	var dagIns *workflowEntity.DagInstance
	dbErr := m.dbProvider.Transaction(ctx, func(ctx context.Context) error {
		err := m.bucketRepo.Save(ctx, bucket)
		if err != nil {
			return err
		}
		bucketConfiguration.BucketID = bucket.ID
		err = m.bucketConfigurationRepo.Save(ctx, bucketConfiguration)
		if err != nil {
			return err
		}

		// start wfs
		dagVars := m.buildCreateBucketDagVars(bucket, bucketConfiguration)
		var dagErr error
		dagIns, dagErr = m.workflowClient.RunDagWithTags(
			workflow.DagCreateBucketAWS, dagVars,
			map[string]string{
				workflow.BusinessTypeKey:   string(workflow.BusinessTypeBucket),
				workflow.BusinessActionKey: string(workflow.BusinessActionBucketCreate),
				workflow.BusinessID:        strconv.FormatUint(uint64(bucket.ID), 10),
			})
		if dagErr != nil {
			return dagErr
		}
		logger.Info("create bucket dag started", zap.Any("request", req), zap.Any("dag id", dagIns.ID))
		return nil
	})
	if dbErr != nil {
		logger.Error("create bucket failed", zap.Any("request", req), zap.Error(dbErr))
		return nil, nil, bizerr_wrapper.NewInternalErr(appconfig.ErrInternalError)
	}
	return bucket, dagIns, nil
}

func (m *eksBucketModel) genCreateBucketConfigurationEntity(
	req *domain.CreateBucketReqV2,
) *entity.AWSBucketConfiguration {
	bucketConfiguration := &entity.AWSBucketConfiguration{
		VersioningEnable: pointer.To(false),
		GCRuleEnable:     pointer.To(false),
	}
	if *req.BusinessPurpose == domain.BucketBusinessPurposeDualRegion ||
		*req.BusinessPurpose == domain.BucketBusinessPurposeRecoveryGroup {
		bucketConfiguration.VersioningEnable = pointer.To(true)
		bucketConfiguration.GCRuleEnable = pointer.To(true)
	}
	return bucketConfiguration
}

func (m *eksBucketModel) genCreateBucketEntity(
	ctx context.Context, logger *zap.Logger, req *domain.CreateBucketReqV2,
) (*entity.Bucket, bizerr.BizError) {
	bucket := &entity.Bucket{
		Name:              *req.Name,
		TenantID:          strconv.FormatUint(*req.TenantId, 10),
		ProjectID:         strconv.FormatUint(*req.ProjectId, 10),
		Regions:           req.Regions,
		Provider:          string(domain.InfraProviderTypeAWS),
		Scope:             *req.Scope,
		Status:            entity.BucketStatusCreating,
		InfraProviderType: string(*req.InfraCloudProvider),
	}
	accountID, err := m.getAccountID(ctx, logger, req)
	if err != nil {
		return nil, err
	}
	bucket.CloudAccountID = accountID

	cmekArn, err := m.getCmekArn(ctx, logger, req)
	if err != nil {
		return nil, err
	}
	if cmekArn != nil {
		bucket.AwsKmsArn = *cmekArn
	}
	return bucket, nil
}

func (m *eksBucketModel) getCmekArn(
	ctx context.Context, logger *zap.Logger,
	req *domain.CreateBucketReqV2,
) (*string, bizerr.BizError) {
	cmek, err := m.coreCli.GetAwsCmek(ctx, &core.GetAwsCmekReq{
		ProjectId:  *req.ProjectId,
		RegionName: &req.Regions[0],
	})
	if err != nil {
		logger.Error("get aws cmek failed", zap.Error(err))
		return nil, bizerr_wrapper.NewInternalErr(appconfig.ErrInternalError)
	}
	if len(cmek.GetCmeks()) > 0 {
		// one region only have one cmek
		return &cmek.GetCmeks()[0].KmsArn, nil
	}
	return nil, nil
}

func (m *eksBucketModel) buildCreateBucketDagVars(
	bucket *entity.Bucket, bucketConfiguration *entity.AWSBucketConfiguration,
) map[string]string {
	runVar := map[string]string{}
	runVar[workflow.KeyBucketID] = strconv.FormatUint(uint64(bucket.ID), 10)
	if bucketConfiguration.VersioningEnable != nil && *bucketConfiguration.VersioningEnable {
		runVar[workflow.KeyBucketVersioningSwitch] = workflow.ValueBucketVersioningSwitchTrue
	}
	if bucket.AwsKmsArn != "" {
		runVar[workflow.KeyBucketEncryptionSwitch] = workflow.ValueBucketEncryptionSwitchTrue
	}
	if bucketConfiguration.GCRuleEnable != nil && *bucketConfiguration.GCRuleEnable {
		runVar[workflow.KeyBucketGCRuleSwitch] = workflow.ValueBucketGCRuleSwitchTrue
	}
	return runVar
}

func (m *eksBucketModel) getAccountID(
	ctx context.Context, logger *zap.Logger, req *domain.CreateBucketReqV2,
) (*string, bizerr.BizError) {
	if req.CloudAccountID != nil {
		return req.CloudAccountID, nil
	}

	networkReq := &grpcli.RegionalNetworkReq{
		ProjectID:    *req.ProjectId,
		ProviderType: string(*req.InfraCloudProvider),
		Region:       req.Regions[0],
	}
	accountID, err := m.tidbClusterInfraClient.GetCloudAccountID(ctx, networkReq)
	if err != nil {
		logger.Error("get cloud account id failed", zap.Error(err))
		return nil, bizerr_wrapper.NewInternalErr(appconfig.ErrInternalError)
	}
	return &accountID, nil
}

func (m *eksBucketModel) queryBucketIsAlreadyExisted(
	ctx context.Context, logger *zap.Logger, req *domain.CreateBucketReqV2,
) (*entity.Bucket, bizerr.BizError) {
	accountID, err := m.getAccountID(ctx, logger, req)
	if err != nil {
		return nil, err
	}
	buckets, err := m.bucketRepo.ListByFilter(ctx, repo.BucketFilter{
		Name:           req.Name,
		CloudAccountID: accountID,
		DeletedTime:    pointer.To(int64(0)),
	})
	if err != nil {
		return nil, err
	}
	if len(buckets) == 0 {
		return nil, nil
	}
	if len(buckets) > 1 {
		logger.Error("bucket count is more than 1", zap.Any("request", req))
		return nil, bizerr_wrapper.NewInternalErr(appconfig.ErrInternalError)
	}
	if buckets[0].Status == entity.BucketStatusDeleting ||
		buckets[0].Status == entity.BucketStatusDeleted {
		logger.Warn("bucket is deleting or deleted", zap.Any("request", req))
		return nil, bizerr_wrapper.NewInternalErr(appconfig.ErrInternalError)
	}
	return buckets[0], nil
}

func (m *eksBucketModel) checkCreateBucketParams(
	logger *zap.Logger, req *domain.CreateBucketReqV2,
) bizerr.BizError {
	err := utils.CheckParamInvalidError("regions", req.Regions, func(param []string) bool {
		return len(param) == 1
	})
	if err != nil {
		logger.Warn("check param invalid error failed", zap.Error(err))
		return err
	}

	err = utils.CheckParamInvalidError("bucket_scope", req.Scope,
		func(param *entity.BucketScope) bool {
			return *param == entity.BucketScopeProject
		})
	if err != nil {
		logger.Warn("check param invalid error failed", zap.Error(err))
		return err
	}
	return nil
}

func (m *eksBucketModel) DeleteBucket(
	ctx context.Context, req *domain.DeleteBucketReqV2,
) (*domain.DeleteBucketRespV2, bizerr.BizError) {
	ctx = log.WithLogID(ctx)
	logger := log.GetLogger(ctx)

	bucket, err := m.bucketRepo.Load(ctx, *req.BucketID)
	// Reentrant processing
	if err != nil {
		if bizerr.ErrCodeRecordNotFound == err.ErrCode() {
			return &domain.DeleteBucketRespV2{}, nil
		}
		logger.Error("load bucket failed", zap.Any("request", req), zap.Error(err))
		return nil, bizerr_wrapper.NewInternalErr(appconfig.ErrInternalError)
	}
	if bucket.Status == entity.BucketStatusDeleting || bucket.Status == entity.BucketStatusDeleted {
		return &domain.DeleteBucketRespV2{}, nil
	}

	isFriendly := m.statusMatrix.IsActionFriendly2Status(status_matrix.CurrentStatus{
		Status: string(bucket.Status),
	}, string(domain.DeleteBucket))
	if !isFriendly {
		logger.Warn("bucket status is not friendly",
			zap.Any("request", req), zap.Any("action", "delete"))
		return nil, bizerr_wrapper.NewInvalidInputError(appconfig.ErrActionParam, bucket.Status, domain.DeleteBucket)
	}

	dagIns, err := m.deleteBucketEntity(ctx, logger, bucket, req)
	if err != nil {
		return nil, err
	}
	return &domain.DeleteBucketRespV2{
		WorkflowID: &dagIns.ID,
	}, nil
}

func (m *eksBucketModel) deleteBucketEntity(
	ctx context.Context, logger *zap.Logger, bucket *entity.Bucket, req *domain.DeleteBucketReqV2,
) (*workflowEntity.DagInstance, bizerr.BizError) {
	dagVars := m.buildDeleteBucketDagVars(bucket)
	var dagIns *workflowEntity.DagInstance
	dbErr := m.dbProvider.Transaction(ctx, func(ctx context.Context) error {
		// to deleting
		nextStatus := m.statusMatrix.CalculateNextStatusForAction(status_matrix.CurrentStatus{
			Status: string(bucket.Status),
		}, string(domain.DeleteBucket))
		bucket.Status = entity.BucketStatus(nextStatus.Status)
		err := m.bucketRepo.Save(ctx, bucket)
		if err != nil {
			return err
		}
		var dagErr error
		dagIns, dagErr = m.workflowClient.RunDagWithTags(workflow.DagDeleteBucketAWS, dagVars, map[string]string{
			workflow.BusinessTypeKey:   string(workflow.BusinessTypeBucket),
			workflow.BusinessActionKey: string(workflow.BusinessActionBucketDelete),
			workflow.BusinessID:        strconv.FormatUint(uint64(bucket.ID), 10),
		})
		if dagErr != nil {
			return dagErr
		}
		logger.Info("delete bucket dag started", zap.Any("request", req), zap.Any("dag id", dagIns.ID))
		return nil
	})
	if dbErr != nil {
		logger.Error("delete bucket failed", zap.Any("request", req), zap.Error(dbErr))
		return nil, bizerr_wrapper.NewInternalErr(appconfig.ErrInternalError)
	}
	return dagIns, nil
}

func (m *eksBucketModel) buildDeleteBucketDagVars(bucket *entity.Bucket) map[string]string {
	runVar := map[string]string{
		workflow.KeyBucketID: strconv.FormatUint(uint64(bucket.ID), 10),
	}
	return runVar
}

func (m *eksBucketModel) GrantAccessToBucket(
	ctx context.Context, req *domain.GrantAccessToBucketReq,
) (*domain.GrantAccessToBucketResp, bizerr.BizError) {
	ctx = log.WithLogID(ctx)
	logger := log.GetLogger(ctx)

	bucket, err := m.queryBucketAndCheckActionsIsValid(ctx, req.BucketID, domain.EditBucketPolicy)
	if err != nil {
		return nil, err
	}

	bucketConfig, err := m.bucketConfigurationRepo.Get(ctx,
		func(qs entity.AWSBucketConfigurationQuerySet) entity.AWSBucketConfigurationQuerySet {
			return qs.BucketIDEq(entity.ID(*req.BucketID))
		})
	if err != nil {
		logger.Error("load bucket configuration error", zap.Any("bucket id", req.BucketID), zap.Error(err))
		return nil, err
	}

	updateAWSBucketPolicyReq := &domain.AddAWSAllowRoleArnToBucketPolicyReq{
		AWSBucketReq: domain.AWSBucketReq{
			Name:         bucket.Name,
			AWSAccountID: *bucket.CloudAccountID,
			Region:       bucket.Regions[0],
		},
		IAMRoleARNs: req.CloudProviderIdentities,
	}

	if err := m.dbProvider.Transaction(ctx, func(ctx context.Context) error {
		// lock the bucket configuration to avoid race condition
		if err := m.bucketConfigurationRepo.Lock(ctx, bucketConfig.ID.Uint64()); err != nil {
			return err
		}

		resp, err := m.awsBucketManager.AddAllowRoleArnToBucketPolicy(ctx, updateAWSBucketPolicyReq)
		if err != nil {
			return err
		}
		bucketConfig.BucketPolicy.AllowARNs = resp.IAMRoleARNs
		return m.bucketConfigurationRepo.Save(ctx, bucketConfig)
	}); err != nil {
		logger.Error("update bucket policy error", zap.Any("bucket id", bucket.ID), zap.Error(err))
		return nil, bizerr_wrapper.NewInternalErr(appconfig.ErrInternalError)
	}
	return &domain.GrantAccessToBucketResp{}, nil
}

func (m *eksBucketModel) RevokeAccessToBucket(
	ctx context.Context, req *domain.RevokeAccessToBucketReq,
) (*domain.RevokeAccessToBucketResp, bizerr.BizError) {
	ctx = log.WithLogID(ctx)
	logger := log.GetLogger(ctx)

	bucket, err := m.queryBucketAndCheckActionsIsValid(ctx, req.BucketID, domain.EditBucketPolicy)
	if err != nil {
		return nil, err
	}

	bucketConfig, err := m.bucketConfigurationRepo.Get(ctx,
		func(qs entity.AWSBucketConfigurationQuerySet) entity.AWSBucketConfigurationQuerySet {
			return qs.BucketIDEq(entity.ID(*req.BucketID))
		})
	if err != nil {
		logger.Error("load bucket configuration error", zap.Any("bucket id", req.BucketID), zap.Error(err))
		return nil, err
	}

	if err := m.dbProvider.Transaction(ctx, func(ctx context.Context) error {
		// Serialize revoke with grants and other policy updates for this bucket.
		if err := m.bucketConfigurationRepo.Lock(ctx, bucketConfig.ID.Uint64()); err != nil {
			return err
		}

		deleteReq := &domain.DeleteAWSRoleArnFromBucketPolicyReq{
			AWSBucketReq: domain.AWSBucketReq{
				Name:         bucket.Name,
				AWSAccountID: *bucket.CloudAccountID,
				Region:       bucket.Regions[0],
			},
			IAMRoleARNs: req.CloudProviderIdentities,
		}
		resp, err := m.awsBucketManager.DeleteRoleArnFromBucketPolicy(ctx, deleteReq)
		if err != nil {
			return err
		}
		bucketConfig.BucketPolicy.AllowARNs = resp.IAMRoleARNs
		return m.bucketConfigurationRepo.Save(ctx, bucketConfig)
	}); err != nil {
		logger.Error("update bucket policy error", zap.Any("bucket id", bucket.ID), zap.Error(err))
		return nil, bizerr_wrapper.NewInternalErr(appconfig.ErrInternalError)
	}
	return &domain.RevokeAccessToBucketResp{}, nil
}

func (m *eksBucketModel) CreateBucketReplication(
	ctx context.Context, req *domain.CreateBucketReplicationReq,
) (*domain.CreateBucketReplicationResp, bizerr.BizError) {
	ctx = log.WithLogID(ctx)
	logger := log.GetLogger(ctx)

	sourceBucket, err := m.queryBucketAndCheckActionsIsValid(ctx, req.SourceBucketID, domain.EditBucketReplications)
	if err != nil {
		return nil, err
	}

	targetBucket, err := m.queryBucketAndCheckActionsIsValid(ctx, req.TargetBucketID, domain.EditBucketReplications)
	if err != nil {
		return nil, err
	}

	sourceBucketConfig, err := m.bucketConfigurationRepo.Get(ctx,
		func(qs entity.AWSBucketConfigurationQuerySet) entity.AWSBucketConfigurationQuerySet {
			return qs.BucketIDEq(entity.ID(*req.SourceBucketID))
		})
	if err != nil {
		logger.Error("load bucket configuration error", zap.Any("bucket id", req.SourceBucketID), zap.Error(err))
		return nil, err
	}

	replicationEntity, err := m.createAWSReplicationEntity(ctx, req, sourceBucket, targetBucket)
	if err != nil {
		logger.Error("create bucket replicationEntity error", zap.Any("source bucket id", sourceBucket.ID),
			zap.Any("target bucket id", targetBucket.ID), zap.Error(err))
		return nil, bizerr_wrapper.NewInternalErr(appconfig.ErrInternalError)
	}

	updateReq := m.buildAddAWSBucketReplicationReq(req, sourceBucket, targetBucket)
	if err := m.dbProvider.Transaction(ctx, func(ctx context.Context) error {
		resp, bizError := m.awsBucketManager.AddBucketReplication(ctx, updateReq)
		if bizError != nil {
			return bizError
		}
		sourceBucketConfig.BucketReplicationPolicy.RuleIDs = resp.RuleIDs
		bizError = m.bucketConfigurationRepo.Save(ctx, sourceBucketConfig)
		if bizError != nil {
			return bizError
		}
		replicationEntity.Status = entity.AWSBucketReplicationStatusCreated
		return m.bucketReplicationRepo.Save(ctx, replicationEntity)
	}); err != nil {
		logger.Error("create bucket replicationEntity error", zap.Any("source bucket id", sourceBucket.ID),
			zap.Any("target bucket id", targetBucket.ID), zap.Error(err))
		return nil, bizerr_wrapper.NewInternalErr(appconfig.ErrInternalError)
	}
	return &domain.CreateBucketReplicationResp{}, nil
}

func (m *eksBucketModel) buildAddAWSBucketReplicationReq(req *domain.CreateBucketReplicationReq,
	sourceBucket *entity.Bucket, targetBucket *entity.Bucket,
) *domain.AddAWSBucketReplicationReq {
	updateReq := &domain.AddAWSBucketReplicationReq{
		AWSBucketReq: domain.AWSBucketReq{
			Name:         sourceBucket.Name,
			AWSAccountID: *sourceBucket.CloudAccountID,
			Region:       sourceBucket.Regions[0],
		},
		ReplicationRule: &domain.AWSReplicationRule{
			ID: req.ReplicationPrefix,
			Destination: &domain.AWSDestination{
				Bucket: &targetBucket.Name,
			},
			Filter: &domain.AWSReplicationRuleFilter{
				Prefix: req.ReplicationPrefix,
			},
		},
	}
	return updateReq
}

func (m *eksBucketModel) createAWSReplicationEntity(ctx context.Context,
	req *domain.CreateBucketReplicationReq, sourceBucket *entity.Bucket, targetBucket *entity.Bucket,
) (*entity.AWSBucketReplication, bizerr.BizError) {
	replicationEntity, err := m.bucketReplicationRepo.Get(ctx,
		func(qs entity.AWSBucketReplicationQuerySet) entity.AWSBucketReplicationQuerySet {
			return qs.SourceBucketIDEq(sourceBucket.ID).
				TargetBucketIDEq(targetBucket.ID).
				PrefixEq(*req.ReplicationPrefix)
		})
	if err != nil {
		if err.ErrCode() != bizerr.ErrCodeRecordNotFound {
			log.GetLogger(ctx).Warn("load bucket replicationEntity error")
			return nil, err
		}
		replicationEntity = &entity.AWSBucketReplication{
			SourceBucketID: sourceBucket.ID,
			TargetBucketID: targetBucket.ID,
			Prefix:         *req.ReplicationPrefix,
			ClusterID:      entity.ID(*req.ClusterID),
			Status:         entity.AWSBucketReplicationStatusCreating,
		}
	}
	replicationEntity.Status = entity.AWSBucketReplicationStatusCreating
	err = m.bucketReplicationRepo.Save(ctx, replicationEntity)
	if err != nil {
		return nil, err
	}
	return replicationEntity, nil
}

func (m *eksBucketModel) queryBucketAndCheckActionsIsValid(
	ctx context.Context, bucketID *uint64, action domain.BucketAction,
) (*entity.Bucket, bizerr.BizError) {
	logger := log.GetLogger(ctx)
	err := utils.CheckParamEmptyError("bucket id", bucketID)
	if err != nil {
		logger.Warn("bucket id is empty", zap.Any("bucket id", bucketID))
		return nil, err
	}
	bucket, err := m.bucketRepo.Load(ctx, *bucketID)
	if err != nil {
		logger.Error("load bucket failed", zap.Any("bucketID", bucketID), zap.Error(err))
		return nil, bizerr_wrapper.NewInternalErr(appconfig.ErrInternalError)
	}

	isFriendly := m.statusMatrix.IsActionFriendly2Status(status_matrix.CurrentStatus{
		Status: string(bucket.Status),
	}, string(action))
	if !isFriendly {
		logger.Warn("bucket status is not friendly", zap.Any("bucket id", bucket.ID),
			zap.Any("bucket status", bucket.Status), zap.Any("action", action))
		return nil, bizerr_wrapper.NewInvalidInputError(appconfig.ErrActionParam, bucket.Status, action)
	}
	return bucket, nil
}

// DeleteBucketReplication just mark bucket replication deleting, worker will reconcile to clean up
func (m *eksBucketModel) DeleteBucketReplication(
	ctx context.Context, req *domain.DeleteBucketReplicationReq,
) (*domain.DeleteBucketReplicationResp, bizerr.BizError) {
	ctx = log.WithLogID(ctx)
	logger := log.GetLogger(ctx)

	replicationEntity, err := m.bucketReplicationRepo.Get(ctx,
		func(qs entity.AWSBucketReplicationQuerySet) entity.AWSBucketReplicationQuerySet {
			return qs.SourceBucketIDEq(entity.ID(*req.SourceBucketID)).
				TargetBucketIDEq(entity.ID(*req.TargetBucketID)).
				PrefixEq(*req.ReplicationPrefix)
		})
	if err != nil {
		if err.ErrCode() == bizerr.ErrCodeRecordNotFound {
			logger.Info("bucket replicationEntity not found")
			return &domain.DeleteBucketReplicationResp{}, nil
		}
		logger.Error("load bucket replicationEntity error")
		return nil, err
	}

	sourceBucket, err := m.bucketRepo.Load(ctx, *req.SourceBucketID)
	if err != nil {
		return nil, bizerr_wrapper.NewInternalErr(appconfig.ErrInternalError)
	}

	targetBucket, err := m.bucketRepo.Load(ctx, *req.TargetBucketID)
	if err != nil {
		return nil, bizerr_wrapper.NewInternalErr(appconfig.ErrInternalError)
	}

	sourceBucketConfig, err := m.bucketConfigurationRepo.Get(ctx,
		func(qs entity.AWSBucketConfigurationQuerySet) entity.AWSBucketConfigurationQuerySet {
			return qs.BucketIDEq(entity.ID(*req.SourceBucketID))
		})
	if err != nil {
		logger.Error("load bucket configuration error", zap.Any("bucket id", req.SourceBucketID), zap.Error(err))
		return nil, err
	}

	if err := m.dbProvider.Transaction(ctx, func(ctx context.Context) error {
		sourceBucketConfig.BucketReplicationPolicy.RuleIDs = pie.Of(sourceBucketConfig.BucketReplicationPolicy.RuleIDs).
			Filter(func(ruleID string) bool {
				return ruleID != *req.ReplicationPrefix
			}).Result
		if err := m.bucketConfigurationRepo.Save(ctx, sourceBucketConfig); err != nil {
			logger.Error("update bucket configuration error",
				zap.Any("bucket id", req.SourceBucketID), zap.Error(err))
			return err
		}

		replicationEntity.Status = entity.AWSBucketReplicationStatusDeleting
		return m.bucketReplicationRepo.Save(ctx, replicationEntity)
	}); err != nil {
		logger.Error("create bucket replicationEntity error", zap.Any("source bucket id", sourceBucket.ID),
			zap.Any("target bucket id", targetBucket.ID), zap.Error(err))
		return nil, bizerr_wrapper.NewInternalErr(appconfig.ErrInternalError)
	}
	return &domain.DeleteBucketReplicationResp{}, nil
}