status-go/protocol/messenger_mailserver.go

1052 lines
27 KiB
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 = 20
// tolerance is how many seconds of potentially out-of-order messages we want to fetch
tolerance uint32 = 60
mailserverRequestTimeout = 30 * time.Second
oneMonthInSeconds uint32 = 31 * 24 * 60 * 60
mailserverMaxTries uint = 2
mailserverMaxFailedRequests uint = 2
OneDayInSeconds = 86400
)
// 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) {
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() {
_, err := m.performMailserverRequest(func() (*MessengerResponse, error) {
response, err := m.syncChatWithFilters(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(fn func() (*MessengerResponse, error)) (*MessengerResponse, error) {
m.mailserverCycle.RLock()
defer m.mailserverCycle.RUnlock()
var tries uint = 0
for tries < mailserverMaxTries {
if !m.isActiveMailserverAvailable() {
return nil, errors.New("mailserver not available")
}
m.logger.Info("trying performing mailserver requests", zap.Uint("try", tries))
activeMailserver := m.getActiveMailserver()
// Make sure we are connected to a mailserver
if activeMailserver == nil {
return nil, errors.New("mailserver not available")
}
// Peform request
response, err := fn()
if err == nil {
// Reset failed requests
m.logger.Debug("mailserver request performed successfully",
zap.String("mailserverID", activeMailserver.ID))
activeMailserver.FailedRequests = 0
return response, nil
}
m.logger.Error("failed to perform mailserver request",
zap.String("mailserverID", activeMailserver.ID),
zap.Uint("tries", tries),
zap.Error(err),
)
tries++
// Increment failed requests
activeMailserver.FailedRequests++
// Change mailserver
if activeMailserver.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() {
_, err := m.performMailserverRequest(func() (*MessengerResponse, error) {
response, err := m.syncFilters(filters)
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 {
return uint32(m.getTimesource().GetCurrentTime() / 1000)
}
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(chatID string) (*MessengerResponse, error) {
filters, err := m.filtersForChat(chatID)
if err != nil {
return nil, err
}
return m.syncFilters(filters)
}
func (m *Messenger) syncBackup() error {
filter := m.transport.PersonalTopicFilter()
if filter == nil {
return errors.New("personal topic filter not loaded")
}
to := m.calculateMailserverTo()
from := uint32(m.getTimesource().GetCurrentTime()/1000) - oneMonthInSeconds
batch := MailserverBatch{From: from, To: to, Topics: []types.TopicType{filter.ContentTopic}}
err := m.processMailserverBatch(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) RequestAllHistoricMessagesWithRetries(forceFetchingBackup bool) (*MessengerResponse, error) {
return m.performMailserverRequest(func() (*MessengerResponse, error) {
return m.RequestAllHistoricMessages(forceFetchingBackup)
})
}
// RequestAllHistoricMessages requests all the historic messages for any topic
func (m *Messenger) RequestAllHistoricMessages(forceFetchingBackup 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)
response, err := m.syncFilters(filters)
if err != nil {
return nil, err
}
return response, nil
}
func getPrioritizedBatches() []int {
return []int{1, 5, 10}
}
func (m *Messenger) syncFiltersFrom(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 - OneDayInSeconds
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
}
}
i := 0
for _, batch := range batches24h {
i++
err := m.processMailserverBatch(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(filters []*transport.Filter) (*MessengerResponse, error) {
return m.syncFiltersFrom(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) processMailserverBatch(batch MailserverBatch) error {
if m.featureFlags.StoreNodesDisabled {
return nil
}
mailserverID, err := m.activeMailserverID()
if err != nil {
return err
}
return processMailserverBatch(m.ctx, m.transport, batch, mailserverID, m.logger, defaultStoreNodeRequestPageSize, nil, false)
}
func (m *Messenger) processMailserverBatchWithOptions(batch MailserverBatch, pageLimit uint32, shouldProcessNextPage func(int) (bool, uint32), processEnvelopes bool) error {
if m.featureFlags.StoreNodesDisabled {
return nil
}
mailserverID, err := m.activeMailserverID()
if err != nil {
return err
}
return processMailserverBatch(m.ctx, m.transport, batch, mailserverID, m.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) {
var from uint32
_, err := m.performMailserverRequest(func() (*MessengerResponse, error) {
pubsubTopic, topics, err := m.topicsForChat(chatID)
if err != nil {
return nil, nil
}
chat, ok := m.allChats.Load(chatID)
if !ok {
return nil, ErrChatNotFound
}
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(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
}
_, 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)
}
err = m.processMailserverBatch(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.sendDataSync)
if err != nil {
m.logger.Error("failed to start datasync", zap.Error(err))
}
}
m.connectionState = state
}
func (m *Messenger) fetchMessages(chatID string, duration uint32) (uint32, error) {
to := uint32(m.getTimesource().GetCurrentTime() / 1000)
from := to - duration
_, err := m.performMailserverRequest(func() (*MessengerResponse, error) {
pubsubTopic, topics, err := m.topicsForChat(chatID)
if err != nil {
return nil, nil
}
chat, ok := m.allChats.Load(chatID)
if !ok {
return nil, ErrChatNotFound
}
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(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
}