//nolint:unused
package impl

import (
	"context"
	"strconv"
	"strings"

	"github.com/AlekSi/pointer"
	workflow_entity "github.com/shiningrush/fastflow/pkg/entity"
	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 gkeBucketModel struct {
	baseBucketModel
	dbProvider     db.TransProvider
	bucketRepo     repo.BucketRepo
	bucketConfRepo repo.GCPBucketConfigurationRepo

	bucketManager          bucket.GCPBucketManagerModel
	coreCli                grpcli.CoreGrpcClient
	tidbClusterInfraClient grpcli.TiDBClusterInfraClient
	workflowClient         workflow.Workflow
	statusMatrix           status_matrix.StatusMatrix
}

func NewGKEBucketModel(
	dbProvider db.TransProvider,
	bucketRepo repo.BucketRepo,
	bucketConfRepo repo.GCPBucketConfigurationRepo,
	gcpBucketManager bucket.GCPBucketManagerModel,
	coreCli grpcli.CoreGrpcClient,
	tidbClusterInfraClient grpcli.TiDBClusterInfraClient,
	workflowClient workflow.Workflow,
) bucket.BucketModelV2 {
	return &gkeBucketModel{
		baseBucketModel: baseBucketModel{
			bucketRepo: bucketRepo,
		},

		dbProvider:     dbProvider,
		bucketRepo:     bucketRepo,
		bucketConfRepo: bucketConfRepo,

		bucketManager:          gcpBucketManager,
		coreCli:                coreCli,
		tidbClusterInfraClient: tidbClusterInfraClient,
		workflowClient:         workflowClient,
		statusMatrix:           NewBucketStatusMatrix(),
	}
}

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

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

	bucket, dagIns, err := m.createBucketEntityAndDag(ctx, logger, req)
	if err != nil {
		return nil, err
	}

	logger.Info("create bucket entity and dag successfully",
		zap.Uint64("bucket_id", bucket.ID.Uint64()),
		zap.Uint64("cluster_id", *req.ClusterId),
		zap.String("dag_ins_id", dagIns.ID))

	return &domain.CreateBucketRespV2{
		BucketID:   bucket.ID.Uint64(),
		Status:     bucket.Status,
		WorkflowID: &dagIns.ID,
	}, nil
}

func (m *gkeBucketModel) createBucketEntityAndDag(
	ctx context.Context, logger *zap.Logger, req *domain.CreateBucketReqV2,
) (*entity.Bucket, *workflow_entity.DagInstance, bizerr.BizError) {
	bkt, err := m.genBucketEntity(ctx, logger, req)
	if err != nil {
		return nil, nil, err
	}

	var dagIns *workflow_entity.DagInstance
	txErr := m.dbProvider.Transaction(ctx, func(ctx context.Context) error {
		if err := m.bucketRepo.Save(ctx, bkt); err != nil {
			return err
		}

		bucketConf := m.genBucketConfigurationEntity(bkt)
		if err := m.bucketConfRepo.Save(ctx, bucketConf); err != nil {
			return err
		}

		dagVars, dagTags := m.buildCreateBucketDagVarsAndTags(bkt)
		var dagErr error
		dagIns, dagErr = m.workflowClient.RunDagWithTags(workflow.DagCreateBucketGCP, dagVars, dagTags)
		return dagErr
	})
	if txErr != nil {
		logger.Warn("create bucket entity and dag error",
			zap.String("name", *req.Name), zap.Uint64("cluster_id", *req.ClusterId), zap.Error(err))
		return nil, nil, bizerr_wrapper.NewInternalErr(appconfig.ErrInternalError)
	}
	return bkt, dagIns, nil
}

func (m *gkeBucketModel) genBucketEntity(
	ctx context.Context, logger *zap.Logger, req *domain.CreateBucketReqV2,
) (*entity.Bucket, bizerr.BizError) {
	bkt := &entity.Bucket{
		Name:              *req.Name,
		TenantID:          strconv.FormatUint(*req.TenantId, 10),
		ProjectID:         strconv.FormatUint(*req.ProjectId, 10),
		Regions:           req.Regions,
		Provider:          string(domain.InfraProviderTypeGCP),
		Scope:             *req.Scope,
		Status:            entity.BucketStatusCreating,
		InfraProviderType: string(*req.InfraCloudProvider),
	}
	accountID, err := m.getAccountID(ctx, logger, req)
	if err != nil {
		return nil, err
	}
	bkt.CloudAccountID = accountID
	if req.BusinessPurpose != nil {
		bkt.Extension = map[string]string{"type": string(*req.BusinessPurpose)}
	}
	return bkt, nil
}

func (m *gkeBucketModel) genBucketConfigurationEntity(bucket *entity.Bucket) *entity.GCPBucketConfiguration {
	return &entity.GCPBucketConfiguration{
		BucketID: bucket.ID,
		BucketPolicy: entity.GCPBucketPolicy{
			ServiceAccounts: []string{},
		},
	}
}

func (m *gkeBucketModel) buildCreateBucketDagVarsAndTags(bucket *entity.Bucket) (map[string]string, map[string]string) {
	bucketIDStr := strconv.FormatUint(uint64(bucket.ID), 10)
	vars := map[string]string{
		workflow.KeyBucketID: bucketIDStr,
	}

	tags := map[string]string{
		workflow.BusinessTypeKey:   string(workflow.BusinessTypeBucket),
		workflow.BusinessActionKey: string(workflow.BusinessActionBucketCreate),
		workflow.BusinessID:        bucketIDStr,
	}
	return vars, tags
}

func (m *gkeBucketModel) checkCreateBucketParams(logger *zap.Logger, req *domain.CreateBucketReqV2) bizerr.BizError {
	err := utils.CheckParamInvalidError("infra_provider_type", req.InfraCloudProvider,
		func(param *domain.InfraProviderType) bool {
			return *param == domain.InfraProviderTypeGCP || *param == domain.InfraProviderTypeGKE
		})
	if err != nil {
		logger.Warn("check param invalid error failed", zap.Error(err))
		return err
	}

	err = utils.CheckParamEmptyError("regions", req.Regions)
	if err != nil {
		logger.Warn("check param empty 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
	}

	err = utils.CheckParamEmptyError("project_id", req.ProjectId)
	if err != nil {
		logger.Warn("check param empty error failed", zap.Error(err))
		return err
	}

	return nil
}

func (m *gkeBucketModel) queryBucketIsAlreadyExisted(
	ctx context.Context, logger *zap.Logger, req *domain.CreateBucketReqV2,
) (*entity.Bucket, bizerr.BizError) {
	cloudAccountId, err := m.getAccountID(ctx, logger, req)
	if err != nil {
		return nil, err
	}
	// previously, for gcp dual region bucket, the bucket scope is cluster.
	// now we decide to modify dual region bucket to project scope.
	// when the user enable dual region firstly or reopen dual region switch, we will use project scope bucket directly.
	// so the cluster scope dual region bucket will be disused gradually
	buckets, err := m.bucketRepo.ListByFilter(ctx, repo.BucketFilter{
		Name:           req.Name,
		CloudAccountID: cloudAccountId,
		DeletedTime:    pointer.ToInt64(0),
	})
	if err != nil {
		logger.Warn("get bucket from db error",
			zap.String("name", *req.Name), zap.String("cloud_account_id", *cloudAccountId), zap.Error(err))
		return nil, bizerr_wrapper.NewInternalErr(appconfig.ErrInternalError)
	}

	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 *gkeBucketModel) 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 *gkeBucketModel) DeleteBucket(
	ctx context.Context, req *domain.DeleteBucketReqV2,
) (*domain.DeleteBucketRespV2, bizerr.BizError) {
	ctx = log.WithLogID(ctx)
	logger := log.GetLogger(ctx)
	if err := m.checkDeleteBucketParam(logger, req); err != nil {
		return nil, err
	}

	bucket, err := m.bucketRepo.Load(ctx, *req.BucketID)
	if err != nil {
		if bizerr.ErrCodeRecordNotFound == err.ErrCode() {
			logger.Info("not found bucket to delete, ignore it", zap.Uint64("bucket_id", *req.BucketID))
			return &domain.DeleteBucketRespV2{}, nil
		}
		logger.Warn("load bucket error", zap.Uint64("bucket_id", *req.BucketID), zap.Error(err))
		return nil, bizerr_wrapper.NewInternalErr(appconfig.ErrInternalError)
	}

	if bucket.Status == entity.BucketStatusDeleting || bucket.Status == entity.BucketStatusDeleted {
		logger.Info("bucket is deleting or deleted, ignore it",
			zap.Uint64("bucket_id", *req.BucketID), zap.String("bucket_status", string(bucket.Status)))
		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.String("action", "delete"))
		return nil, bizerr_wrapper.NewInvalidInputError(appconfig.ErrActionParam, domain.DeleteBucket, bucket.Status)
	}

	dagIns, err := m.deleteBucket(ctx, logger, bucket)
	if err != nil {
		return nil, err
	}

	return &domain.DeleteBucketRespV2{
		WorkflowID: &dagIns.ID,
	}, nil
}

func (m *gkeBucketModel) deleteBucket(
	ctx context.Context, logger *zap.Logger, bucket *entity.Bucket,
) (*workflow_entity.DagInstance, bizerr.BizError) {
	nextStatus := m.statusMatrix.CalculateNextStatusForAction(status_matrix.CurrentStatus{
		Status: string(bucket.Status),
	}, string(domain.DeleteBucket))
	bucket.Status = entity.BucketStatus(nextStatus.Status)

	dagVars, dagTags := m.buildDeleteBucketDagVarsAndTags(bucket)
	var dagIns *workflow_entity.DagInstance

	err := m.dbProvider.Transaction(ctx, func(ctx context.Context) error {
		if dbErr := m.bucketRepo.Save(ctx, bucket); dbErr != nil {
			return dbErr
		}

		var dagErr error
		if dagIns, dagErr = m.workflowClient.RunDagWithTags(workflow.DagDeleteBucketGCP, dagVars, dagTags); dagErr != nil {
			return dagErr
		}
		return nil
	})
	if err != nil {
		logger.Warn("update bucket to deleting and create dag error",
			zap.Uint64("bucket_id", bucket.ID.Uint64()), zap.String("bucket_name", bucket.Name), zap.Error(err))
		return nil, bizerr_wrapper.NewInternalErr(appconfig.ErrInternalError)
	}

	return dagIns, nil
}

func (m *gkeBucketModel) buildDeleteBucketDagVarsAndTags(bucket *entity.Bucket) (map[string]string, map[string]string) {
	vars := map[string]string{
		workflow.KeyBucketID: strconv.FormatUint(uint64(bucket.ID), 10),
	}

	tags := map[string]string{
		workflow.BusinessTypeKey:   string(workflow.BusinessTypeBucket),
		workflow.BusinessActionKey: string(workflow.BusinessActionBucketDelete),
		workflow.BusinessID:        strconv.FormatUint(uint64(bucket.ID), 10),
	}
	return vars, tags
}

func (m *gkeBucketModel) checkDeleteBucketParam(logger *zap.Logger, req *domain.DeleteBucketReqV2) bizerr.BizError {
	if err := utils.CheckParamEmptyError("bucket_id", req.BucketID); err != nil {
		logger.Warn("bucket_id is nil", zap.Error(err))
		return err
	}
	return nil
}

func (m *gkeBucketModel) GrantAccessToBucket(
	ctx context.Context, req *domain.GrantAccessToBucketReq,
) (*domain.GrantAccessToBucketResp, bizerr.BizError) {
	ctx = log.WithLogID(ctx)
	logger := log.GetLogger(ctx)
	if err := m.checkGrantAccessParam(logger, req); err != nil {
		return nil, err
	}

	bucket, bucketConf, err := m.loadBucketEntities(ctx, logger, *req.BucketID)
	if err != nil {
		return nil, err
	}

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

	bucketReq := &domain.GCPAddAllowPolicyToBucketReq{
		BucketName:         bucket.Name,
		IAMServiceAccounts: req.CloudProviderIdentities,
	}

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

		bucketResp, err := m.bucketManager.AddAllowPolicyToBucket(ctx, bucketReq)
		if err != nil {
			return err
		}

		bucketConf.BucketPolicy.ServiceAccounts = bucketResp.IAMServiceAccounts
		if err := m.bucketConfRepo.Save(ctx, bucketConf); err != nil {
			return err
		}
		return nil
	})
	if txErr != nil {
		logger.Warn("add allow policy to bucket error",
			zap.String("bucket_name", bucket.Name),
			zap.String("service_accounts", strings.Join(req.CloudProviderIdentities, ",")),
			zap.Uint64("bucket_id", *req.BucketID),
			zap.Error(txErr))
		return nil, bizerr_wrapper.NewInternalErr(appconfig.ErrInternalError)
	}

	return &domain.GrantAccessToBucketResp{}, nil
}

func (m *gkeBucketModel) loadBucketEntities(
	ctx context.Context, logger *zap.Logger, bucketID uint64,
) (*entity.Bucket, *entity.GCPBucketConfiguration, bizerr.BizError) {
	bucket, err := m.bucketRepo.Load(ctx, bucketID)
	if err != nil {
		logger.Warn("load bucket error", zap.Uint64("bucket_id", bucketID), zap.Error(err))
		return nil, nil, bizerr_wrapper.NewInternalErr(appconfig.ErrInternalError)
	}

	bucketConf, err := m.bucketConfRepo.Get(ctx, &repo.GCPBucketConfigurationFilter{
		BucketID: &bucketID,
	})
	if err != nil {
		logger.Warn("get gcp bucket configuration error", zap.Uint64("bucket_id", bucketID), zap.Error(err))
		return nil, nil, bizerr_wrapper.NewInternalErr(appconfig.ErrInternalError)
	}

	return bucket, bucketConf, nil
}

func (m *gkeBucketModel) checkGrantAccessParam(logger *zap.Logger, req *domain.GrantAccessToBucketReq) bizerr.BizError {
	if err := utils.CheckParamEmptyError("bucket_id", req.BucketID); err != nil {
		logger.Warn("bucket_id is nil", zap.Error(err))
		return err
	}
	if err := utils.CheckParamEmptyError("identities", req.CloudProviderIdentities); err != nil {
		logger.Warn("cloud_provider_identities is empty", zap.Error(err))
		return err
	}
	return nil
}

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

	bucket, bucketConf, err := m.loadBucketEntities(ctx, logger, *req.BucketID)
	if err != nil {
		return nil, err
	}

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

	bucketReq := &domain.GCPDeleteAllowPolicyFromBucketReq{
		BucketName:         bucket.Name,
		IAMServiceAccounts: req.CloudProviderIdentities,
	}

	txErr := m.dbProvider.Transaction(ctx, func(ctx context.Context) error {
		bucketResp, err := m.bucketManager.DeleteAllowPolicyFromBucket(ctx, bucketReq)
		if err != nil {
			return err
		}

		bucketConf.BucketPolicy.ServiceAccounts = bucketResp.IAMServiceAccounts
		if err := m.bucketConfRepo.Save(ctx, bucketConf); err != nil {
			return err
		}
		return nil
	})
	if txErr != nil {
		logger.Warn("delete allow policy from bucket error",
			zap.String("bucket_name", bucket.Name),
			zap.String("service_accounts", strings.Join(req.CloudProviderIdentities, ",")),
			zap.Uint64("bucket_id", *req.BucketID),
			zap.Error(txErr))
		return nil, bizerr_wrapper.NewInternalErr(appconfig.ErrInternalError)
	}

	return &domain.RevokeAccessToBucketResp{}, nil
}

func (m *gkeBucketModel) checkRevokeAccessParam(
	logger *zap.Logger, req *domain.RevokeAccessToBucketReq,
) bizerr.BizError {
	if err := utils.CheckParamEmptyError("bucket_id", req.BucketID); err != nil {
		logger.Warn("bucket_id is nil", zap.Error(err))
		return err
	}
	if err := utils.CheckParamEmptyError("identities", req.CloudProviderIdentities); err != nil {
		logger.Warn("cloud_provider_identities is empty", zap.Error(err))
		return err
	}
	return nil
}