func Test_AnIncomingMessageIsNotAllowed(t *testing.T) {
	ctrl, finish := testutil.NewMockCtrl(t)
	defer finish()

	wsconn, routerMock, _ := createDefaultMocks([]string{})

	tam := NewMockAccessManager(ctrl)
	tam.EXPECT().IsAllowed(auth.READ, "testuser", protocol.Path("/foo")).Return(false)
	handler := NewWebSocket(
		testWSHandler(routerMock, tam),
		wsconn,
		"testuser",
	)
	go func() {
		handler.Start()
	}()
	time.Sleep(time.Millisecond * 2)

	handler.sendChannel <- aTestMessage.Bytes()
	time.Sleep(time.Millisecond * 2)
	//nothing shall have been sent

	//now allow
	tam.EXPECT().IsAllowed(auth.READ, "testuser", protocol.Path("/foo")).Return(true)

	wsconn.EXPECT().Send(aTestMessage.Bytes())

	time.Sleep(time.Millisecond * 2)

	handler.sendChannel <- aTestMessage.Bytes()
	time.Sleep(time.Millisecond * 2)
}

func TestRouter_ReplacingOfRoutesMatchingAppID(t *testing.T) {
	a := assert.New(t)

	// Given a Router with a route
	router, _, _, _ := aStartedRouter()

	matcherFunc := func(route, other RouteConfig, keys ...string) bool {
		return route.Path == other.Path && route.Get("application_id") == other.Get("application_id")
	}
	router.Subscribe(NewRoute(
		RouteConfig{
			RouteParams: RouteParams{"application_id": "appid01", "user_id": "user01"},
			Path:        protocol.Path("/blah"),
			Matcher:     matcherFunc,
		},
	))

	// when: i add another route with the same Application Id and Same Path
	router.Subscribe(NewRoute(
		RouteConfig{
			RouteParams: RouteParams{"application_id": "appid01", "user_id": "newUserId"},
			Path:        protocol.Path("/blah"),
			Matcher:     matcherFunc,
		},
	))

	// then: the router only contains the new route
	a.Equal(1, len(router.routes))
	a.Equal(1, len(router.routes["/blah"]))
	a.Equal("newUserId", router.routes["/blah"][0].Get("user_id"))
}

func TestNexmoSender_SendWithError(t *testing.T) {
	a := assert.New(t)
	sender, err := NewNexmoSender(KEY, SECRET)
	a.NoError(err)

	sms := NexmoSms{
		To:   "toNumber",
		From: "FromNUmber",
		Text: "body",
	}
	d, err := json.Marshal(&sms)
	a.NoError(err)

	msg := protocol.Message{
		Path:          protocol.Path(SMSDefaultTopic),
		UserID:        "samsa",
		ApplicationID: "sms",
		ID:            uint64(4),
		Body:          d,
	}

	err = sender.Send(&msg)
	a.Error(err)
	a.Equal(ErrIncompleteSMSSent, err)
}

func (c *conn) initRoute() {
	c.route = router.NewRoute(router.RouteConfig{
		Path:         protocol.Path(*c.config.SMSTopic),
		ChannelSize:  10,
		FetchRequest: c.fetchRequest(),
	})
}

// Post creates a new subscriber
func (c *connector) Post(w http.ResponseWriter, req *http.Request) {
	params := mux.Vars(req)
	log.WithField("params", params).Debug("Create subscription")

	topic, ok := params[TopicParam]
	if !ok {
		fmt.Fprintf(w, "Missing topic parameter.")
		return
	}
	delete(params, TopicParam)

	subscriber, err := c.manager.Create(protocol.Path("/"+topic), params)
	if err != nil {
		if err == ErrSubscriberExists {
			fmt.Fprintf(w, `{"error":"subscription already exists"}`)
		} else {
			http.Error(w, fmt.Sprintf(`{"error":"unknown error: %s"}`, err.Error()), http.StatusInternalServerError)
		}
		return
	}

	go c.Run(subscriber)
	log.WithField("topic", topic).Debug("Subscription created")
	fmt.Fprintf(w, `{"subscribed":"/%v"}`, topic)
}

func (api *RestMessageAPI) ServeHTTP(w http.ResponseWriter, r *http.Request) {
	if r.Method != http.MethodPost {
		http.Error(w, "Method not allowed", http.StatusMethodNotAllowed)
		return
	}

	body, err := ioutil.ReadAll(r.Body)
	if err != nil {
		http.Error(w, `Can not read body`, http.StatusBadRequest)
		return
	}

	topic, err := api.extractTopic(r.URL.Path)
	if err != nil {
		if err == errNotFound {
			http.NotFound(w, r)
			return
		}
		http.Error(w, "Server error.", http.StatusInternalServerError)
		return
	}

	msg := &protocol.Message{
		Path:          protocol.Path(topic),
		Body:          body,
		UserID:        q(r, `userId`),
		ApplicationID: xid.New().String(),
		OptionalID:    q(r, `messageId`),
		HeaderJSON:    headersToJSON(r.Header),
		NodeID:        *config.Cluster.NodeID,
	}

	api.router.HandleMessage(msg)
}

// Ensure the subscription is started when posting
func TestConnector_PostSubscription(t *testing.T) {
	_, finish := testutil.NewMockCtrl(t)
	defer finish()

	a := assert.New(t)

	recorder := httptest.NewRecorder()
	conn, mocks := getTestConnector(t, Config{
		Name:       "test",
		Schema:     "test",
		Prefix:     "/connector/",
		URLPattern: "/{device_token}/{user_id}/{topic:.*}",
	}, true, false)

	mocks.manager.EXPECT().Load().Return(nil)
	mocks.manager.EXPECT().List().Return(make([]Subscriber, 0))
	err := conn.Start()
	a.NoError(err)
	defer conn.Stop()

	subscriber := NewMockSubscriber(testutil.MockCtrl)
	mocks.manager.EXPECT().Create(gomock.Eq(protocol.Path("/topic1")), gomock.Eq(router.RouteParams{
		"device_token": "device1",
		"user_id":      "user1",
	})).Return(subscriber, nil)

	subscriber.EXPECT().Loop(gomock.Any(), gomock.Any())
	r := router.NewRoute(router.RouteConfig{
		Path: protocol.Path("topic1"),
		RouteParams: router.RouteParams{
			"device_token": "device1",
			"user_id":      "user1",
		},
	})
	subscriber.EXPECT().Route().Return(r)
	mocks.router.EXPECT().Subscribe(gomock.Eq(r)).Return(r, nil)

	req, err := http.NewRequest(http.MethodPost, "/connector/device1/user1/topic1", strings.NewReader(""))
	a.NoError(err)
	conn.ServeHTTP(recorder, req)
	a.Equal(`{"subscribed":"/topic1"}`, recorder.Body.String())
	time.Sleep(100 * time.Millisecond)
}

func (c *conn) fetchRequest() (fr *store.FetchRequest) {
	if c.LastIDSent > 0 {
		fr = store.NewFetchRequest(
			protocol.Path(*c.config.SMSTopic).Partition(),
			c.LastIDSent+1,
			0,
			store.DirectionForward, -1)
	}
	return
}

func aRouterRoute(unused int) (*router, *Route) {
	router, _, _, _ := aStartedRouter()
	route, _ := router.Subscribe(NewRoute(
		RouteConfig{
			RouteParams: RouteParams{"application_id": "appid01", "user_id": "user01"},
			Path:        protocol.Path("/blah"),
			ChannelSize: chanSize,
		},
	))
	return router, route
}

func (ws *WebSocket) handleCancelCmd(cmd *protocol.Cmd) {
	if len(cmd.Arg) == 0 {
		ws.sendError(protocol.ERROR_BAD_REQUEST, "- command requires a path argument, but none given")
		return
	}
	path := protocol.Path(cmd.Arg)
	rec, exist := ws.receivers[path]
	if exist {
		rec.Stop()
		delete(ws.receivers, path)
	}
}

func TestRouter_SubscribeNotAllowed(t *testing.T) {
	ctrl, finish := testutil.NewMockCtrl(t)
	defer finish()
	a := assert.New(t)

	am := NewMockAccessManager(ctrl)
	msMock := NewMockMessageStore(ctrl)
	kvsMock := NewMockKVStore(ctrl)

	am.EXPECT().IsAllowed(auth.READ, "user01", protocol.Path("/blah")).Return(false)

	router := New(am, msMock, kvsMock, nil).(*router)
	router.Start()

	_, e := router.Subscribe(NewRoute(
		RouteConfig{
			RouteParams: RouteParams{"application_id": "appid01", "user_id": "user01"},
			Path:        protocol.Path("/blah"),
			ChannelSize: chanSize,
		},
	))

	// default TestAccessManager denies all
	a.NotNil(e)

	// now add permissions
	am.EXPECT().IsAllowed(auth.READ, "user01", protocol.Path("/blah")).Return(true)

	// and user shall be allowed to subscribe
	_, e = router.Subscribe(NewRoute(
		RouteConfig{
			RouteParams: RouteParams{"application_id": "appid01", "user_id": "user01"},
			Path:        protocol.Path("/blah"),
			ChannelSize: chanSize,
		},
	))

	a.Nil(e)
}

func (router *router) GetSubscribersForTopic(topicPath string) ([]byte, error) {
	subscribers := make([]RouteParams, 0)
	routes, present := router.routes[protocol.Path(topicPath)]
	if present {
		for index, currRoute := range routes {
			logger.WithFields(log.Fields{
				"index":       index,
				"routeParams": currRoute.RouteParams,
			}).Debug("Added route to slice")
			subscribers = append(subscribers, currRoute.RouteParams)
		}
	}
	return json.Marshal(subscribers)
}

func Test_WebSocket_SubscribeAndUnsubscribe(t *testing.T) {
	_, finish := testutil.NewMockCtrl(t)
	defer finish()

	a := assert.New(t)

	messages := []string{"+ /foo", "+ /bar", "- /foo"}
	wsconn, routerMock, messageStore := createDefaultMocks(messages)

	var wg sync.WaitGroup
	wg.Add(3)
	doneGroup := func(bytes []byte) error {
		wg.Done()
		return nil
	}

	routerMock.EXPECT().Subscribe(routeMatcher{"/foo"}).Return(nil, nil)
	wsconn.EXPECT().
		Send([]byte("#" + protocol.SUCCESS_SUBSCRIBED_TO + " /foo")).
		Do(doneGroup)

	routerMock.EXPECT().Subscribe(routeMatcher{"/bar"}).Return(nil, nil)
	wsconn.EXPECT().
		Send([]byte("#" + protocol.SUCCESS_SUBSCRIBED_TO + " /bar")).
		Do(doneGroup)

	routerMock.EXPECT().Unsubscribe(routeMatcher{"/foo"})
	wsconn.EXPECT().
		Send([]byte("#" + protocol.SUCCESS_CANCELED + " /foo")).
		Do(doneGroup)

	websocket := runNewWebSocket(wsconn, routerMock, messageStore, nil)
	wg.Wait()

	a.Equal(1, len(websocket.receivers))
	a.Equal(protocol.Path("/bar"), websocket.receivers[protocol.Path("/bar")].path)
}

// NewReceiverFromCmd parses the info in the command
func NewReceiverFromCmd(
	applicationId string,
	cmd *protocol.Cmd,
	sendChannel chan []byte,
	router server.Router,
	userId string) (rec *Receiver, err error) {

	messageStore, err := router.MessageStore()
	if err != nil {
		return nil, err
	}

	rec = &Receiver{
		applicationId:       applicationId,
		sendC:               sendChannel,
		router:              router,
		messageStore:        messageStore,
		cancelC:             make(chan bool, 1),
		enableNotifications: true,
		userId:              userId,
	}
	if len(cmd.Arg) == 0 || cmd.Arg[0] != '/' {
		return nil, fmt.Errorf("command requires at least a path argument, but non given")
	}

	args := strings.SplitN(cmd.Arg, " ", 3)
	rec.path = protocol.Path(args[0])

	if len(args) > 1 {
		rec.doFetch = true
		rec.startId, err = strconv.ParseInt(args[1], 10, 64)
		if err != nil {
			return nil, fmt.Errorf("startid has to be empty or int, but was %q: %v", args[1], err)
		}
	}

	rec.doSubscription = true
	if len(args) > 2 {
		rec.doSubscription = false
		rec.maxCount, err = strconv.Atoi(args[2])
		if err != nil {
			return nil, fmt.Errorf("maxCount has to be empty or int, but was %q: %v", args[1], err)
		}
	}

	return rec, nil
}

func TestConnector_StartWithSubscriptions(t *testing.T) {
	_, finish := testutil.NewMockCtrl(t)
	defer finish()

	a := assert.New(t)
	conn, mocks := getTestConnector(t, Config{
		Name:       "test",
		Schema:     "test",
		Prefix:     "/connector/",
		URLPattern: "/{device_token}/{user_id}/{topic:.*}",
	}, false, false)

	entriesC := make(chan [2]string)
	mocks.kvstore.EXPECT().Iterate(gomock.Eq("test"), gomock.Eq("")).Return(entriesC)
	close(entriesC)
	mocks.kvstore.EXPECT().Put(gomock.Any(), gomock.Any(), gomock.Any()).Times(4)

	err := conn.Start()
	a.NoError(err)

	routes := make([]*router.Route, 0, 4)
	mocks.router.EXPECT().Subscribe(gomock.Any()).Do(func(r *router.Route) (*router.Route, error) {
		routes = append(routes, r)
		return r, nil
	}).Times(4)

	// create subscriptions
	createSubscriptions(t, conn, 4)
	time.Sleep(100 * time.Millisecond)

	mocks.sender.EXPECT().Send(gomock.Any()).Return(nil, nil).Times(4)

	// send message in route channel
	for i, r := range routes {
		r.Deliver(&protocol.Message{
			ID:   uint64(i),
			Path: protocol.Path("/topic"),
			Body: []byte("test body"),
		})
	}

	time.Sleep(100 * time.Millisecond)

	err = conn.Stop()
	a.NoError(err)
}

func TestRouter_RoutingWithSubTopics(t *testing.T) {
	ctrl, finish := testutil.NewMockCtrl(t)
	defer finish()
	a := assert.New(t)

	// Given a Router with route
	router, _, _, _ := aStartedRouter()

	msMock := NewMockMessageStore(ctrl)
	router.messageStore = msMock
	// expect a message to `blah` partition first and `blahblub` second
	firstStore := msMock.EXPECT().
		StoreMessage(gomock.Any(), gomock.Any()).
		Do(func(m *protocol.Message, nodeID uint8) (int, error) {
			a.Equal("/blah/blub", string(m.Path))
			return 0, nil
		})

	msMock.EXPECT().
		StoreMessage(gomock.Any(), gomock.Any()).After(firstStore).
		Do(func(m *protocol.Message, nodeID uint8) (int, error) {
			a.Equal("/blahblub", string(m.Path))
			return 0, nil
		})

	r, _ := router.Subscribe(NewRoute(
		RouteConfig{
			RouteParams: RouteParams{"application_id": "appid01", "user_id": "user01"},
			Path:        protocol.Path("/blah"),
			ChannelSize: chanSize,
		},
	))

	// when i send a message to a subroute
	router.HandleMessage(&protocol.Message{Path: "/blah/blub", Body: aTestByteMessage})

	// then I can receive the message
	assertChannelContainsMessage(a, r.MessagesChannel(), aTestByteMessage)

	// but, when i send a message to a resource, which is just a substring
	router.HandleMessage(&protocol.Message{Path: "/blahblub", Body: aTestByteMessage})

	// then the message gets not delivered
	a.Equal(0, len(r.MessagesChannel()))
}

// NewRoute creates a new route pointer
func NewRoute(path, applicationID, userID string, c chan *MessageForRoute) *Route {
	queueSize := 0

	route := &Route{
		messagesC:     c,
		queue:         newQueue(queueSize),
		queueSize:     queueSize,
		timeout:       -1,
		closeC:        make(chan struct{}),
		Path:          protocol.Path(path),
		UserID:        userID,
		ApplicationID: applicationID,

		logger: logger.WithFields(log.Fields{
			"path":          path,
			"applicationID": applicationID,
			"userID":        userID,
		}),
	}
	return route
}

func TestSender_Send(t *testing.T) {
	_, finish := testutil.NewMockCtrl(t)
	defer finish()
	a := assert.New(t)

	// given
	routeParams := make(map[string]string)
	routeParams["device_id"] = "1234"
	routeConfig := router.RouteConfig{
		Path:        protocol.Path("path"),
		RouteParams: routeParams,
	}
	route := router.NewRoute(routeConfig)

	msg := &protocol.Message{
		Body: []byte("{}"),
	}

	mSubscriber := NewMockSubscriber(testutil.MockCtrl)
	mSubscriber.EXPECT().Route().Return(route).AnyTimes()

	mRequest := NewMockRequest(testutil.MockCtrl)
	mRequest.EXPECT().Subscriber().Return(mSubscriber).AnyTimes()
	mRequest.EXPECT().Message().Return(msg).AnyTimes()

	mPusher := NewMockPusher(testutil.MockCtrl)
	mPusher.EXPECT().Push(gomock.Any()).Return(nil, nil)

	// and
	s, err := NewSenderUsingPusher(mPusher, "com.myapp")
	a.NoError(err)

	// when
	rsp, err := s.Send(mRequest)

	// then
	a.NoError(err)
	a.Nil(rsp)
}

func (ws *WebSocket) handleSendCmd(cmd *protocol.Cmd) {
	logger.WithFields(log.Fields{
		"cmd": string(cmd.Bytes()),
	}).Debug("Sending ")

	if len(cmd.Arg) == 0 {
		ws.sendError(protocol.ERROR_BAD_REQUEST, "send command requires a path argument, but none given")
		return
	}

	args := strings.SplitN(cmd.Arg, " ", 2)
	msg := &protocol.Message{
		Path:          protocol.Path(args[0]),
		ApplicationID: ws.applicationID,
		UserID:        ws.userID,
		HeaderJSON:    cmd.HeaderJSON,
		Body:          cmd.Body,
	}

	ws.router.HandleMessage(msg)

	ws.sendOK(protocol.SUCCESS_SEND, "")
}

// ServeHTTP is an http.Handler.
// It is a part of the service.endpoint implementation.
func (api *RestMessageAPI) ServeHTTP(w http.ResponseWriter, r *http.Request) {
	if r.Method == http.MethodHead {
		return
	}

	if r.Method == http.MethodGet {
		log.WithField("url", r.URL.Path).Debug("GET")

		topic, err := api.extractTopic(r.URL.Path, subscribersPrefix)
		if err != nil {
			log.WithError(err).Error("Extracting topic failed")
			if err == errNotFound {
				http.NotFound(w, r)
				return
			}
			http.Error(w, "Server error.", http.StatusInternalServerError)
			return
		}

		resp, err := api.router.GetSubscribersForTopic(topic)
		w.Header().Set("Content-Type", "application/json")

		_, err = w.Write(resp)
		if err != nil {
			log.WithField("error", err.Error()).Error("Writing to byte stream failed")
			http.Error(w, "Server error.", http.StatusInternalServerError)
			return
		}
		return
	}

	if r.Method != http.MethodPost {
		http.Error(w, "Method not allowed", http.StatusMethodNotAllowed)
		return
	}

	body, err := ioutil.ReadAll(r.Body)
	if err != nil {
		http.Error(w, "Can not read body", http.StatusBadRequest)
		return
	}

	topic, err := api.extractTopic(r.URL.Path, "/message")
	if err != nil {
		if err == errNotFound {
			http.NotFound(w, r)
			return
		}
		http.Error(w, "Server error.", http.StatusInternalServerError)
		return
	}

	msg := &protocol.Message{
		Path:          protocol.Path(topic),
		Body:          body,
		UserID:        q(r, "userId"),
		ApplicationID: xid.New().String(),
		HeaderJSON:    headersToJSON(r.Header),
	}

	// add filters
	api.setFilters(r, msg)

	api.router.HandleMessage(msg)
	fmt.Fprintf(w, "OK")
}