package vo

import (
	"fmt"
	"regexp"
	"sort"
	"strings"

	"github.com/tidbcloud/serverless-service/internal/domain/constant"
)

type CommonImportErr uint

const (
	ErrNotCommonErr CommonImportErr = iota
	ErrS3AccessDenied
	ErrGcsAccessDenied
	ErrAzBlobAccessDenied
	ErrBucketRegionError
	ErrRegionUnbalanced
	ErrPreCheckFailed
	ErrTargetTablesNotEmpty
	ErrCsvHeaderConfigMismatch
	ErrSyntaxError
	ErrSchemaNotExists
	ErrUnknownColumns
	ErrChecksumMismatch
	ErrNoFiles
	ErrClusterDeleted
	ErrCreateSchema
	ErrChecksumErr
	ErrReadSourceFile
	ErrCastValue
	ErrSchemaDataMismatch
	ErrInternal
	ErrTooManyDuplicatedKeys
	ErrNoSuchTable
	ErrInvalidSchemaFile
	WarnDuplicatedUniqueKeys
	ErrNotSupportedParquetType
	ErrGeneratedColumnExpression
)

var (
	ignoreColumnsRe           = regexp.MustCompile(` or use tables\.ignoreColumns to ignore .*($|\n)`)
	unknownColumnsRe          = regexp.MustCompile(`unknown columns in header \((.*)\) for table (.*)`)
	tidbSchemaRe              = regexp.MustCompile("TiDB schema (`.*`\\.`.*`)")
	bucketRegionRe            = regexp.MustCompile("the bucket is not in '.*' region at endpoint")
	s3AccessDeniedRe          = regexp.MustCompile("unknown storage error: AccessDenied: ")
	s3AccessDenied1Re         = regexp.MustCompile("AccessDenied: ")
	googleAccessDeniedRe      = regexp.MustCompile("googleapi: Error 403: ")
	unknownErrRe              = regexp.MustCompile("unknown error: ")
	unknownStorageErrRe       = regexp.MustCompile("unknown storage error: ")
	preCheckFailedRe          = regexp.MustCompile("pre-check failed: ")
	targetTablesNotEmptyRe    = regexp.MustCompile(`table\(s\) .* are not empty`)
	csvHeaderConfigMismatchRe = regexp.MustCompile(
		`source csv files contains header row but .*mydumper\.csv\.header.* is false`,
	)
	errorSchemaNotExistsRe     = regexp.MustCompile(`\[Lightning:Restore:ErrSchemaNotExists]`)
	encodeErrRe                = regexp.MustCompile("encode kv error in file ")
	errCastValueRFCRe          = regexp.MustCompile(`\[(Import|Lightning:Restore):ErrCastValue]`)
	errCheckCsvHeaderRe        = regexp.MustCompile(`\[Lightning:PreCheck:ErrCheckCSVHeader]`)
	errEncodeKVCodeRe          = regexp.MustCompile(`\[Lightning:Restore:ErrEncodeKV]`)
	lightningPreCheckFailedRe  = regexp.MustCompile("tidb-lightning pre-check failed: ")
	errEmptyDirRe              = regexp.MustCompile(`\[Lightning:Storage:ErrEmptySourceDir]`)
	errCreateSchemaRe          = regexp.MustCompile(`\[Lightning:Restore:ErrCreateSchema]`)
	errRestoreTableRe          = regexp.MustCompile(`\[Lightning:Restore:ErrRestoreTable]`)
	errInvalidSchemaFileRe     = regexp.MustCompile(`\[Lightning:Loader:ErrInvalidSchemaFile]`)
	errUnknownRe               = regexp.MustCompile(`\[Lightning:Common:ErrUnknown]`)
	errListFilesRe             = regexp.MustCompile(`list file failed: `)
	schemaNotExistsDetailRe    = regexp.MustCompile("table `([^`]*)`\\.`[^`]*` schema not found")
	generateTableInfoFromSQLRe = regexp.MustCompile(`generate table info from SQL error:\s*(.*)$`)
	lightningErrorCodePrefixRe = regexp.MustCompile(`^\[Lightning:[^\]]+]\s*`)
)

type PrettyError struct {
	CommonErr CommonImportErr
	Detail    string
}

var orderedMsgRules = []struct {
	re  *regexp.Regexp
	typ CommonImportErr
}{
	{regexp.MustCompile("AccessDenied"), ErrS3AccessDenied},
	{regexp.MustCompile("googleapi: Error 403: "), ErrGcsAccessDenied},
	{regexp.MustCompile("ERROR CODE: AuthenticationFailed"), ErrAzBlobAccessDenied},
	{regexp.MustCompile("BucketRegionError"), ErrBucketRegionError},
	{regexp.MustCompile("Region distribution is unbalanced"), ErrRegionUnbalanced},
	{targetTablesNotEmptyRe, ErrTargetTablesNotEmpty},
	{csvHeaderConfigMismatchRe, ErrCsvHeaderConfigMismatch},
	{regexp.MustCompile("ErrSchemaNotExists"), ErrSchemaNotExists},
	{regexp.MustCompile("syntax error"), ErrSyntaxError},
	{regexp.MustCompile("unknown columns in header"), ErrUnknownColumns},
	{regexp.MustCompile("checksum mismatched"), ErrChecksumMismatch},
	{regexp.MustCompile("doesn't exist or contains no files"), ErrNoFiles},
	{regexp.MustCompile("Lightning:Config:ErrInvalidConfig"), ErrInternal},
	{regexp.MustCompile("ErrCreateSchema"), ErrCreateSchema},
	{regexp.MustCompile("failed to cast value"), ErrCastValue},
	{errCastValueRFCRe, ErrCastValue},
	{regexp.MustCompile("please check table schema"), ErrSchemaDataMismatch},
	{regexp.MustCompile(`\[types:1292]`), ErrCastValue},
	{regexp.MustCompile("Error 1146"), ErrNoSuchTable},
	{regexp.MustCompile("unknown value"), ErrNotSupportedParquetType},
	{regexp.MustCompile("generate table info from SQL error:"), ErrSyntaxError},
	{regexp.MustCompile("failed to evaluate generated column expression"), ErrGeneratedColumnExpression},
	{errInvalidSchemaFileRe, ErrInvalidSchemaFile},
	{regexp.MustCompile(`exists some duplicate (unique )?keys`), WarnDuplicatedUniqueKeys},
	{regexp.MustCompile("TiKV server timeout"), ErrInternal},
	{regexp.MustCompile("The available disk of TiKV"), ErrInternal},
	{regexp.MustCompile("failed to send request"), ErrInternal},
	{regexp.MustCompile(`chunk-\d+ not found`), ErrInternal},
	{regexp.MustCompile("region has overlap data"), ErrInternal},
	{regexp.MustCompile("cluster is deleted"), ErrClusterDeleted},
	{regexp.MustCompile("Region is unavailable"), ErrChecksumErr},
	{regexp.MustCompile("TiKV server is busy"), ErrInternal},
	{regexp.MustCompile("PD server timeout"), ErrInternal},
	{regexp.MustCompile("TiKV server reports stale command"), ErrInternal},
	{regexp.MustCompile("SharedConfigLoadError"), ErrInternal},
	{regexp.MustCompile("rpc error:"), ErrInternal},
	{regexp.MustCompile("too many duplicate key"), ErrTooManyDuplicatedKeys},
	{regexp.MustCompile("encode kv error(.|\n)*?connection reset by peer"), ErrReadSourceFile},
}

func (e CommonImportErr) String() string {
	switch e {
	case ErrNotCommonErr:
		return "NotCommonErr"
	case ErrS3AccessDenied:
		return "S3AccessDenied"
	case ErrGcsAccessDenied:
		return "GcsAccessDenied"
	case ErrAzBlobAccessDenied:
		return "AzBlobAccessDenied"
	case ErrBucketRegionError:
		return "BucketRegionError"
	case ErrRegionUnbalanced:
		return "RegionUnbalanced"
	case ErrPreCheckFailed:
		return "PreCheckFailed"
	case ErrTargetTablesNotEmpty:
		return "TargetTablesNotEmpty"
	case ErrCsvHeaderConfigMismatch:
		return "CsvHeaderConfigMismatch"
	case ErrSyntaxError:
		return "SyntaxError"
	case ErrSchemaNotExists:
		return "SchemaNotExists"
	case ErrUnknownColumns:
		return "UnknownColumns"
	case ErrChecksumMismatch:
		return "ChecksumMismatch"
	case ErrNoFiles:
		return "FileNotExists"
	case ErrClusterDeleted:
		return "ClusterDeleted"
	case ErrCreateSchema:
		return "CreateSchemaError"
	case ErrChecksumErr:
		return "ChecksumError"
	case ErrReadSourceFile:
		return "ReadSourceFileError"
	case ErrCastValue:
		return "CastValueError"
	case ErrSchemaDataMismatch:
		return "SchemaDataMismatch"
	case ErrInternal:
		return "InternalError"
	case ErrTooManyDuplicatedKeys:
		return "TooManyDuplicatedKeys"
	case ErrNoSuchTable:
		return "NoSuchTable"
	case ErrInvalidSchemaFile:
		return "InvalidSchemaFile"
	case WarnDuplicatedUniqueKeys:
		return "DuplicatedUniqueKeys"
	case ErrNotSupportedParquetType:
		return "NotSupportedParquetType"
	case ErrGeneratedColumnExpression:
		return "GeneratedColumnExpressionError"
	default:
		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 constant.DataFormat) string {
	switch typ {
	case ErrS3AccessDenied:
		shortErr := getStringAfter(s3AccessDeniedRe, msg)
		if shortErr != msg {
			return shortErr
		}
		shortErr = getStringAfter(s3AccessDenied1Re, msg)
		if shortErr != msg {
			return shortErr
		}
		return msg
	case ErrGcsAccessDenied:
		return getStringAfter(googleAccessDeniedRe, msg)
	case ErrAzBlobAccessDenied:
		msg = getStringAfter(unknownErrRe, msg)
		msg = getStringAfter(unknownStorageErrRe, msg)
		// extract the request method and URL with response status code, ignore raw response body.
		// https://github.com/Azure/azure-sdk-for-go/blob/sdk/azcore/v1.16.0/sdk/azcore/internal/exported/response_error.go#L127
		splitMsg := strings.Split(msg, "\n")
		msg = splitMsg[0] + ". " + splitMsg[2]
		return msg
	case ErrBucketRegionError:
		if idxes := bucketRegionRe.FindStringIndex(msg); idxes != nil {
			begIdx := idxes[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:
		msg = removeIgnoreColumnsHints(msg)
		return getStringAfter(preCheckFailedRe, msg)
	case ErrTargetTablesNotEmpty:
		return "Target table(s) are not empty. Truncate the target table(s) or import into new empty table(s), then retry."
	case ErrCsvHeaderConfigMismatch:
		return "Header row detected, but 'With Header' is disabled. Enable 'With Header' or remove the header row, then retry."
	case ErrSchemaNotExists:
		detail := getStringAfter(errorSchemaNotExistsRe, msg)
		subMatch := schemaNotExistsDetailRe.FindSubmatch([]byte(detail))
		if subMatch == nil {
			return detail
		}
		schemaName := string(subMatch[1])
		return fmt.Sprintf("Target schema '%s' does not exist. Create it in TiDB first, then retry import.", schemaName)
	case ErrSyntaxError:
		if subMatch := generateTableInfoFromSQLRe.FindSubmatch([]byte(msg)); subMatch != nil {
			tableName := strings.TrimSpace(string(subMatch[1]))
			if tableName != "" {
				return fmt.Sprintf(
					"Failed to parse table schema SQL for table '%s'. Please check the table schema SQL and retry.",
					tableName,
				)
			}
			return "Failed to parse table schema SQL. Please check the table schema SQL and retry."
		}
		handledMsg := getStringAfter(encodeErrRe, msg)
		if handledMsg == msg {
			handledMsg = getStringAfter(errCheckCsvHeaderRe, msg)
		}
		if strings.Contains(handledMsg, "cannot have consecutive fields without separator") {
			return "CSV format appears invalid. Check separator and quote settings, then retry."
		}
		return handledMsg
	case ErrCastValue:
		msg = strings.ReplaceAll(msg, "\n", ". ")
		msg = getStringAfter(errUnknownRe, msg)
		msg = getStringAfter(encodeErrRe, msg)
		msg = errEncodeKVCodeRe.ReplaceAllString(msg, "")
		msg = errCastValueRFCRe.ReplaceAllString(msg, "")
		msg = strings.TrimSpace(msg)
		if msg == "" {
			return "Please correct the source data and retry."
		}
		return msg + " Please correct the source data and retry."
	case ErrSchemaDataMismatch:
		msg = removeIgnoreColumnsHints(msg)
		msg = getStringAfter(lightningPreCheckFailedRe, msg)
		errDetailLines := strings.Split(msg, "\n")
		errTableNameMap := map[string]struct{}{}
		for _, errDetail := range errDetailLines {
			subMatch := tidbSchemaRe.FindSubmatch([]byte(errDetail))
			if subMatch != nil {
				tableName := string(subMatch[1])
				if _, ok := errTableNameMap[tableName]; !ok {
					errTableNameMap[tableName] = struct{}{}
				}
			}
		}
		errTableNames := []string{}
		for tblName := range errTableNameMap {
			errTableNames = append(errTableNames, tblName)
		}
		sort.Strings(errTableNames)
		var errDigestMsg string
		errDigestMsg += "affected tables: [" + strings.Join(errTableNames, ",") + "]; error details: "
		const digestErrCnt = 3
		if len(errDetailLines) > digestErrCnt {
			errDigestMsg += strings.Join(errDetailLines[:digestErrCnt], "; ")
			errDigestMsg += fmt.Sprintf(" ... and %d more errors", len(errDetailLines)-digestErrCnt)
		} else {
			errDigestMsg += strings.Join(errDetailLines, "; ")
		}
		return fmt.Sprintf(
			"Source data is inconsistent with the table schema: %s.  Please try to fix the source data and try again",
			errDigestMsg,
		)
	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 constant.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 constant.ImportDataFormatParquet, constant.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 "Please don't modify any data on the target table during the import process, otherwise a checksum mismatch error will lead to import failure. If there is no modification and the import fails, please contact support for assistance."
	case ErrNoFiles:
		return getStringAfter(errEmptyDirRe, msg)
	case ErrClusterDeleted:
		return "Cluster is deleted."
	case ErrCreateSchema:
		return "It is recommended to create the table in TiDB manually and try again later. Details: " + getStringAfter(
			errCreateSchemaRe,
			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 ErrInternal:
		return "An internal error occurred during import. Please contact support for assistance."
	case ErrTooManyDuplicatedKeys:
		return "There are too many duplicate primary keys or unique keys in the source data. Please check the source data and try again."
	case ErrNoSuchTable:
		return getStringAfter(errRestoreTableRe, msg)
	case ErrInvalidSchemaFile:
		return getStringAfter(errInvalidSchemaFileRe, msg)
	case WarnDuplicatedUniqueKeys:
		return msg
	case ErrNotSupportedParquetType:
		msg = getStringAfter(errListFilesRe, msg)
		return getStringAfter(unknownErrRe, msg)
	case ErrGeneratedColumnExpression:
		msg = getStringAfter(errUnknownRe, msg)
		msg = strings.TrimSpace(msg)
		return msg + ". Please check your generated column definitions and source data."
	case ErrNotCommonErr:
		return stripLightningErrorCode(msg)
	default:
		return "An internal error occurred during import. Please contact support for assistance."
	}
}

func extractUnknownColumnsForDisplay(msg string) ([]string, string) {
	subMatch := unknownColumnsRe.FindSubmatch([]byte(msg))
	if subMatch == nil {
		return nil, ""
	}
	colNames := strings.Split(string(subMatch[1]), ",")
	tableName := string(subMatch[2])
	finalColLen := 0
	const maxColLen int = 50
	var finalColNames []string
	for _, colName := range colNames {
		trimSpaceCol := strings.TrimSpace(colName)
		if trimSpaceCol == "" {
			continue
		}
		if finalColLen+len(trimSpaceCol) > maxColLen {
			finalColNames = append(finalColNames, "...")
			break
		}
		finalColNames = append(finalColNames, trimSpaceCol)
		finalColLen += len(trimSpaceCol)
	}
	return finalColNames, strings.TrimSpace(tableName)
}

// getStringAfter returns the string after the given prefix.
// or return the original string.
func getStringAfter(pattern *regexp.Regexp, msg string) string {
	if idxes := pattern.FindStringIndex(msg); idxes != nil {
		begIdx := idxes[1]
		return msg[begIdx:]
	}
	return msg
}

func NewPrettyError(msg string, dataFormat constant.DataFormat) *PrettyError {
	typ := ErrNotCommonErr
	for _, rule := range orderedMsgRules {
		if rule.re.MatchString(msg) {
			typ = rule.typ
			break
		}
	}

	if typ == ErrNotCommonErr && strings.Contains(msg, "ErrPreCheckFailed") {
		typ = ErrPreCheckFailed
	}
	return &PrettyError{
		CommonErr: typ,
		Detail:    detailExtract(typ, msg, dataFormat),
	}
}

func stripLightningErrorCode(msg string) string {
	return strings.TrimSpace(lightningErrorCodePrefixRe.ReplaceAllString(msg, ""))
}

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
}

// PrettyMessage the error from lightning contains the stack, so we trim it.
func PrettyMessage(msg string, dataFormat constant.DataFormat) string {
	if msg == "" {
		return ""
	}

	idx := strings.Index(msg, "github.com")
	if idx != -1 {
		msg = msg[:idx]
	}

	return SimplifyErrorMsg(strings.TrimSpace(msg), dataFormat)
}

// SimplifyErrorMsg simplifies the error message returned from lightning.
// It removes detailed info and simply returns the type of errors.
func SimplifyErrorMsg(msg string, dataFormat constant.DataFormat) string {
	return NewPrettyError(msg, dataFormat).Error()
}

// RemoveIgnoreColumnsHints removes the hints about ignore columns from the error message.
func removeIgnoreColumnsHints(text string) string {
	// keep the newline character if it exists
	return ignoreColumnsRe.ReplaceAllString(text, "${1}")
}