//nolint:gomnd
package create

import (
	"fmt"
	"strings"
	"time"

	"github.com/AlekSi/pointer"
	"github.com/avast/retry-go/v4"
	"github.com/elliotchance/pie/v2"
	"github.com/shiningrush/fastflow/pkg/entity/run"
	"github.com/tidbcloud/dataflow-service/app/models/bucket"
	"github.com/tidbcloud/dataflow-service/app/models/domain"
	"github.com/tidbcloud/dataflow-service/app/repo"
	backupRepo "github.com/tidbcloud/dataflow-service/app/repo/backup"
	"github.com/tidbcloud/dataflow-service/app/repo/entity"
	"github.com/tidbcloud/dataflow-service/app/utils"
	"github.com/tidbcloud/dataflow-service/app/workflow"
	"github.com/tidbcloud/dataflow-service/app/workflow/action"
	"github.com/tidbcloud/dataflow-service/app/workflow/action/backup_setting/base"
	service_config "github.com/tidbcloud/dataflow-service/internal/config"
	"github.com/tidbcloud/pkgv2/db"
	"go.uber.org/zap"
)

type createBucketActionParam struct {
	BackupSettingId uint64 `json:"backup_setting_id"`
	Region          string `json:"region"`
}

type CreateAndCheckBucketReadyAction struct {
	base.BaseAction
}

func NewCreateAndCheckBucketReadyAction(
	transProvider db.TransProvider,
	bucketRepo repo.BucketRepo,
	backupSettingRepo backupRepo.BackupSettingRepo,
	bucketModel bucket.BucketModelV2,
) *CreateAndCheckBucketReadyAction {
	return &CreateAndCheckBucketReadyAction{
		BaseAction: base.NewBaseAction(
			transProvider,
			bucketRepo,
			backupSettingRepo,
			bucketModel,
		),
	}
}

func (a *CreateAndCheckBucketReadyAction) Name() string {
	return workflow.ActionCreateAndCheckBucketReady
}

func (a *CreateAndCheckBucketReadyAction) Run(ctx run.ExecuteContext, params interface{}) error {
	logger := action.GetLogger(ctx).Named(a.Name())
	input := params.(*createBucketActionParam)
	logger.Info("action run", zap.Any("input", input))
	err := action.InvokeRunActionWithRetry(ctx.Context(), logger, func() error {
		return a.checkAndCreateBucket(ctx, logger, input)
	})
	if err != nil {
		logger.Error("action run error", zap.Error(err))
		return err
	}
	logger.Info("action run finished", zap.Any("input", input))
	return nil
}

func (a *CreateAndCheckBucketReadyAction) checkAndCreateBucket(
	ctx run.ExecuteContext, logger *zap.Logger, input *createBucketActionParam,
) error {
	backupSetting, err := a.BackupSettingRepo.Load(ctx.Context(), entity.ID(input.BackupSettingId))
	if err != nil {
		logger.Warn("load backup setting error", zap.Error(err))
		return err
	}
	// only gcp bucket has multiple regions
	regions := strings.Split(input.Region, workflow.SepBucketRegions)
	bucketName := utils.UniqueIdentifierLte63V2(backupSetting.GetBucketName(service_config.Global().Env, "", regions))

	req := &domain.CreateBucketReqV2{
		BusinessPurpose:    pointer.To(domain.BucketBusinessPurposeDefault),
		Scope:              pointer.To(entity.BucketScopeProject),
		TenantId:           pointer.To(uint64(backupSetting.TenantID)),
		ProjectId:          pointer.To(uint64(backupSetting.ProjectID)),
		ClusterId:          pointer.To(uint64(backupSetting.ClusterID)),
		Name:               pointer.To(bucketName),
		InfraCloudProvider: pointer.To(domain.InfraProviderType(backupSetting.InfraProviderType)),
		Regions:            regions,
		CloudAccountID:     backupSetting.CloudAccountID,
	}
	if backupSetting.Bucket.DualRegionEnable {
		req.BusinessPurpose = pointer.To(domain.BucketBusinessPurposeDualRegion)
	}
	bucket, err := a.BucketModel.CreateBucket(ctx.Context(), req)
	if err != nil {
		logger.Error("create bucket error", zap.Error(err))
		return err
	}
	ctx.ShareData().Set(workflow.KeyBucketID+input.Region, fmt.Sprintf("%v", bucket.BucketID))
	return nil
}

func (a *CreateAndCheckBucketReadyAction) RunAfter(ctx run.ExecuteContext, params interface{}) error {
	logger := action.GetLogger(ctx).Named(a.Name())
	input := params.(*createBucketActionParam)
	logger.Info("after action run start ", zap.Any("input", input))
	bucketId, err := action.GetUint64FromShareData(ctx.ShareData(), workflow.KeyBucketID+input.Region)
	if err != nil {
		logger.Error("get bucket id from context error", zap.Error(err))
		return err
	}
	err = retry.Do(func() error {
		return a.checkBucketStatus(ctx, logger, bucketId, input.BackupSettingId)
	},
		retry.Context(ctx.Context()),
		retry.DelayType(retry.FixedDelay),
		retry.Delay(time.Second*10),
		retry.Attempts(0),
		retry.OnRetry(func(n uint, err error) {
			logger.Info("check bucket status error", zap.Error(err), zap.Uint("retry_times", n))
		}),
		retry.LastErrorOnly(true),
	)
	if err != nil {
		logger.Error("check bucket status error", zap.Error(err))
		return err
	}
	logger.Info("after action run finish")
	return nil
}

func (a *CreateAndCheckBucketReadyAction) checkBucketStatus(
	ctx run.ExecuteContext, logger *zap.Logger, bucketID uint64, backupSettingID uint64,
) error {
	bucket, err := a.BucketRepo.Load(ctx.Context(), bucketID)
	if err != nil {
		logger.Warn("load bucket error", zap.Error(err))
		return err
	}
	if bucket.Status == entity.BucketStatusCreated || bucket.Status == entity.BucketStatusEditing {
		return a.updateBucketInfoToBackupSetting(ctx, logger, backupSettingID, bucket)
	}
	return fmt.Errorf("bucket status is %s", bucket.Status)
}

func (a *CreateAndCheckBucketReadyAction) updateBucketInfoToBackupSetting(
	ctx run.ExecuteContext, logger *zap.Logger, backupSettingID uint64, bucket *entity.Bucket,
) error {
	backupSetting, err := a.BackupSettingRepo.Load(ctx.Context(), entity.ID(backupSettingID))
	if err != nil {
		logger.Warn("load backup setting error", zap.Error(err))
		return err
	}
	for i, item := range backupSetting.Bucket.Buckets {
		result := pie.Of(item.Regions).Any(func(region string) bool {
			return region == bucket.Regions[0]
		})
		if result {
			backupSetting.Bucket.Buckets[i].BucketId = uint64(bucket.ID)
			err := a.BackupSettingRepo.Save(ctx.Context(), backupSetting)
			if err != nil {
				logger.Warn("save backup setting error", zap.Error(err))
				return err
			}
			return nil
		}
	}
	return fmt.Errorf("bucket region %s not in backup setting", bucket.Regions[0])
}

func (a *CreateAndCheckBucketReadyAction) ParameterNew() interface{} {
	return &createBucketActionParam{}
}