package utils

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

	"github.com/tidbcloud/pkgv2/log"
	"github.com/tidbcloud/serverless-service/idl/pbgen/proto/console"
	"github.com/tidbcloud/serverless-service/internal/domain/constant"
	"go.uber.org/zap"
)

const (
	DefaultSQLFilePatternWithTargetTable            = `(?i)^%s.%s(?:(?:\.\d+)|(?:\.?(?:_\d+){3}))?\.sql(?:\.(\w+))?$`
	DefaultCSVFilePatternWithTargetTable            = `(?i)^%s.%s(?:(?:\.\d+)|(?:\.?(?:_\d+){3}))?\.csv(?:\.(\w+))?$`
	DefaultRawParquetFilePatternWithTargetTable     = `(?i)^%s.%s(?:(?:\.\d+)|(?:\.?(?:_\d+){3}))?(?:\.(?:snappy|gz|zst|zstd|gzip))?\.parquet$`
	DefaultAuroraSnapshotFilePatternWithTargetTable = `(?i)^%s/%s.%s/(?:[^/]*/)*(?:[a-z0-9\-_.]+\.parquet)$`

	DefaultDBSchemaFilePattern       = `(?i)^(?:[^/]*/)*([^/.]+)-schema-create\.sql(?:\.(\w+))?$`
	DefaultTableSchemaFilePattern    = `(?i)^(?:[^/]*/)*([^/.]+)\.(.*?)-schema\.sql(?:\.(\w+))?$`
	DefaultViewSchemaFilePattern     = `(?i)^(?:[^/]*/)*([^/.]+)\.(.*?)-schema-view\.sql(?:\.(\w+))?$`
	DefaultFilePattern               = `(?i)^(?:[^/]*/)*([^/.]+)\.([^/.]+?)(?:(?:\.\d+)|(?:\.?(?:_\d+){3}))?\.%s(?:\.(\w+))?$`
	DefaultAuroraSnapshotFilePattern = `(?i)^(?:[^/]*/)*([a-z0-9_]+)\.([a-z0-9_]+)/(?:[^/]*/)*(?:[a-z0-9\-_.]+\.parquet)$`
	DefaultRawParquetFilePattern     = `(?i)^(?:[^/]*/)*([^/.]+)\.([^/.]+?)(?:(?:\.\d+)|(?:\.?(?:_\d+){3}))?(?:\.(?:snappy|gz|zst|zstd|gzip))?\.parquet$`
)

var (
	DefaultSQLFilePattern = fmt.Sprintf(DefaultFilePattern, "sql")
	DefaultCSVFilePattern = fmt.Sprintf(DefaultFilePattern, "csv")
)

var (
	RegexpCsv            = regexp.MustCompile(`.*\.csv(\.(snappy|gz|lzo))?$`)
	RegexpSql            = regexp.MustCompile(`.*\.sql(\.(snappy|gz|lzo))?$`)
	RegexpParquet        = regexp.MustCompile(`.*(\.(snappy|gz|lzo))?\.parquet$`)
	RegexpAuroraSnapshot = regexp.MustCompile(`.*\.parquet$`)
)

var (
	DefaultCSVFileRegexp            = regexp.MustCompile(DefaultCSVFilePattern)
	DefaultSQLFileRegexp            = regexp.MustCompile(DefaultSQLFilePattern)
	DefaultRawParquetFileRegexp     = regexp.MustCompile(DefaultRawParquetFilePattern)
	DefaultAuroraSnapshotFileRegexp = regexp.MustCompile(DefaultAuroraSnapshotFilePattern)
)

type CustomMultiFileMatcher struct {
	dataFormat constant.DataFormat
}

func NewCustomMatcher(dataFormat constant.DataFormat) *CustomMultiFileMatcher {
	return &CustomMultiFileMatcher{
		dataFormat: dataFormat,
	}
}

// Match returns the matched uri
func (m *CustomMultiFileMatcher) Match(
	ctx context.Context,
	info *console.ImportTargetTableInfo,
	objectUris []string,
) (string, error) {
	for _, uri := range objectUris {
		match, err := matchCustomNamingRule(uri, info.CustomFile, m.dataFormat)
		if err != nil {
			return "", err
		}
		if match {
			return uri, nil
		}
	}

	return "", nil
}

func matchCustomNamingRule(fileName, customNamingRule string, dataFormat constant.DataFormat) (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 buildDefaultCustomNamingRuleRegexp(dataFormat constant.DataFormat) (*regexp.Regexp, error) {
	switch dataFormat {
	case constant.ImportDataFormatCsv:
		return RegexpCsv, nil
	case constant.ImportDataFormatSql:
		return RegexpSql, nil
	case constant.ImportDataFormatParquet:
		return RegexpParquet, nil
	case constant.ImportDataFormatAuroraSnapshot:
		return RegexpAuroraSnapshot, nil
	}
	return nil, fmt.Errorf("unsupported data format %d", dataFormat)
}

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()
}

// TiDB Lightning's FieldExtractor treats "$" as extractor syntax (for example, "$1").
// Escape "$" to "$$" so user-provided schema/table names are kept literal in file routing rules.
func EscapeFieldExtractor(v string) string {
	return strings.ReplaceAll(v, "$", "$$")
}

type DefaultMultiFileMatcher struct {
	dataFormat constant.DataFormat
}

func NewDefaultMultiFileMatcher(dataFormat constant.DataFormat) *DefaultMultiFileMatcher {
	return &DefaultMultiFileMatcher{
		dataFormat: dataFormat,
	}
}

func (m *DefaultMultiFileMatcher) Match(
	ctx context.Context,
	info *console.ImportTargetTableInfo,
	objectUris []string,
) (string, error) {
	for _, uri := range objectUris {
		match, err := MatchDefaultNamingRule(ctx, uri, info.TargetTable.Database, info.TargetTable.Table, m.dataFormat)
		if err != nil {
			return "", err
		}
		if match {
			return uri, nil
		}
	}

	return "", nil
}

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

	matchedStrs := re.FindStringSubmatch(fileName)
	if len(matchedStrs) < 3 {
		return false, nil
	}
	if matchedStrs[1] != schema || matchedStrs[2] != table {
		return false, nil
	}
	return true, nil
}

func buildDefaultNamingRuleRegexp(dataFormat constant.DataFormat) (*regexp.Regexp, error) {
	switch dataFormat {
	case constant.ImportDataFormatCsv:
		return DefaultCSVFileRegexp, nil
	case constant.ImportDataFormatSql:
		return DefaultSQLFileRegexp, nil
	case constant.ImportDataFormatParquet:
		return DefaultRawParquetFileRegexp, nil
	case constant.ImportDataFormatAuroraSnapshot:
		return DefaultAuroraSnapshotFileRegexp, nil
	}
	return nil, fmt.Errorf("unsupported data format %d", dataFormat)
}