package loader

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

	"github.com/pingcap/errors"
	"github.com/pingcap/tidb/pkg/executor/importer"
	"github.com/pingcap/tidb/pkg/lightning/mydump"
	"github.com/pingcap/tidb/pkg/objstore"
	"github.com/pingcap/tidb/pkg/objstore/storeapi"
	"github.com/pingcap/tidb/pkg/util/stringutil"
	pkglog "github.com/tidbcloud/dataflow-service/dataflow-agent-ng/pkg/log"
	"go.uber.org/zap"
)

var (
	scanFilesOutOfLimit = errors.New("scan files out of limit")
	// ErrFileNotFound is returned when a single file path does not exist in storage.
	ErrFileNotFound = errors.New("file not found")

	// getStoreFn is the function used to create external storage from a URL.
	// It can be replaced in tests to inject in-memory storage.
	getStoreFn = func(ctx context.Context, url string) (storeapi.Storage, error) {
		return importer.GetSortStore(ctx, url)
	}
)

func parseFileType(path string) string {
	path = strings.ToLower(path)
	ext := filepath.Ext(path)
	// avoid duplicate compress extension
	if ext == ".gz" || ext == ".gzip" || ext == ".zstd" || ext == ".zst" || ext == ".snappy" {
		path = strings.TrimSuffix(path, ext)
		ext = filepath.Ext(path)
	}
	switch ext {
	case ".sql":
		return importer.DataFormatSQL
	case ".parquet":
		return importer.DataFormatParquet
	default:
		// if file do not contain file extension, use ".csv" as default format
		return importer.DataFormatCSV
	}
}

func getSourceType(format string) mydump.SourceType {
	switch format {
	case importer.DataFormatParquet:
		return mydump.SourceTypeParquet
	case importer.DataFormatDelimitedData, importer.DataFormatCSV:
		return mydump.SourceTypeCSV
	default:
		// DataFormatSQL
		return mydump.SourceTypeSQL
	}
}

func ScanDataFiles(ctx context.Context, url string, maxScanFiles int) ([]*mydump.SourceFileMeta, int64, error) {
	logger := pkglog.GetLogger(ctx).With(zap.String("url", url), zap.Int("maxScanFiles", maxScanFiles))
	logger.Info("scan data files start")
	u, err := objstore.ParseRawURL(url)
	if err != nil {
		return nil, 0, err
	}

	var fileNameKey string
	if objstore.IsLocal(u) {
		err := errors.New("local file is not supported")
		return nil, 0, err
	} else {
		fileNameKey = strings.Trim(u.Path, "/")
	}
	// try to find pattern error in advance
	_, err = filepath.Match(stringutil.EscapeGlobQuestionMark(fileNameKey), "")
	if err != nil {
		return nil, 0, errors.New("Glob pattern error: " + err.Error())
	}
	u.Path = ""
	s, err := getStoreFn(ctx, u.String())
	if err != nil {
		return nil, 0, err
	}

	var (
		totalSize  int64
		sourceType mydump.SourceType
	)
	dataFiles := []*mydump.SourceFileMeta{}
	// check glob pattern is present in filename.
	idx := strings.IndexAny(fileNameKey, "*[")
	// simple path when the path represent one file
	if idx == -1 {
		logger.With(zap.String("fileNameKey", fileNameKey)).Info("scan single file")
		// Use WalkDir to get file metadata instead of Open + Seek
		// This avoids issues with Azure Blob Storage Seek implementation
		var fileFound bool
		if err := s.WalkDir(ctx, &storeapi.WalkOption{ObjPrefix: fileNameKey, SkipSubDir: true},
			func(remotePath string, size int64) error {
				// Only process exact match
				if remotePath != fileNameKey {
					return nil
				}
				fileFound = true
				sourceType = getSourceType(parseFileType(remotePath))
				compressTp := mydump.ParseCompressionOnFileExtension(remotePath)
				fileMeta := mydump.SourceFileMeta{
					Path:        remotePath,
					FileSize:    size,
					Compression: compressTp,
					Type:        sourceType,
				}
				dataFiles = append(dataFiles, &fileMeta)
				totalSize = size
				return nil
			}); err != nil {
			return nil, 0, err
		}
		if !fileFound {
			return nil, 0, fmt.Errorf("%w: %s", ErrFileNotFound, fileNameKey)
		}
	} else {
		commonPrefix := fileNameKey[:idx]
		escapedPath := stringutil.EscapeGlobQuestionMark(fileNameKey)
		logger.With(zap.String("fileNameKey", fileNameKey), zap.String("commonPrefix", commonPrefix)).Info("scan multi files")
		allFiles := make([]mydump.RawFile, 0, 16)
		if err := s.WalkDir(ctx, &storeapi.WalkOption{ObjPrefix: commonPrefix, SkipSubDir: true},
			func(remotePath string, size int64) error {
				allFiles = append(allFiles, mydump.RawFile{Path: remotePath, Size: size})
				if maxScanFiles > 0 && len(allFiles) >= maxScanFiles {
					return scanFilesOutOfLimit
				}
				return nil
			}); err != nil {
			if !errors.ErrorEqual(err, scanFilesOutOfLimit) {
				return nil, 0, errors.New(errors.GetErrStackMsg(err) + "failed to walk dir")
			}
		}

		var err error
		var processedFiles []*mydump.SourceFileMeta
		var once sync.Once
		if processedFiles, err = mydump.ParallelProcess(ctx, allFiles, 2,
			func(ctx context.Context, f mydump.RawFile) (*mydump.SourceFileMeta, error) {
				// we have checked in LoadDataExec.Next
				//nolint: errcheck
				match, _ := filepath.Match(escapedPath, f.Path)
				logger.With(zap.String("path", f.Path), zap.String("fileNameKey", fileNameKey), zap.String("commonPrefix", commonPrefix), zap.Bool("match", match)).Info("filter multi files")
				if !match {
					return nil, nil
				}
				path, size := f.Path, f.Size
				// pick arbitrary one file to detect the format.
				once.Do(func() {
					sourceType = getSourceType(parseFileType(path))
				})
				compressTp := mydump.ParseCompressionOnFileExtension(path)
				fileMeta := mydump.SourceFileMeta{
					Path:        path,
					FileSize:    size,
					Compression: compressTp,
					Type:        sourceType,
				}
				return &fileMeta, nil
			}); err != nil {
			return nil, 0, err
		}
		// filter unmatch files
		for _, f := range processedFiles {
			if f != nil {
				dataFiles = append(dataFiles, f)
				totalSize += f.FileSize
			}
		}
	}
	logger.With(zap.String("fileNameKey", fileNameKey), zap.Int("totalFiles", len(dataFiles)), zap.Int64("totalSize", totalSize)).Info("filter multi files")
	return dataFiles, totalSize, nil
}