package impl

import (
	"context"

	"github.com/samber/lo"
	"github.com/tidbcloud/account/idl/pbgen/proto/account"
	appconfig "github.com/tidbcloud/dataflow-service/app/config"
	"github.com/tidbcloud/dataflow-service/app/models/domain"
	restore_model "github.com/tidbcloud/dataflow-service/app/models/restore"
	backup_repo "github.com/tidbcloud/dataflow-service/app/repo/backup"
	restore_repo "github.com/tidbcloud/dataflow-service/app/repo/restore"
	"github.com/tidbcloud/dataflow-service/app/utils"
	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"
	globalv1 "github.com/tidbcloud/tidb-management-service/api/spec/global/dedicated/v1"
	mgmt "github.com/tidbcloud/tidb-management-service/api/spec/tidb_mgmt_service/v1"
	"go.uber.org/zap"
)

type restoreModelRouter struct {
	baseRestoreModel

	restoreModelMap      map[domain.RestoreMode]restore_model.RestoreModel
	backupRepo           backup_repo.BackupRepo
	accountSvcGrpcClient grpcli.AccountSvcGrpcClient
	configCenterCli      grpcli.ConfigCenterGrpcClient
}

func NewRestoreModelRouter(
	dbProvider db.TransProvider,
	restoreRepo restore_repo.RestoreRepo,
	backupRepo backup_repo.BackupRepo,
	snapshotRestoreModel restore_model.RestoreModel,
	pitRestoreModel restore_model.RestoreModel,
	coreCli grpcli.CoreGrpcClient,
	mgmtSvcCli grpcli.MgmtSvcGrpcClient,
	dedicatedGlobalClient grpcli.DedicatedGlobalGrpcClient,
	accountSvcGrpcClient grpcli.AccountSvcGrpcClient,
	configCenterCli grpcli.ConfigCenterGrpcClient,
) restore_model.RestoreModel {
	proxy := &restoreModelRouter{
		baseRestoreModel: baseRestoreModel{
			dbProvider:            dbProvider,
			restoreRepo:           restoreRepo,
			coreCli:               coreCli,
			mgmtSvcCli:            mgmtSvcCli,
			dedicatedGlobalClient: dedicatedGlobalClient,
			configCenterCli:       configCenterCli,
		},
		backupRepo:           backupRepo,
		accountSvcGrpcClient: accountSvcGrpcClient,
	}
	proxy.restoreModelMap = make(map[domain.RestoreMode]restore_model.RestoreModel)
	proxy.restoreModelMap[domain.RestoreModeSnapshot] = snapshotRestoreModel
	proxy.restoreModelMap[domain.RestoreModePITR] = pitRestoreModel
	return proxy
}

func (r *restoreModelRouter) getModel(
	ctx context.Context, restoreModeType domain.RestoreMode,
) (restore_model.RestoreModel, bizerr.BizError) {
	logger := log.GetLogger(ctx)
	err := utils.CheckParamEmptyError("restoreModeType", restoreModeType)
	if err != nil {
		return nil, err
	}
	model, ok := r.restoreModelMap[restoreModeType]
	if !ok {
		logger.Warn("get restore model failed")
		return nil, bizerr_wrapper.NewInternalErr(appconfig.ErrInternalError)
	}
	return model, nil
}

func (r *restoreModelRouter) Create(
	ctx context.Context, req *domain.CreateRestoreReqV2,
) (*domain.CreateRestoreRespV2, bizerr.BizError) {
	ctx = log.WithTraceID(ctx)
	model, err := r.getModel(ctx, req.Mode)
	if err != nil {
		return nil, err
	}
	if err := r.validate(ctx, req); err != nil {
		return nil, err
	}
	return model.Create(ctx, req)
}

//nolint:funlen
func (r *restoreModelRouter) validate(ctx context.Context, req *domain.CreateRestoreReqV2) bizerr.BizError {
	logger := log.GetLogger(ctx)
	var clusterID uint64
	switch req.Mode {
	case domain.RestoreModeSnapshot:
		// The backupID is required when the restore mode is snapshot
		if req.BackupID == 0 {
			logger.Warn("backupID is 0 when restore mode is snapshot")
			return bizerr.New(bizerr.ErrCodeParams, "backupID is 0 when restore mode is snapshot")
		}

		// Get clusterID from backup
		backup, err := r.backupRepo.Load(ctx, req.BackupID)
		if err != nil {
			return err
		}

		// ensure no cross-tenant restore
		if req.TenantID != backup.TenantID.Uint64() {
			logger.Info("cross-tenant restore is not allowed",
				zap.Uint64("reqTenantId", req.TenantID), zap.Uint64("backupTenantId", backup.TenantID.Uint64()),
				zap.Uint64("reqProjectId", req.ProjectID), zap.Uint64("backupProjectId", backup.ProjectID.Uint64()))
			return bizerr.New(bizerr.ErrCodeParams, "cross-tenant restore is not allowed")
		}

		clusterID = backup.ClusterID.Uint64() // use backup's clusterID
	case domain.RestoreModePITR:
		// The clusterID is required when the restore mode is pitr
		if req.ClusterID == 0 {
			logger.Warn("clusterID is 0 when restore mode is pitr")
			return bizerr.New(bizerr.ErrCodeParams, "clusterID is 0 when restore mode is pitr")
		}
		clusterID = req.ClusterID
	default:
		logger.Warn("invalid restore mode", zap.String("mode", string(req.Mode)))
		return bizerr.New(bizerr.ErrCodeParams, "invalid restore mode")
	}

	// ensure no cross-tenant restore
	projectResp, bizErr := r.accountSvcGrpcClient.GetProjectByID(ctx, &account.GetProjectByIDRequest{
		Id: req.ProjectID,
	})
	if bizErr != nil {
		logger.Error("fail to get project", zap.Error(bizErr))
		return bizErr
	}
	if projectResp.Project.TenantId != req.TenantID {
		logger.Info("cross-tenant restore is not allowed",
			zap.Uint64("reqTenantId", req.TenantID), zap.Uint64("projectTenantId", projectResp.Project.TenantId),
			zap.Uint64("reqProjectId", req.ProjectID))
		return bizerr.New(bizerr.ErrCodeParams, "cross-tenant restore is not allowed")
	}

	// get cluster for get cloudProviderName
	cluster, err := r.mgmtSvcCli.GetAnyDedicatedClusterBasicByID(ctx, clusterID)
	if err != nil {
		return err
	}
	// ensure CIDR is set for the restored cluster region if creating a new cluster
	if req.ClusterConfig.ClusterID == 0 {
		if err := r.ensureCIDRSet(ctx, cluster, req); err != nil {
			return err
		}
	}

	return nil
}

func (r *restoreModelRouter) Cancel(
	ctx context.Context, req *domain.CancelRestoreReqV2,
) (*domain.CancelRestoreRespV2, bizerr.BizError) {
	ctx = log.WithTraceID(ctx)
	logger := log.GetLogger(ctx)
	restore, err := r.restoreRepo.Load(ctx, req.RestoreID)
	// Reentrant processing
	if err != nil {
		if bizerr.ErrCodeRecordNotFound == err.ErrCode() {
			return &domain.CancelRestoreRespV2{}, nil
		}
		logger.Error("load restore failed", zap.Any("request", req), zap.Error(err))
		return nil, bizerr_wrapper.NewInternalErr(appconfig.ErrInternalError)
	}
	model, err := r.getModel(ctx, domain.RestoreMode(restore.Mode))
	if err != nil {
		return nil, err
	}
	return model.Cancel(ctx, req)
}

func (r *restoreModelRouter) ensureCIDRSet(ctx context.Context, cluster *mgmt.Cluster, req *domain.CreateRestoreReqV2) bizerr.BizError {
	// Check whether the CIDR is set if the target cluster ID is not specified
	resp, err := r.dedicatedGlobalClient.QueryCidrs(
		ctx, &globalv1.QueryCidrsRequest{
			Filter: &globalv1.QueryCidrsRequest_QueryCidrsFilter{
				ProjectIds:        []uint64{req.ProjectID},
				CloudProviderName: cluster.CloudProvider,
			},
			GlobalMetaOnly: lo.ToPtr(true), // avoid querying regional control-plane for regional failure tolerance
		},
	)
	if err != nil {
		return err
	}
	found := false
	for _, each := range resp.Items {
		if each.RegionName == req.ClusterConfig.Region {
			found = true
			break
		}
	}

	if !found {
		return bizerr.New(bizerr.ErrCodeRecordNotFound, "Please set CIDR for this region (%s) in project(%d) first", req.ClusterConfig.Region, req.ProjectID)
	}

	return nil
}