package discovery

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

	"gitee.com/titan-kit/titan/log"
	"gitee.com/titan-kit/titan/registry"

	"google.golang.org/grpc/attributes"
	"google.golang.org/grpc/resolver"
)

type Option func(o *builder)

func WithLogger(logger log.Logger) Option {
	return func(o *builder) {
		o.logger = logger
	}
}

type builder struct {
	discoverer registry.Discovery
	logger     log.Logger
}

// NewDiscoveryBuilder 创建一个用于服务发现服务的构建器.
func NewDiscoveryBuilder(d registry.Discovery, opts ...Option) resolver.Builder {
	b := &builder{discoverer: d, logger: log.DefaultLogger}
	for _, o := range opts {
		o(b)
	}
	return b
}

func (d *builder) Build(target resolver.Target, cc resolver.ClientConn, opts resolver.BuildOptions) (resolver.Resolver, error) {
	w, err := d.discoverer.Watch(context.Background(), target.Endpoint)
	if err != nil {
		return nil, err
	} else {
		ctx, cancel := context.WithCancel(context.Background())
		r := &discoveryResolver{w: w, cc: cc, ctx: ctx, cancel: cancel, log: log.NewSlf4g("starter/rpc/resolver/discovery", d.logger)}
		go r.watch()
		return r, nil
	}
}

func (d *builder) Scheme() string {
	return "discovery"
}

type discoveryResolver struct {
	w      registry.Watcher
	cc     resolver.ClientConn
	log    *log.Slf4g
	ctx    context.Context
	cancel context.CancelFunc
}

func (r *discoveryResolver) watch() {
	for {
		select {
		case <-r.ctx.Done():
			return
		default:
		}

		ins, err := r.w.Next()
		if err != nil {
			r.log.ErrorF("Failed to watch discovery endpoint: %v", err)
			time.Sleep(time.Second)
			continue
		}
		r.update(ins)
	}
}

func (r *discoveryResolver) update(ins []*registry.ServiceInstance) {
	var address []resolver.Address
	for _, in := range ins {
		endpoint, err := parseEndpoint(in.Endpoints)
		if err != nil {
			r.log.ErrorF("Failed to parse discovery endpoint: %v", err)
			continue
		}
		if endpoint == "" {
			continue
		}
		addr := resolver.Address{ServerName: in.Name, Attributes: parseAttributes(in.Metadata), Addr: endpoint}
		address = append(address, addr)
	}
	_ = r.cc.UpdateState(resolver.State{Addresses: address})
}

func (r *discoveryResolver) Close() {
	r.cancel()
	_ = r.w.Stop()
}

func (r *discoveryResolver) ResolveNow(options resolver.ResolveNowOptions) {}

func parseEndpoint(endpoints []string) (string, error) {
	for _, e := range endpoints {
		u, err := url.Parse(e)
		if err != nil {
			return "", err
		}
		if u.Scheme == "rpc" {
			return u.Host, nil
		}
	}
	return "", nil
}

func parseAttributes(md map[string]string) *attributes.Attributes {
	pairs := make([]interface{}, 0, len(md))
	for k, v := range md {
		pairs = append(pairs, k, v)
	}
	return attributes.New(pairs...)
}