/*
 * 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 client

import (
	"context"
	"errors"
	"net"
	"testing"

	"github.com/golang/mock/gomock"

	"github.com/cloudwego/gopkg/protocol/thrift"

	"github.com/cloudwego/kitex/internal/mocks"
	mocksdiscovery "github.com/cloudwego/kitex/internal/mocks/discovery"
	mocksproxy "github.com/cloudwego/kitex/internal/mocks/proxy"
	"github.com/cloudwego/kitex/internal/test"
	"github.com/cloudwego/kitex/pkg/discovery"
	"github.com/cloudwego/kitex/pkg/endpoint"
	"github.com/cloudwego/kitex/pkg/event"
	"github.com/cloudwego/kitex/pkg/kerrors"
	"github.com/cloudwego/kitex/pkg/proxy"
	"github.com/cloudwego/kitex/pkg/remote/codec/protobuf"
	"github.com/cloudwego/kitex/pkg/remote/trans/nphttp2/status"
	"github.com/cloudwego/kitex/pkg/rpcinfo"
	"github.com/cloudwego/kitex/pkg/rpcinfo/remoteinfo"
)

var (
	instance404 = []discovery.Instance{
		discovery.NewInstance("tcp", "localhost:404", 1, make(map[string]string)),
	}
	instance505 = discovery.NewInstance("tcp", "localhost:505", 1, make(map[string]string))

	ctx = func() context.Context {
		ctx := context.Background()
		ctx = context.WithValue(ctx, endpoint.CtxEventBusKey, event.NewEventBus())
		ctx = context.WithValue(ctx, endpoint.CtxEventQueueKey, event.NewQueue(10))
		return ctx
	}()
	tcpAddrStr = "127.0.0.1:9909"
)

func resolver404(ctrl *gomock.Controller) discovery.Resolver {
	resolver := mocksdiscovery.NewMockResolver(ctrl)
	resolver.EXPECT().Resolve(gomock.Any(), gomock.Any()).Return(discovery.Result{
		Cacheable: true,
		CacheKey:  "test",
		Instances: instance404,
	}, nil).AnyTimes()
	resolver.EXPECT().Diff(gomock.Any(), gomock.Any(), gomock.Any()).Return(discovery.Change{}, false).AnyTimes()
	resolver.EXPECT().Name().Return("middlewares_test").AnyTimes()
	resolver.EXPECT().Target(gomock.Any(), gomock.Any()).AnyTimes()
	return resolver
}

func TestNoResolver(t *testing.T) {
	svcInfo := mocks.ServiceInfo()
	cli, err := NewClient(svcInfo, WithDestService("destService"))
	test.Assert(t, err == nil)

	err = cli.Call(context.Background(), "mock", mocks.NewMockArgs(), mocks.NewMockResult())
	test.Assert(t, errors.Is(err, kerrors.ErrNoResolver))
}

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

	var invoked bool
	cli := newMockClient(t, ctrl).(*kcFinalizerClient)
	mw := newResolveMWBuilder(cli.lbf)(ctx)
	ep := func(ctx context.Context, request, response interface{}) error {
		invoked = true
		return nil
	}

	to := remoteinfo.NewRemoteInfo(&rpcinfo.EndpointBasicInfo{}, "")
	ri := rpcinfo.NewRPCInfo(nil, to, rpcinfo.NewInvocation("", ""), nil, rpcinfo.NewRPCStats())

	ctx := rpcinfo.NewCtxWithRPCInfo(context.Background(), ri)
	req := new(MockTStruct)
	res := new(MockTStruct)
	err := mw(ep)(ctx, req, res)
	test.Assert(t, err == nil)
	test.Assert(t, invoked)
	test.Assert(t, to.GetInstance() == instance404[0])
}

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

	resolver := &discovery.SynthesizedResolver{
		ResolveFunc: func(ctx context.Context, key string) (discovery.Result, error) {
			return discovery.Result{}, nil
		},
		NameFunc: func() string { return t.Name() },
	}
	var invoked bool
	cli := newMockClient(t, ctrl, WithResolver(resolver)).(*kcFinalizerClient)
	mw := newResolveMWBuilder(cli.lbf)(ctx)
	ep := func(ctx context.Context, request, response interface{}) error {
		invoked = true
		return nil
	}

	to := remoteinfo.NewRemoteInfo(&rpcinfo.EndpointBasicInfo{}, "")
	ri := rpcinfo.NewRPCInfo(nil, to, rpcinfo.NewInvocation("", ""), nil, rpcinfo.NewRPCStats())

	ctx := rpcinfo.NewCtxWithRPCInfo(context.Background(), ri)
	req := new(MockTStruct)
	res := new(MockTStruct)

	err := mw(ep)(ctx, req, res)
	test.Assert(t, err != nil, err)
	test.Assert(t, errors.Is(err, kerrors.ErrNoMoreInstance))
	test.Assert(t, to.GetInstance() == nil)
	test.Assert(t, !invoked)
}

func TestDefaultErrorHandler(t *testing.T) {
	tcpAddr, _ := net.ResolveTCPAddr("tcp", tcpAddrStr)
	ri := rpcinfo.NewRPCInfo(nil, rpcinfo.NewEndpointInfo("mockService", "mockMethod", tcpAddr, nil),
		rpcinfo.NewInvocation("", ""), nil, rpcinfo.NewRPCStats())
	reqCtx := rpcinfo.NewCtxWithRPCInfo(context.Background(), ri)

	// Test TApplicationException
	err := DefaultClientErrorHandler(context.Background(), thrift.NewApplicationException(100, "mock"))
	test.Assert(t, err.Error() == "remote or network error[remote]: mock", err.Error())
	var te *thrift.ApplicationException
	ok := errors.As(err, &te)
	test.Assert(t, ok)
	test.Assert(t, te.TypeID() == 100)
	// Test TApplicationException with remote addr
	err = ClientErrorHandlerWithAddr(reqCtx, thrift.NewApplicationException(100, "mock"))
	test.Assert(t, err.Error() == "remote or network error[remote-"+tcpAddrStr+"]: mock", err.Error())
	ok = errors.As(err, &te)
	test.Assert(t, ok)
	test.Assert(t, te.TypeID() == 100)

	// Test PbError
	err = DefaultClientErrorHandler(context.Background(), protobuf.NewPbError(100, "mock"))
	test.Assert(t, err.Error() == "remote or network error[remote]: mock")
	var pe protobuf.PBError
	ok = errors.As(err, &pe)
	test.Assert(t, ok)
	test.Assert(t, te.TypeID() == 100)
	// Test PbError with remote addr
	err = ClientErrorHandlerWithAddr(reqCtx, protobuf.NewPbError(100, "mock"))
	test.Assert(t, err.Error() == "remote or network error[remote-"+tcpAddrStr+"]: mock", err.Error())
	ok = errors.As(err, &pe)
	test.Assert(t, ok)
	test.Assert(t, te.TypeID() == 100)

	// Test status.Error
	err = DefaultClientErrorHandler(context.Background(), status.Err(100, "mock"))
	test.Assert(t, err.Error() == "remote or network error: rpc error: code = 100 desc = mock", err.Error())
	// Test status.Error with remote addr
	err = ClientErrorHandlerWithAddr(reqCtx, status.Err(100, "mock"))
	test.Assert(t, err.Error() == "remote or network error["+tcpAddrStr+"]: rpc error: code = 100 desc = mock", err.Error())

	// Test other error
	err = DefaultClientErrorHandler(context.Background(), errors.New("mock"))
	test.Assert(t, err.Error() == "remote or network error: mock")
	// Test other error with remote addr
	err = ClientErrorHandlerWithAddr(reqCtx, errors.New("mock"))
	test.Assert(t, err.Error() == "remote or network error["+tcpAddrStr+"]: mock")
}

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

	p := mocksproxy.NewMockForwardProxy(ctrl)
	p.EXPECT().ResolveProxyInstance(gomock.Any()).Return(nil).Times(1)

	mw := newProxyMW(p)

	err := mw(func(ctx context.Context, req, resp interface{}) (err error) {
		return nil
	})(context.Background(), nil, nil)
	test.Assert(t, err == nil)

	tp := mocksproxy.NewMockWithMiddleware(ctrl)
	tp.EXPECT().ProxyMiddleware().DoAndReturn(func() endpoint.Middleware {
		return func(e endpoint.Endpoint) endpoint.Endpoint {
			return func(ctx context.Context, req, resp interface{}) (err error) {
				return errors.New("execute proxy middleware")
			}
		}
	}).Times(1)
	mw = newProxyMW(&struct {
		proxy.ForwardProxy
		proxy.WithMiddleware
	}{
		WithMiddleware: tp,
	})
	err = mw(func(ctx context.Context, req, resp interface{}) (err error) {
		return nil
	})(context.Background(), nil, nil)
	test.Assert(t, err.Error() == "execute proxy middleware")
}

func BenchmarkResolverMW(b *testing.B) {
	ctrl := gomock.NewController(b)
	defer ctrl.Finish()

	cli := newMockClient(b, ctrl).(*kcFinalizerClient)
	mw := newResolveMWBuilder(cli.lbf)(ctx)
	ep := func(ctx context.Context, request, response interface{}) error { return nil }
	ri := rpcinfo.NewRPCInfo(nil, nil, rpcinfo.NewInvocation("", ""), nil, rpcinfo.NewRPCStats())

	ctx := rpcinfo.NewCtxWithRPCInfo(context.Background(), ri)
	req := new(MockTStruct)
	res := new(MockTStruct)

	b.ResetTimer()
	for i := 0; i < b.N; i++ {
		mw(ep)(ctx, req, res)
	}
}

func BenchmarkResolverMWParallel(b *testing.B) {
	ctrl := gomock.NewController(b)
	defer ctrl.Finish()

	cli := newMockClient(b, ctrl).(*kcFinalizerClient)
	mw := newResolveMWBuilder(cli.lbf)(ctx)
	ep := func(ctx context.Context, request, response interface{}) error { return nil }
	ri := rpcinfo.NewRPCInfo(nil, nil, rpcinfo.NewInvocation("", ""), nil, rpcinfo.NewRPCStats())

	ctx := rpcinfo.NewCtxWithRPCInfo(context.Background(), ri)
	req := new(MockTStruct)
	res := new(MockTStruct)

	b.ResetTimer()
	b.RunParallel(func(pb *testing.PB) {
		for pb.Next() {
			mw(ep)(ctx, req, res)
		}
	})
}

func TestDiscoveryEventHandler(t *testing.T) {
	bus, queue := event.NewEventBus(), event.NewQueue(200)
	h := discoveryEventHandler(discovery.ChangeEventName, bus, queue)
	ins := []discovery.Instance{discovery.NewInstance("tcp", "addr", 10, nil)}
	cacheKey := "testCacheKey"
	c := &discovery.Change{
		Result: discovery.Result{
			Cacheable: true,
			CacheKey:  cacheKey,
		},
		Added: ins,
	}
	h(c)
	events := queue.Dump().([]*event.Event)
	test.Assert(t, len(events) == 1)
	extra, ok := events[0].Extra.(map[string]interface{})
	test.Assert(t, ok)
	test.Assert(t, extra["Cacheable"] == true)
	test.Assert(t, extra["CacheKey"] == cacheKey)
	added := extra["Added"].([]*instInfo)
	test.Assert(t, len(added) == 1)
}
