package impl

import (
	"context"
	"os"
	"strconv"
	"time"

	"github.com/AlekSi/pointer"
	workflowEntity "github.com/shiningrush/fastflow/pkg/entity"
	appconfig "github.com/tidbcloud/dataflow-service/app/config"
	"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"
	bizerr_wrapper "github.com/tidbcloud/dataflow-service/pkg/bizerr-wrapper"
	encryptor_wrapper "github.com/tidbcloud/dataflow-service/pkg/encryptor-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"
	mgmt "github.com/tidbcloud/tidb-management-service/api/spec/tidb_mgmt_service/v1"
	"go.uber.org/zap"
)

const (
	expiration = 10 * 365 * 24 * time.Hour // 10 year
)

type aksBucketModel struct {
	baseBucketModel
	dbProvider db.TransProvider
	bucketRepo repo.BucketRepo

	encryptor      encryptor_wrapper.Encryptor
	bucketManager  bucket.AzureBucketManagerModel
	workflowClient workflow.Workflow

	mgmtCli grpcli.MgmtSvcGrpcClient
}

func NewAKSBucketModel(
	dbProvider db.TransProvider,
	bucketRepo repo.BucketRepo,
	encryptor encryptor_wrapper.Encryptor,
	bucketManager bucket.AzureBucketManagerModel,
	workflowClient workflow.Workflow,
	mgmtCli grpcli.MgmtSvcGrpcClient,
) bucket.BucketModelV2 {
	return &aksBucketModel{
		baseBucketModel: baseBucketModel{
			bucketRepo: bucketRepo,
		},

		dbProvider: dbProvider,
		bucketRepo: bucketRepo,

		encryptor:      encryptor,
		bucketManager:  bucketManager,
		workflowClient: workflowClient,
		mgmtCli:        mgmtCli,
	}
}

//nolint:funlen
func (m *aksBucketModel) CreateBucket(
	ctx context.Context, req *domain.CreateBucketReqV2,
) (*domain.CreateBucketRespV2, bizerr.BizError) {
	ctx = log.WithLogID(ctx)
	logger := log.GetLogger(ctx)

	cloudAccount, err := m.getCloudAccount()
	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
	}

	bkt := &entity.Bucket{
		Name:              *req.Name,
		TenantID:          strconv.FormatUint(*req.TenantId, 10),
		ProjectID:         strconv.FormatUint(*req.ProjectId, 10),
		CloudAccountID:    cloudAccount,
		Regions:           req.Regions,
		Provider:          string(domain.InfraProviderTypeAzure),
		Scope:             *req.Scope,
		Status:            entity.BucketStatusCreating,
		InfraProviderType: string(*req.InfraCloudProvider),
	}
	if req.BusinessPurpose != nil {
		bkt.Extension = map[string]string{"type": string(*req.BusinessPurpose)}
	}
	if err := m.setCmekIfNeed(ctx, bkt); err != nil {
		return nil, err
	}

	var dagInstance *workflowEntity.DagInstance
	dbErr := m.dbProvider.Transaction(ctx, func(ctx context.Context) error {
		if err := m.bucketRepo.Save(ctx, bkt); err != nil {
			return err
		}
		bucketIDStr := strconv.FormatUint(uint64(bkt.ID), 10)
		dagVars := m.buildCreateBucketDagVars(bkt)
		tags := map[string]string{
			workflow.BusinessTypeKey:   string(workflow.BusinessTypeBucket),
			workflow.BusinessActionKey: string(workflow.BusinessActionBucketCreate),
			workflow.BusinessID:        bucketIDStr,
		}
		var dagErr error
		dagInstance, dagErr = m.workflowClient.RunDagWithTags(workflow.DagCreateBucketAzure, dagVars, tags)
		if dagErr != nil {
			return dagErr
		}
		logger.Info("create bucket dag started", zap.Any("request", req), zap.Any("dag id", dagInstance.ID))
		return nil
	})
	if dbErr != nil {
		logger.Error("create bucket failed", zap.Any("request", req), zap.Error(dbErr))
		return nil, bizerr_wrapper.NewInternalErr(appconfig.ErrInternalError)
	}

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

func (m *aksBucketModel) setCmekIfNeed(ctx context.Context, bkt *entity.Bucket) bizerr.BizError {
	logger := log.GetLogger(ctx).With(
		zap.Uint64("bucketID", uint64(bkt.ID)),
		zap.String("projectID", bkt.ProjectID),
		zap.String("cloudProvider", bkt.Provider),
		zap.Strings("regions", bkt.Regions),
	)

	req := &mgmt.ListCustomerManagedEncryptionKeysRequest{
		ProjectId:     bkt.ProjectID,
		CloudProvider: bkt.Provider,
		RegionId:      bkt.Regions[0],
	}
	resp, err := m.mgmtCli.ListCustomerManagedEncryptionKey(ctx, req)
	if err != nil {
		return bizerr.WrapGrpcError(err, "call mgmt svc to list cmek failed")
	}
	if len(resp.Cmeks) == 0 {
		logger.Sugar().Infof("no cmek found, skip setting cmek")
		return nil
	}
	if len(resp.Cmeks) > 1 {
		return bizerr_wrapper.NewInternalErr("multiple cmeks found, count: %d", len(resp.Cmeks))
	}

	cmek := resp.Cmeks[0]
	az := cmek.GetAzureKeyVault()
	if az == nil || az.AppRegistration == nil {
		return bizerr_wrapper.NewInternalErr("cmek is not azure key vault type: id=%s,name=%s",
			cmek.CustomerManagedEncryptionKeyId, cmek.Name)
	}
	bkt.AzureCmekParams = &entity.AzureCmekParams{
		ApplicationID:             az.AppRegistration.AppRegistrationId,
		ManagedIdentityResourceID: az.AppRegistration.AppRegistrationUmi,
		KeyVaultName:              az.KeyVaultName,
		KeyName:                   az.KeyVaultKey,
	}
	logger.Info("set cmek to bucket", zap.Any("cmek", az))
	return nil
}

func (m *aksBucketModel) buildCreateBucketDagVars(
	bucket *entity.Bucket,
) map[string]string {
	runVar := map[string]string{}
	runVar[workflow.KeyBucketID] = strconv.FormatUint(uint64(bucket.ID), 10)
	if bucket.AzureCmekParams != nil {
		runVar[workflow.KeyBucketEncryptionSwitch] = workflow.ValueBucketEncryptionSwitchTrue
	}
	return runVar
}

func (m *aksBucketModel) 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)
	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
	}

	var dagInstance *workflowEntity.DagInstance
	dbErr := m.dbProvider.Transaction(ctx, func(ctx context.Context) error {
		dagVars := 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),
		}
		var dagErr error
		dagInstance, dagErr = m.workflowClient.RunDagWithTags(workflow.DagDeleteBucketAzure, dagVars, tags)
		if err != nil {
			return dagErr
		}
		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 &domain.DeleteBucketRespV2{
		WorkflowID: &dagInstance.ID,
	}, nil
}

// GrantAccessToBucket generates a SAS token and save in db
// Backup job will retrieve this token and pass it to regional api
func (m *aksBucketModel) GrantAccessToBucket(
	ctx context.Context, req *domain.GrantAccessToBucketReq,
) (*domain.GrantAccessToBucketResp, bizerr.BizError) {
	ctx = log.WithLogID(ctx)
	logger := log.GetLogger(ctx)

	err := m.dbProvider.Transaction(ctx, func(ctx context.Context) error {
		bkt, err := m.bucketRepo.Load(ctx, *req.BucketID)
		if err != nil {
			return err
		}
		if _, err := bkt.GetSASToken(m.encryptor); err == nil {
			logger.Info("SAS token already generated")
			return nil
		}

		// Generate SAS token
		tokenReq := &domain.GenerateAzureSASTokenReq{
			Name:           bkt.Name,
			SubscriptionID: *bkt.CloudAccountID,
			Expiration:     expiration,
		}
		token, err := m.bucketManager.GenerateSASToken(ctx, tokenReq)
		if err != nil {
			return nil
		}
		if err := bkt.SetSASToken(m.encryptor, token); err != nil {
			logger.Warn("set SAS token error", zap.Error(err))
			return err
		}
		if err := m.bucketRepo.Save(ctx, bkt); err != nil {
			logger.Warn("save SAS token error", zap.Error(err))
			return err
		}
		return nil
	})
	if err != nil {
		logger.Error("generate SAS token error", zap.Any("bucketID", req.BucketID), zap.Error(err))
		return nil, bizerr_wrapper.NewInternalErr(appconfig.ErrInternalError)
	}

	return &domain.GrantAccessToBucketResp{}, nil
}

// We dont't need to revoke access to azure bucket in normal use cases
// Only when there is security incident, in this case we revoke access manually
func (m *aksBucketModel) RevokeAccessToBucket(
	ctx context.Context, req *domain.RevokeAccessToBucketReq,
) (*domain.RevokeAccessToBucketResp, bizerr.BizError) {
	return &domain.RevokeAccessToBucketResp{}, nil
}

func (m *aksBucketModel) queryBucketIsAlreadyExisted(
	ctx context.Context, logger *zap.Logger, req *domain.CreateBucketReqV2,
) (*entity.Bucket, bizerr.BizError) {
	accountID, err := m.getCloudAccount()
	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 *aksBucketModel) getCloudAccount() (*string, bizerr.BizError) {
	// Use environment variable for test
	// TODO: We need to get the cloud account of a cluster from tidb-mgmt in 3-layer arch
	// Once tidb-mgmt provides such API
	account := os.Getenv("AZURE_DATAPLANE_SUBSCRIPTION")
	return &account, nil
}