Include V2 query for Mongo messages

app/common.go

@@ -2,6 +2,7 @@ package app
 
 import (
 	"context"
+	"encoding/json"
 	"time"
 
 	"github.com/topfreegames/mqtt-history/cassandra"
@@ -69,3 +70,61 @@ func SelectFromCollection(ctx context.Context, topic string, from int64, limit i
 	}
 	return searchResults
 }
+
+// SelectFromCollectionV2 expects the message to be stored in mongo with a specific structure,
+// the main difference being that the payload field is now referred to as "original_payload" and
+// is a JSON object, not a string, and also the timestamp is int64 seconds since Unix epoch, not an ISODate on Mongo
+func SelectFromCollectionV2(ctx context.Context, topic string, from int64, limit int64, collection string) []*models.Message {
+	type message struct {
+		Timestamp       int64  `bson:"timestamp"`
+		OriginalPayload bson.M `bson:"original_payload"`
+	}
+
+	searchResults := make([]message, 0)
+
+	callback := func(coll *mongo.Collection) error {
+		query := bson.M{
+			"topic": topic,
+			"timestamp": bson.M{
+				"$lte": from, // less than or equal
+			},
+		}
+
+		sort := bson.D{
+			{"topic", 1},
+			{"timestamp", -1},
+		}
+
+		opts := options.Find()
+		opts.SetSort(sort)
+		opts.SetLimit(limit)
+
+		cursor, err := coll.Find(ctx, query, opts)
+		if err != nil {
+			return err
+		}
+
+		return cursor.All(ctx, &searchResults)
+	}
+
+	err := mongoclient.GetCollection(collection, callback)
+	if err != nil {
+		return []*models.Message{}
+	}
+
+	messages := make([]*models.Message, 0)
+	for _, result := range searchResults {
+		payload := result.OriginalPayload
+		bytes, _ := json.Marshal(payload)
+
+		finalStr := string(bytes)
+		message := &models.Message{
+			Timestamp: time.Unix(result.Timestamp, 0),
+			Payload:   finalStr,
+			Topic:     topic,
+		}
+		messages = append(messages, message)
+	}
+
+	return messages
+}