backfill: skip unparseable history messages without content (#1046)

Co-authored-by: Indent <noreply@indent.com>

pkg/connector/backfill.go

@@ -293,11 +293,17 @@ func (wa *WhatsAppClient) handleWAHistorySync(
 		for i, rawMsg := range rawMessages {
 			// Don't store messages that will just be skipped.
 			msgEvt, err := wa.Client.ParseWebMessage(jid, rawMsg.GetMessage())
-			if err != nil {
+			if err != nil && getMessageType(rawMsg.GetMessage().GetMessage()) == "ignore" {
+				ignoredTypes++
+				continue
+			} else if err != nil {
 				log.Warn().Err(err).
 					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()).
@@ -399,8 +405,6 @@ 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").
@@ -444,9 +448,6 @@ 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 {
@@ -487,7 +488,6 @@ 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,
@@ -503,7 +503,6 @@ 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()
 				},
 			},
@@ -511,7 +510,6 @@ 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
 		}
@@ -519,9 +517,6 @@ 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 {

pkg/connector/client.go

@@ -50,7 +50,6 @@ 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),
@@ -115,7 +114,6 @@ 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

pkg/connector/handlewhatsapp.go

@@ -374,6 +374,14 @@ 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).