package sbox

import (
	"gitee.com/dennis-kk/rpc-go-backend/idlrpc"
	"gitee.com/dennis-kk/rpc-go-backend/idlrpc/pkg/transport"
	"gitee.com/dennis-kk/service-box-go/common"
	_ "gitee.com/dennis-kk/service-box-go/internal/zookeeper"
	"gitee.com/dennis-kk/service-box-go/util/errors"
	"gitee.com/dennis-kk/service-box-go/util/service_infra"
	"gitee.com/dennis-kk/service-box-go/util/slog"
	"gitee.com/dennis-kk/service-box-go/util/slog/zap"
	"sync/atomic"
	"testing"
	"time"
)

var (
	testLayer *serviceLayer
	logger    slog.BoxLogger
	res       = struct {
		name  string
		value string
		err   error
	}{}

	testCh *BoxChannel
)

func init() {
	logger, _ = zap.NewLogger(slog.WithCallerSkipCount(2))
	slog.SetDefaultLog(logger)
	testCh = &BoxChannel{
		status:    common.ChannelClose,
		localHost: "127.0.0.1:4666",
		peerHost:  "127.0.0.1:2333",
		conn:      &testConn{},
	}
}

func testChangedHandle(name string, eventType service_infra.ChangedType, changeInfo *service_infra.ChangedInfo, err error) {
	if err != nil {
		return
	}
	switch eventType {
	case service_infra.ServiceChange:
		res.name = name
		res.err = err
		for _, h := range changeInfo.Hosts {
			res.value = h
		}
	case service_infra.ServiceDelete:
	}
}

func testInitLayer() error {
	res = struct {
		name  string
		value string
		err   error
	}{}
	testLayer, _ = makeNewLayer("192.168.82.22:2181", testConnectToservice)
	return nil
}

func testConnectToservice(name, network, host string) {
	go func() {
		testCh.status = common.ChannelRunning
		res.err = testLayer.onConnect(host, testCh)
		if res.err != nil {
			return
		}
	}()
}

func testCallBack(name string, trans *BoxChannel, err error) {
	res.name = name
	res.value = trans.RemoteAddr()
	res.err = err
	logger.Info("CallBack Service Successful !")
}

func testProxyFinder(uuid uint64, trans transport.ITransport) (idlrpc.IProxy, error) {
	return nil, nil
}

func makeNewLayer(host string, ch ConnectHandle) (*serviceLayer, error) {
	layer := makeServiceLayer(ch, logger)
	cfg := &serviceLayerConfig{
		MiddlewareType: "zookeeper",
		Prefix:         "/",
		Hosts: []string{
			host,
		},
	}
	if err := layer.init(cfg, testProxyFinder); err != nil {
		return nil, err
	}
	err := layer.svs.AddListener(layer.onServiceChanged)
	if err != nil {
		return nil, err
	}

	if err := layer.start(); err != nil {
		return nil, err
	}

	return layer, nil
}

func TestRegisterService(t *testing.T) {

	if err := testInitLayer(); err != nil {
		t.Fatalf("init service layer %v", err)
	}

	if err := testLayer.registerService("5871407834537456905", "127.0.0.1:2333"); err != nil {
		t.Fatalf("register Service Error %v", err)
	}

	if err := testLayer.getTransport("5871407834537456905", testCallBack); err != nil {
		t.Fatalf("get transport error %v !", err)
	}

	ticker := time.NewTimer(100 * time.Millisecond)
	run := true
	for run {
		select {
		case <-ticker.C:
			run = false
		default:
			testLayer.tick()
		}
	}

	if res.err != nil {
		t.Errorf("get service error %v", res.err)
	}
	if res.name != "5871407834537456905" || res.value != "127.0.0.1:2333" {
		t.Errorf("test error %s, %s", res.name, res.value)
	}
}

func TestAddServiceWatcher(t *testing.T) {
	if err := testInitLayer(); err != nil {
		t.Fatalf("init service layer %v", err)
	}

	watcher := func(eType ServiceEvent, name string, host string, proxy idlrpc.IProxy) {}

	err := testLayer.addWatcher(5871407834537456905, "5871407834537456905", watcher)
	if err != nil {
		t.Fatalf("add service watcher error !!!")
	}
}

func TestRepeatedAddWatcher(t *testing.T) {
	if err := testInitLayer(); err != nil {
		t.Fatalf("init service layer %v", err)
	}

	watcher := func(eType ServiceEvent, name string, host string, proxy idlrpc.IProxy) {}

	err := testLayer.addWatcher(5871407834537456905, "5871407834537456905", watcher)
	if err != nil {
		t.Fatalf("add service watcher error !!!")
	}

	err = testLayer.addWatcher(5871407834537456905, "5871407834537456905", watcher)
	if err != errors.RepeatedWatcher {
		t.Fatalf("unexcpetion error %v", err)
	}

	testLayer.stop()
}

func TestMultiAddWatcher(t *testing.T) {
	if err := testInitLayer(); err != nil {
		t.Fatalf("init service layer %v", err)
	}

	watcher1 := func(eType ServiceEvent, name string, host string, proxy idlrpc.IProxy) {}
	watcher2 := func(eType ServiceEvent, name string, host string, proxy idlrpc.IProxy) {}

	err := testLayer.addWatcher(5871407834537456905, "5871407834537456905", watcher1)
	if err != nil {
		t.Fatalf("add service watcher error !!!")
	}

	err = testLayer.addWatcher(5871407834537456905, "5871407834537456905", watcher2)
	if err != nil {
		t.Fatalf("unexcpetion error %v", err)
	}

	testLayer.stop()
}

func TestSingleWatcher(t *testing.T) {
	if err := testInitLayer(); err != nil {
		t.Fatalf("init service layer %v", err)
	}

	except := struct {
		name string
		host string
	}{}

	watcher := func(eType ServiceEvent, name string, host string, proxy idlrpc.IProxy) {
		except.name = name
		except.host = host
	}

	err := testLayer.addWatcher(5871407834537456905, "5871407834537456905", watcher)
	if err != nil {
		t.Fatalf("add service watcher error !!!")
	}

	if err := testLayer.registerService("5871407834537456905", "127.0.0.1:2333"); err != nil {
		t.Fatalf("register Service Error %v", err)
	}

	if err := testLayer.getTransport("5871407834537456905", testCallBack); err != nil {
		t.Fatalf("get transport error %v !", err)
	}

	ticker := time.NewTimer(100 * time.Millisecond)
	run := true
	for run {
		select {
		case <-ticker.C:
			run = false
		default:
			testLayer.tick()
		}
	}

	if except.name != "5871407834537456905" {
		t.Errorf("unexception service name %s:5871407834537456905", except.name)
	}

	if except.host != "127.0.0.1:2333" {
		t.Errorf("unexception service host %s:127.0.0.1:2333", except.host)
	}

	testLayer.stop()
}

//TestAddWatcherBeforeRegisterService 注册服务前就已然添加监听
func TestAddWatcherBeforeRegisterService(t *testing.T) {
	if err := testInitLayer(); err != nil {
		t.Fatalf("init service layer %v", err)
	}

	except := struct {
		name string
		host string
	}{}

	watcher := func(eType ServiceEvent, name string, host string, proxy idlrpc.IProxy) {
		except.name = name
		except.host = host
	}

	err := testLayer.addWatcher(5871407834537456905, "5871407834537456905", watcher)
	if err != nil {
		t.Fatalf("add service watcher error  %v!!!", err)
	}

	if err := testLayer.registerService("5871407834537456905", "127.0.0.1:2333"); err != nil {
		t.Fatalf("register Service Error %v", err)
	}

	ticker := time.NewTimer(100 * time.Millisecond)
	run := true
	for run {
		select {
		case <-ticker.C:
			run = false
		default:
			testLayer.tick()
		}
	}

	if except.name != "5871407834537456905" {
		t.Errorf("unexception service name %s:5871407834537456905", except.name)
	}

	if except.host != "127.0.0.1:2333" {
		t.Errorf("unexception service host %s:127.0.0.1:2333", except.host)
	}

	testLayer.stop()
}

//TestAddWatcherWithAlreadyDiscovered 添加监听时，服务已经进行过发现
func TestAddWatcherWithAlreadyDiscovered(t *testing.T) {
	if err := testInitLayer(); err != nil {
		t.Fatalf("init service layer %v", err)
	}

	except := struct {
		name string
		host string
	}{}

	watcher := func(eType ServiceEvent, name string, host string, proxy idlrpc.IProxy) {
		except.name = name
		except.host = host
	}

	if err := testLayer.registerService("5871407834537456905", "127.0.0.1:2333"); err != nil {
		t.Fatalf("register Service Error %v", err)
	}

	if err := testLayer.getTransport("5871407834537456905", testCallBack); err != nil {
		t.Fatalf("get transport error %v !", err)
	}
	//确定已经发现到服务
	ticker := time.NewTimer(100 * time.Millisecond)
	run := true
	for run {
		select {
		case <-ticker.C:
			run = false
		default:
			testLayer.tick()
		}
	}

	if res.name != "5871407834537456905" {
		t.Fatalf("discover Service Error %s", res.name)
	}

	err := testLayer.addWatcher(5871407834537456905, "5871407834537456905", watcher)
	if err != nil {
		t.Fatalf("add service watcher error  %v!!!", err)
	}

	ticker = time.NewTimer(100 * time.Millisecond)
	run = true
	for run {
		select {
		case <-ticker.C:
			run = false
		default:
			testLayer.tick()
		}
	}

	if except.name != "5871407834537456905" {
		t.Errorf("unexception service name %s:5871407834537456905", except.name)
	}

	if except.host != "127.0.0.1:2333" {
		t.Errorf("unexception service host %s:127.0.0.1:2333", except.host)
	}

	testLayer.stop()
}

func TestTriggerMultiWatcher(t *testing.T) {
	if err := testInitLayer(); err != nil {
		t.Fatalf("init service layer %v", err)
	}

	except := struct {
		name string
		host string
		time int32
	}{}

	watcher1 := func(eType ServiceEvent, name string, host string, proxy idlrpc.IProxy) {
		except.name = name
		except.host = host
		atomic.AddInt32(&except.time, 1)
	}
	watcher2 := func(eType ServiceEvent, name string, host string, proxy idlrpc.IProxy) {
		atomic.AddInt32(&except.time, 1)
	}

	err := testLayer.addWatcher(5871407834537456905, "5871407834537456905", watcher1)
	if err != nil {
		t.Fatalf("add service watcher error !!!")
	}

	err = testLayer.addWatcher(5871407834537456905, "5871407834537456905", watcher2)
	if err != nil {
		t.Fatalf("unexcpetion error %v", err)
	}

	if err := testLayer.registerService("5871407834537456905", "127.0.0.1:2333"); err != nil {
		t.Fatalf("register Service Error %v", err)
	}

	if err := testLayer.getTransport("5871407834537456905", testCallBack); err != nil {
		t.Fatalf("get transport error %v !", err)
	}

	ticker := time.NewTimer(100 * time.Millisecond)
	run := true
	for run {
		select {
		case <-ticker.C:
			run = false
		default:
			testLayer.tick()
		}
	}

	if except.name != "5871407834537456905" {
		t.Errorf("unexception service name %s:5871407834537456905", except.name)
	}

	if except.host != "127.0.0.1:2333" {
		t.Errorf("unexception service host %s:127.0.0.1:2333", except.host)
	}

	if except.time != 2 {
		t.Errorf("unexception service time %d:2", except.time)
	}

	testLayer.stop()
}

func TestRemoveTrigger(t *testing.T) {
	if err := testInitLayer(); err != nil {
		t.Fatalf("init service layer %v", err)
	}

	except := struct {
		name string
		host string
	}{}

	watcher1 := func(eType ServiceEvent, name string, host string, proxy idlrpc.IProxy) {
		except.name = name
		except.host = host
	}

	err := testLayer.addWatcher(5871407834537456905, "5871407834537456905", watcher1)
	if err != nil {
		t.Fatalf("add service watcher error !!!")
	}

	err = testLayer.removeWatcher("5871407834537456905", watcher1)
	if err != nil {
		t.Fatalf("remove service watcher error !!!")
	}

	if _, ok := testLayer.watcher.watchersMap["5871407834537456905"]; ok {
		t.Fatalf("remove service watcher failed !!!")
	}

	testLayer.stop()
}

func TestRemoveUnExistWatcher(t *testing.T) {
	if err := testInitLayer(); err != nil {
		t.Fatalf("init service layer %v", err)
	}

	except := struct {
		name string
		host string
	}{}

	watcher1 := func(eType ServiceEvent, name string, host string, proxy idlrpc.IProxy) {
		except.name = name
		except.host = host
	}

	err := testLayer.removeWatcher("5871407834537456905", watcher1)
	if err != nil {
		t.Fatalf("remove service watcher error !!!")
	}

	testLayer.stop()
}

//TestServiceRemove 启动服务，删除节点，其他集群能够正常收到删除消息并且剔除已
func TestServiceRemove(t *testing.T) {
	if err := testInitLayer(); err != nil {
		t.Fatalf("init service layer %v", err)
	}

	layer2, _ := makeNewLayer("192.168.82.22:2181", testConnectToservice)
	if layer2 == nil {
		t.Fatalf("invalid layer")
	}

	// layer2 注册服务
	if err := layer2.registerService("5871407834537456905", "127.0.0.1:2333"); err != nil {
		t.Fatalf("register Service Error %v", err)
	}
	// tick 1秒
	ticker := time.NewTimer(1 * time.Second)
	run := true
	for run {
		select {
		case <-ticker.C:
			run = false
		default:
			testLayer.tick()
			layer2.tick()
		}
	}
	res.name = ""
	//layer2 反注册
	layer2.stop()

	ticker = time.NewTimer(1 * time.Second)
	// 获取layer2
	if err := testLayer.getTransport("5871407834537456905", testCallBack); err != nil {
		t.Fatalf("get transport error %v !", err)
	}
	// tick 1秒 触发超时
	ticker = time.NewTimer(1 * time.Second)
	run = true
	for run {
		select {
		case <-ticker.C:
			run = false
		default:
			testLayer.tick()
			layer2.tick()
		}
	}

	if len(res.name) != 0 {
		t.Fatalf("dirty data of nodes %s", res.name)
	}

	testLayer.stop()
}

func TestFindServiceAfterReboot(t *testing.T) {
	if err := testInitLayer(); err != nil {
		t.Fatalf("init service layer %v", err)
	}

	layer2, _ := makeNewLayer("192.168.82.22:2181", testConnectToservice)
	if layer2 == nil {
		t.Fatalf("invalid layer")
	}
	// layer2 注册服务
	if err := layer2.registerService("5871407834537456905", "127.0.0.1:2333"); err != nil {
		t.Fatalf("register Service Error %v", err)
	}
	// tick 1秒
	ticker := time.NewTimer(1 * time.Second)
	run := true
	for run {
		select {
		case <-ticker.C:
			run = false
		default:
			testLayer.tick()
			layer2.tick()
		}
	}
	//layer2 反注册
	layer2.stop()

	//重新注册
	layer2, _ = makeNewLayer("192.168.82.22:2181", testConnectToservice)
	if layer2 == nil {
		t.Fatalf("invalid layer")
	}
	// layer2 注册服务
	if err := layer2.registerService("5871407834537456905", "127.0.0.1:2333"); err != nil {
		t.Fatalf("register Service Error %v", err)
	}
	// 获取layer2
	if err := testLayer.getTransport("5871407834537456905", testCallBack); err != nil {
		t.Fatalf("get transport error %v !", err)
	}
	// tick 1秒
	ticker = time.NewTimer(2 * time.Second)
	run = true
	for run {
		select {
		case <-ticker.C:
			run = false
		default:
			testLayer.tick()
			layer2.tick()
		}
	}

	if res.name != "5871407834537456905" || res.value != "127.0.0.1:2333" {
		t.Errorf("test error %s, %s", res.name, res.value)
	}
}

//TestRefinedServiceAfterReboot 测试一个已经发现过的服务重启之后，能否正常更新对应的connect
func TestRefinedServiceAfterReboot(t *testing.T) {
	// 第一次初始化准备服务
	if err := testInitLayer(); err != nil {
		t.Fatalf("init service layer %v", err)
	}

	layer2, _ := makeNewLayer("192.168.82.22:2181", testConnectToservice)
	if layer2 == nil {
		t.Fatalf("invalid layer")
	}
	// layer2 注册服务
	if err := layer2.registerService("5871407834537456905", "127.0.0.1:2333"); err != nil {
		t.Fatalf("register Service Error %v", err)
	}
	// 第一次尝试发现
	if err := testLayer.getTransport("5871407834537456905", testCallBack); err != nil {
		t.Fatalf("get transport error %v !", err)
	}
	//tick 1秒
	ticker := time.NewTimer(1 * time.Second)
	run := true
	for run {
		select {
		case <-ticker.C:
			run = false
		default:
			testLayer.tick()
			layer2.tick()
		}
	}
	//layer2 反注册
	layer2.stop()
	//test layer close
	testLayer.onClose("127.0.0.1:2333")
	//ch 清理
	testCh.status = common.ChannelClose
	// tick 1秒
	ticker = time.NewTimer(1 * time.Second)
	run = true
	for run {
		select {
		case <-ticker.C:
			run = false
		default:
			testLayer.tick()
			layer2.tick()
		}
	}
	//重新注册
	layer2, _ = makeNewLayer("192.168.82.22:2181", testConnectToservice)
	if layer2 == nil {
		t.Fatalf("invalid layer")
	}
	// layer2 注册服务
	if err := layer2.registerService("5871407834537456905", "127.0.0.1:2333"); err != nil {
		t.Fatalf("register Service Error %v", err)
	}
	// 第一次尝试发现
	if err := testLayer.getTransport("5871407834537456905", testCallBack); err != nil {
		t.Fatalf("get transport error %v !", err)
	}

	// tick 1秒
	ticker = time.NewTimer(1 * time.Second)
	run = true
	for run {
		select {
		case <-ticker.C:
			run = false
		default:
			testLayer.tick()
			layer2.tick()
		}
	}

	if testCh.status != common.ChannelRunning {
		t.Fatalf("error for channel status %d ", testCh.status)
	}

	testLayer.stop()
}