package importcfg

import (
	"fmt"
	"regexp"
	"strings"
	"unicode/utf8"

	"github.com/pingcap/errors"
	"github.com/pingcap/log"
	"github.com/tidbcloud/dataflow-service/idl/pbgen/proto/dataflow"
	"go.uber.org/zap"
)

func buildDefaultNamingRuleRegexp(dataFormat uint32) (*regexp.Regexp, error) {
	switch dataflow.ImportDataFormat(dataFormat) {
	case dataflow.ImportDataFormat_CSV:
		return regexp.Compile(DefaultCSVFilePattern)
	case dataflow.ImportDataFormat_SQL:
		return regexp.Compile(DefaultSQLFilePattern)
	case dataflow.ImportDataFormat_PARQUET:
		return regexp.Compile(DefaultRawParquetFilePattern)
	case dataflow.ImportDataFormat_AURORA_SNAPSHOT:
		return regexp.Compile(DefaultAuroraSnapshotFilePattern)
	}
	return nil, errors.New(fmt.Sprintf("unsupported data format %d", dataFormat))
}

func buildDefaultCustomNamingRuleRegexp(dataFormat uint32) (*regexp.Regexp, error) {
	switch dataflow.ImportDataFormat(dataFormat) {
	case dataflow.ImportDataFormat_CSV:
		return regexp.Compile(`.*\.csv(\.(snappy|gz|zst|zstd|gzip))?$`)
	case dataflow.ImportDataFormat_SQL:
		return regexp.Compile(`.*\.sql(\.(snappy|gz|zst|zstd|gzip))?$`)
	case dataflow.ImportDataFormat_PARQUET:
		return regexp.Compile(`.*(\.(snappy|gz|zst|zstd|gzip))?\.parquet$`)
	case dataflow.ImportDataFormat_AURORA_SNAPSHOT:
		return regexp.Compile(DefaultAuroraSnapshotFilePattern)
	}
	return nil, errors.New(fmt.Sprintf("unsupported data format %d", dataFormat))
}

func MatchDefaultNamingRule(fileName, schema, table string, dataFormat uint32) bool {
	re, err := buildDefaultNamingRuleRegexp(dataFormat)
	if err != nil {
		log.L().Error("failed to compile regexp", zap.Error(err))
		return false
	}

	matchedStrs := re.FindStringSubmatch(fileName)
	if len(matchedStrs) < 3 {
		log.L().Warn("failed to match file name", zap.String("file name", fileName), zap.Strings("matched strs", matchedStrs))
		return false
	}
	if matchedStrs[1] != schema || matchedStrs[2] != table {
		log.L().Warn("schema or table not match", zap.String("schema", schema), zap.String("table", table), zap.Strings("matched strs", matchedStrs))
		return false
	}
	return true
}

func MatchCustomNamingRule(fileName, customNamingRule string, dataFormat uint32) (bool, error) {
	re, err := regexp.Compile(genCustomNamingRuleRegexp(customNamingRule))
	if err != nil {
		return false, err
	}
	defaultRe, err := buildDefaultCustomNamingRuleRegexp(dataFormat)
	if err != nil {
		return false, err
	}
	return re.MatchString(fileName) && defaultRe.MatchString(fileName), nil
}

func genCustomNamingRuleRegexp(wildcards string) string {
	// regexp syntax: https://en.wikipedia.org/wiki/Regular_expression
	var sb strings.Builder

	sb.WriteRune('^')

	cursor := 0
	maxlen := len(wildcards)

	for cursor < maxlen {
		r, sz := utf8.DecodeRuneInString(wildcards[cursor:])

		switch r {
		// defined in https://en.wikipedia.org/wiki/Glob_(programming)#Syntax
		case '{', '}', '(', ')', '^', '$', '.', '|', '+', '\\':
			sb.WriteString(`\`)
			sb.WriteRune(r)
		case '?':
			sb.WriteRune('.')
		case '*':
			sb.WriteString(".*")
		default:
			sb.WriteRune(r)
		}

		cursor += sz
	}

	sb.WriteRune('$')
	return sb.String()
}