status-go/services/shhext/history_test.go

package shhext

import (
	"context"
	"testing"
	"time"

	"github.com/ethereum/go-ethereum/rlp"
	"github.com/status-im/status-go/db"
	"github.com/status-im/status-go/eth-node/types"
	"github.com/status-im/status-go/mailserver"
	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/require"
)

func newTestContext(t *testing.T) Context {
	mdb, err := db.NewMemoryDB()
	require.NoError(t, err)
	return NewContext(context.Background(), time.Now, NewRequestsRegistry(0), db.NewLevelDBStorage(mdb))
}

func createInMemStore(t *testing.T) db.HistoryStore {
	mdb, err := db.NewMemoryDB()
	require.NoError(t, err)
	return db.NewHistoryStore(db.NewLevelDBStorage(mdb))
}

func TestRenewRequest(t *testing.T) {
	req := db.HistoryRequest{}
	duration := time.Hour
	req.AddHistory(db.TopicHistory{Duration: duration})

	firstNow := time.Now()
	RenewRequests([]db.HistoryRequest{req}, firstNow)

	initial := firstNow.Add(-duration).Unix()

	th := req.Histories()[0]
	require.Equal(t, initial, th.Current.Unix())
	require.Equal(t, initial, th.First.Unix())
	require.Equal(t, firstNow.Unix(), th.End.Unix())

	secondNow := time.Now()
	RenewRequests([]db.HistoryRequest{req}, secondNow)

	require.Equal(t, initial, th.Current.Unix())
	require.Equal(t, initial, th.First.Unix())
	require.Equal(t, secondNow.Unix(), th.End.Unix())
}

func TestCreateTopicOptionsFromRequest(t *testing.T) {
	req := db.HistoryRequest{}
	topic := types.TopicType{1}
	now := time.Now()
	req.AddHistory(db.TopicHistory{Topic: topic, Current: now, End: now})
	options := CreateTopicOptionsFromRequest(req)
	require.Len(t, options, len(req.Histories()),
		"length must be equal to the number of topic histories attached to request")
	require.Equal(t, topic, options[0].Topic)
	require.Equal(t, uint64(now.Add(-WhisperTimeAllowance).Unix()), options[0].Range.Start,
		"start of the range must be adjusted by the whisper time allowance")
	require.Equal(t, uint64(now.Unix()), options[0].Range.End)
}

func TestTopicOptionsToBloom(t *testing.T) {
	options := TopicOptions{
		{Topic: types.TopicType{1}, Range: Range{Start: 1, End: 10}},
		{Topic: types.TopicType{2}, Range: Range{Start: 3, End: 12}},
	}
	bloom := options.ToBloomFilterOption()
	require.Equal(t, uint64(3), bloom.Range.Start, "Start must be the latest Start across all options")
	require.Equal(t, uint64(12), bloom.Range.End, "End must be the latest End across all options")
	require.Equal(t, topicsToBloom(options[0].Topic, options[1].Topic), bloom.Filter)
}

func TestBloomFilterToMessageRequestPayload(t *testing.T) {
	var (
		start   uint32 = 10
		end     uint32 = 20
		filter         = []byte{1, 1, 1, 1}
		message        = mailserver.MessagesRequestPayload{
			Lower: start,
			Upper: end,
			Bloom: filter,
			Batch: true,
			Limit: 10000,
		}
		bloomOption = BloomFilterOption{
			Filter: filter,
			Range: Range{
				Start: uint64(start),
				End:   uint64(end),
			},
		}
	)
	expected, err := rlp.EncodeToBytes(message)
	require.NoError(t, err)
	payload, err := bloomOption.ToMessagesRequestPayload()
	require.NoError(t, err)
	require.Equal(t, expected, payload)
}

func TestCreateRequestsEmptyState(t *testing.T) {
	ctx := newTestContext(t)
	reactor := NewHistoryUpdateReactor()
	requests, err := reactor.CreateRequests(ctx, []TopicRequest{
		{Topic: types.TopicType{1}, Duration: time.Hour},
		{Topic: types.TopicType{2}, Duration: time.Hour},
		{Topic: types.TopicType{3}, Duration: 10 * time.Hour},
	})
	require.NoError(t, err)
	require.Len(t, requests, 2)
	var (
		oneTopic, twoTopic db.HistoryRequest
	)
	if len(requests[0].Histories()) == 1 {
		oneTopic, twoTopic = requests[0], requests[1]
	} else {
		oneTopic, twoTopic = requests[1], requests[0]
	}
	require.Len(t, oneTopic.Histories(), 1)
	require.Len(t, twoTopic.Histories(), 2)

}

func TestCreateRequestsWithExistingRequest(t *testing.T) {
	ctx := newTestContext(t)
	store := ctx.HistoryStore()
	req := store.NewRequest()
	req.ID = types.Hash{1}
	th := store.NewHistory(types.TopicType{1}, time.Hour)
	req.AddHistory(th)
	require.NoError(t, req.Save())
	reactor := NewHistoryUpdateReactor()
	requests, err := reactor.CreateRequests(ctx, []TopicRequest{
		{Topic: types.TopicType{1}, Duration: time.Hour},
		{Topic: types.TopicType{2}, Duration: time.Hour},
		{Topic: types.TopicType{3}, Duration: time.Hour},
	})
	require.NoError(t, err)
	require.Len(t, requests, 2)

	var (
		oneTopic, twoTopic db.HistoryRequest
	)
	if len(requests[0].Histories()) == 1 {
		oneTopic, twoTopic = requests[0], requests[1]
	} else {
		oneTopic, twoTopic = requests[1], requests[0]
	}
	assert.Len(t, oneTopic.Histories(), 1)
	assert.Len(t, twoTopic.Histories(), 2)
}

func TestCreateMultiRequestsWithSameTopic(t *testing.T) {
	ctx := newTestContext(t)
	store := ctx.HistoryStore()
	reactor := NewHistoryUpdateReactor()
	topic := types.TopicType{1}
	requests, err := reactor.CreateRequests(ctx, []TopicRequest{
		{Topic: topic, Duration: time.Hour},
	})
	require.NoError(t, err)
	require.Len(t, requests, 1)
	requests[0].ID = types.Hash{1}
	require.NoError(t, requests[0].Save())

	// duration changed. request wasn't finished
	requests, err = reactor.CreateRequests(ctx, []TopicRequest{
		{Topic: topic, Duration: 10 * time.Hour},
	})
	require.NoError(t, err)
	require.Len(t, requests, 2)
	longest := 0
	for i := range requests {
		r := &requests[i]
		r.ID = types.Hash{byte(i)}
		require.NoError(t, r.Save())
		require.Len(t, r.Histories(), 1)
		if r.Histories()[0].Duration == 10*time.Hour {
			longest = i
		}
	}
	require.Equal(t, requests[longest].Histories()[0].End, requests[longest^1].Histories()[0].First)

	for _, r := range requests {
		require.NoError(t, reactor.UpdateFinishedRequest(ctx, r.ID))
	}
	requests, err = reactor.CreateRequests(ctx, []TopicRequest{
		{Topic: topic, Duration: 10 * time.Hour},
	})
	require.NoError(t, err)
	require.Len(t, requests, 1)

	topics, err := store.GetHistoriesByTopic(topic)
	require.NoError(t, err)
	require.Len(t, topics, 1)
	require.Equal(t, 10*time.Hour, topics[0].Duration)
}

func TestRequestFinishedUpdate(t *testing.T) {
	ctx := newTestContext(t)
	store := ctx.HistoryStore()
	req := store.NewRequest()
	req.ID = types.Hash{1}
	now := ctx.Time()
	thOne := store.NewHistory(types.TopicType{1}, time.Hour)
	thOne.End = now
	thTwo := store.NewHistory(types.TopicType{2}, time.Hour)
	thTwo.End = now
	req.AddHistory(thOne)
	req.AddHistory(thTwo)
	require.NoError(t, req.Save())

	reactor := NewHistoryUpdateReactor()
	require.NoError(t, reactor.UpdateTopicHistory(ctx, thOne.Topic, now.Add(-time.Minute)))
	require.NoError(t, reactor.UpdateFinishedRequest(ctx, req.ID))
	_, err := store.GetRequest(req.ID)
	require.EqualError(t, err, "leveldb: not found")

	require.NoError(t, thOne.Load())
	require.NoError(t, thTwo.Load())
	require.Equal(t, now.Unix(), thOne.Current.Unix())
	require.Equal(t, now.Unix(), thTwo.Current.Unix())
}

func TestTopicHistoryUpdate(t *testing.T) {
	ctx := newTestContext(t)
	store := ctx.HistoryStore()
	reqID := types.Hash{1}
	request := store.NewRequest()
	request.ID = reqID
	now := time.Now()
	require.NoError(t, request.Save())
	th := store.NewHistory(types.TopicType{1}, time.Hour)
	th.RequestID = request.ID
	th.End = now
	require.NoError(t, th.Save())
	reactor := NewHistoryUpdateReactor()
	timestamp := now.Add(-time.Minute)

	require.NoError(t, reactor.UpdateTopicHistory(ctx, th.Topic, timestamp))
	require.NoError(t, th.Load())
	require.Equal(t, timestamp.Unix(), th.Current.Unix())

	require.NoError(t, reactor.UpdateTopicHistory(ctx, th.Topic, now))
	require.NoError(t, th.Load())
	require.Equal(t, timestamp.Unix(), th.Current.Unix())
}

func TestGroupHistoriesByRequestTimestamp(t *testing.T) {
	requests := GroupHistoriesByRequestTimespan(createInMemStore(t), []db.TopicHistory{
		{Topic: types.TopicType{1}, Duration: time.Hour},
		{Topic: types.TopicType{2}, Duration: time.Hour},
		{Topic: types.TopicType{3}, Duration: 2 * time.Hour},
		{Topic: types.TopicType{4}, Duration: 2 * time.Hour},
		{Topic: types.TopicType{5}, Duration: 3 * time.Hour},
		{Topic: types.TopicType{6}, Duration: 3 * time.Hour},
	})
	require.Len(t, requests, 3)
	for _, req := range requests {
		require.Len(t, req.Histories(), 2)
	}
}

// initial creation of the history index. no other histories in store
func TestAdjustHistoryWithNoOtherHistories(t *testing.T) {
	store := createInMemStore(t)
	th := store.NewHistory(types.TopicType{1}, time.Hour)
	adjusted, err := adjustRequestedHistories(store, []db.TopicHistory{th})
	require.NoError(t, err)
	require.Len(t, adjusted, 1)
	require.Equal(t, th.Topic, adjusted[0].Topic)
}

// Duration for the history index with same topic was gradually incresed:
// {Duration: 1h} {Duration: 2h} {Duration: 3h}
// But actual request wasn't sent
// So when we receive {Duration: 4h} we can merge all of them into single index
// that covers all of them e.g. {Duration: 4h}
func TestAdjustHistoryWithExistingLowerRanges(t *testing.T) {
	store := createInMemStore(t)
	topic := types.TopicType{1}
	histories := make([]db.TopicHistory, 3)
	i := 0
	for i = range histories {
		histories[i] = store.NewHistory(topic, time.Duration(i+1)*time.Hour)
		require.NoError(t, histories[i].Save())
	}
	i++
	th := store.NewHistory(topic, time.Duration(i+1)*time.Hour)
	adjusted, err := adjustRequestedHistories(store, []db.TopicHistory{th})
	require.NoError(t, err)
	require.Len(t, adjusted, 1)
	require.Equal(t, th.Duration, adjusted[0].Duration)

	all, err := store.GetHistoriesByTopic(topic)
	require.NoError(t, err)
	require.Len(t, all, 1)
	require.Equal(t, th.Duration, all[0].Duration)
}

// Precondition is based on the previous test. We have same information in the database
// but now every history index request was successfully completed. And End timstamp is set to the First of the next index.
// So, we have:
// {First: now-1h, End: now} {First: now-2h, End: now-1h} {First: now-3h: End: now-2h}
// When we want to create new request with {Duration: 4h}
// We see that there is no reason to keep all indexes and we can squash them.
func TestAdjustHistoriesWithExistingCoveredLowerRanges(t *testing.T) {
	store := createInMemStore(t)
	topic := types.TopicType{1}
	histories := make([]db.TopicHistory, 3)
	i := 0
	now := time.Now()
	for i = range histories {
		duration := time.Duration(i+1) * time.Hour
		prevduration := time.Duration(i) * time.Hour
		histories[i] = store.NewHistory(topic, duration)
		histories[i].First = now.Add(-duration)
		histories[i].Current = now.Add(-prevduration)
		require.NoError(t, histories[i].Save())
	}
	i++
	th := store.NewHistory(topic, time.Duration(i+1)*time.Hour)
	th.Current = now.Add(-time.Duration(i) * time.Hour)
	adjusted, err := adjustRequestedHistories(store, []db.TopicHistory{th})
	require.NoError(t, err)
	require.Len(t, adjusted, 1)
	require.Equal(t, th.Duration, adjusted[0].Duration)
}

func TestAdjustHistoryReplaceTopicWithHigherDuration(t *testing.T) {
	store := createInMemStore(t)
	topic := types.TopicType{1}
	hour := store.NewHistory(topic, time.Hour)
	require.NoError(t, hour.Save())
	minute := store.NewHistory(topic, time.Minute)
	adjusted, err := adjustRequestedHistories(store, []db.TopicHistory{minute})
	require.NoError(t, err)
	require.Len(t, adjusted, 1)
	require.Equal(t, hour.Duration, adjusted[0].Duration)
}

// if client requested lower duration than the one we have in the index already it will
// it will be discarded and we will use existing index
func TestAdjustHistoryRemoveTopicIfPendingWithHigherDuration(t *testing.T) {
	store := createInMemStore(t)
	topic := types.TopicType{1}
	hour := store.NewHistory(topic, time.Hour)
	hour.RequestID = types.Hash{1}
	require.NoError(t, hour.Save())
	minute := store.NewHistory(topic, time.Minute)
	adjusted, err := adjustRequestedHistories(store, []db.TopicHistory{minute})
	require.NoError(t, err)
	require.Len(t, adjusted, 0)
}