package gcp

import (
	"context"
	"fmt"
	"time"

	"github.com/avast/retry-go/v4"
	"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"
	"github.com/tidbcloud/dataflow-service/app/repo/entity"
	"github.com/tidbcloud/dataflow-service/app/workflow"
	"github.com/tidbcloud/dataflow-service/app/workflow/action"
	"github.com/tidbcloud/dataflow-service/app/workflow/action/bucket/base"
	"github.com/tidbcloud/pkgv2/db"
	"go.uber.org/zap"
)

type CreateBucketAction struct {
	base.BaseAction
	GcpBucketManager bucket.GCPBucketManagerModel
}

func NewCreateBucketAction(
	transProvider db.TransProvider,
	bucketRepo repo.BucketRepo,
	gcpBucketManager bucket.GCPBucketManagerModel,
) *CreateBucketAction {
	return &CreateBucketAction{
		BaseAction:       base.NewBaseAction(transProvider, bucketRepo),
		GcpBucketManager: gcpBucketManager,
	}
}

func (a *CreateBucketAction) Name() string {
	return workflow.ActionCreateBucketGCP
}

func (a *CreateBucketAction) Run(ctx run.ExecuteContext, params interface{}) error {
	logger := action.GetLogger(ctx).Named(a.Name())
	logger.Info("action run start")
	err := action.InvokeRunActionWithRetry(ctx.Context(), logger, func() error {
		bucket, err := a.QueryBucket(ctx, logger)
		if err != nil {
			return err
		}
		isReady := a.checkBucketIsReady(ctx.Context(), logger, bucket)
		if isReady {
			logger.Info("bucket is ready", zap.Any("bucket id", bucket.ID))
			return nil
		}
		return a.createBucket(ctx, logger, bucket)
	})
	if err != nil {
		logger.Error("action run error", zap.Error(err))
		return err
	}
	logger.Info("action run finished")
	return nil
}

func (a *CreateBucketAction) createBucket(ctx run.ExecuteContext, logger *zap.Logger, bucket *entity.Bucket) error {
	tags := map[string]string{
		"environment": action.GenInfraCREnvironmentTagValue(),
		"servicetype": "dedicated",
		"usedby":      "custom-tidb",
		"component":   bucket.GenResourceComponentTagValue(),
		"tenant":      bucket.TenantID,
		"project":     bucket.ProjectID,
	}
	createBucketReq := &domain.CreateGCPBucketReq{
		Name:      bucket.Name,
		Regions:   bucket.Regions,
		ProjectID: *bucket.CloudAccountID,
		Tags:      tags,
	}
	err := a.GcpBucketManager.CreateBucket(ctx.Context(), createBucketReq)
	if err != nil {
		logger.Warn("create aws bucket error", zap.Any("bucket id", bucket.ID), zap.Error(err))
		return err
	}
	return nil
}

func (a *CreateBucketAction) checkBucketIsReady(
	ctx context.Context, logger *zap.Logger, bucket *entity.Bucket,
) bool {
	err := a.GcpBucketManager.CheckBucketIfExists(ctx, &domain.CheckGCPBucketReq{
		Name: bucket.Name,
	})
	if err != nil {
		logger.Warn("check bucket if exists error", zap.String("bucket_name", bucket.Name), zap.Error(err))
		return false
	}
	return true
}

//nolint:gomnd
func (a *CreateBucketAction) RunAfter(ctx run.ExecuteContext, params interface{}) error {
	logger := action.GetLogger(ctx).Named(a.Name())
	logger.Info("action runAfter start")
	err := retry.Do(func() error {
		bucket, err := a.QueryBucket(ctx, logger)
		if err != nil {
			return err
		}
		isReady := a.checkBucketIsReady(ctx.Context(), logger, bucket)
		if isReady {
			logger.Info("bucket is ready", zap.Any("bucket id", bucket.ID))
			return nil
		}
		logger.Info("bucket is not ready", zap.Any("bucket id", bucket.ID))
		return fmt.Errorf("bucket is not ready")
	},
		retry.Context(ctx.Context()),
		retry.DelayType(retry.FixedDelay),
		retry.Delay(time.Second*3),
		retry.Attempts(0),
		retry.OnRetry(func(n uint, err error) {
			logger.Info("action runAfter error", zap.Uint("retry_times", n))
		}),
		retry.LastErrorOnly(true),
	)
	if err != nil {
		logger.Error("action runAfter error", zap.Error(err))
		return err
	}
	logger.Info("action runAfter finished")
	return nil
}