package central

import (
	"context"
	"encoding/json"
	"fmt"
	"io"
	"net/http"
	"time"

	"github.com/pingcap/log"
	"go.uber.org/zap"

	"github.com/tidbcloud/clusterproxy/resolve"
	"github.com/tidbcloud/clusterproxy/route"
	"github.com/tidbcloud/clusterproxy/route/router"
)

const (
	connectionTypeKafka            = "kafka"
	connectionTypeMysql            = "mysql"
	connectionTypeS3               = "s3"
	connectionTypeGCS              = "gcs"
	connectionTypeAzureBlobStorage = "azblob"
	connectionTypeSnowflake        = "snowflake"
	connectionTypeRedshift         = "redshift"
	connectionTypeBigQuery         = "bigquery"
	connectionTypeCluster          = "cluster"
	connectionTypePulsar           = "pulsar"
)

type pulsarAuthType string

const (
	pulsarAuthTypeBasic pulsarAuthType = "basic"
	pulsarAuthTypeToken pulsarAuthType = "token"
)

const passwordMask = "****"

func init() {
	router.RegisterRoute(testConnectionRoute)
}

var testConnectionRoute = &route.Route{
	Name:    "test-connection",
	Path:    "test_connection",
	Version: route.V1,
	Schema: &route.Schema{
		PathRegex:       "^$",
		RequiredHeaders: []string{},
		JSONBodyModel:   &connectionModel{},
	},
	GetHandler: connectionGetHandler,
}

type kafkaItem struct {
	ReplicationFactor       uint32 `json:"replication_factor"`
	PartitionNum            uint32 `json:"partition_num"`
	SchemaRegistryEndpoints string `json:"schema_registry_endpoints"`
}

type s3Parameter struct {
	S3Uri           string `json:"s3_uri"`
	AccessKeyId     string `json:"access_key_id"`
	SecretAccessKey string `json:"secret_access_key"`
	RoleArn         string `json:"role_arn"`
	ExternalID      string `json:"external_id"`
}

type gcsParameter struct {
	GcsUrl                    string `json:"gcs_url"`
	Region                    string `json:"region"`
	ImpersonateServiceAccount string `json:"impersonate_service_account"`
}

type azureBlobStorageParameter struct {
	BlobUri  string `json:"blob_uri"`
	SASToken string `json:"sas_token"`
}

type pulsarParameter struct {
	ServiceURL         string         `json:"service_url"`
	AuthenticationType pulsarAuthType `json:"authentication_type"`
	Username           string         `json:"username"`
	Password           string         `json:"password"`
	Token              string         `json:"token"`
	ConnectionTimeout  uint32         `json:"connection_timeout"`
	Tenant             string         `json:"tenant"`
	Namespace          string         `json:"namespace"`
	Topic              string         `json:"topic"`
}

type bigqueryParameter struct {
	ProjectID string `json:"project_id"`
	DatasetID string `json:"dataset_id"`
}

// mysqlParameter holds MySQL-specific connection options used for tests.
type mysqlParameter struct {
	// PEM encoded bytes
	SSLCABytes   []byte `json:"ssl_ca_bytes"`
	SSLCertBytes []byte `json:"ssl_cert_bytes"`
	SSLKeyBytes  []byte `json:"ssl_key_bytes"`
}

type connectionModel struct {
	Type     string `json:"type"`
	User     string `json:"user"`
	Password string `json:"password"`
	// Notice: when it's Kafka, e.g. b1.example.com:9092,b2.example.com:9092...
	Url       string `json:"url"`
	Topic     string `json:"topic"`
	Database  string `json:"database"`
	AuthType  string `json:"auth_type"`
	EnableTls bool   `json:"enable_tls"`
	// MySQL specific parameters
	MysqlParam            mysqlParameter            `json:"mysql_param"`
	KafkaItem             kafkaItem                 `json:"kafka_item"`
	S3Param               s3Parameter               `json:"s3_param"`
	GCSParam              gcsParameter              `json:"gcs_param"`
	BQParam               bigqueryParameter         `json:"bigquery_param"`
	PulsarParam           pulsarParameter           `json:"pulsar_param"`
	AzureBlobStorageParam azureBlobStorageParameter `json:"azure_blob_storage_param"`
}

const ConvertFailed = "CONVERT connectionModel FAILED"

func (t *connectionModel) ToLoggingString() string {
	data, err := json.Marshal(t)
	if err != nil {
		return ConvertFailed
	}
	var dest connectionModel
	err = json.Unmarshal(data, &dest)
	if err != nil {
		return ConvertFailed
	}

	if dest.User != "" {
		dest.User = passwordMask
	}
	if dest.Password != "" {
		dest.Password = passwordMask
	}
	if dest.S3Param.AccessKeyId != "" {
		dest.S3Param.AccessKeyId = passwordMask
	}
	if dest.S3Param.SecretAccessKey != "" {
		dest.S3Param.SecretAccessKey = passwordMask
	}
	if dest.AzureBlobStorageParam.SASToken != "" {
		dest.AzureBlobStorageParam.SASToken = passwordMask
	}
	if dest.PulsarParam.Username != "" {
		dest.PulsarParam.Username = passwordMask
	}
	if dest.PulsarParam.Password != "" {
		dest.PulsarParam.Password = passwordMask
	}
	if dest.PulsarParam.Token != "" {
		dest.PulsarParam.Token = passwordMask
	}
	// Avoid logging raw TLS material
	if len(dest.MysqlParam.SSLCABytes) > 0 {
		dest.MysqlParam.SSLCABytes = []byte(passwordMask)
	}
	if len(dest.MysqlParam.SSLCertBytes) > 0 {
		dest.MysqlParam.SSLCertBytes = []byte(passwordMask)
	}
	if len(dest.MysqlParam.SSLKeyBytes) > 0 {
		dest.MysqlParam.SSLKeyBytes = []byte(passwordMask)
	}

	data, err = json.Marshal(dest)
	if err != nil {
		return ConvertFailed
	}

	return string(data)
}

func connectionGetHandler(c *route.Context, cl resolve.Cluster) http.Handler {
	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		body, err := io.ReadAll(r.Body)
		defer r.Body.Close() // nolint
		if err != nil {
			http.Error(w, "failed to read body contents", http.StatusInternalServerError)
			return
		}

		model := &connectionModel{}
		err = json.Unmarshal(body, model)
		if err != nil {
			// Surface non-internal errors to the response.
			http.Error(w, fmt.Sprintf("failed to parse the body contents: %v", err), http.StatusBadRequest)
			return
		}

		// ctx for dial kafka/mysql
		ctx, cancel := context.WithTimeout(context.TODO(), 10*time.Second)
		defer cancel()
		switch model.Type {
		case connectionTypeKafka:
			if err := checkKafkaAccessV2(ctx, model); err != nil {
				http.Error(w, err.Error(), http.StatusBadRequest)
				return
			}
		case connectionTypeMysql:
			if err := checkMySQLAccess(ctx, model); err != nil {
				http.Error(w, err.Error(), http.StatusBadRequest)
				return
			}
		case connectionTypeS3:
			s3Param := model.S3Param
			if err := checkS3Access(ctx, s3Param.S3Uri, s3Param.AccessKeyId, s3Param.SecretAccessKey, s3Param.RoleArn, s3Param.ExternalID); err != nil {
				http.Error(w, fmt.Sprintf("Connect S3 `%s` failed: %v", s3Param.S3Uri, err), http.StatusBadRequest)
				return
			}
		case connectionTypeGCS:
			gcsParam := model.GCSParam
			gcsOpt := &GcsOptions{}
			if gcsParam.ImpersonateServiceAccount != "" {
				gcsOpt.ImpersonateServiceAccount = &gcsParam.ImpersonateServiceAccount
			}
			if err := checkGcsAccess(ctx, gcsParam.GcsUrl, gcsParam.Region, gcsOpt); err != nil {
				http.Error(w, fmt.Sprintf("Connect GCS `%s` failed: %v", gcsParam.GcsUrl, err), http.StatusBadRequest)
				return
			}
		case connectionTypeAzureBlobStorage:
			azBlobParam := model.AzureBlobStorageParam
			if err := checkAzureBlobStorageAccess(ctx, azBlobParam.BlobUri, azBlobParam.SASToken); err != nil {
				log.Warn("access azure blob storage error", zap.Error(err), zap.String("connectionModel", model.ToLoggingString()))
				http.Error(w, fmt.Sprintf("Connect azure blob storage `%s` failed: %v", azBlobParam.BlobUri, err), http.StatusBadRequest)
				return
			}
		case connectionTypeSnowflake:
			if err := checkSnowflakeAccess(ctx, model); err != nil {
				http.Error(w, fmt.Sprintf("Connect Snowflake `%s` (using user `%s`) failed: %v", model.Url, model.User, err), http.StatusBadRequest)
				return
			}
		case connectionTypeRedshift:
			if err := checkRedshiftAccess(ctx, model); err != nil {
				http.Error(w, fmt.Sprintf("Connect Redshift `%s` (using user `%s`) failed: %v", model.Url, model.User, err), http.StatusBadRequest)
				return
			}
		case connectionTypeBigQuery:
			bqParam := model.BQParam
			if err := checkBigQueryAccess(ctx, bqParam.ProjectID, bqParam.DatasetID); err != nil {
				http.Error(w, fmt.Sprintf(
					"Connect BigQuery (project_id: `%s`, dataset_id `%s`) failed: %v", bqParam.ProjectID, bqParam.DatasetID, err,
				), http.StatusBadRequest)
				return
			}
		case connectionTypeCluster:
			if err := checkClusterAccess(ctx, c, cl, model.User, model.Password); err != nil {
				http.Error(w, fmt.Sprintf("Connect TiDB Cloud cluster (using user `%s`) failed: %v", model.User, err), http.StatusBadRequest)
				return
			}
		case connectionTypePulsar:
			if err := checkPulsarAccess(ctx, &model.PulsarParam); err != nil {
				log.Warn("access pulsar error", zap.Error(err), zap.String("connectionModel", model.ToLoggingString()))
				http.Error(w, err.Error(), http.StatusBadRequest)
				return
			}
		default:
			http.Error(w, fmt.Sprintf("Unknown connection type %s", model.Type), http.StatusBadRequest)
			return
		}
	})
}