package precheck

import (
	"context"
	"regexp"
	"slices"
	"strconv"
	"strings"

	"github.com/juju/errors"
	"github.com/tidbcloud/pkgv2/bizerr"
	"github.com/tidbcloud/pkgv2/log"
	"go.uber.org/zap"
)

var expandVariablePattern = regexp.MustCompile(`\$(?:\$|[\pL\p{Nd}_]+|\{[\pL\p{Nd}_]+\})`)

// FileRouteRule is the rule for routing files.
type FileRouteRule struct {
	Pattern string
	Schema  string
	Table   string
	Type    string
}

// FileRouter provides some operations to apply a rule to route file path to target schema/table
type FileRouter interface {
	// Route apply rule to path. Return nil if path doesn't match route rule;
	// return error if path match route rule but the captured value for field is invalid
	Route(path string) (*RouteResult, error)
}

// NewFileRouter creates a new file router with the rule.
func NewFileRouter(ctx context.Context, cfg []*FileRouteRule) (FileRouter, bizerr.BizError) {
	res := make([]FileRouter, 0, len(cfg))
	p := regexRouterParser{}
	for _, c := range cfg {
		rule, err := p.Parse(ctx, c)
		if err != nil {
			return nil, err
		}
		res = append(res, rule)
	}
	return chainRouters(res), nil
}

type RegexRouter struct {
	pattern    *regexp.Regexp
	extractors []patExpander
}

// Route routes a file path to a source file type.
func (r *RegexRouter) Route(path string) (*RouteResult, error) {
	indexes := r.pattern.FindStringSubmatchIndex(path)
	if len(indexes) == 0 {
		return nil, nil
	}
	result := &RouteResult{}
	for _, e := range r.extractors {
		err := e.Expand(r.pattern, path, indexes, result)
		if err != nil {
			return nil, err
		}
	}
	result.FilePath = path
	return result, nil
}

type regexRouterParser struct{}

func (p regexRouterParser) Parse(ctx context.Context, r *FileRouteRule) (*RegexRouter, bizerr.BizError) {
	logger := log.WithContextL(ctx)
	rule := &RegexRouter{}
	if r.Pattern == "" {
		return nil, bizerr.New(bizerr.ErrCodeParams, "rule pattern must not be empty")
	}

	pattern, err := regexp.Compile(r.Pattern)
	if err != nil {
		logger.Error("failed to compile rule pattern", zap.String("pattern", r.Pattern), zap.Error(err))
		return nil, bizerr.New(bizerr.ErrCodeParams, "rule pattern is invalid")
	}
	rule.pattern = pattern

	err = p.parseFieldExtractor(rule, "type", r.Type, func(result *RouteResult, value string) error {
		result.Type = value
		return nil
	})
	if err != nil {
		logger.Error("failed to parse rule type", zap.String("pattern", r.Pattern), zap.Error(err))
		return nil, bizerr.New(bizerr.ErrCodeUnknown, "failed to parse rule type")
	}

	err = p.parseFieldExtractor(rule, "schema", r.Schema, func(result *RouteResult, value string) error {
		result.Schema = strings.ToLower(value)
		return nil
	})
	if err != nil {
		logger.Error("failed to parse rule schema", zap.String("pattern", r.Pattern), zap.Error(err))
		return nil, bizerr.New(bizerr.ErrCodeUnknown, "failed to parse rule schema")
	}

	// special case: when the pattern is for db schema, should not parse table name
	if r.Type != SchemaSchema {
		err = p.parseFieldExtractor(rule, "table", r.Table, func(result *RouteResult, value string) error {
			result.Name = strings.ToLower(value)
			return nil
		})
		if err != nil {
			logger.Error("failed to parse rule table", zap.String("pattern", r.Pattern), zap.Error(err))
			return nil, bizerr.New(bizerr.ErrCodeUnknown, "failed to parse rule table")
		}
	}

	return rule, nil
}

// parse each field extractor in `p.r` and set them to p.rule
func (p regexRouterParser) parseFieldExtractor(
	rule *RegexRouter,
	field,
	fieldPattern string,
	applyFn func(result *RouteResult, value string) error,
) error {
	// pattern is empty, return default rule
	if len(fieldPattern) == 0 {
		return errors.Errorf("field '%s' match pattern can't be empty", field)
	}

	// check and parse regexp template
	if err := p.checkSubPatterns(rule.pattern, fieldPattern); err != nil {
		return errors.Trace(err)
	}
	rule.extractors = append(rule.extractors, patExpander{
		template: fieldPattern,
		applyFn:  applyFn,
	})
	return nil
}

func (regexRouterParser) checkSubPatterns(pat *regexp.Regexp, t string) error {
	subPats := expandVariablePattern.FindAllString(t, -1)
	for _, subVar := range subPats {
		var tmplName string
		switch {
		case subVar == "$$":
			continue
		case strings.HasPrefix(subVar, "${"):
			tmplName = subVar[2 : len(subVar)-1]
		default:
			tmplName = subVar[1:]
		}
		if number, err := strconv.Atoi(tmplName); err == nil {
			if number > pat.NumSubexp() {
				return errors.Errorf("sub pattern capture '%s' out of range", subVar)
			}
		} else if slices.IndexFunc(pat.SubexpNames(), func(name string) bool {
			return name == tmplName
		}) < 0 {
			return errors.Errorf("invalid named capture '%s'", subVar)
		}

	}

	return nil
}

// patExpander extract string by expanding template with the regexp pattern
type patExpander struct {
	template string
	applyFn  func(result *RouteResult, value string) error
}

func (p *patExpander) Expand(pattern *regexp.Regexp, path string, matchIndex []int, result *RouteResult) error {
	value := pattern.ExpandString([]byte{}, p.template, path, matchIndex)
	return p.applyFn(result, string(value))
}

type chainRouters []FileRouter

func (c chainRouters) Route(path string) (*RouteResult, error) {
	for _, r := range c {
		res, err := r.Route(path)
		if err != nil {
			return nil, err
		}
		if res != nil {
			return res, nil
		}
	}
	return nil, nil
}

type RouteResult struct {
	Table
	Type     string
	FilePath string
}

type Table struct {
	// Schema is the name of the schema (database) containing this table.
	Schema string `toml:"db-name" json:"db-name" yaml:"db-name"`
	// Name is the unqualified table name.
	Name string `toml:"tbl-name" json:"tbl-name" yaml:"tbl-name"`
}

const (
	// SchemaSchema is the source type value for schema file for DB.
	SchemaSchema = "schema-schema"
	// TableSchema is the source type value for schema file for table.
	TableSchema = "table-schema"
	// TypeSQL is the source type value for sql data file.
	TypeSQL = "sql"
	// TypeCSV is the source type value for csv data file.
	TypeCSV = "csv"
	// TypeParquet is the source type value for parquet data file.
	TypeParquet = "parquet"
)