package pimport

import (
	"context"
	"encoding/json"

	"github.com/pingcap/errcode"
	"github.com/pingcap/errors"
	"github.com/tidbcloud/dataflow-service/idl/pbgen/proto/dataflow"
	"github.com/tidbcloud/dataflow-service/internal/db/entity"
	pimport "github.com/tidbcloud/dataflow-service/internal/model/import"
	"github.com/tidbcloud/dataflow-service/internal/utils"
	"github.com/tidbcloud/pkgv2/log"
	"go.uber.org/zap"
)

func buildImportForValidation(ctx context.Context, req *dataflow.ImportValidateReq) (*pimport.ImportTask, error) {
	imp := &pimport.ImportTask{
		Import: entity.Import{
			ProjectID:        entity.ID(req.ProjectId),
			TenantID:         entity.ID(req.OrgId),
			ClusterID:        entity.ID(req.ClusterId),
			SourceURL:        req.SourceUrl,
			BucketRegion:     req.Region,
			DataSourceType:   pbDataSource2Entity(dataflow.ImportDataSource(req.DataSourceType)),
			DataFormat:       pbDataFormat2Entity(dataflow.ImportDataFormat(req.DataFormat)),
			AwsRoleArn:       req.AwsRoleArn,
			TargetTableInfos: pbTargetTableInfos2Entity(req.TargetTableInfos),
		},
	}

	var err error
	imp.CustomCSVFormat, err = buildCsvCfg(ctx, req.CsvFormat)
	if err != nil {
		log.Error(ctx, "build csv config failed", zap.Error(err))
		return nil, err
	}

	var accessConfig *entity.ImportSourceAccessConfig
	switch req.GetDataSourceType() {
	case uint32(dataflow.ImportDataSource_S3_SOURCE):
		accessConfig = &entity.ImportSourceAccessConfig{
			AWSAccessKeyID:     req.GetAwsAccessKeyId(),
			AWSSecretAccessKey: req.GetAwsSecretAccessKey(),
			AWSRoleARN:         req.GetAwsRoleArn(),
			// external will be added later, when adjusting the import entity
		}
	case uint32(dataflow.ImportDataSource_AZBLOB_SOURCE):
		accessConfig = &entity.ImportSourceAccessConfig{
			AzblobSASToken: req.AzblobSasToken,
		}
	}
	log.Info(ctx, "buildImportForValidation before set accessConfig", zap.Any("accessConfig", accessConfig), zap.Any("req", req), zap.Any("imp", imp))
	if err := imp.SetAccessConfigInSourceURL(accessConfig); err != nil {
		return nil, errors.Annotate(err, "cannot set the access config for the source URL")
	}
	log.Info(ctx, "buildImportForValidation after set accessConfig", zap.Any("imp", imp))
	return imp, nil
}

func getDupHandlingForSQLConfigValue(enumValue dataflow.DupHandlingForSql) string {
	switch enumValue {
	case dataflow.DupHandlingForSql_DUP_HANDLING_FOR_SQL_REPLACE:
		return "replace"
	case dataflow.DupHandlingForSql_DUP_HANDLING_FOR_SQL_IGNORE:
		return "ignore"
	case dataflow.DupHandlingForSql_DUP_HANDLING_FOR_SQL_ERROR:
		return "error"
	default:
		return ""
	}
}

func buildImport(ctx context.Context, req *dataflow.ImportCreateReq) (*pimport.ImportTask, error) {
	csvCfg, err := buildCsvCfg(ctx, req.CsvFormat)
	if err != nil {
		errMsg := "build csv config failed, please check the correctness of your csv format."
		log.Error(ctx, errMsg, zap.Error(err))
		return nil, errcode.NewInvalidInputErr(errors.New(errMsg))
	}

	imp := &pimport.ImportTask{
		Import: entity.Import{
			ProjectID:       entity.ID(req.ProjectId),
			TenantID:        entity.ID(req.OrgId),
			Name:            req.GetName(),
			ClusterID:       entity.ID(req.ClusterId),
			SourceURL:       req.SourceUrl,
			BucketRegion:    req.Region,
			DataSourceType:  pbDataSource2Entity(dataflow.ImportDataSource(req.DataSourceType)),
			DataFormat:      pbDataFormat2Entity(dataflow.ImportDataFormat(req.DataFormat)),
			AwsRoleArn:      req.AwsRoleArn,
			CustomCSVFormat: csvCfg,
			OptionalArgs: &entity.OptionalArgs{
				DuplicationHandlingForSQL: getDupHandlingForSQLConfigValue(req.GetDupHandlingForSql()),
				SkipChecksum:              req.GetSkipChecksum(),
			},
			TargetTableInfos: pbTargetTableInfos2Entity(req.TargetTableInfos),
			NeedDropTables:   req.NeedDropTables,
			ConnMethod:       entity.ConnMethod(req.ConnMethod),
		},
	}

	if req.EndpointServiceName != nil {
		imp.EndpointServiceName = *req.EndpointServiceName
	}

	if req.LocalImportCfg != nil {
		imp.LocalImportCfg = &entity.LocalImportCfg{
			FileName:         req.LocalImportCfg.FileName,
			TargetDb:         req.LocalImportCfg.TargetTable.Schema,
			TargetTable:      req.LocalImportCfg.TargetTable.Table,
			OriginalFileName: req.LocalImportCfg.OriginalFileName,
			TotalSize:        req.LocalImportCfg.TotalSize,
		}
	}

	var accessConfig *entity.ImportSourceAccessConfig
	switch req.GetDataSourceType() {
	case uint32(dataflow.ImportDataSource_S3_SOURCE):
		accessConfig = &entity.ImportSourceAccessConfig{
			AWSAccessKeyID:     req.GetAwsAccessKeyId(),
			AWSSecretAccessKey: req.GetAwsSecretAccessKey(),
			AWSRoleARN:         req.GetAwsRoleArn(),
			// external will be added later, when adjusting the import entity
		}
	case uint32(dataflow.ImportDataSource_AZBLOB_SOURCE):
		accessConfig = &entity.ImportSourceAccessConfig{
			AzblobSASToken: req.GetAzblobSasToken(),
		}
	}
	if err := imp.SetAccessConfigInSourceURL(accessConfig); err != nil {
		return nil, errors.Annotate(err, "cannot set the access config for the source URL")
	}
	return imp, nil
}

func adjustCSVConfig(ctx context.Context, csvFormat *dataflow.CustomCSVFormat) {
	logger := log.WithContextL(ctx)
	if csvFormat == nil {
		return
	}
	actualDelimiter, err := utils.RestoreSpecialChars(csvFormat.Delimiter)
	if err != nil {
		logger.Error("restore special chars of CSV delimiter error", zap.Error(err))
		actualDelimiter = csvFormat.Delimiter
	}
	csvFormat.Delimiter = actualDelimiter
	actualSeparator, err := utils.RestoreSpecialChars(csvFormat.Separator)
	if err != nil {
		logger.Error("restore actual chars of CSV separator error", zap.Error(err))
		actualSeparator = csvFormat.Separator
	}
	csvFormat.Separator = actualSeparator
}

func buildCsvCfg(ctx context.Context, csvFormat *dataflow.CustomCSVFormat) (string, error) {
	if csvFormat == nil {
		return "", nil
	}

	adjustCSVConfig(ctx, csvFormat)

	csvConfigBytes, err := json.Marshal(csvFormat)
	if err != nil {
		log.Error(ctx, "failed to marshal csv config", zap.Error(err))
		return "", err
	}

	return string(csvConfigBytes), nil
}