test(store): make queries (#752)

* test(store): make queries

* test: most recent timestamp and count of msg

* nit: add comment for pageSize+1

waku/persistence/store.go

@@ -410,6 +410,10 @@ func (d *DBStore) prepareQuerySQL(query *pb.HistoryQuery) (string, []interface{}
  paramCnt++
 
  sqlQuery += fmt.Sprintf("LIMIT $%d", paramCnt)
+ // Always search for _max page size_ + 1. If the extra row does not exist, do not return pagination info.
+ pageSize := query.PagingInfo.PageSize + 1
+ parameters = append(parameters, pageSize)
+
  sqlQuery = fmt.Sprintf(sqlQuery, conditionStr, orderDirection, orderDirection, orderDirection, orderDirection)
  d.log.Info(fmt.Sprintf("sqlQuery: %s", sqlQuery))
 
@@ -434,10 +438,7 @@ func (d *DBStore) Query(query *pb.HistoryQuery) (*pb.Index, []StoredMessage, err
  return nil, nil, err
  }
  defer stmt.Close()
- pageSize := query.PagingInfo.PageSize + 1
-
- parameters = append(parameters, pageSize)
-
+ //
  measurementStart := time.Now()
  rows, err := stmt.Query(parameters...)
  if err != nil {
@@ -458,6 +459,7 @@ func (d *DBStore) Query(query *pb.HistoryQuery) (*pb.Index, []StoredMessage, err
 
  var cursor *pb.Index
  if len(result) != 0 {
+ // since there are more rows than pagingInfo.PageSize, we need to return a cursor, for pagination
  if len(result) > int(query.PagingInfo.PageSize) {
  result = result[0:query.PagingInfo.PageSize]
  lastMsgIdx := len(result) - 1

waku/persistence/store_test.go

@@ -3,6 +3,7 @@ package persistence
 import (
  "context"
  "database/sql"
+ "fmt"
  "testing"
  "time"
 
@@ -14,6 +15,7 @@ import (
  "github.com/waku-org/go-waku/waku/persistence/migrate"
  sqlitemigrations "github.com/waku-org/go-waku/waku/persistence/sqlite/migrations"
  "github.com/waku-org/go-waku/waku/v2/protocol"
+ "github.com/waku-org/go-waku/waku/v2/protocol/store/pb"
  "github.com/waku-org/go-waku/waku/v2/timesource"
  "github.com/waku-org/go-waku/waku/v2/utils"
  "go.uber.org/zap"
@@ -95,4 +97,92 @@ func TestStoreRetention(t *testing.T) {
  require.Equal(t, "test5", dbResults[0].Message.ContentTopic)
  require.Equal(t, "test6", dbResults[1].Message.ContentTopic)
  require.Equal(t, "test7", dbResults[2].Message.ContentTopic)
+ // checking the number of all the message in the db
+ msgCount, err := store.Count()
+ require.NoError(t, err)
+ require.Equal(t, msgCount, 3)
+}
+
+func TestQuery(t *testing.T) {
+ db := NewMock()
+ store, err := NewDBStore(prometheus.DefaultRegisterer, utils.Logger(), WithDB(db), WithMigrations(Migrate), WithRetentionPolicy(5, 20*time.Second))
+ require.NoError(t, err)
+
+ insertTime := time.Now()
+ //////////////////////////////////
+ _ = store.Put(protocol.NewEnvelope(tests.CreateWakuMessage("test1", insertTime.Add(-40*time.Second).UnixNano()), insertTime.Add(-10*time.Second).UnixNano(), "test"))
+ _ = store.Put(protocol.NewEnvelope(tests.CreateWakuMessage("test2", insertTime.Add(-30*time.Second).UnixNano()), insertTime.Add(-10*time.Second).UnixNano(), "test"))
+ _ = store.Put(protocol.NewEnvelope(tests.CreateWakuMessage("test3", insertTime.Add(-20*time.Second).UnixNano()), insertTime.Add(-10*time.Second).UnixNano(), "test"))
+ _ = store.Put(protocol.NewEnvelope(tests.CreateWakuMessage("test3", insertTime.Add(-20*time.Second).UnixNano()), insertTime.Add(-10*time.Second).UnixNano(), "test2"))
+ _ = store.Put(protocol.NewEnvelope(tests.CreateWakuMessage("test4", insertTime.Add(-10*time.Second).UnixNano()), insertTime.Add(-10*time.Second).UnixNano(), "test"))
+
+ // Range [startTime-endTime]
+ // Check: matching ContentTopics and pubsubTopic, and ts of msg in range
+ // this filters test1,test2 contentTopics. test3 matches list of contentTopic but its not within the time range
+ cursor, msgs, err := store.Query(&pb.HistoryQuery{
+ PubsubTopic: "test",
+ ContentFilters: []*pb.ContentFilter{
+ {ContentTopic: "test1"},
+ {ContentTopic: "test2"},
+ {ContentTopic: "test3"},
+ },
+ PagingInfo: &pb.PagingInfo{PageSize: 10},
+ StartTime: insertTime.Add(-41 * time.Second).UnixNano(),
+ EndTime: insertTime.Add(-21 * time.Second).UnixNano(),
+ })
+ require.NoError(t, err)
+ require.Len(t, msgs, 2)
+
+ _ = store.Put(protocol.NewEnvelope(tests.CreateWakuMessage("test5", insertTime.UnixNano()), insertTime.Add(-10*time.Second).UnixNano(), "test"))
+
+ // Range [cursor-endTime]
+ // Check: matching ContentTopic,pubsubTopic, pageSize
+ // cursor has last message id of test2 which is now-30second
+ // endTime is now+1sec
+ // matched messages are both test4,test5, but the len of returned result depends on pageSize
+ var cursor2 *pb.Index
+ for _, pageSize := range []int{1, 2} {
+ cursorLocal, msgs, err := store.Query(&pb.HistoryQuery{
+ PubsubTopic: "test",
+ ContentFilters: []*pb.ContentFilter{
+ {ContentTopic: "test4"},
+ {ContentTopic: "test5"},
+ },
+ PagingInfo: &pb.PagingInfo{Cursor: cursor, PageSize: uint64(pageSize), Direction: pb.PagingInfo_FORWARD},
+ EndTime: insertTime.Add(1 * time.Second).UnixNano(),
+ })
+ require.NoError(t, err)
+ require.Len(t, msgs, pageSize) // due to pageSize
+ require.Equal(t, msgs[0].Message.ContentTopic, "test4")
+ if len(msgs) > 1 {
+ require.Equal(t, msgs[1].Message.ContentTopic, "test5")
+ }
+ cursor2 = cursorLocal
+ }
+
+ // range [startTime-cursor(test5_ContentTopic_Msg)],
+ // Check: backend range with cursor excludes test1 ContentTopic, matching ContentTopic
+ // check backward pagination
+ _, msgs, err = store.Query(&pb.HistoryQuery{
+ PubsubTopic: "test",
+ ContentFilters: []*pb.ContentFilter{
+ {ContentTopic: "test1"}, // contentTopic test1 is out of the range
+ {ContentTopic: "test2"},
+ {ContentTopic: "test3"},
+ {ContentTopic: "test4"},
+ },
+ PagingInfo: &pb.PagingInfo{Cursor: cursor2, PageSize: 4, Direction: pb.PagingInfo_BACKWARD},
+ StartTime: insertTime.Add(-39 * time.Second).UnixNano(),
+ })
+ require.NoError(t, err)
+ require.Len(t, msgs, 3) // due to pageSize
+ // Check:this also makes returned messages are sorted ascending
+ for ind, msg := range msgs {
+ require.Equal(t, msg.Message.ContentTopic, fmt.Sprintf("test%d", ind+2)) // test2,test3,test4
+ }
+
+ // checking most recent timestamp in db
+ timestamp, err := store.MostRecentTimestamp()
+ require.NoError(t, err)
+ require.Equal(t, timestamp, insertTime.UnixNano())
 }