From cf4df5fcd25b93bbd8206603df61304e00b04311 Mon Sep 17 00:00:00 2001 From: Eli Yukelzon Date: Fri, 20 Nov 2020 11:00:52 +0200 Subject: [PATCH] MM-30048 added thread related socket messages (#16234) Co-authored-by: Mattermod --- api4/user_test.go | 92 +++++++++++++++++++++++++++++++++ app/notification.go | 52 +++++++++++++++---- app/user.go | 11 ++++ i18n/en.json | 4 ++ model/config.go | 15 ++++++ model/preference.go | 28 +++++----- model/thread.go | 6 +++ model/websocket_message.go | 3 ++ services/telemetry/telemetry.go | 1 + store/sqlstore/post_store.go | 2 +- store/storetest/thread_store.go | 2 +- 11 files changed, 190 insertions(+), 26 deletions(-) diff --git a/api4/user_test.go b/api4/user_test.go index 65a3b0deb4..93d3f33ded 100644 --- a/api4/user_test.go +++ b/api4/user_test.go @@ -5401,6 +5401,98 @@ func TestGetThreadsForUser(t *testing.T) { }) } +func TestThreadSocketEvents(t *testing.T) { + th := Setup(t).InitBasic() + defer th.TearDown() + th.App.UpdateConfig(func(cfg *model.Config) { + *cfg.ServiceSettings.ThreadAutoFollow = true + *cfg.ServiceSettings.CollapsedThreads = model.COLLAPSED_THREADS_DEFAULT_ON + }) + + userWSClient, err := th.CreateWebSocketClient() + require.Nil(t, err) + defer userWSClient.Close() + userWSClient.Listen() + + Client := th.Client + + rpost, resp := Client.CreatePost(&model.Post{ChannelId: th.BasicChannel.Id, Message: "testMsg"}) + CheckNoError(t, resp) + CheckCreatedStatus(t, resp) + + _, err = th.App.CreatePostAsUser(&model.Post{ChannelId: th.BasicChannel.Id, Message: "testReply", UserId: th.BasicUser2.Id, RootId: rpost.Id}, th.App.Session().Id, false) + require.Nil(t, err) + defer th.App.Srv().Store.Post().PermanentDeleteByUser(th.BasicUser.Id) + defer th.App.Srv().Store.Post().PermanentDeleteByUser(th.BasicUser2.Id) + + t.Run("Listed for update event", func(t *testing.T) { + var caught bool + func() { + for { + select { + case ev := <-userWSClient.EventChannel: + if ev.EventType() == model.WEBSOCKET_EVENT_THREAD_UPDATED { + caught = true + thread, err := model.ThreadFromJson(ev.GetData()["thread"].(string)) + require.Nil(t, err) + require.Contains(t, thread.Participants, th.BasicUser.Id) + require.Contains(t, thread.Participants, th.BasicUser2.Id) + } + case <-time.After(1 * time.Second): + return + } + } + }() + require.Truef(t, caught, "User should have received %s event", model.WEBSOCKET_EVENT_THREAD_UPDATED) + }) + + resp = th.Client.UpdateThreadFollowForUser(th.BasicUser.Id, rpost.Id, false) + CheckNoError(t, resp) + CheckOKStatus(t, resp) + + t.Run("Listed for follow event", func(t *testing.T) { + var caught bool + func() { + for { + select { + case ev := <-userWSClient.EventChannel: + if ev.EventType() == model.WEBSOCKET_EVENT_THREAD_FOLLOW_CHANGED { + caught = true + require.Equal(t, ev.GetData()["state"], false) + } + case <-time.After(1 * time.Second): + return + } + } + }() + require.Truef(t, caught, "User should have received %s event", model.WEBSOCKET_EVENT_THREAD_FOLLOW_CHANGED) + }) + + resp = th.Client.UpdateThreadReadForUser(th.BasicUser.Id, rpost.Id, 123) + CheckNoError(t, resp) + CheckOKStatus(t, resp) + + t.Run("Listed for read event", func(t *testing.T) { + var caught bool + func() { + for { + select { + case ev := <-userWSClient.EventChannel: + if ev.EventType() == model.WEBSOCKET_EVENT_THREAD_READ_CHANGED { + caught = true + require.EqualValues(t, ev.GetData()["timestamp"], 123) + } + case <-time.After(1 * time.Second): + return + } + } + }() + + require.Truef(t, caught, "User should have received %s event", model.WEBSOCKET_EVENT_THREAD_READ_CHANGED) + }) + +} + func TestFollowThreads(t *testing.T) { th := Setup(t).InitBasic() defer th.TearDown() diff --git a/app/notification.go b/app/notification.go index 7eb3b26891..a21b608e36 100644 --- a/app/notification.go +++ b/app/notification.go @@ -6,6 +6,7 @@ package app import ( "net/http" "sort" + "strconv" "strings" "unicode" "unicode/utf8" @@ -160,24 +161,31 @@ func (a *App) SendNotifications(post *model.Post, team *model.Team, channel *mod mentionedUsersList := make([]string, 0, len(mentions.Mentions)) updateMentionChans := []chan *model.AppError{} mentionAutofollowChans := []chan *model.AppError{} + threadParticipants := []string{post.UserId} + if *a.Config().ServiceSettings.ThreadAutoFollow && post.RootId != "" { + if parentPostList != nil { + threadParticipants = append(threadParticipants, parentPostList.Posts[parentPostList.Order[0]].UserId) + } + for id := range mentions.Mentions { + threadParticipants = append(threadParticipants, id) + } + // for each mention, make sure to update thread autofollow + for _, id := range threadParticipants { + mac := make(chan *model.AppError, 1) + go func(userId string) { + defer close(mac) - // for each mention, make sure to update thread autofollow - for id := range mentions.Mentions { - mac := make(chan *model.AppError, 1) - go func(userId string) { - defer close(mac) - if *a.Config().ServiceSettings.ThreadAutoFollow && post.RootId != "" { nErr := a.Srv().Store.Thread().CreateMembershipIfNeeded(userId, post.RootId, true) if nErr != nil { mac <- model.NewAppError("SendNotifications", "app.channel.autofollow.app_error", nil, nErr.Error(), http.StatusInternalServerError) return } - } - mac <- nil - }(id) - mentionAutofollowChans = append(mentionAutofollowChans, mac) - } + mac <- nil + }(id) + mentionAutofollowChans = append(mentionAutofollowChans, mac) + } + } for id := range mentions.Mentions { mentionedUsersList = append(mentionedUsersList, id) @@ -403,6 +411,28 @@ func (a *App) SendNotifications(post *model.Post, team *model.Team, channel *mod } a.Publish(message) + // If this is a reply in a thread, notify participants + if *a.Config().ServiceSettings.CollapsedThreads != model.COLLAPSED_THREADS_DISABLED && post.RootId != "" { + thread, err := a.Srv().Store.Thread().Get(post.RootId) + if err != nil { + mlog.Error("Cannot get thread", mlog.String("id", post.RootId)) + return nil, err + } + payload := thread.ToJson() + for _, uid := range thread.Participants { + sendEvent := *a.Config().ServiceSettings.CollapsedThreads == model.COLLAPSED_THREADS_DEFAULT_ON + // check if a participant has overridden collapsed threads settings + if preference, err := a.Srv().Store.Preference().Get(uid, model.PREFERENCE_CATEGORY_COLLAPSED_THREADS_SETTINGS, model.PREFERENCE_NAME_COLLAPSED_THREADS_ENABLED); err == nil { + sendEvent, _ = strconv.ParseBool(preference.Value) + } + if sendEvent { + message := model.NewWebSocketEvent(model.WEBSOCKET_EVENT_THREAD_UPDATED, "", "", uid, nil) + message.Add("thread", payload) + a.Publish(message) + } + } + + } return mentionedUsersList, nil } diff --git a/app/user.go b/app/user.go index 0eeb7544ed..9be5a2799d 100644 --- a/app/user.go +++ b/app/user.go @@ -2388,6 +2388,9 @@ func (a *App) UpdateThreadsReadForUser(userId string, timestamp int64) *model.Ap if err != nil { return model.NewAppError("UpdateThreadsReadForUser", "app.user.update_threads_read_for_user.app_error", nil, err.Error(), http.StatusInternalServerError) } + message := model.NewWebSocketEvent(model.WEBSOCKET_EVENT_THREAD_READ_CHANGED, "", "", userId, nil) + message.Add("timestamp", timestamp) + a.Publish(message) return nil } @@ -2396,6 +2399,10 @@ func (a *App) UpdateThreadFollowForUser(userId, threadId string, state bool) *mo if err != nil { return model.NewAppError("UpdateThreadFollowForUser", "app.user.update_thread_follow_for_user.app_error", nil, err.Error(), http.StatusInternalServerError) } + message := model.NewWebSocketEvent(model.WEBSOCKET_EVENT_THREAD_FOLLOW_CHANGED, "", "", userId, nil) + message.Add("thread_id", threadId) + message.Add("state", state) + a.Publish(message) return nil } @@ -2404,5 +2411,9 @@ func (a *App) UpdateThreadReadForUser(userId, threadId string, timestamp int64) if err != nil { return model.NewAppError("UpdateThreadReadForUser", "app.user.update_thread_read_for_user.app_error", nil, err.Error(), http.StatusInternalServerError) } + message := model.NewWebSocketEvent(model.WEBSOCKET_EVENT_THREAD_READ_CHANGED, "", "", userId, nil) + message.Add("thread_id", threadId) + message.Add("timestamp", timestamp) + a.Publish(message) return nil } diff --git a/i18n/en.json b/i18n/en.json index 23a5dc0a27..974715b8a5 100644 --- a/i18n/en.json +++ b/i18n/en.json @@ -7022,6 +7022,10 @@ "id": "model.config.is_valid.cluster_email_batching.app_error", "translation": "Unable to enable email batching when clustering is enabled." }, + { + "id": "model.config.is_valid.collapsed_threads.app_error", + "translation": "CollapsedThreads setting must be either disabled,default_on or default_off" + }, { "id": "model.config.is_valid.data_retention.deletion_job_start_time.app_error", "translation": "Data retention job start time must be a 24-hour time stamp in the form HH:MM." diff --git a/model/config.go b/model/config.go index f43dd14577..c37235bb08 100644 --- a/model/config.go +++ b/model/config.go @@ -83,6 +83,10 @@ const ( GROUP_UNREAD_CHANNELS_DEFAULT_ON = "default_on" GROUP_UNREAD_CHANNELS_DEFAULT_OFF = "default_off" + COLLAPSED_THREADS_DISABLED = "disabled" + COLLAPSED_THREADS_DEFAULT_ON = "default_on" + COLLAPSED_THREADS_DEFAULT_OFF = "default_off" + EMAIL_BATCHING_BUFFER_SIZE = 256 EMAIL_BATCHING_INTERVAL = 30 @@ -351,6 +355,7 @@ type ServiceSettings struct { FeatureFlagSyncIntervalSeconds *int `access:"environment,write_restrictable"` DebugSplit *bool `access:"environment,write_restrictable"` ThreadAutoFollow *bool `access:"experimental"` + CollapsedThreads *string `access:"experimental"` ManagedResourcePaths *string `access:"environment,write_restrictable,cloud_restrictable"` } @@ -786,6 +791,10 @@ func (s *ServiceSettings) SetDefaults(isUpdate bool) { s.ThreadAutoFollow = NewBool(true) } + if s.CollapsedThreads == nil { + s.CollapsedThreads = NewString(COLLAPSED_THREADS_DISABLED) + } + if s.ManagedResourcePaths == nil { s.ManagedResourcePaths = NewString("") } @@ -3437,6 +3446,12 @@ func (s *ServiceSettings) isValid() *AppError { return NewAppError("Config.IsValid", "model.config.is_valid.group_unread_channels.app_error", nil, "", http.StatusBadRequest) } + if *s.CollapsedThreads != COLLAPSED_THREADS_DISABLED && + *s.CollapsedThreads != COLLAPSED_THREADS_DEFAULT_ON && + *s.CollapsedThreads != COLLAPSED_THREADS_DEFAULT_OFF { + return NewAppError("Config.IsValid", "model.config.is_valid.collapsed_threads.app_error", nil, "", http.StatusBadRequest) + } + return nil } diff --git a/model/preference.go b/model/preference.go index e752bb54c0..3cb6ec98d6 100644 --- a/model/preference.go +++ b/model/preference.go @@ -13,20 +13,22 @@ import ( ) const ( - PREFERENCE_CATEGORY_DIRECT_CHANNEL_SHOW = "direct_channel_show" - PREFERENCE_CATEGORY_GROUP_CHANNEL_SHOW = "group_channel_show" - PREFERENCE_CATEGORY_TUTORIAL_STEPS = "tutorial_step" - PREFERENCE_CATEGORY_ADVANCED_SETTINGS = "advanced_settings" - PREFERENCE_CATEGORY_FLAGGED_POST = "flagged_post" - PREFERENCE_CATEGORY_FAVORITE_CHANNEL = "favorite_channel" - PREFERENCE_CATEGORY_SIDEBAR_SETTINGS = "sidebar_settings" + PREFERENCE_CATEGORY_DIRECT_CHANNEL_SHOW = "direct_channel_show" + PREFERENCE_CATEGORY_GROUP_CHANNEL_SHOW = "group_channel_show" + PREFERENCE_CATEGORY_TUTORIAL_STEPS = "tutorial_step" + PREFERENCE_CATEGORY_ADVANCED_SETTINGS = "advanced_settings" + PREFERENCE_CATEGORY_FLAGGED_POST = "flagged_post" + PREFERENCE_CATEGORY_FAVORITE_CHANNEL = "favorite_channel" + PREFERENCE_CATEGORY_SIDEBAR_SETTINGS = "sidebar_settings" + PREFERENCE_CATEGORY_COLLAPSED_THREADS_SETTINGS = "collapsed_threads_settings" - PREFERENCE_CATEGORY_DISPLAY_SETTINGS = "display_settings" - PREFERENCE_NAME_CHANNEL_DISPLAY_MODE = "channel_display_mode" - PREFERENCE_NAME_COLLAPSE_SETTING = "collapse_previews" - PREFERENCE_NAME_MESSAGE_DISPLAY = "message_display" - PREFERENCE_NAME_NAME_FORMAT = "name_format" - PREFERENCE_NAME_USE_MILITARY_TIME = "use_military_time" + PREFERENCE_CATEGORY_DISPLAY_SETTINGS = "display_settings" + PREFERENCE_NAME_COLLAPSED_THREADS_ENABLED = "collapsed_threads_enabled" + PREFERENCE_NAME_CHANNEL_DISPLAY_MODE = "channel_display_mode" + PREFERENCE_NAME_COLLAPSE_SETTING = "collapse_previews" + PREFERENCE_NAME_MESSAGE_DISPLAY = "message_display" + PREFERENCE_NAME_NAME_FORMAT = "name_format" + PREFERENCE_NAME_USE_MILITARY_TIME = "use_military_time" PREFERENCE_CATEGORY_THEME = "theme" // the name for theme props is the team id diff --git a/model/thread.go b/model/thread.go index 80e655b8ef..3c79b043a8 100644 --- a/model/thread.go +++ b/model/thread.go @@ -56,6 +56,12 @@ func (o *Thread) ToJson() string { return string(b) } +func ThreadFromJson(s string) (*Thread, error) { + var t Thread + err := json.Unmarshal([]byte(s), &t) + return &t, err +} + func (o *Thread) Etag() string { return Etag(o.PostId, o.LastReplyAt) } diff --git a/model/websocket_message.go b/model/websocket_message.go index a4f92f80b2..3d667725c5 100644 --- a/model/websocket_message.go +++ b/model/websocket_message.go @@ -70,6 +70,9 @@ const ( WEBSOCKET_WARN_METRIC_STATUS_RECEIVED = "warn_metric_status_received" WEBSOCKET_WARN_METRIC_STATUS_REMOVED = "warn_metric_status_removed" WEBSOCKET_EVENT_CLOUD_PAYMENT_STATUS_UPDATED = "cloud_payment_status_updated" + WEBSOCKET_EVENT_THREAD_UPDATED = "thread_updated" + WEBSOCKET_EVENT_THREAD_FOLLOW_CHANGED = "thread_follow_changed" + WEBSOCKET_EVENT_THREAD_READ_CHANGED = "thread_read_changed" ) type WebSocketMessage interface { diff --git a/services/telemetry/telemetry.go b/services/telemetry/telemetry.go index a197f54e02..7bf477db00 100644 --- a/services/telemetry/telemetry.go +++ b/services/telemetry/telemetry.go @@ -405,6 +405,7 @@ func (ts *TelemetryService) trackConfig() { "enable_tutorial": *cfg.ServiceSettings.EnableTutorial, "experimental_enable_default_channel_leave_join_messages": *cfg.ServiceSettings.ExperimentalEnableDefaultChannelLeaveJoinMessages, "experimental_group_unread_channels": *cfg.ServiceSettings.ExperimentalGroupUnreadChannels, + "collapsed_threads": *cfg.ServiceSettings.CollapsedThreads, "websocket_url": isDefault(*cfg.ServiceSettings.WebsocketURL, ""), "allow_cookies_for_subdomains": *cfg.ServiceSettings.AllowCookiesForSubdomains, "enable_api_team_deletion": *cfg.ServiceSettings.EnableAPITeamDeletion, diff --git a/store/sqlstore/post_store.go b/store/sqlstore/post_store.go index 89d60c50ff..8b790693b8 100644 --- a/store/sqlstore/post_store.go +++ b/store/sqlstore/post_store.go @@ -1976,7 +1976,7 @@ func (s *SqlPostStore) updateThreadsFromPosts(transaction *gorp.Transaction, pos if thread, found := threadByRoot[rootId]; !found { // calculate participants var participants model.StringArray - if _, err := transaction.Select(&participants, "SELECT DISTINCT UserId FROM Posts WHERE RootId=:RootId", map[string]interface{}{"RootId": rootId}); err != nil { + if _, err := transaction.Select(&participants, "SELECT DISTINCT UserId FROM Posts WHERE RootId=:RootId OR Id=:RootId", map[string]interface{}{"RootId": rootId}); err != nil { return err } // calculate reply count diff --git a/store/storetest/thread_store.go b/store/storetest/thread_store.go index fa9e6ed34c..adea5c69da 100644 --- a/store/storetest/thread_store.go +++ b/store/storetest/thread_store.go @@ -221,7 +221,7 @@ func testThreadStorePopulation(t *testing.T, ss store.Store) { thread1, err := ss.Thread().Get(newPosts1[0].Id) require.Nil(t, err) require.EqualValues(t, thread1.ReplyCount, 1) - require.Len(t, thread1.Participants, 1) + require.Len(t, thread1.Participants, 2) err = ss.Post().PermanentDeleteByUser(rootPost.UserId) require.Nil(t, err)