backfill: delete history sync messages up to the newest bridged message on backward fetches

pkg/connector/backfill.go

@@ -293,17 +293,11 @@ 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 && getMessageType(rawMsg.GetMessage().GetMessage()) == "ignore" {
+			if err != nil {
-				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()).
@@ -572,6 +566,15 @@ func (wa *WhatsAppClient) FetchMessages(ctx context.Context, params bridgev2.Fet
 			anchorID = parsedID.ID
 		}
 	}
+	var lastMessageTS uint64
+	if !params.Forward {
+		lastMessage, err := wa.Main.Bridge.DB.Message.GetLastPartAtOrBeforeTime(ctx, params.Portal.PortalKey, time.Now())
+		if err != nil {
+			return nil, fmt.Errorf("failed to get last message in portal: %w", err)
+		} else if lastMessage != nil {
+			lastMessageTS = uint64(lastMessage.Timestamp.Unix())
+		}
+	}
 	var hasMore bool
 	if !params.Forward && wa.Main.Config.HistorySync.BackwardsOnDemand {
 		hasMore = conv != nil && ptr.Val(conv.EndOfHistoryTransferType) == waHistorySync.Conversation_COMPLETE_BUT_MORE_MESSAGES_REMAIN_ON_PRIMARY
@@ -580,7 +583,11 @@ func (wa *WhatsAppClient) FetchMessages(ctx context.Context, params bridgev2.Fet
 	if err != nil {
 		return nil, fmt.Errorf("failed to load messages from database: %w", err)
 	} else if len(messages) == 0 || (len(messages) == 1 && anchorID != "" && messages[0].GetKey().GetID() == anchorID) {
-		wa.deleteHistorySyncMessages(ctx, portalJID, 0, 0)
+		if startTime != nil {
+			wa.deleteHistorySyncMessages(ctx, portalJID, 1<<62, uint64(startTime.Unix()))
+		} else {
+			wa.deleteHistorySyncMessages(ctx, portalJID, lastMessageTS, 0)
+		}
 		hasMore = hasMore && params.AnchorMessage != nil
 		if hasMore && !params.AllowSlowFetch {
 			return &bridgev2.FetchMessagesResponse{
@@ -608,7 +615,14 @@ func (wa *WhatsAppClient) FetchMessages(ctx context.Context, params bridgev2.Fet
 			}
 		}
 	}
-	resp, err := wa.convertHistorySyncMessages(ctx, params.Portal, portalJID, messages, true, !params.Forward && !hasMore)
+	var deleteUpTo uint64
+	if !params.Forward && !hasMore {
+		deleteUpTo = 1 << 62
+		if lastMessageTS != 0 {
+			deleteUpTo = lastMessageTS
+		}
+	}
+	resp, err := wa.convertHistorySyncMessages(ctx, params.Portal, portalJID, messages, true, deleteUpTo)
 	if err != nil {
 		return nil, fmt.Errorf("failed to convert messages: %w", err)
 	}
@@ -670,7 +684,7 @@ func (wa *WhatsAppClient) convertHistorySyncMessages(
 	portalJID types.JID,
 	messages []*waWeb.WebMessageInfo,
 	explodeOnError bool,
-	deleteEverything bool,
+	deleteUpTo uint64,
 ) (*bridgev2.FetchMessagesResponse, error) {
 	cli := wa.Client
 	if cli == nil {
@@ -729,11 +743,7 @@ func (wa *WhatsAppClient) convertHistorySyncMessages(
 		Messages: convertedMessages,
 		Cursor:   networkid.PaginationCursor(strconv.FormatUint(oldestTS, 10)),
 		CompleteCallback: func() {
-			if deleteEverything {
+			wa.deleteHistorySyncMessages(ctx, portalJID, max(newestTS, deleteUpTo), oldestTS)
-				wa.deleteHistorySyncMessages(ctx, portalJID, 1<<62, oldestTS)
-			} else {
-				wa.deleteHistorySyncMessages(ctx, portalJID, newestTS, oldestTS)
-			}
 			if len(mediaRequests) > 0 {
 				go func(ctx context.Context) {
 					for _, req := range mediaRequests {
@@ -896,7 +906,7 @@ func (wa *WhatsAppClient) handleOnDemandHistorySync(ctx context.Context, blob *w
 					Int("message_count", len(messages)).
 					Stringer("end_of_history_type", conv.GetEndOfHistoryTransferType()).
 					Msg("Converting messages to bridge from on-demand history sync")
-				resp, err := wa.convertHistorySyncMessages(ctx, portal, portalJID, messages, false, false)
+				resp, err := wa.convertHistorySyncMessages(ctx, portal, portalJID, messages, false, 0)
 				if err != nil {
 					return nil, err
 				}

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).

pkg/msgconv/from-whatsapp.go

@@ -322,7 +322,6 @@ func (mc *MessageConverter) ToMatrix(
 		cm.ThreadRoot = ptr.Ptr(waid.MakeMessageID(chat, pcp, commentTarget.GetID()))
 	}
 	mc.cutOffLongMessage(part.Content)
-	logOversizedPart(ctx, part)
 
 	return cm
 }

pkg/msgconv/wa-text.go

@@ -18,7 +18,6 @@ package msgconv
 
 import (
 	"context"
-	"encoding/json"
 	"fmt"
 	"html"
 	"regexp"
@@ -26,7 +25,6 @@ import (
 	"strconv"
 	"strings"
 
-	"github.com/rs/zerolog"
 	"go.mau.fi/whatsmeow/proto/waE2E"
 	"maunium.net/go/mautrix/bridgev2"
 	"maunium.net/go/mautrix/event"
@@ -74,36 +72,6 @@ func (mc *MessageConverter) cutOffLongMessage(content *event.MessageEventContent
 	}
 }
 
-func logOversizedPart(ctx context.Context, part *bridgev2.ConvertedMessagePart) {
-	if part == nil || part.Content == nil {
-		return
-	}
-	contentJSON, _ := json.Marshal(part.Content)
-	extraJSON, _ := json.Marshal(part.Extra)
-	if len(contentJSON)+len(extraJSON) <= 40000 {
-		return
-	}
-	extraKeys := make([]string, 0, len(part.Extra))
-	for key := range part.Extra {
-		extraKeys = append(extraKeys, key)
-	}
-	var mentionCount int
-	if part.Content.Mentions != nil {
-		mentionCount = len(part.Content.Mentions.UserIDs)
-	}
-	linkPreviewJSON, _ := json.Marshal(part.Content.BeeperLinkPreviews)
-	zerolog.Ctx(ctx).Warn().
-		Str("msg_type", string(part.Content.MsgType)).
-		Int("content_json_size", len(contentJSON)).
-		Int("extra_json_size", len(extraJSON)).
-		Int("body_size", len(part.Content.Body)).
-		Int("formatted_body_size", len(part.Content.FormattedBody)).
-		Int("mention_count", mentionCount).
-		Int("link_preview_json_size", len(linkPreviewJSON)).
-		Strs("extra_keys", extraKeys).
-		Msg("Converted message part is larger than the cut-off limit")
-}
-
 func (mc *MessageConverter) parseFormatting(content *event.MessageEventContent, allowInlineURL, forceHTML bool) {
 	parsedHTML := parseWAFormattingToHTML(content.Body, allowInlineURL)
 	if forceHTML || parsedHTML != event.TextToHTML(content.Body) {