package inner

import (
	"context"
	"net/url"
	"strings"
	"time"

	"github.com/AlekSi/pointer"
	"github.com/pingcap/errcode"
	"github.com/pingcap/errors"
	"github.com/tidbcloud/core-svc/idl/pbgen/proto/core"
	"github.com/tidbcloud/dataflow-service/app/config"
	"github.com/tidbcloud/dataflow-service/app/models/backup"
	"github.com/tidbcloud/dataflow-service/app/models/bucket"
	restoreaccess "github.com/tidbcloud/dataflow-service/app/models/classic_restore_access"
	"github.com/tidbcloud/dataflow-service/app/models/domain"
	"github.com/tidbcloud/dataflow-service/app/models/restore"
	"github.com/tidbcloud/dataflow-service/app/repo/entity"
	"github.com/tidbcloud/dataflow-service/idl/pbgen/proto/dataflow"
	"github.com/tidbcloud/dataflow-service/internal/grpc/grpcli"
	bizerr_wrapper "github.com/tidbcloud/dataflow-service/pkg/bizerr-wrapper"
	pkggrpcli "github.com/tidbcloud/dataflow-service/pkg/grpc/grpcli"
	"github.com/tidbcloud/dataflow-service/pkg/log"
	"github.com/tidbcloud/pkgv2/bizerr"
	mgmt "github.com/tidbcloud/tidb-management-service/api/spec/tidb_mgmt_service/v1"
	"go.uber.org/zap"
	"google.golang.org/protobuf/types/known/emptypb"
	"google.golang.org/protobuf/types/known/timestamppb"
)

type BackupInnerController struct {
	dataflow.UnimplementedBackupInternalServiceServer
	snapshotBackupModel backup.SnapshotBackupModel
	restoreRouter       restore.RestoreModel
	bucketModelProxy    bucket.BucketModelV2
	restoreAccessModel  restoreaccess.Model
	mgmtCli             pkggrpcli.MgmtSvcGrpcClient
}

func NewBackupInnerController(
	model backup.SnapshotBackupModel,
	bucketModelProxy bucket.BucketModelV2,
	restoreRouter restore.RestoreModel,
	mgmtCli pkggrpcli.MgmtSvcGrpcClient,
	restoreAccessModel restoreaccess.Model,
) BackupInnerController {
	return BackupInnerController{
		snapshotBackupModel: model,
		bucketModelProxy:    bucketModelProxy,
		restoreRouter:       restoreRouter,
		mgmtCli:             mgmtCli,
		restoreAccessModel:  restoreAccessModel,
	}
}

func (s *BackupInnerController) PrepareClassicRestoreAccess(
	ctx context.Context, req *dataflow.PrepareClassicRestoreAccessReq,
) (*dataflow.PrepareClassicRestoreAccessResp, error) {
	logger := log.GetLogger(ctx)
	resp, err := s.restoreAccessModel.Prepare(ctx, &restoreaccess.PrepareReq{
		BackupID: req.BackupId, BucketID: req.BucketId, RestoreID: req.RestoreId, TrustedRoleARN: req.TrustedRoleArn,
	})
	if err != nil {
		logger.Error("failed to prepare classic restore access", zap.Error(err), zap.Uint64("backup_id", req.BackupId),
			zap.Uint64("bucket_id", req.BucketId), zap.String("restore_id", req.RestoreId))
		return nil, bizerr_wrapper.NewInternalErr("failed to prepare classic restore access: %s", err.Error())
	}
	return &dataflow.PrepareClassicRestoreAccessResp{SourceRoleArn: resp.SourceRoleARN, ExternalId: resp.ExternalID}, nil
}

func (s *BackupInnerController) CleanupClassicRestoreAccess(
	ctx context.Context, req *dataflow.CleanupClassicRestoreAccessReq,
) (*emptypb.Empty, error) {
	logger := log.GetLogger(ctx)
	err := s.restoreAccessModel.Cleanup(ctx, &restoreaccess.CleanupReq{BucketID: req.BucketId, RestoreID: req.RestoreId})
	if err != nil {
		logger.Error("failed to clean up classic restore access", zap.Error(err), zap.Uint64("bucket_id", req.BucketId),
			zap.String("restore_id", req.RestoreId))
		return nil, bizerr_wrapper.NewInternalErr("failed to clean up classic restore access: %s", err.Error())
	}
	return &emptypb.Empty{}, nil
}

func (s *BackupInnerController) DeleteDeletedClusterBackups(
	ctx context.Context, req *dataflow.ClusterIdReq,
) (*emptypb.Empty, error) {
	logger := log.GetLogger(ctx)
	ctx = log.WithTraceID(ctx)
	backups, err := s.snapshotBackupModel.List(ctx,
		&domain.ListBackupReq{
			Filter: &domain.ListBackupFilter{
				ClusterIDs: []uint64{req.ClusterId},
			},
		})
	if err != nil {
		logger.Error("failed to get backups of deleted cluster",
			zap.Uint64("clusterID", req.ClusterId), zap.Error(err))
		return &emptypb.Empty{}, bizerr_wrapper.NewInternalErr(config.ErrMsgInternalError, req.ClusterId)
	}
	for _, backup := range backups.Items {
		_, deleteErr := s.snapshotBackupModel.AsyncDelete(ctx,
			&domain.AsyncDeleteBackupReq{
				BackupID: backup.ID,
			})
		if deleteErr != nil {
			logger.Error("failed to deleted backup",
				zap.Uint64("clusterID", req.ClusterId), zap.Error(deleteErr), zap.Uint64("backupId", backup.ID))
			return &emptypb.Empty{}, bizerr_wrapper.NewInternalErr(config.ErrMsgInternalError, req.ClusterId)
		}
	}

	return &emptypb.Empty{}, nil
}

func (s *BackupInnerController) CreateFinalBackup(
	ctx context.Context, req *dataflow.CreateFinalBackupReq,
) (_ *dataflow.CreateFinalBackupResp, err error) {
	logger := log.GetLogger(ctx)
	backupId, err := s.snapshotBackupModel.CreateFinalBackup(ctx,
		&domain.CreateFinalBackupReq{ClusterID: req.ClusterId})
	if err != nil {
		logger.Error("Fial to create final backup.", zap.Error(err))
		return nil, err
	}
	return &dataflow.CreateFinalBackupResp{
		BackupId: uint64(backupId),
	}, nil
}

func (s *BackupInnerController) DeleteAllManualBackups(
	ctx context.Context, req *dataflow.DeleteAllManualBackupsReq,
) (*emptypb.Empty, error) {
	logger := log.GetLogger(ctx)
	ctx = log.WithTraceID(ctx)
	orgID := entity.ID(req.OrgId)

	deletedOnly := true
	listReq := &mgmt.ListClustersRequest{
		Filter: &mgmt.ClusterFilter{
			TenantIds:   []uint64{uint64(orgID)},
			View:        mgmt.ClusterFilter_CLUSTER_VIEW_BASIC,
			Deleted:     &deletedOnly,
			ClusterType: mgmt.DeployType_DEPLOY_TYPE_DEDICATED.Enum(),
		},
	}
	resp, err := s.mgmtCli.ListClusters(ctx, listReq)
	if err != nil {
		logger.Error("Fail to get clusters when DeleteAllManualBackups", zap.Error(err))
		return nil, errors.Annotate(err, "fetch deleted no cleaned up")
	}
	if len(resp.GetClusters()) == 0 {
		return &emptypb.Empty{}, nil
	}
	for _, cluster := range resp.GetClusters() {
		err = s.markBackupsAsDeleted(ctx, cluster.GetClusterId())
		if err != nil {
			break
		}
	}

	if err != nil {
		logger.Error("failed to delete all manual backups", zap.Uint64("orgID", req.OrgId), zap.Error(err))
		return nil, errcode.NewInternalErr(errors.Errorf(
			"failed to delete all manual backups: %s", err.Error(),
		))
	}

	return &emptypb.Empty{}, nil
}

func (s *BackupInnerController) markBackupsAsDeleted(
	ctx context.Context, clusterId uint64,
) bizerr.BizError {
	logger := log.GetLogger(ctx)
	ctx = log.WithTraceID(ctx)
	backups, err := s.snapshotBackupModel.List(ctx,
		&domain.ListBackupReq{
			Filter: &domain.ListBackupFilter{
				ClusterIDs: []uint64{clusterId},
			},
		})
	if err != nil {
		logger.Error("failed to get backups of deleted cluster",
			zap.Uint64("clusterID", clusterId), zap.Error(err))
		return bizerr_wrapper.NewInternalErr(config.ErrMsgInternalError, clusterId)
	}

	for _, backup := range backups.Items {
		if backup.Type == domain.BackupType_Auto {
			logger.Info("skip auto backup", zap.Any("backup", backup))
			continue
		}
		deleteReq := &domain.AsyncDeleteBackupReq{
			BackupID: uint64(backup.ID),
		}
		resp, deleteErr := s.snapshotBackupModel.AsyncDelete(ctx, deleteReq)
		if deleteErr != nil {
			logger.Error("fail to delete backup",
				zap.Uint64("clusterID", clusterId), zap.Uint64("backupId", backup.ID), zap.Error(deleteErr))
			return deleteErr
		}
		logger.Info("delete backup", zap.Uint64("backupID", backup.ID), zap.Any("resp", resp))
	}
	return nil
}

func (s *BackupInnerController) DeleteAllBackups(
	ctx context.Context, req *dataflow.DeleteAllBackupsReq,
) (*emptypb.Empty, error) {
	logger := log.GetLogger(ctx)
	ctx = log.WithTraceID(ctx)

	if req.OrgId == 0 {
		return &emptypb.Empty{}, bizerr_wrapper.NewInvalidInputError(config.ErrInvalidParam, "OrgId", req.OrgId)
	}
	listReq := &core.ListClustersReq{
		Filter: &core.ClusterFilter{
			TenantIds: []uint64{req.OrgId},
			Statuses:  grpcli.DeletedClusterStatus,
		},
	}
	listResp, err := grpcli.ListClusters(ctx, listReq)
	if err != nil {
		return &emptypb.Empty{}, errors.Annotate(err, "list clusters")
	}

	clusterIDs := []uint64{}
	for _, clusterID := range listResp.OrderedClusterIds {
		_, ok := listResp.Clusters[clusterID]
		if !ok {
			continue
		}
		clusterIDs = append(clusterIDs, clusterID)
	}

	backups, err := s.snapshotBackupModel.List(ctx,
		&domain.ListBackupReq{
			Filter: &domain.ListBackupFilter{
				ClusterIDs: clusterIDs,
			},
		})
	if err != nil {
		logger.Error("failed to get backups of tenant",
			zap.Uint64("tenantID", req.OrgId), zap.Error(err))
		return &emptypb.Empty{}, bizerr_wrapper.NewInternalErr(config.ErrMsgInternalError, req.OrgId)
	}
	for _, backup := range backups.Items {
		_, deleteErr := s.snapshotBackupModel.AsyncDelete(ctx,
			&domain.AsyncDeleteBackupReq{
				BackupID: backup.ID,
			})
		if deleteErr != nil {
			logger.Error("failed to delete backup",
				zap.Error(deleteErr), zap.Uint64("backupId", backup.ID))
			return &emptypb.Empty{}, bizerr_wrapper.NewInternalErr(config.ErrMsgInternalError, req.OrgId)
		}
	}
	return &emptypb.Empty{}, nil
}

func (s *BackupInnerController) GetBackup(
	ctx context.Context, req *dataflow.GetBackupReq,
) (*dataflow.GetBackupResp, error) {
	logger := log.GetLogger(ctx)
	ctx = log.WithTraceID(ctx)

	backup, err := s.snapshotBackupModel.Query(ctx, &domain.QueryBackupReq{
		Filter: &domain.ListBackupFilter{
			BackupIDs: []uint64{req.BackupId},
		},
	})
	if err != nil {
		logger.Error("failed to get backup", zap.Error(err), zap.Any("request", req))
		return nil, bizerr_wrapper.NewInternalErr("failed to get backup: %s", err.Error())
	}
	if backup == nil {
		logger.Error("backup not found", zap.Any("request", req))
		return nil, bizerr_wrapper.NewNotFoundErr("backup not found")
	}
	scheme, key, rawErr := parseBackupStoragePath(backup.BackupEntity.Storage)
	if rawErr != nil {
		logger.Warn("failed to parse backup storage path", zap.Error(rawErr), zap.Any("request", req))
		return nil, bizerr_wrapper.NewInternalErr("invalid backup storage path: %s", backup.BackupEntity.Storage)
	}

	// Notice: only support AWS now
	var cloudProvider domain.CloudProviderType
	switch strings.ToLower(scheme) {
	case "s3":
		cloudProvider = domain.CloudProviderTypeAWS
	default:
		logger.Warn("unsupported cloud provider", zap.String("scheme", scheme))
		return nil, bizerr_wrapper.NewInvalidInputError(config.ErrInvalidParam, "CloudProvider", scheme)
	}

	var storageInfos []*dataflow.BackupStorageInfo
	for _, b := range backup.BackupEntity.BucketSettingSnapshot.Buckets {
		bucket, err := s.bucketModelProxy.GetBucket(ctx, &domain.GetBucketReq{
			BucketID:           b.BucketId,
			InfraCloudProvider: pointer.To(domain.InfraProviderType(cloudProvider)),
		})
		if err != nil {
			logger.Error("failed to get bucket", zap.Error(err), zap.Uint64("bucketId", b.BucketId))
			return nil, bizerr_wrapper.NewInternalErr("failed to get bucket: %s", err.Error())
		}
		storageInfos = append(storageInfos, &dataflow.BackupStorageInfo{
			BucketId:   b.BucketId,
			BucketName: bucket.Bucket.Name,
			Prefix:     key,
			Region:     bucket.Bucket.Regions[0],
			AwsKmsArn:  bucket.Bucket.AwsKmsArn,
		})
	}
	return &dataflow.GetBackupResp{
		BackupId:       req.BackupId,
		TenantId:       uint64(backup.BackupEntity.TenantID),
		ClusterId:      uint64(backup.BackupEntity.ClusterID),
		ClusterVersion: backup.BackupEntity.ClusterVersion,
		DataSizeBytes:  uint64(backup.BackupEntity.BackupSize),
		CloudProvider:  string(cloudProvider),
		StorageInfos:   storageInfos,
	}, nil
}

func parseBackupStoragePath(backupStoragePath string) (scheme, key string, err error) {
	u, err := url.Parse(backupStoragePath)
	if err != nil {
		return "", "", err
	}
	return u.Scheme, strings.TrimPrefix(u.Path, "/"), nil
}

//nolint:dupl
func (s *BackupInnerController) GrantAccessBackupBucket(
	ctx context.Context, req *dataflow.GrantAccessBackupBucketReq,
) (*emptypb.Empty, error) {
	logger := log.GetLogger(ctx)
	ctx = log.WithTraceID(ctx)

	if req.CloudProvider != string(domain.CloudProviderTypeAWS) {
		logger.Warn("unsupported cloud provider", zap.String("cloudProvider", req.CloudProvider))
		return nil, bizerr_wrapper.NewInvalidInputError(config.ErrInvalidParam, "CloudProvider", req.CloudProvider)
	}

	_, err := s.bucketModelProxy.GrantAccessToBucket(ctx, &domain.GrantAccessToBucketReq{
		BucketID:                pointer.To(req.BucketId),
		CloudProviderIdentities: []string{req.CloudProviderIdentity},
		InfraCloudProvider:      pointer.To(domain.InfraProviderType(req.CloudProvider)),
	})
	if err != nil {
		logger.Error("failed to grant access to backup bucket", zap.Error(err), zap.Any("request", req))
		return nil, bizerr_wrapper.NewInternalErr("failed to grant access to backup bucket: %s", err.Error())
	}
	return &emptypb.Empty{}, nil
}

//nolint:dupl
func (s *BackupInnerController) RevokeAccessBackupBucket(
	ctx context.Context, req *dataflow.RevokeAccessBackupBucketReq,
) (*emptypb.Empty, error) {
	logger := log.GetLogger(ctx)
	ctx = log.WithTraceID(ctx)

	if req.CloudProvider != string(domain.CloudProviderTypeAWS) {
		logger.Warn("unsupported cloud provider", zap.String("cloudProvider", req.CloudProvider))
		return nil, bizerr_wrapper.NewInvalidInputError(config.ErrInvalidParam, "CloudProvider", req.CloudProvider)
	}

	_, err := s.bucketModelProxy.RevokeAccessToBucket(ctx, &domain.RevokeAccessToBucketReq{
		BucketID:                pointer.To(req.BucketId),
		CloudProviderIdentities: []string{req.CloudProviderIdentity},
		InfraCloudProvider:      pointer.To(domain.InfraProviderType(req.CloudProvider)),
	})
	if err != nil {
		logger.Error("failed to revoke access to backup bucket", zap.Error(err), zap.Any("request", req))
		return nil, bizerr_wrapper.NewInternalErr("failed to revoke access to backup bucket: %s", err.Error())
	}
	return &emptypb.Empty{}, nil
}

func (s *BackupInnerController) Restore(
	ctx context.Context, req *dataflow.BackupInternalRestoreReq,
) (*dataflow.RestoreResp, error) {
	logger := log.GetLogger(ctx)
	ctx = log.WithTraceID(ctx)
	// Only	support snapshot mode
	if req.RestoreMode != dataflow.RestoreMode_RESTORE_MODE_SNAPSHOT {
		logger.Error("unsupported restore mode", zap.Any("request", req))
		return nil, bizerr_wrapper.NewInvalidInputError(config.ErrInvalidParam, "RestoreMode", req.RestoreMode)
	}

	cluster, err := s.mgmtCli.GetDedicatedClusterBasicByID(ctx, req.TargetClusterId)
	if err != nil {
		logger.Error("failed to get target cluster", zap.Error(err), zap.Uint64("clusterID", req.TargetClusterId))
		return nil, bizerr_wrapper.NewInternalErr("failed to get target cluster: %s", err.Error())
	}

	restoreReq := &domain.CreateRestoreReqV2{
		TenantID:  req.OrgId,
		ProjectID: req.TargetProjectId,
		Mode:      domain.RestoreModeSnapshot,
		BackupID:  req.Id,
		ClusterConfig: domain.ClusterConfig{
			ClusterID: req.TargetClusterId,
			Cloned:    true,
			Name:      cluster.GetClusterName(),
			Region:    cluster.GetRegion(),
		},
		BackupRegion:    req.BackupRegion,
		BRVersion:       req.BrVersion,
		BRImageTag:      req.BrImageTag,
		BRCustomOptions: strings.TrimSpace(req.BrCustomOptions),
	}
	createRestoreResp, err := s.restoreRouter.Create(ctx, restoreReq)
	if err != nil {
		return nil, err
	}
	return &dataflow.RestoreResp{
		Id:        createRestoreResp.ID,
		ClusterId: createRestoreResp.ClusterID,
		OrgId:     req.OrgId,
	}, nil
}

func (s *BackupInnerController) GetRestore(
	ctx context.Context, req *dataflow.GetRestoreReq,
) (*dataflow.GetRestoreResp, error) {
	ctx = log.WithTraceID(ctx)
	qp, err := s.restoreRouter.Query(ctx, &domain.QueryRestoreReqV2{
		Filter: &domain.ListRestoreFilter{
			RestoreIDs: []uint64{req.RestoreId},
		},
	})
	if err != nil {
		return nil, err
	}
	resp := &dataflow.GetRestoreResp{
		Id:              qp.Id,
		CreateTimestamp: timestamppb.New(time.Unix(qp.CreateTime.Unix(), 0)),
		BackupId:        qp.BackupId,
		ClusterId:       qp.RestoreClusterID,
		Status:          domainRestoreStatusToDataflow(qp.Status),
	}
	return resp, nil
}

func domainRestoreStatusToDataflow(
	restore domain.RestoreStatusEnum,
) dataflow.GetRestoreResp_StatusEnum {
	switch restore {
	case domain.RestoreStatusEnumPending:
		return dataflow.GetRestoreResp_PENDING
	case domain.RestoreStatusEnumRunning:
		return dataflow.GetRestoreResp_RUNNING
	case domain.RestoreStatusEnumFailed:
		return dataflow.GetRestoreResp_FAILED
	case domain.RestoreStatusEnumSuccess:
		return dataflow.GetRestoreResp_SUCCESS
	case domain.RestoreStatusEnumTerminated:
		return dataflow.GetRestoreResp_TERMINATED
	default:
		return dataflow.GetRestoreResp_RESTORE_STATUS_UNSPECIFIED
	}
}