package v1beta1

import (
	"context"
	"net/url"
	"slices"
	"strconv"
	"strings"
	"unicode/utf8"

	"github.com/aws/aws-sdk-go-v2/aws"
	"github.com/pingcap/errors"
	"github.com/tidbcloud/pkgv2/bizerr"
	"github.com/tidbcloud/pkgv2/log"
	"github.com/tidbcloud/serverless-service/idl/pbgen/proto/v1beta1"
	"go.uber.org/zap"
)

var (
	defaultDelimiter = "\""
	defaultNullValue = "\\N"
)

func validateImportReq(ctx context.Context, req *v1beta1.CreateImportRequest) bizerr.BizError {
	if req.GetSource() == nil {
		return bizerr.New(bizerr.ErrCodeParams, "missing required parameter")
	}
	if req.GetSource().Type == v1beta1.ImportSourceType_LOCAL {
		if req.GetSource().GetLocal() == nil {
			return bizerr.New(bizerr.ErrCodeParams, "missing required parameter")
		}
		if req.GetSource().GetLocal().GetUploadId() == "" {
			return bizerr.New(bizerr.ErrCodeParams, "parameter upload_id is empty")
		}
		if req.GetSource().GetLocal().GetTargetDatabase() == "" {
			return bizerr.New(bizerr.ErrCodeParams, "parameter target_database is empty")
		}
		if req.GetSource().GetLocal().GetTargetTable() == "" {
			return bizerr.New(bizerr.ErrCodeParams, "parameter target_table is empty")
		}
	} else if req.GetSource().Type == v1beta1.ImportSourceType_S3 {
		if req.GetSource().GetS3() == nil {
			return bizerr.New(bizerr.ErrCodeParams, "missing required parameter")
		}
		uri := strings.TrimSpace(req.GetSource().GetS3().GetUri())
		if uri == "" {
			return bizerr.New(bizerr.ErrCodeParams, "parameter uri is empty")
		}
		if validateUrlSchema(uri, "s3") != nil {
			return bizerr.New(bizerr.ErrCodeParams, "invalid s3 uri")
		}
		if !strings.HasSuffix(uri, "/") {
			return bizerr.New(bizerr.ErrCodeParams, "s3 uri must end with '/'")
		}
		if req.GetSource().GetS3().GetAuthType() == v1beta1.ImportS3AuthType_ACCESS_KEY {
			if req.GetSource().GetS3().GetAccessKey() == nil || req.GetSource().GetS3().GetAccessKey().GetSecret() == "" ||
				req.GetSource().GetS3().GetAccessKey().GetId() == "" {
				return bizerr.New(bizerr.ErrCodeParams, "missing required access key")
			}
		} else if req.GetSource().GetS3().GetAuthType() == v1beta1.ImportS3AuthType_ROLE_ARN {
			if req.GetSource().GetS3().GetRoleArn() == "" {
				return bizerr.New(bizerr.ErrCodeParams, "missing required role arn")
			}
		} else {
			return bizerr.New(bizerr.ErrCodeParams, "unsupported auth type")
		}
	} else if req.GetSource().Type == v1beta1.ImportSourceType_GCS {
		if req.GetSource().GetGcs() == nil {
			return bizerr.New(bizerr.ErrCodeParams, "missing required parameter")
		}
		uri := strings.TrimSpace(req.GetSource().GetGcs().GetUri())
		if uri == "" {
			return bizerr.New(bizerr.ErrCodeParams, "parameter uri is empty")
		}
		if validateUrlSchema(uri, "gs", "gcs") != nil {
			return bizerr.New(bizerr.ErrCodeParams, "invalid gcs uri")
		}
		if !strings.HasSuffix(uri, "/") {
			return bizerr.New(bizerr.ErrCodeParams, "gcs uri must end with '/'")
		}
		if req.GetSource().GetGcs().GetAuthType() == v1beta1.ImportGcsAuthType_SERVICE_ACCOUNT_KEY {
			if req.GetSource().GetGcs().GetServiceAccountKey() == "" {
				return bizerr.New(bizerr.ErrCodeParams, "missing required credentials")
			}
		} else {
			return bizerr.New(bizerr.ErrCodeParams, "unsupported auth type")
		}
	} else if req.GetSource().Type == v1beta1.ImportSourceType_AZURE_BLOB {
		if req.GetSource().GetAzureBlob() == nil {
			return bizerr.New(bizerr.ErrCodeParams, "missing required parameter")
		}
		uri := strings.TrimSpace(req.GetSource().GetAzureBlob().GetUri())
		if uri == "" {
			return bizerr.New(bizerr.ErrCodeParams, "missing required Azure Blob URI")
		}
		if validateUrlSchema(uri, "azure", "https") != nil {
			return bizerr.New(bizerr.ErrCodeParams, "invalid Azure Blob URI")
		}
		if !strings.HasSuffix(uri, "/") {
			return bizerr.New(bizerr.ErrCodeParams, "Azure Blob URI must end with '/'")
		}
		if req.GetSource().GetAzureBlob().GetAuthType() == v1beta1.ImportAzureBlobAuthType_SAS_TOKEN {
			if req.GetSource().GetAzureBlob().GetSasToken() == "" {
				return bizerr.New(bizerr.ErrCodeParams, "missing sas token")
			}
		} else {
			return bizerr.New(bizerr.ErrCodeParams, "unsupported auth type")
		}
	} else if req.GetSource().Type == v1beta1.ImportSourceType_S3_COMPATIBLE {
		if req.GetSource().GetS3Compatible() == nil {
			return bizerr.New(bizerr.ErrCodeParams, "missing required parameter")
		}
		uri := strings.TrimSpace(req.GetSource().GetS3Compatible().GetUri())
		if uri == "" {
			return bizerr.New(bizerr.ErrCodeParams, "parameter uri is empty")
		}
		if validateUrlSchema(uri, "s3") != nil {
			return bizerr.New(bizerr.ErrCodeParams, "invalid uri")
		}
		if !strings.HasSuffix(uri, "/") {
			return bizerr.New(bizerr.ErrCodeParams, "uri must end with '/'")
		}
		if req.GetSource().GetS3Compatible().GetAuthType() == v1beta1.ImportS3CompatibleAuthType_ACCESS_KEY {
			if req.GetSource().GetS3Compatible().GetAccessKey() == nil || req.GetSource().GetS3Compatible().GetAccessKey().GetSecret() == "" ||
				req.GetSource().GetS3Compatible().GetAccessKey().GetId() == "" {
				return bizerr.New(bizerr.ErrCodeParams, "missing required access key")
			}
		} else {
			return bizerr.New(bizerr.ErrCodeParams, "unsupported auth type")
		}
		if req.GetSource().GetS3Compatible().GetEndpoint() != "" && validateUrlSchema(req.GetSource().GetS3Compatible().GetEndpoint(), "https") != nil {
			return bizerr.New(bizerr.ErrCodeParams, "S3 endpoint must start with https://")
		}
	} else if req.GetSource().Type == v1beta1.ImportSourceType_OSS {
		if req.GetSource().GetOss() == nil {
			return bizerr.New(bizerr.ErrCodeParams, "missing required parameter")
		}
		uri := strings.TrimSpace(req.GetSource().GetOss().GetUri())
		if uri == "" {
			return bizerr.New(bizerr.ErrCodeParams, "parameter uri is empty")
		}
		if validateUrlSchema(uri, "oss") != nil {
			return bizerr.New(bizerr.ErrCodeParams, "invalid uri")
		}
		if !strings.HasSuffix(uri, "/") {
			return bizerr.New(bizerr.ErrCodeParams, "uri must end with '/'")
		}
		if req.GetSource().GetOss().GetAuthType() == v1beta1.ImportOSSAuthType_ACCESS_KEY {
			if req.GetSource().GetOss().GetAccessKey() == nil || req.GetSource().GetOss().GetAccessKey().GetSecret() == "" ||
				req.GetSource().GetOss().GetAccessKey().GetId() == "" {
				return bizerr.New(bizerr.ErrCodeParams, "missing required access key")
			}
		} else {
			return bizerr.New(bizerr.ErrCodeParams, "unsupported auth type")
		}
	} else {
		return bizerr.New(bizerr.ErrCodeParams, "unsupported source type")
	}

	if req.GetImportOptions() == nil {
		return bizerr.New(bizerr.ErrCodeParams, "missing required parameter")
	}
	if req.GetImportOptions().GetFileType() == v1beta1.ImportFileType_CSV {
		if req.GetImportOptions().GetCsvFormat() == nil {
			req.GetImportOptions().Format = &v1beta1.ImportOptions_CsvFormat{
				CsvFormat: &v1beta1.CSVFormat{},
			}
		}
		if bizErr := validateCSVFormat(ctx, req.GetImportOptions().GetCsvFormat()); bizErr != nil {
			return bizErr
		}
	} else if req.GetImportOptions().GetFileType() == v1beta1.ImportFileType_UNSPECIFIED {
		return bizerr.New(bizerr.ErrCodeParams, "unsupported file type")
	}

	return nil
}

func validateUrlSchema(u string, schemas ...string) error {
	parsedUrl, err := url.Parse(u)
	if err != nil {
		return err
	}
	if !slices.Contains(schemas, parsedUrl.Scheme) {
		return bizerr.New(bizerr.ErrCodeParams, "invalid schema %s", parsedUrl.Scheme)
	}

	return nil
}

func validateCSVFormat(ctx context.Context, csvFormat *v1beta1.CSVFormat) bizerr.BizError {
	if csvFormat == nil {
		return nil
	}

	// defaults
	if csvFormat.Separator == "" {
		csvFormat.Separator = ","
	}
	if csvFormat.Delimiter == nil {
		csvFormat.Delimiter = &defaultDelimiter
	}
	if csvFormat.Null == nil {
		csvFormat.Null = &defaultNullValue
	}
	if csvFormat.BackslashEscape == nil {
		csvFormat.BackslashEscape = aws.Bool(true)
	}
	if csvFormat.Header == nil {
		csvFormat.Header = aws.Bool(true)
	}
	if csvFormat.TrimLastSeparator == nil {
		csvFormat.TrimLastSeparator = aws.Bool(false)
	}
	if csvFormat.NotNull == nil {
		csvFormat.NotNull = aws.Bool(false)
	}

	logger := log.WithContextL(ctx)
	actualDelimiter, err := restoreEscapes(*csvFormat.Delimiter)
	if err != nil {
		logger.Error("restore CSV delimiter failed", zap.Error(err), zap.String("delimiter", *csvFormat.Delimiter))
		return bizerr.New(bizerr.ErrCodeParams, "invalid CSV delimiter")
	}
	csvFormat.Delimiter = &actualDelimiter
	actualSeparator, err := restoreEscapes(csvFormat.Separator)
	if err != nil {
		logger.Error("restore CSV separator failed", zap.Error(err), zap.String("separator", csvFormat.Separator))
		return bizerr.New(bizerr.ErrCodeParams, "invalid CSV separator")
	}
	csvFormat.Separator = actualSeparator
	if actualDelimiter != "" && (strings.HasPrefix(actualSeparator, actualDelimiter) ||
		strings.HasPrefix(actualDelimiter, actualSeparator)) {
		return bizerr.New(
			bizerr.ErrCodeParams,
			"The CSV separator and delimiter cannot be the same, and neither can be a prefix of the other.",
		)
	}

	return nil
}

func restoreEscapes(str string) (string, error) {
	// https://docs.pingcap.com/tidb/stable/migrate-from-csv-using-tidb-lightning
	replaced := make([]rune, 0, len(str))
	s := []byte(str)
	r := 0
	for r < len(s) {
		if s[r] != '\\' {
			c, size := utf8.DecodeRune(s[r:])
			r += size
			replaced = append(replaced, c)
			continue
		}
		r++
		if r >= len(s) {
			return "", errors.New("escape sequence at end of string")
		}
		switch s[r] {
		default:
			return "", errors.Errorf("Expected valid escape code after \\, but got %q.", s[r])
		case ' ', '\t':
			return "", errors.Errorf("invalid escape: '\\%c'", s[r])
		case 'b':
			replaced = append(replaced, rune(0x0008))
			r++
		case 't':
			replaced = append(replaced, rune(0x0009))
			r++
		case 'n':
			replaced = append(replaced, rune(0x000A))
			r++
		case 'f':
			replaced = append(replaced, rune(0x000C))
			r++
		case 'r':
			replaced = append(replaced, rune(0x000D))
			r++
		case '"':
			replaced = append(replaced, rune(0x0022))
			r++
		case '\\':
			replaced = append(replaced, rune(0x005C))
			r++
		case 'u':
			// At this point, we know we have a Unicode escape of the form
			// `uXXXX` at [r, r+5). (Because the lexer guarantees this
			// for us.)
			escaped, err := asciiEscapeToUnicode(s[r+1 : r+5])
			if err != nil {
				return "", errors.Annotate(err, "ascii escape to unicode has error")
			}
			replaced = append(replaced, escaped)
			r += 5
		case 'U':
			// At this point, we know we have a Unicode escape of the form
			// `uXXXX` at [r, r+9). (Because the lexer guarantees this
			// for us.)
			escaped, err := asciiEscapeToUnicode(s[r+1 : r+9])
			if err != nil {
				return "", errors.Annotate(err, "ascii escape to unicode has error")
			}
			replaced = append(replaced, escaped)
			r += 9
		}
	}
	return string(replaced), nil
}

func asciiEscapeToUnicode(bs []byte) (rune, error) {
	s := string(bs)
	hex, err := strconv.ParseUint(strings.ToLower(s), 16, 32)
	if err != nil {
		return 0, errors.Annotatef(err, "Could not parse '%s' as a hexadecimal number, but the lexer claims it's OK", s)
	}
	if !utf8.ValidRune(rune(hex)) {
		return 0, errors.Annotatef(err, "Escaped character '\\u%s' is not valid UTF-8.", s)
	}
	return rune(hex), nil
}