wadb: index history sync messages by chat and timestamp (#1051)

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

pkg/connector/chatinfo.go

@@ -4,7 +4,6 @@ import (
 	"context"
 	"errors"
 	"fmt"
-	"maps"
 	"time"
 
 	"github.com/rs/zerolog"
@@ -245,13 +244,6 @@ const (
 	defaultPL    = 0
 )
 
-var alwaysAllowedGroupEvents = map[event.Type]int{
-	event.EventEncrypted:            defaultPL, // Need to always allow encrypted events for poll responses
-	event.EventReaction:             defaultPL,
-	event.EventRedaction:            defaultPL,
-	event.EventUnstablePollResponse: defaultPL,
-}
-
 func setDefaultSubGroupFlag(isCommunityAnnouncementGroup bool) bridgev2.ExtraUpdater[*bridgev2.Portal] {
 	return func(_ context.Context, portal *bridgev2.Portal) bool {
 		meta := portal.Metadata.(*waid.PortalMetadata)
@@ -319,6 +311,10 @@ func (wa *WhatsAppClient) wrapGroupInfo(ctx context.Context, info *types.GroupIn
 					event.StateRoomName:                metaChangePL,
 					event.StateRoomAvatar:              metaChangePL,
 					event.StateTopic:                   metaChangePL,
+					event.EventEncrypted:               defaultPL, // Need to always allow encrypted events for poll responses
+					event.EventReaction:                defaultPL,
+					event.EventRedaction:               defaultPL,
+					event.EventUnstablePollResponse:    defaultPL,
 					event.StateBeeperDisappearingTimer: metaChangePL,
 				},
 			},
@@ -330,7 +326,6 @@ func (wa *WhatsAppClient) wrapGroupInfo(ctx context.Context, info *types.GroupIn
 		},
 		ExtraUpdates: extraUpdater,
 	}
-	maps.Copy(wrapped.Members.PowerLevels.Events, alwaysAllowedGroupEvents)
 	var hasSelf bool
 	for _, pcp := range info.Participants {
 		member := bridgev2.ChatMember{
@@ -441,25 +436,24 @@ func (wa *WhatsAppClient) wrapGroupInfoChange(ctx context.Context, evt *events.G
 		if memberChanges == nil {
 			memberChanges = &bridgev2.ChatMemberList{}
 		}
-		memberChanges.PowerLevels = &bridgev2.PowerLevelOverrides{
-			Events: make(map[event.Type]int),
-		}
+		memberChanges.PowerLevels = &bridgev2.PowerLevelOverrides{}
 		if evt.Announce != nil {
 			if evt.Announce.IsAnnounce {
 				memberChanges.PowerLevels.EventsDefault = ptr.Ptr(adminPL)
 			} else {
 				memberChanges.PowerLevels.EventsDefault = ptr.Ptr(defaultPL)
 			}
-			maps.Copy(memberChanges.PowerLevels.Events, alwaysAllowedGroupEvents)
 		}
 		if evt.Locked != nil {
 			metaChangePL := defaultPL
 			if evt.Locked.IsLocked {
 				metaChangePL = adminPL
 			}
-			memberChanges.PowerLevels.Events[event.StateRoomName] = metaChangePL
-			memberChanges.PowerLevels.Events[event.StateRoomAvatar] = metaChangePL
-			memberChanges.PowerLevels.Events[event.StateTopic] = metaChangePL
+			memberChanges.PowerLevels.Events = map[event.Type]int{
+				event.StateRoomName:   metaChangePL,
+				event.StateRoomAvatar: metaChangePL,
+				event.StateTopic:      metaChangePL,
+			}
 		}
 	}
 	return &bridgev2.ChatInfoChange{

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