[MM-45868] Add teamId to Threads table (#20915)

* Add teamId to Threads table

* Get rid of multiple teamId reads

* Fix failed test

* Add teamId to standard queries

* Fix linter

* Get teamId from db

db/migrations/migrations.list

@@ -186,6 +186,8 @@ db/migrations/mysql/000092_add_createat_to_teammembers.down.sql
 db/migrations/mysql/000092_add_createat_to_teammembers.up.sql
 db/migrations/mysql/000093_notify_admin.down.sql
 db/migrations/mysql/000093_notify_admin.up.sql
+db/migrations/mysql/000094_threads_teamid.down.sql
+db/migrations/mysql/000094_threads_teamid.up.sql
 db/migrations/postgres/000001_create_teams.down.sql
 db/migrations/postgres/000001_create_teams.up.sql
 db/migrations/postgres/000002_create_team_members.down.sql
@@ -372,3 +374,5 @@ db/migrations/postgres/000092_add_createat_to_teamembers.down.sql
 db/migrations/postgres/000092_add_createat_to_teamembers.up.sql
 db/migrations/postgres/000093_notify_admin.down.sql
 db/migrations/postgres/000093_notify_admin.up.sql
+db/migrations/postgres/000094_threads_teamid.down.sql
+db/migrations/postgres/000094_threads_teamid.up.sql

db/migrations/mysql/000094_threads_teamid.down.sql

@@ -0,0 +1,14 @@
+SET @preparedStatement = (SELECT IF(
+    EXISTS(
+        SELECT 1 FROM INFORMATION_SCHEMA.STATISTICS
+        WHERE table_name = 'Threads'
+        AND table_schema = DATABASE()
+        AND column_name = 'TeamId'
+    ),
+    'ALTER TABLE Threads DROP COLUMN TeamId;',
+    'SELECT 1;'
+));
+
+PREPARE removeColumnIfExists FROM @preparedStatement;
+EXECUTE removeColumnIfExists;
+DEALLOCATE PREPARE removeColumnIfExists;

db/migrations/mysql/000094_threads_teamid.up.sql

@@ -0,0 +1,19 @@
+SET @preparedStatement = (SELECT IF(
+    NOT EXISTS(
+        SELECT 1 FROM INFORMATION_SCHEMA.COLUMNS
+        WHERE table_name = 'Threads'
+        AND table_schema = DATABASE()
+        AND column_name = 'TeamId'
+    ),
+    'ALTER TABLE Threads ADD COLUMN TeamId varchar(26) DEFAULT NULL;',
+    'SELECT 1;'
+));
+
+PREPARE addColumnIfNotExists FROM @preparedStatement;
+EXECUTE addColumnIfNotExists;
+DEALLOCATE PREPARE addColumnIfNotExists;
+
+UPDATE Threads, Channels
+SET Threads.TeamId = Channels.TeamId
+WHERE Channels.Id = Threads.ChannelId
+AND Threads.TeamId IS NULL;

db/migrations/postgres/000094_threads_teamid.down.sql

@@ -0,0 +1 @@
+ALTER TABLE threads DROP COLUMN IF EXISTS teamid;

db/migrations/postgres/000094_threads_teamid.up.sql

@@ -0,0 +1,2 @@
+ALTER TABLE threads ADD COLUMN IF NOT EXISTS teamid VARCHAR(26);
+UPDATE threads SET teamid = channels.teamid FROM channels WHERE threads.teamid IS NULL AND channels.id = threads.channelid;

model/thread.go

@@ -26,6 +26,9 @@ type Thread struct {
 	// DeleteAt is a denormalized copy of the root posts's DeleteAt. In the database, it's
 	// named ThreadDeleteAt to avoid introducing a query conflict with older server versions.
 	DeleteAt int64 `json:"delete_at"`
+
+	// TeamId is a denormalized copy of the Channel's teamId.
+	TeamId string `json:"team_id"`
 }
 
 type ThreadResponse struct {

store/sqlstore/integrity.go

@@ -468,6 +468,7 @@ func checkPostsIntegrity(ss *SqlStore, results chan<- model.IntegrityCheckResult
 	results <- checkPostsFileInfoIntegrity(ss)
 	results <- checkPostsPostsRootIdIntegrity(ss)
 	results <- checkPostsReactionsIntegrity(ss)
+	results <- checkThreadsTeamsIntegrity(ss)
 }
 
 func checkSchemesIntegrity(ss *SqlStore, results chan<- model.IntegrityCheckResult) {
@@ -511,6 +512,16 @@ func checkUsersIntegrity(ss *SqlStore, results chan<- model.IntegrityCheckResult
 	results <- checkUsersUserAccessTokensIntegrity(ss)
 }
 
+func checkThreadsTeamsIntegrity(ss *SqlStore) model.IntegrityCheckResult {
+	return checkParentChildIntegrity(ss, relationalCheckConfig{
+		parentName:         "Teams",
+		parentIdAttr:       "TeamId",
+		childName:          "Threads",
+		childIdAttr:        "PostId",
+		canParentIdBeEmpty: false,
+	})
+}
+
 func CheckRelationalIntegrity(ss *SqlStore, results chan<- model.IntegrityCheckResult) {
 	mlog.Info("Starting relational integrity checks...")
 	checkChannelsIntegrity(ss, results)

store/sqlstore/integrity_test.go

@@ -650,9 +650,10 @@ func TestCheckPostsPostsRootIdIntegrity(t *testing.T) {
 		})
 
 		t.Run("should generate a report with one record", func(t *testing.T) {
-			root := createPost(ss, model.NewId(), model.NewId(), "", "")
+			channel := createChannel(ss, model.NewId(), model.NewId())
+			root := createPost(ss, channel.Id, model.NewId(), "", "")
 			rootId := root.Id
-			post := createPost(ss, model.NewId(), model.NewId(), root.Id, root.Id)
+			post := createPost(ss, channel.Id, model.NewId(), root.Id, root.Id)
 			dbmap.Exec(`DELETE FROM Posts WHERE Id=?`, root.Id)
 			result := checkPostsPostsRootIdIntegrity(store)
 			require.NoError(t, result.Err)
@@ -663,6 +664,8 @@ func TestCheckPostsPostsRootIdIntegrity(t *testing.T) {
 				ChildId:  &post.Id,
 			}, data.Records[0])
 			dbmap.Exec(`DELETE FROM Posts WHERE Id=?`, post.Id)
+			dbmap.Exec(`DELETE FROM Channels WHERE Id=?`, channel.Id)
+			dbmap.Exec(`DELETE FROM Threads WHERE PostId=?`, rootId)
 		})
 	})
 }
@@ -1602,3 +1605,39 @@ func TestCheckUsersUserAccessTokensIntegrity(t *testing.T) {
 		})
 	})
 }
+
+func TestCheckThreadsTeamsIntegrity(t *testing.T) {
+	StoreTest(t, func(t *testing.T, ss store.Store) {
+		store := ss.(*SqlStore)
+		dbmap := store.GetMasterX()
+
+		t.Run("should generate a report with no records", func(t *testing.T) {
+			result := checkThreadsTeamsIntegrity(store)
+			require.NoError(t, result.Err)
+			data := result.Data.(model.RelationalIntegrityCheckData)
+			require.Empty(t, data.Records)
+		})
+
+		t.Run("should generate a report with one record", func(t *testing.T) {
+			team := createTeam(ss)
+			channel := createChannel(ss, team.Id, model.NewId())
+			root := createPost(ss, channel.Id, model.NewId(), "", "")
+			post := createPost(ss, channel.Id, model.NewId(), root.Id, root.Id)
+
+			dbmap.Exec(`DELETE FROM Teams WHERE Id=?`, team.Id)
+			result := checkThreadsTeamsIntegrity(store)
+			require.NoError(t, result.Err)
+			data := result.Data.(model.RelationalIntegrityCheckData)
+			require.Len(t, data.Records, 1)
+
+			require.Equal(t, model.OrphanedRecord{
+				ParentId: &team.Id,
+				ChildId:  &root.Id,
+			}, data.Records[0])
+			dbmap.Exec(`DELETE FROM Posts WHERE Id=?`, post.Id)
+			dbmap.Exec(`DELETE FROM Posts WHERE Id=?`, root.Id)
+			dbmap.Exec(`DELETE FROM Channels WHERE Id=?`, channel.Id)
+			dbmap.Exec(`DELETE FROM Threads WHERE PostId=?`, root.Id)
+		})
+	})
+}

store/sqlstore/post_store.go

@@ -2948,7 +2948,8 @@ func (s *SqlPostStore) updateThreadsFromPosts(transaction *sqlxTxWrapper, posts
 	}
 
 	threadsByRoots := []*model.Thread{}
-	if err := transaction.Select(&threadsByRoots, threadsByRootsSql, threadsByRootsArgs...); err != nil {
+	err = transaction.Select(&threadsByRoots, threadsByRootsSql, threadsByRootsArgs...)
+	if err != nil {
 		return err
 	}
 
@@ -2957,6 +2958,8 @@ func (s *SqlPostStore) updateThreadsFromPosts(transaction *sqlxTxWrapper, posts
 		threadByRoot[thread.PostId] = thread
 	}
 
+	teamIdByChannelId := map[string]string{}
+
 	for rootId, posts := range postsByRoot {
 		if thread, found := threadByRoot[rootId]; !found {
 			data := []struct {
@@ -2986,16 +2989,30 @@ func (s *SqlPostStore) updateThreadsFromPosts(transaction *sqlxTxWrapper, posts
 			if err != nil {
 				return err
 			}
+
+			channelId := posts[0].ChannelId
+			teamId, ok := teamIdByChannelId[channelId]
+			if !ok {
+				// get teamId for channel
+				err = transaction.Get(&teamId, "SELECT COALESCE(Channels.TeamId, '') FROM Channels WHERE Channels.Id=?", channelId)
+				if err != nil {
+					return err
+				}
+
+				// store teamId for channel for efficiency
+				teamIdByChannelId[channelId] = teamId
+			}
 			// no metadata entry, create one
 			if _, err := transaction.NamedExec(`INSERT INTO Threads
-				(PostId, ChannelId, ReplyCount, LastReplyAt, Participants)
+				(PostId, ChannelId, ReplyCount, LastReplyAt, Participants, TeamId)
 				VALUES
-				(:PostId, :ChannelId, :ReplyCount, :LastReplyAt, :Participants)`, &model.Thread{
+				(:PostId, :ChannelId, :ReplyCount, :LastReplyAt, :Participants, :TeamId)`, &model.Thread{
 				PostId:       rootId,
-				ChannelId:    posts[0].ChannelId,
+				ChannelId:    channelId,
 				ReplyCount:   count,
 				LastReplyAt:  lastReplyAt,
 				Participants: participants,
+				TeamId:       teamId,
 			}); err != nil {
 				return err
 			}

store/sqlstore/thread_store.go

@@ -51,6 +51,7 @@ func (s *SqlThreadStore) initializeQueries() {
 			"Threads.LastReplyAt",
 			"Threads.Participants",
 			"COALESCE(Threads.ThreadDeleteAt, 0) AS DeleteAt",
+			"COALESCE(Threads.TeamId, '') AS TeamId",
 		).
 		From("Threads")
 
@@ -62,6 +63,7 @@ func (s *SqlThreadStore) initializeQueries() {
 			"Threads.LastReplyAt",
 			"Threads.Participants",
 			"COALESCE(Threads.ThreadDeleteAt, 0) AS ThreadDeleteAt",
+			"COALESCE(Threads.TeamId, '') AS TeamId",
 		).
 		From("Threads")
 }
@@ -95,10 +97,9 @@ func (s *SqlThreadStore) getTotalThreadsQuery(userId, teamId string, opts model.
 
 	if teamId != "" {
 		query = query.
-			LeftJoin("Channels ON Threads.ChannelId = Channels.Id").
 			Where(sq.Or{
-				sq.Eq{"Channels.TeamId": teamId},
-				sq.Eq{"Channels.TeamId": ""},
+				sq.Eq{"Threads.TeamId": teamId},
+				sq.Eq{"Threads.TeamId": ""},
 			})
 	}
 
@@ -158,10 +159,9 @@ func (s *SqlThreadStore) GetTotalUnreadMentions(userId, teamId string, opts mode
 
 	if teamId != "" {
 		query = query.
-			LeftJoin("Channels ON Threads.ChannelId = Channels.Id").
 			Where(sq.Or{
-				sq.Eq{"Channels.TeamId": teamId},
-				sq.Eq{"Channels.TeamId": ""},
+				sq.Eq{"Threads.TeamId": teamId},
+				sq.Eq{"Threads.TeamId": ""},
 			})
 	}
 
@@ -192,6 +192,7 @@ func (s *SqlThreadStore) GetThreadsForUser(userId, teamId string, opts model.Get
 		UnreadMentions int64
 		Participants   model.StringArray
 		ThreadDeleteAt int64
+		TeamId         string
 		model.Post
 	}
 
@@ -223,10 +224,9 @@ func (s *SqlThreadStore) GetThreadsForUser(userId, teamId string, opts model.Get
 	// a team at all.
 	if teamId != "" {
 		query = query.
-			Join("Channels ON Threads.ChannelId = Channels.Id").
 			Where(sq.Or{
-				sq.Eq{"Channels.TeamId": teamId},
-				sq.Eq{"Channels.TeamId": ""},
+				sq.Eq{"Threads.TeamId": teamId},
+				sq.Eq{"Threads.TeamId": ""},
 			})
 	}
 
@@ -322,7 +322,7 @@ func (s *SqlThreadStore) GetTeamsUnreadForUser(userID string, teamIDs []string)
 	fetchConditions := sq.And{
 		sq.Eq{"ThreadMemberships.UserId": userID},
 		sq.Eq{"ThreadMemberships.Following": true},
-		sq.Eq{"Channels.TeamId": teamIDs},
+		sq.Eq{"Threads.TeamId": teamIDs},
 		sq.Eq{"COALESCE(Threads.ThreadDeleteAt, 0)": 0},
 	}
 
@@ -348,10 +348,9 @@ func (s *SqlThreadStore) GetTeamsUnreadForUser(userID string, teamIDs []string)
 			Select("COUNT(Threads.PostId) AS Count, TeamId").
 			From("Threads").
 			LeftJoin("ThreadMemberships ON Threads.PostId = ThreadMemberships.PostId").
-			LeftJoin("Channels ON Threads.ChannelId = Channels.Id").
 			Where(fetchConditions).
 			Where("Threads.LastReplyAt > ThreadMemberships.LastViewed").
-			GroupBy("Channels.TeamId")
+			GroupBy("Threads.TeamId")
 
 		err := s.GetReplicaX().SelectBuilder(&unreadThreads, repliesQuery)
 		if err != nil {
@@ -366,9 +365,8 @@ func (s *SqlThreadStore) GetTeamsUnreadForUser(userID string, teamIDs []string)
 			Select("COALESCE(SUM(ThreadMemberships.UnreadMentions),0) AS Count, TeamId").
 			From("ThreadMemberships").
 			LeftJoin("Threads ON Threads.PostId = ThreadMemberships.PostId").
-			LeftJoin("Channels ON Threads.ChannelId = Channels.Id").
 			Where(fetchConditions).
-			GroupBy("Channels.TeamId")
+			GroupBy("Threads.TeamId")
 
 		err := s.GetReplicaX().SelectBuilder(&unreadMentions, mentionsQuery)
 		if err != nil {
@@ -449,6 +447,7 @@ func (s *SqlThreadStore) GetThreadForUser(teamId string, threadMembership *model
 		UnreadMentions int64
 		Participants   model.StringArray
 		ThreadDeleteAt int64
+		TeamId         string
 		model.Post
 	}
 
@@ -462,7 +461,7 @@ func (s *SqlThreadStore) GetThreadForUser(teamId string, threadMembership *model
 		})
 
 	fetchConditions := sq.And{
-		sq.Or{sq.Eq{"Channels.TeamId": teamId}, sq.Eq{"Channels.TeamId": ""}},
+		sq.Or{sq.Eq{"Threads.TeamId": teamId}, sq.Eq{"Threads.TeamId": ""}},
 		sq.Eq{"Threads.PostId": threadMembership.PostId},
 	}
 
@@ -476,7 +475,6 @@ func (s *SqlThreadStore) GetThreadForUser(teamId string, threadMembership *model
 	query = query.
 		Column(sq.Alias(unreadRepliesQuery, "UnreadReplies")).
 		LeftJoin("Posts ON Posts.Id = Threads.PostId").
-		LeftJoin("Channels ON Posts.ChannelId = Channels.Id").
 		Where(fetchConditions)
 
 	err := s.GetReplicaX().GetBuilder(&thread, query)
@@ -671,9 +669,8 @@ func (s *SqlThreadStore) GetMembershipsForUser(userId, teamId string) ([]*model.
 	query := s.getQueryBuilder().
 		Select("ThreadMemberships.*").
 		Join("Threads ON Threads.PostId = ThreadMemberships.PostId").
-		Join("Channels ON Threads.ChannelId = Channels.Id").
 		From("ThreadMemberships").
-		Where(sq.Or{sq.Eq{"Channels.TeamId": teamId}, sq.Eq{"Channels.TeamId": ""}}).
+		Where(sq.Or{sq.Eq{"Threads.TeamId": teamId}, sq.Eq{"Threads.TeamId": ""}}).
 		Where(sq.Eq{"ThreadMemberships.UserId": userId})
 
 	err := s.GetReplicaX().SelectBuilder(&memberships, query)

store/storetest/channel_store.go

@@ -6894,11 +6894,19 @@ func testChannelStoreGetPinnedPosts(t *testing.T, ss store.Store) {
 	require.Empty(t, pl.Posts, "wasn't supposed to return posts")
 
 	t.Run("with correct ReplyCount", func(t *testing.T) {
-		channelId := model.NewId()
+		teamId := model.NewId()
+		channel, err := ss.Channel().Save(&model.Channel{
+			TeamId:      teamId,
+			DisplayName: "DisplayName",
+			Name:        "channel" + model.NewId(),
+			Type:        model.ChannelTypeOpen,
+		}, -1)
+		require.NoError(t, err)
+
 		userId := model.NewId()
 
 		post1, err := ss.Post().Save(&model.Post{
-			ChannelId: channelId,
+			ChannelId: channel.Id,
 			UserId:    userId,
 			Message:   "message",
 			IsPinned:  true,
@@ -6907,7 +6915,7 @@ func testChannelStoreGetPinnedPosts(t *testing.T, ss store.Store) {
 		time.Sleep(time.Millisecond)
 
 		post2, err := ss.Post().Save(&model.Post{
-			ChannelId: channelId,
+			ChannelId: channel.Id,
 			UserId:    userId,
 			Message:   "message",
 			IsPinned:  true,
@@ -6916,7 +6924,7 @@ func testChannelStoreGetPinnedPosts(t *testing.T, ss store.Store) {
 		time.Sleep(time.Millisecond)
 
 		post3, err := ss.Post().Save(&model.Post{
-			ChannelId: channelId,
+			ChannelId: channel.Id,
 			UserId:    userId,
 			RootId:    post1.Id,
 			Message:   "message",
@@ -6925,7 +6933,7 @@ func testChannelStoreGetPinnedPosts(t *testing.T, ss store.Store) {
 		require.NoError(t, err)
 		time.Sleep(time.Millisecond)
 
-		posts, err := ss.Channel().GetPinnedPosts(channelId)
+		posts, err := ss.Channel().GetPinnedPosts(channel.Id)
 		require.NoError(t, err)
 		require.Len(t, posts.Posts, 3)
 		require.Equal(t, posts.Posts[post1.Id].ReplyCount, int64(1))

store/storetest/thread_store.go

@@ -106,8 +106,17 @@ func testThreadStorePopulation(t *testing.T, ss store.Store) {
 		require.Equal(t, int64(2), thread.ReplyCount)
 		require.ElementsMatch(t, model.StringArray{newPosts[0].UserId, newPosts[1].UserId}, thread.Participants)
 
+		teamId := model.NewId()
+		channel, err := ss.Channel().Save(&model.Channel{
+			TeamId:      teamId,
+			DisplayName: "DisplayName1",
+			Name:        "channel" + model.NewId(),
+			Type:        model.ChannelTypeOpen,
+		}, -1)
+		require.NoError(t, err)
+
 		o5 := model.Post{}
-		o5.ChannelId = model.NewId()
+		o5.ChannelId = channel.Id
 		o5.UserId = model.NewId()
 		o5.RootId = newPosts[0].Id
 		o5.Message = NewTestId()
@@ -141,9 +150,18 @@ func testThreadStorePopulation(t *testing.T, ss store.Store) {
 	})
 
 	t.Run("Update reply should update the UpdateAt of the thread", func(t *testing.T) {
+		teamId := model.NewId()
+		channel, err := ss.Channel().Save(&model.Channel{
+			TeamId:      teamId,
+			DisplayName: "DisplayName",
+			Name:        "channel" + model.NewId(),
+			Type:        model.ChannelTypeOpen,
+		}, -1)
+		require.NoError(t, err)
+
 		rootPost := model.Post{}
 		rootPost.RootId = model.NewId()
-		rootPost.ChannelId = model.NewId()
+		rootPost.ChannelId = channel.Id
 		rootPost.UserId = model.NewId()
 		rootPost.Message = NewTestId()
 
@@ -188,8 +206,17 @@ func testThreadStorePopulation(t *testing.T, ss store.Store) {
 	})
 
 	t.Run("Deleting reply should update the thread", func(t *testing.T) {
+		teamId := model.NewId()
+		channel, err := ss.Channel().Save(&model.Channel{
+			TeamId:      teamId,
+			DisplayName: "DisplayName",
+			Name:        "channel" + model.NewId(),
+			Type:        model.ChannelTypeOpen,
+		}, -1)
+		require.NoError(t, err)
+
 		o1 := model.Post{}
-		o1.ChannelId = model.NewId()
+		o1.ChannelId = channel.Id
 		o1.UserId = model.NewId()
 		o1.Message = NewTestId()
 		rootPost, err := ss.Post().Save(&o1)
@@ -240,8 +267,17 @@ func testThreadStorePopulation(t *testing.T, ss store.Store) {
 	})
 
 	t.Run("Deleting root post should delete the thread", func(t *testing.T) {
+		teamId := model.NewId()
+		channel, err := ss.Channel().Save(&model.Channel{
+			TeamId:      teamId,
+			DisplayName: "DisplayName",
+			Name:        "channel" + model.NewId(),
+			Type:        model.ChannelTypeOpen,
+		}, -1)
+		require.NoError(t, err)
+
 		rootPost := model.Post{}
-		rootPost.ChannelId = model.NewId()
+		rootPost.ChannelId = channel.Id
 		rootPost.UserId = model.NewId()
 		rootPost.Message = NewTestId()