// This is just a test that shows how to instantiate a gossip component
func TestNewGossipCryptoService(t *testing.T) {
	s1 := grpc.NewServer()
	s2 := grpc.NewServer()
	s3 := grpc.NewServer()

	ll1, _ := net.Listen("tcp", fmt.Sprintf("%s:%d", "", 5611))
	ll2, _ := net.Listen("tcp", fmt.Sprintf("%s:%d", "", 5612))
	ll3, _ := net.Listen("tcp", fmt.Sprintf("%s:%d", "", 5613))

	endpoint1 := "localhost:5611"
	endpoint2 := "localhost:5612"
	endpoint3 := "localhost:5613"

	g1 := NewGossipComponent(endpoint1, s1, []grpc.DialOption{grpc.WithInsecure()})
	g2 := NewGossipComponent(endpoint2, s2, []grpc.DialOption{grpc.WithInsecure()}, endpoint1)
	g3 := NewGossipComponent(endpoint3, s3, []grpc.DialOption{grpc.WithInsecure()}, endpoint1)
	go s1.Serve(ll1)
	go s2.Serve(ll2)
	go s3.Serve(ll3)

	time.Sleep(time.Second * 5)
	fmt.Println(g1.GetPeers())
	fmt.Println(g2.GetPeers())
	fmt.Println(g3.GetPeers())
	time.Sleep(time.Second)
}

// serverWithAuth builds a gRPC server, possibly with authentication if key / cert files are given.
func serverWithAuth(keyFile, certFile, caCertFile string) *grpc.Server {
	if keyFile == "" {
		return grpc.NewServer(grpc.MaxMsgSize(maxMsgSize)) // No auth.
	}
	log.Debug("Loading x509 key pair from key: %s cert: %s", keyFile, certFile)
	cert, err := tls.LoadX509KeyPair(certFile, keyFile)
	if err != nil {
		log.Fatalf("Failed to load x509 key pair: %s", err)
	}
	config := tls.Config{
		Certificates: []tls.Certificate{cert},
		ClientAuth:   tls.RequestClientCert,
	}
	if caCertFile != "" {
		cert, err := ioutil.ReadFile(caCertFile)
		if err != nil {
			log.Fatalf("Failed to read CA cert file: %s", err)
		}
		config.ClientCAs = x509.NewCertPool()
		if !config.ClientCAs.AppendCertsFromPEM(cert) {
			log.Fatalf("Failed to find any PEM certificates in CA cert")
		}
	}
	return grpc.NewServer(grpc.Creds(credentials.NewTLS(&config)), grpc.MaxMsgSize(maxMsgSize))
}

func main() {
	configfn := "nameserver.conf"
	data, err := ioutil.ReadFile(configfn)
	if err != nil {
		log.Fatalf("server: cannot load configuration file[%s] (%v)", configfn, err)
	}

	var conf config.Server
	if _, err := toml.Decode(string(data), &conf); err != nil {
		log.Fatalf("server: configuration file[%s] is not valid (%v)", configfn, err)
	}
	server := NewServer()
	for i, v := range conf.Disks {
		log.Infof("Adding %v to disks", v)
		server.registeredDisks = append(server.registeredDisks, &conf.Disks[i])
	}
	log.Infof("server: starting server...")

	lis, err := net.Listen("tcp", net.JoinHostPort(conf.Bind, conf.Port))

	if err != nil {
		log.Fatalf("server: failed to listen: %v", err)
	}

	log.Infof("server: listening on %s", net.JoinHostPort(conf.Bind, conf.Port))

	s := grpc.NewServer()
	pb.RegisterNameServer(s, server)
	log.Infof("server: ready to serve clients")
	s.Serve(lis)
}

// StartBroker starts a new Broker.
func (b *Broker) StartBroker() {
	fmt.Printf("Broker started.\n")

	pos := strings.Index(b.localAddr, ":")
	port := b.localAddr[pos:]

	listener, err := net.Listen("tcp", port)
	if err != nil {
		fmt.Printf("Error: %v\n", err)
	} else {
		fmt.Printf("Listener started on %v\n", port)
	}

	grpcServer := grpc.NewServer()
	pb.RegisterPubBrokerServer(grpcServer, b)
	pb.RegisterSubBrokerServer(grpcServer, b)
	pb.RegisterInterBrokerServer(grpcServer, b)
	b.connectToOtherBrokers()
	go b.handleMessages()
	go b.checkBusy()

	fmt.Printf("*** Ready to serve incoming requests. ***\n")
	err = grpcServer.Serve(listener)
	if err != nil {
		fmt.Printf("Error: %v\n", err)
	}
}

func main() {
	grpc.EnableTracing = false

	flag.Parse()
	lis, err := net.Listen("tcp", ":"+strconv.Itoa(*driverPort))
	if err != nil {
		grpclog.Fatalf("failed to listen: %v", err)
	}
	grpclog.Printf("worker listening at port %v", *driverPort)

	s := grpc.NewServer()
	stop := make(chan bool)
	testpb.RegisterWorkerServiceServer(s, &workerServer{
		stop:       stop,
		serverPort: *serverPort,
	})

	go func() {
		<-stop
		// Wait for 1 second before stopping the server to make sure the return value of QuitWorker is sent to client.
		// TODO revise this once server graceful stop is supported in gRPC.
		time.Sleep(time.Second)
		s.Stop()
	}()

	s.Serve(lis)
}

// This test makes sure the go rpc service works
func TestGoRPCTabletConn(t *testing.T) {
	// fake service
	service := tabletconntest.CreateFakeServer(t)

	// listen on a random port
	listener, err := net.Listen("tcp", ":0")
	if err != nil {
		t.Fatalf("Cannot listen: %v", err)
	}
	host := listener.Addr().(*net.TCPAddr).IP.String()
	port := listener.Addr().(*net.TCPAddr).Port

	// Create a gRPC server and listen on the port
	server := grpc.NewServer()
	grpcqueryservice.RegisterForTest(server, service)
	go server.Serve(listener)

	// Create a gRPC client connecting to the server
	ctx := context.Background()
	client, err := DialTablet(ctx, &pb.EndPoint{
		Host: host,
		PortMap: map[string]int32{
			"grpc": int32(port),
		},
	}, tabletconntest.TestKeyspace, tabletconntest.TestShard, 30*time.Second)
	if err != nil {
		t.Fatalf("dial failed: %v", err)
	}

	// run the test suite
	tabletconntest.TestSuite(t, client, service)

	// and clean up
	client.Close()
}

func startTLSCA(t *testing.T) {
	LogInit(ioutil.Discard, os.Stdout, os.Stdout, os.Stderr, os.Stdout)

	eca_s = NewECA()
	tlsca_s = NewTLSCA(eca_s)

	var opts []grpc.ServerOption
	creds, err := credentials.NewServerTLSFromFile(viper.GetString("server.tls.certfile"), viper.GetString("server.tls.keyfile"))
	if err != nil {
		t.Logf("Failed creating credentials for TLS-CA service: %s", err)
		t.Fail()
	}

	opts = []grpc.ServerOption{grpc.Creds(creds)}

	srv = grpc.NewServer(opts...)

	eca_s.Start(srv)
	tlsca_s.Start(srv)

	sock, err := net.Listen("tcp", viper.GetString("server.port"))
	if err != nil {
		t.Logf("Failed to start TLS-CA service: %s", err)
		t.Fail()
	}

	srv.Serve(sock)
}

// set up the GRPC server
func setupGRPCServer(signerConfig signer.Config) (*grpc.Server, net.Listener, error) {

	//RPC server setup
	kms := &api.KeyManagementServer{
		CryptoServices: signerConfig.CryptoServices,
	}
	ss := &api.SignerServer{
		CryptoServices: signerConfig.CryptoServices,
	}
	hs := ghealth.NewServer()

	lis, err := net.Listen("tcp", signerConfig.GRPCAddr)
	if err != nil {
		return nil, nil, fmt.Errorf("grpc server failed to listen on %s: %v",
			signerConfig.GRPCAddr, err)
	}

	creds := credentials.NewTLS(signerConfig.TLSConfig)
	opts := []grpc.ServerOption{grpc.Creds(creds)}
	grpcServer := grpc.NewServer(opts...)

	pb.RegisterKeyManagementServer(grpcServer, kms)
	pb.RegisterSignerServer(grpcServer, ss)
	healthpb.RegisterHealthServer(grpcServer, hs)

	// Set status for both of the grpc service "KeyManagement" and "Signer", these are
	// the only two we have at present, if we add more grpc service in the future,
	// we should add a new line for that service here as well.
	hs.SetServingStatus(notary.HealthCheckKeyManagement, healthpb.HealthCheckResponse_SERVING)
	hs.SetServingStatus(notary.HealthCheckSigner, healthpb.HealthCheckResponse_SERVING)

	return grpcServer, lis, nil
}

func TestReflectionEnd2end(t *testing.T) {
	// Start server.
	lis, err := net.Listen("tcp", "localhost:0")
	if err != nil {
		t.Fatalf("failed to listen: %v", err)
	}
	s := grpc.NewServer()
	pb.RegisterSearchServiceServer(s, &server{})
	// Register reflection service on s.
	Register(s)
	go s.Serve(lis)

	// Create client.
	conn, err := grpc.Dial(lis.Addr().String(), grpc.WithInsecure())
	if err != nil {
		t.Fatalf("cannot connect to server: %v", err)
	}
	defer conn.Close()

	c := rpb.NewServerReflectionClient(conn)
	stream, err := c.ServerReflectionInfo(context.Background())

	testFileByFilename(t, stream)
	testFileByFilenameError(t, stream)
	testFileContainingSymbol(t, stream)
	testFileContainingSymbolError(t, stream)
	testFileContainingExtension(t, stream)
	testFileContainingExtensionError(t, stream)
	testAllExtensionNumbersOfType(t, stream)
	testAllExtensionNumbersOfTypeError(t, stream)
	testListServices(t, stream)

	s.Stop()
}

func startTLSCA(t *testing.T) {
	CacheConfiguration() // Cache configuration
	ecaS = NewECA(nil)
	tlscaS = NewTLSCA(ecaS)

	var opts []grpc.ServerOption
	creds, err := credentials.NewServerTLSFromFile(viper.GetString("server.tls.cert.file"), viper.GetString("server.tls.key.file"))
	if err != nil {
		t.Logf("Failed creating credentials for TLS-CA service: %s", err)
		t.Fail()
	}

	opts = []grpc.ServerOption{grpc.Creds(creds)}

	srv = grpc.NewServer(opts...)

	ecaS.Start(srv)
	tlscaS.Start(srv)

	sock, err := net.Listen("tcp", viper.GetString("server.port"))
	if err != nil {
		t.Logf("Failed to start TLS-CA service: %s", err)
		t.Fail()
	}

	srv.Serve(sock)
}

func Start(host string, options ...RemotingOption) {

	lis, err := net.Listen("tcp", host)
	if err != nil {
		log.Fatalf("failed to listen: %v", err)
	}
	config := defaultRemoteConfig()
	for _, option := range options {
		option(config)
	}

	host = lis.Addr().String()
	log.Printf("Host is %v", host)
	actor.ProcessRegistry.RegisterHostResolver(remoteHandler)
	actor.ProcessRegistry.Host = host
	props := actor.
		FromProducer(newEndpointManager(config)).
		WithMailbox(actor.NewBoundedMailbox(1000, 100000))

	endpointManagerPID = actor.Spawn(props)

	s := grpc.NewServer(config.serverOptions...)
	messages.RegisterRemotingServer(s, &server{})
	log.Printf("Starting GAM server on %v.", host)
	go s.Serve(lis)
}

// NewNode creates a new raft node to use for tests
func NewNode(t *testing.T, clockSource *fakeclock.FakeClock, tc *cautils.TestCA, opts ...raft.NodeOptions) *TestNode {
	l, err := net.Listen("tcp", "127.0.0.1:0")
	require.NoError(t, err, "can't bind to raft service port")
	wrappedListener := NewWrappedListener(l)

	securityConfig, err := tc.NewNodeConfig(ca.ManagerRole)
	require.NoError(t, err)

	serverOpts := []grpc.ServerOption{grpc.Creds(securityConfig.ServerTLSCreds)}
	s := grpc.NewServer(serverOpts...)

	cfg := raft.DefaultNodeConfig()

	stateDir, err := ioutil.TempDir("", "test-raft")
	require.NoError(t, err, "can't create temporary state directory")

	keyRotator := NewSimpleKeyRotator(raft.EncryptionKeys{CurrentDEK: []byte("current")})
	newNodeOpts := raft.NodeOptions{
		ID:             securityConfig.ClientTLSCreds.NodeID(),
		Addr:           l.Addr().String(),
		Config:         cfg,
		StateDir:       stateDir,
		ClockSource:    clockSource,
		TLSCredentials: securityConfig.ClientTLSCreds,
		KeyRotator:     keyRotator,
	}

	if len(opts) > 1 {
		panic("more than one optional argument provided")
	}
	if len(opts) == 1 {
		newNodeOpts.JoinAddr = opts[0].JoinAddr
		if opts[0].Addr != "" {
			newNodeOpts.Addr = opts[0].Addr
		}
	}

	n := raft.NewNode(newNodeOpts)

	healthServer := health.NewHealthServer()
	api.RegisterHealthServer(s, healthServer)
	raft.Register(s, n)

	go func() {
		// After stopping, we should receive an error from Serve
		assert.Error(t, s.Serve(wrappedListener))
	}()

	healthServer.SetServingStatus("Raft", api.HealthCheckResponse_SERVING)

	return &TestNode{
		Node:           n,
		Listener:       wrappedListener,
		SecurityConfig: securityConfig,
		Address:        newNodeOpts.Addr,
		StateDir:       newNodeOpts.StateDir,
		Server:         s,
		KeyRotator:     keyRotator,
	}
}

// This test makes sure the go rpc service works
func TestGoRPCTabletConn(t *testing.T) {
	// fake service
	service := tabletconntest.CreateFakeServer(t)

	// listen on a random port
	listener, err := net.Listen("tcp", ":0")
	if err != nil {
		t.Fatalf("Cannot listen: %v", err)
	}
	host := listener.Addr().(*net.TCPAddr).IP.String()
	port := listener.Addr().(*net.TCPAddr).Port

	// Create a gRPC server and listen on the port
	server := grpc.NewServer()
	grpcqueryservice.RegisterForTest(server, service)
	go server.Serve(listener)

	// run the test suite
	tabletconntest.TestSuite(t, protocolName, &pb.EndPoint{
		Host: host,
		PortMap: map[string]int32{
			"grpc": int32(port),
		},
	}, service)
}

func TestInitGossipService(t *testing.T) {
	// Test whenever gossip service is indeed singleton
	grpcServer := grpc.NewServer()
	socket, error := net.Listen("tcp", fmt.Sprintf("%s:%d", "", 5611))
	assert.NoError(t, error)

	go grpcServer.Serve(socket)
	defer grpcServer.Stop()

	wg := sync.WaitGroup{}
	wg.Add(10)
	for i := 0; i < 10; i++ {
		go func() {
			InitGossipService("localhost:5611", grpcServer)
			wg.Done()
		}()
	}
	wg.Wait()

	defer GetGossipService().Stop()
	gossip := GetGossipService()

	for i := 0; i < 10; i++ {
		go func(gossipInstance GossipService) {
			assert.Equal(t, gossip, GetGossipService())
		}(gossip)
	}
}

// Serve serves stuff.
func Serve(
	registerFunc func(*grpc.Server),
	options ServeOptions,
	serveEnv ServeEnv,
) (retErr error) {
	defer func(start time.Time) { logServerFinished(start, retErr) }(time.Now())
	if registerFunc == nil {
		return ErrMustSpecifyRegisterFunc
	}
	if serveEnv.GRPCPort == 0 {
		serveEnv.GRPCPort = 7070
	}
	grpcServer := grpc.NewServer(
		grpc.MaxConcurrentStreams(math.MaxUint32),
		grpc.UnaryInterceptor(protorpclog.LoggingUnaryServerInterceptor),
	)
	registerFunc(grpcServer)
	if options.Version != nil {
		protoversion.RegisterAPIServer(grpcServer, protoversion.NewAPIServer(options.Version, protoversion.APIServerOptions{}))
	}
	listener, err := net.Listen("tcp", fmt.Sprintf(":%d", serveEnv.GRPCPort))
	if err != nil {
		return err
	}
	errC := make(chan error)
	go func() { errC <- grpcServer.Serve(listener) }()
	protolion.Info(
		&ServerStarted{
			Port: uint32(serveEnv.GRPCPort),
		},
	)
	return <-errC
}

func serveGRPC(l net.Listener) {
	s := grpc.NewServer(grpc.CustomCodec(&query.Codec{}))
	graph.RegisterDgraphServer(s, &grpcServer{})
	if err := s.Serve(l); err != nil {
		log.Fatalf("While serving gRpc request: %v", err)
	}
}

// NewVtctlPipe creates a new VtctlPipe based on the given topo server.
func NewVtctlPipe(t *testing.T, ts topo.Server) *VtctlPipe {
	// Register all vtctl commands
	servenvInitialized.Do(func() {
		// make sure we use the right protocol
		flag.Set("vtctl_client_protocol", "grpc")

		// Enable all query groups
		flag.Set("enable_queries", "true")
		servenv.FireRunHooks()
	})

	// Listen on a random port
	listener, err := net.Listen("tcp", ":0")
	if err != nil {
		t.Fatalf("Cannot listen: %v", err)
	}

	// Create a gRPC server and listen on the port
	server := grpc.NewServer()
	grpcvtctlserver.StartServer(server, ts)
	go server.Serve(listener)

	// Create a VtctlClient gRPC client to talk to the fake server
	client, err := vtctlclient.New(listener.Addr().String(), 30*time.Second)
	if err != nil {
		t.Fatalf("Cannot create client: %v", err)
	}

	return &VtctlPipe{
		listener: listener,
		client:   client,
		t:        t,
	}
}

func startPKI() {
	var opts []grpc.ServerOption
	if viper.GetBool("peer.pki.tls.enabled") {
		// TLS configuration
		creds, err := credentials.NewServerTLSFromFile(
			filepath.Join(viper.GetString("server.rootpath"), "tlsca.cert"),
			filepath.Join(viper.GetString("server.rootpath"), "tlsca.priv"),
		)
		if err != nil {
			panic("Failed creating credentials for OBC-CA: " + err.Error())
		}
		opts = []grpc.ServerOption{grpc.Creds(creds)}
	}

	fmt.Printf("open socket...\n")
	sockp, err := net.Listen("tcp", viper.GetString("server.port"))
	if err != nil {
		panic("Cannot open port: " + err.Error())
	}
	fmt.Printf("open socket...done\n")

	server = grpc.NewServer(opts...)

	eca.Start(server)
	tca.Start(server)
	tlsca.Start(server)

	fmt.Printf("start serving...\n")
	server.Serve(sockp)
}

// This test makes sure the go rpc service works
func testGRPCVTGateConn(t *testing.T, rpcOnlyInReply bool) {
	// fake service
	service := vtgateconntest.CreateFakeServer(t)

	// listen on a random port
	listener, err := net.Listen("tcp", ":0")
	if err != nil {
		t.Fatalf("Cannot listen: %v", err)
	}

	// Create a gRPC server and listen on the port
	server := grpc.NewServer()
	grpcvtgateservice.RegisterForTest(server, service)
	*vtgate.RPCErrorOnlyInReply = rpcOnlyInReply
	go server.Serve(listener)

	// Create a Go RPC client connecting to the server
	ctx := context.Background()
	client, err := dial(ctx, listener.Addr().String(), 30*time.Second)
	if err != nil {
		t.Fatalf("dial failed: %v", err)
	}

	// run the test suite
	vtgateconntest.TestSuite(t, client, service)

	// and clean up
	client.Close()
}

func newJoinNode(t *testing.T, id uint64, join string) *Node {
	l, err := net.Listen("tcp", "127.0.0.1:0")
	assert.NoError(t, err, "Can't bind to raft service port")
	s := grpc.NewServer()

	cfg := DefaultNodeConfig()
	cfg.Logger = raftLogger

	n, err := NewNode(id, l.Addr().String(), cfg, nil)
	assert.NoError(t, err, "Can't create raft node")
	n.Listener = l
	n.Server = s

	go n.Start()

	c, err := GetRaftClient(join, 100*time.Millisecond)
	assert.NoError(t, err, "Can't initiate connection with existing raft")

	resp, err := c.JoinRaft(n.Ctx, &NodeInfo{ID: id, Addr: l.Addr().String()})
	assert.NoError(t, err, "Can't join existing Raft")

	err = n.RegisterNodes(resp.Nodes)
	assert.NoError(t, err, "Can't add nodes to the local cluster list")

	Register(s, n)
	go s.Serve(l)

	time.Sleep(1 * time.Second)
	return n
}