connector/backfill: implement reaction backfill

Signed-off-by: Sumner Evans <sumner.evans@automattic.com>

pkg/connector/backfill.go

@@ -84,7 +84,20 @@ func (l *LinkedInClient) FetchMessages(ctx context.Context, fetchParams bridgev2
 			Timestamp:        msg.DeliveredAt.Time,
 		}
 
-		// TODO reactions
+		for _, rs := range msg.ReactionSummaries {
+			reactors, err := l.client.GetReactors(ctx, msg.EntityURN, rs.Emoji)
+			if err != nil {
+				log.Err(err).Msg("failed to get reactors")
+				continue
+			}
+			for _, reactor := range reactors.Elements {
+				backfillMessage.Reactions = append(backfillMessage.Reactions, &bridgev2.BackfillReaction{
+					Sender:  l.makeSender(reactor),
+					EmojiID: networkid.EmojiID(rs.Emoji),
+					Emoji:   rs.Emoji,
+				})
+			}
+		}
 
 		resp.Messages = append(resp.Messages, &backfillMessage)
 	}

pkg/connector/client.go

@@ -212,9 +212,10 @@ func (l *LinkedInClient) onRealtimeMessage(ctx context.Context, msg linkedingo.M
 		Timestamp:    msg.DeliveredAt.Time,
 	}
 
+	chatInfo, _ := l.conversationToChatInfo(msg.Conversation)
 	l.main.Bridge.QueueRemoteEvent(l.userLogin, &simplevent.ChatResync{
 		EventMeta:       meta.WithType(bridgev2.RemoteEventChatResync),
-		ChatInfo:        ptr.Ptr(l.conversationToChatInfo(msg.Conversation)),
+		ChatInfo:        &chatInfo,
 		LatestMessageTS: msg.DeliveredAt.Time,
 	})
 
@@ -354,7 +355,7 @@ func (l *LinkedInClient) getMessagingParticipantUserInfo(participant linkedingo.
 	return
 }
 
-func (l *LinkedInClient) conversationToChatInfo(conv linkedingo.Conversation) (ci bridgev2.ChatInfo) {
+func (l *LinkedInClient) conversationToChatInfo(conv linkedingo.Conversation) (ci bridgev2.ChatInfo, userInChat bool) {
 	if conv.Title != "" {
 		ci.Name = &conv.Title
 	}
@@ -374,6 +375,7 @@ func (l *LinkedInClient) conversationToChatInfo(conv linkedingo.Conversation) (c
 		MemberMap:        map[networkid.UserID]bridgev2.ChatMember{},
 	}
 	for _, participant := range conv.ConversationParticipants {
+		userInChat = userInChat || networkid.UserID(participant.EntityURN.ID()) == l.userID
 		sender := l.makeSender(participant)
 		ci.Members.MemberMap[sender.Sender] = bridgev2.ChatMember{
 			EventSender: sender,

pkg/connector/sync.go

@@ -5,7 +5,6 @@ import (
 	"time"
 
 	"github.com/rs/zerolog"
-	"go.mau.fi/util/ptr"
 	"maunium.net/go/mautrix/bridgev2"
 	"maunium.net/go/mautrix/bridgev2/simplevent"
 )
@@ -41,6 +40,11 @@ func (l *LinkedInClient) syncConversations(ctx context.Context) {
 		}
 
 		for _, conv := range conversations.Elements {
+			log := log.With().
+				Stringer("conversation_urn", conv.EntityURN).
+				Time("last_activity_at", conv.LastActivityAt.Time).
+				Logger()
+
 			if conv.LastActivityAt.Before(updatedBefore) {
 				updatedBefore = conv.LastActivityAt.Time
 			}
@@ -71,8 +75,13 @@ func (l *LinkedInClient) syncConversations(ctx context.Context) {
 					latestMessageTS = msg.DeliveredAt.Time
 				}
 			}
+			chatInfo, userInChat := l.conversationToChatInfo(conv)
+			if !userInChat {
+				log.Debug().Msg("User not in chat")
+				continue
+			}
 			l.main.Bridge.QueueRemoteEvent(l.userLogin, &simplevent.ChatResync{
-				ChatInfo:        ptr.Ptr(l.conversationToChatInfo(conv)),
+				ChatInfo:        &chatInfo,
 				EventMeta:       meta.WithType(bridgev2.RemoteEventChatResync),
 				LatestMessageTS: latestMessageTS,
 			})

pkg/linkedingo/conversations.go

@@ -17,9 +17,10 @@ type GraphQlResponse struct {
 }
 
 type GraphQLData struct {
-	MessengerConversationsByCategoryQuery *CollectionResponse[ConversationCursorMetadata, Conversation] `json:"messengerConversationsByCategoryQuery,omitempty"`
-	MessengerMessagesByAnchorTimestamp    *CollectionResponse[MessageMetadata, Message]                 `json:"messengerMessagesByAnchorTimestamp,omitempty"`
-	MessengerMessagesByConversation       *CollectionResponse[MessageMetadata, Message]                 `json:"messengerMessagesByConversation,omitempty"`
+	MessengerConversationsByCategoryQuery           *CollectionResponse[ConversationCursorMetadata, Conversation] `json:"messengerConversationsByCategoryQuery,omitempty"`
+	MessengerMessagesByAnchorTimestamp              *CollectionResponse[MessageMetadata, Message]                 `json:"messengerMessagesByAnchorTimestamp,omitempty"`
+	MessengerMessagesByConversation                 *CollectionResponse[MessageMetadata, Message]                 `json:"messengerMessagesByConversation,omitempty"`
+	MessengerMessagingParticipantsByMessageAndEmoji *CollectionResponse[any, MessagingParticipant]                `json:"messengerMessagingParticipantsByMessageAndEmoji,omitempty"`
 }
 
 // CollectionResponse represents a
@@ -93,9 +94,6 @@ func (c *Client) GetConversationsUpdatedBefore(ctx context.Context, updatedBefor
 			"count":             "20",
 			"query":             "(predicateUnions:List((conversationCategoryPredicate:(category:PRIMARY_INBOX))))",
 		}).
-		WithCSRF().
-		WithXLIHeaders().
-		WithHeader("accept", contentTypeGraphQL).
 		Do(ctx)
 	if err != nil {
 		return nil, err

pkg/linkedingo/messages.go

@@ -83,6 +83,7 @@ type Message struct {
 	BackendConversationURN  URN                     `json:"backendConversationUrn,omitempty"`
 	Conversation            Conversation            `json:"conversation,omitempty"`
 	RenderContent           []RenderContent         `json:"renderContent,omitempty"`
+	ReactionSummaries       []ReactionSummary       `json:"reactionSummaries,omitempty"`
 }
 
 func (m Message) MessageID() networkid.MessageID {
@@ -201,9 +202,6 @@ func (c *Client) GetMessagesBefore(ctx context.Context, conversationURN URN, bef
 			"countBefore":     strconv.Itoa(count),
 			"countAfter":      "0",
 		}).
-		WithCSRF().
-		WithXLIHeaders().
-		WithHeader("accept", contentTypeGraphQL).
 		Do(ctx)
 	if err != nil {
 		return nil, err
@@ -223,9 +221,6 @@ func (c *Client) GetMessagesWithPrevCursor(ctx context.Context, conversationURN
 			"count":           strconv.Itoa(count),
 			"prevCursor":      url.QueryEscape(prevCursor),
 		}).
-		WithCSRF().
-		WithXLIHeaders().
-		WithHeader("accept", contentTypeGraphQL).
 		Do(ctx)
 	if err != nil {
 		return nil, err

pkg/linkedingo/reactions.go

@@ -2,8 +2,10 @@ package linkedingo
 
 import (
 	"context"
+	"encoding/json"
 	"fmt"
 	"net/http"
+	"net/url"
 )
 
 type DecoratedReactionSummary struct {
@@ -55,3 +57,20 @@ func (c *Client) doReactAction(ctx context.Context, messageURN URN, emoji, actio
 	}
 	return nil
 }
+
+func (c *Client) GetReactors(ctx context.Context, messageURN URN, emoji string) (*CollectionResponse[any, MessagingParticipant], error) {
+	resp, err := c.newAuthedRequest(http.MethodGet, linkedInVoyagerMessagingGraphQLURL).
+		WithGraphQLQuery("messengerMessagingParticipants.6bedbcf9406fa19045dc627ffc51f286", map[string]string{
+			"messageUrn": url.QueryEscape(messageURN.String()),
+			"emoji":      url.QueryEscape(emoji),
+		}).
+		Do(ctx)
+	if err != nil {
+		return nil, err
+	} else if resp.StatusCode != http.StatusOK {
+		return nil, fmt.Errorf("failed to get reactors for message %s with emoji %s (statusCode=%d)", messageURN, emoji, resp.StatusCode)
+	}
+
+	var response GraphQlResponse
+	return response.Data.MessengerMessagingParticipantsByMessageAndEmoji, json.NewDecoder(resp.Body).Decode(&response)
+}

pkg/linkedingo/request.go

@@ -70,6 +70,9 @@ func (a *authedRequest) WithRawQuery(raw string) *authedRequest {
 
 func (a *authedRequest) WithGraphQLQuery(queryID string, variables map[string]string) *authedRequest {
 	a.WithHeader("accept", contentTypeGraphQL)
+	a.WithCSRF()
+	a.WithXLIHeaders()
+
 	var queryStr strings.Builder
 	queryStr.WriteString("queryId=")
 	queryStr.WriteString(queryID)