package console

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/console"
	"github.com/tidbcloud/serverless-service/internal/errorcode"
	globalserverlessv1 "github.com/tidbcloud/tidb-management-service/api/spec/global/serverless/v1"
	tidbmgmtservicev1 "github.com/tidbcloud/tidb-management-service/api/spec/tidb_mgmt_service/v1"
	"go.uber.org/zap"
)

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

func (s *importService) validateCluster(
	ctx context.Context,
	clusterID string,
	logger *zap.Logger,
) (*tidbmgmtservicev1.Cluster, bizerr.BizError) {
	cID, err := strconv.ParseUint(clusterID, 10, 64)
	if err != nil {
		return nil, bizerr.New(bizerr.ErrCodeParams, "invalid cluster id")
	}
	cluster, bizErr := s.mgmtSvc.GetCluster(ctx, cID, tidbmgmtservicev1.ClusterFilter_CLUSTER_VIEW_BASIC)
	if bizErr != nil {
		logger.Error("GetClusterByID failed", zap.String("cluster_id", clusterID), zap.Error(bizErr))
		return nil, bizErr
	}

	if cluster.GetServerlessCluster().
		GetUserStatus() !=
		globalserverlessv1.UserClusterStatus_USER_CLUSTER_STATUS_NORMAL {
		return nil, bizerr.New(
			errorcode.ErrCodeClusterNotRunning,
			"Cluster ID %s is not in normal state, please try again later.",
			clusterID,
		)
	}
	return cluster, nil
}

func validateSourceCheckReq(req *console.ValidateImportReq) bizerr.BizError {
	if req.GetSource().Type == console.ImportSourceType_LOCAL {
		return bizerr.New(bizerr.ErrCodeParams, "local import does not support this operation")
	}
	return validateSource(req.GetSource())
}

func validatePrecheckReq(req *console.PrecheckReq) bizerr.BizError {
	if req.GetImportOptions() == nil {
		return bizerr.New(bizerr.ErrCodeParams, "missing required importOptions")
	}
	if req.GetImportOptions().GetFileType() == console.ImportFileType_UNSPECIFIED {
		return bizerr.New(bizerr.ErrCodeParams, "unsupported file type")
	}
	if req.GetSource().Type == console.ImportSourceType_LOCAL {
		return bizerr.New(bizerr.ErrCodeParams, "local import does not support this operation")
	}

	return validateSource(req.GetSource())
}

func validateImportReq(ctx context.Context, req *console.CreateImportReq) bizerr.BizError {
	bizError := validateSource(req.GetSource())
	if bizError != nil {
		return bizError
	}

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

	return nil
}

func validateSource(source *console.ImportSource) bizerr.BizError {
	if source == nil {
		return bizerr.New(bizerr.ErrCodeParams, "missing required parameter")
	}
	switch source.Type {
	case console.ImportSourceType_LOCAL:
		if source.GetLocal() == nil {
			return bizerr.New(bizerr.ErrCodeParams, "missing required parameter")
		}
		if source.GetLocal().GetUploadId() == "" {
			return bizerr.New(bizerr.ErrCodeParams, "parameter upload_id is empty")
		}
		if source.GetLocal().GetTargetDatabase() == "" {
			return bizerr.New(bizerr.ErrCodeParams, "parameter target_database is empty")
		}
		if source.GetLocal().GetTargetTable() == "" {
			return bizerr.New(bizerr.ErrCodeParams, "parameter target_table is empty")
		}
	case console.ImportSourceType_S3:
		if source.GetS3() == nil {
			return bizerr.New(bizerr.ErrCodeParams, "missing required parameter")
		}
		if source.GetS3().GetUri() == "" {
			return bizerr.New(bizerr.ErrCodeParams, "parameter uri is empty")
		}
		if validateUrlSchema(source.GetS3().GetUri(), "s3") != nil {
			return bizerr.New(bizerr.ErrCodeParams, "invalid S3 uri")
		}
		if source.GetS3().GetAuthType() == console.ImportS3AuthType_ACCESS_KEY {
			if source.GetS3().GetAccessKey() == nil || source.GetS3().GetAccessKey().GetSecret() == "" ||
				source.GetS3().GetAccessKey().GetId() == "" {
				return bizerr.New(bizerr.ErrCodeParams, "missing required access key")
			}
		} else if source.GetS3().GetAuthType() == console.ImportS3AuthType_ROLE_ARN {
			if source.GetS3().GetRoleArn() == "" {
				return bizerr.New(bizerr.ErrCodeParams, "missing required role arn")
			}
		} else {
			return bizerr.New(bizerr.ErrCodeParams, "unsupported auth type")
		}
	case console.ImportSourceType_GCS:
		if source.GetGcs() == nil {
			return bizerr.New(bizerr.ErrCodeParams, "missing required parameter")
		}
		if source.GetGcs().GetUri() == "" {
			return bizerr.New(bizerr.ErrCodeParams, "parameter uri is empty")
		}
		if validateUrlSchema(source.GetGcs().GetUri(), "gs", "gcs") != nil {
			return bizerr.New(bizerr.ErrCodeParams, "invalid GCS uri")
		}
		if source.GetGcs().GetAuthType() == console.ImportGcsAuthType_SERVICE_ACCOUNT_KEY {
			if source.GetGcs().GetServiceAccountKey() == "" {
				return bizerr.New(bizerr.ErrCodeParams, "missing required credentials")
			}
		} else {
			return bizerr.New(bizerr.ErrCodeParams, "unsupported auth type")
		}
	case console.ImportSourceType_AZURE_BLOB:
		if source.GetAzureBlob() == nil {
			return bizerr.New(bizerr.ErrCodeParams, "missing required parameter")
		}
		if source.GetAzureBlob().GetAuthType() == console.ImportAzureBlobAuthType_SAS_TOKEN {
			if source.GetAzureBlob().GetUri() == "" {
				return bizerr.New(bizerr.ErrCodeParams, "missing required uri")
			}
			if validateUrlSchema(source.GetAzureBlob().GetUri(), "azure", "https") != nil {
				return bizerr.New(bizerr.ErrCodeParams, "invalid Azure Blob uri")
			}
		} else {
			return bizerr.New(bizerr.ErrCodeParams, "unsupported auth type")
		}
	case console.ImportSourceType_S3_COMPATIBLE:
		if source.GetS3Compatible() == nil {
			return bizerr.New(bizerr.ErrCodeParams, "missing required S3 Compatible source")
		}
		if source.GetS3Compatible().GetUri() == "" {
			return bizerr.New(bizerr.ErrCodeParams, "parameter uri is empty")
		}
		if validateUrlSchema(source.GetS3Compatible().GetUri(), "s3") != nil {
			return bizerr.New(bizerr.ErrCodeParams, "invalid uri")
		}
		if source.GetS3Compatible().GetAuthType() == console.ImportS3CompatibleAuthType_ACCESS_KEY {
			if source.GetS3Compatible().GetAccessKey() == nil ||
				source.GetS3Compatible().GetAccessKey().GetSecret() == "" ||
				source.GetS3Compatible().GetAccessKey().GetId() == "" {
				return bizerr.New(bizerr.ErrCodeParams, "missing required access key")
			}
		} else {
			return bizerr.New(bizerr.ErrCodeParams, "unsupported auth type")
		}
		if source.GetS3Compatible().GetEndpoint() != "" &&
			validateUrlSchema(source.GetS3Compatible().GetEndpoint(), "https") != nil {
			return bizerr.New(bizerr.ErrCodeParams, "S3 endpoint must start with https://")
		}
	case console.ImportSourceType_OSS:
		if source.GetOss() == nil {
			return bizerr.New(bizerr.ErrCodeParams, "missing required OSS source")
		}
		if source.GetOss().GetUri() == "" {
			return bizerr.New(bizerr.ErrCodeParams, "parameter uri is empty")
		}
		if validateUrlSchema(source.GetOss().GetUri(), "oss") != nil {
			return bizerr.New(bizerr.ErrCodeParams, "invalid uri")
		}
		if source.GetOss().GetAuthType() == console.ImportOSSAuthType_ACCESS_KEY {
			if source.GetOss().GetAccessKey() == nil || source.GetOss().GetAccessKey().GetSecret() == "" ||
				source.GetOss().GetAccessKey().GetId() == "" {
				return bizerr.New(bizerr.ErrCodeParams, "missing required access key")
			}
		} else if source.GetOss().GetAuthType() == console.ImportOSSAuthType_ROLE_ARN {
			if source.GetOss().GetRoleArn() == "" {
				return bizerr.New(bizerr.ErrCodeParams, "missing required role arn")
			}
		} else {
			return bizerr.New(bizerr.ErrCodeParams, "unsupported auth type")
		}
	default:
		return bizerr.New(bizerr.ErrCodeParams, "unsupported source 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 *console.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
}