status-go/protocol/messenger_mailserver.go

1194 lines
31 KiB
Go
Raw Normal View History

2021-03-25 15:15:22 +00:00
package protocol
import (
"context"
"fmt"
"math"
"sort"
"sync"
2021-07-27 11:52:10 +00:00
"time"
2021-03-25 15:15:22 +00:00
"github.com/pkg/errors"
"go.uber.org/zap"
2021-05-14 10:55:42 +00:00
"github.com/status-im/status-go/connection"
2021-03-25 15:15:22 +00:00
"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
2021-05-14 10:55:42 +00:00
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")
2021-05-14 10:55:42 +00:00
func (m *Messenger) shouldSync() (bool, error) {
// TODO (pablo) support community store node as well
if m.mailserverCycle.activeMailserver == nil || !m.Online() {
2021-05-14 10:55:42 +00:00
return false, nil
}
2021-03-25 15:15:22 +00:00
2021-05-14 10:55:42 +00:00
useMailserver, err := m.settings.CanUseMailservers()
2021-03-25 15:15:22 +00:00
if err != nil {
m.logger.Error("failed to get use mailservers", zap.Error(err))
2021-05-14 10:55:42 +00:00
return false, err
2021-03-25 15:15:22 +00:00
}
return useMailserver, nil
2021-05-14 10:55:42 +00:00
}
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
2021-03-25 15:15:22 +00:00
}
go func() {
ms := m.getActiveMailserver(chat.CommunityID)
_, err = m.performMailserverRequest(ms, func(mailServer mailservers.Mailserver) (*MessengerResponse, error) {
response, err := m.syncChatWithFilters(mailServer, chat.ID)
2021-03-25 15:15:22 +00:00
if err != nil {
m.logger.Error("failed to sync chat", zap.Error(err))
return nil, err
}
2021-03-25 15:15:22 +00:00
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))
2021-03-25 15:15:22 +00:00
}
}()
2021-05-14 10:55:42 +00:00
return true, nil
2021-03-25 15:15:22 +00:00
}
func (m *Messenger) connectToNewMailserverAndWait() error {
// Handle pinned mailservers
m.logger.Info("disconnecting mailserver")
pinnedMailserver, err := m.getPinnedMailserver()
2021-05-14 10:55:42 +00:00
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(graylistBackoff)
2021-05-14 10:55:42 +00:00
}
2021-03-25 15:15:22 +00:00
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")
2021-03-25 15:15:22 +00:00
}
2021-10-27 10:59:43 +00:00
2021-05-14 10:55:42 +00:00
func (m *Messenger) scheduleSyncFilters(filters []*transport.Filter) (bool, error) {
shouldSync, err := m.shouldSync()
2021-03-25 15:15:22 +00:00
if err != nil {
2021-05-14 10:55:42 +00:00
m.logger.Error("failed to get shouldSync", zap.Error(err))
return false, err
2021-03-25 15:15:22 +00:00
}
2021-05-14 10:55:42 +00:00
if !shouldSync {
return false, nil
2021-03-25 15:15:22 +00:00
}
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)
2021-03-25 15:15:22 +00:00
if err != nil {
m.logger.Error("failed to sync filter", zap.Error(err))
return nil, err
}
2021-03-25 15:15:22 +00:00
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))
}
2021-03-25 15:15:22 +00:00
}
}()
2021-05-14 10:55:42 +00:00
return true, nil
2021-03-25 15:15:22 +00:00
}
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
2021-03-25 15:15:22 +00:00
}
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
2021-03-25 15:15:22 +00:00
}
filters = []*transport.Filter{filter}
}
return filters, nil
}
func (m *Messenger) topicsForChat(chatID string) (string, []types.TopicType, error) {
2021-03-25 15:15:22 +00:00
filters, err := m.filtersForChat(chatID)
if err != nil {
return "", nil, err
2021-03-25 15:15:22 +00:00
}
var contentTopics []types.TopicType
2021-03-25 15:15:22 +00:00
for _, filter := range filters {
contentTopics = append(contentTopics, filter.ContentTopic)
2021-03-25 15:15:22 +00:00
}
return filters[0].PubsubTopic, contentTopics, nil
2021-03-25 15:15:22 +00:00
}
func (m *Messenger) syncChatWithFilters(ms mailservers.Mailserver, chatID string) (*MessengerResponse, error) {
2021-03-25 15:15:22 +00:00
filters, err := m.filtersForChat(chatID)
if err != nil {
return nil, err
}
return m.syncFilters(ms, filters)
2021-03-25 15:15:22 +00:00
}
2021-10-27 10:59:43 +00:00
func (m *Messenger) syncBackup() error {
filter := m.transport.PersonalTopicFilter()
if filter == nil {
return errors.New("personal topic filter not loaded")
}
canSync, err := m.canSyncWithStoreNodes()
if err != nil {
return err
}
if !canSync {
return nil
}
2021-10-27 10:59:43 +00:00
from, to := m.calculateMailserverTimeBounds(oneMonthDuration)
2023-11-15 15:58:15 +00:00
batch := MailserverBatch{From: from, To: to, Topics: []types.TopicType{filter.ContentTopic}}
ms := m.getActiveMailserver(filter.ChatID)
err = m.processMailserverBatch(*ms, batch)
2021-10-27 10:59:43 +00:00
if err != nil {
return err
}
return m.settings.SetBackupFetched(true)
}
2021-06-01 09:29:37 +00:00
func (m *Messenger) defaultSyncPeriodFromNow() (uint32, error) {
defaultSyncPeriod, err := m.settings.GetDefaultSyncPeriod()
if err != nil {
return 0, err
}
2021-05-31 14:35:14 +00:00
return uint32(m.getTimesource().GetCurrentTime()/1000) - defaultSyncPeriod, nil
2021-03-25 15:15:22 +00:00
}
2021-06-01 09:29:37 +00:00
// 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
}
2021-03-25 15:15:22 +00:00
if d > period {
return d, nil
2021-03-25 15:15:22 +00:00
}
return period - tolerance, nil
2021-03-25 15:15:22 +00:00
}
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
}
2021-03-25 15:15:22 +00:00
// RequestAllHistoricMessages requests all the historic messages for any topic
func (m *Messenger) RequestAllHistoricMessages(forceFetchingBackup, withRetries bool) (*MessengerResponse, error) {
2021-05-14 10:55:42 +00:00
shouldSync, err := m.shouldSync()
2021-03-25 15:15:22 +00:00
if err != nil {
return nil, err
}
2021-05-14 10:55:42 +00:00
if !shouldSync {
2021-03-25 15:15:22 +00:00
return nil, nil
}
2021-10-27 10:59:43 +00:00
backupFetched, err := m.settings.BackupFetched()
if err != nil {
return nil, err
}
if forceFetchingBackup || !backupFetched {
2021-10-27 10:59:43 +00:00
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
}
if response != nil {
allResponses.AddChats(response.Chats())
allResponses.AddMessages(response.Messages())
}
continue
}
response, err := m.syncFilters(*ms, filtersForMs)
if err != nil {
return nil, err
}
if response != nil {
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}
2021-03-25 15:15:22 +00:00
}
func (m *Messenger) syncFiltersFrom(ms mailservers.Mailserver, filters []*transport.Filter, lastRequest uint32) (*MessengerResponse, error) {
canSync, err := m.canSyncWithStoreNodes()
if err != nil {
return nil, err
}
if !canSync {
return nil, nil
}
2021-03-25 15:15:22 +00:00
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
2021-03-25 15:15:22 +00:00
}
batches := make(map[string]map[int]MailserverBatch)
2021-03-25 15:15:22 +00:00
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)
2021-03-25 15:15:22 +00:00
for _, filter := range filters {
if !filter.Listen || filter.Ephemeral {
continue
}
contentTopics, ok := contentTopicsPerPubsubTopic[filter.PubsubTopic]
2021-03-25 15:15:22 +00:00
if !ok {
contentTopics = make(map[string]*transport.Filter)
2021-03-25 15:15:22 +00:00
}
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}
}
2021-03-25 15:15:22 +00:00
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)
}
2021-03-25 15:15:22 +00:00
}
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)
2021-03-25 15:15:22 +00:00
if err != nil {
m.logger.Error("error syncing topics", zap.Error(err))
2021-03-25 15:15:22 +00:00
return nil, err
}
}
2021-10-29 14:29:28 +00:00
m.logger.Debug("topics synced")
if m.config.messengerSignalsHandler != nil {
m.config.messengerSignalsHandler.HistoryRequestCompleted()
}
2021-03-25 15:15:22 +00:00
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
}
2021-03-25 15:15:22 +00:00
chat.SyncedTo = to
2021-03-25 15:15:22 +00:00
err = m.persistence.SetSyncTimestamps(chat.SyncedFrom, chat.SyncedTo, chat.ID)
if err != nil {
return nil, err
}
2021-03-25 15:15:22 +00:00
response.AddChat(chat)
if gap != nil {
response.AddMessage(gap)
messagesToBeSaved = append(messagesToBeSaved, gap)
}
2021-03-25 15:15:22 +00:00
}
}
}
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)
}
2021-03-25 15:15:22 +00:00
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{
2021-03-25 15:15:22 +00:00
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
}
2023-04-27 17:42:31 +00:00
type messageRequester interface {
SendMessagesRequestForTopics(
ctx context.Context,
peerID []byte,
from, to uint32,
previousCursor []byte,
previousStoreCursor types.StoreRequestCursor,
pubsubTopic string,
contentTopics []types.TopicType,
limit uint32,
2023-04-27 17:42:31 +00:00
waitForResponse bool,
processEnvelopes bool,
) (cursor []byte, storeCursor types.StoreRequestCursor, envelopesCount int, err error)
2023-04-27 17:42:31 +00:00
}
func processMailserverBatch(
ctx context.Context,
messageRequester messageRequester,
batch MailserverBatch,
mailserverID []byte,
logger *zap.Logger,
pageLimit uint32,
shouldProcessNextPage func(int) (bool, uint32),
processEnvelopes bool,
) error {
2021-10-29 14:29:28 +00:00
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)))
2021-10-29 14:29:28 +00:00
logger.Info("syncing topic")
2021-07-27 11:52:10 +00:00
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
2023-04-27 17:42:31 +00:00
errCh := make(chan error)
ctx, cancel := context.WithCancel(ctx)
defer cancel()
// Producer
wg.Add(1)
go func() {
2023-04-27 17:42:31 +00:00
defer func() {
logger.Debug("mailserver batch producer complete")
wg.Done()
2023-04-27 17:42:31 +00:00
}()
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 {
2023-04-27 17:42:31 +00:00
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()
2021-03-25 15:15:22 +00:00
if err != nil {
logger.Debug("failed to send request", zap.Error(err))
2023-04-27 17:42:31 +00:00
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
}
2021-03-25 15:15:22 +00:00
}
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()
2023-04-27 17:42:31 +00:00
logger.Info("synced topic", zap.NamedError("hasError", result))
return result
}
func (m *Messenger) canSyncWithStoreNodes() (bool, error) {
if m.featureFlags.StoreNodesDisabled {
return false, nil
}
if m.connectionState.IsExpensive() {
return m.settings.CanSyncOnMobileNetwork()
}
return true, nil
}
func (m *Messenger) DisableStoreNodes() {
m.featureFlags.StoreNodesDisabled = true
}
func (m *Messenger) processMailserverBatch(ms mailservers.Mailserver, batch MailserverBatch) error {
canSync, err := m.canSyncWithStoreNodes()
if err != nil {
return err
}
if !canSync {
return nil
}
mailserverID, err := ms.IDBytes()
2023-04-27 17:42:31 +00:00
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 {
canSync, err := m.canSyncWithStoreNodes()
if err != nil {
return err
}
if !canSync {
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)
2021-03-25 15:15:22 +00:00
}
type MailserverBatch struct {
From uint32
To uint32
Cursor string
PubsubTopic string
Topics []types.TopicType
ChatIDs []string
2021-03-25 15:15:22 +00:00
}
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) {
canSync, err := m.canSyncWithStoreNodes()
if err != nil {
return nil, err
}
if !canSync {
return nil, nil
}
pubsubTopic, topics, err := m.topicsForChat(chatID)
if err != nil {
return nil, nil
}
2021-03-25 15:15:22 +00:00
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)
}
2021-03-25 15:15:22 +00:00
err = m.processMailserverBatch(ms, batch)
if err != nil {
return nil, err
}
2021-03-25 15:15:22 +00:00
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
})
2021-03-25 15:15:22 +00:00
if err != nil {
return 0, err
}
return from, nil
2021-03-25 15:15:22 +00:00
}
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)
2021-03-25 15:15:22 +00:00
if !ok {
return errors.New("chat not existing")
}
pubsubTopic, topics, err := m.topicsForChat(chatID)
2021-03-25 15:15:22 +00:00
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,
2021-03-25 15:15:22 +00:00
}
if m.config.messengerSignalsHandler != nil {
m.config.messengerSignalsHandler.HistoryRequestStarted(1)
}
ms := m.getActiveMailserver(chat.CommunityID)
err = m.processMailserverBatch(*ms, batch)
2021-03-25 15:15:22 +00:00
if err != nil {
return err
}
if m.config.messengerSignalsHandler != nil {
m.config.messengerSignalsHandler.HistoryRequestCompleted()
}
2021-03-25 15:15:22 +00:00
return m.persistence.DeleteMessages(messageIDs)
}
func (m *Messenger) waitUntilP2PMessagesProcessed() { // nolint: unused
2021-11-26 12:30:35 +00:00
ticker := time.NewTicker(50 * time.Millisecond)
for { //nolint: gosimple
select {
case <-ticker.C:
if !m.transport.ProcessingP2PMessages() {
ticker.Stop()
return
}
}
}
}
2021-03-25 15:15:22 +00:00
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
}
m.disconnectActiveMailserver(backoffByUserAction)
if value {
m.cycleMailservers()
return nil
}
return nil
}
func (m *Messenger) SetPinnedMailservers(mailservers map[string]string) error {
err := m.settings.SetPinnedMailservers(mailservers)
if err != nil {
return err
}
m.disconnectActiveMailserver(backoffByUserAction)
m.cycleMailservers()
return nil
}
2021-03-25 15:15:22 +00:00
func (m *Messenger) RemoveFilters(filters []*transport.Filter) error {
return m.transport.RemoveFilters(filters)
}
2021-05-14 10:55:42 +00:00
func (m *Messenger) ConnectionChanged(state connection.State) {
m.transport.ConnectionChanged(state)
2021-10-21 12:39:19 +00:00
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))
}
2021-10-21 12:39:19 +00:00
}
2021-05-14 10:55:42 +00:00
m.connectionState = state
}
2023-11-24 14:51:09 +00:00
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) {
canSync, err := m.canSyncWithStoreNodes()
if err != nil {
return nil, err
}
if !canSync {
return nil, nil
}
m.logger.Debug("fetching messages", zap.String("chatID", chatID), zap.String("mailserver", ms.Name))
2023-11-24 14:51:09 +00:00
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)
2023-11-24 14:51:09 +00:00
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
}