Adding /histories route

app/app.go

@@ -115,6 +115,7 @@ func (app *App) configureApplication() {
 	a.Get("/healthcheck", HealthCheckHandler(app))
 	a.Get("/historysince/*", HistorySinceHandler(app))
 	a.Get("/history/*", HistoryHandler(app))
+  a.Get("/histories/*", HistoriesHandler(app))
 	a.Get("/:other", NotFoundHandler(app))
 
 	app.RedisClient = redisclient.GetRedisClient(viper.GetString("redis.host"), viper.GetInt("redis.port"), viper.GetString("redis.password"))

app/histories.go

@@ -0,0 +1,81 @@
+package app
+
+import (
+	"fmt"
+	"net/http"
+	"reflect"
+	"strconv"
+	"strings"
+
+	"github.com/labstack/echo"
+	"github.com/topfreegames/mqtt-history/es"
+	"github.com/topfreegames/mqtt-history/logger"
+	"gopkg.in/olivere/elastic.v3"
+)
+
+// HistoriesHandler is the handler responsible for sending multiples rooms history to the player
+func HistoriesHandler(app *App) func(c echo.Context) error {
+	return func(c echo.Context) error {
+		esclient := es.GetESClient()
+		c.Set("route", "Histories")
+		topicPrefix := c.ParamValues()[0]
+		authorizedTopics := []string{}
+		userID := c.QueryParam("userid")
+		topicsSuffix := strings.Split(c.QueryParam("topics"), ",")
+		topics := make([]string, len(topicsSuffix))
+		from, err := strconv.Atoi(c.QueryParam("from"))
+		limit, err := strconv.Atoi(c.QueryParam("limit"))
+		for i, topicSuffix := range topicsSuffix {
+			topics[i] = topicPrefix + topicSuffix
+		}
+		if limit == 0 {
+			limit = 10
+		}
+
+		logger.Logger.Debugf("user %s is asking for histories for topicPrefix %s with args topics=%s from=%d and limit=%d", userID, topicPrefix, topics, from, limit)
+		rc := app.RedisClient.Pool.Get()
+		defer rc.Close()
+		rc.Send("MULTI")
+		rc.Send("GET", userID)
+		for _, topic := range topics {
+			rc.Send("GET", fmt.Sprintf("%s-%s", userID, topic))
+		}
+		r, err := rc.Do("EXEC")
+		if err != nil {
+			return err
+		}
+		redisResults := (r.([]interface{}))
+		for i, redisResp := range redisResults[1:] {
+			if redisResp != nil {
+				authorizedTopics = append(authorizedTopics, topics[i])
+			}
+		}
+
+		if redisResults[0] != nil && len(authorizedTopics) > 0 {
+			boolQuery := elastic.NewBoolQuery()
+			topicBoolQuery := elastic.NewBoolQuery()
+			topicBoolQuery.Should(elastic.NewTermsQuery("topic", authorizedTopics))
+			boolQuery.Must(topicBoolQuery)
+
+			var searchResults *elastic.SearchResult
+			err = WithSegment("elasticsearch", c, func() error {
+				searchResults, err = esclient.Search().Index("chat").Query(boolQuery).
+					Sort("timestamp", false).From(from).Size(limit).Do()
+				return err
+			})
+
+			if err != nil {
+				return err
+			}
+			messages := []Message{}
+			var ttyp Message
+			for _, item := range searchResults.Each(reflect.TypeOf(ttyp)) {
+				if t, ok := item.(Message); ok {
+					messages = append(messages, t)
+				}
+			}
+			return c.JSON(http.StatusOK, messages)
+		}
+		return c.String(echo.ErrUnauthorized.Code, echo.ErrUnauthorized.Message)
+	}
+}

app/histories_test.go

@@ -0,0 +1,169 @@
+// mqtt-history
+// https://github.com/topfreegames/mqtt-history
+// Licensed under the MIT license:
+// http://www.opensource.org/licenses/mit-license
+// Copyright © 2016 Top Free Games <backend@tfgco.com>
+
+package app_test
+
+import (
+	"encoding/json"
+	"fmt"
+	"net/http"
+	"strings"
+	"testing"
+	"time"
+
+	. "github.com/franela/goblin"
+	. "github.com/onsi/gomega"
+	"github.com/satori/go.uuid"
+	. "github.com/topfreegames/mqtt-history/app"
+	"github.com/topfreegames/mqtt-history/es"
+	"github.com/topfreegames/mqtt-history/redisclient"
+	. "github.com/topfreegames/mqtt-history/testing"
+)
+
+func TestHistoriesHandler(t *testing.T) {
+	g := Goblin(t)
+
+	// special hook for gomega
+	RegisterFailHandler(func(m string, _ ...int) { g.Fail(m) })
+
+	g.Describe("Histories", func() {
+		esclient := es.GetESClient()
+
+		g.BeforeEach(func() {
+			refreshIndex()
+		})
+
+		g.Describe("Histories Handler", func() {
+			g.It("It should return 401 if the user is not authorized into the topics", func() {
+				a := GetDefaultTestApp()
+				testId := strings.Replace(uuid.NewV4().String(), "-", "", -1)
+				path := fmt.Sprintf("/history/chat/test_?userid=test:test&topics=%s", testId)
+				status, _ := Get(a, path, t)
+				g.Assert(status).Equal(http.StatusUnauthorized)
+			})
+
+			g.It("It should return 200 if the user is authorized into the topics", func() {
+				a := GetDefaultTestApp()
+				testId := strings.Replace(uuid.NewV4().String(), "-", "", -1)
+				testId2 := strings.Replace(uuid.NewV4().String(), "-", "", -1)
+				topic := fmt.Sprintf("chat/test_%s", testId)
+				topic2 := fmt.Sprintf("chat/test_%s", testId2)
+				authStr := fmt.Sprintf("test:test-%s", topic)
+				authStr2 := fmt.Sprintf("test:test-%s", topic2)
+				rc := redisclient.GetRedisClient("localhost", 4444, "")
+				_, err := rc.Pool.Get().Do("set", "test:test", "lalala")
+				_, err = rc.Pool.Get().Do("set", authStr, 2)
+				_, err = rc.Pool.Get().Do("set", authStr2, 2)
+				Expect(err).To(BeNil())
+
+				testMessage := Message{
+					Timestamp: time.Now().AddDate(0, 0, -1),
+					Payload:   "{\"test1\":\"test2\"}",
+					Topic:     topic,
+				}
+
+				testMessage2 := Message{
+					Timestamp: time.Now(),
+					Payload:   "{\"test3\":\"test4\"}",
+					Topic:     topic2,
+				}
+
+				_, err = esclient.Index().Index("chat").Type("message").BodyJson(testMessage).Do()
+				Expect(err).To(BeNil())
+
+				_, err = esclient.Index().Index("chat").Type("message").BodyJson(testMessage2).Do()
+				Expect(err).To(BeNil())
+
+				refreshIndex()
+				path := fmt.Sprintf("/histories/chat/test_?userid=test:test&topics=%s,%s", testId, testId2)
+				status, body := Get(a, path, t)
+				g.Assert(status).Equal(http.StatusOK)
+
+				var messages []Message
+				err = json.Unmarshal([]byte(body), &messages)
+				Expect(err).To(BeNil())
+				g.Assert(messages[0].Payload).Equal("{\"test3\":\"test4\"}")
+				g.Assert(messages[1].Payload).Equal("{\"test1\":\"test2\"}")
+			})
+
+			g.It("It should return 200 if the user is authorized into at least one topic", func() {
+				a := GetDefaultTestApp()
+				testId := strings.Replace(uuid.NewV4().String(), "-", "", -1)
+				testId2 := strings.Replace(uuid.NewV4().String(), "-", "", -1)
+				topic := fmt.Sprintf("chat/test_%s", testId)
+				topic2 := fmt.Sprintf("chat/test_%s", testId2)
+				authStr := fmt.Sprintf("test:test-%s", topic)
+				rc := redisclient.GetRedisClient("localhost", 4444, "")
+				_, err := rc.Pool.Get().Do("set", "test:test", "lalala")
+				_, err = rc.Pool.Get().Do("set", authStr, 2)
+				Expect(err).To(BeNil())
+
+				testMessage := Message{
+					Timestamp: time.Now().AddDate(0, 0, -1),
+					Payload:   "{\"test1\":\"test2\"}",
+					Topic:     topic,
+				}
+
+				testMessage2 := Message{
+					Timestamp: time.Now(),
+					Payload:   "{\"test3\":\"test4\"}",
+					Topic:     topic2,
+				}
+
+				_, err = esclient.Index().Index("chat").Type("message").BodyJson(testMessage).Do()
+				Expect(err).To(BeNil())
+
+				_, err = esclient.Index().Index("chat").Type("message").BodyJson(testMessage2).Do()
+				Expect(err).To(BeNil())
+
+				refreshIndex()
+				path := fmt.Sprintf("/histories/chat/test_?userid=test:test&topics=%s,%s", testId, testId2)
+				status, body := Get(a, path, t)
+				g.Assert(status).Equal(http.StatusOK)
+
+				var messages []Message
+				err = json.Unmarshal([]byte(body), &messages)
+				Expect(err).To(BeNil())
+				g.Assert(messages[0].Payload).Equal("{\"test1\":\"test2\"}")
+				g.Assert(len(messages)).Equal(1)
+			})
+
+			g.It("It should return 401 if the user is not authorized in any topic", func() {
+				a := GetDefaultTestApp()
+				testId := strings.Replace(uuid.NewV4().String(), "-", "", -1)
+				testId2 := strings.Replace(uuid.NewV4().String(), "-", "", -1)
+				topic := fmt.Sprintf("chat/test_%s", testId)
+				topic2 := fmt.Sprintf("chat/test_%s", testId2)
+				rc := redisclient.GetRedisClient("localhost", 4444, "")
+				_, err := rc.Pool.Get().Do("set", "test:test", "lalala")
+				Expect(err).To(BeNil())
+
+				testMessage := Message{
+					Timestamp: time.Now().AddDate(0, 0, -1),
+					Payload:   "{\"test1\":\"test2\"}",
+					Topic:     topic,
+				}
+
+				testMessage2 := Message{
+					Timestamp: time.Now(),
+					Payload:   "{\"test3\":\"test4\"}",
+					Topic:     topic2,
+				}
+
+				_, err = esclient.Index().Index("chat").Type("message").BodyJson(testMessage).Do()
+				Expect(err).To(BeNil())
+
+				_, err = esclient.Index().Index("chat").Type("message").BodyJson(testMessage2).Do()
+				Expect(err).To(BeNil())
+
+				refreshIndex()
+				path := fmt.Sprintf("/histories/chat/test_?userid=test:test&topics=%s,%s", testId, testId2)
+				status, _ := Get(a, path, t)
+				g.Assert(status).Equal(http.StatusUnauthorized)
+			})
+		})
+	})
+}