/*
 * Copyright 2021 CloudWeGo Authors
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package server

import (
	"context"
	"errors"
	"log"
	"net"
	"os"
	"reflect"
	"strings"
	"sync"
	"sync/atomic"
	"testing"
	"time"

	"github.com/bytedance/gopkg/cloud/metainfo"
	"github.com/cloudwego/localsession"
	"github.com/cloudwego/localsession/backup"
	"github.com/golang/mock/gomock"

	"github.com/cloudwego/kitex/internal/mocks"
	mockslimiter "github.com/cloudwego/kitex/internal/mocks/limiter"
	internal_server "github.com/cloudwego/kitex/internal/server"
	"github.com/cloudwego/kitex/internal/test"
	"github.com/cloudwego/kitex/pkg/endpoint"
	"github.com/cloudwego/kitex/pkg/limit"
	"github.com/cloudwego/kitex/pkg/limiter"
	"github.com/cloudwego/kitex/pkg/registry"
	"github.com/cloudwego/kitex/pkg/remote"
	"github.com/cloudwego/kitex/pkg/remote/bound"
	"github.com/cloudwego/kitex/pkg/remote/trans"
	"github.com/cloudwego/kitex/pkg/remote/trans/nphttp2"
	"github.com/cloudwego/kitex/pkg/rpcinfo"
	"github.com/cloudwego/kitex/pkg/serviceinfo"
	"github.com/cloudwego/kitex/pkg/stats"
	"github.com/cloudwego/kitex/pkg/streaming"
	"github.com/cloudwego/kitex/pkg/transmeta"
	"github.com/cloudwego/kitex/pkg/utils"
	"github.com/cloudwego/kitex/transport"
)

var svcInfo = mocks.ServiceInfo()

// NOTE: always use this method to get addr for server listening
// should be used with `WithServiceAddr(addr)`
func getAddrForListener() net.Addr {
	addr := test.GetLocalAddress()
	ret, _ := net.ResolveTCPAddr("tcp", addr)
	return ret
}

// NewTestServer calls NewServer with a random addr
// DO NOT USE `NewServer` and `s.Run()` without specifying addr, it listens on :8888 ...
func NewTestServer(ops ...Option) (Server, net.Addr) {
	addr := getAddrForListener()
	svr := NewServer(append(ops, WithServiceAddr(addr))...)
	return svr, addr
}

// WaitServer waits utils its listener is ready
func WaitServer(s Server) {
	// sleep 5ms, and likely it's up
	time.Sleep(5 * time.Millisecond)
	// wait at most 100*10ms ~ 1s or panic
	for i := 0; i < 100; i++ {
		svr := s.(*server)
		svr.Lock()
		remotesvr := svr.svr
		svr.Unlock()
		if remotesvr == nil || remotesvr.Address() == nil {
			time.Sleep(10 * time.Millisecond)
			continue
		}
		a := remotesvr.Address()
		log.Printf("[WaitServer] %s/%s is up", a.Network(), a.String())
		return
	}
	panic("server " + s.(*server).svr.Address().String() + " not ready")
}

func goWaitAndStop(t *testing.T, s Server) {
	go func() {
		WaitServer(s)
		// give more time for server to init other stuffs after listener is ready
		time.Sleep(10 * time.Millisecond)
		err := s.Stop()
		test.Assert(t, err == nil, err)
	}()
}

func TestServerRun(t *testing.T) {
	var opts []Option
	opts = append(opts, WithMetaHandler(noopMetahandler{}))
	svr, _ := NewTestServer(opts...)
	err := svr.RegisterService(mocks.ServiceInfo(), mocks.MyServiceHandler())
	test.Assert(t, err == nil)

	var runHook int32
	var shutdownHook int32
	RegisterStartHook(func() {
		atomic.AddInt32(&runHook, 1)
	})
	RegisterShutdownHook(func() {
		atomic.AddInt32(&shutdownHook, 1)
	})

	goWaitAndStop(t, svr)
	err = svr.Run()
	test.Assert(t, err == nil, err)

	test.Assert(t, atomic.LoadInt32(&runHook) == 1)
	test.Assert(t, atomic.LoadInt32(&shutdownHook) == 1)
}

func TestReusePortServerRun(t *testing.T) {
	addr := getAddrForListener()
	var opts []Option
	opts = append(opts, WithReusePort(true))
	opts = append(opts, WithServiceAddr(addr), WithExitWaitTime(time.Microsecond*10))

	var wg sync.WaitGroup
	for i := 0; i < 8; i++ {
		wg.Add(2)
		go func() {
			defer wg.Done()
			svr := NewServer(opts...)
			time.AfterFunc(time.Millisecond*100, func() {
				defer wg.Done()
				err := svr.Stop()
				test.Assert(t, err == nil, err)
			})
			err := svr.RegisterService(mocks.ServiceInfo(), mocks.MyServiceHandler())
			test.Assert(t, err == nil)
			err = svr.Run()
			test.Assert(t, err == nil, err)
		}()
	}
	wg.Wait()
}

func TestInitOrResetRPCInfo(t *testing.T) {
	var opts []Option
	rwTimeout := time.Millisecond
	opts = append(opts, WithReadWriteTimeout(rwTimeout))
	svr := &server{
		opt:  internal_server.NewOptions(opts),
		svcs: newServices(),
	}
	svr.init()
	err := svr.RegisterService(mocks.ServiceInfo(), mocks.MyServiceHandler())
	test.Assert(t, err == nil)

	remoteAddr := utils.NewNetAddr("tcp", "to")
	conn := new(mocks.Conn)
	conn.RemoteAddrFunc = func() (r net.Addr) {
		return remoteAddr
	}
	rpcInfoInitFunc := svr.initOrResetRPCInfoFunc()
	ri := rpcInfoInitFunc(nil, conn.RemoteAddr())
	test.Assert(t, ri != nil)
	test.Assert(t, ri.From().Address().String() == remoteAddr.String())
	test.Assert(t, ri.Config().ReadWriteTimeout() == rwTimeout)

	// modify rpcinfo
	fi := rpcinfo.AsMutableEndpointInfo(ri.From())
	fi.SetServiceName("mock service")
	fi.SetMethod("mock method")
	fi.SetTag("key", "value")

	ti := rpcinfo.AsMutableEndpointInfo(ri.To())
	ti.SetServiceName("mock service")
	ti.SetMethod("mock method")
	ti.SetTag("key", "value")

	if setter, ok := ri.Invocation().(rpcinfo.InvocationSetter); ok {
		setter.SetSeqID(123)
		setter.SetPackageName("mock package")
		setter.SetServiceName("mock service")
		setter.SetMethodName("mock method")
	}

	mc := rpcinfo.AsMutableRPCConfig(ri.Config())
	mc.SetTransportProtocol(transport.TTHeader)
	mc.SetConnectTimeout(10 * time.Second)
	mc.SetRPCTimeout(20 * time.Second)
	mc.SetInteractionMode(rpcinfo.Streaming)
	mc.SetIOBufferSize(1024)
	mc.SetReadWriteTimeout(30 * time.Second)

	rpcStats := rpcinfo.AsMutableRPCStats(ri.Stats())
	rpcStats.SetRecvSize(1024)
	rpcStats.SetSendSize(1024)
	rpcStats.SetPanicked(errors.New("panic"))
	rpcStats.SetError(errors.New("err"))
	rpcStats.SetLevel(stats.LevelDetailed)

	// check setting
	test.Assert(t, ri.From().ServiceName() == "mock service")
	test.Assert(t, ri.From().Method() == "mock method")
	value, exist := ri.From().Tag("key")
	test.Assert(t, exist && value == "value")

	test.Assert(t, ri.To().ServiceName() == "mock service")
	test.Assert(t, ri.To().Method() == "mock method")
	value, exist = ri.To().Tag("key")
	test.Assert(t, exist && value == "value")

	test.Assert(t, ri.Invocation().SeqID() == 123)
	test.Assert(t, ri.Invocation().PackageName() == "mock package")
	test.Assert(t, ri.Invocation().ServiceName() == "mock service")
	test.Assert(t, ri.Invocation().MethodName() == "mock method")

	test.Assert(t, ri.Config().TransportProtocol() == transport.TTHeader)
	test.Assert(t, ri.Config().ConnectTimeout() == 10*time.Second)
	test.Assert(t, ri.Config().RPCTimeout() == 20*time.Second)
	test.Assert(t, ri.Config().InteractionMode() == rpcinfo.Streaming)
	test.Assert(t, ri.Config().IOBufferSize() == 1024)
	test.Assert(t, ri.Config().ReadWriteTimeout() == rwTimeout)

	test.Assert(t, ri.Stats().RecvSize() == 1024)
	test.Assert(t, ri.Stats().SendSize() == 1024)
	hasPanicked, _ := ri.Stats().Panicked()
	test.Assert(t, hasPanicked)
	test.Assert(t, ri.Stats().Error() != nil)
	test.Assert(t, ri.Stats().Level() == stats.LevelDetailed)

	// test reset
	pOld := reflect.ValueOf(ri).Pointer()
	ri = rpcInfoInitFunc(ri, conn.RemoteAddr())
	pNew := reflect.ValueOf(ri).Pointer()
	test.Assert(t, pOld == pNew, pOld, pNew)
	test.Assert(t, ri.From().Address().String() == remoteAddr.String())
	test.Assert(t, ri.Config().ReadWriteTimeout() == rwTimeout)

	test.Assert(t, ri.From().ServiceName() == "")
	test.Assert(t, ri.From().Method() == "")
	_, exist = ri.From().Tag("key")
	test.Assert(t, !exist)

	test.Assert(t, ri.To().ServiceName() == "")
	test.Assert(t, ri.To().Method() == "")
	_, exist = ri.To().Tag("key")
	test.Assert(t, !exist)

	test.Assert(t, ri.Invocation().SeqID() == 0)
	test.Assert(t, ri.Invocation().PackageName() == "")
	test.Assert(t, ri.Invocation().ServiceName() == "")
	test.Assert(t, ri.Invocation().MethodName() == "")

	test.Assert(t, ri.Config().TransportProtocol() == transport.PurePayload)
	test.Assert(t, ri.Config().ConnectTimeout() == 50*time.Millisecond)
	test.Assert(t, ri.Config().RPCTimeout() == 0)
	test.Assert(t, ri.Config().InteractionMode() == rpcinfo.PingPong)
	test.Assert(t, ri.Config().IOBufferSize() == 4096)
	test.Assert(t, ri.Config().ReadWriteTimeout() == rwTimeout)

	test.Assert(t, ri.Stats().RecvSize() == 0)
	test.Assert(t, ri.Stats().SendSize() == 0)
	_, panicked := ri.Stats().Panicked()
	test.Assert(t, panicked == nil)
	test.Assert(t, ri.Stats().Error() == nil)
	test.Assert(t, ri.Stats().Level() == 0)

	// test reset after rpcInfoPool is disabled
	t.Run("reset with pool disabled", func(t *testing.T) {
		backupState := rpcinfo.PoolEnabled()
		defer rpcinfo.EnablePool(backupState)
		rpcinfo.EnablePool(false)

		riNew := rpcInfoInitFunc(ri, conn.RemoteAddr())
		pOld, pNew := reflect.ValueOf(ri).Pointer(), reflect.ValueOf(riNew).Pointer()
		test.Assert(t, pOld != pNew, pOld, pNew)
	})
}

func TestServiceRegisterFailed(t *testing.T) {
	t.Parallel() // slow test, use Parallel

	mockRegErr := errors.New("mock register error")
	var rCount int
	var drCount int
	mockRegistry := MockRegistry{
		RegisterFunc: func(info *registry.Info) error {
			rCount++
			return mockRegErr
		},
		DeregisterFunc: func(info *registry.Info) error {
			drCount++
			return nil
		},
	}
	var opts []Option
	opts = append(opts, WithRegistry(mockRegistry))
	svr, _ := NewTestServer(opts...)
	err := svr.RegisterService(mocks.ServiceInfo(), mocks.MyServiceHandler())
	test.Assert(t, err == nil)

	err = svr.Run()
	test.Assert(t, err != nil)
	test.Assert(t, strings.Contains(err.Error(), mockRegErr.Error()))
	test.Assert(t, drCount == 1)
}

func TestServiceDeregisterFailed(t *testing.T) {
	t.Parallel() // slow test, use Parallel

	mockDeregErr := errors.New("mock deregister error")
	var rCount int
	var drCount int
	mockRegistry := MockRegistry{
		RegisterFunc: func(info *registry.Info) error {
			rCount++
			return nil
		},
		DeregisterFunc: func(info *registry.Info) error {
			drCount++
			return mockDeregErr
		},
	}
	var opts []Option
	opts = append(opts, WithRegistry(mockRegistry))
	svr, _ := NewTestServer(opts...)
	err := svr.RegisterService(mocks.ServiceInfo(), mocks.MyServiceHandler())
	test.Assert(t, err == nil)

	time.AfterFunc(2000*time.Millisecond, func() {
		err := svr.Stop()
		test.Assert(t, strings.Contains(err.Error(), mockDeregErr.Error()))
	})
	err = svr.Run()
	test.Assert(t, err == nil, err)
	test.Assert(t, rCount == 1)
}

func TestServiceRegistryInfo(t *testing.T) {
	t.Parallel() // slow test, use Parallel

	registryInfo := &registry.Info{
		Weight: 100,
		Tags:   map[string]string{"aa": "bb"},
	}
	checkInfo := func(info *registry.Info) {
		test.Assert(t, info.PayloadCodec == serviceinfo.Thrift.String(), info.PayloadCodec)
		test.Assert(t, info.Weight == registryInfo.Weight, info.Addr)
		test.Assert(t, len(info.Tags) == len(registryInfo.Tags), info.Tags)
		test.Assert(t, info.Tags["aa"] == registryInfo.Tags["aa"], info.Tags)
	}
	var rCount int
	var drCount int
	mockRegistry := MockRegistry{
		RegisterFunc: func(info *registry.Info) error {
			checkInfo(info)
			rCount++
			return nil
		},
		DeregisterFunc: func(info *registry.Info) error {
			checkInfo(info)
			drCount++
			return nil
		},
	}
	var opts []Option
	opts = append(opts, WithRegistry(mockRegistry))
	opts = append(opts, WithRegistryInfo(registryInfo))
	svr, _ := NewTestServer(opts...)
	err := svr.RegisterService(mocks.ServiceInfo(), mocks.MyServiceHandler())
	test.Assert(t, err == nil)

	time.AfterFunc(2000*time.Millisecond, func() {
		err := svr.Stop()
		test.Assert(t, err == nil, err)
	})
	err = svr.Run()
	test.Assert(t, err == nil, err)
	test.Assert(t, rCount == 1)
	test.Assert(t, drCount == 1)
}

func TestServiceRegistryNoInitInfo(t *testing.T) {
	t.Parallel() // slow test, use Parallel

	checkInfo := func(info *registry.Info) {
		test.Assert(t, info.PayloadCodec == serviceinfo.Thrift.String(), info.PayloadCodec)
	}
	var rCount int
	var drCount int
	mockRegistry := MockRegistry{
		RegisterFunc: func(info *registry.Info) error {
			checkInfo(info)
			rCount++
			return nil
		},
		DeregisterFunc: func(info *registry.Info) error {
			checkInfo(info)
			drCount++
			return nil
		},
	}
	var opts []Option
	opts = append(opts, WithRegistry(mockRegistry))
	svr, _ := NewTestServer(opts...)
	err := svr.RegisterService(mocks.ServiceInfo(), mocks.MyServiceHandler())
	test.Assert(t, err == nil)

	time.AfterFunc(2000*time.Millisecond, func() {
		err := svr.Stop()
		test.Assert(t, err == nil, err)
	})
	err = svr.Run()
	test.Assert(t, err == nil, err)
	test.Assert(t, rCount == 1)
	test.Assert(t, drCount == 1)
}

// TestServiceRegistryInfoWithNilTags is to check the Tags val. If Tags of RegistryInfo is nil,
// the Tags of ServerBasicInfo will be assigned to Tags of RegistryInfo
func TestServiceRegistryInfoWithNilTags(t *testing.T) {
	t.Parallel() // slow test, use Parallel

	registryInfo := &registry.Info{
		Weight: 100,
	}
	checkInfo := func(info *registry.Info) {
		test.Assert(t, info.PayloadCodec == serviceinfo.Thrift.String(), info.PayloadCodec)
		test.Assert(t, info.Weight == registryInfo.Weight, info.Weight)
		test.Assert(t, info.Tags["aa"] == "bb", info.Tags)
	}
	var rCount int
	var drCount int
	mockRegistry := MockRegistry{
		RegisterFunc: func(info *registry.Info) error {
			checkInfo(info)
			rCount++
			return nil
		},
		DeregisterFunc: func(info *registry.Info) error {
			checkInfo(info)
			drCount++
			return nil
		},
	}
	var opts []Option
	opts = append(opts, WithRegistry(mockRegistry))
	opts = append(opts, WithRegistryInfo(registryInfo))
	opts = append(opts, WithServerBasicInfo(&rpcinfo.EndpointBasicInfo{
		Tags: map[string]string{"aa": "bb"},
	}))
	svr, _ := NewTestServer(opts...)
	err := svr.RegisterService(mocks.ServiceInfo(), mocks.MyServiceHandler())
	test.Assert(t, err == nil)

	time.AfterFunc(2000*time.Millisecond, func() {
		err := svr.Stop()
		test.Assert(t, err == nil, err)
	})
	err = svr.Run()
	test.Assert(t, err == nil, err)
	test.Assert(t, rCount == 1)
	test.Assert(t, drCount == 1)
}

func TestServiceRegistryInfoWithSkipListenAddr(t *testing.T) {
	t.Parallel() // slow test, use Parallel

	realAddr, _ := net.ResolveTCPAddr("tcp", test.GetLocalAddress())
	registryInfo := &registry.Info{
		Weight:         100,
		Addr:           realAddr,
		SkipListenAddr: true,
	}
	checkInfo := func(info *registry.Info) {
		test.Assert(t, info.Weight == registryInfo.Weight, info.Weight)
		test.Assert(t, info.SkipListenAddr, info.SkipListenAddr)
		test.Assert(t, info.Addr.String() == realAddr.String(), info.Addr)
	}
	var rCount int
	var drCount int
	mockRegistry := MockRegistry{
		RegisterFunc: func(info *registry.Info) error {
			checkInfo(info)
			rCount++
			return nil
		},
		DeregisterFunc: func(info *registry.Info) error {
			checkInfo(info)
			drCount++
			return nil
		},
	}
	var opts []Option
	opts = append(opts, WithRegistry(mockRegistry))
	opts = append(opts, WithRegistryInfo(registryInfo))
	svr, _ := NewTestServer(opts...)
	err := svr.RegisterService(mocks.ServiceInfo(), mocks.MyServiceHandler())
	test.Assert(t, err == nil)

	time.AfterFunc(1500*time.Millisecond, func() {
		err := svr.Stop()
		test.Assert(t, err == nil, err)
	})
	err = svr.Run()
	test.Assert(t, err == nil, err)
	test.Assert(t, rCount == 1)
	test.Assert(t, drCount == 1)
}

func TestServiceRegistryInfoWithoutSkipListenAddr(t *testing.T) {
	t.Parallel() // slow test, use Parallel

	realAddr, _ := net.ResolveTCPAddr("tcp", test.GetLocalAddress())
	registryInfo := &registry.Info{
		Weight: 100,
		Addr:   realAddr,
	}
	checkInfo := func(info *registry.Info) {
		test.Assert(t, info.Weight == registryInfo.Weight, info.Weight)
		test.Assert(t, !info.SkipListenAddr, info.SkipListenAddr)
		test.Assert(t, info.Addr.String() != realAddr.String(), info.Addr)
	}
	var rCount int
	var drCount int
	mockRegistry := MockRegistry{
		RegisterFunc: func(info *registry.Info) error {
			checkInfo(info)
			rCount++
			return nil
		},
		DeregisterFunc: func(info *registry.Info) error {
			checkInfo(info)
			drCount++
			return nil
		},
	}
	var opts []Option
	opts = append(opts, WithRegistry(mockRegistry))
	opts = append(opts, WithRegistryInfo(registryInfo))
	svr, _ := NewTestServer(opts...)
	err := svr.RegisterService(mocks.ServiceInfo(), mocks.MyServiceHandler())
	test.Assert(t, err == nil)

	time.AfterFunc(1500*time.Millisecond, func() {
		err := svr.Stop()
		test.Assert(t, err == nil, err)
	})
	err = svr.Run()
	test.Assert(t, err == nil, err)
	test.Assert(t, rCount == 1)
	test.Assert(t, drCount == 1)
}

func TestGRPCServerMultipleServices(t *testing.T) {
	var opts []Option
	opts = append(opts, withGRPCTransport())
	svr, _ := NewTestServer(opts...)
	err := svr.RegisterService(mocks.ServiceInfo(), mocks.MyServiceHandler())
	test.Assert(t, err == nil)
	err = svr.RegisterService(mocks.Service2Info(), mocks.MyServiceHandler())
	test.Assert(t, err == nil)

	test.DeepEqual(t, svr.(*server).svcs.SearchService("", mocks.MockMethod, false, serviceinfo.Thrift), mocks.ServiceInfo())
	test.DeepEqual(t, svr.(*server).svcs.SearchService("", mocks.Mock2Method, false, serviceinfo.Thrift), mocks.Service2Info())
	goWaitAndStop(t, svr)
	err = svr.Run()
	test.Assert(t, err == nil, err)
}

func TestServerBoundHandler(t *testing.T) {
	ctrl := gomock.NewController(t)
	defer ctrl.Finish()

	interval := time.Millisecond * 100
	cases := []struct {
		opts          []Option
		wantInbounds  []remote.InboundHandler
		wantOutbounds []remote.OutboundHandler
	}{
		{
			opts: []Option{
				WithLimit(&limit.Option{
					MaxConnections: 1000,
					MaxQPS:         10000,
				}),
				WithMetaHandler(noopMetahandler{}),
			},
			wantInbounds: []remote.InboundHandler{
				bound.NewTransMetaHandler([]remote.MetaHandler{transmeta.MetainfoServerHandler, noopMetahandler{}}),
				bound.NewServerLimiterHandler(limiter.NewConnectionLimiter(1000), limiter.NewQPSLimiter(interval, 10000), nil, false),
			},
			wantOutbounds: []remote.OutboundHandler{
				bound.NewTransMetaHandler([]remote.MetaHandler{transmeta.MetainfoServerHandler, noopMetahandler{}}),
			},
		},
		{
			opts: []Option{
				WithConnectionLimiter(mockslimiter.NewMockConcurrencyLimiter(ctrl)),
			},
			wantInbounds: []remote.InboundHandler{
				bound.NewTransMetaHandler([]remote.MetaHandler{transmeta.MetainfoServerHandler}),
				bound.NewServerLimiterHandler(mockslimiter.NewMockConcurrencyLimiter(ctrl), &limiter.DummyRateLimiter{}, nil, false),
			},
			wantOutbounds: []remote.OutboundHandler{
				bound.NewTransMetaHandler([]remote.MetaHandler{transmeta.MetainfoServerHandler}),
			},
		},
		{
			opts: []Option{
				WithQPSLimiter(mockslimiter.NewMockRateLimiter(ctrl)),
			},
			wantInbounds: []remote.InboundHandler{
				bound.NewTransMetaHandler([]remote.MetaHandler{transmeta.MetainfoServerHandler}),
				bound.NewServerLimiterHandler(&limiter.DummyConcurrencyLimiter{}, mockslimiter.NewMockRateLimiter(ctrl), nil, true),
			},
			wantOutbounds: []remote.OutboundHandler{
				bound.NewTransMetaHandler([]remote.MetaHandler{transmeta.MetainfoServerHandler}),
			},
		},
		{
			opts: []Option{
				WithConnectionLimiter(mockslimiter.NewMockConcurrencyLimiter(ctrl)),
				WithQPSLimiter(mockslimiter.NewMockRateLimiter(ctrl)),
			},
			wantInbounds: []remote.InboundHandler{
				bound.NewTransMetaHandler([]remote.MetaHandler{transmeta.MetainfoServerHandler}),
				bound.NewServerLimiterHandler(mockslimiter.NewMockConcurrencyLimiter(ctrl), mockslimiter.NewMockRateLimiter(ctrl), nil, true),
			},
			wantOutbounds: []remote.OutboundHandler{
				bound.NewTransMetaHandler([]remote.MetaHandler{transmeta.MetainfoServerHandler}),
			},
		},
		{
			opts: []Option{
				WithLimit(&limit.Option{
					MaxConnections: 1000,
					MaxQPS:         10000,
				}),
				WithConnectionLimiter(mockslimiter.NewMockConcurrencyLimiter(ctrl)),
			},
			wantInbounds: []remote.InboundHandler{
				bound.NewTransMetaHandler([]remote.MetaHandler{transmeta.MetainfoServerHandler}),
				bound.NewServerLimiterHandler(mockslimiter.NewMockConcurrencyLimiter(ctrl), limiter.NewQPSLimiter(interval, 10000), nil, false),
			},
			wantOutbounds: []remote.OutboundHandler{
				bound.NewTransMetaHandler([]remote.MetaHandler{transmeta.MetainfoServerHandler}),
			},
		},
		{
			opts: []Option{
				WithLimit(&limit.Option{
					MaxConnections: 1000,
					MaxQPS:         10000,
				}),
				WithConnectionLimiter(mockslimiter.NewMockConcurrencyLimiter(ctrl)),
				WithQPSLimiter(mockslimiter.NewMockRateLimiter(ctrl)),
			},
			wantInbounds: []remote.InboundHandler{
				bound.NewTransMetaHandler([]remote.MetaHandler{transmeta.MetainfoServerHandler}),
				bound.NewServerLimiterHandler(mockslimiter.NewMockConcurrencyLimiter(ctrl), mockslimiter.NewMockRateLimiter(ctrl), nil, true),
			},
			wantOutbounds: []remote.OutboundHandler{
				bound.NewTransMetaHandler([]remote.MetaHandler{transmeta.MetainfoServerHandler}),
			},
		},
		{
			opts: []Option{
				WithLimit(&limit.Option{
					MaxConnections: 1000,
					MaxQPS:         10000,
				}),
				WithConnectionLimiter(mockslimiter.NewMockConcurrencyLimiter(ctrl)),
				WithQPSLimiter(mockslimiter.NewMockRateLimiter(ctrl)),
				WithMuxTransport(),
			},
			wantInbounds: []remote.InboundHandler{
				bound.NewTransMetaHandler([]remote.MetaHandler{transmeta.MetainfoServerHandler}),
				bound.NewServerLimiterHandler(mockslimiter.NewMockConcurrencyLimiter(ctrl), mockslimiter.NewMockRateLimiter(ctrl), nil, true),
			},
			wantOutbounds: []remote.OutboundHandler{
				bound.NewTransMetaHandler([]remote.MetaHandler{transmeta.MetainfoServerHandler}),
			},
		},
		{
			opts: []Option{
				WithLimit(&limit.Option{
					MaxConnections: 1000,
					MaxQPS:         10000,
				}),
				WithMuxTransport(),
			},
			wantInbounds: []remote.InboundHandler{
				bound.NewTransMetaHandler([]remote.MetaHandler{transmeta.MetainfoServerHandler}),
				bound.NewServerLimiterHandler(limiter.NewConnectionLimiter(1000), limiter.NewQPSLimiter(interval, 10000), nil, true),
			},
			wantOutbounds: []remote.OutboundHandler{
				bound.NewTransMetaHandler([]remote.MetaHandler{transmeta.MetainfoServerHandler}),
			},
		},
		{
			opts: []Option{
				WithMetaHandler(noopMetahandler{}),
			},
			wantInbounds: []remote.InboundHandler{
				bound.NewTransMetaHandler([]remote.MetaHandler{transmeta.MetainfoServerHandler, noopMetahandler{}}),
			},
			wantOutbounds: []remote.OutboundHandler{
				bound.NewTransMetaHandler([]remote.MetaHandler{transmeta.MetainfoServerHandler, noopMetahandler{}}),
			},
		},
	}
	for _, tcase := range cases {
		opts := append(tcase.opts, WithExitWaitTime(time.Millisecond*10))
		svr, _ := NewTestServer(opts...)
		err := svr.RegisterService(mocks.ServiceInfo(), mocks.MyServiceHandler())
		test.Assert(t, err == nil)

		goWaitAndStop(t, svr)
		err = svr.Run()
		test.Assert(t, err == nil, err)

		// may have data race issue. make sure will NOT have any connection in this case.
		// Read: DeepEqual
		// Write: OnActive/OnInactive
		iSvr := svr.(*server)
		test.Assert(t, inboundDeepEqual(iSvr.opt.RemoteOpt.Inbounds, tcase.wantInbounds))
		test.Assert(t, reflect.DeepEqual(iSvr.opt.RemoteOpt.Outbounds, tcase.wantOutbounds))
	}
}

func TestInvokeHandlerWithContextBackup(t *testing.T) {
	testInvokeHandlerWithSession(t, true, "localhost:0")
	os.Setenv(localsession.SESSION_CONFIG_KEY, "true,100,1h")
	testInvokeHandlerWithSession(t, false, "localhost:0")
}

func testInvokeHandlerWithSession(t *testing.T, fail bool, ad string) {
	callMethod := "mock"
	var opts []Option
	mwExec := false
	opts = append(opts, WithMiddleware(func(next endpoint.Endpoint) endpoint.Endpoint {
		return func(ctx context.Context, req, resp interface{}) (err error) {
			mwExec = true
			return next(ctx, req, resp)
		}
	}))

	k1, v1 := "1", "1"
	var backupHandler backup.BackupHandler
	if !fail {
		opts = append(opts, WithContextBackup(true, true))
		backupHandler = func(prev, cur context.Context) (context.Context, bool) {
			v := prev.Value(k1)
			if v != nil {
				cur = context.WithValue(cur, k1, v)
				return cur, true
			}
			return cur, true
		}
	}

	opts = append(opts, WithCodec(&mockCodec{}))
	transHdlrFact := &mockSvrTransHandlerFactory{}
	exitCh := make(chan bool)
	var ln net.Listener
	transSvr := &mocks.MockTransServer{
		BootstrapServerFunc: func(net.Listener) error {
			{ // mock server call
				ri := rpcinfo.NewRPCInfo(nil, nil, rpcinfo.NewInvocation(svcInfo.ServiceName, callMethod), nil, rpcinfo.NewRPCStats())
				ctx := rpcinfo.NewCtxWithRPCInfo(context.Background(), ri)
				inkSetter := ri.Invocation().(rpcinfo.InvocationSetter)
				inkSetter.SetMethodInfo(svcInfo.MethodInfo(context.Background(), callMethod))
				recvMsg := remote.NewMessage(nil, ri, remote.Call, remote.Server)
				recvMsg.NewData(callMethod)
				sendMsg := remote.NewMessage(svcInfo.MethodInfo(context.Background(), callMethod).NewResult(), ri, remote.Reply, remote.Server)

				// inject kvs here
				ctx = metainfo.WithPersistentValue(ctx, "a", "b")
				ctx = context.WithValue(ctx, k1, v1)

				_, err := transHdlrFact.hdlr.OnMessage(ctx, recvMsg, sendMsg)
				test.Assert(t, err == nil, err)
			}
			<-exitCh
			return nil
		},
		ShutdownFunc: func() error {
			if ln != nil {
				ln.Close()
			}
			exitCh <- true
			return nil
		},
		CreateListenerFunc: func(addr net.Addr) (net.Listener, error) {
			var err error
			ln, err = net.Listen("tcp", ad)
			return ln, err
		},
	}
	opts = append(opts, WithTransServerFactory(mocks.NewMockTransServerFactory(transSvr)))
	opts = append(opts, WithTransHandlerFactory(transHdlrFact))

	svr := NewServer(opts...)
	time.AfterFunc(100*time.Millisecond, func() {
		err := svr.Stop()
		test.Assert(t, err == nil, err)
	})
	serviceHandler := false
	err := svr.RegisterService(mocks.ServiceInfo(), mocks.MockFuncHandler(func(ctx context.Context, req *mocks.MyRequest) (r *mocks.MyResponse, err error) {
		serviceHandler = true

		wg := sync.WaitGroup{}
		wg.Add(1)
		go func() {
			defer wg.Done()

			// miss context here
			ctx := backup.RecoverCtxOnDemands(context.Background(), backupHandler)

			if !fail {
				b, _ := metainfo.GetPersistentValue(ctx, "a")
				test.Assert(t, b == "b", "can't get metainfo")
				test.Assert(t, ctx.Value(k1) == v1, "can't get v1")
			} else {
				_, ok := metainfo.GetPersistentValue(ctx, "a")
				test.Assert(t, !ok, "can get metainfo")
				test.Assert(t, ctx.Value(k1) != v1, "can get v1")
			}
		}()
		wg.Wait()

		return &mocks.MyResponse{Name: "mock"}, nil
	}))
	test.Assert(t, err == nil)
	err = svr.Run()
	test.Assert(t, err == nil, err)
	test.Assert(t, mwExec)
	test.Assert(t, serviceHandler)
}

func TestInvokeHandlerExec(t *testing.T) {
	callMethod := "mock"
	var opts []Option
	mwExec := false
	opts = append(opts, WithMiddleware(func(next endpoint.Endpoint) endpoint.Endpoint {
		return func(ctx context.Context, req, resp interface{}) (err error) {
			mwExec = true
			return next(ctx, req, resp)
		}
	}))
	opts = append(opts, WithCodec(&mockCodec{}))
	transHdlrFact := &mockSvrTransHandlerFactory{}
	exitCh := make(chan bool)
	var ln net.Listener
	transSvr := &mocks.MockTransServer{
		BootstrapServerFunc: func(net.Listener) error {
			{ // mock server call
				ri := rpcinfo.NewRPCInfo(nil, nil, rpcinfo.NewInvocation(svcInfo.ServiceName, callMethod), nil, rpcinfo.NewRPCStats())
				ctx := rpcinfo.NewCtxWithRPCInfo(context.Background(), ri)
				inkSetter := ri.Invocation().(rpcinfo.InvocationSetter)
				inkSetter.SetMethodInfo(svcInfo.MethodInfo(context.Background(), callMethod))
				recvMsg := remote.NewMessage(nil, ri, remote.Call, remote.Server)
				recvMsg.NewData(callMethod)
				sendMsg := remote.NewMessage(svcInfo.MethodInfo(context.Background(), callMethod).NewResult(), ri, remote.Reply, remote.Server)

				_, err := transHdlrFact.hdlr.OnMessage(ctx, recvMsg, sendMsg)
				test.Assert(t, err == nil, err)
			}
			<-exitCh
			return nil
		},
		ShutdownFunc: func() error {
			if ln != nil {
				ln.Close()
			}
			exitCh <- true
			return nil
		},
		CreateListenerFunc: func(addr net.Addr) (net.Listener, error) {
			var err error
			ln, err = net.Listen("tcp", "localhost:0")
			return ln, err
		},
	}
	opts = append(opts, WithTransServerFactory(mocks.NewMockTransServerFactory(transSvr)))
	opts = append(opts, WithTransHandlerFactory(transHdlrFact))

	svr := NewServer(opts...)
	time.AfterFunc(100*time.Millisecond, func() {
		err := svr.Stop()
		test.Assert(t, err == nil, err)
	})
	serviceHandler := false
	err := svr.RegisterService(mocks.ServiceInfo(), mocks.MockFuncHandler(func(ctx context.Context, req *mocks.MyRequest) (r *mocks.MyResponse, err error) {
		serviceHandler = true
		return &mocks.MyResponse{Name: "mock"}, nil
	}))
	test.Assert(t, err == nil)
	err = svr.Run()
	test.Assert(t, err == nil, err)
	test.Assert(t, mwExec)
	test.Assert(t, serviceHandler)
}

func TestInvokeHandlerPanic(t *testing.T) {
	callMethod := "mock"
	var opts []Option
	mwExec := false
	opts = append(opts, WithMiddleware(func(next endpoint.Endpoint) endpoint.Endpoint {
		return func(ctx context.Context, req, resp interface{}) (err error) {
			mwExec = true
			return next(ctx, req, resp)
		}
	}))
	opts = append(opts, WithCodec(&mockCodec{}))
	transHdlrFact := &mockSvrTransHandlerFactory{}
	exitCh := make(chan bool)
	var ln net.Listener
	transSvr := &mocks.MockTransServer{
		BootstrapServerFunc: func(net.Listener) error {
			{
				// mock server call
				ri := rpcinfo.NewRPCInfo(nil, nil, rpcinfo.NewInvocation(svcInfo.ServiceName, callMethod), nil, rpcinfo.NewRPCStats())
				ctx := rpcinfo.NewCtxWithRPCInfo(context.Background(), ri)
				inkSetter := ri.Invocation().(rpcinfo.InvocationSetter)
				inkSetter.SetMethodInfo(svcInfo.MethodInfo(context.Background(), callMethod))
				recvMsg := remote.NewMessage(nil, ri, remote.Call, remote.Server)
				recvMsg.NewData(callMethod)
				sendMsg := remote.NewMessage(svcInfo.MethodInfo(context.Background(), callMethod).NewResult(), ri, remote.Reply, remote.Server)

				_, err := transHdlrFact.hdlr.OnMessage(ctx, recvMsg, sendMsg)
				test.Assert(t, strings.Contains(err.Error(), "happened in biz handler"))
			}
			<-exitCh
			return nil
		},
		ShutdownFunc: func() error {
			ln.Close()
			exitCh <- true
			return nil
		},
		CreateListenerFunc: func(addr net.Addr) (net.Listener, error) {
			var err error
			ln, err = net.Listen("tcp", "localhost:0")
			return ln, err
		},
	}
	opts = append(opts, WithTransServerFactory(mocks.NewMockTransServerFactory(transSvr)))
	opts = append(opts, WithTransHandlerFactory(transHdlrFact))

	svr := NewServer(opts...)
	time.AfterFunc(100*time.Millisecond, func() {
		err := svr.Stop()
		test.Assert(t, err == nil, err)
	})
	serviceHandler := false
	err := svr.RegisterService(mocks.ServiceInfo(), mocks.MockFuncHandler(func(ctx context.Context, req *mocks.MyRequest) (r *mocks.MyResponse, err error) {
		serviceHandler = true
		panic("test")
	}))
	test.Assert(t, err == nil)
	err = svr.Run()
	test.Assert(t, err == nil, err)
	test.Assert(t, mwExec)
	test.Assert(t, serviceHandler)
}

func TestRegisterService(t *testing.T) {
	{
		svr := NewServer()
		time.AfterFunc(time.Second, func() {
			err := svr.Stop()
			test.Assert(t, err == nil, err)
		})

		svr.Run()

		test.PanicAt(t, func() {
			_ = svr.RegisterService(mocks.ServiceInfo(), mocks.MyServiceHandler())
		}, func(err interface{}) bool {
			if errMsg, ok := err.(string); ok {
				return strings.Contains(errMsg, "server is running")
			}
			return true
		})
		svr.Stop()
	}

	{
		svr := NewServer()
		time.AfterFunc(time.Second, func() {
			err := svr.Stop()
			test.Assert(t, err == nil, err)
		})

		test.PanicAt(t, func() {
			_ = svr.RegisterService(nil, mocks.MyServiceHandler())
		}, func(err interface{}) bool {
			if errMsg, ok := err.(string); ok {
				return strings.Contains(errMsg, "svcInfo is nil")
			}
			return true
		})

		test.PanicAt(t, func() {
			_ = svr.RegisterService(mocks.ServiceInfo(), nil)
		}, func(err interface{}) bool {
			if errMsg, ok := err.(string); ok {
				return strings.Contains(errMsg, "handler is nil")
			}
			return true
		})

		test.PanicAt(t, func() {
			_ = svr.RegisterService(mocks.ServiceInfo(), mocks.MyServiceHandler(), WithFallbackService())
			_ = svr.RegisterService(mocks.ServiceInfo(), mocks.MyServiceHandler())
		}, func(err interface{}) bool {
			if errMsg, ok := err.(string); ok {
				return strings.Contains(errMsg, "service [MockService] has already been registered")
			}
			return true
		})

		test.PanicAt(t, func() {
			_ = svr.RegisterService(mocks.Service2Info(), mocks.MyServiceHandler(), WithFallbackService())
		}, func(err interface{}) bool {
			if errMsg, ok := err.(string); ok {
				return strings.Contains(errMsg, "multiple fallback services cannot be registered")
			}
			return true
		})
		svr.Stop()
	}

	{
		svr := NewServer()
		time.AfterFunc(time.Second, func() {
			err := svr.Stop()
			test.Assert(t, err == nil, err)
		})

		_ = svr.RegisterService(mocks.ServiceInfo(), mocks.MyServiceHandler())
		_ = svr.RegisterService(mocks.Service3Info(), mocks.MyServiceHandler())
		err := svr.Run()
		test.Assert(t, err != nil)
		test.Assert(t, err.Error() == "method name [mock] is conflicted between services but no fallback service is specified")
		svr.Stop()
	}
}

type noopMetahandler struct{}

func (noopMetahandler) WriteMeta(ctx context.Context, msg remote.Message) (context.Context, error) {
	return ctx, nil
}

func (noopMetahandler) ReadMeta(ctx context.Context, msg remote.Message) (context.Context, error) {
	return ctx, nil
}
func (noopMetahandler) OnConnectStream(ctx context.Context) (context.Context, error) { return ctx, nil }
func (noopMetahandler) OnReadStream(ctx context.Context) (context.Context, error)    { return ctx, nil }

type mockSvrTransHandlerFactory struct {
	hdlr remote.ServerTransHandler
}

func (f *mockSvrTransHandlerFactory) NewTransHandler(opt *remote.ServerOption) (remote.ServerTransHandler, error) {
	f.hdlr, _ = trans.NewDefaultSvrTransHandler(opt, &mockExtension{})
	return f.hdlr, nil
}

type mockExtension struct{}

func (m mockExtension) SetReadTimeout(ctx context.Context, conn net.Conn, cfg rpcinfo.RPCConfig, role remote.RPCRole) {
}

func (m mockExtension) NewWriteByteBuffer(ctx context.Context, conn net.Conn, msg remote.Message) remote.ByteBuffer {
	return remote.NewWriterBuffer(0)
}

func (m mockExtension) NewReadByteBuffer(ctx context.Context, conn net.Conn, msg remote.Message) remote.ByteBuffer {
	return remote.NewReaderBuffer(nil)
}

func (m mockExtension) ReleaseBuffer(buffer remote.ByteBuffer, err error) error {
	return nil
}

func (m mockExtension) IsTimeoutErr(err error) bool {
	return false
}

func (m mockExtension) IsRemoteClosedErr(err error) bool {
	return false
}

type mockCodec struct {
	EncodeFunc func(ctx context.Context, msg remote.Message, out remote.ByteBuffer) error
	DecodeFunc func(ctx context.Context, msg remote.Message, in remote.ByteBuffer) error
}

func (m *mockCodec) Name() string {
	return "Mock"
}

func (m *mockCodec) Encode(ctx context.Context, msg remote.Message, out remote.ByteBuffer) (err error) {
	if m.EncodeFunc != nil {
		return m.EncodeFunc(ctx, msg, out)
	}
	return
}

func (m *mockCodec) Decode(ctx context.Context, msg remote.Message, in remote.ByteBuffer) (err error) {
	if m.DecodeFunc != nil {
		return m.DecodeFunc(ctx, msg, in)
	}
	return
}

func TestDuplicatedRegisterInfoPanic(t *testing.T) {
	svcs := newServices()
	svcs.addService(mocks.ServiceInfo(), nil, &RegisterOptions{})
	s := &server{
		opt:  internal_server.NewOptions(nil),
		svcs: svcs,
	}
	s.init()

	test.Panic(t, func() {
		_ = s.RegisterService(mocks.ServiceInfo(), mocks.MyServiceHandler())
	})
}

func TestRunServiceWithoutSvcInfo(t *testing.T) {
	svr := NewServer()
	time.AfterFunc(100*time.Millisecond, func() {
		_ = svr.Stop()
	})
	err := svr.Run()
	test.Assert(t, err != nil)
	test.Assert(t, strings.Contains(err.Error(), "no service"))
}

func inboundDeepEqual(inbound1, inbound2 []remote.InboundHandler) bool {
	if len(inbound1) != len(inbound2) {
		return false
	}
	for i := 0; i < len(inbound1); i++ {
		if !bound.DeepEqual(inbound1[i], inbound2[i]) {
			return false
		}
	}
	return true
}

func withGRPCTransport() Option {
	return Option{F: func(o *internal_server.Options, di *utils.Slice) {
		o.RemoteOpt.SvrHandlerFactory = nphttp2.NewSvrTransHandlerFactory()
	}}
}

type mockStream struct {
	streaming.ServerStream
}

type mockGRPCStream struct {
	streaming.Stream
}

func (s *mockStream) GetGRPCStream() streaming.Stream {
	return &mockGRPCStream{}
}

type streamingMethodArg struct {
	methodName string
	mode       serviceinfo.StreamingMode
	streamHdlr serviceinfo.MethodHandler
}

func newStreamingServer(svcInfo *serviceinfo.ServiceInfo, mws []endpoint.Middleware) *server {
	var opts []Option
	for _, mw := range mws {
		opts = append(opts, WithMiddleware(mw))
	}
	svr := &server{
		svcs: newServices(),
		opt:  internal_server.NewOptions(opts),
	}
	svr.svcs.addService(svcInfo, nil, &RegisterOptions{})
	return svr
}

func TestStreamCtxDiverge(t *testing.T) {
	testcases := []struct {
		methodName string
		mode       serviceinfo.StreamingMode
	}{
		{
			methodName: "ClientStreaming",
			mode:       serviceinfo.StreamingClient,
		},
		{
			methodName: "ServerStreaming",
			mode:       serviceinfo.StreamingServer,
		},
		{
			methodName: "BidiStreaming",
			mode:       serviceinfo.StreamingBidirectional,
		},
	}

	testKey := "key"
	testVal := "val"
	testService := "test"
	streamHdlr := func(ctx context.Context, handler, arg, result interface{}) error {
		st, ok := arg.(*streaming.Args)
		test.Assert(t, ok)
		val, ok := st.Stream.Context().Value(testKey).(string)
		test.Assert(t, ok)
		test.Assert(t, val == testVal)
		return nil
	}

	var args []streamingMethodArg
	for _, tc := range testcases {
		args = append(args, streamingMethodArg{
			methodName: tc.methodName,
			mode:       tc.mode,
			streamHdlr: streamHdlr,
		})
	}
	mws := []endpoint.Middleware{
		// treat it as user middleware
		// user would modify the ctx here
		func(next endpoint.Endpoint) endpoint.Endpoint {
			return func(ctx context.Context, req, resp interface{}) (err error) {
				ctx = context.WithValue(ctx, testKey, testVal)
				return next(ctx, req, resp)
			}
		},
	}
	methods := make(map[string]serviceinfo.MethodInfo)
	for _, arg := range args {
		methods[arg.methodName] = serviceinfo.NewMethodInfo(arg.streamHdlr, nil, nil, false, serviceinfo.WithStreamingMode(arg.mode))
	}
	svcInfo := &serviceinfo.ServiceInfo{
		ServiceName: testService,
		Methods:     methods,
	}
	svr := newStreamingServer(svcInfo, mws)
	svr.buildInvokeChain(context.Background())

	for _, tc := range testcases {
		t.Run(tc.methodName, func(t *testing.T) {
			ri := svr.initOrResetRPCInfoFunc()(nil, nil)
			ink, ok := ri.Invocation().(rpcinfo.InvocationSetter)
			test.Assert(t, ok)
			ink.SetServiceName(testService)
			ink.SetMethodName(tc.methodName)
			ink.SetMethodInfo(svcInfo.MethodInfo(context.Background(), tc.methodName))
			ctx := rpcinfo.NewCtxWithRPCInfo(context.Background(), ri)
			mock := &mockStream{}
			err := svr.eps(ctx, &streaming.Args{ServerStream: mock}, nil)
			test.Assert(t, err == nil, err)
		})
	}
}
