package portforward

import (
	"context"
	"fmt"
	"strconv"

	"github.com/pingcap/errors"
	"google.golang.org/grpc/resolver"
)

// RegisterClusterWithSchema registers a kubernetes cluster with a schema.
func RegisterClusterWithSchema(manager *Manager, schema string) {
	resolver.Register(NewBuilder(manager, schema))
}

type targetInfo struct {
	serviceName       string
	serviceNamespace  string
	port              string
	resolveByPortName bool
	useFirstPort      bool
}

func parseResolverTarget(target resolver.Target) (targetInfo, error) {
	var service, port, namespace string
	if target.URL.Host == "" {
		// kubernetes:///service.namespace:port
		service, port, namespace = splitServicePortNamespace(target.Endpoint())
	} else if target.URL.Port() == "" && target.Endpoint() != "" {
		// kubernetes://namespace/service:port
		service, port, _ = splitServicePortNamespace(target.Endpoint())
		namespace = target.URL.Hostname()
	} else {
		// kubernetes://service.namespace:port
		service, port, namespace = splitServicePortNamespace(target.URL.Host)
	}

	if service == "" {
		return targetInfo{}, fmt.Errorf("target %s must specify a service", &target.URL)
	}

	resolveByPortName := false
	useFirstPort := false
	if port == "" {
		useFirstPort = true
	} else if _, err := strconv.Atoi(port); err != nil {
		resolveByPortName = true
	}

	return targetInfo{
		serviceName:       service,
		serviceNamespace:  namespace,
		port:              port,
		resolveByPortName: resolveByPortName,
		useFirstPort:      useFirstPort,
	}, nil
}

func (ti targetInfo) String() string {
	return fmt.Sprintf("kubernetes://%s/%s:%s", ti.serviceNamespace, ti.serviceName, ti.port)
}

// NewBuilder creates a kubeBuilder which is used by grpc resolver.
func NewBuilder(manager *Manager, schema string) resolver.Builder {
	return &kubeBuilder{
		manager: manager,
		schema:  schema,
	}
}

// Build creates a new resolver for the given target.
//
// gRPC dial calls Build synchronously, and fails if the returned error is
// not nil.
func (b *kubeBuilder) Build(target resolver.Target, cc resolver.ClientConn, _ resolver.BuildOptions) (resolver.Resolver, error) {
	ti, err := parseResolverTarget(target)
	if err != nil {
		return nil, err
	}
	if ti.serviceNamespace == "" {
		ti.serviceNamespace = "nightly-ms"
	}
	ctx := context.Background()
	portNum, err := strconv.ParseUint(ti.port, 10, 32)
	if err != nil {
		return nil, errors.Trace(err)
	}
	localAddr, err := b.manager.GetForwardedAddress(ctx, ti.serviceName, ti.serviceNamespace, int32(portNum))
	if err != nil {
		return nil, err
	}
	cc.NewAddress([]resolver.Address{{Addr: localAddr}}) //nolint: staticcheck
	r := &kResolver{
		target: ti,
		cc:     cc,
	}
	return r, nil
}

// Scheme returns the scheme supported by this resolver.
// Scheme is defined at https://github.com/grpc/grpc/blob/master/doc/naming.md.
func (b *kubeBuilder) Scheme() string {
	return b.schema
}

type kubeBuilder struct {
	manager *Manager
	schema  string
}

type kResolver struct {
	target      targetInfo
	cc          resolver.ClientConn
	stopChannel chan struct{}
}

// ResolveNow will be called by gRPC to try to resolve the target name again.
// It's just a hint, resolver can ignore this if it's not necessary.
func (k *kResolver) ResolveNow(resolver.ResolveNowOptions) {
}

// Close closes the resolver.
func (k *kResolver) Close() {
	k.stopChannel <- struct{}{}
}