package streaming

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

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

	"github.com/tidbcloud/service-proxy/pkg/checker/core"
	"github.com/tidbcloud/service-proxy/pkg/checker/util"
)

func init() {
	core.DefaultRegistry.Register("pulsar", func(req *core.CheckRequest) (core.Checker, error) {
		if req.Pulsar == nil {
			return nil, errors.New("pulsar params required")
		}
		return NewPulsarChecker(req.Pulsar), nil
	})
}

const (
	defaultPulsarTimeout = 15 * time.Second
)

// PulsarChecker checks Apache Pulsar connectivity.
type PulsarChecker struct {
	params *core.PulsarParams
}

// NewPulsarChecker creates a new Pulsar checker.
func NewPulsarChecker(params *core.PulsarParams) *PulsarChecker {
	return &PulsarChecker{params: params}
}

// Type returns the checker type.
func (c *PulsarChecker) Type() string {
	return "pulsar"
}

// getTimeout returns the connection timeout, using default if not specified.
func (c *PulsarChecker) getTimeout() time.Duration {
	if c.params.ConnectionTimeout == 0 {
		return defaultPulsarTimeout
	}
	return time.Second * time.Duration(c.params.ConnectionTimeout)
}

// Check performs Pulsar connectivity check.
func (c *PulsarChecker) Check(ctx context.Context) error {
	if ctx == nil {
		return errors.New("context is required")
	}

	// Validate params
	if strings.TrimSpace(c.params.ServiceURL) == "" {
		return fmt.Errorf("service_url is required")
	}

	// Validate auth (no auth-none support - matching clusterproxy behavior)
	switch c.params.AuthType {
	case core.PulsarAuthTypeBasic:
		if c.params.Username == "" || c.params.Password == "" {
			return fmt.Errorf("username and password required for basic auth")
		}
	case core.PulsarAuthTypeToken:
		if c.params.Token == "" {
			return fmt.Errorf("token required for token auth")
		}
	default:
		return fmt.Errorf("unknown pulsar auth type: %s", c.params.AuthType)
	}

	// Validate topic
	if c.params.Topic != "" {
		if c.params.Tenant == "" || c.params.Namespace == "" {
			return fmt.Errorf("tenant and namespace required when topic is specified")
		}
	}

	select {
	case <-ctx.Done():
		return ctx.Err()
	default:
	}

	return c.checkAccess(ctx)
}

func (c *PulsarChecker) checkAccess(ctx context.Context) (err error) {
	startTime := time.Now()
	log.Info("start to connect to pulsar service", zap.String("service_url", util.SanitizeURL(c.params.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)))
		}
	}()

	select {
	case <-ctx.Done():
		return ctx.Err()
	default:
	}

	if err = c.checkAllBrokersReachability(ctx); err != nil {
		return err
	}
	if c.params.Topic != "" {
		if err = c.checkTopicAuthorization(); err != nil {
			return err
		}
	}
	return nil
}

func (c *PulsarChecker) checkAllBrokersReachability(ctx context.Context) error {
	brokerURL, err := url.ParseRequestURI(c.params.ServiceURL)
	if err != nil {
		return fmt.Errorf("failed to parse pulsar broker url: %w", err)
	}
	addrs := strings.Split(brokerURL.Host, ",")
	connTimeout := c.getTimeout()
	for _, addr := range addrs {
		if err := ctx.Err(); err != nil {
			return err
		}

		_, err = net.DialTimeout("tcp", addr, connTimeout)
		if err != nil {
			return fmt.Errorf("failed to dial pulsar broker: %w", err)
		}
		log.Info("dial pulsar broker successfully", zap.String("broker", addr))
	}
	return nil
}

func (c *PulsarChecker) checkTopicAuthorization() error {
	timeout := c.getTimeout()
	opts := pulsar.ClientOptions{
		URL:               c.params.ServiceURL,
		ConnectionTimeout: timeout,
		OperationTimeout:  timeout,
	}
	var auth pulsar.Authentication
	var err error
	switch c.params.AuthType {
	case core.PulsarAuthTypeBasic:
		auth, err = pulsar.NewAuthenticationBasic(c.params.Username, c.params.Password)
		if err != nil {
			return fmt.Errorf("init pulsar basic authentication: %w", err)
		}
	case core.PulsarAuthTypeToken:
		auth = pulsar.NewAuthenticationToken(c.params.Token)
	default:
		return fmt.Errorf("unknown pulsar authentication type: %s", c.params.AuthType)
	}
	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", c.params.Tenant, c.params.Namespace, c.params.Topic)
	_, err = cli.TopicPartitions(topic)
	if err != nil {
		log.Error("get pulsar topic partitions error",
			zap.String("pulsar_service", util.SanitizeURL(c.params.ServiceURL)), zap.String("topic", topic), zap.Error(err))
		return fmt.Errorf("failed to access pulsar topic %s: %w", topic, err)
	}
	return nil
}