/*
 * Copyright 2024 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 test

import (
	"context"
	"fmt"
	"io"
	"net"
	"reflect"
	"strings"
	"sync"
	"testing"
	"time"

	"github.com/tidwall/gjson"

	"github.com/cloudwego/kitex/client/genericclient"
	kt "github.com/cloudwego/kitex/internal/mocks/thrift"
	"github.com/cloudwego/kitex/internal/test"
	"github.com/cloudwego/kitex/pkg/generic"
	"github.com/cloudwego/kitex/pkg/kerrors"
	"github.com/cloudwego/kitex/server"
)

var idl = "./idl/api.thrift"

func TestClientStreaming(t *testing.T) {
	ctx := context.Background()
	addr := test.GetLocalAddress()

	svr := initMockTestServer(new(StreamingTestImpl), addr)
	defer svr.Stop()

	cli := initStreamingClient(t, addr, idl, false)
	testClientStreaming(t, ctx, cli)

	// with dynamicgo
	cli = initStreamingClient(t, addr, idl, true)
	testClientStreaming(t, ctx, cli)
}

func TestServerStreaming(t *testing.T) {
	ctx := context.Background()
	addr := test.GetLocalAddress()

	svr := initMockTestServer(new(StreamingTestImpl), addr)
	defer svr.Stop()

	cli := initStreamingClient(t, addr, idl, false)
	testServerStreaming(t, ctx, cli)

	// with dynamicgo
	cli = initStreamingClient(t, addr, idl, true)
	testServerStreaming(t, ctx, cli)
}

func TestBidirectionalStreaming(t *testing.T) {
	ctx := context.Background()
	addr := test.GetLocalAddress()

	svr := initMockTestServer(new(StreamingTestImpl), addr)
	defer svr.Stop()

	cli := initStreamingClient(t, addr, idl, false)
	testBidirectionalStreaming(t, ctx, cli)

	// with dynamicgo
	cli = initStreamingClient(t, addr, idl, true)
	testBidirectionalStreaming(t, ctx, cli)
}

func TestUnary(t *testing.T) {
	ctx := context.Background()
	addr := test.GetLocalAddress()

	svr := initMockTestServer(new(StreamingTestImpl), addr)
	defer svr.Stop()

	cli := initStreamingClient(t, addr, idl, false)
	testUnary(t, ctx, cli)

	// with dynamicgo
	cli = initStreamingClient(t, addr, idl, true)
	testUnary(t, ctx, cli)
}

func TestBizException(t *testing.T) {
	ctx := context.Background()
	addr := test.GetLocalAddress()

	svr := initMockTestServer(new(StreamingTestImpl), addr)
	defer svr.Stop()

	cli := initStreamingClient(t, addr, idl, false)
	streamCli, err := genericclient.NewClientStreaming(ctx, cli, "EchoBizException")
	test.Assert(t, err == nil)
	for i := 0; i < 3; i++ {
		req := fmt.Sprintf(`{"message": "grpc client streaming generic %dth request"}`, i)
		err = streamCli.Send(req)
		test.Assert(t, err == nil, err)
		time.Sleep(100 * time.Millisecond)
	}
	resp, err := streamCli.CloseAndRecv()
	test.Assert(t, err != nil)
	bizStatusErr, ok := kerrors.FromBizStatusError(err)
	test.Assert(t, ok)
	test.Assert(t, bizStatusErr.BizStatusCode() == 404)
	test.Assert(t, bizStatusErr.BizMessage() == "not found")
	test.Assert(t, resp == nil)
}

func TestNoneStreaming(t *testing.T) {
	ctx := context.Background()
	addr := test.GetLocalAddress()

	svr := initMockTestServer(new(StreamingTestImpl), addr)
	defer svr.Stop()

	cli := initGenericClient(t, addr, idl, false)
	testNoneStreaming(t, ctx, cli)

	// with dynamicgo
	cli = initGenericClient(t, addr, idl, true)
	testNoneStreaming(t, ctx, cli)
}

func initStreamingClient(t *testing.T, addr, idl string, enableDynamicgo bool) genericclient.Client {
	g, err := getJsonThriftGeneric(idl, enableDynamicgo)
	test.Assert(t, err == nil)
	return newGenericStreamingClient(g, addr)
}

func initGenericClient(t *testing.T, addr, idl string, enableDynamicgo bool) genericclient.Client {
	g, err := getJsonThriftGeneric(idl, enableDynamicgo)
	test.Assert(t, err == nil)
	return newGenericClient(g, addr)
}

func getJsonThriftGeneric(idl string, enableDynamicgo bool) (generic.Generic, error) {
	var p generic.DescriptorProvider
	var err error
	if enableDynamicgo {
		p, err = generic.NewThriftFileProviderWithDynamicGo(idl)
		if err != nil {
			return nil, err
		}
	} else {
		p, err = generic.NewThriftFileProvider(idl)
		if err != nil {
			return nil, err
		}
	}
	g, err := generic.JSONThriftGeneric(p)
	if err != nil {
		return nil, err
	}
	return g, nil
}

func initMockTestServer(handler kt.TestService, address string) server.Server {
	addr, _ := net.ResolveTCPAddr("tcp", address)
	svr := newMockServer(handler, addr)
	test.WaitServerStart(addr.String())
	return svr
}

func testClientStreaming(t *testing.T, ctx context.Context, cli genericclient.Client) {
	streamCli, err := genericclient.NewClientStreaming(ctx, cli, "EchoClient")
	test.Assert(t, err == nil)
	for i := 0; i < 3; i++ {
		req := fmt.Sprintf(`{"message": "grpc client streaming generic %dth request"}`, i)
		err = streamCli.Send(req)
		test.Assert(t, err == nil, err)
		time.Sleep(100 * time.Millisecond)
	}
	resp, err := streamCli.CloseAndRecv()
	test.Assert(t, err == nil, err)
	strResp, ok := resp.(string)
	test.Assert(t, ok)
	fmt.Printf("clientStreaming message received: %v\n", strResp)
	test.Assert(t, reflect.DeepEqual(gjson.Get(strResp, "message").String(),
		"all message: grpc client streaming generic 0th request, grpc client streaming generic 1th request, grpc client streaming generic 2th request"))
}

func testServerStreaming(t *testing.T, ctx context.Context, cli genericclient.Client) {
	streamCli, err := genericclient.NewServerStreaming(ctx, cli, "EchoServer", `{"message": "grpc server streaming generic request"}`)
	test.Assert(t, err == nil, err)
	for {
		resp, err := streamCli.Recv()
		if err != nil {
			test.Assert(t, err == io.EOF, err)
			fmt.Println("serverStreaming message receive done")
			break
		} else {
			strResp, ok := resp.(string)
			test.Assert(t, ok)
			fmt.Printf("serverStreaming message received: %s\n", strResp)
			test.Assert(t, strings.Contains(gjson.Get(strResp, "message").String(), "grpc server streaming generic request ->"))
		}
		time.Sleep(100 * time.Millisecond)
	}
}

func testBidirectionalStreaming(t *testing.T, ctx context.Context, cli genericclient.Client) {
	streamCli, err := genericclient.NewBidirectionalStreaming(ctx, cli, "Echo")
	test.Assert(t, err == nil)

	wg := &sync.WaitGroup{}
	wg.Add(2)

	go func() {
		defer wg.Done()
		defer streamCli.Close()
		for i := 0; i < 3; i++ {
			req := fmt.Sprintf(`{"message": "grpc bidirectional streaming generic %dth request"}`, i)
			err = streamCli.Send(req)
			test.Assert(t, err == nil)
			fmt.Printf("Echo send: req = %s\n", req)
		}
	}()

	go func() {
		defer wg.Done()
		for {
			resp, err := streamCli.Recv()
			if err != nil {
				test.Assert(t, err == io.EOF, err)
				fmt.Println("bidirectionalStreaming message receive done")
				break
			} else {
				strResp, ok := resp.(string)
				test.Assert(t, ok)
				fmt.Printf("bidirectionalStreaming message received: %s\n", strResp)
				test.Assert(t, strings.Contains(gjson.Get(strResp, "message").String(), "th response"))
			}
			time.Sleep(100 * time.Millisecond)
		}
	}()
	wg.Wait()
}

func testUnary(t *testing.T, ctx context.Context, cli genericclient.Client) {
	resp, err := cli.GenericCall(ctx, "EchoUnary", `{"message": "unary request"}`)
	test.Assert(t, err == nil, err)
	strResp, ok := resp.(string)
	test.Assert(t, ok)
	test.Assert(t, reflect.DeepEqual(gjson.Get(strResp, "message").String(), "hello unary request"))
}

func testNoneStreaming(t *testing.T, ctx context.Context, cli genericclient.Client) {
	resp, err := cli.GenericCall(ctx, "EchoPingPong", `{"message": "ping pong request"}`)
	test.Assert(t, err == nil, err)
	strResp, ok := resp.(string)
	test.Assert(t, ok)
	test.Assert(t, reflect.DeepEqual(gjson.Get(strResp, "message").String(), "hello ping pong request"))
}
