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

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

pkg/connector/client.go

@@ -56,7 +56,6 @@ func (wa *WhatsAppConnector) LoadUserLogin(ctx context.Context, login *bridgev2.
 		pushNamesSynced:           exsync.NewEvent(),
 		appStateFullSyncAttempted: make(map[appstate.WAPatchName]time.Time),
 		previousOwnLIDs:           exsync.NewSetWithItems(login.Metadata.(*waid.UserLoginMetadata).PreviousOwnLIDs),
-		museRequests:              exsync.NewMap[string, chan museResponse](),
 
 		onDemandHistorySyncRequests: make(map[types.JID]onDemandHistorySyncRequest),
 
@@ -137,7 +136,6 @@ type WhatsAppClient struct {
 	wasaResyncLock         sync.Mutex
 	museProfileLock        sync.Mutex
 	lastMuseProfileRequest time.Time
-	museRequests           *exsync.Map[string, chan museResponse]
 
 	onDemandHistorySyncLock     sync.Mutex
 	onDemandHistorySyncRequests map[types.JID]onDemandHistorySyncRequest

pkg/connector/connector.go

@@ -111,7 +111,6 @@ func (wa *WhatsAppConnector) Init(bridge *bridgev2.Bridge) {
 	wa.MsgConv.DB = wa.DB
 	wa.Bridge.Commands.(*commands.Processor).AddHandlers(
 		cmdAccept, cmdSync, cmdResetAppState, cmdInviteLink, cmdResolveLink, cmdJoin,
-		cmdMuseApprovalAllow, cmdMuseApprovalDeny,
 	)
 	wa.mediaEditCache = make(MediaEditCache)
 	wa.unmigratedDMs = exsync.NewSet[networkid.PortalKey]()

pkg/connector/handlewhatsapp.go

@@ -353,7 +353,7 @@ func (wa *WhatsAppClient) handleWAMessage(ctx context.Context, evt *events.Messa
 	}
 	if protocol := evt.Message.GetProtocolMessage(); evt.Info.Chat == types.MuseJID && protocol.GetType() == waE2E.ProtocolMessage_AI_METADATA_OPERATION {
 		if sync := protocol.GetAiMetadataOperation().GetHatchMetadataSync(); sync != nil {
-			return wa.handleMuseMetadata(ctx, sync)
+			return wa.handleMuseMetadata(ctx, sync.GetData())
 		}
 		return
 	}

pkg/connector/muse.go

@@ -1,14 +1,11 @@
 package connector
 
 import (
-	"cmp"
 	"context"
 	"encoding/base64"
 	"encoding/json"
 	"fmt"
 	"runtime/debug"
-	"slices"
-	"strings"
 	"time"
 
 	"github.com/google/uuid"
@@ -21,11 +18,9 @@ import (
 	"go.mau.fi/whatsmeow/types"
 	"google.golang.org/protobuf/proto"
 	"maunium.net/go/mautrix/bridgev2"
-	"maunium.net/go/mautrix/bridgev2/commands"
 	"maunium.net/go/mautrix/bridgev2/database"
 	"maunium.net/go/mautrix/bridgev2/networkid"
 	"maunium.net/go/mautrix/bridgev2/simplevent"
-	"maunium.net/go/mautrix/event"
 
 	"go.mau.fi/mautrix-whatsapp/pkg/waid"
 )
@@ -136,74 +131,28 @@ func museAvatarReady(portal *database.Portal) bool {
 		(portal.AvatarMXC != "" && portal.AvatarHash == [32]byte(expectedHash))
 }
 
-type museApprovalRecord struct {
+func (wa *WhatsAppClient) handleMuseMetadata(ctx context.Context, data []byte) bool {
-	ApprovalID         string `json:"approval_id"`
-	Decision           string `json:"decision"`
-	Status             string `json:"status"`
-	RequestExpiresAtMS int64  `json:"request_expires_at_ms"`
-	DecisionOptions    []struct {
-		Kind      string `json:"kind"`
-		LabelText string `json:"label_text"`
-	} `json:"decision_options"`
-	Display struct {
-		PermissionQuestion struct {
-			Text string `json:"text"`
-		} `json:"permission_question"`
-		PurposeSummary string `json:"purpose_summary"`
-		SummaryTitle   string `json:"summary_title"`
-	} `json:"display"`
-}
-
-func (wa *WhatsAppClient) handleMuseMetadata(ctx context.Context, sync *waAICommon.HatchMetadataSync) bool {
-	data := sync.GetData()
 	var msg struct {
 		Type    string `json:"type"`
-		Status  string `json:"status"`
 		Payload struct {
-			Body      json.RawMessage `json:"body"`
+			Event   string `json:"event"`
-			Event     string          `json:"event"`
+			Payload struct {
-			Timestamp int64           `json:"ts_ms"`
+				Name   string `json:"name"`
-			Payload   json.RawMessage `json:"payload"`
+				Avatar struct {
+					Image museImage `json:"secure_image"`
+				} `json:"avatar"`
+			} `json:"payload"`
 		} `json:"payload"`
 	}
 	if err := json.Unmarshal(data, &msg); err != nil {
 		zerolog.Ctx(ctx).Warn().Err(err).Msg("Failed to decode Muse metadata")
 		return true
-	} else if msg.Type == "res" {
+	}
-		if ch, ok := wa.museRequests.Pop(sync.GetRequestID()); ok {
+	if msg.Type != "event" || msg.Payload.Event != "identity.updated" || msg.Payload.Payload.Name == "" {
-			ch <- museResponse{Status: msg.Status, Body: msg.Payload.Body}
-		}
-		return true
-	} else if msg.Type != "event" {
 		return true
 	}
-	switch msg.Payload.Event {
+	name := msg.Payload.Payload.Name
-	case "identity.updated":
+	avatar := wa.museAvatar(msg.Payload.Payload.Avatar.Image)
-		return wa.handleMuseIdentity(ctx, msg.Payload.Payload)
-	case "hitl.approval_record":
-		return wa.handleMuseApproval(ctx, msg.Payload.Payload, msg.Payload.Timestamp)
-	default:
-		return true
-	}
-}
-
-func (wa *WhatsAppClient) handleMuseIdentity(ctx context.Context, data json.RawMessage) bool {
-	var identity struct {
-		Name   string `json:"name"`
-		Avatar struct {
-			Image museImage `json:"secure_image"`
-		} `json:"avatar"`
-	}
-	if len(data) == 0 {
-		return true
-	} else if err := json.Unmarshal(data, &identity); err != nil {
-		zerolog.Ctx(ctx).Warn().Err(err).Msg("Failed to decode Muse identity")
-		return true
-	} else if identity.Name == "" {
-		return true
-	}
-	name := identity.Name
-	avatar := wa.museAvatar(identity.Avatar.Image)
 	return wa.UserLogin.QueueRemoteEvent(&simplevent.ChatInfoChange{
 		EventMeta: simplevent.EventMeta{
 			Type:         bridgev2.RemoteEventChatInfoChange,
@@ -225,278 +174,6 @@ func (wa *WhatsAppClient) handleMuseIdentity(ctx context.Context, data json.RawM
 	}).Success
 }
 
-func (wa *WhatsAppClient) handleMuseApproval(ctx context.Context, data json.RawMessage, tsMS int64) bool {
-	var wrapper struct {
-		Lifecycle      string              `json:"lifecycle"`
-		Record         *museApprovalRecord `json:"record"`
-		ApprovalRecord *museApprovalRecord `json:"approval_record"`
-		Approval       *museApprovalRecord `json:"approval"`
-		museApprovalRecord
-	}
-	if err := json.Unmarshal(data, &wrapper); err != nil {
-		zerolog.Ctx(ctx).Warn().Err(err).Msg("Failed to decode Muse approval record")
-		return true
-	}
-	record := &wrapper.museApprovalRecord
-	if wrapper.Record != nil {
-		record = wrapper.Record
-	} else if wrapper.ApprovalRecord != nil {
-		record = wrapper.ApprovalRecord
-	} else if wrapper.Approval != nil {
-		record = wrapper.Approval
-	}
-	approvalID := strings.TrimSpace(record.ApprovalID)
-	if approvalID == "" {
-		return true
-	}
-	status := strings.TrimSpace(wrapper.Lifecycle)
-	if status == "" {
-		status = strings.TrimSpace(record.Status)
-	}
-	resolution := strings.TrimSpace(record.Decision)
-	if resolution == "" && status != "" && status != "pending" {
-		resolution = status
-	}
-	log := zerolog.Ctx(ctx).With().Str("approval_id", approvalID).Str("resolution", resolution).Logger()
-	ts := time.Now()
-	if tsMS > 0 {
-		ts = time.UnixMilli(tsMS)
-	}
-	meta := &waid.MuseApprovalMeta{
-		ID:       approvalID,
-		Question: museApprovalQuestion(record),
-		Summary:  strings.TrimSpace(record.Display.PurposeSummary),
-	}
-	for _, opt := range record.DecisionOptions {
-		if kind := strings.TrimSpace(opt.Kind); kind != "" {
-			meta.Options = append(meta.Options, kind)
-		}
-	}
-	evt := &simplevent.Message[*waid.MuseApprovalMeta]{
-		EventMeta: simplevent.EventMeta{
-			Type:      bridgev2.RemoteEventMessage,
-			PortalKey: wa.makeWAPortalKey(types.MuseJID),
-			Sender:    wa.makeEventSender(ctx, types.MuseJID),
-			Timestamp: ts,
-		},
-		Data:               meta,
-		ID:                 waid.MakeFakeMessageID(types.MuseJID, types.MuseJID, "approval-"+approvalID),
-		ConvertMessageFunc: convertMuseApproval,
-	}
-	if resolution != "" {
-		evt.Type = bridgev2.RemoteEventEdit
-		evt.TargetMessage = evt.ID
-		evt.ConvertEditFunc = func(ctx context.Context, portal *bridgev2.Portal, intent bridgev2.MatrixAPI, existing []*database.Message, data *waid.MuseApprovalMeta) (*bridgev2.ConvertedEdit, error) {
-			msgMeta := existing[0].Metadata.(*waid.MessageMetadata)
-			if prev := msgMeta.MuseApproval; prev != nil {
-				if data.Question == "" {
-					data.Question = prev.Question
-				}
-				if data.Summary == "" {
-					data.Summary = prev.Summary
-				}
-			}
-			data.Options = nil
-			data.Resolved = true
-			msgMeta.MuseApproval = data
-			return &bridgev2.ConvertedEdit{
-				ModifiedParts: []*bridgev2.ConvertedEditPart{{
-					Part:    existing[0],
-					Type:    event.EventMessage,
-					Content: museApprovalContent(portal, data, museApprovalResolutionText(resolution)),
-				}},
-			}, nil
-		}
-	} else if len(meta.Options) == 0 {
-		log.Debug().Msg("Ignoring Muse approval record without decision options")
-		return true
-	} else if record.RequestExpiresAtMS > 0 && time.UnixMilli(record.RequestExpiresAtMS).Before(time.Now()) {
-		log.Debug().Msg("Ignoring expired Muse approval record")
-		return true
-	} else {
-		evt.CreatePortal = true
-	}
-	log.Debug().Msg("Received Muse approval record")
-	return wa.UserLogin.QueueRemoteEvent(evt).Success
-}
-
-func museApprovalResolutionText(resolution string) string {
-	switch strings.ToLower(resolution) {
-	case "allow", "allowed", "allow_once", "allow_session", "allow_today", "allow_always", "approve", "approved":
-		return "Allowed."
-	case "deny", "denied", "deny_always", "reject", "rejected", "decline", "declined":
-		return "Denied."
-	case "expired", "timeout", "timed_out":
-		return "This request expired."
-	default:
-		return "No longer waiting for approval."
-	}
-}
-
-func museApprovalQuestion(record *museApprovalRecord) string {
-	if question := strings.TrimSpace(record.Display.PermissionQuestion.Text); question != "" {
-		return question
-	}
-	return strings.TrimSpace(record.Display.SummaryTitle)
-}
-
-func museApprovalContent(portal *bridgev2.Portal, meta *waid.MuseApprovalMeta, footer string) *event.MessageEventContent {
-	lines := make([]string, 0, 3)
-	if meta.Question != "" {
-		assistant := portal.Name
-		if assistant == "" {
-			assistant = "Muse"
-		}
-		lines = append(lines, strings.ReplaceAll(meta.Question, "{assistant}", assistant))
-	} else {
-		lines = append(lines, "Muse is asking for your approval.")
-	}
-	if meta.Summary != "" {
-		lines = append(lines, meta.Summary)
-	}
-	lines = append(lines, footer)
-	return &event.MessageEventContent{
-		MsgType: event.MsgNotice,
-		Body:    strings.Join(lines, "\n\n"),
-	}
-}
-
-func convertMuseApproval(ctx context.Context, portal *bridgev2.Portal, intent bridgev2.MatrixAPI, meta *waid.MuseApprovalMeta) (*bridgev2.ConvertedMessage, error) {
-	prefix := portal.Bridge.Config.CommandPrefix
-	footer := fmt.Sprintf("Reply to this message with %s approval-allow or %s approval-deny, or use the WhatsApp app.", prefix, prefix)
-	return &bridgev2.ConvertedMessage{
-		Parts: []*bridgev2.ConvertedMessagePart{{
-			Type:       event.EventMessage,
-			Content:    museApprovalContent(portal, meta, footer),
-			DBMetadata: &waid.MessageMetadata{MuseApproval: meta},
-		}},
-	}, nil
-}
-
-type museResponse struct {
-	Status string
-	Body   json.RawMessage
-}
-
-func (wa *WhatsAppClient) decideMuseApproval(ctx context.Context, approvalID, decision string) error {
-	payload, err := json.Marshal(map[string]any{
-		"version": 1,
-		"type":    "req",
-		"payload": map[string]any{
-			"method": "hitl.approval.decide",
-			"params": map[string]any{
-				"approval_id": approvalID,
-				"decision":    decision,
-			},
-		},
-	})
-	if err != nil {
-		return err
-	}
-	requestID := uuid.NewString()
-	ch := make(chan museResponse, 1)
-	wa.museRequests.Set(requestID, ch)
-	defer wa.museRequests.Delete(requestID)
-	_, err = wa.Client.SendMessage(ctx, types.MuseJID, &waE2E.Message{
-		ProtocolMessage: &waE2E.ProtocolMessage{
-			Type: waE2E.ProtocolMessage_AI_METADATA_OPERATION.Enum(),
-			AiMetadataOperation: &waAICommon.AIMetadataOperation{
-				HatchMetadataSync: &waAICommon.HatchMetadataSync{
-					Data:        payload,
-					TimestampMS: proto.Int64(time.Now().UnixMilli()),
-					RequestID:   proto.String(requestID),
-				},
-			},
-		},
-	})
-	if err != nil {
-		return fmt.Errorf("failed to send decision: %w", err)
-	}
-	select {
-	case resp := <-ch:
-		if resp.Status != "ok" {
-			var body struct {
-				ErrorMessage string `json:"error_message"`
-				ErrorCode    string `json:"error_code"`
-			}
-			_ = json.Unmarshal(resp.Body, &body)
-			if body.ErrorMessage != "" {
-				return fmt.Errorf("muse rejected the decision: %s", body.ErrorMessage)
-			}
-			return fmt.Errorf("muse rejected the decision (%s)", cmp.Or(body.ErrorCode, resp.Status))
-		}
-		return nil
-	case <-time.After(30 * time.Second):
-		return fmt.Errorf("muse didn't confirm the decision")
-	case <-ctx.Done():
-		return ctx.Err()
-	}
-}
-
-var cmdMuseApprovalAllow = &commands.FullHandler{
-	Func: fnMuseApproval,
-	Name: "approval-allow",
-	Help: commands.HelpMeta{
-		Section:     commands.HelpSectionChats,
-		Description: "Allow a pending Muse request. This can only be used in reply to a Muse approval message.",
-	},
-	RequiresLogin:  true,
-	RequiresPortal: true,
-}
-
-var cmdMuseApprovalDeny = &commands.FullHandler{
-	Func: fnMuseApproval,
-	Name: "approval-deny",
-	Help: commands.HelpMeta{
-		Section:     commands.HelpSectionChats,
-		Description: "Deny a pending Muse request. This can only be used in reply to a Muse approval message.",
-	},
-	RequiresLogin:  true,
-	RequiresPortal: true,
-}
-
-func museDecisionKind(options []string, allow bool) string {
-	if !allow {
-		if slices.Contains(options, "deny") {
-			return "deny"
-		}
-		return ""
-	}
-	for _, kind := range []string{"allow_once", "allow"} {
-		if slices.Contains(options, kind) {
-			return kind
-		}
-	}
-	return ""
-}
-
-func fnMuseApproval(ce *commands.Event) {
-	allow := ce.Command == "approval-allow"
-	if len(ce.ReplyTo) == 0 {
-		ce.Reply("Reply to a Muse approval message when using this command.")
-	} else if message, err := ce.Bridge.DB.Message.GetPartByMXID(ce.Ctx, ce.ReplyTo); err != nil {
-		ce.Log.Err(err).Stringer("reply_to_mxid", ce.ReplyTo).Msg("Failed to get reply target event for Muse approval command")
-		ce.Reply("Failed to get reply event")
-	} else if message == nil || message.Room != ce.Portal.PortalKey {
-		ce.Reply("Reply event not found")
-	} else if meta := message.Metadata.(*waid.MessageMetadata).MuseApproval; meta == nil {
-		ce.Reply("That doesn't look like a Muse approval message.")
-	} else if meta.Resolved {
-		ce.Reply("This request has already been decided.")
-	} else if kind := museDecisionKind(meta.Options, allow); kind == "" {
-		ce.Reply("Muse doesn't offer that choice for this request. Use the WhatsApp app instead.")
-	} else if login := ce.Bridge.GetCachedUserLoginByID(ce.Portal.Receiver); login == nil || login.UserMXID != ce.User.MXID {
-		ce.Reply("Login not found")
-	} else if !login.Client.IsLoggedIn() {
-		ce.Reply("Not logged in")
-	} else if err = login.Client.(*WhatsAppClient).decideMuseApproval(ce.Ctx, meta.ID, kind); err != nil {
-		ce.Log.Err(err).Str("approval_id", meta.ID).Str("decision", kind).Msg("Failed to send Muse approval decision")
-		ce.Reply("Couldn't send the decision: %v. Use the WhatsApp app instead.", err)
-	} else {
-		ce.React("✅")
-	}
-}
-
 func (wa *WhatsAppClient) museAvatar(image museImage) *bridgev2.Avatar {
 	if image.DirectPath == "" || image.Length <= 0 || image.Length > 5<<20 {
 		return nil

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 {
+	} else if cli := getClient(ctx); user.Server == types.HiddenUserServer && cli != nil {
-		cli := getClient(ctx)
 		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) {

pkg/waid/dbmeta.go

@@ -82,14 +82,6 @@ const (
 	MsgErrMediaNotFound    MessageErrorType = "media_not_found"
 )
 
-type MuseApprovalMeta struct {
-	ID       string   `json:"id"`
-	Question string   `json:"question,omitempty"`
-	Summary  string   `json:"summary,omitempty"`
-	Options  []string `json:"options,omitempty"`
-	Resolved bool     `json:"resolved,omitzero"`
-}
-
 type GroupInviteMeta struct {
 	JID        types.JID `json:"jid"`
 	Code       string    `json:"code"`
@@ -105,7 +97,6 @@ type MessageMetadata struct {
 	Error            MessageErrorType  `json:"error,omitempty"`
 	BroadcastListJID *types.JID        `json:"broadcast_list_jid,omitempty"`
 	GroupInvite      *GroupInviteMeta  `json:"group_invite,omitempty"`
-	MuseApproval     *MuseApprovalMeta `json:"muse_approval,omitempty"`
 	FailedMediaMeta  json.RawMessage   `json:"media_meta,omitempty"`
 	DirectMediaMeta  json.RawMessage   `json:"direct_media_meta,omitempty"`
 	IsMatrixPoll     bool              `json:"is_matrix_poll,omitzero"`
@@ -128,9 +119,6 @@ func (mm *MessageMetadata) CopyFrom(other any) {
 	if otherMM.GroupInvite != nil {
 		mm.GroupInvite = otherMM.GroupInvite
 	}
-	if otherMM.MuseApproval != nil {
-		mm.MuseApproval = otherMM.MuseApproval
-	}
 	mm.IsMatrixPoll = mm.IsMatrixPoll || otherMM.IsMatrixPoll
 }