package registry

import (
	"bytes"
	"encoding/gob"
	"errors"
	"github.com/hashicorp/consul/api"
	"github.com/hashicorp/consul/watch"
	"reflect"
	"strconv"
	"sync"
	"time"
)

//consul注册器
type consulRegistry struct {
	masterClient         *api.Client            //主客户端
	watchClients         []*api.Client          //监视器客户端
	servicesWatchParse   *watch.Plan            //服务列表监视器
	serviceWatchParseMap map[string]*watch.Plan //服务监视器映射
	observerMap          *sync.Map              //观察者映射
	serviceMap           map[string]*Service    //服务映射
	serviceMapLock       *sync.RWMutex          //服务映射锁
}

//新建consul注册器
func newConsulRegistry() Registry {
	return &consulRegistry{
		masterClient:         nil,
		watchClients:         make([]*api.Client, 0),
		servicesWatchParse:   nil,
		serviceWatchParseMap: make(map[string]*watch.Plan),
		observerMap:          new(sync.Map),
		serviceMap:           make(map[string]*Service),
		serviceMapLock:       new(sync.RWMutex),
	}
}

//内部初始化
func init() {
	if _, ok := newRegistryMap["consul"]; !ok {
		newRegistryMap["consul"] = newConsulRegistry
	}
}

//初始化
//参数名         类型         格式		默认值
//cluster		[]string	ip:port		["127.0.0.1:8500"]
//dc			string		string		"dc1"
func (cr *consulRegistry) Init(options map[string]interface{}) error {
	//参数过滤
	cluster := []string{"127.0.0.1:8500"}
	dc := "dc1"
	if v, ok := options["cluster"]; ok {
		if _, ok := v.([]string); ok {
			cluster = v.([]string)
		} else {
			return errors.New("consul  cluster []string")
		}
	}
	if v, ok := options["dc"]; ok {
		if _, ok := v.(string); ok {
			dc = v.(string)
		} else {
			return errors.New("consul dc string")
		}
	}

	//初始化参数
	cr.masterClient = newConsulClient(cluster[0], dc, time.Millisecond*100)
	for _, address := range cluster {
		cr.watchClients = append(cr.watchClients, newConsulClient(address, dc, time.Second*300))
	}

	//服务初始化
	parse, _ := watch.Parse(map[string]interface{}{"type": "services"})
	parse.Handler = cr.handlerServices
	cr.servicesWatchParse = parse
	go cr.watch(cr.servicesWatchParse)

	return nil
}

//新建consul客户端
func newConsulClient(address string, dc string, waitTime time.Duration) *api.Client {
	config := api.DefaultConfig()
	config.Address = address
	config.Datacenter = dc
	config.WaitTime = waitTime
	client, _ := api.NewClient(config)
	return client
}

//监控
func (cr *consulRegistry) watch(parse *watch.Plan) {
	sentry := 0

	defer func() {
		if e := recover(); e != nil {
			time.Sleep(time.Second * 1)
			if sentry < len(cr.watchClients)-1 {
				sentry++
			} else {
				sentry = 0
			}
		}
	}()

	_ = parse.RunWithClientAndLogger(cr.watchClients[sentry], nil)
}

//监控服务列表的变化
func (cr *consulRegistry) handlerServices(_ uint64, data interface{}) {
	services := data.(map[string][]string)

	//锁
	cr.serviceMapLock.Lock()
	defer cr.serviceMapLock.Unlock()

	for service, tags := range services {
		//如果服务不存在，则创建服务、监听服务
		if _, ok := cr.serviceWatchParseMap[service]; !ok {
			parse, _ := watch.Parse(map[string]interface{}{"type": "service", "service": service})
			parse.Handler = cr.handlerService
			cr.serviceWatchParseMap[service] = parse
			cr.serviceMap[service] = NewService(service)
			go cr.watch(parse)
		}

		//更新标签
		tagMap := make(map[string]struct{})
		for _, tag := range tags {
			tagMap[tag] = struct{}{}
		}
		if !reflect.DeepEqual(tagMap, cr.serviceMap[service].TagMap) {
			cr.serviceMap[service].TagMap = tagMap
		}
	}

	//处理被删除的服务
	for service, parse := range cr.serviceWatchParseMap {
		if _, ok := services[service]; !ok {
			//删除服务
			parse.Stop()
			delete(cr.serviceWatchParseMap, service)
			delete(cr.serviceMap, service)

			//执行观察者
			cr.observerMap.Range(func(_, value interface{}) bool {
				observer := value.(Observer)
				observer.DeleteService(service)
				return true
			})
		}
	}
}

//观察服务的变化
func (cr *consulRegistry) handlerService(_ uint64, data interface{}) {
	services := data.([]*api.ServiceEntry)

	cr.serviceMapLock.Lock()
	defer cr.serviceMapLock.Unlock()

	serviceName := services[0].Service.Service
	nodeMap := make(map[string]*Node)
	for _, service := range services {
		if service.Service.ID == "consul" {
			continue
		}

		//更新节点
		node := NewNode(service.Service.Service, service.Service.ID, service.Service.Address, service.Service.Port)
		node.Meta = service.Service.Meta
		for _, health := range service.Checks {
			if node.Id == health.ServiceID {
				node.Status = health.Status
			}
		}

		nodeMap[service.Service.ID] = node
	}

	cr.serviceMap[serviceName].NodeMap = nodeMap
	cr.observerMap.Range(func(_, value interface{}) bool {
		observer := value.(Observer)
		observer.UpdateNodes(cr.serviceMap[serviceName])
		return true
	})
}

//服务注册
func (cr *consulRegistry) Register(node *Node) error {
	check := &api.AgentServiceCheck{
		HTTP:                           "http://" + node.Address + ":" + strconv.Itoa(node.Port) + "/tech/health/check",
		Interval:                       "5s",
		Timeout:                        "10s",
//		DeregisterCriticalServiceAfter: "24h",
		DeregisterCriticalServiceAfter: "1m",
		CheckID:                        node.Id,
	}
	registration := api.AgentServiceRegistration{
		ID:      node.Id,
		Name:    node.ServiceName,
		Port:    node.Port,
		Address: node.Address,
		Check:   check,
		Meta:    node.Meta,
	}

	//可以重试一次
	for i := 0; i < 2; i++ {
		if e := cr.masterClient.Agent().ServiceRegister(&registration); e == nil {
			cr.observerMap.Range(func(_, value interface{}) bool {
				observer := value.(Observer)
				observer.AddNode(node)
				return true
			})
			return nil
		}
	}


	return errors.New("service register fail " + node.Id)
}

//服务注销
func (cr *consulRegistry) Deregister(node *Node) error {
	//可重试一次
	for i := 0; i < 2; i++ {
		if e := cr.masterClient.Agent().ServiceDeregister(node.Id); e == nil {
			cr.observerMap.Range(func(_, value interface{}) bool {
				observer := value.(Observer)
				observer.DelNode(node)
				return true
			})
			return nil
		}
	}


	return errors.New("service deregister failed " + node.ServiceName + ":" + node.Id)
}

//获取服务映射
func (cr *consulRegistry) GetServiceMap() map[string]*Service {
	cr.serviceMapLock.RLock()
	defer cr.serviceMapLock.RUnlock()

	//深拷贝
	rtn := make(map[string]*Service)
	buffer := bytes.NewBuffer(nil)
	_ = gob.NewEncoder(buffer).Encode(&cr.serviceMap)
	_ = gob.NewDecoder(bytes.NewBuffer(buffer.Bytes())).Decode(&rtn)

	return rtn
}

//获取服务
func (cr *consulRegistry) GetService(serviceName string) (*Service, error) {
	cr.serviceMapLock.RLock()
	defer cr.serviceMapLock.RUnlock()

	//深拷贝
	if v, ok := cr.serviceMap[serviceName]; ok {
		rtn := NewService(serviceName)
		rtn.TagMap = v.TagMap
		for kk, vv := range v.NodeMap {
			rtn.NodeMap[kk] = vv
		}
		return rtn, nil
	} else {
		return nil, errors.New("service not found")
	}
}

//设置观察者
func (cr *consulRegistry) SetObserver(name string, observer Observer) error {
	cr.serviceMapLock.RLock()
	defer cr.serviceMapLock.RUnlock()

	if _, ok := cr.observerMap.Load(name); ok {
		return errors.New("observer is exists")
	} else {
		for _, service := range cr.serviceMap {
			observer.UpdateNodes(service)
		}
		cr.observerMap.Store(name, observer)
	}

	return nil
}

//删除观察
func (cr *consulRegistry) DelObserver(name string) error {
	if _, ok := cr.observerMap.Load(name); ok {
		cr.observerMap.Delete(name)
	} else {
		return errors.New("observer is not exists")
	}

	return nil
}
