package console

import (
	"context"
	"fmt"
	"strconv"
	"strings"

	"github.com/tidbcloud/pkgv2/bizerr"
	"github.com/tidbcloud/pkgv2/clusterproxy"
	"github.com/tidbcloud/pkgv2/import/cloud_storage"
	"github.com/tidbcloud/pkgv2/log"
	"github.com/tidbcloud/pkgv2/tracing"
	"github.com/tidbcloud/serverless-service/idl/pbgen/proto/console"
	"github.com/tidbcloud/serverless-service/internal/domain/entity"
	"github.com/tidbcloud/serverless-service/internal/domain/facade"
	"github.com/tidbcloud/serverless-service/internal/domain/repo"
	"github.com/tidbcloud/serverless-service/internal/domain/service"
	"github.com/tidbcloud/serverless-service/internal/domain/vo"
	"github.com/tidbcloud/serverless-service/internal/errorcode"
	stashing "github.com/tidbcloud/serverless-service/internal/infra/stashing_area"
	"github.com/tidbcloud/serverless-service/internal/service/import_svc"
	"github.com/tidbcloud/serverless-service/internal/utils"
	globalserverlessv1 "github.com/tidbcloud/tidb-management-service/api/spec/global/serverless/v1"
	tidbmgmtservicev1 "github.com/tidbcloud/tidb-management-service/api/spec/tidb_mgmt_service/v1"
	"go.opentelemetry.io/otel"
	"go.opentelemetry.io/otel/attribute"
	"go.opentelemetry.io/otel/trace"
	"go.uber.org/zap"
)

var Tracer = otel.Tracer("import.console")

var (
	ListSupportedOrderByField = []string{"create_time"}
	ListOrderByFieldMap       = map[string]string{
		"create_time": "created_at",
	}
)

type importService struct {
	uploadRepo                 repo.UploadRepo
	stashingAreaProvider       *stashing.StashingAreaProvider
	importHelper               *import_svc.ImportHelperService
	importSvc                  service.ImportService
	accountFacade              facade.AccountFacade
	clusterProxyProviderFacade facade.ClusterProxyProviderFacade
	mgmtSvc                    facade.MgmtSvcFacade
	console.UnimplementedImportServiceServer
}

func NewImportService(
	uploadRepo repo.UploadRepo,
	stashingAreaProvider *stashing.StashingAreaProvider,
	importHelper *import_svc.ImportHelperService,
	importSvc service.ImportService,
	accountFacade facade.AccountFacade,
	clusterProxyProviderFacade facade.ClusterProxyProviderFacade,
	mgmtSvc facade.MgmtSvcFacade,
) console.ImportServiceServer {
	return &importService{
		uploadRepo:                 uploadRepo,
		stashingAreaProvider:       stashingAreaProvider,
		importSvc:                  importSvc,
		importHelper:               importHelper,
		accountFacade:              accountFacade,
		clusterProxyProviderFacade: clusterProxyProviderFacade,
		mgmtSvc:                    mgmtSvc,
	}
}

func (s *importService) ListImports(
	ctx context.Context,
	req *console.ListImportsReq,
) (_ *console.ListImportsResp, err error) {
	ctx, span := Tracer.Start(ctx, "console.import.List")
	defer func() {
		tracing.EndSpan(span, err)
		err = errorcode.PrepareImportErrorMessage(ctx, err)
	}()
	logger := log.WithContextL(ctx).With(zap.String("cluster_id", req.ClusterId))

	pager := utils.NewPagerNumberPager()
	err = pager.Normalize(req)
	if err != nil {
		return nil, err
	}

	// check orderBy
	err = utils.CheckOrderBy(req, ListSupportedOrderByField)
	if err != nil {
		return nil, bizerr.New(
			errorcode.ErrCodeInvalidOrderby,
			"%s",
			errorcode.ExportErrorMessages[errorcode.ErrCodeInvalidOrderby]+" Error: "+err.Error(),
		)
	}
	orderBy := req.OrderBy
	for k, v := range ListOrderByFieldMap {
		orderBy = strings.ReplaceAll(orderBy, k, v)
	}

	span.AddEvent("call ListImports")
	total, imports, bizErr := s.importSvc.ListImports(ctx, &vo.ListImportOpt{
		ClusterID: req.ClusterId,
		Offset:    int64(pager.Offset()),
		PageSize:  int32(pager.PageSize()),
		OrderBy:   orderBy,
	})
	if bizErr != nil {
		return nil, bizErr
	}
	nextPageToken := pager.NextPageToken(uint32(total))
	imps, err := vo.ListImport2Console(ctx, imports, total, nextPageToken)
	if err != nil {
		logger.Error("ListImport2Console failed", zap.Error(err))
		return nil, bizerr.New(errorcode.ErrCodeInternalError, "")
	}
	return imps, nil
}

func (s *importService) CreateImport(
	ctx context.Context,
	req *console.CreateImportReq,
) (resp *console.Import, err error) {
	ctx, span := Tracer.Start(ctx, "console.import.Create")
	defer func() {
		tracing.EndSpan(span, err)
		err = errorcode.PrepareImportErrorMessage(ctx, err)
		go recordCreateImportAuditLog(ctx, req, resp, err)
	}()

	bizErr := validateImportReq(ctx, req)
	if bizErr != nil {
		return nil, bizErr
	}

	switch req.GetSource().GetType() {
	case console.ImportSourceType_LOCAL:
		return s.createImportFromLocalFile(ctx, req, span)
	case console.ImportSourceType_S3, console.ImportSourceType_S3_COMPATIBLE, console.ImportSourceType_OSS:
		return s.createImportFromS3(ctx, req, span)
	case console.ImportSourceType_GCS:
		return s.createImportFromGCS(ctx, req, span)
	case console.ImportSourceType_AZURE_BLOB:
		return s.createImportFromAzBlob(ctx, req, span)
	default:
		return nil, bizerr.New(bizerr.ErrCodeParams, "unrecognized import type")
	}
}

func (s *importService) GetImport(ctx context.Context, req *console.ImportIds) (_ *console.Import, err error) {
	ctx, span := Tracer.Start(ctx, "console.import.Get")
	defer func() {
		tracing.EndSpan(span, err)
		err = errorcode.PrepareImportErrorMessage(ctx, err)
	}()
	logger := log.WithContextL(ctx).With(zap.String("cluster_id", req.ClusterId), zap.String("import_id", req.Id))

	span.AddEvent("call GetImport")
	var i *entity.Import
	idCompat, err := strconv.ParseUint(req.Id, 10, 64)
	var bizErr bizerr.BizError
	if err == nil {
		i, bizErr = s.importSvc.GetImportByIDCompat(ctx, req.ClusterId, idCompat)
		if bizErr != nil {
			logger.Error("failed to get import", zap.Error(bizErr))
			return nil, bizErr
		}
	} else {
		i, bizErr = s.importSvc.GetImport(ctx, req.ClusterId, req.Id)
		if bizErr != nil {
			logger.Error("failed to get import", zap.Error(bizErr))
			return nil, bizErr
		}
	}

	imp, err := vo.GetImport2Console(ctx, i)
	if err != nil {
		logger.Error("GetImport2Console failed", zap.Error(err))
		return nil, bizerr.New(errorcode.ErrCodeInternalError, "")
	}
	return imp, nil
}

func (s *importService) CancelImport(ctx context.Context, req *console.ImportIds) (resp *console.Import, err error) {
	ctx, span := Tracer.Start(ctx, "console.import.Cancel")
	defer func() {
		tracing.EndSpan(span, err)
		err = errorcode.PrepareImportErrorMessage(ctx, err)
		go recordCancelImportAuditLog(ctx, req, err)
	}()
	logger := log.WithContextL(ctx).With(zap.String("cluster_id", req.ClusterId), zap.String("import_id", req.Id))

	var id string
	idCompat, err := strconv.ParseUint(req.Id, 10, 64)
	var bizErr bizerr.BizError
	if err == nil {
		i, bizErr := s.importSvc.GetImportByIDCompat(ctx, req.ClusterId, idCompat)
		if bizErr != nil {
			return nil, bizErr
		}
		id = i.ID
	} else {
		id = req.Id
	}

	span.AddEvent("call CancelImport")
	bizErr = s.importSvc.CancelImport(ctx, req.ClusterId, id)
	if bizErr != nil {
		logger.Error("failed to cancel import", zap.Error(bizErr))
		return nil, bizErr
	}

	span.AddEvent("call GetImport")
	i, bizErr := s.importSvc.GetImport(ctx, req.ClusterId, req.Id)
	if bizErr != nil {
		logger.Error("failed to get import", zap.Error(bizErr))
		return nil, bizErr
	}

	imp, err := vo.GetImport2Console(ctx, i)
	if err != nil {
		logger.Error("GetImport2Console failed", zap.Error(err))
		return nil, bizerr.New(errorcode.ErrCodeInternalError, "")
	}
	return imp, nil
}

func (s *importService) ValidateImport(
	ctx context.Context,
	req *console.ValidateImportReq,
) (_ *console.ValidateImportResp, err error) {
	ctx, span := Tracer.Start(ctx, "console.import.ValidateImport")
	defer func() {
		tracing.EndSpan(span, err)
		err = errorcode.PrepareImportErrorMessage(ctx, err)
	}()
	logger := log.WithContextL(ctx).With(zap.String("cluster_id", req.ClusterId))

	switch req.GetValidationType() {
	case console.ImportValidationType_SOURCE_ACCESS_CHECK:
		cluster, bizErr := s.validateCluster(ctx, req.ClusterId, log.WithContextL(ctx))
		if bizErr != nil {
			return nil, bizErr
		}

		bizErr = validateSourceCheckReq(req)
		if bizErr != nil {
			return nil, bizErr
		}

		getExternalIdFunc := func() (string, bizerr.BizError) {
			profile, bizErr := s.importSvc.GetDataIngestionProfile(ctx, cluster)
			if bizErr != nil {
				logger.Error("failed to get data ingestion profile", zap.Error(bizErr))
				return "", bizErr
			}
			return profile.ExternalId, nil
		}
		cs, bizErr := buildCloudStorageFromImportSource(req.GetSource(), getExternalIdFunc)
		if bizErr != nil {
			logger.Error("failed to build cloud storage from import source", zap.Error(bizErr))
			return nil, bizErr
		}

		bizErr = s.mgmtSvc.CheckImportPermission(ctx, &globalserverlessv1.CheckImportPermissionRequest{
			CloudStorage: cs,
			ClusterId:    req.ClusterId,
		})
		if bizErr != nil {
			return nil, bizErr
		}
	}

	return &console.ValidateImportResp{}, nil
}

func (s *importService) getClusterProxy(
	ctx context.Context,
	span trace.Span,
	logger *zap.Logger,
	clusterID string,
) (facade.ClusterProxyFacade, bizerr.BizError) {
	user, err := utils.GetUserFromContext(ctx)
	if err != nil {
		logger.Error("user not found", zap.Error(err))
		return nil, bizerr.New(bizerr.ErrCodeUnknown, "user not found")
	}
	logger = log.WithContextL(ctx).With(zap.Uint64("user_id", user.ID))
	span.SetAttributes(
		attribute.String("cluster_id", clusterID),
		attribute.String("user_id", fmt.Sprint(user.ID)),
	)

	cID, err := strconv.ParseUint(clusterID, 10, 64)
	if err != nil {
		logger.Error("invalid cluster id", zap.Error(err))
		return nil, bizerr.New(bizerr.ErrCodeParams, "invalid cluster id")
	}

	span.AddEvent("get JWT")
	jwt, err := s.accountFacade.GetJwtByUser(ctx, user.Email, user.ID, cID)
	if err != nil {
		logger.Error("failed to get jwt", zap.Error(err))
		return nil, bizerr.New(errorcode.ErrCodeInternalError, "")
	}

	span.AddEvent("get cluster")
	cluster, err := s.mgmtSvc.GetCluster(ctx, cID, tidbmgmtservicev1.ClusterFilter_CLUSTER_VIEW_BASIC)
	if err != nil {
		logger.Error("failed to get cluster", zap.Error(err))
		return nil, bizerr.New(errorcode.ErrCodeInternalError, "")
	}
	if cluster.GetServerlessCluster().
		GetUserStatus() !=
		globalserverlessv1.UserClusterStatus_USER_CLUSTER_STATUS_NORMAL {
		return nil, bizerr.New(errorcode.ErrCodeClusterNotRunning, "cluster is not running")
	}

	logger = logger.With(zap.String("region", cluster.Region))
	span.AddEvent("get cluster proxy")
	operator := &clusterproxy.ClusterOperator{
		Username:   jwt.UserName,
		AuthMethod: clusterproxy.AuthMethodToken,
		Credential: jwt.Jwt,
	}
	proxy, err := s.clusterProxyProviderFacade.GetClusterProxy(ctx, cluster, operator)
	if err != nil {
		logger.Error("failed to get cluster proxy", zap.Error(err))
		return nil, bizerr.New(errorcode.ErrCodeSvcClusterProxy, "failed to get cluster proxy")
	}
	return proxy, nil
}

func (s *importService) getS3bucketRegion(
	ctx context.Context,
	source *console.ImportSource,
	cluster *tidbmgmtservicev1.Cluster,
) (string, bizerr.BizError) {
	var cs cloud_storage.CloudStorage
	var bizErr bizerr.BizError
	switch source.GetType() {
	case console.ImportSourceType_S3_COMPATIBLE:
		accessKey := source.GetS3Compatible().GetAccessKey()
		cs, bizErr = cloud_storage.NewS3Compatible(ctx, source.GetS3Compatible().GetUri(),
			cloud_storage.WithS3CompatibleAccessKey(accessKey.GetId(), accessKey.GetSecret()),
			cloud_storage.WithS3CompatibleEndpoint(source.GetS3Compatible().GetEndpoint()),
		)
		if bizErr != nil {
			return "", bizErr
		}
		return cs.(*cloud_storage.S3).Region, nil
	case console.ImportSourceType_OSS:
		oss := source.GetOss()
		options := []cloud_storage.OSSOption{cloud_storage.WithOSSAccessKey(
			oss.GetAccessKey().GetId(), oss.GetAccessKey().GetSecret())}
		var region string
		region, bizErr = cloud_storage.GetOssBucketRegion(ctx, oss.GetUri(), options...)
		if bizErr != nil {
			return "", bizErr
		}
		return region, nil
	case console.ImportSourceType_S3:
		switch source.GetS3().GetAuthType() {
		case console.ImportS3AuthType_ROLE_ARN:
			var profile *entity.DataIngestionProfile
			profile, bizErr = s.importSvc.GetDataIngestionProfile(ctx, cluster)
			if bizErr != nil {
				return "", bizErr
			}
			cs, bizErr = cloud_storage.NewS3(ctx, source.GetS3().GetUri(),
				cloud_storage.WithAssumeRole(source.GetS3().GetRoleArn(), profile.ExternalId),
			)
			if bizErr != nil {
				return "", bizErr
			}
		case console.ImportS3AuthType_ACCESS_KEY:
			accessKey := source.GetS3().GetAccessKey()
			cs, bizErr = cloud_storage.NewS3(ctx, source.GetS3().GetUri(),
				cloud_storage.WithAccessKey(accessKey.GetId(), accessKey.GetSecret()),
			)
			if bizErr != nil {
				return "", bizErr
			}
		}
		return cs.(*cloud_storage.S3).Region, nil
	}

	return "", bizerr.New(bizerr.ErrCodeParams, "unsupported import source type for S3 bucket: %s", source.GetType())
}