MM-35396 Refactor GetThreadForUser store func to take membership as argument to prevent replica lag issues and reduce joins in query (#17754)

* Update store function GetThreadForUser to use master DB to fix replica lag

* Refactor GetThreadForUser store func to take membership as argument to prevent replica lag issues and reduce joins in query

* Add translation

* Fix test

* Updates per feedback

* Minor clean-up per feedback

app/app_iface.go

@@ -759,7 +759,8 @@ type AppIface interface {
 	GetTeamsForUser(userID string) ([]*model.Team, *model.AppError)
 	GetTeamsUnreadForUser(excludeTeamId string, userID string) ([]*model.TeamUnread, *model.AppError)
 	GetTermsOfService(id string) (*model.TermsOfService, *model.AppError)
-	GetThreadForUser(userID, teamID, threadId string, extended bool) (*model.ThreadResponse, *model.AppError)
+	GetThreadForUser(teamID string, threadMembership *model.ThreadMembership, extended bool) (*model.ThreadResponse, *model.AppError)
+	GetThreadMembershipForUser(userId, threadId string) (*model.ThreadMembership, *model.AppError)
 	GetThreadMembershipsForUser(userID, teamID string) ([]*model.ThreadMembership, error)
 	GetThreadsForUser(userID, teamID string, options model.GetUserThreadsOpts) (*model.Threads, *model.AppError)
 	GetUploadSession(uploadId string) (*model.UploadSession, *model.AppError)

app/channel.go

@@ -2500,7 +2500,7 @@ func (a *App) MarkChannelAsUnreadFromPost(postID string, userID string, collapse
 				if nErr != nil {
 					return nil, model.NewAppError("MarkChannelAsUnreadFromPost", "app.channel.update_last_viewed_at_post.app_error", nil, nErr.Error(), http.StatusInternalServerError)
 				}
-				thread, _ := a.Srv().Store.Thread().GetThreadForUser(userID, channel.TeamId, threadId, true)
+				thread, _ := a.Srv().Store.Thread().GetThreadForUser(channel.TeamId, threadMembership, true)
 				a.sanitizeProfiles(thread.Participants, false)
 				thread.Post.SanitizeProps()
 
@@ -2598,11 +2598,11 @@ func (a *App) markChannelAsUnreadFromPostCRTUnsupported(postID string, userID st
 	if err != nil {
 		return nil, err
 	}
-	_, nErr = a.Srv().Store.Thread().UpdateMembership(threadMembership)
+	threadMembership, nErr = a.Srv().Store.Thread().UpdateMembership(threadMembership)
 	if nErr != nil {
 		return nil, model.NewAppError("MarkChannelAsUnreadFromPost", "app.channel.update_last_viewed_at_post.app_error", nil, nErr.Error(), http.StatusInternalServerError)
 	}
-	thread, nErr := a.Srv().Store.Thread().GetThreadForUser(userID, channel.TeamId, threadId, true)
+	thread, nErr := a.Srv().Store.Thread().GetThreadForUser(channel.TeamId, threadMembership, true)
 	if nErr != nil {
 		return nil, model.NewAppError("MarkChannelAsUnreadFromPost", "app.channel.update_last_viewed_at_post.app_error", nil, nErr.Error(), http.StatusInternalServerError)
 	}

app/notification.go

@@ -464,7 +464,11 @@ func (a *App) SendNotifications(post *model.Post, team *model.Team, channel *mod
 			}
 			if sendEvent {
 				message := model.NewWebSocketEvent(model.WEBSOCKET_EVENT_THREAD_UPDATED, team.Id, "", uid, nil)
-				userThread, err := a.Srv().Store.Thread().GetThreadForUser(uid, channel.TeamId, post.RootId, true)
+				threadMembership, err := a.Srv().Store.Thread().GetMembershipForUser(uid, post.RootId)
+				if err != nil {
+					return nil, errors.Wrapf(err, "cannot get thread membership %q for user %q", post.RootId, uid)
+				}
+				userThread, err := a.Srv().Store.Thread().GetThreadForUser(channel.TeamId, threadMembership, true)
 				if err != nil {
 					return nil, errors.Wrapf(err, "cannot get thread %q for user %q", post.RootId, uid)
 				}

app/opentracing/opentracing_layer.go

@@ -9294,7 +9294,7 @@ func (a *OpenTracingAppLayer) GetTermsOfService(id string) (*model.TermsOfServic
 	return resultVar0, resultVar1
 }
 
-func (a *OpenTracingAppLayer) GetThreadForUser(userID string, teamID string, threadId string, extended bool) (*model.ThreadResponse, *model.AppError) {
+func (a *OpenTracingAppLayer) GetThreadForUser(teamID string, threadMembership *model.ThreadMembership, extended bool) (*model.ThreadResponse, *model.AppError) {
 	origCtx := a.ctx
 	span, newCtx := tracing.StartSpanWithParentByContext(a.ctx, "app.GetThreadForUser")
 
@@ -9306,7 +9306,29 @@ func (a *OpenTracingAppLayer) GetThreadForUser(userID string, teamID string, thr
 	}()
 
 	defer span.Finish()
-	resultVar0, resultVar1 := a.app.GetThreadForUser(userID, teamID, threadId, extended)
+	resultVar0, resultVar1 := a.app.GetThreadForUser(teamID, threadMembership, extended)
+
+	if resultVar1 != nil {
+		span.LogFields(spanlog.Error(resultVar1))
+		ext.Error.Set(span, true)
+	}
+
+	return resultVar0, resultVar1
+}
+
+func (a *OpenTracingAppLayer) GetThreadMembershipForUser(userId string, threadId string) (*model.ThreadMembership, *model.AppError) {
+	origCtx := a.ctx
+	span, newCtx := tracing.StartSpanWithParentByContext(a.ctx, "app.GetThreadMembershipForUser")
+
+	a.ctx = newCtx
+	a.app.Srv().Store.SetContext(newCtx)
+	defer func() {
+		a.app.Srv().Store.SetContext(origCtx)
+		a.ctx = origCtx
+	}()
+
+	defer span.Finish()
+	resultVar0, resultVar1 := a.app.GetThreadMembershipForUser(userId, threadId)
 
 	if resultVar1 != nil {
 		span.LogFields(spanlog.Error(resultVar1))

app/post_test.go

@@ -1952,21 +1952,27 @@ func TestFollowThreadSkipsParticipants(t *testing.T) {
 	_, err = th.App.CreatePost(th.Context, &model.Post{RootId: p1.Id, UserId: user.Id, ChannelId: channel.Id, Message: "Hola"}, channel, false, false)
 	require.Nil(t, err)
 
-	thread, err := th.App.GetThreadForUser(user.Id, th.BasicTeam.Id, p1.Id, false)
+	threadMembership, err := th.App.GetThreadMembershipForUser(user.Id, p1.Id)
+	require.Nil(t, err)
+	thread, err := th.App.GetThreadForUser(th.BasicTeam.Id, threadMembership, false)
 	require.Nil(t, err)
 	require.Len(t, thread.Participants, 1) // length should be 1, the original poster, since sysadmin was just mentioned but didn't post
 
 	_, err = th.App.CreatePost(th.Context, &model.Post{RootId: p1.Id, UserId: sysadmin.Id, ChannelId: channel.Id, Message: "sysadmin reply"}, channel, false, false)
 	require.Nil(t, err)
 
-	thread, err = th.App.GetThreadForUser(user.Id, th.BasicTeam.Id, p1.Id, false)
+	threadMembership, err = th.App.GetThreadMembershipForUser(user.Id, p1.Id)
+	require.Nil(t, err)
+	thread, err = th.App.GetThreadForUser(th.BasicTeam.Id, threadMembership, false)
 	require.Nil(t, err)
 	require.Len(t, thread.Participants, 2) // length should be 2, the original poster and sysadmin, since sysadmin participated now
 
 	// another user follows the thread
 	th.App.UpdateThreadFollowForUser(user2.Id, th.BasicTeam.Id, p1.Id, true)
 
-	thread, err = th.App.GetThreadForUser(user2.Id, th.BasicTeam.Id, p1.Id, false)
+	threadMembership, err = th.App.GetThreadMembershipForUser(user2.Id, p1.Id)
+	require.Nil(t, err)
+	thread, err = th.App.GetThreadForUser(th.BasicTeam.Id, threadMembership, false)
 	require.Nil(t, err)
 	require.Len(t, thread.Participants, 2) // length should be 2, since follow shouldn't update participant list, only user1 and sysadmin are participants
 	for _, p := range thread.Participants {

app/user.go

@@ -2285,8 +2285,19 @@ func (a *App) GetThreadsForUser(userID, teamID string, options model.GetUserThre
 	return threads, nil
 }
 
-func (a *App) GetThreadForUser(userID, teamID, threadId string, extended bool) (*model.ThreadResponse, *model.AppError) {
-	thread, err := a.Srv().Store.Thread().GetThreadForUser(userID, teamID, threadId, extended)
+func (a *App) GetThreadMembershipForUser(userId, threadId string) (*model.ThreadMembership, *model.AppError) {
+	threadMembership, err := a.Srv().Store.Thread().GetMembershipForUser(userId, threadId)
+	if err != nil {
+		return nil, model.NewAppError("GetThreadMembershipForUser", "app.user.get_thread_membership_for_user.app_error", nil, err.Error(), http.StatusInternalServerError)
+	}
+	if threadMembership == nil {
+		return nil, model.NewAppError("GetThreadMembershipForUser", "app.user.get_thread_membership_for_user.not_found", nil, "thread membership not found/followed", http.StatusNotFound)
+	}
+	return threadMembership, nil
+}
+
+func (a *App) GetThreadForUser(teamID string, threadMembership *model.ThreadMembership, extended bool) (*model.ThreadResponse, *model.AppError) {
+	thread, err := a.Srv().Store.Thread().GetThreadForUser(teamID, threadMembership, extended)
 	if err != nil {
 		return nil, model.NewAppError("GetThreadForUser", "app.user.get_threads_for_user.app_error", nil, err.Error(), http.StatusInternalServerError)
 	}
@@ -2349,14 +2360,16 @@ func (a *App) UpdateThreadReadForUser(userID, teamID, threadID string, timestamp
 		return nil, model.NewAppError("UpdateThreadsReadForUser", "app.user.update_threads_read_for_user.app_error", nil, nErr.Error(), http.StatusInternalServerError)
 	}
 
+	membership.LastViewed = timestamp
 	nErr = a.Srv().Store.Thread().MarkAsRead(userID, threadID, timestamp)
 	if nErr != nil {
 		return nil, model.NewAppError("UpdateThreadReadForUser", "app.user.update_thread_read_for_user.app_error", nil, nErr.Error(), http.StatusInternalServerError)
 	}
-	thread, err := a.GetThreadForUser(userID, teamID, threadID, false)
+	thread, err := a.GetThreadForUser(teamID, membership, false)
 	if err != nil {
 		return nil, err
 	}
+
 	message := model.NewWebSocketEvent(model.WEBSOCKET_EVENT_THREAD_READ_CHANGED, teamID, "", userID, nil)
 	message.Add("thread_id", threadID)
 	message.Add("timestamp", timestamp)