MM-36234,MM-37030,MM-37031: CRT, desktop thread notifications (#18088)

* CRT: desktop thread notifications

* Fixes go lint

* Adds default for desktop CRT notifications

* Adds email and push notifications for CRT threads

Adds user ids of thread followers with CRT to crtMentions so they will get
notified appropriately.

* Minor change

* Refactor a bit

CRTMentions.addMention had a bug on the return and de-duplication.
This commit fixes duplicate notifications by looking up if the user is to be
notified on CRT on both email and push notifications.

* Minor refactor

* Changes according to review comments

- Fixes adding to followers a user that had explicitly unfollowed a
  thread.
- Simplified send email according to email_threads option
- Send mentions and followers in separate arrays via the websocket
- Fixes push notifications message for push_threads

* Adds a comment on a buggy use case

* Updates comment to correct ticket link

* Fixes when user notifications is set to all

There was a bug where if user had set notifications to all
then they would receive desktop notifications even for non following threads.

A similar bug existed in push notifications, where if a user has set it
to all the threads setting would still be considered.

This commit fixes that by adding users to notificationsForCRT
StringArray when they have the non thread setting to 'all'.

* Fixes notifications to users unfollowing threads

Users which had previously explicitly unfollowed a thread
should not receive notifications about those threads.

* Update store mocks

* Fixes push notifications for CRT

Push notification about replies for CRT users should have a title of
"Reply to Thread".

CRT users with global user setting to 'UserNotifyAll' should not get
notifications for unfollowed threads.

This commit fixes those issues.

* Fixes i18n error

Co-authored-by: Mattermod <mattermod@users.noreply.github.com>

app/notification.go

@@ -27,6 +27,8 @@ func (a *App) SendNotifications(post *model.Post, team *model.Team, channel *mod
 		return []string{}, nil
 	}
 
+	isCRTAllowed := a.Config().FeatureFlags.CollapsedThreads && *a.Config().ServiceSettings.CollapsedThreads != model.CollapsedThreadsDisabled
+
 	pchan := make(chan store.StoreResult, 1)
 	go func() {
 		props, err := a.Srv().Store.User().GetAllProfilesInChannel(context.Background(), channel.Id, true)
@@ -61,6 +63,16 @@ func (a *App) SendNotifications(post *model.Post, team *model.Team, channel *mod
 		}()
 	}
 
+	var tchan chan store.StoreResult
+	if isCRTAllowed && post.RootId != "" {
+		tchan = make(chan store.StoreResult, 1)
+		go func() {
+			followers, err := a.Srv().Store.Thread().GetThreadFollowers(post.RootId, true)
+			tchan <- store.StoreResult{Data: followers, NErr: err}
+			close(tchan)
+		}()
+	}
+
 	result := <-pchan
 	if result.NErr != nil {
 		return nil, result.NErr
@@ -73,6 +85,15 @@ func (a *App) SendNotifications(post *model.Post, team *model.Team, channel *mod
 	}
 	channelMemberNotifyPropsMap := result.Data.(map[string]model.StringMap)
 
+	followers := make(model.StringArray, 0)
+	if tchan != nil {
+		result = <-tchan
+		if result.NErr != nil {
+			return nil, result.NErr
+		}
+		followers = result.Data.([]string)
+	}
+
 	groups := make(map[string]*model.Group)
 	if gchan != nil {
 		result = <-gchan
@@ -158,11 +179,13 @@ func (a *App) SendNotifications(post *model.Post, team *model.Team, channel *mod
 		}()
 
 		// find which users in the channel are set up to always receive mobile notifications
+		// excludes CRT users since those should be added in notificationsForCRT
 		for _, profile := range profileMap {
 			if (profile.NotifyProps[model.PushNotifyProp] == model.UserNotifyAll ||
 				channelMemberNotifyPropsMap[profile.Id][model.PushNotifyProp] == model.ChannelNotifyAll) &&
 				(post.UserId != profile.Id || post.GetProp("from_webhook") == "true") &&
-				!post.IsSystemMessage() {
+				!post.IsSystemMessage() &&
+				!(a.isCRTEnabledForUser(profile.Id) && post.RootId != "") {
 				allActivityPushUserIds = append(allActivityPushUserIds, profile.Id)
 			}
 		}
@@ -174,6 +197,7 @@ func (a *App) SendNotifications(post *model.Post, team *model.Team, channel *mod
 	threadParticipants := map[string]bool{post.UserId: true}
 	participantMemberships := map[string]*model.ThreadMembership{}
 	membershipsMutex := &sync.Mutex{}
+	followersMutex := &sync.Mutex{}
 	if *a.Config().ServiceSettings.ThreadAutoFollow && post.RootId != "" {
 		var rootMentions *ExplicitMentions
 		if parentPostList != nil {
@@ -230,6 +254,14 @@ func (a *App) SendNotifications(post *model.Post, team *model.Team, channel *mod
 					mac <- model.NewAppError("SendNotifications", "app.channel.autofollow.app_error", nil, err.Error(), http.StatusInternalServerError)
 					return
 				}
+
+				// add new followers to existing followers
+				if threadMembership.Following && !followers.Contains(userID) {
+					followersMutex.Lock()
+					followers = append(followers, userID)
+					followersMutex.Unlock()
+				}
+
 				membershipsMutex.Lock()
 				participantMemberships[userID] = threadMembership
 				membershipsMutex.Unlock()
@@ -255,6 +287,23 @@ func (a *App) SendNotifications(post *model.Post, team *model.Team, channel *mod
 		updateMentionChans = append(updateMentionChans, umc)
 	}
 
+	notificationsForCRT := &CRTNotifiers{}
+	if isCRTAllowed && post.RootId != "" {
+		for _, uid := range followers {
+			profile := profileMap[uid]
+			if profile == nil || !a.isCRTEnabledForUser(uid) {
+				continue
+			}
+
+			if post.GetProp("from_webhook") != "true" && uid == post.UserId {
+				continue
+			}
+
+			// add user id to notificationsForCRT depending on threads notify props
+			notificationsForCRT.addUserToNotify(profile, mentions)
+		}
+	}
+
 	notification := &PostNotification{
 		Post:       post.Clone(),
 		Channel:    channel,
@@ -263,7 +312,10 @@ func (a *App) SendNotifications(post *model.Post, team *model.Team, channel *mod
 	}
 
 	if *a.Config().EmailSettings.SendEmailNotifications {
-		for _, id := range mentionedUsersList {
+		emailReceipients := append(mentionedUsersList, notificationsForCRT.Email...)
+		emailReceipients = model.RemoveDuplicateStrings(emailReceipients)
+
+		for _, id := range emailReceipients {
 			if profileMap[id] == nil {
 				continue
 			}
@@ -362,7 +414,7 @@ func (a *App) SendNotifications(post *model.Post, team *model.Team, channel *mod
 
 	if sendPushNotifications {
 		for _, id := range mentionedUsersList {
-			if profileMap[id] == nil {
+			if profileMap[id] == nil || notificationsForCRT.Push.Contains(id) {
 				continue
 			}
 
@@ -402,7 +454,7 @@ func (a *App) SendNotifications(post *model.Post, team *model.Team, channel *mod
 		}
 
 		for _, id := range allActivityPushUserIds {
-			if profileMap[id] == nil {
+			if profileMap[id] == nil || notificationsForCRT.Push.Contains(id) {
 				continue
 			}
 
@@ -433,6 +485,37 @@ func (a *App) SendNotifications(post *model.Post, team *model.Team, channel *mod
 				}
 			}
 		}
+
+		for _, id := range notificationsForCRT.Push {
+			if profileMap[id] == nil {
+				continue
+			}
+
+			var status *model.Status
+			var err *model.AppError
+			if status, err = a.GetStatus(id); err != nil {
+				status = &model.Status{UserId: id, Status: model.StatusOffline, Manual: false, LastActivityAt: 0, ActiveChannel: ""}
+			}
+
+			if DoesStatusAllowPushNotification(profileMap[id].NotifyProps, status, post.ChannelId) {
+				a.sendPushNotification(
+					notification,
+					profileMap[id],
+					profileMap[id].NotifyProps[model.PushThreadsNotifyProp] == model.UserNotifyMention,
+					false,
+					model.UserNotifyAll,
+				)
+			} else {
+				// register that a notification was not sent
+				a.NotificationsLog().Debug("Notification not sent",
+					mlog.String("ackId", ""),
+					mlog.String("type", model.PushTypeMessage),
+					mlog.String("userId", id),
+					mlog.String("postId", post.Id),
+					mlog.String("status", model.PushNotSent),
+				)
+			}
+		}
 	}
 
 	message := model.NewWebSocketEvent(model.WebsocketEventPosted, "", post.ChannelId, "", nil)
@@ -469,6 +552,10 @@ func (a *App) SendNotifications(post *model.Post, team *model.Team, channel *mod
 		message.Add("mentions", model.ArrayToJson(mentionedUsersList))
 	}
 
+	if len(notificationsForCRT.Desktop) != 0 {
+		message.Add("followers", model.ArrayToJson(notificationsForCRT.Desktop))
+	}
+
 	published, err := a.publishWebsocketEventForPermalinkPost(post, message)
 	if err != nil {
 		return nil, err
@@ -478,28 +565,25 @@ func (a *App) SendNotifications(post *model.Post, team *model.Team, channel *mod
 	}
 
 	// If this is a reply in a thread, notify participants
-	if a.Config().FeatureFlags.CollapsedThreads && *a.Config().ServiceSettings.CollapsedThreads != model.CollapsedThreadsDisabled && post.RootId != "" {
-		followers, err := a.Srv().Store.Thread().GetThreadFollowers(post.RootId)
-		if err != nil {
-			return nil, errors.Wrapf(err, "cannot get thread %q followers", post.RootId)
-		}
+	if isCRTAllowed && post.RootId != "" {
 		for _, uid := range followers {
-			sendEvent := *a.Config().ServiceSettings.CollapsedThreads == model.CollapsedThreadsDefaultOn
-			// check if a participant has overridden collapsed threads settings
-			if preference, prefErr := a.Srv().Store.Preference().Get(uid, model.PreferenceCategoryDisplaySettings, model.PreferenceNameCollapsedThreadsEnabled); prefErr == nil {
-				sendEvent = preference.Value == "on"
+			// A user following a thread but had left the channel won't get a notification
+			// https://mattermost.atlassian.net/browse/MM-36769
+			if profileMap[uid] == nil {
+				continue
 			}
-			if sendEvent {
+			if a.isCRTEnabledForUser(uid) {
 				message := model.NewWebSocketEvent(model.WebsocketEventThreadUpdated, team.Id, "", uid, nil)
 				threadMembership := participantMemberships[uid]
 				if threadMembership == nil {
-					threadMembership, err = a.Srv().Store.Thread().GetMembershipForUser(uid, post.RootId)
+					tm, err := a.Srv().Store.Thread().GetMembershipForUser(uid, post.RootId)
 					if err != nil {
 						return nil, errors.Wrapf(err, "Missing thread membership for participant in notifications. user_id=%q thread_id=%q", uid, post.RootId)
 					}
-					if threadMembership == nil {
+					if tm == nil {
 						continue
 					}
+					threadMembership = tm
 				}
 				userThread, err := a.Srv().Store.Thread().GetThreadForUser(channel.TeamId, threadMembership, true)
 				if err != nil {
@@ -525,14 +609,15 @@ func (a *App) SendNotifications(post *model.Post, team *model.Team, channel *mod
 				}
 			}
 		}
-
 	}
 	return mentionedUsersList, nil
 }
 
 func (a *App) userAllowsEmail(user *model.User, channelMemberNotificationProps model.StringMap, post *model.Post) bool {
 	userAllowsEmails := user.NotifyProps[model.EmailNotifyProp] != "false"
-	if channelEmail, ok := channelMemberNotificationProps[model.EmailNotifyProp]; ok {
+
+	// if CRT is ON for user and the post is a reply disregard the channelEmail setting
+	if channelEmail, ok := channelMemberNotificationProps[model.EmailNotifyProp]; ok && !(a.isCRTEnabledForUser(user.Id) && post.RootId != "") {
 		if channelEmail != model.ChannelNotifyDefault {
 			userAllowsEmails = channelEmail != "false"
 		}
@@ -1228,3 +1313,40 @@ func (a *App) GetNotificationNameFormat(user *model.User) string {
 
 	return data.Value
 }
+
+type CRTNotifiers struct {
+	// Desktop contains the user IDs of thread followers to receive desktop notification
+	Desktop model.StringArray
+
+	// Email contains the user IDs of thread followers to receive email notification
+	Email model.StringArray
+
+	// Push contains the user IDs of thread followers to receive push notification
+	Push model.StringArray
+}
+
+func (c *CRTNotifiers) addUserToNotify(user *model.User, mentions *ExplicitMentions) {
+	// user notify props
+	desktop := user.NotifyProps[model.DesktopNotifyProp]
+	push := user.NotifyProps[model.PushNotifyProp]
+	shouldEmail := user.NotifyProps[model.EmailNotifyProp] == "true"
+
+	// user thread notify props
+	desktopThreads := user.NotifyProps[model.DesktopThreadsNotifyProp]
+	emailThreads := user.NotifyProps[model.EmailThreadsNotifyProp]
+	pushThreads := user.NotifyProps[model.PushThreadsNotifyProp]
+
+	_, userWasMentioned := mentions.Mentions[user.Id]
+
+	if desktop != model.UserNotifyNone && (userWasMentioned || desktopThreads == model.UserNotifyAll || desktop == model.UserNotifyAll) {
+		c.Desktop = append(c.Desktop, user.Id)
+	}
+
+	if shouldEmail && (userWasMentioned || emailThreads == model.UserNotifyAll) {
+		c.Email = append(c.Email, user.Id)
+	}
+
+	if push != model.UserNotifyNone && (userWasMentioned || pushThreads == model.UserNotifyAll || push == model.UserNotifyAll) {
+		c.Push = append(c.Push, user.Id)
+	}
+}

app/notification_email.go

@@ -236,7 +236,10 @@ func (a *App) getNotificationEmailBody(recipient *model.User, post *model.Post,
 	data.Props["NotificationFooterInfoLogin"] = translateFunc("app.notification.footer.infoLogin")
 	data.Props["NotificationFooterInfo"] = translateFunc("app.notification.footer.info")
 
-	if channel.Type == model.ChannelTypeDirect {
+	if a.isCRTEnabledForUser(recipient.Id) && post.RootId != "" {
+		data.Props["Title"] = translateFunc("app.notification.body.thread.title", map[string]interface{}{"SenderName": senderName})
+		data.Props["SubTitle"] = translateFunc("app.notification.body.thread.subTitle", map[string]interface{}{"SenderName": senderName})
+	} else if channel.Type == model.ChannelTypeDirect {
 		// Direct Messages
 		data.Props["Title"] = translateFunc("app.notification.body.dm.title", map[string]interface{}{"SenderName": senderName})
 		data.Props["SubTitle"] = translateFunc("app.notification.body.dm.subTitle", map[string]interface{}{"SenderName": senderName})

app/notification_push.go

@@ -198,6 +198,10 @@ func (a *App) getPushNotificationMessage(contentsConfig, postMessage string, exp
 		return senderName + userLocale("api.post.send_notification_and_forget.push_comment_on_thread")
 	}
 
+	if replyToThreadType == model.UserNotifyAll {
+		return senderName + userLocale("api.post.send_notification_and_forget.push_comment_on_crt_thread")
+	}
+
 	return senderName + userLocale("api.post.send_notifications_and_forget.push_general_message")
 }
 
@@ -559,9 +563,13 @@ func (a *App) buildFullPushNotificationMessage(contentsConfig string, post *mode
 		IsIdLoaded: false,
 	}
 
+	userLocale := i18n.GetUserTranslations(user.Locale)
 	cfg := a.Config()
 	if contentsConfig != model.GenericNoChannelNotification || channel.Type == model.ChannelTypeDirect {
 		msg.ChannelName = channelName
+		if a.isCRTEnabledForUser(user.Id) && post.RootId != "" {
+			msg.ChannelName = userLocale("api.push_notification.title.collapsed_threads")
+		}
 	}
 
 	msg.SenderName = senderName
@@ -591,7 +599,6 @@ func (a *App) buildFullPushNotificationMessage(contentsConfig string, post *mode
 		}
 	}
 
-	userLocale := i18n.GetUserTranslations(user.Locale)
 	hasFiles := post.FileIds != nil && len(post.FileIds) > 0
 
 	msg.Message = a.getPushNotificationMessage(

i18n/en.json

@@ -2230,6 +2230,10 @@
     "id": "api.post.send_notification_and_forget.push_channel_mention",
     "translation": " notified the channel."
   },
+  {
+    "id": "api.post.send_notification_and_forget.push_comment_on_crt_thread",
+    "translation": " commented on a thread you are following."
+  },
   {
     "id": "api.post.send_notification_and_forget.push_comment_on_post",
     "translation": " commented on your post."
@@ -2310,6 +2314,10 @@
     "id": "api.push_notification.id_loaded.fetch.app_error",
     "translation": "An error occurred fetching the ID-loaded push notification."
   },
+  {
+    "id": "api.push_notification.title.collapsed_threads",
+    "translation": "Reply to Thread"
+  },
   {
     "id": "api.push_notifications.message.parse.app_error",
     "translation": "An error occurred building the push notification message."
@@ -5470,6 +5478,14 @@
     "id": "app.notification.body.mention.title",
     "translation": "{{.SenderName}} mentioned you in a message"
   },
+  {
+    "id": "app.notification.body.thread.subTitle",
+    "translation": "While you were away, {{.SenderName}} replied to a thread you're following."
+  },
+  {
+    "id": "app.notification.body.thread.title",
+    "translation": "{{.SenderName}} replied to a thread"
+  },
   {
     "id": "app.notification.footer.info",
     "translation": " and go to Account Settings > Notifications"

model/user.go

@@ -43,6 +43,9 @@ const (
 	FirstNameNotifyProp            = "first_name"
 	AutoResponderActiveNotifyProp  = "auto_responder_active"
 	AutoResponderMessageNotifyProp = "auto_responder_message"
+	DesktopThreadsNotifyProp       = "desktop_threads"
+	PushThreadsNotifyProp          = "push_threads"
+	EmailThreadsNotifyProp         = "email_threads"
 
 	DefaultLocale        = "en"
 	UserAuthServiceEmail = "email"
@@ -449,6 +452,9 @@ func (u *User) SetDefaultNotifications() {
 	u.NotifyProps[PushStatusNotifyProp] = StatusAway
 	u.NotifyProps[CommentsNotifyProp] = CommentsNotifyNever
 	u.NotifyProps[FirstNameNotifyProp] = "false"
+	u.NotifyProps[DesktopThreadsNotifyProp] = UserNotifyAll
+	u.NotifyProps[EmailThreadsNotifyProp] = UserNotifyAll
+	u.NotifyProps[PushThreadsNotifyProp] = UserNotifyAll
 }
 
 func (u *User) UpdateMentionKeysFromUsername(oldUsername string) {

store/opentracinglayer/opentracinglayer.go

@@ -9033,7 +9033,7 @@ func (s *OpenTracingLayerThreadStore) GetPosts(threadID string, since int64) ([]
 	return result, err
 }
 
-func (s *OpenTracingLayerThreadStore) GetThreadFollowers(threadID string) ([]string, error) {
+func (s *OpenTracingLayerThreadStore) GetThreadFollowers(threadID string, fetchOnlyActive bool) ([]string, error) {
 	origCtx := s.Root.Store.Context()
 	span, newCtx := tracing.StartSpanWithParentByContext(s.Root.Store.Context(), "ThreadStore.GetThreadFollowers")
 	s.Root.Store.SetContext(newCtx)
@@ -9042,7 +9042,7 @@ func (s *OpenTracingLayerThreadStore) GetThreadFollowers(threadID string) ([]str
 	}()
 
 	defer span.Finish()
-	result, err := s.ThreadStore.GetThreadFollowers(threadID)
+	result, err := s.ThreadStore.GetThreadFollowers(threadID, fetchOnlyActive)
 	if err != nil {
 		span.LogFields(spanlog.Error(err))
 		ext.Error.Set(span, true)

store/retrylayer/retrylayer.go

@@ -9842,11 +9842,11 @@ func (s *RetryLayerThreadStore) GetPosts(threadID string, since int64) ([]*model
 
 }
 
-func (s *RetryLayerThreadStore) GetThreadFollowers(threadID string) ([]string, error) {
+func (s *RetryLayerThreadStore) GetThreadFollowers(threadID string, fetchOnlyActive bool) ([]string, error) {
 
 	tries := 0
 	for {
-		result, err := s.ThreadStore.GetThreadFollowers(threadID)
+		result, err := s.ThreadStore.GetThreadFollowers(threadID, fetchOnlyActive)
 		if err == nil {
 			return result, nil
 		}

store/sqlstore/thread_store.go

@@ -359,12 +359,24 @@ func (s *SqlThreadStore) GetThreadsForUser(userId, teamId string, opts model.Get
 	return result, nil
 }
 
-func (s *SqlThreadStore) GetThreadFollowers(threadID string) ([]string, error) {
+func (s *SqlThreadStore) GetThreadFollowers(threadID string, fetchOnlyActive bool) ([]string, error) {
 	var users []string
+
+	fetchConditions := sq.And{
+		sq.Eq{"PostId": threadID},
+	}
+
+	if fetchOnlyActive {
+		fetchConditions = sq.And{
+			sq.Eq{"Following": true},
+			fetchConditions,
+		}
+	}
+
 	query, args, _ := s.getQueryBuilder().
 		Select("ThreadMemberships.UserId").
 		From("ThreadMemberships").
-		Where(sq.Eq{"PostId": threadID}).ToSql()
+		Where(fetchConditions).ToSql()
 	_, err := s.GetReplica().Select(&users, query, args...)
 
 	if err != nil {

store/store.go

@@ -279,7 +279,7 @@ type ChannelMemberHistoryStore interface {
 	PermanentDeleteBatch(endTime int64, limit int64) (int64, error)
 }
 type ThreadStore interface {
-	GetThreadFollowers(threadID string) ([]string, error)
+	GetThreadFollowers(threadID string, fetchOnlyActive bool) ([]string, error)
 
 	SaveMultiple(thread []*model.Thread) ([]*model.Thread, int, error)
 	Save(thread *model.Thread) (*model.Thread, error)

store/storetest/mocks/ThreadStore.go

@@ -179,13 +179,13 @@ func (_m *ThreadStore) GetPosts(threadID string, since int64) ([]*model.Post, er
 	return r0, r1
 }
 
-// GetThreadFollowers provides a mock function with given fields: threadID
-func (_m *ThreadStore) GetThreadFollowers(threadID string) ([]string, error) {
-	ret := _m.Called(threadID)
+// GetThreadFollowers provides a mock function with given fields: threadID, fetchOnlyActive
+func (_m *ThreadStore) GetThreadFollowers(threadID string, fetchOnlyActive bool) ([]string, error) {
+	ret := _m.Called(threadID, fetchOnlyActive)
 
 	var r0 []string
-	if rf, ok := ret.Get(0).(func(string) []string); ok {
-		r0 = rf(threadID)
+	if rf, ok := ret.Get(0).(func(string, bool) []string); ok {
+		r0 = rf(threadID, fetchOnlyActive)
 	} else {
 		if ret.Get(0) != nil {
 			r0 = ret.Get(0).([]string)
@@ -193,8 +193,8 @@ func (_m *ThreadStore) GetThreadFollowers(threadID string) ([]string, error) {
 	}
 
 	var r1 error
-	if rf, ok := ret.Get(1).(func(string) error); ok {
-		r1 = rf(threadID)
+	if rf, ok := ret.Get(1).(func(string, bool) error); ok {
+		r1 = rf(threadID, fetchOnlyActive)
 	} else {
 		r1 = ret.Error(1)
 	}

store/timerlayer/timerlayer.go

@@ -8135,10 +8135,10 @@ func (s *TimerLayerThreadStore) GetPosts(threadID string, since int64) ([]*model
 	return result, err
 }
 
-func (s *TimerLayerThreadStore) GetThreadFollowers(threadID string) ([]string, error) {
+func (s *TimerLayerThreadStore) GetThreadFollowers(threadID string, fetchOnlyActive bool) ([]string, error) {
 	start := timemodule.Now()
 
-	result, err := s.ThreadStore.GetThreadFollowers(threadID)
+	result, err := s.ThreadStore.GetThreadFollowers(threadID, fetchOnlyActive)
 
 	elapsed := float64(timemodule.Since(start)) / float64(timemodule.Second)
 	if s.Root.Metrics != nil {