package import_svc

import (
	"bytes"
	"context"
	"fmt"
	"io"
	"strings"
	"time"
	"unicode/utf8"

	"github.com/dustin/go-humanize"
	"github.com/pingcap/errors"
	"github.com/pingcap/tidb/pkg/lightning/config"
	"github.com/pingcap/tidb/pkg/lightning/mydump"
	"github.com/pingcap/tidb/pkg/lightning/worker"
	"github.com/pingcap/tidb/pkg/parser/mysql"
	ptypes "github.com/pingcap/tidb/pkg/parser/types"
	"github.com/pingcap/tidb/pkg/sessionctx/stmtctx"
	"github.com/pingcap/tidb/pkg/types"
	agentpb "github.com/tidbcloud/dataflow-service/idl/pbgen/proto/agent"
	str_utils "github.com/tidbcloud/dataflow-service/pkg/string-utils"
	"github.com/tidbcloud/pkgv2/bizerr"
	"github.com/tidbcloud/pkgv2/log"
	"github.com/tidbcloud/serverless-service/internal/domain/constant"
	"github.com/tidbcloud/serverless-service/internal/errorcode"
	"go.uber.org/zap"
)

const (
	defaultPreviewLimit = 5
	maxColumnPreview    = 50
)

type ErrTooManyColumns struct{}

func (e *ErrTooManyColumns) Error() string {
	return "too many columns"
}

type ErrCSVSyntax struct {
	err error
}

func (e *ErrCSVSyntax) Error() string {
	// Example error message: "syntax error: unterminated quoted field"
	msg := e.err.Error()
	idx := strings.Index(msg, "github.com")
	if idx != -1 {
		msg = msg[:idx]
	}
	return strings.TrimSpace(msg)
}

type ErrCSVRowOverflow struct {
	RowNum int
}

func (e *ErrCSVRowOverflow) Error() string {
	return fmt.Sprintf("the data num of one row is greater than header on %s row", humanize.Ordinal(e.RowNum+1))
}

type ErrHeaderDataColumnMismatch struct {
	HeaderColumnsCount int
	DataColumnsCount   int
}

func (e *ErrHeaderDataColumnMismatch) Error() string {
	return fmt.Sprintf(
		"the header column count (%d) does not equal to data column count (%d).",
		e.HeaderColumnsCount,
		e.DataColumnsCount,
	)
}

type ErrPreviewHasWarning struct {
	WarningMessage string
}

func (e *ErrPreviewHasWarning) Error() string {
	return fmt.Sprintf("the preview has warning: %s", e.WarningMessage)
}

type ImportPreviewer interface {
	Preview(ctx context.Context, limitCount uint) ([]*agentpb.Schema, error)
}

// PreviewFileMapping defines the source file to target table mapping.
type PreviewFileMapping struct {
	DatabaseName string
	TableName    string
}

// LocalImportPreviewer is the implementation that the preview can be run on the server locally without any problems.
// Usually this means:
//   - The server can access the source URL without any problems. (no assuming role issues)
//   - The list of files to preview is clear from the import entity:
//     no need to scan the source to get how many tables are there,
//     and no need to find the first data file name for each target table to do the preview.
//
// As a result, currently the previewer only applies to local file import situations, which meets all the pre-conditions.
type LocalImportPreviewer struct {
	PreviewFileMappings []*PreviewFileMapping
	CSVConfig           *config.CSVConfig
	DataFormat          constant.DataFormat
	File                io.ReadSeekCloser
}

func (p *LocalImportPreviewer) Preview(ctx context.Context, limitCount uint) ([]*agentpb.Schema, error) {
	schemaPreviewMap := map[string]*agentpb.Schema{}
	for _, fileMapping := range p.PreviewFileMappings {
		previewTable, err := p.doSingleTablePreview(ctx, fileMapping, limitCount)
		if err != nil {
			return nil, err
		}
		schemaName := fileMapping.DatabaseName
		schemaPreview, ok := schemaPreviewMap[schemaName]
		if !ok {
			schemaPreview = &agentpb.Schema{
				SchemaName: schemaName,
			}
			schemaPreviewMap[schemaName] = schemaPreview
		}
		schemaPreview.Tables = append(schemaPreview.GetTables(), previewTable)
	}
	result := []*agentpb.Schema{}
	for _, schemaPreview := range schemaPreviewMap {
		result = append(result, schemaPreview)
	}
	return result, nil
}

func (p *LocalImportPreviewer) newParser(ctx context.Context) (mydump.Parser, error) {
	var parser mydump.Parser
	switch p.DataFormat {
	case constant.ImportDataFormatCsv:
		if p.CSVConfig == nil {
			return nil, errors.New("CSV format configuration is missing")
		}
		ioWorkers := worker.NewPool(ctx, 5, "preview_local")

		cfg := config.MydumperRuntime{
			CSV: *p.CSVConfig,
		}

		charsetConvertor, err := mydump.NewCharsetConvertor(cfg.DataCharacterSet, cfg.DataInvalidCharReplace)
		if err != nil {
			return nil, errors.Annotate(err, "failed to new charset convertor")
		}
		parser, err = mydump.NewCSVParser(
			ctx,
			&cfg.CSV,
			p.File,
			int64(config.ReadBlockSize),
			ioWorkers,
			cfg.CSV.Header,
			charsetConvertor,
		)
		if err != nil {
			return nil, errors.Annotate(err, "failed to create csv parser")
		}
	default:
		return nil, errors.Errorf("unsupported data format: %v", p.DataFormat)
	}
	return parser, nil
}

func (p *LocalImportPreviewer) doSingleTablePreview(
	ctx context.Context,
	fileMapping *PreviewFileMapping,
	limitCount uint,
) (*agentpb.Table, error) {
	logger := log.WithContextL(ctx)
	parser, err := p.newParser(ctx)
	if err != nil {
		return nil, errors.Annotate(err, "create source data parser error")
	}

	defer parser.Close()

	result := &agentpb.Table{
		SchemaName: fileMapping.DatabaseName,
		TableName:  fileMapping.TableName,
		RowValues:  make([]*agentpb.Row, limitCount),
	}

	var rowDataTypes []*ptypes.FieldType
	for i := 0; i < int(limitCount); i++ {
		if err := parser.ReadRow(); err != nil {
			if errors.Cause(err) == io.EOF {
				result.RowValues = result.RowValues[:i]
				break
			}
			if strings.Contains(err.Error(), "syntax error") {
				return nil, &ErrCSVSyntax{err: err}
			}

			logger.Error("failed to read row in parser", zap.Error(err))
			return nil, errors.Annotate(err, "failed to read row in parser")
		}

		readRow := parser.LastRow().Row
		writeRow := &agentpb.Row{ColumnValues: make([]string, 0, len(readRow))}
		if len(rowDataTypes) == 0 {
			rowDataTypes = make([]*ptypes.FieldType, len(readRow))
		}
		for j, datum := range readRow {
			columnVal, err := datum.ToString()
			if err != nil {
				return nil, errors.Annotate(err, "failed to get string from a datum")
			}
			writeRow.ColumnValues = append(writeRow.ColumnValues, str_utils.ForceValidUTF8(columnVal))
			fp := inferDataType(datum, writeRow.ColumnValues[j])

			if len(rowDataTypes) <= j {
				logger.Warn(
					"the data num of one row is greater than previous row",
					zap.String("row", humanize.Ordinal(i+1)),
				)
				return nil, &ErrCSVRowOverflow{RowNum: i}
			}
			if rowDataTypes[j] == nil {
				rowDataTypes[j] = fp
			} else {
				rowDataTypes[j] = types.AggFieldType([]*types.FieldType{rowDataTypes[j], fp})
			}
		}

		result.RowValues[i] = writeRow
	}
	for _, tp := range rowDataTypes {
		result.RowDataTypes = append(result.RowDataTypes, strings.ToUpper(tp.CompactStr()))
	}
	if len(parser.Columns()) > maxColumnPreview {
		return nil, &ErrTooManyColumns{}
	}

	if len(parser.Columns()) != 0 {
		result.RowHeaders = parser.Columns()
		if len(parser.Columns()) != len(rowDataTypes) {
			logger.Error(
				"the data header num does not equal to data type.",
				zap.Int("data type num", len(rowDataTypes)),
			)
			return nil, &ErrHeaderDataColumnMismatch{
				HeaderColumnsCount: len(parser.Columns()),
				DataColumnsCount:   len(rowDataTypes),
			}
		}
	} else {
		result.RowHeaders = make([]string, len(rowDataTypes))
	}

	return result, nil
}

// inferDataType infer the type by a datum and its string
// return the type and priority.
func inferDataType(strDatum types.Datum, str string) *ptypes.FieldType {
	var (
		ft  *ptypes.FieldType
		err error
		sc  = stmtctx.NewStmtCtxWithTimeZone(time.Local)
	)

	contain := func(s string, ru rune) bool {
		for _, ch := range s {
			if ch == ru {
				return true
			}
		}
		return false
	}

	// bigint
	ft = ptypes.NewFieldType(mysql.TypeLonglong)
	_, err = strDatum.ConvertTo(sc.TypeCtx(), ft)
	if err == nil && !contain(str, '.') {
		return ft
	}

	// double
	ft = ptypes.NewFieldType(mysql.TypeDouble)
	_, err = strDatum.ConvertTo(sc.TypeCtx(), ft)
	if err == nil {
		return ft
	}

	// date(0000-00-00)
	ft = ptypes.NewFieldType(mysql.TypeDate)
	_, err = strDatum.ConvertTo(sc.TypeCtx(), ft)
	if err == nil && !contain(str, ':') {
		return ft
	}

	// time(00:00:00)
	ft = ptypes.NewFieldType(mysql.TypeDuration)
	_, err = strDatum.ConvertTo(sc.TypeCtx(), ft)
	if err == nil && !contain(str, '-') {
		return ft
	}

	// datetime(0000-00-00 00:00:00)
	ft = ptypes.NewFieldType(mysql.TypeDatetime)
	_, err = strDatum.ConvertTo(sc.TypeCtx(), ft)
	if err == nil {
		return ft
	}

	// json
	ft = ptypes.NewFieldType(mysql.TypeJSON)
	_, err = strDatum.ConvertTo(sc.TypeCtx(), ft)
	if err == nil {
		return ft
	}

	s, _ := strDatum.ToString()
	length := utf8.RuneCountInString(s)
	if length > 64*1024 {
		return ptypes.NewFieldType(mysql.TypeLongBlob)
	}
	if length > 512 {
		return ptypes.NewFieldType(mysql.TypeBlob)
	}
	// varchar(512)
	ft = ptypes.NewFieldType(mysql.TypeVarchar)
	ft.SetFlen(512)
	return ft
}

func PreviewLocalImport(
	ctx context.Context,
	f io.ReadCloser,
	csvConfig *config.CSVConfig,
	format constant.DataFormat,
) (*agentpb.Schema, bizerr.BizError) {
	defer func() {
		_ = f.Close()
	}()
	logger := log.WithContextL(ctx)

	file, err := io.ReadAll(f)
	if err != nil {
		logger.Error("ReadAll failed", zap.Error(err))
		return nil, bizerr.New(errorcode.ErrCodeInternalError, errorcode.InternalErrorMessage)
	}
	previewer := &LocalImportPreviewer{
		CSVConfig:  csvConfig,
		DataFormat: format,
		PreviewFileMappings: []*PreviewFileMapping{
			{
				DatabaseName: "target_db",    // this is a fake name.
				TableName:    "target_table", // this is a fake name.
			},
		},
		File: NewByteReadSeekCloser(file),
	}
	dbPreviewResults, err := previewer.Preview(ctx, uint(defaultPreviewLimit))
	if err != nil {
		logger.Error("preview table data has errors", zap.Error(err))
		switch concreteErr := errors.Cause(err).(type) {
		case *ErrCSVSyntax:
			return nil, bizerr.New(bizerr.ErrCodeParams, "%s", err.Error())
		case *ErrCSVRowOverflow:
			return nil, bizerr.New(bizerr.ErrCodeParams, "The CSV file is invalid, your %s row data has overflowed.", humanize.Ordinal(concreteErr.RowNum+1))
		case *ErrHeaderDataColumnMismatch:
			return nil, bizerr.New(bizerr.ErrCodeParams, "Invalid CSV file. Your row data has %d columns, but your headers have %d columns. Please modify the CSV files to ensure them match.",
				concreteErr.DataColumnsCount, concreteErr.HeaderColumnsCount)
		case *ErrTooManyColumns:
			return nil, bizerr.New(bizerr.ErrCodeParams, "The CSV file has too many columns. Please import data file from cloud storage.")
		default:
			return nil, bizerr.New(errorcode.ErrCodeInternalError, "Internal error when previewing table data. Please seek support.")
		}
	}
	if len(dbPreviewResults) < 1 {
		return nil, bizerr.New(bizerr.ErrCodeParams, "No data found in the file.")
	}
	dbPreviewResult := dbPreviewResults[0]
	return dbPreviewResult, nil
}

type ByteReadSeekCloser struct {
	reader *bytes.Reader
}

func NewByteReadSeekCloser(b []byte) *ByteReadSeekCloser {
	return &ByteReadSeekCloser{
		reader: bytes.NewReader(b),
	}
}

func (r *ByteReadSeekCloser) Close() error {
	return nil
}

func (r *ByteReadSeekCloser) Read(b []byte) (n int, err error) {
	return r.reader.Read(b)
}

func (r *ByteReadSeekCloser) Seek(offset int64, whence int) (int64, error) {
	return r.reader.Seek(offset, whence)
}