package central

import (
	"context"
	"fmt"
	"net"
	"net/url"
	"strings"
	"time"

	"github.com/apache/pulsar-client-go/pulsar"
	"github.com/pingcap/log"
	"go.uber.org/zap"
)

func checkPulsarAccess(ctx context.Context, pulsarParam *pulsarParameter) (err error) {
	if pulsarParam == nil {
		return fmt.Errorf("pulsar parameter can not be nil")
	}

	startTime := time.Now()
	log.Info("start to connect to pulsar service", zap.String("service url", pulsarParam.ServiceURL))
	defer func() {
		if err != nil {
			log.Error("connect to pulsar service failed",
				zap.Error(err), zap.Duration("cost time", time.Since(startTime)))
		} else {
			log.Info("connect to pulsar service successfully", zap.Duration("cost time", time.Since(startTime)))
		}
	}()

	if err = checkAllPulsarBrokersReachability(pulsarParam); err != nil {
		return err
	}
	if pulsarParam.Topic != "" {
		if err = checkPulsarTopicAuthorization(pulsarParam); err != nil {
			return err
		}
	}
	return nil
}

func checkAllPulsarBrokersReachability(pulsarParam *pulsarParameter) error {
	brokerURL, err := url.ParseRequestURI(pulsarParam.ServiceURL)
	if err != nil {
		return fmt.Errorf("parse pulsar broker url %s failed: %w", pulsarParam.ServiceURL, err)
	}
	addrs := strings.Split(brokerURL.Host, ",")
	connTimeout := time.Second * time.Duration(pulsarParam.ConnectionTimeout)
	for _, addr := range addrs {
		_, err = net.DialTimeout("tcp", addr, connTimeout)
		if err != nil {
			return fmt.Errorf("dial pulsar broker %s failed: %w", addr, err)
		}
		log.Info("dial pulsar broker successfully", zap.String("broker", addr))
	}
	return nil
}

func checkPulsarTopicAuthorization(pulsarParam *pulsarParameter) error {
	timeout := time.Second * time.Duration(pulsarParam.ConnectionTimeout)
	opts := pulsar.ClientOptions{
		URL:               pulsarParam.ServiceURL,
		ConnectionTimeout: timeout,
		OperationTimeout:  timeout,
	}
	var auth pulsar.Authentication
	var err error
	switch pulsarParam.AuthenticationType {
	case pulsarAuthTypeBasic:
		auth, err = pulsar.NewAuthenticationBasic(pulsarParam.Username, pulsarParam.Password)
		if err != nil {
			return fmt.Errorf("init pulsar basic authentication: %w", err)
		}
	case pulsarAuthTypeToken:
		auth = pulsar.NewAuthenticationToken(pulsarParam.Token)
	default:
		return fmt.Errorf("unknown pulsar authentication type: %s", pulsarParam.AuthenticationType)
	}
	opts.Authentication = auth

	cli, err := pulsar.NewClient(opts)
	if err != nil {
		return fmt.Errorf("new pulsar client error: %w", err)
	}
	defer cli.Close()

	topic := fmt.Sprintf("persistent://%s/%s/%s", pulsarParam.Tenant, pulsarParam.Namespace, pulsarParam.Topic)
	_, err = cli.TopicPartitions(topic)
	if err != nil {
		log.Error("get pulsar topic partitions error",
			zap.String("pulsar_service", pulsarParam.ServiceURL), zap.String("topic", topic), zap.Error(err))
		return fmt.Errorf(
			"connect to pulsar service error. Please check if the authentication or pulsar protocol is wrong")
	}
	return nil
}