MM-56548: [AI assisted]Add support for incremental thread loading using UpdateAt timestamp (#30486)

Every time we load the RHS, we used to load the FULL thread always. Although the actual ThreadViewer React component is virtualized, and the server side API call is paginated, we still went through all the pages, to get the full thread and passed it on to the ThreadViewer. This would be for first loads, and subsequent loads of the same thread. This was a bug originally, but then it was a necessity after we applied websocket event scope because now we won't get emoji reactions of a thread if the user is not on the thread. To fix that, we enhance the thread loading functionality by adding support for fetching thread updates based on the UpdateAt timestamp. Now, for subsequent loads, we only get the changed posts in a thread. The implementation: - Adds new API parameters: fromUpdateAt and updatesOnly to the GetPostThread endpoint - Updates database queries to support sorting and filtering by UpdateAt - Implements thread state management to track the last update timestamp - Adds client-side support to use incremental loading for improved performance - Ensures proper validation for parameter combinations and error handling This change enables more efficient thread loading, particularly for long threads with frequent updates, by only fetching posts that have been updated since the last view. Caveats: For delta updates, the SQL query won't use the best index possible because we have an index for (CreateAt, Id), but no index for (UpdateAt, Id). However, from my tests, it is not as bad as it looks: ``` [loadtest] # EXPLAIN (ANALYZE, BUFFERS) SELECT * FROM Posts WHERE Posts.DeleteAt = 0 AND Posts.RootId = 'qbr5gctu9iyg8c36hpcq6f3w8e' AND Posts.UpdateAt > 1623445795824 ORDER BY UpdateAt ASC, Id ASC LIMIT 61; QUERY PLAN ------------------------------------------------------------------------------------------------------------------------------------------------- Limit (cost=8.31..8.31 rows=1 width=216) (actual time=0.047..0.049 rows=0 loops=1) Buffers: shared hit=2 -> Sort (cost=8.31..8.31 rows=1 width=216) (actual time=0.044..0.045 rows=0 loops=1) Sort Key: updateat, id Sort Method: quicksort Memory: 25kB Buffers: shared hit=2 -> Index Scan using idx_posts_root_id_delete_at on posts (cost=0.28..8.30 rows=1 width=216) (actual time=0.031..0.032 rows=0 loops=1) Index Cond: (((rootid)::text = 'qbr5gctu9iyg8c36hpcq6f3w8e'::text) AND (deleteat = 0)) Filter: (updateat > '1623445795824'::bigint) Buffers: shared hit=2 Planning: Buffers: shared hit=3 Planning Time: 0.508 ms Execution Time: 0.106 ms (14 rows) ``` We still get an index scan with index cond. Although there's a filter element, but atleast we get the whole thread with the index. My thinking is that while the whole thread might be large, but after that, updates on a thread should be incremental. Therefore, we should be okay without adding yet another index on the posts table. This is just the first step in what could be potentially improved further. 1. We shouldn't even be loading the full thread always. But rather let the virtualized viewer load more posts on demand. 2. If a post has been just reacted to, then we need not send the whole post down, but just the reaction. This further saves bandwidth. https://mattermost.atlassian.net/browse/MM-56548 TBD: Add load-test coverage to update the thread loading code ```release-note NONE ``` --------- Co-authored-by: Mattermost Build <build@mattermost.com>
2025-04-22 10:43:13 +05:30
--- a/server/channels/api4/post.go
+++ b/server/channels/api4/post.go
@@ -671,6 +671,27 @@ func getPostThread(c *Context, w http.ResponseWriter, r *http.Request) {
 		return
 	}

+	var fromUpdateAt int64
+	if fromUpdateAtStr := r.URL.Query().Get("fromUpdateAt"); fromUpdateAtStr != "" {
+		var err error
+		fromUpdateAt, err = strconv.ParseInt(fromUpdateAtStr, 10, 64)
+		if err != nil {
+			c.SetInvalidParamWithErr("fromUpdateAt", err)
+			return
+		}
+	}
+
+	if fromUpdateAt != 0 && fromCreateAt != 0 {
+		c.SetInvalidParamWithDetails("fromUpdateAt", "both fromUpdateAt and fromCreateAt cannot be set")
+		return
+	}
+
+	updatesOnly := r.URL.Query().Get("updatesOnly") == "true"
+	if updatesOnly && fromUpdateAt == 0 {
+		c.SetInvalidParamWithDetails("fromUpdateAt", "fromUpdateAt must be set if updatesOnly is set")
+		return
+	}
+
 	direction := ""
 	if dir := r.URL.Query().Get("direction"); dir != "" {
 		if dir != "up" && dir != "down" {
@@ -679,14 +700,22 @@ func getPostThread(c *Context, w http.ResponseWriter, r *http.Request) {
 		}
 		direction = dir
 	}
+
+	if updatesOnly && direction == "up" {
+		c.SetInvalidParamWithDetails("updatesOnly", "updatesOnly flag cannot be used with up direction")
+		return
+	}
+
 	opts := model.GetPostsOptions{
 		SkipFetchThreads:         r.URL.Query().Get("skipFetchThreads") == "true",
 		CollapsedThreads:         r.URL.Query().Get("collapsedThreads") == "true",
 		CollapsedThreadsExtended: r.URL.Query().Get("collapsedThreadsExtended") == "true",
+		UpdatesOnly:              updatesOnly,
 		PerPage:                  perPage,
 		Direction:                direction,
 		FromPost:                 fromPost,
 		FromCreateAt:             fromCreateAt,
+		FromUpdateAt:             fromUpdateAt,
 	}
 	list, err := c.App.GetPostThread(c.Params.PostId, opts, c.AppContext.Session().UserId)
 	if err != nil {
--- a/server/channels/api4/post_test.go
+++ b/server/channels/api4/post_test.go
@@ -3638,7 +3638,60 @@ func TestGetPostThread(t *testing.T) {
 	require.Error(t, err)
 	CheckForbiddenStatus(t, resp)

+	// Test the new query parameters - updatesOnly, fromUpdateAt
 	// Sending some bad params
+	_, resp, err = client.GetPostThreadWithOpts(context.Background(), th.BasicPost.Id, "", model.GetPostsOptions{
+		UpdatesOnly: true, // updatesOnly is true but fromUpdateAt is not set
+	})
+	require.Error(t, err)
+	CheckBadRequestStatus(t, resp)
+
+	// Test error when both fromUpdateAt and fromCreateAt are set
+	_, resp, err = client.GetPostThreadWithOpts(context.Background(), th.BasicPost.Id, "", model.GetPostsOptions{
+		FromUpdateAt: 12345,
+		FromCreateAt: 12345,
+	})
+	require.Error(t, err)
+	CheckBadRequestStatus(t, resp)
+
+	// Test error when updatesOnly is used with direction="up"
+	_, resp, err = client.GetPostThreadWithOpts(context.Background(), th.BasicPost.Id, "", model.GetPostsOptions{
+		UpdatesOnly:  true,
+		FromUpdateAt: 12345,
+		Direction:    "up",
+	})
+	require.Error(t, err)
+	CheckBadRequestStatus(t, resp)
+
+	// Test valid parameters
+	// This should work with proper parameters
+	_, resp, err = client.GetPostThreadWithOpts(context.Background(), th.BasicPost.Id, "", model.GetPostsOptions{
+		UpdatesOnly:  true,
+		FromUpdateAt: 12345,
+		Direction:    "down",
+	})
+	require.NoError(t, err)
+	CheckOKStatus(t, resp)
+
+	list, resp, err = client.GetPostThreadWithOpts(context.Background(), th.BasicPost.Id, "", model.GetPostsOptions{
+		UpdatesOnly:  true,
+		Direction:    "down",
+		FromUpdateAt: post.UpdateAt,
+	})
+	require.NoError(t, err)
+	CheckOKStatus(t, resp)
+	assert.Len(t, list.Order, 1)
+	assert.Len(t, list.Posts, 1)
+	require.Equal(t, th.BasicPost.Id, list.Order[0], "wrong order")
+
+	// Test with just fromUpdateAt parameter
+	_, resp, err = client.GetPostThreadWithOpts(context.Background(), th.BasicPost.Id, "", model.GetPostsOptions{
+		FromUpdateAt: 12345,
+	})
+	require.NoError(t, err)
+	CheckOKStatus(t, resp)
+
+	// Sending other bad params unrelated to the new changes
 	_, resp, err = client.GetPostThreadWithOpts(context.Background(), th.BasicPost.Id, "", model.GetPostsOptions{
 		CollapsedThreads: true,
 		FromPost:         "something",
--- a/server/channels/store/sqlstore/post_store.go
+++ b/server/channels/store/sqlstore/post_store.go
@@ -627,36 +627,50 @@ func (s *SqlPostStore) getPostWithCollapsedThreads(id, userID string, opts model
 		}
 	}
 	if sort != "" {
-		query = query.OrderBy("CreateAt " + sort + ", Id " + sort)
+		if opts.UpdatesOnly {
+			query = query.OrderBy("UpdateAt " + sort + ", Id " + sort)
+		} else {
+			query = query.OrderBy("CreateAt " + sort + ", Id " + sort)
+		}
 	}

 	if opts.FromCreateAt != 0 {
+		var direction sq.Sqlizer
+		var pagination sq.Sqlizer
+
 		if opts.Direction == "down" {
-			direction := sq.Gt{"Posts.CreateAt": opts.FromCreateAt}
-			if opts.FromPost != "" {
-				query = query.Where(sq.Or{
-					direction,
-					sq.And{
-						sq.Eq{"Posts.CreateAt": opts.FromCreateAt},
-						sq.Gt{"Posts.Id": opts.FromPost},
-					},
-				})
-			} else {
-				query = query.Where(direction)
-			}
+			direction = sq.Gt{"Posts.CreateAt": opts.FromCreateAt}
+			pagination = sq.Gt{"Posts.Id": opts.FromPost}
 		} else {
-			direction := sq.Lt{"Posts.CreateAt": opts.FromCreateAt}
-			if opts.FromPost != "" {
-				query = query.Where(sq.Or{
-					direction,
-					sq.And{
-						sq.Eq{"Posts.CreateAt": opts.FromCreateAt},
-						sq.Lt{"Posts.Id": opts.FromPost},
-					},
-				})
-			} else {
-				query = query.Where(direction)
-			}
+			direction = sq.Lt{"Posts.CreateAt": opts.FromCreateAt}
+			pagination = sq.Lt{"Posts.Id": opts.FromPost}
+		}
+
+		if opts.FromPost != "" {
+			query = query.Where(sq.Or{
+				direction,
+				sq.And{
+					sq.Eq{"Posts.CreateAt": opts.FromCreateAt},
+					pagination,
+				},
+			})
+		} else {
+			query = query.Where(direction)
+		}
+	}
+
+	if opts.FromUpdateAt != 0 && opts.Direction == "down" {
+		direction := sq.Gt{"Posts.UpdateAt": opts.FromUpdateAt}
+		if opts.FromPost != "" {
+			query = query.Where(sq.Or{
+				direction,
+				sq.And{
+					sq.Eq{"Posts.UpdateAt": opts.FromUpdateAt},
+					sq.Gt{"Posts.Id": opts.FromPost},
+				},
+			})
+		} else {
+			query = query.Where(direction)
 		}
 	}

@@ -773,7 +787,11 @@ func (s *SqlPostStore) Get(ctx context.Context, id string, opts model.GetPostsOp
 			}
 		}
 		if sort != "" {
-			query = query.OrderBy("CreateAt " + sort + ", Id " + sort)
+			if opts.UpdatesOnly {
+				query = query.OrderBy("UpdateAt " + sort + ", Id " + sort)
+			} else {
+				query = query.OrderBy("CreateAt " + sort + ", Id " + sort)
+			}
 		}

 		if opts.FromCreateAt != 0 {
@@ -806,6 +824,36 @@ func (s *SqlPostStore) Get(ctx context.Context, id string, opts model.GetPostsOp
 			}
 		}

+		if opts.FromUpdateAt != 0 {
+			if opts.Direction == "down" {
+				direction := sq.Gt{"p.UpdateAt": opts.FromUpdateAt}
+				if opts.FromPost != "" {
+					query = query.Where(sq.Or{
+						direction,
+						sq.And{
+							sq.Eq{"p.UpdateAt": opts.FromUpdateAt},
+							sq.Gt{"p.Id": opts.FromPost},
+						},
+					})
+				} else {
+					query = query.Where(direction)
+				}
+			} else {
+				direction := sq.Lt{"p.UpdateAt": opts.FromUpdateAt}
+				if opts.FromPost != "" {
+					query = query.Where(sq.Or{
+						direction,
+						sq.And{
+							sq.Eq{"p.UpdateAt": opts.FromUpdateAt},
+							sq.Lt{"p.Id": opts.FromPost},
+						},
+					})
+				} else {
+					query = query.Where(direction)
+				}
+			}
+		}
+
 		if opts.PerPage != 0 {
 			query = query.Limit(uint64(opts.PerPage + 1))
 		}
--- a/server/channels/store/storetest/post_store.go
+++ b/server/channels/store/storetest/post_store.go
@@ -838,7 +838,7 @@ func testPostStoreGetForThread(t *testing.T, rctx request.CTX, ss store.Store) {
 		}
 		r1, err = ss.Post().Get(context.Background(), o1.Id, opts, o1.UserId, map[string]bool{})
 		require.NoError(t, err)
-		require.Len(t, r1.Order, 2) // including the root post
+		require.Len(t, r1.Order, 2)
 		require.Len(t, r1.Posts, 2)
 		assert.True(t, *r1.HasNext)

@@ -876,7 +876,7 @@ func testPostStoreGetForThread(t *testing.T, rctx request.CTX, ss store.Store) {
 		}
 		r1, err = ss.Post().Get(context.Background(), o1.Id, opts, o1.UserId, map[string]bool{})
 		require.NoError(t, err)
-		require.Len(t, r1.Order, 2) // including the root post
+		require.Len(t, r1.Order, 2)
 		require.Len(t, r1.Posts, 2)
 		assert.LessOrEqual(t, r1.Posts[r1.Order[1]].CreateAt, firstPostCreateAt)
 		assert.False(t, *r1.HasNext)
@@ -896,6 +896,126 @@ func testPostStoreGetForThread(t *testing.T, rctx request.CTX, ss store.Store) {
 		assert.GreaterOrEqual(t, r1.Posts[r1.Order[1]].CreateAt, m1.CreateAt)
 		assert.True(t, *r1.HasNext)
 	})
+
+	t.Run("Pagination with UpdateAt", func(t *testing.T) {
+		teamID := model.NewId()
+		channel, err := ss.Channel().Save(rctx, &model.Channel{
+			TeamId:      teamID,
+			DisplayName: "DisplayName1",
+			Name:        "channel" + model.NewId(),
+			Type:        model.ChannelTypeOpen,
+		}, -1)
+		require.NoError(t, err)
+
+		now := model.GetMillis()
+		o1 := &model.Post{CreateAt: now, ChannelId: channel.Id, UserId: model.NewId(), Message: NewTestID()}
+		o1, err = ss.Post().Save(rctx, o1)
+		require.NoError(t, err)
+
+		// Create replies with explicit UpdateAt timestamps
+		o2 := &model.Post{CreateAt: now + 1, UpdateAt: now + 1, ChannelId: o1.ChannelId, UserId: model.NewId(), Message: NewTestID(), RootId: o1.Id}
+		_, err = ss.Post().Save(rctx, o2)
+		require.NoError(t, err)
+
+		m1 := &model.Post{CreateAt: now + 2, UpdateAt: now + 2, ChannelId: o1.ChannelId, UserId: model.NewId(), Message: NewTestID(), RootId: o1.Id}
+		m1, err = ss.Post().Save(rctx, m1)
+		require.NoError(t, err)
+
+		o3 := &model.Post{CreateAt: now + 3, UpdateAt: now + 3, ChannelId: o1.ChannelId, UserId: model.NewId(), Message: NewTestID(), RootId: o1.Id}
+		_, err = ss.Post().Save(rctx, o3)
+		require.NoError(t, err)
+
+		o4 := &model.Post{CreateAt: now + 4, UpdateAt: now + 4, ChannelId: o1.ChannelId, UserId: model.NewId(), Message: NewTestID(), RootId: o1.Id}
+		o4, err = ss.Post().Save(rctx, o4)
+		require.NoError(t, err)
+
+		// Test pagination with UpdateAt in "down" direction
+		opts := model.GetPostsOptions{
+			UpdatesOnly:      true,
+			CollapsedThreads: true,
+			PerPage:          2,
+			Direction:        "down",
+		}
+		r1, err := ss.Post().Get(context.Background(), o1.Id, opts, o1.UserId, map[string]bool{})
+		require.NoError(t, err)
+		require.Len(t, r1.Order, 3) // including the root post
+		require.Len(t, r1.Posts, 3)
+		assert.Equal(t, r1.Posts[r1.Order[0]].UpdateAt, o4.CreateAt) // The root post always get updated with the createAt of the latest post in the thread.
+		assert.True(t, *r1.HasNext)
+
+		lastPostID := r1.Order[len(r1.Order)-1]
+		lastPostUpdateAt := r1.Posts[lastPostID].UpdateAt
+
+		// Continue pagination using UpdateAt
+		opts = model.GetPostsOptions{
+			UpdatesOnly:      true,
+			CollapsedThreads: true,
+			PerPage:          2,
+			Direction:        "down",
+			FromPost:         lastPostID,
+			FromUpdateAt:     lastPostUpdateAt,
+		}
+		r1, err = ss.Post().Get(context.Background(), o1.Id, opts, o1.UserId, map[string]bool{})
+		require.NoError(t, err)
+		require.Len(t, r1.Order, 3) // including the root post
+		require.Len(t, r1.Posts, 3)
+		assert.GreaterOrEqual(t, r1.Posts[r1.Order[len(r1.Order)-1]].UpdateAt, lastPostUpdateAt)
+		assert.Equal(t, r1.Posts[r1.Order[0]].UpdateAt, o4.CreateAt) // The root post always get updated with the createAt of the latest post in the thread.
+		assert.False(t, *r1.HasNext)
+
+		// Non-CRT mode with UpdateAt pagination
+		opts = model.GetPostsOptions{
+			UpdatesOnly:      true,
+			CollapsedThreads: false,
+			PerPage:          2,
+			Direction:        "down",
+			SkipFetchThreads: false,
+		}
+		r1, err = ss.Post().Get(context.Background(), o1.Id, opts, o1.UserId, map[string]bool{})
+		require.NoError(t, err)
+		// Ordering by updateAt will move the root post down, so we will get more posts in the thread.
+		require.Len(t, r1.Order, 3)
+		require.Len(t, r1.Posts, 3)
+		require.True(t, *r1.HasNext)
+
+		lastPostID = r1.Order[len(r1.Order)-1]
+		lastPostUpdateAt = r1.Posts[lastPostID].UpdateAt
+
+		opts = model.GetPostsOptions{
+			UpdatesOnly:      true,
+			CollapsedThreads: false,
+			PerPage:          3,
+			Direction:        "down",
+			FromPost:         lastPostID,
+			FromUpdateAt:     lastPostUpdateAt,
+			SkipFetchThreads: false,
+		}
+		r1, err = ss.Post().Get(context.Background(), o1.Id, opts, o1.UserId, map[string]bool{})
+		require.NoError(t, err)
+		require.Len(t, r1.Order, 3)
+		require.Len(t, r1.Posts, 3)
+		require.Equal(t, r1.Posts[r1.Order[0]].ReplyCount, int64(4))
+		require.Equal(t, r1.Posts[r1.Order[1]].ReplyCount, int64(4))
+		require.Equal(t, r1.Posts[r1.Order[2]].ReplyCount, int64(4))
+		require.GreaterOrEqual(t, r1.Posts[r1.Order[len(r1.Order)-1]].UpdateAt, lastPostUpdateAt)
+		assert.False(t, *r1.HasNext)
+
+		// Only with UpdateAt - direction down
+		opts = model.GetPostsOptions{
+			UpdatesOnly:      true,
+			CollapsedThreads: false,
+			PerPage:          1,
+			Direction:        "down",
+			FromUpdateAt:     m1.UpdateAt,
+			SkipFetchThreads: false,
+		}
+		r1, err = ss.Post().Get(context.Background(), o1.Id, opts, o1.UserId, map[string]bool{})
+		require.NoError(t, err)
+		require.Len(t, r1.Order, 2)
+		require.Len(t, r1.Posts, 2)
+		require.GreaterOrEqual(t, r1.Posts[r1.Order[1]].UpdateAt, m1.UpdateAt)
+		require.True(t, *r1.HasNext)
+	})
 }

 func testPostStoreGetSingle(t *testing.T, rctx request.CTX, ss store.Store) {
--- a/server/public/model/client4.go
+++ b/server/public/model/client4.go
@@ -4168,6 +4168,9 @@ func (c *Client4) GetPostThreadWithOpts(ctx context.Context, postID string, etag
 	if opts.SkipFetchThreads {
 		values.Set("skipFetchThreads", "true")
 	}
+	if opts.UpdatesOnly {
+		values.Set("updatesOnly", "true")
+	}
 	if opts.PerPage != 0 {
 		values.Set("perPage", strconv.Itoa(opts.PerPage))
 	}
@@ -4177,6 +4180,9 @@ func (c *Client4) GetPostThreadWithOpts(ctx context.Context, postID string, etag
 	if opts.FromCreateAt != 0 {
 		values.Set("fromCreateAt", strconv.FormatInt(opts.FromCreateAt, 10))
 	}
+	if opts.FromUpdateAt != 0 {
+		values.Set("fromUpdateAt", strconv.FormatInt(opts.FromUpdateAt, 10))
+	}
 	if opts.Direction != "" {
 		values.Set("direction", opts.Direction)
 	}
--- a/server/public/model/post.go
+++ b/server/public/model/post.go
@@ -403,7 +403,9 @@ type GetPostsOptions struct {
 	CollapsedThreadsExtended bool
 	FromPost                 string // PostId after which to send the items
 	FromCreateAt             int64  // CreateAt after which to send the items
+	FromUpdateAt             int64  // UpdateAt after which to send the items. This cannot be used with FromCreateAt.
 	Direction                string // Only accepts up|down. Indicates the order in which to send the items.
+	UpdatesOnly              bool   // This flag is used to make the API work with the updateAt value.
 	IncludeDeleted           bool
 	IncludePostPriority      bool
 }