package fileparser

import (
	"context"
	"path/filepath"
	"strings"

	"github.com/pingcap/errors"
	"github.com/pingcap/tidb/br/pkg/storage"
	"github.com/pingcap/tidb/pkg/lightning/config"
	lnconfig "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"
	"github.com/tidbcloud/dataflow-service/idl/pbgen/proto/agent"
	"github.com/tidbcloud/dataflow-service/internal/db/entity"
	str_utils "github.com/tidbcloud/dataflow-service/pkg/string-utils"
)

type ParserBuilder struct {
	ctx        context.Context
	dataFormat entity.DataFormat
	csvConfig  *agent.CsvConfig
	opts       *storage.BackendOptions
}

func NewParserBuilder(
	ctx context.Context,
	dataFormat entity.DataFormat,
	csvConfig *agent.CsvConfig,
	opts *storage.BackendOptions,
) *ParserBuilder {
	return &ParserBuilder{
		ctx:        ctx,
		dataFormat: dataFormat,
		csvConfig:  csvConfig,
		opts:       opts,
	}
}

func (b *ParserBuilder) Build(sourceUri string) (mydump.Parser, error) {
	switch b.dataFormat {
	case entity.ImportDataFormatCsv:
		return NewCsvParser(b.ctx, sourceUri, b.csvConfig, b.opts)
	case entity.ImportDataFormatSql:
		return NewSQLParser(b.ctx, sourceUri, b.opts)
	default:
		return nil, errors.Errorf("data format %d need not precheck", b.dataFormat)
	}
}

func NewCsvParser(
	ctx context.Context,
	sourceUri string,
	csvConfig *agent.CsvConfig,
	opts *storage.BackendOptions,
) (mydump.Parser, error) {
	if csvConfig == nil {
		return nil, errors.New("CSV format configuration is missing")
	}

	reader, err := NewParserReader(ctx, sourceUri, opts)
	if err != nil {
		return nil, errors.Annotatef(err, "failed to open the path: %s", sourceUri)
	}
	ioWorkers := worker.NewPool(ctx, 5, "preview_csv")

	cfg := lnconfig.MydumperRuntime{
		CSV: lnconfig.CSVConfig{
			FieldsTerminatedBy: csvConfig.Separator,
			FieldsEnclosedBy:   csvConfig.Delimiter,
			BackslashEscape:    csvConfig.BackslashEscape,
			Header:             csvConfig.Header,
			HeaderSchemaMatch:  csvConfig.Header,
			TrimLastEmptyField: csvConfig.GetTrimLastSeparator(),
		},
	}
	if csvConfig.BackslashEscape {
		cfg.CSV.FieldsEscapedBy = "\\"
	}
	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, reader, int64(lnconfig.ReadBlockSize), ioWorkers, csvConfig.Header, charsetConvertor)
	if err != nil {
		return nil, errors.Annotate(err, "failed to create csv parser")
	}
	return parser, nil
}

func parseCompressionType(t string) (storage.CompressType, error) {
	switch strings.ToLower(strings.TrimSpace(t)) {
	case ".gz", ".gzip":
		return storage.Gzip, nil
	case ".snappy":
		return storage.Snappy, nil
	case ".zstd", ".zst":
		return storage.Zstd, nil
	case ".xz", ".lz4", ".lzo":
		return storage.NoCompression, errors.Errorf("unsupported compression type '%s'", t)
	case ".csv", ".parquet", ".sql":
		return storage.NoCompression, nil
	default:
		return storage.NoCompression, errors.Errorf("invalid type '%s'", t)
	}
}

func ReadRow(parser mydump.Parser) (*agent.Row, error) {
	if err := parser.ReadRow(); err != nil {
		return nil, errors.Annotate(err, "failed to read row in parser")
	}

	readRow := parser.LastRow().Row
	writeRow := &agent.Row{ColumnValues: make([]string, 0, len(readRow))}

	for _, 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))
	}

	return writeRow, nil
}

func NewParserReader(ctx context.Context, sourceUri string, opts *storage.BackendOptions) (storage.ExternalFileReader, error) {
	base := filepath.Base(sourceUri)
	dir := strings.TrimRight(sourceUri, base)

	storageBackend, err := storage.ParseBackend(dir, opts)
	if err != nil {
		return nil, errors.Annotate(err, "failed to parse external storage backend")
	}
	store, err := storage.New(ctx, storageBackend, &storage.ExternalStorageOptions{})
	if err != nil {
		return nil, errors.Annotate(err, "failed to new storage backend")
	}

	compression, err := parseCompressionType(filepath.Ext(base))
	if err != nil {
		return nil, err
	}
	return storage.WithCompression(store, compression, storage.DecompressConfig{
		ZStdDecodeConcurrency: 1,
	}).Open(ctx, base, nil)
}

func NewSQLParser(
	ctx context.Context,
	sourceUri string,
	opts *storage.BackendOptions,
) (mydump.Parser, error) {
	reader, err := NewParserReader(ctx, sourceUri, opts)
	if err != nil {
		return nil, errors.Annotatef(err, "failed to open the path: %s", sourceUri)
	}
	ioWorkers := worker.NewPool(ctx, 5, "sql")
	defaultSQLMode, _ := mysql.GetSQLMode(mysql.DefaultSQLMode)

	return mydump.NewChunkParser(ctx, defaultSQLMode, reader, int64(config.ReadBlockSize), ioWorkers), nil
}