package vo

import (
	"context"
	"encoding/json"
	"net/url"
	"path/filepath"
	"strconv"

	"github.com/aws/aws-sdk-go-v2/aws"
	"github.com/aws/aws-sdk-go-v2/feature/s3/manager"
	"github.com/aws/aws-sdk-go-v2/service/s3"
	"github.com/juju/errors"
	"github.com/pingcap/tidb/pkg/lightning/config"
	"github.com/tidbcloud/pkgv2/bizerr"
	"github.com/tidbcloud/pkgv2/import/cloud_storage"
	"github.com/tidbcloud/pkgv2/log"
	"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/errorcode"
	"github.com/tidbcloud/serverless-service/internal/utils"
	tidbmgmtservicev1 "github.com/tidbcloud/tidb-management-service/api/spec/tidb_mgmt_service/v1"
	"go.uber.org/zap"
	"google.golang.org/protobuf/types/known/timestamppb"
)

func ListImport2V1beta(imports []*entity.Import, total int64) (*v1beta.ListImportsResp, bizerr.BizError) {
	v := new(v1beta.ListImportsResp)
	v.Total = total

	for _, i := range imports {
		df2v1beta, err := GetImport2V1beta(i)
		if err != nil {
			return nil, err
		}
		v.Imports = append(v.Imports, df2v1beta)
	}
	return v, nil
}

func GetImport2V1beta(i *entity.Import) (*v1beta.GetImportResp, bizerr.BizError) {
	if i == nil {
		return nil, bizerr.New(bizerr.ErrCodeUnknown, "import is nil")
	}
	req, bizErr := ImportEntity2V1betaCreationDetails(i)
	if bizErr != nil {
		return nil, bizErr
	}

	resp := &v1beta.GetImportResp{
		SourceUrl:        i.SourceURL,
		CreatedAt:        timestamppb.New(i.CreatedAt),
		Status:           entityImportStatus2V1beta(i.Status),
		CompletedPercent: i.CompletedPercent,
		DataFormat:       entityDataFormat2V1beta(i.DataFormat),
		Message:          PrettyMessage(i.Message, i.DataFormat),
		Id:               i.IDCompat,
		CreationDetails:  req,
	}

	clusterID, err := strconv.ParseUint(i.ClusterID, 10, 64)
	if err != nil {
		return nil, bizerr.New(bizerr.ErrCodeParams, "invalid cluster id")
	}
	resp.ClusterId = clusterID

	if i.TotalSize != nil {
		resp.TotalSize = *i.TotalSize
	}

	if i.CompletedAt != nil {
		resp.ElapsedTimeSeconds = uint32(i.CompletedAt.Sub(i.CreatedAt).Seconds())
	}

	return resp, nil
}

func BuildV1betaImportWithS3Uri(
	ctx context.Context,
	req *v1beta.CreateImportReq,
	cluster *tidbmgmtservicev1.Cluster,
	s3Uri string,
	upload *entity.Upload,
) (*entity.ImportTask, bizerr.BizError) {
	uris, err := url.Parse(cloud_storage.EscapePath(s3Uri))
	if err != nil {
		return nil, bizerr.New(bizerr.ErrCodeParams, "invalid s3 uri")
	}
	bucketName := uris.Host
	// get region from bucket and check if bucket exists
	awsCfg := aws.Config{
		Region: "us-west-2",
	}
	s3Client := s3.NewFromConfig(awsCfg)
	region, err := manager.GetBucketRegion(ctx, s3Client, bucketName)
	if err != nil {
		var bnf manager.BucketNotFound
		if errors.As(err, &bnf) {
			return nil, bizerr.New(errorcode.ErrCodeS3BucketNotFound, errorcode.ErrMsgBucketNotFound)
		}
		return nil, bizerr.WrapError(err, "GetBucketRegion failed")
	}
	imp := &entity.ImportTask{
		Import: &entity.Import{
			ID:             entity.GenerateIDWithPrefix(constant.ImportIDPrefix),
			ProjectID:      strconv.FormatUint(cluster.GetOriginalProjectId(), 10),
			TenantID:       strconv.FormatUint(cluster.OrganizationId, 10),
			ClusterID:      strconv.FormatUint(req.ClusterId, 10),
			SourceURL:      s3Uri,
			BucketRegion:   region,
			DataSourceType: v1betaSourceType2Entity(req.Type),
			DataFormat:     V1betaDataFormat2Entity(req.DataFormat),
			NeedDropTables: false,
		},
		Cluster: cluster,
	}

	if req.DataFormat == v1beta.DataFormat_CSV {
		csvBytes, err := json.Marshal(req.CsvFormat)
		if err != nil {
			log.Error(ctx, "failed to marshal CSV format", zap.Error(err))
			return nil, bizerr.New(errorcode.ErrCodeInternalError, errorcode.InternalErrorMessage)
		}
		imp.CSVFormat = string(csvBytes)
	}

	if req.Type == v1beta.CreateImportReq_LOCAL {
		imp.LocalImportCfg = &entity.LocalImportCfg{
			FileName:    filepath.Base(upload.BucketKey),
			TargetDB:    req.GetTargetTable().GetSchema(),
			TargetTable: req.GetTargetTable().GetTable(),
			UploadID:    upload.ID,
		}
	}

	if req.Type == v1beta.CreateImportReq_S3 {
		if req.AwsRoleArn != "" {
			imp.S3RoleArn = req.AwsRoleArn
		}
	}

	imp.Creator = utils.BuildCreatorID(ctx)

	return imp, nil
}

func V1betaDataFormat2Entity(dataFormat v1beta.DataFormat) constant.DataFormat {
	switch dataFormat {
	case v1beta.DataFormat_CSV:
		return constant.ImportDataFormatCsv
	case v1beta.DataFormat_SqlFile:
		return constant.ImportDataFormatSql
	case v1beta.DataFormat_AuroraSnapshot:
		return constant.ImportDataFormatAuroraSnapshot
	case v1beta.DataFormat_Parquet:
		return constant.ImportDataFormatParquet
	}
	return constant.ImportDataFormatUnknown
}

func v1betaSourceType2Entity(sourceType v1beta.CreateImportReq_ImportType) constant.DataSource {
	switch sourceType {
	case v1beta.CreateImportReq_LOCAL:
		return constant.ImportDataSourceLocal
	case v1beta.CreateImportReq_S3:
		return constant.ImportDataSourceS3
	}
	return constant.ImportDataSourceUnknown
}

func entitySourceType2V1beta(sourceType constant.DataSource) v1beta.CreateImportReq_ImportType {
	switch sourceType {
	case constant.ImportDataSourceLocal:
		return v1beta.CreateImportReq_LOCAL
	case constant.ImportDataSourceS3:
		return v1beta.CreateImportReq_S3
	}
	return v1beta.CreateImportReq_UNKNOWN
}

func entityImportStatus2V1beta(entityResult constant.ImportStatus) v1beta.GetImportResp_Status {
	switch entityResult {
	case constant.ImportStatusUnknown:
		return v1beta.GetImportResp_UNKNOWN
	case constant.ImportStatusPreparing:
		return v1beta.GetImportResp_PREPARING
	case constant.ImportStatusImporting:
		return v1beta.GetImportResp_IMPORTING
	case constant.ImportStatusCompleted:
		return v1beta.GetImportResp_COMPLETED
	case constant.ImportStatusFailed:
		return v1beta.GetImportResp_FAILED
	case constant.ImportStatusCanceling:
		return v1beta.GetImportResp_CANCELING
	case constant.ImportStatusCanceled:
		return v1beta.GetImportResp_CANCELED
	default:
		return v1beta.GetImportResp_UNKNOWN
	}
}

func ImportEntity2V1betaCreationDetails(imp *entity.Import) (*v1beta.CreateImportReq, bizerr.BizError) {
	var csvFormat v1beta.CustomCSVFormat
	if imp.DataFormat == constant.ImportDataFormatCsv && imp.CSVFormat != "" {
		if err := json.Unmarshal([]byte(imp.CSVFormat), &csvFormat); err != nil {
			return nil, bizerr.New(bizerr.ErrCodeParams, "invalid custom CSV format")
		}
	}
	resp := &v1beta.CreateImportReq{
		Type:       entitySourceType2V1beta(imp.DataSourceType),
		DataFormat: entityDataFormat2V1beta(imp.DataFormat),
		CsvFormat:  &csvFormat,
		SourceUrl:  imp.SourceURL,
		AwsRoleArn: imp.S3RoleArn,

		TargetTable: nil,
	}
	clusterID, err := strconv.ParseUint(imp.ClusterID, 10, 64)
	if err != nil {
		return nil, bizerr.New(bizerr.ErrCodeParams, "invalid cluster id")
	}
	resp.ClusterId = clusterID
	if imp.LocalImportCfg != nil {
		resp.FileName = imp.LocalImportCfg.FileName
	}

	return resp, nil
}

func entityDataFormat2V1beta(format constant.DataFormat) v1beta.DataFormat {
	switch format {
	case constant.ImportDataFormatCsv:
		return v1beta.DataFormat_CSV
	case constant.ImportDataFormatSql:
		return v1beta.DataFormat_SqlFile
	case constant.ImportDataFormatAuroraSnapshot:
		return v1beta.DataFormat_AuroraSnapshot
	case constant.ImportDataFormatParquet:
		return v1beta.DataFormat_Parquet
	default:
		return v1beta.DataFormat_CanNotBeUnknown
	}
}

func V1beta2CsvConfig(format *v1beta.CustomCSVFormat) *config.CSVConfig {
	if format == nil {
		return nil
	}

	c := &config.CSVConfig{
		Separator:         format.GetSeparator(),
		Delimiter:         format.GetDelimiter(),
		Header:            format.GetHeader(),
		TrimLastSep:       format.GetTrimLastSeparator(),
		NotNull:           format.GetNotNull(),
		Null:              config.StringOrStringSlice{format.GetNull()},
		HeaderSchemaMatch: true,
	}
	if format.GetBackslashEscape() {
		c.EscapedBy = `\`
	}

	return c
}