package pimport

import (
	"context"
	"encoding/json"
	"fmt"
	"regexp"
	"sort"
	"strconv"
	"time"

	"github.com/pingcap/errors"
	"github.com/pingcap/log"
	"github.com/tidbcloud/dataflow-service/idl/pbgen/proto/agent"
	commonpb "github.com/tidbcloud/dataflow-service/idl/pbgen/proto/common"
	"github.com/tidbcloud/dataflow-service/idl/pbgen/proto/dataflow"
	"github.com/tidbcloud/dataflow-service/internal/common"
	"github.com/tidbcloud/dataflow-service/internal/db/entity"
	pimport "github.com/tidbcloud/dataflow-service/internal/model/import"
	"github.com/tidbcloud/dataflow-service/internal/utils"
	errhandler "github.com/tidbcloud/dataflow-service/pkg/error-handler"
	pkggrpcli "github.com/tidbcloud/dataflow-service/pkg/grpc/grpcli"
	cloudv1 "github.com/tidbcloud/infra-api/api/v1"
	"github.com/tidbcloud/pkgv2/bizerr"
	"go.uber.org/zap"
	"google.golang.org/protobuf/types/known/timestamppb"
)

const (
	syntaxErrInstructionForCSV = "Please truncate the target table, check your CSV configuration, modify it, and re-import again. ( For example, if you use two double quotes to escape the double quote mark in your CSV, consider set `backslash-escape` to `false` )"
	syntaxErrInstructionCommon = "Please provide the correct data file then truncate the target table, create a new import task to import the data again."
)

func getUserImportFile(ctx context.Context, sourceUrl, roleArn string, start, end int64, clusterID uint64, dataSourceType uint32) (string, error) {
	ctx, cancel := context.WithTimeout(ctx, 30*time.Second)
	defer cancel()
	ctx = common.ContextWithImport(ctx)
	cluster, network, agentClient, err := common.GetClusterNetworkAgentById(ctx, clusterID)
	if err != nil {
		return "", errors.Annotate(err, "fail to get agent client")
	}

	var nodeRole string
	if cluster.GetAttributes().GetUseNewDedicatedArch() {
		var bizErr bizerr.BizError
		nodeRole, bizErr = pkggrpcli.InfraClient().GetClusterNodeRole(ctx, cluster.Id)
		if bizErr != nil {
			return "", errors.Annotate(bizErr, "get node role error")
		}
	}

	req := &agent.GetUserFileReq{
		SourceUrl:      sourceUrl,
		StartOffset:    start,
		EndOffset:      end,
		RoleArn:        roleArn,
		ExternalId:     network.AWSExternalID(),
		DataSourceType: dataSourceType,
		GcsOptions: &agent.GcsOptions{
			ImpersonateServiceAccount: nodeRole,
		},
		SasToken: utils.GetSasTokenFromSourceURL(sourceUrl),
	}
	resp, err := agentClient.GetUserFileContent(ctx, req)
	if err != nil {
		return "", errors.Annotate(err, "fail to get user file content by agent")
	}
	return resp.Content, nil
}

func findFileNameOffset(msg string, startIdx int) (string, int64, error) {
	var (
		idxs []int
		err  error
	)
	// find the offset
	r := regexp.MustCompile("at offset ")
	if idxs = r.FindIndex([]byte(msg)); idxs == nil {
		// unknown syntax error
		return "", 0, errors.New("not found `at offset` for syntax error")
	}
	begIdx := idxs[0]
	fileNameEndIdx := -1
	for i := begIdx; i > startIdx; i-- {
		if msg[i] == ':' {
			fileNameEndIdx = i
			break
		}
	}
	if fileNameEndIdx == -1 {
		return "", 0, errors.New("not found file name for syntax error")
	}
	fileName := msg[startIdx:fileNameEndIdx]
	begIdx = idxs[1]
	offsetEndIdx := -1
	for i := begIdx; i < len(msg); i++ {
		if msg[i] == ':' {
			offsetEndIdx = i
			break
		}
	}
	if offsetEndIdx == -1 {
		return "", 0, errors.New("not found offset for syntax error")
	}
	offset, err := strconv.ParseUint(msg[begIdx:offsetEndIdx], 10, 64)
	if err != nil {
		return "", 0, errors.Annotate(err, "fail to convert offset to int")
	}
	return fileName, int64(offset), nil
}

func convertSyntaxErr(ctx context.Context, impResp *dataflow.ImportGetResp, imp *pimport.ImportTask, logger *zap.Logger) {
	r := regexp.MustCompile("SyntaxError: ")
	var idxs []int
	if idxs = r.FindIndex([]byte(impResp.Message)); idxs == nil {
		// not syntax error
		return
	}
	// Parquet files' data are written in binary format. So the printed data from parquet here is
	// not human-readable and meaningless. Then we directly print the instruction here.
	if imp.DataFormat == entity.ImportDataFormatCsv {
		impResp.Message = impResp.Message + "\n" + syntaxErrInstructionCommon
		return
	}
	begIdx := idxs[1]
	fileName, offset, err := findFileNameOffset(impResp.Message, begIdx)
	if err != nil {
		logger.Warn("fail to find file name and offset for syntax error", zap.Error(err))
		return
	}
	// get content from source
	endOffset := offset + 100
	startOffset := offset - 100
	if startOffset < 0 {
		startOffset = 0
	}
	sourceFileURL := fmt.Sprintf("%s/%s", imp.SourceURL, fileName)
	var (
		targetContent string
		instruction   string
	)
	content, err := getUserImportFile(ctx, sourceFileURL, imp.AwsRoleArn, startOffset, endOffset, uint64(imp.ClusterID),
		uint32(entityDataSource2PB(imp.DataSourceType)))
	if err != nil {
		logger.Warn("fail to read file from s3", zap.Error(err))
		return
	}
	if imp.DataFormat == entity.ImportDataFormatCsv {
		instruction = syntaxErrInstructionForCSV
		csvConfig := &dataflow.CustomCSVFormat{}
		if err := json.Unmarshal([]byte(imp.CustomCSVFormat), csvConfig); err != nil {
			logger.Warn("fail to unmarshal custom csv format", zap.Error(err))
			return
		}
		delimiter := byte('"')
		if csvConfig.Delimiter != "" {
			delimiter = []byte(csvConfig.Delimiter)[0]
		}
		targetContent = extractFileSyntaxErr(offset-startOffset, content, delimiter, '\n')
	} else {
		instruction = syntaxErrInstructionCommon
		targetContent = extractFileSyntaxErr(offset-startOffset, content, '"', ')')
	}

	if targetContent != "" {
		impResp.Message = impResp.Message + "\n Corresponding data: " + targetContent + " \n" + instruction
	} else {
		impResp.Message = impResp.Message + "\n" + instruction
	}
}

func extractFileSyntaxErr(start int64, content string, delimiter, separator byte) string {
	if start >= int64(len(content)) {
		// TODO: lightning will record an invalid offset that can exceed the range of the file content
		// when encountering unterminated quote. We should fix it in lightning.
		return ""
	}
	quoted := false
	targetOffset := int64(-1)
loop:
	for i := start; i >= 0; i-- {
		switch content[i] {
		case delimiter:
			quoted = !quoted
		case '\r', '\n', separator:
			if !quoted {
				targetOffset = i + 1
				break loop
			}
		}
	}
	targetContent := content
	if targetOffset >= 0 && targetOffset < int64(len(content)) {
		targetContent = content[targetOffset:]
	}
	return targetContent
}

func ImportEntity2PB(ctx context.Context, logger *zap.Logger, clusterId uint64, imp *pimport.ImportTask, needShowSyntaxErrorContent bool) *dataflow.ImportGetResp {
	resp := &dataflow.ImportGetResp{
		Id:                         uint64(imp.ID),
		ClusterId:                  clusterId,
		CompletedTables:            imp.CompletedTables,
		PendingTables:              imp.PendingTables,
		TotalTablesCount:           imp.TotalTablesCount,
		CompletedPercent:           imp.CompletedPercent,
		PostImportCompletedPercent: imp.PostImportCompletedPercent,
		Message:                    errhandler.PrettyMessage(imp.Message, uint32(imp.DataFormat)),
		DataFormat:                 uint32(entityDataFormat2PB(imp.DataFormat)),
		DataSourceType:             uint32(entityDataSource2PB(imp.DataSourceType)),
		Name:                       imp.Name,
		CreatedBy:                  imp.Creator,
	}
	var err error
	resp.SourceUrl, err = EscapeQuery(imp.SourceURL)
	if err != nil {
		logger.Error("failed to parse source url", zap.Error(err), zap.String("source url", imp.SourceURL))
	}

	createdAt := timestamppb.New(imp.CreatedAt)
	if err := createdAt.CheckValid(); err != nil {
		logger.Error("failed to parse import's CreatedAt field", zap.Uint64("id", uint64(imp.ID)), zap.Error(err))
	} else {
		resp.CreatedAt = createdAt
	}

	if imp.TotalSize != nil {
		resp.TotalSize = *imp.TotalSize
	}
	if imp.ProcessedDataSize != nil {
		resp.ProcessedSourceDataSize = *imp.ProcessedDataSize
	}
	if imp.TotalFiles != nil {
		resp.TotalFiles = *imp.TotalFiles
	}
	resp.Status = entityStatusToPBStatus(imp.Status)
	switch resp.Status {
	case dataflow.ImportGetResp_PREPARING:
		resp.CompletedPercent = 0
		resp.PostImportCompletedPercent = 0
	case dataflow.ImportGetResp_COMPLETED:
		resp.CompletedPercent = 100
		resp.PostImportCompletedPercent = 100
	case dataflow.ImportGetResp_UNKNOWN:
		logger.Error("unknown status", zap.Stringer("status", imp.Status))
	}
	end := time.Now()
	if imp.FinishedAt != nil {
		end = *imp.FinishedAt
	}
	resp.ElapsedTimeSeconds = uint32(end.Sub(imp.CreatedAt).Seconds())

	for _, t := range imp.CurrentTables.Tables {
		respT := &dataflow.CurrentTable{
			Name:             t.Name,
			Size:             t.Size,
			CompletedPercent: t.CompletedPercent,
		}
		resp.CurrentTables = append(resp.CurrentTables, respT)
	}

	tableCompleteInfos := []*dataflow.ImportTableCompletionInfo{}
	switch resp.Status {
	case dataflow.ImportGetResp_COMPLETED, dataflow.ImportGetResp_FAILED:
		if imp.LocalImportCfg != nil {
			resp.AllCompletedTables = []*dataflow.ImportTableCompletionInfo{
				{
					TableName: pimport.GenTableID(imp.LocalImportCfg.TargetDb, imp.LocalImportCfg.TargetTable),
					Result:    entityImportStatus2PBTableResult(resp.Status),
				},
			}
			break
		}
		for _, t := range imp.TableResults.Tables {
			completionInfo := &dataflow.ImportTableCompletionInfo{
				TableName: t.TableName,
				Result:    entityTableResult2PBTableResult(t.Result),
				Message:   t.Message,
			}
			tableCompleteInfos = append(tableCompleteInfos, completionInfo)
		}
		resp.AllCompletedTables = tableCompleteInfos
	}

	createReq, err := ImportEntity2CreateReq(imp.Import)
	if err != nil {
		logger.Error("construct the create request from the import entity failed", zap.Error(err))
	}
	resp.ImportCreateReq = createReq

	logger.Info("get import finished", zap.Reflect("response", resp))
	if needShowSyntaxErrorContent {
		convertSyntaxErr(ctx, resp, imp, logger)
	}

	if imp.LocalImportCfg != nil {
		// don't show the sourceUrl when import local file
		resp.SourceUrl = ""
		resp.TotalSize = uint64(imp.LocalImportCfg.TotalSize)
	}
	// don't show the preparing status because the progress will increase in preparing now
	if resp.Status == dataflow.ImportGetResp_PREPARING {
		resp.Status = dataflow.ImportGetResp_IMPORTING
	}
	return resp
}

func ImportEntity2CreateReq(imp entity.Import) (*dataflow.ImportCreateReq, error) {
	var customCSVFormat dataflow.CustomCSVFormat
	if imp.DataFormat == entity.ImportDataFormatCsv && imp.CustomCSVFormat != "" {
		if err := json.Unmarshal([]byte(imp.CustomCSVFormat), &customCSVFormat); err != nil {
			return nil, errors.Annotate(err, "parse custom CSV format failed")
		}
	}
	resp := &dataflow.ImportCreateReq{
		ClusterId:      uint64(imp.ClusterID),
		Region:         imp.BucketRegion,
		DataFormat:     uint32(entityDataFormat2PB(imp.DataFormat)),
		DataSourceType: uint32(entityDataSource2PB(imp.DataSourceType)),
		CsvFormat:      &customCSVFormat,
		AwsRoleArn:     imp.AwsRoleArn,
	}
	if imp.OptionalArgs != nil {
		resp.SkipChecksum = imp.OptionalArgs.SkipChecksum
	}

	if imp.LocalImportCfg != nil {
		resp.LocalImportCfg = &dataflow.LocalImportCfg{
			TargetTable: &commonpb.Table{
				Schema: imp.LocalImportCfg.TargetDb,
				Table:  imp.LocalImportCfg.TargetTable,
			},
			OriginalFileName: imp.LocalImportCfg.OriginalFileName,
			TotalSize:        imp.LocalImportCfg.TotalSize,
		}
		resp.DataSourceType = uint32(dataflow.ImportDataSource_LOCAL_SOURCE)
	} else {
		resp.TargetTableInfos = entityImportTargetTableInfos2PB(imp.TargetTableInfos)
		var err error
		resp.SourceUrl, err = EscapeQuery(imp.SourceURL)
		if err != nil {
			log.L().Error("failed to parse source url", zap.Error(err), zap.String("source url", imp.SourceURL))
		}
	}
	return resp, nil
}

func saveImportStatusIntoPBResp(importStatusFromInfra *cloudv1.ImportStatus, resp *dataflow.ImportGetResp) {
	resp.TotalSize = importStatusFromInfra.TotalSize
	resp.CompletedPercent = importStatusFromInfra.CompletedPercent
	resp.PostImportCompletedPercent = importStatusFromInfra.PostImportCompletedPercent
	resp.TotalTablesCount = importStatusFromInfra.TotalTables
	resp.CompletedTables = importStatusFromInfra.CompletedTables
	resp.PendingTables = importStatusFromInfra.PendingTables
	resp.Message = importStatusFromInfra.Message
	resp.ProcessedSourceDataSize = importStatusFromInfra.ProcessedDataSize

	tableCompleteInfos := []*dataflow.ImportTableCompletionInfo{}
	for _, tblDetail := range importStatusFromInfra.AllTableDetails {
		switch tblDetail.Phase {
		case cloudv1.TableImportPhaseComplete:
			completionInfo := &dataflow.ImportTableCompletionInfo{
				TableName: tblDetail.Name,
				Result:    dataflow.ImportTableCompletionInfo_SUCCESS,
			}
			tableCompleteInfos = append(tableCompleteInfos, completionInfo)
		case cloudv1.TableImportPhaseFailed:
			completionInfo := &dataflow.ImportTableCompletionInfo{
				TableName: tblDetail.Name,
				Result:    dataflow.ImportTableCompletionInfo_ERROR,
				Message:   tblDetail.Message,
			}
			tableCompleteInfos = append(tableCompleteInfos, completionInfo)
		}
	}
	sort.Slice(tableCompleteInfos, func(i, j int) bool { return tableCompleteInfos[i].TableName < tableCompleteInfos[j].TableName })
	resp.AllCompletedTables = tableCompleteInfos

	importingTables := []*dataflow.CurrentTable{}
	for _, curTbl := range importStatusFromInfra.CurrentTables {
		importingTables = append(importingTables, &dataflow.CurrentTable{
			Name:             curTbl.Name,
			Size:             curTbl.Size,
			CompletedPercent: curTbl.CompletedPercent,
		})
	}
	sort.Slice(importingTables, func(i, j int) bool { return importingTables[i].Name < importingTables[j].Name })
	resp.CurrentTables = importingTables
}

func pb2ImportStatus(status dataflow.ImportGetResp_Status) entity.ImportStatus {
	switch status {
	case dataflow.ImportGetResp_PREPARING:
		return entity.ImportStatusPreparing
	case dataflow.ImportGetResp_IMPORTING:
		return entity.ImportStatusImporting
	case dataflow.ImportGetResp_COMPLETED:
		return entity.ImportStatusCompleted
	case dataflow.ImportGetResp_FAILED:
		return entity.ImportStatusFailed
	case dataflow.ImportGetResp_CANCELED:
		return entity.ImportStatusCanceled
	case dataflow.ImportGetResp_CANCELING:
		return entity.ImportStatusCanceling
	default:
		return entity.ImportStatusUnknown
	}
}

func entityStatusToPBStatus(status entity.ImportStatus) dataflow.ImportGetResp_Status {
	switch status {
	case entity.ImportStatusPreparing:
		return dataflow.ImportGetResp_PREPARING
	case entity.ImportStatusImporting:
		return dataflow.ImportGetResp_IMPORTING
	case entity.ImportStatusCompleted:
		return dataflow.ImportGetResp_COMPLETED
	case entity.ImportStatusFailed:
		return dataflow.ImportGetResp_FAILED
	case entity.ImportStatusCanceled:
		return dataflow.ImportGetResp_CANCELED
	case entity.ImportStatusCanceling:
		return dataflow.ImportGetResp_CANCELING
	default:
		return dataflow.ImportGetResp_UNKNOWN
	}
}

func entityImportStatus2PBTableResult(entityResult dataflow.ImportGetResp_Status) dataflow.ImportTableCompletionInfo_Result {
	switch entityResult {
	case dataflow.ImportGetResp_COMPLETED:
		return dataflow.ImportTableCompletionInfo_SUCCESS
	case dataflow.ImportGetResp_FAILED:
		return dataflow.ImportTableCompletionInfo_ERROR
	default:
		return dataflow.ImportTableCompletionInfo_UNKNOWN
	}
}

func entityTableResult2PBTableResult(entityResult entity.ImportTableResult) dataflow.ImportTableCompletionInfo_Result {
	switch entityResult {
	case entity.ImportTableResultSuccess:
		return dataflow.ImportTableCompletionInfo_SUCCESS
	case entity.ImportTableResultWarning:
		return dataflow.ImportTableCompletionInfo_WARNING
	case entity.ImportTableResultError:
		return dataflow.ImportTableCompletionInfo_ERROR
	default:
		return dataflow.ImportTableCompletionInfo_UNKNOWN
	}
}

func pbStatusList2ImportStatusList(pbStatuses []dataflow.ImportGetResp_Status) []entity.ImportStatus {
	statuses := make([]entity.ImportStatus, len(pbStatuses))
	for i, st := range pbStatuses {
		statuses[i] = pb2ImportStatus(st)
	}
	return statuses
}

func importEntitiesToListRespPB(ctx context.Context, clusterID entity.ID, totalCount int64, importEntities []*pimport.ImportTask, logger *zap.Logger) (*dataflow.ListImportResp, error) {
	resImports := []*dataflow.ImportGetResp{}
	for _, imp := range importEntities {
		getResp := ImportEntity2PB(ctx, logger, uint64(clusterID), imp, false)
		if imp.LocalImportCfg == nil {
			tweakImportProgress(imp, getResp)
		}
		resImports = append(resImports, getResp)
	}

	return &dataflow.ListImportResp{
		Total:   totalCount,
		Imports: resImports,
	}, nil
}

func pbTargetTableInfos2Entity(infos []*dataflow.ImportTargetTableInfo) entity.TargetTableInfos {
	ret := make([]entity.TargetTableInfo, 0, len(infos))

	for _, info := range infos {
		ret = append(ret, entity.TargetTableInfo{
			Schema:      info.TargetTable.Schema,
			Table:       info.TargetTable.Table,
			CustomFile:  info.CustomFile,
			MatchedFile: info.MatchedUri,
		})
	}
	return entity.TargetTableInfos{Infos: ret}
}

func entityImportTargetTableInfos2PB(infos entity.TargetTableInfos) []*dataflow.ImportTargetTableInfo {
	ret := make([]*dataflow.ImportTargetTableInfo, 0, len(infos.Infos))
	for _, m := range infos.Infos {
		ret = append(ret, &dataflow.ImportTargetTableInfo{
			TargetTable: &commonpb.Table{
				Schema: m.Schema,
				Table:  m.Table,
			},
			CustomFile: m.CustomFile,
		})
	}
	return ret
}

func pbDataFormat2Entity(dataFormat dataflow.ImportDataFormat) entity.DataFormat {
	switch dataFormat {
	case dataflow.ImportDataFormat_SQL:
		return entity.ImportDataFormatSql
	case dataflow.ImportDataFormat_CSV:
		return entity.ImportDataFormatCsv
	case dataflow.ImportDataFormat_PARQUET:
		return entity.ImportDataFormatParquet
	case dataflow.ImportDataFormat_AURORA_SNAPSHOT:
		return entity.ImportDataFormatAuroraSnapshot
	}
	return entity.ImportDataFormatUnknown
}

func pbDataSource2Entity(dataSource dataflow.ImportDataSource) entity.DataSource {
	switch dataSource {
	case dataflow.ImportDataSource_S3_SOURCE:
		return entity.ImportDataSourceS3
	case dataflow.ImportDataSource_GCS_SOURCE:
		return entity.ImportDataSourceGcs
	case dataflow.ImportDataSource_AZBLOB_SOURCE:
		return entity.ImportDataSourceAzblob
	}
	return entity.ImportDataSourceUnknown
}

func entityDataFormat2PB(dataFormat entity.DataFormat) dataflow.ImportDataFormat {
	switch dataFormat {
	case entity.ImportDataFormatSql:
		return dataflow.ImportDataFormat_SQL
	case entity.ImportDataFormatCsv:
		return dataflow.ImportDataFormat_CSV
	case entity.ImportDataFormatParquet:
		return dataflow.ImportDataFormat_PARQUET
	case entity.ImportDataFormatAuroraSnapshot:
		return dataflow.ImportDataFormat_AURORA_SNAPSHOT
	}
	return dataflow.ImportDataFormat_UNKNOWN_FORMAT
}

func entityDataSource2PB(dataSource entity.DataSource) dataflow.ImportDataSource {
	switch dataSource {
	case entity.ImportDataSourceS3:
		return dataflow.ImportDataSource_S3_SOURCE
	case entity.ImportDataSourceGcs:
		return dataflow.ImportDataSource_GCS_SOURCE
	case entity.ImportDataSourceAzblob:
		return dataflow.ImportDataSource_AZBLOB_SOURCE
	}
	return dataflow.ImportDataSource_UNKNOWN_SOURCE
}