status-go/protocol/messenger_mailserver.go

package protocol

import (
	"context"
	"fmt"
	"math"
	"sort"
	"sync"
	"time"

	"github.com/pkg/errors"
	"go.uber.org/zap"

	"github.com/status-im/status-go/connection"
	"github.com/status-im/status-go/eth-node/crypto"
	"github.com/status-im/status-go/eth-node/types"
	"github.com/status-im/status-go/protocol/common"
	"github.com/status-im/status-go/protocol/protobuf"
	"github.com/status-im/status-go/protocol/transport"
	"github.com/status-im/status-go/services/mailservers"
)

const (
	initialStoreNodeRequestPageSize = 4
	defaultStoreNodeRequestPageSize = 50

	// tolerance is how many seconds of potentially out-of-order messages we want to fetch
	tolerance uint32 = 60

	mailserverRequestTimeout         = 30 * time.Second
	mailserverMaxTries          uint = 2
	mailserverMaxFailedRequests uint = 2

	oneDayDuration   = 24 * time.Hour
	oneMonthDuration = 31 * oneDayDuration
)

// maxTopicsPerRequest sets the batch size to limit the number of topics per store query
var maxTopicsPerRequest int = 10

var ErrNoFiltersForChat = errors.New("no filter registered for given chat")

func (m *Messenger) shouldSync() (bool, error) {
	// TODO (pablo) support community store node as well
	if m.mailserverCycle.activeMailserver == nil || !m.Online() {
		return false, nil
	}

	useMailserver, err := m.settings.CanUseMailservers()
	if err != nil {
		m.logger.Error("failed to get use mailservers", zap.Error(err))
		return false, err
	}

	return useMailserver, nil
}

func (m *Messenger) scheduleSyncChat(chat *Chat) (bool, error) {
	shouldSync, err := m.shouldSync()
	if err != nil {
		m.logger.Error("failed to get should sync", zap.Error(err))
		return false, err
	}

	if !shouldSync {
		return false, nil
	}

	go func() {
		ms := m.getActiveMailserver(chat.CommunityID)
		_, err = m.performMailserverRequest(ms, func(mailServer mailservers.Mailserver) (*MessengerResponse, error) {
			response, err := m.syncChatWithFilters(mailServer, chat.ID)

			if err != nil {
				m.logger.Error("failed to sync chat", zap.Error(err))
				return nil, err
			}

			if m.config.messengerSignalsHandler != nil {
				m.config.messengerSignalsHandler.MessengerResponse(response)
			}
			return response, nil
		})
		if err != nil {
			m.logger.Error("failed to perform mailserver request", zap.Error(err))
		}
	}()
	return true, nil
}

func (m *Messenger) connectToNewMailserverAndWait() error {
	// Handle pinned mailservers
	m.logger.Info("disconnecting mailserver")
	pinnedMailserver, err := m.getPinnedMailserver()
	if err != nil {
		m.logger.Error("could not obtain the pinned mailserver", zap.Error(err))
		return err
	}
	// If pinned mailserver is not nil, no need to disconnect and wait for it to be available
	if pinnedMailserver == nil {
		m.disconnectActiveMailserver()
	}

	return m.findNewMailserver()
}

func (m *Messenger) performMailserverRequest(ms *mailservers.Mailserver, fn func(mailServer mailservers.Mailserver) (*MessengerResponse, error)) (*MessengerResponse, error) {
	if ms == nil {
		return nil, errors.New("mailserver not available")
	}

	m.mailserverCycle.RLock()
	defer m.mailserverCycle.RUnlock()
	var tries uint = 0
	for tries < mailserverMaxTries {
		if !m.communityStorenodes.IsCommunityStoreNode(ms.ID) && !m.isMailserverAvailable(ms.ID) {
			return nil, errors.New("mailserver not available")
		}
		m.logger.Info("trying performing mailserver requests", zap.Uint("try", tries), zap.String("mailserverID", ms.ID))

		// Peform request
		response, err := fn(*ms) // pass by value because we don't want the fn to modify the mailserver
		if err == nil {
			// Reset failed requests
			m.logger.Debug("mailserver request performed successfully",
				zap.String("mailserverID", ms.ID))
			ms.FailedRequests = 0
			return response, nil
		}

		m.logger.Error("failed to perform mailserver request",
			zap.String("mailserverID", ms.ID),
			zap.Uint("tries", tries),
			zap.Error(err),
		)

		tries++
		// Increment failed requests
		ms.FailedRequests++

		// Change mailserver
		if ms.FailedRequests >= mailserverMaxFailedRequests {
			return nil, errors.New("too many failed requests")
		}
		// Wait a couple of second not to spam
		time.Sleep(2 * time.Second)

	}
	return nil, errors.New("failed to perform mailserver request")
}

func (m *Messenger) scheduleSyncFilters(filters []*transport.Filter) (bool, error) {
	shouldSync, err := m.shouldSync()
	if err != nil {
		m.logger.Error("failed to get shouldSync", zap.Error(err))
		return false, err
	}

	if !shouldSync {
		return false, nil
	}

	go func() {
		// split filters by community store node so we can request the filters to the correct mailserver
		filtersByMs := m.SplitFiltersByStoreNode(filters)
		for communityID, filtersForMs := range filtersByMs {
			ms := m.getActiveMailserver(communityID)
			_, err := m.performMailserverRequest(ms, func(ms mailservers.Mailserver) (*MessengerResponse, error) {
				response, err := m.syncFilters(ms, filtersForMs)

				if err != nil {
					m.logger.Error("failed to sync filter", zap.Error(err))
					return nil, err
				}

				if m.config.messengerSignalsHandler != nil {
					m.config.messengerSignalsHandler.MessengerResponse(response)
				}
				return response, nil
			})
			if err != nil {
				m.logger.Error("failed to perform mailserver request", zap.Error(err))
			}
		}

	}()
	return true, nil
}

func (m *Messenger) calculateMailserverTo() uint32 {
	seconds := float64(m.GetCurrentTimeInMillis()) / 1000
	return uint32(math.Ceil(seconds))
}

func (m *Messenger) calculateMailserverTimeBounds(duration time.Duration) (uint32, uint32) {
	now := float64(m.GetCurrentTimeInMillis()) / 1000
	to := uint32(math.Ceil(now))
	from := uint32(math.Floor(now)) - uint32(duration.Seconds())
	return from, to
}

func (m *Messenger) filtersForChat(chatID string) ([]*transport.Filter, error) {
	chat, ok := m.allChats.Load(chatID)
	if !ok {
		return nil, ErrChatNotFound
	}
	var filters []*transport.Filter

	if chat.OneToOne() {
		// We sync our own topic and any eventual negotiated
		publicKeys := []string{common.PubkeyToHex(&m.identity.PublicKey), chatID}

		filters = m.transport.FiltersByIdentities(publicKeys)

	} else if chat.PrivateGroupChat() {
		var publicKeys []string
		for _, m := range chat.Members {
			publicKeys = append(publicKeys, m.ID)
		}

		filters = m.transport.FiltersByIdentities(publicKeys)

	} else {
		filter := m.transport.FilterByChatID(chatID)
		if filter == nil {
			return nil, ErrNoFiltersForChat
		}
		filters = []*transport.Filter{filter}
	}

	return filters, nil
}

func (m *Messenger) topicsForChat(chatID string) (string, []types.TopicType, error) {
	filters, err := m.filtersForChat(chatID)
	if err != nil {
		return "", nil, err
	}

	var contentTopics []types.TopicType

	for _, filter := range filters {
		contentTopics = append(contentTopics, filter.ContentTopic)
	}

	return filters[0].PubsubTopic, contentTopics, nil
}

func (m *Messenger) syncChatWithFilters(ms mailservers.Mailserver, chatID string) (*MessengerResponse, error) {
	filters, err := m.filtersForChat(chatID)
	if err != nil {
		return nil, err
	}

	return m.syncFilters(ms, filters)
}

func (m *Messenger) syncBackup() error {

	filter := m.transport.PersonalTopicFilter()
	if filter == nil {
		return errors.New("personal topic filter not loaded")
	}

	from, to := m.calculateMailserverTimeBounds(oneMonthDuration)

	batch := MailserverBatch{From: from, To: to, Topics: []types.TopicType{filter.ContentTopic}}
	ms := m.getActiveMailserver(filter.ChatID)
	err := m.processMailserverBatch(*ms, batch)
	if err != nil {
		return err
	}
	return m.settings.SetBackupFetched(true)
}

func (m *Messenger) defaultSyncPeriodFromNow() (uint32, error) {
	defaultSyncPeriod, err := m.settings.GetDefaultSyncPeriod()
	if err != nil {
		return 0, err
	}
	return uint32(m.getTimesource().GetCurrentTime()/1000) - defaultSyncPeriod, nil
}

// capToDefaultSyncPeriod caps the sync period to the default
func (m *Messenger) capToDefaultSyncPeriod(period uint32) (uint32, error) {
	d, err := m.defaultSyncPeriodFromNow()
	if err != nil {
		return 0, err
	}
	if d > period {
		return d, nil
	}
	return period - tolerance, nil
}

func (m *Messenger) updateFiltersPriority(filters []*transport.Filter) {
	for _, filter := range filters {
		chatID := filter.ChatID
		chat := m.Chat(chatID)
		if chat != nil {
			filter.Priority = chat.ReadMessagesAtClockValue
		}
	}
}

func (m *Messenger) resetFiltersPriority(filters []*transport.Filter) {
	for _, filter := range filters {
		filter.Priority = 0
	}
}

func (m *Messenger) SplitFiltersByStoreNode(filters []*transport.Filter) map[string][]*transport.Filter {
	// split filters by community store node so we can request the filters to the correct mailserver
	filtersByMs := make(map[string][]*transport.Filter, len(filters))
	for _, f := range filters {
		communityID := "" // none by default
		if chat, ok := m.allChats.Load(f.ChatID); ok && chat.CommunityChat() && m.communityStorenodes.HasStorenodeSetup(chat.CommunityID) {
			communityID = chat.CommunityID
		}
		if _, exists := filtersByMs[communityID]; !exists {
			filtersByMs[communityID] = make([]*transport.Filter, 0, len(filters))
		}
		filtersByMs[communityID] = append(filtersByMs[communityID], f)
	}
	return filtersByMs
}

// RequestAllHistoricMessages requests all the historic messages for any topic
func (m *Messenger) RequestAllHistoricMessages(forceFetchingBackup, withRetries bool) (*MessengerResponse, error) {
	shouldSync, err := m.shouldSync()
	if err != nil {
		return nil, err
	}

	if !shouldSync {
		return nil, nil
	}

	backupFetched, err := m.settings.BackupFetched()
	if err != nil {
		return nil, err
	}

	if forceFetchingBackup || !backupFetched {
		m.logger.Info("fetching backup")
		err := m.syncBackup()
		if err != nil {
			return nil, err
		}
		m.logger.Info("backup fetched")
	}

	filters := m.transport.Filters()
	m.updateFiltersPriority(filters)
	defer m.resetFiltersPriority(filters)

	filtersByMs := m.SplitFiltersByStoreNode(filters)
	allResponses := &MessengerResponse{}
	for communityID, filtersForMs := range filtersByMs {
		ms := m.getActiveMailserver(communityID)
		if withRetries {
			response, err := m.performMailserverRequest(ms, func(ms mailservers.Mailserver) (*MessengerResponse, error) {
				return m.syncFilters(ms, filtersForMs)
			})
			if err != nil {
				return nil, err
			}
			allResponses.AddChats(response.Chats())
			allResponses.AddMessages(response.Messages())
			continue
		}
		response, err := m.syncFilters(*ms, filtersForMs)
		if err != nil {
			return nil, err
		}
		allResponses.AddChats(response.Chats())
		allResponses.AddMessages(response.Messages())
	}
	return allResponses, nil
}

const missingMessageCheckPeriod = 30 * time.Second

func (m *Messenger) checkForMissingMessagesLoop() {
	t := time.NewTicker(missingMessageCheckPeriod)
	defer t.Stop()

	for {
		select {
		case <-m.quit:
			return

		case <-t.C:
			filters := m.transport.Filters()
			filtersByMs := m.SplitFiltersByStoreNode(filters)
			for communityID, filtersForMs := range filtersByMs {
				ms := m.getActiveMailserver(communityID)
				if ms == nil {
					continue
				}

				peerID, err := ms.PeerID()
				if err != nil {
					m.logger.Error("could not obtain the peerID")
					return
				}
				m.transport.SetCriteriaForMissingMessageVerification(peerID, filtersForMs)
			}
		}
	}
}

func getPrioritizedBatches() []int {
	return []int{1, 5, 10}
}

func (m *Messenger) syncFiltersFrom(ms mailservers.Mailserver, filters []*transport.Filter, lastRequest uint32) (*MessengerResponse, error) {
	response := &MessengerResponse{}
	topicInfo, err := m.mailserversDatabase.Topics()
	if err != nil {
		return nil, err
	}

	topicsData := make(map[string]mailservers.MailserverTopic)
	for _, topic := range topicInfo {
		topicsData[fmt.Sprintf("%s-%s", topic.PubsubTopic, topic.ContentTopic)] = topic
	}

	batches := make(map[string]map[int]MailserverBatch)

	to := m.calculateMailserverTo()
	var syncedTopics []mailservers.MailserverTopic

	sort.Slice(filters[:], func(i, j int) bool {
		p1 := filters[i].Priority
		p2 := filters[j].Priority
		return p1 > p2
	})
	prioritizedBatches := getPrioritizedBatches()
	currentBatch := 0

	if len(filters) == 0 || filters[0].Priority == 0 {
		currentBatch = len(prioritizedBatches)
	}

	defaultPeriodFromNow, err := m.defaultSyncPeriodFromNow()
	if err != nil {
		return nil, err
	}

	contentTopicsPerPubsubTopic := make(map[string]map[string]*transport.Filter)
	for _, filter := range filters {
		if !filter.Listen || filter.Ephemeral {
			continue
		}

		contentTopics, ok := contentTopicsPerPubsubTopic[filter.PubsubTopic]
		if !ok {
			contentTopics = make(map[string]*transport.Filter)
		}
		contentTopics[filter.ContentTopic.String()] = filter
		contentTopicsPerPubsubTopic[filter.PubsubTopic] = contentTopics
	}

	for pubsubTopic, contentTopics := range contentTopicsPerPubsubTopic {
		if _, ok := batches[pubsubTopic]; !ok {
			batches[pubsubTopic] = make(map[int]MailserverBatch)
		}

		for _, filter := range contentTopics {
			var chatID string
			// If the filter has an identity, we use it as a chatID, otherwise is a public chat/community chat filter
			if len(filter.Identity) != 0 {
				chatID = filter.Identity
			} else {
				chatID = filter.ChatID
			}

			topicData, ok := topicsData[filter.PubsubTopic+filter.ContentTopic.String()]
			var capToDefaultSyncPeriod = true
			if !ok {
				if lastRequest == 0 {
					lastRequest = defaultPeriodFromNow
				}
				topicData = mailservers.MailserverTopic{
					PubsubTopic:  filter.PubsubTopic,
					ContentTopic: filter.ContentTopic.String(),
					LastRequest:  int(defaultPeriodFromNow),
				}
			} else if lastRequest != 0 {
				topicData.LastRequest = int(lastRequest)
				capToDefaultSyncPeriod = false
			}

			batchID := topicData.LastRequest

			if currentBatch < len(prioritizedBatches) {
				batch, ok := batches[pubsubTopic][currentBatch]
				if ok {
					prevTopicData, ok := topicsData[batch.PubsubTopic+batch.Topics[0].String()]
					if (!ok && topicData.LastRequest != int(defaultPeriodFromNow)) ||
						(ok && prevTopicData.LastRequest != topicData.LastRequest) {
						currentBatch++
					}
				}
				if currentBatch < len(prioritizedBatches) {
					batchID = currentBatch
					currentBatchCap := prioritizedBatches[currentBatch] - 1
					if currentBatchCap == 0 {
						currentBatch++
					} else {
						prioritizedBatches[currentBatch] = currentBatchCap
					}
				}
			}

			batch, ok := batches[pubsubTopic][batchID]
			if !ok {
				from := uint32(topicData.LastRequest)
				if capToDefaultSyncPeriod {
					from, err = m.capToDefaultSyncPeriod(uint32(topicData.LastRequest))
					if err != nil {
						return nil, err
					}
				}
				batch = MailserverBatch{From: from, To: to}
			}

			batch.ChatIDs = append(batch.ChatIDs, chatID)
			batch.PubsubTopic = pubsubTopic
			batch.Topics = append(batch.Topics, filter.ContentTopic)
			batches[pubsubTopic][batchID] = batch

			// Set last request to the new `to`
			topicData.LastRequest = int(to)
			syncedTopics = append(syncedTopics, topicData)
		}
	}

	if m.config.messengerSignalsHandler != nil {
		m.config.messengerSignalsHandler.HistoryRequestStarted(len(batches))
	}

	var batches24h []MailserverBatch
	for pubsubTopic := range batches {
		batchKeys := make([]int, 0, len(batches[pubsubTopic]))
		for k := range batches[pubsubTopic] {
			batchKeys = append(batchKeys, k)
		}
		sort.Ints(batchKeys)

		keysToIterate := append([]int{}, batchKeys...)
		for {
			// For all batches
			var tmpKeysToIterate []int
			for _, k := range keysToIterate {
				batch := batches[pubsubTopic][k]

				dayBatch := MailserverBatch{
					To:          batch.To,
					Cursor:      batch.Cursor,
					PubsubTopic: batch.PubsubTopic,
					Topics:      batch.Topics,
					ChatIDs:     batch.ChatIDs,
				}

				from := batch.To - uint32(oneDayDuration.Seconds())
				if from > batch.From {
					dayBatch.From = from
					batches24h = append(batches24h, dayBatch)

					// Replace og batch with new dates
					batch.To = from
					batches[pubsubTopic][k] = batch
					tmpKeysToIterate = append(tmpKeysToIterate, k)
				} else {
					batches24h = append(batches24h, batch)
				}
			}

			if len(tmpKeysToIterate) == 0 {
				break
			}
			keysToIterate = tmpKeysToIterate
		}
	}

	for _, batch := range batches24h {
		err := m.processMailserverBatch(ms, batch)
		if err != nil {
			m.logger.Error("error syncing topics", zap.Error(err))
			return nil, err
		}
	}

	m.logger.Debug("topics synced")
	if m.config.messengerSignalsHandler != nil {
		m.config.messengerSignalsHandler.HistoryRequestCompleted()
	}

	err = m.mailserversDatabase.AddTopics(syncedTopics)
	if err != nil {
		return nil, err
	}

	var messagesToBeSaved []*common.Message
	for _, batches := range batches {
		for _, batch := range batches {
			for _, id := range batch.ChatIDs {
				chat, ok := m.allChats.Load(id)
				if !ok || !chat.Active || chat.Timeline() || chat.ProfileUpdates() {
					continue
				}
				gap, err := m.calculateGapForChat(chat, batch.From)
				if err != nil {
					return nil, err
				}
				if chat.SyncedFrom == 0 || chat.SyncedFrom > batch.From {
					chat.SyncedFrom = batch.From
				}

				chat.SyncedTo = to

				err = m.persistence.SetSyncTimestamps(chat.SyncedFrom, chat.SyncedTo, chat.ID)
				if err != nil {
					return nil, err
				}

				response.AddChat(chat)
				if gap != nil {
					response.AddMessage(gap)
					messagesToBeSaved = append(messagesToBeSaved, gap)
				}
			}
		}
	}

	if len(messagesToBeSaved) > 0 {
		err := m.persistence.SaveMessages(messagesToBeSaved)
		if err != nil {
			return nil, err
		}
	}
	return response, nil
}

func (m *Messenger) syncFilters(ms mailservers.Mailserver, filters []*transport.Filter) (*MessengerResponse, error) {
	return m.syncFiltersFrom(ms, filters, 0)
}

func (m *Messenger) calculateGapForChat(chat *Chat, from uint32) (*common.Message, error) {
	// Chat was never synced, no gap necessary
	if chat.SyncedTo == 0 {
		return nil, nil
	}

	// If we filled the gap, nothing to do
	if chat.SyncedTo >= from {
		return nil, nil
	}

	timestamp := m.getTimesource().GetCurrentTime()

	message := &common.Message{
		ChatMessage: &protobuf.ChatMessage{
			ChatId:      chat.ID,
			Text:        "Gap message",
			MessageType: protobuf.MessageType_SYSTEM_MESSAGE_GAP,
			ContentType: protobuf.ChatMessage_SYSTEM_MESSAGE_GAP,
			Clock:       uint64(from) * 1000,
			Timestamp:   timestamp,
		},
		GapParameters: &common.GapParameters{
			From: chat.SyncedTo,
			To:   from,
		},
		From:             common.PubkeyToHex(&m.identity.PublicKey),
		WhisperTimestamp: timestamp,
		LocalChatID:      chat.ID,
		Seen:             true,
		ID:               types.EncodeHex(crypto.Keccak256([]byte(fmt.Sprintf("%s-%d-%d", chat.ID, chat.SyncedTo, from)))),
	}

	return message, m.persistence.SaveMessages([]*common.Message{message})
}

type work struct {
	pubsubTopic   string
	contentTopics []types.TopicType
	cursor        []byte
	storeCursor   *types.StoreRequestCursor
	limit         uint32
}

type messageRequester interface {
	SendMessagesRequestForTopics(
		ctx context.Context,
		peerID []byte,
		from, to uint32,
		previousCursor []byte,
		previousStoreCursor *types.StoreRequestCursor,
		pubsubTopic string,
		contentTopics []types.TopicType,
		limit uint32,
		waitForResponse bool,
		processEnvelopes bool,
	) (cursor []byte, storeCursor *types.StoreRequestCursor, envelopesCount int, err error)
}

func processMailserverBatch(
	ctx context.Context,
	messageRequester messageRequester,
	batch MailserverBatch,
	mailserverID []byte,
	logger *zap.Logger,
	pageLimit uint32,
	shouldProcessNextPage func(int) (bool, uint32),
	processEnvelopes bool,
) error {

	var topicStrings []string
	for _, t := range batch.Topics {
		topicStrings = append(topicStrings, t.String())
	}
	logger = logger.With(zap.Any("chatIDs", batch.ChatIDs),
		zap.String("fromString", time.Unix(int64(batch.From), 0).Format(time.RFC3339)),
		zap.String("toString", time.Unix(int64(batch.To), 0).Format(time.RFC3339)),
		zap.Any("topic", topicStrings),
		zap.Int64("from", int64(batch.From)),
		zap.Int64("to", int64(batch.To)))

	logger.Info("syncing topic")

	wg := sync.WaitGroup{}
	workWg := sync.WaitGroup{}
	workCh := make(chan work, 1000)       // each batch item is split in 10 topics bunch and sent to this channel
	workCompleteCh := make(chan struct{}) // once all batch items are processed, this channel is triggered
	semaphore := make(chan int, 3)        // limit the number of concurrent queries
	errCh := make(chan error)

	ctx, cancel := context.WithCancel(ctx)
	defer cancel()

	// Producer
	wg.Add(1)
	go func() {
		defer func() {
			logger.Debug("mailserver batch producer complete")
			wg.Done()
		}()

		allWorks := int(math.Ceil(float64(len(batch.Topics)) / float64(maxTopicsPerRequest)))
		workWg.Add(allWorks)

		for i := 0; i < len(batch.Topics); i += maxTopicsPerRequest {
			j := i + maxTopicsPerRequest
			if j > len(batch.Topics) {
				j = len(batch.Topics)
			}

			select {
			case <-ctx.Done():
				logger.Debug("processBatch producer - context done")
				return
			default:
				logger.Debug("processBatch producer - creating work")
				workCh <- work{
					pubsubTopic:   batch.PubsubTopic,
					contentTopics: batch.Topics[i:j],
					limit:         pageLimit,
				}
				time.Sleep(50 * time.Millisecond)
			}
		}

		go func() {
			workWg.Wait()
			workCompleteCh <- struct{}{}
		}()

		logger.Debug("processBatch producer complete")
	}()

	var result error

loop:
	for {
		select {
		case <-ctx.Done():
			logger.Debug("processBatch cleanup - context done")
			result = ctx.Err()
			if errors.Is(result, context.Canceled) {
				result = nil
			}
			break loop
		case w, ok := <-workCh:
			if !ok {
				continue
			}

			logger.Debug("processBatch - received work")
			semaphore <- 1
			go func(w work) { // Consumer
				defer func() {
					workWg.Done()
					<-semaphore
				}()

				queryCtx, queryCancel := context.WithTimeout(ctx, mailserverRequestTimeout)
				cursor, storeCursor, envelopesCount, err := messageRequester.SendMessagesRequestForTopics(queryCtx, mailserverID, batch.From, batch.To, w.cursor, w.storeCursor, w.pubsubTopic, w.contentTopics, w.limit, true, processEnvelopes)

				queryCancel()

				if err != nil {
					logger.Debug("failed to send request", zap.Error(err))
					errCh <- err
					return
				}

				processNextPage := true
				nextPageLimit := pageLimit

				if shouldProcessNextPage != nil {
					processNextPage, nextPageLimit = shouldProcessNextPage(envelopesCount)
				}

				if !processNextPage {
					return
				}

				// Check the cursor after calling `shouldProcessNextPage`.
				// The app might use process the fetched envelopes in the callback for own needs.
				if len(cursor) == 0 && storeCursor == nil {
					return
				}

				logger.Debug("processBatch producer - creating work (cursor)")

				workWg.Add(1)
				workCh <- work{
					pubsubTopic:   w.pubsubTopic,
					contentTopics: w.contentTopics,
					cursor:        cursor,
					storeCursor:   storeCursor,
					limit:         nextPageLimit,
				}
			}(w)
		case err := <-errCh:
			logger.Debug("processBatch - received error", zap.Error(err))
			cancel() // Kill go routines
			return err
		case <-workCompleteCh:
			logger.Debug("processBatch - all jobs complete")
			cancel() // Kill go routines
		}
	}

	wg.Wait()

	// NOTE(camellos): Disabling for now, not critical and I'd rather take a bit more time
	// to test it
	//logger.Info("waiting until message processed")
	//m.waitUntilP2PMessagesProcessed()

	logger.Info("synced topic", zap.NamedError("hasError", result))
	return result
}

func (m *Messenger) DisableStoreNodes() {
	m.featureFlags.StoreNodesDisabled = true
}

func (m *Messenger) processMailserverBatch(ms mailservers.Mailserver, batch MailserverBatch) error {
	if m.featureFlags.StoreNodesDisabled {
		return nil
	}

	mailserverID, err := ms.IDBytes()
	if err != nil {
		return err
	}
	logger := m.logger.With(zap.String("mailserverID", ms.ID))
	return processMailserverBatch(m.ctx, m.transport, batch, mailserverID, logger, defaultStoreNodeRequestPageSize, nil, false)
}

func (m *Messenger) processMailserverBatchWithOptions(ms mailservers.Mailserver, batch MailserverBatch, pageLimit uint32, shouldProcessNextPage func(int) (bool, uint32), processEnvelopes bool) error {
	if m.featureFlags.StoreNodesDisabled {
		return nil
	}

	mailserverID, err := ms.IDBytes()
	if err != nil {
		return err
	}
	logger := m.logger.With(zap.String("mailserverID", ms.ID))
	return processMailserverBatch(m.ctx, m.transport, batch, mailserverID, logger, pageLimit, shouldProcessNextPage, processEnvelopes)
}

type MailserverBatch struct {
	From        uint32
	To          uint32
	Cursor      string
	PubsubTopic string
	Topics      []types.TopicType
	ChatIDs     []string
}

func (m *Messenger) SyncChatFromSyncedFrom(chatID string) (uint32, error) {
	chat, ok := m.allChats.Load(chatID)
	if !ok {
		return 0, ErrChatNotFound
	}

	ms := m.getActiveMailserver(chat.CommunityID)
	var from uint32
	_, err := m.performMailserverRequest(ms, func(ms mailservers.Mailserver) (*MessengerResponse, error) {
		pubsubTopic, topics, err := m.topicsForChat(chatID)
		if err != nil {
			return nil, nil
		}

		defaultSyncPeriod, err := m.settings.GetDefaultSyncPeriod()
		if err != nil {
			return nil, err
		}

		batch := MailserverBatch{
			ChatIDs:     []string{chatID},
			To:          chat.SyncedFrom,
			From:        chat.SyncedFrom - defaultSyncPeriod,
			PubsubTopic: pubsubTopic,
			Topics:      topics,
		}
		if m.config.messengerSignalsHandler != nil {
			m.config.messengerSignalsHandler.HistoryRequestStarted(1)
		}

		err = m.processMailserverBatch(ms, batch)
		if err != nil {
			return nil, err
		}

		if m.config.messengerSignalsHandler != nil {
			m.config.messengerSignalsHandler.HistoryRequestCompleted()
		}
		if chat.SyncedFrom == 0 || chat.SyncedFrom > batch.From {
			chat.SyncedFrom = batch.From
		}

		m.logger.Debug("setting sync timestamps", zap.Int64("from", int64(batch.From)), zap.Int64("to", int64(chat.SyncedTo)), zap.String("chatID", chatID))

		err = m.persistence.SetSyncTimestamps(batch.From, chat.SyncedTo, chat.ID)
		from = batch.From
		return nil, err
	})
	if err != nil {
		return 0, err
	}

	return from, nil
}

func (m *Messenger) FillGaps(chatID string, messageIDs []string) error {
	messages, err := m.persistence.MessagesByIDs(messageIDs)
	if err != nil {
		return err
	}

	chat, ok := m.allChats.Load(chatID)
	if !ok {
		return errors.New("chat not existing")
	}

	pubsubTopic, topics, err := m.topicsForChat(chatID)
	if err != nil {
		return err
	}

	var lowestFrom, highestTo uint32

	for _, message := range messages {
		if message.GapParameters == nil {
			return errors.New("can't sync non-gap message")
		}

		if lowestFrom == 0 || lowestFrom > message.GapParameters.From {
			lowestFrom = message.GapParameters.From
		}

		if highestTo < message.GapParameters.To {
			highestTo = message.GapParameters.To
		}
	}

	batch := MailserverBatch{
		ChatIDs:     []string{chatID},
		To:          highestTo,
		From:        lowestFrom,
		PubsubTopic: pubsubTopic,
		Topics:      topics,
	}

	if m.config.messengerSignalsHandler != nil {
		m.config.messengerSignalsHandler.HistoryRequestStarted(1)
	}

	ms := m.getActiveMailserver(chat.CommunityID)
	err = m.processMailserverBatch(*ms, batch)
	if err != nil {
		return err
	}

	if m.config.messengerSignalsHandler != nil {
		m.config.messengerSignalsHandler.HistoryRequestCompleted()
	}

	return m.persistence.DeleteMessages(messageIDs)
}

func (m *Messenger) waitUntilP2PMessagesProcessed() { // nolint: unused

	ticker := time.NewTicker(50 * time.Millisecond)

	for { //nolint: gosimple
		select {
		case <-ticker.C:
			if !m.transport.ProcessingP2PMessages() {
				ticker.Stop()
				return
			}
		}
	}
}

func (m *Messenger) LoadFilters(filters []*transport.Filter) ([]*transport.Filter, error) {
	return m.transport.LoadFilters(filters)
}

func (m *Messenger) ToggleUseMailservers(value bool) error {
	m.mailserverCycle.Lock()
	defer m.mailserverCycle.Unlock()

	err := m.settings.SetUseMailservers(value)
	if err != nil {
		return err
	}

	if value {
		m.cycleMailservers()
		return nil
	}

	m.disconnectActiveMailserver()
	return nil
}

func (m *Messenger) SetPinnedMailservers(mailservers map[string]string) error {
	err := m.settings.SetPinnedMailservers(mailservers)
	if err != nil {
		return err
	}

	m.cycleMailservers()
	return nil
}

func (m *Messenger) RemoveFilters(filters []*transport.Filter) error {
	return m.transport.RemoveFilters(filters)
}

func (m *Messenger) ConnectionChanged(state connection.State) {
	m.transport.ConnectionChanged(state)
	if !m.connectionState.Offline && state.Offline {
		m.sender.StopDatasync()
	}

	if m.connectionState.Offline && !state.Offline {
		err := m.sender.StartDatasync(m.mvdsStatusChangeEvent, m.sendDataSync)
		if err != nil {
			m.logger.Error("failed to start datasync", zap.Error(err))
		}
	}

	m.connectionState = state
}

func (m *Messenger) fetchMessages(chatID string, duration time.Duration) (uint32, error) {
	from, to := m.calculateMailserverTimeBounds(duration)

	chat, ok := m.allChats.Load(chatID)
	if !ok {
		return 0, ErrChatNotFound
	}

	ms := m.getActiveMailserver(chat.CommunityID)
	_, err := m.performMailserverRequest(ms, func(ms mailservers.Mailserver) (*MessengerResponse, error) {
		m.logger.Debug("fetching messages", zap.String("chatID", chatID), zap.String("mailserver", ms.Name))
		pubsubTopic, topics, err := m.topicsForChat(chatID)
		if err != nil {
			return nil, nil
		}

		batch := MailserverBatch{
			ChatIDs:     []string{chatID},
			From:        from,
			To:          to,
			PubsubTopic: pubsubTopic,
			Topics:      topics,
		}
		if m.config.messengerSignalsHandler != nil {
			m.config.messengerSignalsHandler.HistoryRequestStarted(1)
		}

		err = m.processMailserverBatch(ms, batch)
		if err != nil {
			return nil, err
		}

		if m.config.messengerSignalsHandler != nil {
			m.config.messengerSignalsHandler.HistoryRequestCompleted()
		}
		if chat.SyncedFrom == 0 || chat.SyncedFrom > batch.From {
			chat.SyncedFrom = batch.From
		}

		m.logger.Debug("setting sync timestamps", zap.Int64("from", int64(batch.From)), zap.Int64("to", int64(chat.SyncedTo)), zap.String("chatID", chatID))

		err = m.persistence.SetSyncTimestamps(batch.From, chat.SyncedTo, chat.ID)
		from = batch.From
		return nil, err
	})
	if err != nil {
		return 0, err
	}

	return from, nil
}