backfill: don't requeue history sync conversations that are still being created (#1035)

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

pkg/connector/backfill.go

@@ -293,7 +293,10 @@ 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()).
@@ -402,6 +405,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 +450,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 +493,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 +509,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 +517,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 +525,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 {

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

pkg/connector/handlematrix.go

@@ -273,13 +273,10 @@ func (wa *WhatsAppClient) HandleMatrixEdit(ctx context.Context, edit *bridgev2.M
 		return fmt.Errorf("failed to parse portal ID: %w", err)
 	}
 
-	waMsg := wa.Main.MsgConv.CaptionEditToWhatsApp(ctx, wa.Client, edit.Event, edit.Content, edit.Portal)
-	if waMsg == nil {
-		waMsg, _, err = wa.Main.MsgConv.ToWhatsApp(ctx, wa.Client, edit.Event, edit.Content, nil, nil, edit.Portal)
+	waMsg, _, err := wa.Main.MsgConv.ToWhatsApp(ctx, wa.Client, edit.Event, edit.Content, nil, nil, edit.Portal)
 	if err != nil {
 		return wrapConversionError(fmt.Errorf("failed to convert message: %w", err))
 	}
-	}
 	defer wa.mcTrack(edit, time.Now(), &retErr)
 
 	convertedEdit := wa.Client.BuildEdit(messageID.Chat, messageID.ID, waMsg)

pkg/connector/wadb/upgrades/00-latest-schema.sql

@@ -1,4 +1,4 @@
--- v0 -> v11 (compatible with v3+): Latest revision
+-- v0 -> v12 (compatible with v3+): Latest revision
 
 CREATE TABLE whatsapp_poll_option_id (
     bridge_id TEXT  NOT NULL,
@@ -50,6 +50,8 @@ CREATE TABLE whatsapp_history_sync_message (
         REFERENCES whatsapp_history_sync_conversation (bridge_id, user_login_id, chat_jid) ON UPDATE CASCADE ON DELETE CASCADE
 );
 
+CREATE INDEX whatsapp_history_sync_message_chat_ts_idx ON whatsapp_history_sync_message (bridge_id, user_login_id, chat_jid, timestamp);
+
 CREATE TABLE whatsapp_media_backfill_request (
     bridge_id       TEXT    NOT NULL,
     user_login_id   TEXT    NOT NULL,

pkg/connector/wadb/upgrades/12-history-sync-message-timestamp-index.sql

@@ -0,0 +1,2 @@
+-- v12 (compatible with v3+): Add index for history sync message timestamp lookups
+CREATE INDEX whatsapp_history_sync_message_chat_ts_idx ON whatsapp_history_sync_message (bridge_id, user_login_id, chat_jid, timestamp);

pkg/msgconv/from-matrix.go

@@ -204,46 +204,6 @@ func (mc *MessageConverter) ToWhatsApp(
 	return message, extra, nil
 }
 
-func (mc *MessageConverter) CaptionEditToWhatsApp(
-	ctx context.Context,
-	client *whatsmeow.Client,
-	evt *event.Event,
-	content *event.MessageEventContent,
-	portal *bridgev2.Portal,
-) *waE2E.Message {
-	mime := content.GetInfo().MimeType
-	if mime == "" {
-		return nil
-	} else if mime == "image/gif" {
-		content.MsgType = event.MsgVideo
-	}
-	ctx = context.WithValue(ctx, contextKeyClient, client)
-	ctx = context.WithValue(ctx, contextKeyPortal, portal)
-	contextInfo := mc.generateContextInfo(ctx, nil, content.RelatesTo, portal, content.BeeperDisappearingTimer, content.Mentions != nil && content.Mentions.Room)
-	msg := mc.constructMediaMessage(ctx, content, evt, &whatsmeow.UploadResponse{}, nil, contextInfo, mime)
-	switch {
-	case msg.GetImageMessage() != nil:
-		return &waE2E.Message{ImageMessage: &waE2E.ImageMessage{
-			Caption:     msg.ImageMessage.Caption,
-			ContextInfo: contextInfo,
-		}}
-	case msg.GetVideoMessage() != nil:
-		return &waE2E.Message{VideoMessage: &waE2E.VideoMessage{
-			Caption:     msg.VideoMessage.Caption,
-			ContextInfo: contextInfo,
-		}}
-	case msg.GetDocumentMessage() != nil, msg.GetDocumentWithCaptionMessage() != nil:
-		return &waE2E.Message{DocumentWithCaptionMessage: &waE2E.FutureProofMessage{
-			Message: &waE2E.Message{DocumentMessage: &waE2E.DocumentMessage{
-				Caption:     proto.String(msg.GetDocumentWithCaptionMessage().GetMessage().GetDocumentMessage().GetCaption()),
-				ContextInfo: contextInfo,
-			}},
-		}}
-	default:
-		return nil
-	}
-}
-
 func (mc *MessageConverter) constructMediaMessage(
 	ctx context.Context,
 	content *event.MessageEventContent,

pkg/msgconv/from-whatsapp.go

@@ -76,8 +76,7 @@ func (mc *MessageConverter) getBasicUserInfo(ctx context.Context, user types.JID
 	var pnJID types.JID
 	if user.Server == types.DefaultUserServer {
 		pnJID = user
-	} else if user.Server == types.HiddenUserServer {
-		cli := getClient(ctx)
+	} else if cli := getClient(ctx); user.Server == types.HiddenUserServer && cli != nil {
 		if user.User == cli.Store.GetLID().User {
 			pnJID = cli.Store.GetJID()
 		} else {
@@ -322,6 +321,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) {