msgconv: log size breakdown of converted parts over the cut-off limit (#1047)

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()).
@@ -566,15 +572,6 @@ 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
@@ -583,11 +580,7 @@ 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) {
-		if startTime != nil {
-			wa.deleteHistorySyncMessages(ctx, portalJID, 1<<62, uint64(startTime.Unix()))
-		} else {
-			wa.deleteHistorySyncMessages(ctx, portalJID, lastMessageTS, 0)
-		}
+		wa.deleteHistorySyncMessages(ctx, portalJID, 0, 0)
 		hasMore = hasMore && params.AnchorMessage != nil
 		if hasMore && !params.AllowSlowFetch {
 			return &bridgev2.FetchMessagesResponse{
@@ -615,14 +608,7 @@ func (wa *WhatsAppClient) FetchMessages(ctx context.Context, params bridgev2.Fet
 			}
 		}
 	}
-	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)
+	resp, err := wa.convertHistorySyncMessages(ctx, params.Portal, portalJID, messages, true, !params.Forward && !hasMore)
 	if err != nil {
 		return nil, fmt.Errorf("failed to convert messages: %w", err)
 	}
@@ -684,7 +670,7 @@ func (wa *WhatsAppClient) convertHistorySyncMessages(
 	portalJID types.JID,
 	messages []*waWeb.WebMessageInfo,
 	explodeOnError bool,
-	deleteUpTo uint64,
+	deleteEverything bool,
 ) (*bridgev2.FetchMessagesResponse, error) {
 	cli := wa.Client
 	if cli == nil {
@@ -743,7 +729,11 @@ func (wa *WhatsAppClient) convertHistorySyncMessages(
 		Messages: convertedMessages,
 		Cursor:   networkid.PaginationCursor(strconv.FormatUint(oldestTS, 10)),
 		CompleteCallback: func() {
-			wa.deleteHistorySyncMessages(ctx, portalJID, max(newestTS, deleteUpTo), oldestTS)
+			if deleteEverything {
+				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 {
@@ -906,7 +896,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, 0)
+				resp, err := wa.convertHistorySyncMessages(ctx, portal, portalJID, messages, false, false)
 				if err != nil {
 					return nil, err
 				}

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

pkg/msgconv/from-whatsapp.go

@@ -322,6 +322,7 @@ 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,6 +18,7 @@ package msgconv
 
 import (
 	"context"
+	"encoding/json"
 	"fmt"
 	"html"
 	"regexp"
@@ -25,6 +26,7 @@ import (
 	"strconv"
 	"strings"
 
+	"github.com/rs/zerolog"
 	"go.mau.fi/whatsmeow/proto/waE2E"
 	"maunium.net/go/mautrix/bridgev2"
 	"maunium.net/go/mautrix/event"
@@ -72,6 +74,36 @@ 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) {