package aws

import (
	"context"
	"fmt"
	"strconv"
	"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
	AwsBucketManager bucket.AWSBucketManagerModel
}

func NewCreateBucketAction(
	transProvider db.TransProvider,
	bucketRepo repo.BucketRepo,
	awsBucketManager bucket.AWSBucketManagerModel,
) *CreateBucketAction {
	return &CreateBucketAction{
		BaseAction:       base.NewBaseAction(transProvider, bucketRepo),
		AwsBucketManager: awsBucketManager,
	}
}

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

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 {
	projectID, err := strconv.ParseUint(bucket.ProjectID, 10, 64)
	if err != nil {
		logger.Warn("parse project id error", zap.Any("project id", bucket.ProjectID), zap.Error(err))
		return err
	}
	createBucketReq := &domain.CreateAWSBucketReq{
		AWSBucketReq: domain.AWSBucketReq{
			Name:         bucket.Name,
			AWSAccountID: *bucket.CloudAccountID,
			Region:       bucket.Regions[0],
		},
		ProjectID: projectID,
		Region:    bucket.Regions[0],
	}
	err = a.AwsBucketManager.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.AwsBucketManager.CheckBucketIfExists(ctx, &domain.AWSBucketReq{
		Name:         bucket.Name,
		AWSAccountID: *bucket.CloudAccountID,
		Region:       bucket.Regions[0],
	})
	if err != nil {
		logger.Warn("check bucket if exists error", 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
}