package errhandler

import (
	"fmt"
	"regexp"
	"strings"

	"github.com/tidbcloud/dataflow-service/internal/db/entity"
)

type CommonImportErr uint

const (
	ErrNotCommonErr CommonImportErr = iota
	ErrAccessDenied
	ErrBuckeRegionError
	ErrRegionUnbalanced
	ErrPreCheckFailed
	ErrSyntaxError
	ErrSchemaNotExists
	ErrUnknownColumns
	ErrChecksumMismatch
	ErrNoFiles
	ErrInvalidTablePattern
	ErrClusterErr
	ErrShareConfigLoadError
	ErrCreateSchema
	ErrCheckCsvHeader
	ErrTikvLowSpace
	ErrChecksumErr
	ErrReadSourceFile
	ErrRemoteWorker
	ErrTooManyConflict
	ErrInvalidSchemaFile
)

type prettyError struct {
	CommonErr CommonImportErr
	Detail    string
}

var msg2CommonErr = map[string]CommonImportErr{
	"AccessDenied":                       ErrAccessDenied,
	"BucketRegionError":                  ErrBuckeRegionError,
	"Region distribution is unbalanced":  ErrRegionUnbalanced,
	"ErrPreCheckFailed":                  ErrPreCheckFailed,
	"ErrSchemaNotExists":                 ErrSchemaNotExists,
	"ErrInvalidSchemaFile":               ErrInvalidSchemaFile,
	"syntax error":                       ErrSyntaxError,
	"unknown columns in header":          ErrUnknownColumns,
	"checksum mismatched":                ErrChecksumMismatch,
	"doesn't exist or contains no files": ErrNoFiles,
	"missing table pattern":              ErrInvalidTablePattern,
	"wrong table pattern":                ErrInvalidTablePattern,
	"ErrCreateSchema":                    ErrCreateSchema,
	"The available disk of TiKV":         ErrTikvLowSpace,

	"The number of conflict errors exceeds the threshold configured by": ErrTooManyConflict,
	// cluster unavailable errors
	// mask the details of the internal error.
	"TiKV server timeout":               ErrClusterErr,
	"cluster is deleted":                ErrClusterErr,
	"Region is unavailable":             ErrChecksumErr,
	"TiKV server is busy":               ErrClusterErr,
	"PD server timeout":                 ErrClusterErr,
	"TiKV server reports stale command": ErrClusterErr,
	"SharedConfigLoadError":             ErrShareConfigLoadError,

	// the following errors is encountered by serverless
	"encode kv error(.|\n)*?connection reset by peer": ErrReadSourceFile,
	"failed to send request to remote worker":         ErrRemoteWorker,
}

func (e CommonImportErr) String() string {
	switch e {
	case ErrNotCommonErr:
		return "NotCommonErr"
	case ErrAccessDenied:
		return "AccessDenied"
	case ErrBuckeRegionError:
		return "BucketRegionError"
	case ErrRegionUnbalanced:
		return "RegionUnbalanced"
	case ErrPreCheckFailed:
		return "PreCheckFailed"
	case ErrSyntaxError:
		return "SyntaxError"
	case ErrSchemaNotExists:
		return "SchemaNotExists"
	case ErrInvalidSchemaFile:
		return "InvalidSchemaFile"
	case ErrUnknownColumns:
		return "UnknownColumns"
	case ErrChecksumMismatch:
		return "ChecksumMismatch"
	case ErrNoFiles:
		return "FileNotExists"
	case ErrInvalidTablePattern:
		return "InvalidTablePattern"
	case ErrClusterErr:
		return "ClusterErr"
	case ErrShareConfigLoadError:
		return "AWSConfigLoadError"
	case ErrCreateSchema:
		return "CreateSchemaError"
	case ErrTikvLowSpace:
		return "TikvLowSpace"
	case ErrTooManyConflict:
		return "TooManyConflict"
	case ErrChecksumErr:
		return "ChecksumError"
	case ErrReadSourceFile:
		return "ReadSourceFileError"
	case ErrRemoteWorker:
		return "InternalError"
	}
	return ""
}

// detailExtract extracts the detail from the error message.
// if the error message doesn't match any patterns, this function
// returns the original message.
func detailExtract(typ CommonImportErr, msg string, dataFormat entity.DataFormat) string {
	switch typ {
	case ErrNotCommonErr:
		return msg
	case ErrAccessDenied:
		shortErr := getStringAfter("unknown storage error: AccessDenied: ", msg)
		if shortErr != msg {
			return shortErr
		}
		shortErr = getStringAfter("AccessDenied: ", msg)
		if shortErr != msg {
			return shortErr
		}
		return msg
	case ErrBuckeRegionError:
		pattern := "the bucket is not in '.*' region at endpoint"
		r := regexp.MustCompile(pattern)
		if idxs := r.FindIndex([]byte(msg)); idxs != nil {
			begIdx := idxs[0] + len("the bucket is not in '")
			endIdx := begIdx
			for i := begIdx; i < len(msg); i++ {
				if msg[i] == '\'' {
					endIdx = i
					break
				}
			}
			if endIdx > begIdx {
				return "bucket not in `" + msg[begIdx:endIdx] + "`"
			}
		}
		return "bucket not in specific region"
	case ErrRegionUnbalanced:
		return "TiDB-Lightning pre-check failed: The region distribution is unbalanced. Please submit a ticket for assistance."
	case ErrPreCheckFailed:
		return getStringAfter("pre-check failed: ", msg)
	case ErrSchemaNotExists:
		return getStringAfter("\\[Lightning:Restore:ErrSchemaNotExists\\]", msg) + ". Please create it manually in TiDB first before retrying."
	case ErrInvalidSchemaFile:
		return getStringAfter("\\[Lightning:Loader:ErrInvalidSchemaFile\\]", msg) + ". Please check it before retrying."
	case ErrSyntaxError:
		handledMsg := getStringAfter("encode kv error in file ", msg)
		if handledMsg == msg {
			handledMsg = getStringAfter("\\[Lightning:PreCheck:ErrCheckCSVHeader\\]", msg)
		}
		return handledMsg
	case ErrUnknownColumns:
		finalColNames, tableName := extractUnknownColumnsForDisplay(msg)
		if len(finalColNames) <= 0 {
			return msg
		}

		var finalColStr string
		if len(finalColNames) > 3 {
			finalColNames = finalColNames[:3]
			finalColStr = strings.Join(finalColNames, ",") + ",..."
		} else {
			finalColStr = strings.Join(finalColNames, ",")
		}

		switch dataFormat {
		case entity.ImportDataFormatCsv:
			return fmt.Sprintf("unknown columns in CSV file header (%s) for table '%s'. Please check CSV file header line and re-import data. If the CSV file has a header line, ensure column names of the header line match the target table. If the file doesn't have a header line, disable \"With Header\" in \"CSV Configuration\".", finalColStr, tableName)
		case entity.ImportDataFormatParquet, entity.ImportDataFormatAuroraSnapshot:
			return fmt.Sprintf("The columns (%s) in the Parquet file header do not match the columns of the target table '%s'. Please review the Parquet file header, ensure that the column names match those of the target table, and re-import the data.", finalColStr, tableName)
		}
		return msg
	case ErrChecksumMismatch:
		return "Duplicate row detected. To resolve the issue, either resolve conflicts in primary keys or unique keys in the source data, or ensure the primary key or unique key definition for the target table are correct, and modify them if necessary before reimporting."
	case ErrNoFiles:
		return getStringAfter("\\[Lightning:Storage:ErrEmptySourceDir\\]", msg)
	case ErrInvalidTablePattern:
		return getStringAfter("\\[Lightning:Config:ErrInvalidConfig\\]", msg)
	case ErrClusterErr:
		return "Ensure the cluster is available, and retry import again."
	case ErrShareConfigLoadError:
		return getStringAfter("SharedConfigLoadError: ", msg)
	case ErrCreateSchema:
		return "It is recommended to create the table in TiDB manually and try again later. Details: " + getStringAfter("\\[Lightning:Restore:ErrCreateSchema\\]", msg)
	case ErrTikvLowSpace:
		return "Less than 10% of space is available in TiKV. Please clean up the space or increase the TiKV capacity and try again."
	case ErrTooManyConflict:
		return fmt.Sprintf("More than %s conflicts detected. Please check your data before re-importing.", getStringAfter("The number of conflict errors exceeds the threshold configured by `conflict.threshold`: ", msg))
	case ErrChecksumErr:
		return fmt.Sprintf("Data imported successfully, but an error occurred during checksum process (%s). Contact support for verification plan for relevant tables.", msg)
	case ErrReadSourceFile:
		return "Failed to read source file. Error encountered while accessing S3 file."
	case ErrRemoteWorker:
		return "An error occurred during import."
	}
	return msg
}

func extractUnknownColumnsForDisplay(msg string) ([]string, string) {
	re := regexp.MustCompile(`unknown columns in header \((.*)\) for table (.*)`)
	subMatch := re.FindSubmatch([]byte(msg))
	if subMatch == nil {
		return nil, ""
	}
	colNames := strings.Split(string(subMatch[1]), ",")
	tableName := string(subMatch[2])
	finalColLen := 0
	const maxColLen int = 50
	finalColNames := []string{}
	for _, colName := range colNames {
		trimedColName := strings.TrimSpace(colName)
		if len(trimedColName) <= 0 {
			continue
		}
		if finalColLen+len(trimedColName) > maxColLen {
			finalColNames = append(finalColNames, "...")
			break
		}
		finalColNames = append(finalColNames, trimedColName)
		finalColLen = finalColLen + len(trimedColName)
	}
	return finalColNames, strings.TrimSpace(tableName)
}

// getStringAfter returns the string after the given prefix.
// or return the original string.
func getStringAfter(pattern string, msg string) string {
	r := regexp.MustCompile(pattern)
	if idxs := r.FindIndex([]byte(msg)); idxs != nil {
		begIdx := idxs[1]
		return msg[begIdx:]
	}
	return msg
}

// NewPrettyError simplifies the error message returned from lightning.
// It removes detailed info and simply returns the type of errors.
func NewPrettyError(msg string, dataFormat entity.DataFormat) *prettyError {
	idx := strings.Index(msg, "github.com")
	if idx != -1 {
		msg = msg[:idx]
	}
	msg = strings.TrimSpace(msg)

	typ := ErrNotCommonErr
	for key := range msg2CommonErr {
		matched, _ := regexp.MatchString(key, msg)
		if matched {
			typ = msg2CommonErr[key]
			// ErrRegionUnbalanced is prior to ErrPreCheckFailed
			if typ == ErrPreCheckFailed && strings.Contains(msg, "Region distribution is unbalanced") {
				typ = ErrRegionUnbalanced
			}
			break
		}
	}
	return &prettyError{
		CommonErr: typ,
		Detail:    detailExtract(typ, msg, dataFormat),
	}
}

func (e *prettyError) Error() string {
	if e.CommonErr == ErrNotCommonErr {
		// return the original string
		return e.Detail
	}
	ret := e.CommonErr.String()
	if e.Detail != "" {
		ret += ": " + e.Detail
	}
	return ret
}

// the error from lightning contains the stack, so we trim it.
func PrettyMessage(msg string, dataFormat uint32) string {
	return NewPrettyError(msg, entity.DataFormat(dataFormat)).Error()
}