package grpcdemo

import (
	"context"
	"fmt"
	"io"
	"net"
	"sync/atomic"
	"testing"
	"time"

	"google.golang.org/grpc"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/credentials/insecure"
	healthpb "google.golang.org/grpc/health/grpc_health_v1"
	pb "google.golang.org/grpc/interop/grpc_testing"
	"google.golang.org/grpc/metadata"
	"google.golang.org/grpc/status"
	"google.golang.org/grpc/test/bufconn"
	"google.golang.org/protobuf/proto"
)

type streams struct {
	pb.UnimplementedTestServiceServer
}

func (streams) StreamingOutputCall(r *pb.StreamingOutputCallRequest, s grpc.ServerStreamingServer[pb.StreamingOutputCallResponse]) error {
	for i, body := range []string{"a", "b"} {
		if err := s.Send(&pb.StreamingOutputCallResponse{Payload: &pb.Payload{Body: []byte(body)}}); err != nil {
			return err
		}
		if i == 0 && r.ResponseStatus != nil {
			return status.Error(codes.Code(r.ResponseStatus.Code), r.ResponseStatus.Message)
		}
	}
	return nil
}
func (streams) StreamingInputCall(s grpc.ClientStreamingServer[pb.StreamingInputCallRequest, pb.StreamingInputCallResponse]) error {
	var total int32
	for {
		r, err := s.Recv()
		if err == io.EOF {
			return s.SendAndClose(&pb.StreamingInputCallResponse{AggregatedPayloadSize: total})
		}
		if err != nil {
			return err
		}
		total += int32(len(r.GetPayload().GetBody()))
	}
}
func (streams) FullDuplexCall(s grpc.BidiStreamingServer[pb.StreamingOutputCallRequest, pb.StreamingOutputCallResponse]) error {
	for {
		r, err := s.Recv()
		if err == io.EOF {
			return nil
		}
		if err != nil {
			return err
		}
		if err = s.Send(&pb.StreamingOutputCallResponse{Payload: r.Payload}); err != nil {
			return err
		}
	}
}
func TestStreamingModesAndPartialError(t *testing.T) {
	lis := bufconn.Listen(1 << 20)
	var intercepted atomic.Int32
	interceptor := func(srv any, ss grpc.ServerStream, info *grpc.StreamServerInfo, next grpc.StreamHandler) error {
		md, _ := metadata.FromIncomingContext(ss.Context())
		if len(md.Get("x-demo")) != 1 || md.Get("x-demo")[0] != "test" {
			return status.Error(codes.Unauthenticated, "missing metadata")
		}
		intercepted.Add(1)
		return next(srv, ss)
	}
	server := grpc.NewServer(grpc.StreamInterceptor(interceptor))
	pb.RegisterTestServiceServer(server, streams{})
	done := make(chan error, 1)
	go func() { done <- server.Serve(lis) }()
	defer func() { server.Stop(); lis.Close(); <-done }()
	conn, err := grpc.NewClient("passthrough:///local", grpc.WithTransportCredentials(insecure.NewCredentials()), grpc.WithContextDialer(func(ctx context.Context, _ string) (net.Conn, error) { return lis.DialContext(ctx) }))
	if err != nil {
		t.Fatal(err)
	}
	defer conn.Close()
	c := pb.NewTestServiceClient(conn)
	ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
	defer cancel()
	denied, err := c.StreamingOutputCall(ctx, &pb.StreamingOutputCallRequest{})
	if err == nil {
		_, err = denied.Recv()
	}
	if status.Code(err) != codes.Unauthenticated {
		t.Fatal(err)
	}
	ctx = metadata.AppendToOutgoingContext(ctx, "x-demo", "test")
	output, err := c.StreamingOutputCall(ctx, &pb.StreamingOutputCallRequest{})
	if err != nil {
		t.Fatal(err)
	}
	var values []string
	for {
		m, e := output.Recv()
		if e == io.EOF {
			break
		}
		if e != nil {
			t.Fatal(e)
		}
		values = append(values, string(m.GetPayload().GetBody()))
	}
	if fmt.Sprint(values) != "[a b]" {
		t.Fatal(values)
	}
	t.Log("server stream:", values, "then EOF")
	input, err := c.StreamingInputCall(ctx)
	if err != nil {
		t.Fatal(err)
	}
	for _, body := range []string{"ab", "cde"} {
		if err := input.Send(&pb.StreamingInputCallRequest{Payload: &pb.Payload{Body: []byte(body)}}); err != nil {
			t.Fatal(err)
		}
	}
	total, err := input.CloseAndRecv()
	if err != nil || total.GetAggregatedPayloadSize() != 5 {
		t.Fatalf("%v %v", total, err)
	}
	t.Log("client stream total:", total.GetAggregatedPayloadSize())
	bidi, err := c.FullDuplexCall(ctx)
	if err != nil {
		t.Fatal(err)
	}
	if err = bidi.Send(&pb.StreamingOutputCallRequest{Payload: &pb.Payload{Body: []byte("hello")}}); err != nil {
		t.Fatal(err)
	}
	reply, err := bidi.Recv()
	if err != nil || string(reply.GetPayload().GetBody()) != "hello" {
		t.Fatalf("%v %v", reply, err)
	}
	if err = bidi.CloseSend(); err != nil {
		t.Fatal(err)
	}
	if _, err = bidi.Recv(); err != io.EOF {
		t.Fatal(err)
	}
	t.Log("bidi: hello, half-close, EOF")
	partial, err := c.StreamingOutputCall(ctx, &pb.StreamingOutputCallRequest{ResponseStatus: &pb.EchoStatus{Code: int32(codes.Internal), Message: "demo failure"}})
	if err != nil {
		t.Fatal(err)
	}
	first, err := partial.Recv()
	if err != nil || string(first.GetPayload().GetBody()) != "a" {
		t.Fatal(first, err)
	}
	_, err = partial.Recv()
	if status.Code(err) != codes.Internal {
		t.Fatal(err)
	}
	t.Log("partial stream: a, then", status.Code(err))
	if intercepted.Load() != 4 {
		t.Fatal(intercepted.Load())
	}
}
func Example_protobufWire() {
	b, err := proto.Marshal(&healthpb.HealthCheckRequest{Service: "svc"})
	if err != nil {
		panic(err)
	}
	fmt.Printf("%x\n", b)
	var got healthpb.HealthCheckRequest
	if err = proto.Unmarshal(b, &got); err != nil {
		panic(err)
	}
	fmt.Println(got.GetService())
	// Output:
	// 0a03737663
	// svc
}