package v1beta

import (
	"context"
	"fmt"
	"path/filepath"
	"strconv"
	"time"

	"github.com/aws/aws-sdk-go-v2/aws"
	"github.com/aws/aws-sdk-go-v2/service/s3"
	"github.com/tidbcloud/pkgv2/bizerr"
	"github.com/tidbcloud/pkgv2/log"
	"github.com/tidbcloud/pkgv2/tracing"
	"github.com/tidbcloud/serverless-service/idl/pbgen/proto/v1beta"
	"github.com/tidbcloud/serverless-service/internal/domain/constant"
	"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"
	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/trace"
	"go.uber.org/zap"
	"google.golang.org/protobuf/types/known/emptypb"
)

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

type importService struct {
	uploadRepo           repo.UploadRepo
	stashingAreaProvider *stashing.StashingAreaProvider
	importHelper         *import_svc.ImportHelperService
	importSvc            service.ImportService
	mgmtFacade           facade.MgmtSvcFacade

	v1beta.UnimplementedImportServiceServer
}

func NewImportService(
	uploadRepo repo.UploadRepo,
	stashingAreaProvider *stashing.StashingAreaProvider,
	importHelper *import_svc.ImportHelperService,
	importSvc service.ImportService,
	mgmtFacade facade.MgmtSvcFacade,
) v1beta.ImportServiceServer {
	return &importService{
		uploadRepo:           uploadRepo,
		stashingAreaProvider: stashingAreaProvider,
		importHelper:         importHelper,
		importSvc:            importSvc,
		mgmtFacade:           mgmtFacade,
	}
}

func (s *importService) ListImports(
	ctx context.Context,
	req *v1beta.ListImportsReq,
) (_ *v1beta.ListImportsResp, err error) {
	ctx, span := Tracer.Start(ctx, "v1beta.import.List")
	defer tracing.EndSpan(span, err)

	if req.PageSize == nil || *req.PageSize <= 0 {
		req.PageSize = aws.Int32(20)
	}
	if req.Page == nil || *req.Page < 1 {
		req.Page = aws.Int32(1)
	}
	span.AddEvent("call ListImports")
	offset := (*req.Page - 1) * *req.PageSize
	total, imports, err := s.importSvc.ListImports(ctx, &vo.ListImportOpt{
		ClusterID: strconv.FormatUint(req.ClusterId, 10),
		Offset:    int64(offset),
		PageSize:  *req.PageSize,
		OrderBy:   "created_at desc",
	})
	if err != nil {
		log.WithContextL(ctx).Error("ListImports failed", zap.Uint64("clusterId", req.ClusterId), zap.Error(err))
		return nil, err
	}

	return vo.ListImport2V1beta(imports, total)
}

func (s *importService) CreateImport(
	ctx context.Context,
	req *v1beta.CreateImportReq,
) (_ *v1beta.CreateImportResp, err error) {
	ctx, span := Tracer.Start(ctx, "v1beta.import.Create")
	defer tracing.EndSpan(span, err)

	if req.DataFormat == v1beta.DataFormat_CSV && req.CsvFormat == nil {
		req.CsvFormat = &v1beta.CustomCSVFormat{
			Separator:         ",",
			Delimiter:         "\"",
			Header:            true,
			NotNull:           false,
			Null:              "\\n",
			BackslashEscape:   true,
			TrimLastSeparator: false,
		}
	}

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

	switch req.Type {
	case v1beta.CreateImportReq_S3:
		return s.createImportFromS3(ctx, req, cluster, span)
	case v1beta.CreateImportReq_LOCAL:
		return s.createImportFromLocalFile(ctx, req, cluster, span)
	default:
		return nil, bizerr.New(bizerr.ErrCodeParams, "Unknown import type")
	}
}

func (s *importService) createImportFromS3(
	ctx context.Context,
	req *v1beta.CreateImportReq,
	cluster *tidbmgmtservicev1.Cluster,
	span trace.Span,
) (_ *v1beta.CreateImportResp, err error) {
	if req.AwsRoleArn == "" {
		return nil, bizerr.New(bizerr.ErrCodeParams, "Parameter aws_role_arn is empty")
	}
	if req.SourceUrl == "" {
		return nil, bizerr.New(bizerr.ErrCodeParams, "Parameter source_url is empty")
	}
	logger := log.WithContextL(ctx).With(zap.Uint64("cluster_id", req.ClusterId))

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

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

	return &v1beta.CreateImportResp{
		Id: resp.IDCompat,
	}, nil
}

func (s *importService) createImportFromLocalFile(
	ctx context.Context,
	req *v1beta.CreateImportReq,
	cluster *tidbmgmtservicev1.Cluster,
	span trace.Span,
) (_ *v1beta.CreateImportResp, err error) {
	if req.FileName == "" {
		return nil, bizerr.New(bizerr.ErrCodeParams, "Parameter file_name is empty")
	}
	if req.TargetTable == nil {
		return nil, bizerr.New(bizerr.ErrCodeParams, "Parameter target_table is empty")
	}
	if req.TargetTable.Schema == "" {
		return nil, bizerr.New(bizerr.ErrCodeParams, "Parameter target_table.schema is empty")
	}
	if req.TargetTable.Table == "" {
		return nil, bizerr.New(bizerr.ErrCodeParams, "Parameter target_table.table is empty")
	}

	logger := log.WithContextL(ctx).With(zap.Uint64("cluster_id", req.ClusterId))

	span.AddEvent("call GetUploadByID")
	upload, bizErr := s.uploadRepo.GetUploadByID(ctx, req.FileName)
	if bizErr != nil {
		return nil, bizErr
	}
	bizErr = import_svc.ValidateUpload(ctx, upload, cluster.ClusterId)
	if bizErr != nil {
		return nil, bizErr
	}

	span.AddEvent("get import file from stashing area")
	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)
	}

	object, err := area.GetObject(ctx, &s3.GetObjectInput{
		Bucket: aws.String(area.GetBucketName()),
		Key:    aws.String(upload.BucketKey),
		Range:  aws.String(fmt.Sprintf("bytes=0-%d", constant.PreviewReadSize)),
	})
	if err != nil {
		logger.Error(
			"GetObject failed",
			zap.Error(err),
			zap.String("bucket", area.GetBucketName()),
			zap.String("key", upload.BucketKey),
		)
		return nil, bizerr.New(errorcode.ErrCodeInternalError, errorcode.InternalErrorMessage)
	}
	defer object.Body.Close()

	dbPreviewResult, err := import_svc.PreviewLocalImport(
		ctx,
		object.Body,
		vo.V1beta2CsvConfig(req.CsvFormat),
		vo.V1betaDataFormat2Entity(req.DataFormat),
	)
	if err != nil {
		logger.Error("PreviewLocalImport failed", zap.Error(err))
		return nil, err
	}
	if len(dbPreviewResult.GetTables()) < 1 {
		logger.Error("no preview result for the table")
		return nil, bizerr.New(
			bizerr.ErrCodeParams,
			"no preview result for the table, please make sure the CSV file is correct",
		)
	}
	fileColumns := import_svc.ParseColumns(dbPreviewResult.GetTables()[0])

	span.AddEvent("call GetTableColumns")
	tableColumns, bizErr := s.importHelper.GetTableColumns(ctx, cluster, req.TargetTable.Schema, req.TargetTable.Table)
	var targetColumns []*entity.ColumnInfo
	if bizErr != nil {
		if bizErr.RawErrCode() != errorcode.ErrCodeTableNotExist {
			logger.Error("GetTableColumns failed", zap.Error(bizErr))
			return nil, bizErr
		}
	} else {
		targetColumns = import_svc.ParseColumns(tableColumns)
	}

	// if the target table exists, validate the columns
	span.AddEvent("validate table columns")
	if err = import_svc.ValidateColumns(fileColumns, targetColumns); err != nil {
		logger.Error("ValidateColumns failed", zap.Error(err))
		return nil, err
	}

	var columns []*entity.ColumnInfo
	// if targetTable does not exist, add fileColumns in order to create target table later
	if len(targetColumns) == 0 {
		columns = fileColumns
	}

	sourceUrl := fmt.Sprintf("s3://%s/%s", area.GetBucketName(), upload.BucketKey)
	imp, err := vo.BuildV1betaImportWithS3Uri(ctx, req, cluster, sourceUrl, upload)
	if err != nil {
		logger.Error("build import error", zap.Error(err))
		return nil, err
	}

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

	return &v1beta.CreateImportResp{
		Id: resp.IDCompat,
	}, nil
}

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

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

	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
}

func (s *importService) GetImport(ctx context.Context, req *v1beta.ImportIds) (_ *v1beta.GetImportResp, err error) {
	ctx, span := Tracer.Start(ctx, "v1beta.import.Get")
	defer tracing.EndSpan(span, err)

	i, err := s.importSvc.GetImportByIDCompat(ctx, strconv.FormatUint(req.ClusterId, 10), req.Id)
	if err != nil {
		return nil, err
	}

	return vo.GetImport2V1beta(i)
}

func (s *importService) CancelImport(ctx context.Context, req *v1beta.ImportIds) (_ *emptypb.Empty, err error) {
	ctx, span := Tracer.Start(ctx, "v1beta.import.Cancel")
	defer tracing.EndSpan(span, err)

	err = s.importSvc.CancelImportByIDCompat(ctx, strconv.FormatUint(req.ClusterId, 10), req.Id)
	if err != nil {
		return nil, err
	}

	return &emptypb.Empty{}, nil
}

func (s *importService) GenerateUploadURL(
	ctx context.Context,
	req *v1beta.GenerateUploadURLReq,
) (_ *v1beta.GenerateUploadURLResq, err error) {
	ctx, span := Tracer.Start(ctx, "v1beta.import.GenerateUrl")
	defer tracing.EndSpan(span, err)
	logger := log.WithContextL(ctx).With(zap.Uint64("cluster_id", req.ClusterId), zap.String("file_name", req.FileName))
	cluster, err := s.mgmtFacade.GetCluster(ctx, req.ClusterId, tidbmgmtservicev1.ClusterFilter_CLUSTER_VIEW_BASIC)
	if err != nil {
		logger.Error("failed to get cluster by id", zap.Error(err))
		return nil, bizerr.New(bizerr.ErrCodeGRPC, "failed to get cluster by id")
	}
	stashingArea, err := s.stashingAreaProvider.GetStashingArea(cluster.CloudProvider, cluster.Region)
	if err != nil {
		logger.Error("failed to get stashing area", zap.Error(err))
		return nil, bizerr.New(bizerr.ErrCodeUnknown, "failed to get stashing area")
	}

	id := entity.GenerateIDWithPrefix(constant.UploadIDPrefix)
	key := fmt.Sprintf(
		"%s/%s",
		stashingArea.GetObjectPath(strconv.FormatUint(cluster.GetOrganizationId(), 10), stashing.Import, id),
		filepath.Base(req.FileName),
	)

	// check storage limit
	exceeded, err := stashingArea.CheckExceedSize(ctx, strconv.FormatUint(cluster.GetOrganizationId(), 10))
	if err != nil {
		logger.Error("failed to check storage limit", zap.Error(err))
		return nil, bizerr.New(bizerr.ErrCodeUnknown, "failed to check storage limit")
	}
	if exceeded {
		return nil, bizerr.New(errorcode.ErrCodeStorageLimitExceed, "")
	}

	request, err := stashingArea.PresignPutObject(ctx, key, 6*time.Hour)
	if err != nil {
		logger.Error("failed to presign put object", zap.Error(err))
		return nil, bizerr.New(bizerr.ErrCodeUnknown, "failed to presign put object")
	}
	bizErr := s.uploadRepo.Create(ctx, &entity.Upload{
		ID:            id,
		Status:        constant.UploadStatusCompleted,
		Multipart:     false,
		AwsUploadID:   "",
		BucketKey:     key,
		TenantID:      cluster.GetOrganizationId(),
		ClusterID:     req.ClusterId,
		CloudProvider: cluster.GetCloudProvider(),
		RegionName:    cluster.GetRegion(),
	})
	if bizErr != nil {
		return nil, bizErr
	}
	// since only ticloud v0 use this api, we use newFileName to store the upload id for compatibility.
	return &v1beta.GenerateUploadURLResq{
		UploadUrl:   request.URL,
		NewFileName: id,
	}, nil
}

func (s *importService) validateCluster(
	ctx context.Context,
	clusterID uint64,
	logger *zap.Logger,
) (*tidbmgmtservicev1.Cluster, bizerr.BizError) {
	cluster, bizErr := s.mgmtFacade.GetCluster(ctx, clusterID, tidbmgmtservicev1.ClusterFilter_CLUSTER_VIEW_BASIC)
	if bizErr != nil {
		logger.Error("GetClusterByID failed", zap.Uint64("cluster_id", clusterID), zap.Error(bizErr))
		return nil, bizErr
	}

	if cluster.GetServerlessCluster().
		GetUserStatus() !=
		globalserverlessv1.UserClusterStatus_USER_CLUSTER_STATUS_NORMAL {
		return nil, bizerr.New(
			errorcode.ErrCodeClusterNotRunning,
			"Cluster ID %d is not in normal state, please try again later.",
			clusterID,
		)
	}
	return cluster, nil
}