package console

import (
	"context"
	"strconv"

	"github.com/tidbcloud/pkgv2/bizerr"
	"github.com/tidbcloud/pkgv2/log"
	"github.com/tidbcloud/serverless-service/idl/pbgen/proto/console"
	"github.com/tidbcloud/serverless-service/internal/config"
	"github.com/tidbcloud/serverless-service/internal/domain/entity"
	"github.com/tidbcloud/serverless-service/internal/domain/vo"
	"github.com/tidbcloud/serverless-service/internal/errorcode"
	"github.com/tidbcloud/serverless-service/internal/service/import_svc"
	"go.opentelemetry.io/otel/trace"
	"go.uber.org/zap"
)

func (s *importService) createImportFromS3(
	ctx context.Context,
	req *console.CreateImportReq,
	span trace.Span,
) (*console.Import, bizerr.BizError) {
	logger := log.WithContextL(ctx).With(zap.String("cluster_id", req.ClusterId))

	cluster, bizErr := s.validateCluster(ctx, req.ClusterId, logger)
	if bizErr != nil {
		logger.Error("validate cluster failed", zap.Error(bizErr))
		return nil, bizErr
	}
	if cluster.CloudProvider != config.ProviderAWS &&
		req.GetSource().GetS3().GetAuthType() == console.ImportS3AuthType_ROLE_ARN {
		return nil, bizerr.New(bizerr.ErrCodeParams, "RoleArn is only supported for AWS Clusters")
	}

	region, bizErr := s.getS3bucketRegion(ctx, req.GetSource(), cluster)
	if bizErr != nil {
		return nil, bizErr
	}
	imp, bizErr := vo.BuildImportWithS3UriConsole(ctx, req, cluster, region)
	if bizErr != nil {
		logger.Error("build import error", zap.Error(bizErr))
		return nil, bizErr
	}

	span.AddEvent("call CreateImport")
	resp, bizErr := s.create(ctx, imp, nil, nil)
	if bizErr != nil {
		logger.Error("create import failed", zap.Error(bizErr))
		return nil, bizErr
	}

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

func (s *importService) createImportFromGCS(
	ctx context.Context,
	req *console.CreateImportReq,
	span trace.Span,
) (*console.Import, bizerr.BizError) {
	logger := log.WithContextL(ctx).With(zap.String("cluster_id", req.ClusterId))

	cluster, bizErr := s.validateCluster(ctx, req.ClusterId, logger)
	if bizErr != nil {
		logger.Error("validate cluster failed", zap.Error(bizErr))
		return nil, bizErr
	}

	imp, bizErr := vo.BuildConsoleGCSImport(ctx, req, cluster)
	if bizErr != nil {
		logger.Error("build import error", zap.Error(bizErr))
		return nil, bizErr
	}

	span.AddEvent("call CreateImport")
	resp, bizErr := s.create(ctx, imp, nil, nil)
	if bizErr != nil {
		logger.Error("create import failed", zap.Error(bizErr))
		return nil, bizErr
	}

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

func (s *importService) createImportFromAzBlob(
	ctx context.Context,
	req *console.CreateImportReq,
	span trace.Span,
) (*console.Import, bizerr.BizError) {
	logger := log.WithContextL(ctx).With(zap.String("cluster_id", req.ClusterId))

	cluster, bizErr := s.validateCluster(ctx, req.ClusterId, logger)
	if bizErr != nil {
		logger.Error("validate cluster failed", zap.Error(bizErr))
		return nil, bizErr
	}

	imp, bizErr := vo.BuildConsoleAzureImport(ctx, req, cluster)
	if bizErr != nil {
		logger.Error("build import error", zap.Error(bizErr))
		return nil, bizErr
	}

	span.AddEvent("call CreateImport")
	resp, bizErr := s.create(ctx, imp, nil, nil)
	if bizErr != nil {
		logger.Error("create import failed", zap.Error(bizErr))
		return nil, bizErr
	}

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

func (s *importService) createImportFromLocalFile(
	ctx context.Context,
	req *console.CreateImportReq,
	span trace.Span,
) (_ *console.Import, bizErr bizerr.BizError) {
	logger := log.WithContextL(ctx).With(zap.String("cluster_id", req.ClusterId))

	cluster, bizErr := s.validateCluster(ctx, req.ClusterId, logger)
	if bizErr != nil {
		logger.Error("validate cluster failed", zap.Error(bizErr))
		return nil, bizErr
	}

	span.AddEvent("call GetUploadByID")
	idCompat, err := strconv.ParseUint(req.GetSource().GetLocal().GetUploadId(), 10, 64)
	var upload *entity.Upload
	if err == nil {
		upload, bizErr = s.uploadRepo.GetUploadByIDCompat(ctx, idCompat)
		if bizErr != nil {
			logger.Error("GetUploadByID failed", zap.Error(bizErr))
			return nil, bizErr
		}
	} else {
		upload, bizErr = s.uploadRepo.GetUploadByID(ctx, req.GetSource().GetLocal().GetUploadId())
		if bizErr != nil {
			logger.Error("GetUploadByID failed", zap.Error(bizErr))
			return nil, bizErr
		}
	}
	bizErr = import_svc.ValidateUpload(ctx, upload, cluster.ClusterId)
	if bizErr != nil {
		logger.Error("validate upload failed", zap.Error(bizErr))
		return nil, bizErr
	}

	area, err := s.stashingAreaProvider.GetStashingArea(cluster.CloudProvider, cluster.Region)
	if err != nil {
		logger.Error("GetStashingArea failed", zap.Error(err))
		return nil, bizerr.New(errorcode.ErrCodeInternalError, errorcode.InternalErrorMessage)
	}
	imp, bizErr := vo.BuildImportWithS3UriLocal(ctx, req, cluster, upload, area)
	if bizErr != nil {
		logger.Error("build import error", zap.Error(bizErr))
		return nil, bizErr
	}

	span.AddEvent("call CreateImport")
	resp, bizErr := s.create(
		ctx,
		imp,
		vo.Console2Columns(req.GetSource().GetLocal().GetColumns()),
		req.GetSource().GetLocal().GetPkColumns(),
	)
	if bizErr != nil {
		logger.Error("create import failed", zap.Error(bizErr))
		return nil, bizErr
	}

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

func (s *importService) create(
	ctx context.Context,
	imp *entity.ImportTask,
	columns []*entity.ColumnInfo,
	pkColumns []string,
) (_ *entity.Import, bizErr bizerr.BizError) {
	logger := log.WithContextL(ctx).
		With(zap.String("project_id", imp.ProjectID)).
		With(zap.String("cluster_id", imp.ClusterID))

	if imp.LocalImportCfg != nil {
		bizErr = s.importHelper.CreateTargetTable(ctx, imp.Cluster, imp.LocalImportCfg, columns, pkColumns)
		if bizErr != nil {
			logger.Error("CreateTargetTable failed", zap.Error(bizErr))
			return nil, bizErr
		}
	}

	id, bizErr := s.importSvc.CreateImport(ctx, imp)
	if bizErr != nil {
		return nil, bizErr
	}

	logger.Info("create import success", zap.String("import_id", id))
	im, bizErr := s.importSvc.GetImport(ctx, imp.ClusterID, id)
	if bizErr != nil {
		return nil, bizErr
	}

	return im, nil
}