diff --git a/pkg/connector/backfill.go b/pkg/connector/backfill.go index 8f006e0..e2a18b1 100644 --- a/pkg/connector/backfill.go +++ b/pkg/connector/backfill.go @@ -298,9 +298,6 @@ func (wa *WhatsAppClient) handleWAHistorySync( Int("msg_index", i). Str("msg_id", rawMsg.GetMessage().GetKey().GetID()). Uint64("msg_time_seconds", rawMsg.GetMessage().GetMessageTimestamp()). - Stringer("stub_type", rawMsg.GetMessage().GetMessageStubType()). - Str("message_type", getMessageType(rawMsg.GetMessage().GetMessage())). - Bool("from_me", rawMsg.GetMessage().GetKey().GetFromMe()). Msg("Dropping historical message due to parse error") log.Trace(). Any("web_message_info", rawMsg.GetMessage()). @@ -402,6 +399,8 @@ func (wa *WhatsAppClient) handleWAHistorySync( return nil } +const historySyncQueuedTimeout = 10 * time.Minute + func (wa *WhatsAppClient) createPortalsFromHistorySync(ctx context.Context) { log := wa.UserLogin.Log.With(). Str("action", "create portals from history sync"). @@ -445,6 +444,9 @@ func (wa *WhatsAppClient) createPortalsFromHistorySync(ctx context.Context) { // We don't currently support new PSAs, so don't bother backfilling them either wg.Done() continue + } else if queuedAt, ok := wa.historySyncQueued.Get(conv.ChatJID); ok && time.Since(queuedAt) < historySyncQueuedTimeout { + wg.Done() + continue } // TODO can the chat info fetch be avoided entirely? select { @@ -485,6 +487,7 @@ func (wa *WhatsAppClient) createPortalsFromHistorySync(ctx context.Context) { wg.Done() continue } + wa.historySyncQueued.Set(conv.ChatJID, time.Now()) res := wa.UserLogin.QueueRemoteEvent(&simplevent.ChatResync{ EventMeta: simplevent.EventMeta{ Type: bridgev2.RemoteEventChatResync, @@ -500,6 +503,7 @@ func (wa *WhatsAppClient) createPortalsFromHistorySync(ctx context.Context) { if err != nil { zerolog.Ctx(ctx).Err(err).Msg("Failed to mark conversation as bridged") } + wa.historySyncQueued.Delete(conv.ChatJID) wg.Done() }, }, @@ -507,6 +511,7 @@ func (wa *WhatsAppClient) createPortalsFromHistorySync(ctx context.Context) { LatestMessageTS: conv.LastMessageTimestamp, }) if !res.Success { + wa.historySyncQueued.Delete(conv.ChatJID) log.Debug().Msg("Cancelling history sync portal creation loop") return } @@ -514,6 +519,9 @@ func (wa *WhatsAppClient) createPortalsFromHistorySync(ctx context.Context) { log.Info().Int("conversation_count", len(conversations)).Msg("Finished creating portals from history sync") go func() { wg.Wait() + if wa.historySyncQueued.Len() > 0 { + return + } wa.UserLogin.Metadata.(*waid.UserLoginMetadata).HistorySyncPortalsNeedCreating = false err = wa.UserLogin.Save(ctx) if err != nil { diff --git a/pkg/connector/client.go b/pkg/connector/client.go index 678c8d2..9b3f190 100644 --- a/pkg/connector/client.go +++ b/pkg/connector/client.go @@ -50,6 +50,7 @@ func (wa *WhatsAppConnector) LoadUserLogin(ctx context.Context, login *bridgev2. MC: noopMCInstance, historySyncWakeup: make(chan struct{}, 1), + historySyncQueued: exsync.NewMap[types.JID, time.Time](), resyncQueue: make(map[types.JID]resyncQueueItem), directMediaRetries: make(map[networkid.MessageID]*directMediaRetry), mediaRetryLock: semaphore.NewWeighted(wa.Config.HistorySync.MediaRequests.MaxAsyncHandle), @@ -114,6 +115,7 @@ type WhatsAppClient struct { MC mClient historySyncWakeup chan struct{} + historySyncQueued *exsync.Map[types.JID, time.Time] stopLoops atomic.Pointer[context.CancelFunc] resyncQueue map[types.JID]resyncQueueItem resyncQueueLock sync.Mutex diff --git a/pkg/connector/handlewhatsapp.go b/pkg/connector/handlewhatsapp.go index e6174b9..4391233 100644 --- a/pkg/connector/handlewhatsapp.go +++ b/pkg/connector/handlewhatsapp.go @@ -374,14 +374,6 @@ func (wa *WhatsAppClient) handleWAMessage(ctx context.Context, evt *events.Messa if assocType := messageAssoc.GetAssociationType(); assocType == waE2E.MessageAssociation_HD_IMAGE_DUAL_UPLOAD || assocType == waE2E.MessageAssociation_HD_VIDEO_DUAL_UPLOAD { parentKey := messageAssoc.GetParentMessageKey() protocolMsg, shouldHideEdit := makeHDMediaReplacementEdit(evt.Message, parentKey) - if protocolMsg.GetEditedMessage() == nil { - wa.UserLogin.Log.Debug(). - Str("message_id", evt.Info.ID). - Str("parent_id", parentKey.GetID()). - Stringer("assoc_type", assocType). - Msg("Ignoring HD replacement message without replacement content") - return - } dontRenderEdited = shouldHideEdit wa.UserLogin.Log.Debug(). Str("message_id", evt.Info.ID).