package client

import (
	"context"
	"strings"
	"sync"
	"time"

	"github.com/pingcap/log"
	"github.com/tidbcloud/ffs/config"
	"github.com/tidbcloud/ffs/pkg/v1/proto"
	"github.com/tidbcloud/ffs/util/grpcutil"
	"go.etcd.io/etcd/clientv3"
	"google.golang.org/grpc"
	"google.golang.org/grpc/resolver"
)

type Client struct {
	*grpcutil.ServiceClient
}

func NewClient(endpoints ...string) *Client {
	return &Client{
		ServiceClient: &grpcutil.ServiceClient{
			Schema:    config.Universe,
			Endpoints: endpoints,
			NewClient: func(conn *grpc.ClientConn) interface{} {
				return proto.NewUniverseServiceClient(conn)
			}},
	}
}

func NewDiscovery(endpoints ...string) *Client {
	r := newResolver(endpoints)
	resolver.Register(r)
	return &Client{
		ServiceClient: &grpcutil.ServiceClient{
			Target: r.Scheme() + "://" + config.UniversePrefix,
			Options: []grpc.DialOption{
				grpc.WithBalancerName("round_robin"),
				grpc.WithInsecure(),
				grpc.WithDefaultCallOptions(
					grpc.MaxCallRecvMsgSize(config.GrpcMaxMsgSize),
					grpc.MaxCallSendMsgSize(config.GrpcMaxMsgSize),
				)},
			NewClient: func(conn *grpc.ClientConn) interface{} {
				return proto.NewUniverseServiceClient(conn)
			}},
	}
}

func (c *Client) Client() proto.UniverseServiceClient {
	return c.ServiceClient.Client().(proto.UniverseServiceClient)
}

type etcdResolver struct {
	endpoints []string
	cli       *clientv3.Client
	mu        sync.Mutex
	cc        resolver.ClientConn
	state     resolver.State
}

func newResolver(endpoints []string) resolver.Builder {
	r := &etcdResolver{endpoints: endpoints}
	var err error
	r.cli, err = clientv3.New(clientv3.Config{
		Endpoints:   r.endpoints,
		DialTimeout: 15 * time.Second,
	})
	if err != nil {
		log.S().Fatalf("failed to build resolver err %s", err)
	}
	r.watch(config.UniversePrefix)
	log.S().Infof("new etcd resolver endpoints %v", endpoints)
	return r
}

func (r *etcdResolver) Scheme() string {
	return config.Schema
}

func (r *etcdResolver) ResolveNow(rn resolver.ResolveNowOptions) {

}

func (r *etcdResolver) Close() {
	if r.cli != nil {
		r.cli.Close()
	}
}

func (r *etcdResolver) Build(target resolver.Target, clientConn resolver.ClientConn, opts resolver.BuildOptions) (resolver.Resolver, error) {
	r.mu.Lock()
	r.cc = clientConn
	r.cc.UpdateState(r.state)
	r.mu.Unlock()
	return r, nil
}

func (r *etcdResolver) watch(keyPrefix string) {
	resp, err := r.cli.Get(context.Background(), keyPrefix, clientv3.WithPrefix())
	if err != nil {
		log.S().Fatalf("%s", err)
	}
	r.mu.Lock()
	for i := range resp.Kvs {
		r.state.Addresses = append(r.state.Addresses, resolver.Address{Addr: strings.TrimPrefix(string(resp.Kvs[i].Key), keyPrefix)})
	}
	r.mu.Unlock()
	rev := resp.Header.Revision + 1
	go func() {
		rch := r.cli.Watch(context.Background(), keyPrefix, clientv3.WithPrefix(), clientv3.WithRev(rev))
		for n := range rch {
			for _, ev := range n.Events {
				addr := strings.TrimPrefix(string(ev.Kv.Key), keyPrefix)
				switch ev.Type {
				case clientv3.EventTypePut:
					r.mu.Lock()
					if !exists(r.state.Addresses, addr) {
						r.state.Addresses = append(r.state.Addresses, resolver.Address{Addr: addr})
					}
					r.cc.UpdateState(r.state)
					r.mu.Unlock()
				case clientv3.EventTypeDelete:
					r.mu.Lock()
					if s, ok := remove(r.state.Addresses, addr); ok {
						r.state.Addresses = s
					}
					r.cc.UpdateState(r.state)
					r.mu.Unlock()
				}
			}
		}
	}()
}

func exists(l []resolver.Address, addr string) bool {
	for i := range l {
		if l[i].Addr == addr {
			return true
		}
	}
	return false
}

func remove(s []resolver.Address, addr string) ([]resolver.Address, bool) {
	for i := range s {
		if s[i].Addr == addr {
			s[i] = s[len(s)-1]
			return s[:len(s)-1], true
		}
	}
	return nil, false
}