package utils

import (
	"net/url"
	"strings"
)

const (
	sinkUriSchemaKafka = "kafka"
	sinkUriSchemaMySQL = "mysql"
	sinkUriSchemaS3    = "s3"

	kafkaPasswordKey = "sasl-password"
	s3AccessKey      = "access-key"
	s3SecretKey      = "secret-access-key"
	desensitizedStr  = "********"
)

func DesensitiseSinkUri(sinkUri string) string {
	u, err := url.Parse(sinkUri)
	if err != nil {
		return sinkUri
	}

	switch u.Scheme {
	case sinkUriSchemaKafka:
		return desensitiseSinkKafkaUri(sinkUri, u)
	case sinkUriSchemaMySQL:
		return desensitiseSinkMySQLUri(sinkUri, u)
	case sinkUriSchemaS3:
		return desensitiseSinkS3Uri(sinkUri, u)
	default:
		return sinkUri
	}
}

func desensitiseSinkKafkaUri(sinkUri string, u *url.URL) string {
	queryParams := u.Query()
	if queryParams.Has(kafkaPasswordKey) {
		sinkUri = strings.ReplaceAll(sinkUri, url.QueryEscape(queryParams.Get(kafkaPasswordKey)), desensitizedStr)
	}
	return sinkUri
}

func desensitiseSinkMySQLUri(sinkUri string, u *url.URL) string {
	password, ok := u.User.Password()
	if !ok {
		return sinkUri
	}
	return strings.ReplaceAll(sinkUri, url.QueryEscape(password), desensitizedStr)
}

func desensitiseSinkS3Uri(sinkUri string, u *url.URL) string {
	queryParams := u.Query()
	if queryParams.Has(s3AccessKey) {
		sinkUri = strings.ReplaceAll(sinkUri, queryParams.Get(s3AccessKey), desensitizedStr)
	}
	if queryParams.Has(s3SecretKey) {
		sinkUri = strings.ReplaceAll(sinkUri, url.QueryEscape(queryParams.Get(s3SecretKey)), desensitizedStr)
	}
	return sinkUri
}