package main

import (
	"io"
	"log"
	"net"
	"sync"
	"testing"
	"time"

	"code.justin.tv/release/trace/api"
	"code.justin.tv/release/trace/common"
	"golang.org/x/net/context"
	"google.golang.org/grpc"
	"google.golang.org/grpc/codes"
)

func TestFirehoseRoute(t *testing.T) {
	serverTx := make(chan *api.Transaction, 10)
	clientTx := make(chan *api.Transaction, 10)

	manager := newQueryManager()

	s := grpc.NewServer()
	api.RegisterTraceServer(s, newApiServer(manager))

	l, err := net.Listen("tcp", "localhost:0")
	if err != nil {
		t.Fatalf("net.Listen; err = %q", err)
	}
	defer l.Close()

	go func() {
		err := s.Serve(l)
		if err != nil {
			t.Logf("grpc.Server.Serve; err = %q", err)
		}
	}()

	var (
		readyMu sync.Mutex
		ready   bool
	)

	var wg sync.WaitGroup
	wg.Add(1)
	go func() {
		defer wg.Done()
		defer close(clientTx)

		cc, err := grpc.Dial(l.Addr().String(), grpc.WithBlock(), grpc.WithInsecure())
		if err != nil {
			t.Errorf("dial; err = %q", err)
			return
		}
		defer cc.Close()

		c := api.NewTraceClient(cc)

		ctx := context.Background()
		ctx, _ = context.WithTimeout(ctx, 2*time.Second)

		req := &api.FirehoseRequest{}

		resp, err := c.Firehose(ctx, req)
		if err != nil {
			t.Errorf("firehose; err = %q", err)
			return
		}

		for {
			tx, err := resp.Recv()
			if err != nil {
				if err == io.EOF {
					break
				}
				switch grpc.Code(err) {
				case codes.DeadlineExceeded:
				default:
					t.Errorf("recv; err = %q", err)
				}
				break
			}

			// TODO: find a better way of confirming that our query is registered
			readyMu.Lock()
			ready = true
			readyMu.Unlock()
			if cl := tx.GetClient(); cl != nil && cl.Name == "ready" {
				continue
			}

			clientTx <- tx
		}
	}()

	go filterQueries(serverTx, manager, 2)

	for {
		serverTx <- &api.Transaction{Client: &api.Service{Name: "ready"}}
		readyMu.Lock()
		if ready {
			readyMu.Unlock()
			break
		}
		readyMu.Unlock()
		time.Sleep(1 * time.Millisecond)
	}

	tests := []struct {
		Service string
		Method  string
		URIPath string

		Route string
	}{
		{Service: "code.justin.tv/web/discovery", Method: "GET", URIPath: "/debug/health",
			Route: "health"},
		{Service: "code.justin.tv/web/discovery", Method: "POST", URIPath: "/debug/health",
			Route: ""},
		{Service: "code.justin.tv/bogus/service", Method: "GET", URIPath: "/",
			Route: ""},
		{Service: "code.justin.tv/bogus/service", Method: "GET", URIPath: "",
			Route: ""},
	}

	log.SetFlags(log.Lmicroseconds)

	for _, tt := range tests {
		serverTx <- &api.Transaction{
			Root: &api.Call{
				Svc: &api.Service{Name: tt.Service},
				Params: &api.RPCParams{
					Http: &api.HTTPParams{
						Method:  common.Method(common.Method_value[tt.Method]),
						UriPath: tt.URIPath,
					},
				},
			},
		}

		tx := <-clientTx
		if have, want := tx.Root.Svc.Name, tt.Service; have != want {
			t.Errorf("service name; %q != %q", have, want)
		}
		if have, want := tx.Root.Params.Http.Method.String(), tt.Method; have != want {
			t.Errorf("HTTP method; %q != %q", have, want)
		}
		if have, want := tx.Root.Params.Http.UriPath, ""; have != want {
			t.Errorf("HTTP URI path; %q != %q", have, want)
		}
		if have, want := tx.Root.Params.Http.Route, tt.Route; have != want {
			t.Errorf("HTTP route; %q != %q", have, want)
		}
	}

	close(serverTx)

	wg.Wait()
}
