/*
 * 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"
	"strings"
	"sync"

	"github.com/cloudwego/kitex/client"
	"github.com/cloudwego/kitex/client/genericclient"
	"github.com/cloudwego/kitex/internal/mocks/proto/kitex_gen/pbapi/mock"
	"github.com/cloudwego/kitex/pkg/generic"
	"github.com/cloudwego/kitex/server"
	"github.com/cloudwego/kitex/transport"
)

var startChan chan struct{}

func init() {
	startChan = make(chan struct{})
	server.RegisterStartHook(func() {
		startChan <- struct{}{}
	})
}

func newGenericClient(g generic.Generic, targetIPPort string, cliOpts ...client.Option) genericclient.Client {
	cliOpts = append(cliOpts, client.WithTransportProtocol(transport.GRPC), client.WithHostPorts(targetIPPort))
	cli, err := genericclient.NewStreamingClient("destService", g, cliOpts...)
	if err != nil {
		panic(err)
	}
	return cli
}

func newMockTestServer(handler mock.Mock, addr net.Addr, opts ...server.Option) server.Server {
	opts = append(opts, server.WithServiceAddr(addr))
	svr := mock.NewServer(handler, opts...)
	go func() {
		err := svr.Run()
		if err != nil {
			panic(err)
		}
	}()
	// wait for server starting to avoid data race
	<-startChan
	return svr
}

var _ mock.Mock = &StreamingTestImpl{}

type StreamingTestImpl struct{}

func (s *StreamingTestImpl) UnaryTest(ctx context.Context, req *mock.MockReq) (resp *mock.MockResp, err error) {
	fmt.Println("UnaryTest called")
	resp = &mock.MockResp{}
	resp.Message = "hello " + req.Message
	return
}

func (s *StreamingTestImpl) ClientStreamingTest(stream mock.Mock_ClientStreamingTestServer) (err error) {
	fmt.Println("ClientStreamingTest called")
	var msgs []string
	for {
		req, err := stream.Recv()
		if err != nil {
			if err == io.EOF {
				break
			}
			return err
		}
		fmt.Printf("Recv: %s\n", req.Message)
		msgs = append(msgs, req.Message)
	}
	return stream.SendAndClose(&mock.MockResp{Message: "all message: " + strings.Join(msgs, ", ")})
}

func (s *StreamingTestImpl) ServerStreamingTest(req *mock.MockReq, stream mock.Mock_ServerStreamingTestServer) (err error) {
	fmt.Println("ServerStreamingTest called")
	resp := &mock.MockResp{}
	for i := 0; i < 3; i++ {
		resp.Message = fmt.Sprintf("%v -> %dth response", req.Message, i)
		err := stream.Send(resp)
		if err != nil {
			return err
		}
	}
	return
}

func (s *StreamingTestImpl) BidirectionalStreamingTest(stream mock.Mock_BidirectionalStreamingTestServer) (err error) {
	fmt.Println("BidirectionalStreamingTest called")
	wg := &sync.WaitGroup{}
	wg.Add(2)

	go func() {
		defer func() {
			if p := recover(); p != nil {
				err = fmt.Errorf("panic: %v", p)
			}
			wg.Done()
		}()
		for {
			msg, recvErr := stream.Recv()
			if recvErr == io.EOF {
				return
			} else if recvErr != nil {
				err = recvErr
				return
			}
			fmt.Printf("BidirectionaStreamingTest: received message = %s\n", msg.Message)
		}
	}()

	go func() {
		defer func() {
			if p := recover(); p != nil {
				err = fmt.Errorf("panic: %v", p)
			}
			wg.Done()
		}()
		resp := &mock.MockResp{}
		for i := 0; i < 3; i++ {
			resp.Message = fmt.Sprintf("%dth response", i)
			if sendErr := stream.Send(resp); sendErr != nil {
				err = sendErr
				return
			}
			fmt.Printf("BidirectionaStreamingTest: sent message = %s\n", resp)
		}
	}()
	wg.Wait()
	return
}
