MM-36687 CRT push notifications (#18347)

* CRT changes

* Lint fix

* Moved postId check

* Moved postId check

* Lint fix

* Misc

* Added test case for badge count while clearing notifications

* Fixed count when CRT is on and added isCRTEnabled, teamId to isIdLoaded notifications

* test fix

* isCRTEnabledForUser capitalised

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

api4/system.go

@@ -511,11 +511,6 @@ func pushNotificationAck(c *Context, w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
-	if _, appErr := c.App.GetPostIfAuthorized(ack.PostId, c.AppContext.Session()); appErr != nil {
-		c.Err = appErr
-		return
-	}
-
 	if !*c.App.Config().EmailSettings.SendPushNotifications {
 		c.Err = model.NewAppError("pushNotificationAck", "api.push_notification.disabled.app_error", nil, "", http.StatusNotImplemented)
 		return
@@ -533,21 +528,28 @@ func pushNotificationAck(c *Context, w http.ResponseWriter, r *http.Request) {
 			)
 		}
 
-		notificationInterface := c.App.Notification()
+		// Return post data only when PostId is passed.
+		if ack.PostId != "" && ack.NotificationType == model.PushTypeMessage {
+			if _, appErr := c.App.GetPostIfAuthorized(ack.PostId, c.AppContext.Session()); appErr != nil {
+				c.Err = appErr
+				return
+			}
 
-		if notificationInterface == nil {
+			notificationInterface := c.App.Notification()
-			c.Err = model.NewAppError("pushNotificationAck", "api.system.id_loaded.not_available.app_error", nil, "", http.StatusFound)
-			return
-		}
 
-		msg, appError := notificationInterface.GetNotificationMessage(&ack, c.AppContext.Session().UserId)
+			if notificationInterface == nil {
-		if appError != nil {
+				c.Err = model.NewAppError("pushNotificationAck", "api.system.id_loaded.not_available.app_error", nil, "", http.StatusFound)
-			c.Err = model.NewAppError("pushNotificationAck", "api.push_notification.id_loaded.fetch.app_error", nil, appError.Error(), http.StatusInternalServerError)
+				return
-			return
+			}
-		}
 
-		if err2 := json.NewEncoder(w).Encode(msg); err2 != nil {
+			msg, appError := notificationInterface.GetNotificationMessage(&ack, c.AppContext.Session().UserId)
-			mlog.Warn("Error while writing response", mlog.Err(err2))
+			if appError != nil {
+				c.Err = model.NewAppError("pushNotificationAck", "api.push_notification.id_loaded.fetch.app_error", nil, appError.Error(), http.StatusInternalServerError)
+				return
+			}
+			if err2 := json.NewEncoder(w).Encode(msg); err2 != nil {
+				mlog.Warn("Error while writing response", mlog.Err(err2))
+			}
 		}
 
 		return

api4/system_test.go

@@ -771,7 +771,7 @@ func TestPushNotificationAck(t *testing.T) {
 		resp := httptest.NewRecorder()
 		req := httptest.NewRequest("POST", "/api/v4/notifications/ack", nil)
 		req.Header.Set(model.HeaderAuth, "Bearer "+session.Token)
-		req.Body = ioutil.NopCloser(bytes.NewBufferString(fmt.Sprintf(`{"id":"123", "is_id_loaded":true, "post_id":"%s"}`, privatePost.Id)))
+		req.Body = ioutil.NopCloser(bytes.NewBufferString(fmt.Sprintf(`{"id":"123", "is_id_loaded":true, "post_id":"%s", "type": "%s"}`, privatePost.Id, model.PushTypeMessage)))
 
 		handler.ServeHTTP(resp, req)
 		assert.Equal(t, http.StatusForbidden, resp.Code)

api4/user.go

@@ -3048,7 +3048,7 @@ func updateReadStateThreadByUser(c *Context, w http.ResponseWriter, r *http.Requ
 		return
 	}
 
-	thread, err := c.App.UpdateThreadReadForUser(c.Params.UserId, c.Params.TeamId, c.Params.ThreadId, c.Params.Timestamp)
+	thread, err := c.App.UpdateThreadReadForUser(c.AppContext.Session().Id, c.Params.UserId, c.Params.TeamId, c.Params.ThreadId, c.Params.Timestamp)
 	if err != nil {
 		c.Err = err
 		return

app/app_iface.go

@@ -811,6 +811,7 @@ type AppIface interface {
 	InviteGuestsToChannelsGracefully(teamID string, guestsInvite *model.GuestsInvite, senderId string) ([]*model.EmailInviteWithError, *model.AppError)
 	InviteNewUsersToTeam(emailList []string, teamID, senderId string) *model.AppError
 	InviteNewUsersToTeamGracefully(emailList []string, teamID, senderId string) ([]*model.EmailInviteWithError, *model.AppError)
+	IsCRTEnabledForUser(userID string) bool
 	IsFirstUserAccount() bool
 	IsLeader() bool
 	IsPasswordValid(password string) *model.AppError
@@ -1060,7 +1061,7 @@ type AppIface interface {
 	UpdateTeamPrivacy(teamID string, teamType string, allowOpenInvite bool) *model.AppError
 	UpdateTeamScheme(team *model.Team) (*model.Team, *model.AppError)
 	UpdateThreadFollowForUser(userID, teamID, threadID string, state bool) *model.AppError
-	UpdateThreadReadForUser(userID, teamID, threadID string, timestamp int64) (*model.ThreadResponse, *model.AppError)
+	UpdateThreadReadForUser(currentSessionId, userID, teamID, threadID string, timestamp int64) (*model.ThreadResponse, *model.AppError)
 	UpdateThreadsReadForUser(userID, teamID string) *model.AppError
 	UpdateUser(user *model.User, sendNotifications bool) (*model.User, *model.AppError)
 	UpdateUserActive(c *request.Context, userID string, active bool) *model.AppError

app/channel.go

@@ -2493,7 +2493,7 @@ func (a *App) UpdateChannelLastViewedAt(channelIDs []string, userID string) *mod
 	return nil
 }
 
-func (a *App) isCRTEnabledForUser(userID string) bool {
+func (a *App) IsCRTEnabledForUser(userID string) bool {
 	if *a.Config().ServiceSettings.CollapsedThreads == model.CollapsedThreadsDisabled {
 		return false
 	}
@@ -2507,7 +2507,7 @@ func (a *App) isCRTEnabledForUser(userID string) bool {
 
 // MarkChanelAsUnreadFromPost will take a post and set the channel as unread from that one.
 func (a *App) MarkChannelAsUnreadFromPost(postID string, userID string, collapsedThreadsSupported, followThread bool) (*model.ChannelUnreadAt, *model.AppError) {
-	if !collapsedThreadsSupported || !a.isCRTEnabledForUser(userID) {
+	if !collapsedThreadsSupported || !a.IsCRTEnabledForUser(userID) {
 		return a.markChannelAsUnreadFromPostCRTUnsupported(postID, userID)
 	}
 	post, err := a.GetSinglePost(postID)
@@ -2692,7 +2692,7 @@ func (a *App) markChannelAsUnreadFromPostCRTUnsupported(postID string, userID st
 	a.sanitizeProfiles(thread.Participants, false)
 	thread.Post.SanitizeProps()
 
-	if a.isCRTEnabledForUser(userID) {
+	if a.IsCRTEnabledForUser(userID) {
 		payload, jsonErr := json.Marshal(thread)
 		if jsonErr != nil {
 			mlog.Warn("Failed to encode thread to JSON")
@@ -2909,15 +2909,15 @@ func (a *App) MarkChannelsAsViewed(channelIDs []string, userID string, currentSe
 		}
 	}
 	for _, channelID := range channelsToClearPushNotifications {
-		a.clearPushNotification(currentSessionId, userID, channelID)
+		a.clearPushNotification(currentSessionId, userID, channelID, "")
 	}
 
-	if !collapsedThreadsSupported || !a.isCRTEnabledForUser(userID) {
+	if !collapsedThreadsSupported || !a.IsCRTEnabledForUser(userID) {
 		if err := a.Srv().Store.Thread().MarkAllAsReadInChannels(userID, channelIDs); err != nil {
 			return nil, model.NewAppError("MarkChannelsAsViewed", "app.channel.update_last_viewed_at.app_error", nil, err.Error(), http.StatusInternalServerError)
 		}
 
-		if a.isCRTEnabledForUser(userID) {
+		if a.IsCRTEnabledForUser(userID) {
 			timestamp := model.GetMillis()
 			for _, channelID := range channelIDs {
 				message := model.NewWebSocketEvent(model.WebsocketEventThreadReadChanged, "", channelID, userID, nil)

app/notification.go

@@ -186,7 +186,7 @@ func (a *App) SendNotifications(post *model.Post, team *model.Team, channel *mod
 				channelMemberNotifyPropsMap[profile.Id][model.PushNotifyProp] == model.ChannelNotifyAll) &&
 				(post.UserId != profile.Id || post.GetProp("from_webhook") == "true") &&
 				!post.IsSystemMessage() &&
-				!(a.isCRTEnabledForUser(profile.Id) && post.RootId != "") {
+				!(a.IsCRTEnabledForUser(profile.Id) && post.RootId != "") {
 				allActivityPushUserIds = append(allActivityPushUserIds, profile.Id)
 			}
 		}
@@ -320,7 +320,7 @@ func (a *App) SendNotifications(post *model.Post, team *model.Team, channel *mod
 	if isCRTAllowed && post.RootId != "" {
 		for _, uid := range followers {
 			profile := profileMap[uid]
-			if profile == nil || !a.isCRTEnabledForUser(uid) {
+			if profile == nil || !a.IsCRTEnabledForUser(uid) {
 				continue
 			}
 
@@ -580,7 +580,7 @@ func (a *App) SendNotifications(post *model.Post, team *model.Team, channel *mod
 			if profileMap[uid] == nil {
 				continue
 			}
-			if a.isCRTEnabledForUser(uid) {
+			if a.IsCRTEnabledForUser(uid) {
 				message := model.NewWebSocketEvent(model.WebsocketEventThreadUpdated, team.Id, "", uid, nil)
 				threadMembership := participantMemberships[uid]
 				if threadMembership == nil {
@@ -625,7 +625,7 @@ func (a *App) userAllowsEmail(user *model.User, channelMemberNotificationProps m
 	userAllowsEmails := user.NotifyProps[model.EmailNotifyProp] != "false"
 
 	// 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, ok := channelMemberNotificationProps[model.EmailNotifyProp]; ok && !(a.IsCRTEnabledForUser(user.Id) && post.RootId != "") {
 		if channelEmail != model.ChannelNotifyDefault {
 			userAllowsEmails = channelEmail != "false"
 		}

app/notification_email.go

@@ -261,7 +261,7 @@ func (a *App) getNotificationEmailBody(recipient *model.User, post *model.Post,
 	}
 
 	// Override title and subtile for replies with CRT enabled
-	if a.isCRTEnabledForUser(recipient.Id) && post.RootId != "" {
+	if a.IsCRTEnabledForUser(recipient.Id) && post.RootId != "" {
 		// Title is the same in all cases
 		data.Props["Title"] = translateFunc("app.notification.body.thread.title", map[string]interface{}{"SenderName": senderName})
 

app/notification_push.go

@@ -45,6 +45,7 @@ type PushNotification struct {
 	currentSessionId   string
 	userID             string
 	channelID          string
+	rootID             string
 	post               *model.Post
 	user               *model.User
 	channel            *model.Channel
@@ -217,31 +218,41 @@ func (a *App) getPushNotificationMessage(contentsConfig, postMessage string, exp
 	return senderName + userLocale("api.post.send_notifications_and_forget.push_general_message")
 }
 
-func (a *App) clearPushNotificationSync(currentSessionId, userID, channelID string) *model.AppError {
+func (a *App) clearPushNotificationSync(currentSessionId, userID, channelID, rootID string) *model.AppError {
 	msg := &model.PushNotification{
 		Type:             model.PushTypeClear,
 		Version:          model.PushMessageV2,
 		ChannelId:        channelID,
+		RootId:           rootID,
 		ContentAvailable: 1,
+		Badge:            0,
+		IsCRTEnabled:     a.IsCRTEnabledForUser(userID),
 	}
 
 	unreadCount, err := a.Srv().Store.User().GetUnreadCount(userID)
 	if err != nil {
 		return model.NewAppError("clearPushNotificationSync", "app.user.get_unread_count.app_error", nil, err.Error(), http.StatusInternalServerError)
 	}
-
 	msg.Badge = int(unreadCount)
 
+	if msg.IsCRTEnabled {
+		data, err := a.Srv().Store.Thread().GetThreadsForUser(userID, "", model.GetUserThreadsOpts{TotalsOnly: true})
+		if err != nil {
+			return model.NewAppError("clearPushNotificationSync", "app.user.get_thread_count_for_user.app_error", nil, err.Error(), http.StatusInternalServerError)
+		}
+		msg.Badge += int(data.TotalUnreadMentions)
+	}
 	return a.sendPushNotificationToAllSessions(msg, userID, currentSessionId)
 }
 
-func (a *App) clearPushNotification(currentSessionId, userID, channelID string) {
+func (a *App) clearPushNotification(currentSessionId, userID, channelID, rootID string) {
 	select {
 	case a.Srv().PushNotificationsHub.notificationsChan <- PushNotification{
 		notificationType: notificationTypeClear,
 		currentSessionId: currentSessionId,
 		userID:           userID,
 		channelID:        channelID,
+		rootID:           rootID,
 	}:
 	case <-a.Srv().PushNotificationsHub.stopChan:
 		return
@@ -319,7 +330,7 @@ func (hub *PushNotificationsHub) start() {
 				var err *model.AppError
 				switch notification.notificationType {
 				case notificationTypeClear:
-					err = hub.app.clearPushNotificationSync(notification.currentSessionId, notification.userID, notification.channelID)
+					err = hub.app.clearPushNotificationSync(notification.currentSessionId, notification.userID, notification.channelID, notification.rootID)
 				case notificationTypeMessage:
 					err = hub.app.sendPushNotificationSync(
 						notification.post,
@@ -543,7 +554,7 @@ func (a *App) BuildPushNotificationMessage(contentsConfig string, post *model.Po
 	}
 
 	if contentsConfig == model.IdLoadedNotification {
-		msg = a.buildIdLoadedPushNotificationMessage(post, user)
+		msg = a.buildIdLoadedPushNotificationMessage(channel, post, user)
 	} else {
 		msg = a.buildFullPushNotificationMessage(contentsConfig, post, user, channel, channelName, senderName, explicitMention, channelWideMention, replyToThreadType)
 	}
@@ -557,17 +568,20 @@ func (a *App) BuildPushNotificationMessage(contentsConfig string, post *model.Po
 	return msg, nil
 }
 
-func (a *App) buildIdLoadedPushNotificationMessage(post *model.Post, user *model.User) *model.PushNotification {
+func (a *App) buildIdLoadedPushNotificationMessage(channel *model.Channel, post *model.Post, user *model.User) *model.PushNotification {
 	userLocale := i18n.GetUserTranslations(user.Locale)
 	msg := &model.PushNotification{
-		PostId:     post.Id,
+		PostId:       post.Id,
-		ChannelId:  post.ChannelId,
+		ChannelId:    post.ChannelId,
-		Category:   model.CategoryCanReply,
+		RootId:       post.RootId,
-		Version:    model.PushMessageV2,
+		IsCRTEnabled: a.IsCRTEnabledForUser(user.Id),
-		Type:       model.PushTypeMessage,
+		Category:     model.CategoryCanReply,
-		IsIdLoaded: true,
+		Version:      model.PushMessageV2,
-		SenderId:   user.Id,
+		TeamId:       channel.TeamId,
-		Message:    userLocale("api.push_notification.id_loaded.default_message"),
+		Type:         model.PushTypeMessage,
+		IsIdLoaded:   true,
+		SenderId:     user.Id,
+		Message:      userLocale("api.push_notification.id_loaded.default_message"),
 	}
 
 	return msg
@@ -577,15 +591,16 @@ func (a *App) buildFullPushNotificationMessage(contentsConfig string, post *mode
 	explicitMention bool, channelWideMention bool, replyToThreadType string) *model.PushNotification {
 
 	msg := &model.PushNotification{
-		Category:   model.CategoryCanReply,
+		Category:     model.CategoryCanReply,
-		Version:    model.PushMessageV2,
+		Version:      model.PushMessageV2,
-		Type:       model.PushTypeMessage,
+		Type:         model.PushTypeMessage,
-		TeamId:     channel.TeamId,
+		TeamId:       channel.TeamId,
-		ChannelId:  channel.Id,
+		ChannelId:    channel.Id,
-		PostId:     post.Id,
+		PostId:       post.Id,
-		RootId:     post.RootId,
+		RootId:       post.RootId,
-		SenderId:   post.UserId,
+		SenderId:     post.UserId,
-		IsIdLoaded: false,
+		IsCRTEnabled: false,
+		IsIdLoaded:   false,
 	}
 
 	userLocale := i18n.GetUserTranslations(user.Locale)
@@ -594,13 +609,16 @@ func (a *App) buildFullPushNotificationMessage(contentsConfig string, post *mode
 		msg.ChannelName = channelName
 	}
 
-	if a.isCRTEnabledForUser(user.Id) && post.RootId != "" {
+	if a.IsCRTEnabledForUser(user.Id) {
-		if contentsConfig != model.GenericNoChannelNotification {
+		msg.IsCRTEnabled = true
-			props := map[string]interface{}{"channelName": channelName}
+		if post.RootId != "" {
-			msg.ChannelName = userLocale("api.push_notification.title.collapsed_threads", props)
+			if contentsConfig != model.GenericNoChannelNotification {
+				props := map[string]interface{}{"channelName": channelName}
+				msg.ChannelName = userLocale("api.push_notification.title.collapsed_threads", props)
 
-			if channel.Type == model.ChannelTypeDirect {
+				if channel.Type == model.ChannelTypeDirect {
-				msg.ChannelName = userLocale("api.push_notification.title.collapsed_threads_dm")
+					msg.ChannelName = userLocale("api.push_notification.title.collapsed_threads_dm")
+				}
 			}
 		}
 	}

app/notification_push_test.go

@@ -1149,13 +1149,31 @@ func TestClearPushNotificationSync(t *testing.T) {
 		*cfg.EmailSettings.PushNotificationServer = pushServer.URL
 	})
 
-	err := th.App.clearPushNotificationSync(sess1.Id, "user1", "channel1")
+	err := th.App.clearPushNotificationSync(sess1.Id, "user1", "channel1", "")
 	require.Nil(t, err)
 	// Server side verification.
 	// We verify that 1 request has been sent, and also check the message contents.
 	require.Equal(t, 1, handler.numReqs())
 	assert.Equal(t, "channel1", handler.notifications()[0].ChannelId)
 	assert.Equal(t, model.PushTypeClear, handler.notifications()[0].Type)
+
+	// When CRT is enabled, Send badge count adding both "User unreads" + "User thread mentions"
+	th.App.UpdateConfig(func(cfg *model.Config) {
+		*cfg.ServiceSettings.ThreadAutoFollow = true
+		*cfg.ServiceSettings.CollapsedThreads = model.CollapsedThreadsDefaultOn
+	})
+
+	mockPreferenceStore := mocks.PreferenceStore{}
+	mockPreferenceStore.On("Get", mock.AnythingOfType("string"), model.PreferenceCategoryDisplaySettings, model.PreferenceNameCollapsedThreadsEnabled).Return(&model.Preference{Value: "on"}, nil)
+	mockStore.On("Preference").Return(&mockPreferenceStore)
+
+	mockThreadStore := mocks.ThreadStore{}
+	mockThreadStore.On("GetThreadsForUser", mock.AnythingOfType("string"), mock.AnythingOfType("string"), mock.Anything).Return(&model.Threads{TotalUnreadMentions: 3}, nil)
+	mockStore.On("Thread").Return(&mockThreadStore)
+
+	err = th.App.clearPushNotificationSync(sess1.Id, "user1", "channel1", "")
+	require.Nil(t, err)
+	assert.Equal(t, handler.notifications()[1].Badge, 4)
 }
 
 func TestUpdateMobileAppBadgeSync(t *testing.T) {
@@ -1337,7 +1355,7 @@ func TestAllPushNotifications(t *testing.T) {
 		case 2:
 			go func(sessID, userID string) {
 				defer wg.Done()
-				th.App.clearPushNotification(sessID, userID, th.BasicChannel.Id)
+				th.App.clearPushNotification(sessID, userID, th.BasicChannel.Id, "")
 			}(data.session.Id, data.user.Id)
 		}
 	}
@@ -1396,7 +1414,7 @@ func TestPushNotificationRace(t *testing.T) {
 
 		// Now we start sending messages after the PN hub is shut down.
 		// We test all 3 notification types.
-		app.clearPushNotification("currentSessionId", "userId", "channelId")
+		app.clearPushNotification("currentSessionId", "userId", "channelId", "")
 
 		app.UpdateMobileAppBadge("userId")
 
@@ -1565,7 +1583,7 @@ func BenchmarkPushNotificationThroughput(b *testing.B) {
 			case 2:
 				go func(sessID, userID string) {
 					defer wg.Done()
-					th.App.clearPushNotification(sessID, userID, ch.Id)
+					th.App.clearPushNotification(sessID, userID, ch.Id, "")
 				}(data.session.Id, data.user.Id)
 			}
 		}

app/opentracing/opentracing_layer.go

@@ -10787,6 +10787,23 @@ func (a *OpenTracingAppLayer) InviteNewUsersToTeamGracefully(emailList []string,
 	return resultVar0, resultVar1
 }
 
+func (a *OpenTracingAppLayer) IsCRTEnabledForUser(userID string) bool {
+	origCtx := a.ctx
+	span, newCtx := tracing.StartSpanWithParentByContext(a.ctx, "app.IsCRTEnabledForUser")
+
+	a.ctx = newCtx
+	a.app.Srv().Store.SetContext(newCtx)
+	defer func() {
+		a.app.Srv().Store.SetContext(origCtx)
+		a.ctx = origCtx
+	}()
+
+	defer span.Finish()
+	resultVar0 := a.app.IsCRTEnabledForUser(userID)
+
+	return resultVar0
+}
+
 func (a *OpenTracingAppLayer) IsFirstUserAccount() bool {
 	origCtx := a.ctx
 	span, newCtx := tracing.StartSpanWithParentByContext(a.ctx, "app.IsFirstUserAccount")
@@ -16580,7 +16597,7 @@ func (a *OpenTracingAppLayer) UpdateThreadFollowForUser(userID string, teamID st
 	return resultVar0
 }
 
-func (a *OpenTracingAppLayer) UpdateThreadReadForUser(userID string, teamID string, threadID string, timestamp int64) (*model.ThreadResponse, *model.AppError) {
+func (a *OpenTracingAppLayer) UpdateThreadReadForUser(currentSessionId string, userID string, teamID string, threadID string, timestamp int64) (*model.ThreadResponse, *model.AppError) {
 	origCtx := a.ctx
 	span, newCtx := tracing.StartSpanWithParentByContext(a.ctx, "app.UpdateThreadReadForUser")
 
@@ -16592,7 +16609,7 @@ func (a *OpenTracingAppLayer) UpdateThreadReadForUser(userID string, teamID stri
 	}()
 
 	defer span.Finish()
-	resultVar0, resultVar1 := a.app.UpdateThreadReadForUser(userID, teamID, threadID, timestamp)
+	resultVar0, resultVar1 := a.app.UpdateThreadReadForUser(currentSessionId, userID, teamID, threadID, timestamp)
 
 	if resultVar1 != nil {
 		span.LogFields(spanlog.Error(resultVar1))

app/post.go

@@ -66,7 +66,7 @@ func (a *App) CreatePostAsUser(c *request.Context, post *model.Post, currentSess
 	// the post is NOT a reply post with CRT enabled
 	_, fromWebhook := post.GetProps()["from_webhook"]
 	_, fromBot := post.GetProps()["from_bot"]
-	isCRTReply := post.RootId != "" && a.isCRTEnabledForUser(post.UserId)
+	isCRTReply := post.RootId != "" && a.IsCRTEnabledForUser(post.UserId)
 	if !fromWebhook && !fromBot && !isCRTReply {
 		if _, err := a.MarkChannelsAsViewed([]string{post.ChannelId}, post.UserId, currentSessionId, true); err != nil {
 			mlog.Warn(

app/user.go

@@ -2301,7 +2301,7 @@ func (a *App) UpdateThreadFollowForUser(userID, teamID, threadID string, state b
 	return nil
 }
 
-func (a *App) UpdateThreadReadForUser(userID, teamID, threadID string, timestamp int64) (*model.ThreadResponse, *model.AppError) {
+func (a *App) UpdateThreadReadForUser(currentSessionId, userID, teamID, threadID string, timestamp int64) (*model.ThreadResponse, *model.AppError) {
 	user, err := a.GetUser(userID)
 	if err != nil {
 		return nil, err
@@ -2339,6 +2339,11 @@ func (a *App) UpdateThreadReadForUser(userID, teamID, threadID string, timestamp
 		return nil, err
 	}
 
+	// Clear if user has read the messages
+	if thread.UnreadReplies == 0 && a.IsCRTEnabledForUser(userID) {
+		a.clearPushNotification(currentSessionId, userID, post.ChannelId, threadID)
+	}
+
 	message := model.NewWebSocketEvent(model.WebsocketEventThreadReadChanged, teamID, "", userID, nil)
 	message.Add("thread_id", threadID)
 	message.Add("timestamp", timestamp)

app/user_test.go

@@ -1547,7 +1547,7 @@ func TestUpdateThreadReadForUser(t *testing.T) {
 		require.Nil(t, appErr)
 		require.Zero(t, threads.Total)
 
-		_, appErr = th.App.UpdateThreadReadForUser(th.BasicUser.Id, th.BasicChannel.TeamId, rootPost.Id, replyPost.CreateAt)
+		_, appErr = th.App.UpdateThreadReadForUser("currentSessionId", th.BasicUser.Id, th.BasicChannel.TeamId, rootPost.Id, replyPost.CreateAt)
 		require.Nil(t, appErr)
 
 		threads, appErr = th.App.GetThreadsForUser(th.BasicUser.Id, th.BasicTeam.Id, model.GetUserThreadsOpts{})
@@ -1584,7 +1584,7 @@ func TestUpdateThreadReadForUser(t *testing.T) {
 		mockStore.On("User").Return(&mockUserStore)
 		mockStore.On("Thread").Return(&mockThreadStore)
 
-		_, err = th.App.UpdateThreadReadForUser("user1", "team1", "postid", 100)
+		_, err = th.App.UpdateThreadReadForUser("currentSessionId", "user1", "team1", "postid", 100)
 		require.Error(t, err)
 	})
 }

i18n/en.json

@@ -6371,6 +6371,10 @@
     "id": "app.user.get_recently_active_users.app_error",
     "translation": "We encountered an error while finding the recently active users."
   },
+  {
+    "id": "app.user.get_thread_count_for_user.app_error",
+    "translation": "We could not get thread count for the user."
+  },
   {
     "id": "app.user.get_thread_membership_for_user.app_error",
     "translation": "Unable to get user thread membership"

model/push_notification.go

@@ -64,6 +64,7 @@ type PushNotification struct {
 	OverrideIconURL  string `json:"override_icon_url,omitempty"`
 	FromWebhook      string `json:"from_webhook,omitempty"`
 	Version          string `json:"version,omitempty"`
+	IsCRTEnabled     bool   `json:"is_crt_enabled"`
 	IsIdLoaded       bool   `json:"is_id_loaded"`
 }