package grpc

import (
	"context"
	"fmt"
	"github.com/UpmeshLTD/urus-aio/infrastructure/patterns"
	"github.com/UpmeshLTD/urus-aio/infrastructure/server/transport/grpc/test/pb"
	"io"
	"time"


	. "github.com/onsi/ginkgo/v2"
	. "github.com/onsi/gomega"
	"github.com/phayes/freeport"
	"google.golang.org/grpc"
	"google.golang.org/grpc/credentials/insecure"
)

type serviceImpl struct {
	pb.UnimplementedTestServiceServer
	manager patterns.EventManager
}

type serviceEvent struct {
	subscriberID int32
}

func (s *serviceImpl) Subscribe(req *pb.Request, ss pb.TestService_SubscribeServer) error {
	done := make(chan bool)

	_ = s.manager.RegisterEventHandler(serviceEvent{subscriberID: req.Id}, patterns.EventHandlerFunc(func(event interface{}, data interface{}) error {
		if stop, ok := data.(bool); ok && stop {
			close(done)
		}

		return ss.Send(&pb.Response{Data: "test"})
	}))

	// Do not return until sending is done to avoid premature closing of the channel.
	<-done
	return nil
}

func (s *serviceImpl) Foo(ctx context.Context, request *pb.Request) (*pb.Response, error) {
	return &pb.Response{Data: fmt.Sprintf("input-%d", request.Id)}, nil
}

var _ = Describe("Transport", func() {
	var (
		port         int
		t            Transport
		registerFunc = RegisterFunc(func(svr *grpc.Server) error {
			return nil
		})
	)

	BeforeEach(func() {
		port = freeport.GetPort()
	})

	JustBeforeEach(func() {
		t = NewTransport(&Config{Port: port}, registerFunc)
	})

	It("should start and stop", func() {
		var err error

		go func() {
			err = t.StartWithContext(context.TODO())
		}()

		time.Sleep(time.Millisecond * 100)
		Expect(err).Should(BeNil())

		// Ensure the connection is opened.
		conn, err := grpc.Dial(
			fmt.Sprintf("localhost:%d", port),
			grpc.WithTransportCredentials(insecure.NewCredentials()))

		Expect(err).Should(BeNil())
		Expect(conn).ShouldNot(BeNil())
		_ = conn.Close()

		Expect(t.Stop()).Should(BeNil())
		time.Sleep(time.Millisecond * 100)

		Expect(err).Should(BeNil())
	})

	Describe("with registered service", func() {
		BeforeEach(func() {
			registerFunc = func(svr *grpc.Server) error {
				pb.RegisterTestServiceServer(svr, &serviceImpl{})
				return nil
			}
		})

		It("should serve registered service to gRPC client", func() {
			go func() {
				_ = t.StartWithContext(context.TODO())
			}()

			conn, err := grpc.Dial(
				fmt.Sprintf("localhost:%d", port),
				grpc.WithTransportCredentials(insecure.NewCredentials()))

			Expect(err).Should(BeNil())
			Expect(conn).ShouldNot(BeNil())

			client := pb.NewTestServiceClient(conn)
			resp, err := client.Foo(context.TODO(), &pb.Request{Id: 123})

			Expect(err).Should(BeNil())
			Expect(resp.Data).Should(Equal(fmt.Sprintf("input-123")))

			_ = conn.Close()
			_ = t.Stop()
		})
	})

	Describe("integration with event manager", func() {
		var (
			svc     *serviceImpl
			manager patterns.EventManager
			client  pb.TestServiceClient
		)

		type testCase struct {
			event interface{}
			data  interface{}
		}

		BeforeEach(func() {
			port = freeport.GetPort()
			registerFunc = func(svr *grpc.Server) error {
				pb.RegisterTestServiceServer(svr, svc)
				return nil
			}

			manager = patterns.NewEventManager()
			svc = &serviceImpl{
				manager: manager,
			}
		})

		JustBeforeEach(func() {
			go func() {
				_ = t.StartWithContext(context.TODO())
			}()

			time.Sleep(time.Millisecond * 100)
			conn, _ := grpc.Dial(
				fmt.Sprintf("localhost:%d", port),
				grpc.WithTransportCredentials(insecure.NewCredentials()))

			client = pb.NewTestServiceClient(conn)
		})

		JustAfterEach(func() {
			_ = t.Stop()
		})

		DescribeTable("should stream events", func(tc []testCase, expectedCount int) {
			done := make(chan bool)
			sc, _ := client.Subscribe(context.TODO(), &pb.Request{Id: 1})

			count := 0
			go func() {
				for {
					resp, err := sc.Recv()
					if err == io.EOF {
						done <- true //means stream is finished
						return
					}

					if resp.Data == "test" {
						count++
					}
				}
			}()

			time.Sleep(time.Millisecond * 100)

			// Demonstrating that events can be raised separately.
			for _, t := range tc {
				_ = manager.Raise(t.event, t.data)
			}

			<-done

			Expect(count).Should(Equal(expectedCount))
		},
			Entry("case", []testCase{
				{serviceEvent{subscriberID: 1}, true},
			}, 1),
			Entry("case", []testCase{
				{serviceEvent{subscriberID: 1}, false},
				{serviceEvent{subscriberID: 1}, true},
			}, 2),
			Entry("case", []testCase{
				{serviceEvent{subscriberID: 1}, false},
				{serviceEvent{subscriberID: 2}, true},
				{serviceEvent{subscriberID: 1}, true},
			}, 2),
		)
	})
})