feat: Sync Mentions and Replies AC notifications and messages (#4337)

* feat: Marking Mentions and Replies AC notifications as read also marks corresponding message as seen

* feat: Marking message as seen marks as read corresponding notification (if there is so)

* chore: make messenger activity center test less flaky

* Update VERSION

VERSION

@@ -1 +1 @@
-0.171.19
+0.171.20

protocol/activity_center_persistence.go

@@ -605,26 +605,40 @@ func (db sqlitePersistence) GetActivityCenterNotificationsByID(ids []types.HexBy
 	}
 
 	inVector := strings.Repeat("?, ", len(ids)-1) + "?"
-	rows, err := db.db.Query("SELECT a.id, a.read, a.accepted, a.dismissed FROM activity_center_notifications a WHERE a.id IN ("+inVector+") AND NOT a.deleted", idsArgs...) // nolint: gosec
+	// nolint: gosec
+	rows, err := db.db.Query(
+		`
+		SELECT
+		a.id,
+		a.timestamp,
+		a.notification_type,
+		a.chat_id,
+		a.community_id,
+		a.membership_status,
+		a.read,
+		a.accepted,
+		a.dismissed,
+		a.message,
+		c.last_message,
+		a.reply_message,
+		a.contact_verification_status,
+		c.name,
+		a.author,
+		substr('0000000000000000000000000000000000000000000000000000000000000000' || a.timestamp, -64, 64) || hex(a.id) as cursor,
+		a.updated_at
+		FROM activity_center_notifications a
+		LEFT JOIN chats c
+		ON
+		c.id = a.chat_id
+		WHERE a.id IN (`+inVector+`) AND NOT a.deleted`, idsArgs...)
 
 	if err != nil {
 		return nil, err
 	}
 
-	var notifications []*ActivityCenterNotification
+	_, notifications, err := db.unmarshalActivityCenterNotificationRows(rows)
-	for rows.Next() {
+	if err != nil {
-		notification := &ActivityCenterNotification{}
+		return nil, nil
-		err := rows.Scan(
-			&notification.ID,
-			&notification.Read,
-			&notification.Accepted,
-			&notification.Dismissed)
-
-		if err != nil {
-			return nil, err
-		}
-
-		notifications = append(notifications, notification)
 	}
 
 	return notifications, nil

protocol/messenger.go

@@ -3494,8 +3494,8 @@ func (r *ReceivedMessageState) addNewActivityCenterNotification(publicKey ecdsa.
 			Author:        message.From,
 			UpdatedAt:     m.GetCurrentTimeInMillis(),
 			AlbumMessages: albumMessages,
+			Read:          message.Seen,
 		}
-
 		return m.addActivityCenterNotification(r.Response, notification, nil)
 	}
 
@@ -4141,7 +4141,7 @@ func (m *Messenger) DeleteMessagesByChatID(id string) error {
 // MarkMessagesSeen marks messages with `ids` as seen in the chat `chatID`.
 // It returns the number of affected messages or error. If there is an error,
 // the number of affected messages is always zero.
-func (m *Messenger) MarkMessagesSeen(chatID string, ids []string) (uint64, uint64, error) {
+func (m *Messenger) markMessagesSeenImpl(chatID string, ids []string) (uint64, uint64, error) {
 	count, countWithMentions, err := m.persistence.MarkMessagesSeen(chatID, ids)
 	if err != nil {
 		return 0, 0, err
@@ -4154,6 +4154,32 @@ func (m *Messenger) MarkMessagesSeen(chatID string, ids []string) (uint64, uint6
 	return count, countWithMentions, nil
 }
 
+func (m *Messenger) MarkMessagesSeen(chatID string, ids []string) (uint64, uint64, []*ActivityCenterNotification, error) {
+	count, countWithMentions, err := m.markMessagesSeenImpl(chatID, ids)
+	if err != nil {
+		return 0, 0, nil, err
+	}
+
+	hexBytesIds := []types.HexBytes{}
+	for _, id := range ids {
+		hexBytesIds = append(hexBytesIds, types.FromHex(id))
+	}
+
+	// Mark notifications as read in the database
+	updatedAt := m.GetCurrentTimeInMillis()
+	err = m.persistence.MarkActivityCenterNotificationsRead(hexBytesIds, updatedAt)
+	if err != nil {
+		return 0, 0, nil, err
+	}
+
+	notifications, err := m.persistence.GetActivityCenterNotificationsByID(hexBytesIds)
+	if err != nil {
+		return 0, 0, nil, err
+	}
+
+	return count, countWithMentions, notifications, nil
+}
+
 func (m *Messenger) syncChatMessagesRead(ctx context.Context, chatID string, clock uint64, rawMessageHandler RawMessageHandler) error {
 	if !m.hasPairedDevices() {
 		return nil
@@ -4215,30 +4241,40 @@ func (m *Messenger) markAllRead(chatID string, clock uint64, shouldBeSynced bool
 	return m.persistence.SaveChats([]*Chat{chat})
 }
 
-func (m *Messenger) MarkAllRead(ctx context.Context, chatID string) error {
+func (m *Messenger) MarkAllRead(ctx context.Context, chatID string) (*MessengerResponse, error) {
-	_, err := m.DismissAllActivityCenterNotificationsFromChatID(ctx, chatID, m.GetCurrentTimeInMillis())
+	response := &MessengerResponse{}
+
+	notifications, err := m.DismissAllActivityCenterNotificationsFromChatID(ctx, chatID, m.GetCurrentTimeInMillis())
 	if err != nil {
-		return err
+		return nil, err
 	}
+	response.AddActivityCenterNotifications(notifications)
 
 	clock, _ := m.latestIncomingMessageClock(chatID)
 
 	if clock == 0 {
 		chat, ok := m.allChats.Load(chatID)
 		if !ok {
-			return errors.New("chat not found")
+			return nil, errors.New("chat not found")
 		}
 		clock, _ = chat.NextClockAndTimestamp(m.getTimesource())
 	}
 
-	return m.markAllRead(chatID, clock, true)
+	err = m.markAllRead(chatID, clock, true)
-}
-
-func (m *Messenger) MarkAllReadInCommunity(ctx context.Context, communityID string) ([]string, error) {
-	_, err := m.DismissAllActivityCenterNotificationsFromCommunity(ctx, communityID, m.GetCurrentTimeInMillis())
 	if err != nil {
 		return nil, err
 	}
+	return response, nil
+}
+
+func (m *Messenger) MarkAllReadInCommunity(ctx context.Context, communityID string) (*MessengerResponse, error) {
+	response := &MessengerResponse{}
+
+	notifications, err := m.DismissAllActivityCenterNotificationsFromCommunity(ctx, communityID, m.GetCurrentTimeInMillis())
+	if err != nil {
+		return nil, err
+	}
+	response.AddActivityCenterNotifications(notifications)
 
 	chatIDs, err := m.persistence.AllChatIDsByCommunity(nil, communityID)
 	if err != nil {
@@ -4257,14 +4293,12 @@ func (m *Messenger) MarkAllReadInCommunity(ctx context.Context, communityID stri
 			chat.UnviewedMessagesCount = 0
 			chat.UnviewedMentionsCount = 0
 			m.allChats.Store(chat.ID, chat)
+			response.AddChat(chat)
 		} else {
 			err = errors.New(fmt.Sprintf("chat with chatID %s not found", chatID))
 		}
 	}
-	if err != nil {
+	return response, err
-		return chatIDs, err
-	}
-	return chatIDs, err
 }
 
 // MuteChat signals to the messenger that we don't want to be notified

protocol/messenger_activity_center.go

@@ -115,34 +115,17 @@ func (m *Messenger) MarkAsSeenActivityCenterNotifications() (*MessengerResponse,
 }
 
 func (m *Messenger) MarkAllActivityCenterNotificationsRead(ctx context.Context) (*MessengerResponse, error) {
-	response := &MessengerResponse{}
+	ids, err := m.persistence.GetNotReadActivityCenterNotificationIds()
+	if err != nil {
+		return nil, err
+	}
+
 	updateAt := m.GetCurrentTimeInMillis()
-	if m.hasPairedDevices() {
+	return m.MarkActivityCenterNotificationsRead(ctx, toHexBytes(ids), updateAt, true)
-		ids, err := m.persistence.GetNotReadActivityCenterNotificationIds()
-		if err != nil {
-			return nil, err
-		}
-
-		_, err = m.MarkActivityCenterNotificationsRead(ctx, toHexBytes(ids), updateAt, true)
-		return nil, err
-	}
-
-	err := m.persistence.MarkAllActivityCenterNotificationsRead(updateAt)
-	if err != nil {
-		return nil, err
-	}
-
-	state, err := m.persistence.GetActivityCenterState()
-	if err != nil {
-		return nil, err
-	}
-
-	response.SetActivityCenterState(state)
-	return response, nil
 }
 
 func (m *Messenger) MarkActivityCenterNotificationsRead(ctx context.Context, ids []types.HexBytes, updatedAt uint64, sync bool) (*MessengerResponse, error) {
-	response := &MessengerResponse{}
+	// Mark notifications as read in the database
 	if updatedAt == 0 {
 		updatedAt = m.GetCurrentTimeInMillis()
 	}
@@ -151,6 +134,32 @@ func (m *Messenger) MarkActivityCenterNotificationsRead(ctx context.Context, ids
 		return nil, err
 	}
 
+	notifications, err := m.persistence.GetActivityCenterNotificationsByID(ids)
+	if err != nil {
+		return nil, err
+	}
+
+	response := &MessengerResponse{}
+	repliesAndMentions := make(map[string][]string)
+
+	// When marking as read Mention or Reply notification, the corresponding chat message should also be read.
+	for _, notification := range notifications {
+		response.AddActivityCenterNotification(notification)
+
+		if notification.Message != nil &&
+			(notification.Type == ActivityCenterNotificationTypeMention || notification.Type == ActivityCenterNotificationTypeReply) {
+			repliesAndMentions[notification.ChatID] = append(repliesAndMentions[notification.ChatID], notification.Message.ID)
+		}
+	}
+
+	// Mark messages as seen
+	for chatID, messageIDs := range repliesAndMentions {
+		_, _, err := m.markMessagesSeenImpl(chatID, messageIDs)
+		if err != nil {
+			return nil, err
+		}
+	}
+
 	state, err := m.persistence.GetActivityCenterState()
 	if err != nil {
 		return nil, err
@@ -159,10 +168,6 @@ func (m *Messenger) MarkActivityCenterNotificationsRead(ctx context.Context, ids
 	response.SetActivityCenterState(state)
 
 	if !sync {
-		notifications, err := m.persistence.GetActivityCenterNotificationsByID(ids)
-		if err != nil {
-			return nil, err
-		}
 		response2, err := m.processActivityCenterNotifications(notifications, true)
 		if err != nil {
 			return nil, err

protocol/messenger_activity_center_test.go

@@ -3,6 +3,7 @@ package protocol
 import (
 	"context"
 	"crypto/ecdsa"
+	"errors"
 	"testing"
 
 	"github.com/stretchr/testify/suite"
@@ -306,3 +307,178 @@ func (s *MessengerActivityCenterMessageSuite) TestMuteCommunityActivityCenterNot
 	s.Require().True(response.Messages()[0].Mentioned)
 	s.Require().Len(response.ActivityCenterNotifications(), 0)
 }
+
+func (s *MessengerActivityCenterMessageSuite) prepareCommunityChannelWithMentionAndReply() (*Messenger, *Messenger, *common.Message, *common.Message, *communities.Community) {
+	alice := s.m
+	bob := s.newMessenger()
+	_, err := bob.Start()
+	s.Require().NoError(err)
+	defer bob.Shutdown() // nolint: errcheck
+
+	// Create a community
+	community, chat := s.createCommunity(bob)
+	s.Require().NotNil(community)
+	s.Require().NotNil(chat)
+
+	// Alice joins the community
+	s.advertiseCommunityTo(community, bob, alice)
+	s.joinCommunity(community, bob, alice)
+
+	// Bob sends a mention message
+	mentionMessage := common.NewMessage()
+	mentionMessage.ChatId = chat.ID
+	mentionMessage.ContentType = protobuf.ChatMessage_TEXT_PLAIN
+	mentionMessage.Text = "Good news, @" + common.EveryoneMentionTag + " !"
+
+	response, err := bob.SendChatMessage(context.Background(), mentionMessage)
+	s.Require().NoError(err)
+	s.Require().Len(response.Messages(), 1)
+	s.Require().True(response.Messages()[0].Mentioned)
+
+	// check alice got the mention message
+	response, err = WaitOnMessengerResponse(
+		alice,
+		func(r *MessengerResponse) bool {
+			return len(r.Messages()) == 1 && len(r.ActivityCenterNotifications()) == 1 &&
+				r.Messages()[0].ID == r.ActivityCenterNotifications()[0].Message.ID &&
+				r.ActivityCenterNotifications()[0].Type == ActivityCenterNotificationTypeMention
+		},
+		"no messages",
+	)
+	s.Require().NoError(err)
+
+	s.Require().False(response.ActivityCenterNotifications()[0].Read)
+	s.Require().Equal(response.ActivityCenterNotifications()[0].ID.String(), response.ActivityCenterNotifications()[0].Message.ID)
+	mentionMessage = response.Messages()[0]
+
+	// Alice sends a community message
+	inputMessage := common.NewMessage()
+	inputMessage.ChatId = chat.ID
+	inputMessage.ContentType = protobuf.ChatMessage_TEXT_PLAIN
+	inputMessage.Text = "test message"
+
+	response, err = alice.SendChatMessage(context.Background(), inputMessage)
+	s.Require().NoError(err)
+	s.Require().Len(response.Messages(), 1)
+
+	// Check the community message is received by Bob
+	_, err = WaitOnMessengerResponse(
+		bob,
+		func(r *MessengerResponse) bool { return len(r.Messages()) == 1 },
+		"no messages",
+	)
+	s.Require().NoError(err)
+
+	// Bob sends a reply message
+	replyMessage := common.NewMessage()
+	replyMessage.ChatId = chat.ID
+	replyMessage.ContentType = protobuf.ChatMessage_TEXT_PLAIN
+	replyMessage.Text = "test message reply"
+	replyMessage.ResponseTo = response.Messages()[0].ID
+
+	response, err = bob.SendChatMessage(context.Background(), replyMessage)
+	s.Require().NoError(err)
+	s.Require().Len(response.Messages(), 2)
+
+	// Check Alice got the reply message
+	response, err = WaitOnMessengerResponse(
+		alice,
+		func(r *MessengerResponse) bool {
+			return len(r.Messages()) == 2 && len(r.ActivityCenterNotifications()) == 1 &&
+				(r.Messages()[0].ID == r.ActivityCenterNotifications()[0].Message.ID ||
+					r.Messages()[1].ID == r.ActivityCenterNotifications()[0].Message.ID) &&
+				r.ActivityCenterNotifications()[0].Type == ActivityCenterNotificationTypeReply
+		},
+		"no messages",
+	)
+	s.Require().NoError(err)
+	s.Require().False(response.ActivityCenterNotifications()[0].Read)
+
+	// There is an extra message with reply
+	if response.Messages()[0].ID == response.ActivityCenterNotifications()[0].Message.ID {
+		replyMessage = response.Messages()[0]
+	} else if response.Messages()[1].ID == response.ActivityCenterNotifications()[0].Message.ID {
+		replyMessage = response.Messages()[1]
+	} else {
+		s.Error(errors.New("can't find corresponding message in the response"))
+	}
+
+	s.confirmMentionAndReplyNotificationsRead(alice, mentionMessage, replyMessage, false)
+
+	return alice, bob, mentionMessage, replyMessage, community
+}
+
+func (s *MessengerActivityCenterMessageSuite) confirmMentionAndReplyNotificationsRead(user *Messenger, mentionMessage *common.Message, replyMessage *common.Message, read bool) {
+	// Confirm reply notification
+	notifResponse, err := user.ActivityCenterNotifications(ActivityCenterNotificationsRequest{
+		Limit:         8,
+		ReadType:      ActivityCenterQueryParamsReadAll,
+		ActivityTypes: []ActivityCenterType{ActivityCenterNotificationTypeReply},
+	})
+	s.Require().NoError(err)
+	s.Require().Len(notifResponse.Notifications, 1)
+	s.Require().Equal(read, notifResponse.Notifications[0].Read)
+
+	// Confirm mention notification
+	notifResponse, err = user.ActivityCenterNotifications(ActivityCenterNotificationsRequest{
+		Limit:         8,
+		ReadType:      ActivityCenterQueryParamsReadAll,
+		ActivityTypes: []ActivityCenterType{ActivityCenterNotificationTypeMention},
+	})
+	s.Require().NoError(err)
+	s.Require().Len(notifResponse.Notifications, 1)
+	s.Require().Equal(read, notifResponse.Notifications[0].Read)
+}
+
+func (s *MessengerActivityCenterMessageSuite) TestMarkMessagesSeenMarksNotificationsRead() {
+	alice, _, mentionMessage, replyMessage, _ := s.prepareCommunityChannelWithMentionAndReply()
+
+	_, _, notifications, err := alice.MarkMessagesSeen(replyMessage.ChatId, []string{mentionMessage.ID, replyMessage.ID})
+
+	s.Require().NoError(err)
+	s.Require().Len(notifications, 2)
+	s.Require().True(notifications[0].Read)
+	s.Require().True(notifications[1].Read)
+
+	s.confirmMentionAndReplyNotificationsRead(alice, mentionMessage, replyMessage, true)
+}
+
+func (s *MessengerActivityCenterMessageSuite) TestMarkAllReadMarksNotificationsRead() {
+	alice, _, mentionMessage, replyMessage, _ := s.prepareCommunityChannelWithMentionAndReply()
+
+	response, err := alice.MarkAllRead(context.Background(), mentionMessage.ChatId)
+
+	s.Require().NoError(err)
+	s.Require().Len(response.ActivityCenterNotifications(), 2)
+	s.Require().True(response.ActivityCenterNotifications()[0].Read)
+	s.Require().True(response.ActivityCenterNotifications()[1].Read)
+
+	s.confirmMentionAndReplyNotificationsRead(alice, mentionMessage, replyMessage, true)
+}
+
+func (s *MessengerActivityCenterMessageSuite) TestMarkAllReadInCommunityMarksNotificationsRead() {
+	alice, _, mentionMessage, replyMessage, community := s.prepareCommunityChannelWithMentionAndReply()
+
+	response, err := alice.MarkAllReadInCommunity(context.Background(), community.IDString())
+
+	s.Require().NoError(err)
+	s.Require().Len(response.ActivityCenterNotifications(), 2)
+	s.Require().True(response.ActivityCenterNotifications()[0].Read)
+	s.Require().True(response.ActivityCenterNotifications()[1].Read)
+
+	s.confirmMentionAndReplyNotificationsRead(alice, mentionMessage, replyMessage, true)
+}
+
+func (s *MessengerActivityCenterMessageSuite) TestMarkAllActivityCenterNotificationsReadMarksMessagesAsSeen() {
+	alice, _, mentionMessage, replyMessage, _ := s.prepareCommunityChannelWithMentionAndReply()
+
+	response, err := alice.MarkAllActivityCenterNotificationsRead(context.Background())
+
+	s.Require().NoError(err)
+	s.Require().Len(response.ActivityCenterNotifications(), 3)
+	s.Require().True(response.ActivityCenterNotifications()[0].Read)
+	s.Require().True(response.ActivityCenterNotifications()[1].Read)
+	s.Require().True(response.ActivityCenterNotifications()[2].Read)
+
+	s.confirmMentionAndReplyNotificationsRead(alice, mentionMessage, replyMessage, true)
+}

protocol/messenger_sync_chat_test.go

@@ -194,7 +194,7 @@ func (s *MessengerSyncChatSuite) TestMarkChatMessagesRead() {
 	alice2chat := s.alice2.Chat(chatID)
 	s.Require().Equal(alice2chat.UnviewedMessagesCount, uint(1))
 
-	err = s.alice1.MarkAllRead(context.TODO(), chatID)
+	_, err = s.alice1.MarkAllRead(context.TODO(), chatID)
 	s.Require().NoError(err)
 
 	var receivedChat *Chat

protocol/messenger_test.go

@@ -268,16 +268,18 @@ func (s *MessengerSuite) TestMarkMessagesSeen() {
 	err = s.m.SaveMessages([]*common.Message{inputMessage1, inputMessage2})
 	s.Require().NoError(err)
 
-	count, countWithMentions, err := s.m.MarkMessagesSeen(chat.ID, []string{inputMessage1.ID})
+	count, countWithMentions, notifications, err := s.m.MarkMessagesSeen(chat.ID, []string{inputMessage1.ID})
 	s.Require().NoError(err)
 	s.Require().Equal(uint64(1), count)
 	s.Require().Equal(uint64(1), countWithMentions)
+	s.Require().Len(notifications, 0)
 
 	// Make sure that if it's not seen, it does not return a count of 1
-	count, countWithMentions, err = s.m.MarkMessagesSeen(chat.ID, []string{inputMessage1.ID})
+	count, countWithMentions, notifications, err = s.m.MarkMessagesSeen(chat.ID, []string{inputMessage1.ID})
 	s.Require().NoError(err)
 	s.Require().Equal(uint64(0), count)
 	s.Require().Equal(uint64(0), countWithMentions)
+	s.Require().Len(notifications, 0)
 
 	chats := s.m.Chats()
 	for _, c := range chats {
@@ -305,7 +307,7 @@ func (s *MessengerSuite) TestMarkAllRead() {
 	err = s.m.SaveMessages([]*common.Message{inputMessage1, inputMessage2})
 	s.Require().NoError(err)
 
-	err = s.m.MarkAllRead(context.Background(), chat.ID)
+	_, err = s.m.MarkAllRead(context.Background(), chat.ID)
 	s.Require().NoError(err)
 
 	chats := s.m.Chats()

services/ext/api.go

@@ -683,9 +683,10 @@ type ApplicationMessagesResponse struct {
 	Cursor   string            `json:"cursor"`
 }
 
-type MarkMessagSeenResponse struct {
+type MarkMessageSeenResponse struct {
-	Count             uint64 `json:"count"`
+	Count                       uint64                                 `json:"count"`
-	CountWithMentions uint64 `json:"countWithMentions"`
+	CountWithMentions           uint64                                 `json:"countWithMentions"`
+	ActivityCenterNotifications []*protocol.ActivityCenterNotification `json:"activityCenterNotifications,omitempty"`
 }
 
 type ApplicationPinnedMessagesResponse struct {
@@ -801,21 +802,25 @@ func (api *PublicAPI) DeleteMessagesByChatID(id string) error {
 	return api.service.messenger.DeleteMessagesByChatID(id)
 }
 
-func (api *PublicAPI) MarkMessagesSeen(chatID string, ids []string) (*MarkMessagSeenResponse, error) {
+func (api *PublicAPI) MarkMessagesSeen(chatID string, ids []string) (*MarkMessageSeenResponse, error) {
-	count, withMentions, err := api.service.messenger.MarkMessagesSeen(chatID, ids)
+	count, withMentions, notifications, err := api.service.messenger.MarkMessagesSeen(chatID, ids)
 	if err != nil {
 		return nil, err
 	}
 
-	response := &MarkMessagSeenResponse{Count: count, CountWithMentions: withMentions}
+	response := &MarkMessageSeenResponse{
+		Count:                       count,
+		CountWithMentions:           withMentions,
+		ActivityCenterNotifications: notifications,
+	}
 	return response, nil
 }
 
-func (api *PublicAPI) MarkAllRead(ctx context.Context, chatID string) error {
+func (api *PublicAPI) MarkAllRead(ctx context.Context, chatID string) (*protocol.MessengerResponse, error) {
 	return api.service.messenger.MarkAllRead(ctx, chatID)
 }
 
-func (api *PublicAPI) MarkAllReadInCommunity(ctx context.Context, communityID string) ([]string, error) {
+func (api *PublicAPI) MarkAllReadInCommunity(ctx context.Context, communityID string) (*protocol.MessengerResponse, error) {
 	return api.service.messenger.MarkAllReadInCommunity(ctx, communityID)
 }