mirrors/mostlymatter
1
0
Форкнуть 0
Код Задачи Запросы на слияние Действия Пакеты Проекты Релизы Вики Активность

MM-41349: CRT, fix LastUpdated semantics (#19523)

Просмотр исходного кода 
* deadcode: remove UpdateChannelLastViewedAt

* deadcode: remove ThreadStore.(Save(Multiple)|Update|Delete)

* deadcode: followThead in App.MarkChannelAsUnreadFromPost

* document ThreadMembership, Thread structs

* maintain LastUpdated consistently

Whenever we touch a `ThreadMembership` record, we should be setting `LastUpdated` to the current timestamp. The mobile client relies on this to detect changes to these records.

* simplify: never updateThreads from `App.MarkChannelAsUnreadFromPost`

Change all invocations of `ChannelStore.UpdateLastViewedAtPost` from `App.MarkChannelAsUnreadFromPost` to pass `updateThreads` as `false`. When `ChannelStore.UpdateLastViewedAtPost` was invoked with `updateThreads` as `true`, it would in turn call `ThreadStore.UpdateUnreadsByChannel` but pass `updateViewedTimestamp` as `false`. This effectively updated the `LastUpdated` field of the corresponding thread memberships but never touched any of the actual data (such as `LastViewed`).

The overall CRT feature continued to work, because `App.MarkChannelAsUnreadFromPost` directly updates the relevant thread memberships via `ThreadStore.MaintainMembership`.

* deadcode: updateThreads in ChannelStore.UpdateLastViewedAtPost

* simplify: never updateThreads from App.SendNotifications

Change all invocations of `ChannelStore.IncrementMentionCount` from
`App.SendNotifications` to pass `updateThreads` as `false`. When `ChannelStore.IncrementMentionCount` was invoked with `updateThreads` as `true`, it would in turn call `ThreadStore.UpdateUnreadsByChannel` but pass `updateViewedTimestamp` as `false`. This effectively updated the `LastUpdated` field of the corresponding thread memberships but never touched any of the actual data (such as `UnreadMentions`).

The overall CRT feature continued to work, because `App.SendNotifications` directly updates the relevant thread memberships mention counts via `ThreadStore.MaintainMembership`.

* deadcode: updateThreads in ChannelStore.IncrementMentionCount

* fix & rename ThreadStore.UpdateUnreadsByChannel

Rename `ThreadStore.UpdateUnreadsByChannel` to `ThreadStore.UpdateLastViewedByThreadIds`, making it unconditionally set the `LastViewed` for the given threads (as well as `LastUpdated`).

All previous invocations of this method that passed `updateViewedTimestamp` have been previously removed.

* unrelated gofmt -w -s changes to satisfy linter

* always set LastUpdated to model.GetMillis()

* deadcode: ThreadStore.SaveMembership

* fix TestMarkUnreadWithThreads

* GetMasterX
Этот коммит содержится в:
Jesse Hallam
2022-02-28 16:24:34 -04:00
коммит произвёл GitHub
родитель cc900149c6
Коммит 6757edc4e2
20 изменённых файлов: 178 добавлений и 813 удалений
Скачать Patch файл Скачать Diff файл Показать все файлы Свернуть все файлы

2
api4/post.go
Просмотреть файл

@@ -756,7 +756,7 @@ func setPostUnread(c *Context, w http.ResponseWriter, r *http.Request) {
return
}
state, err := c.App.MarkChannelAsUnreadFromPost(c.Params.PostId, c.Params.UserId, collapsedThreadsSupported, false)
state, err := c.App.MarkChannelAsUnreadFromPost(c.Params.PostId, c.Params.UserId, collapsedThreadsSupported)
if err != nil {
c.Err = err
return

3
app/app_iface.go
Просмотреть файл

@@ -223,7 +223,7 @@ type AppIface interface {
// MakeAuditRecord creates a audit record pre-populated with defaults.
MakeAuditRecord(event string, initialStatus string) *audit.Record
// MarkChanelAsUnreadFromPost will take a post and set the channel as unread from that one.
MarkChannelAsUnreadFromPost(postID string, userID string, collapsedThreadsSupported, followThread bool) (*model.ChannelUnreadAt, *model.AppError)
MarkChannelAsUnreadFromPost(postID string, userID string, collapsedThreadsSupported bool) (*model.ChannelUnreadAt, *model.AppError)
// MentionsToPublicChannels returns all the mentions to public channels,
// linking them to their channels
MentionsToPublicChannels(message, teamID string) model.ChannelMentionMap
@@ -1035,7 +1035,6 @@ type AppIface interface {
TriggerWebhook(c *request.Context, payload *model.OutgoingWebhookPayload, hook *model.OutgoingWebhook, post *model.Post, channel *model.Channel)
UnregisterPluginCommand(pluginID, teamID, trigger string)
UpdateActive(c *request.Context, user *model.User, active bool) (*model.User, *model.AppError)
UpdateChannelLastViewedAt(channelIDs []string, userID string) *model.AppError
UpdateChannelMemberNotifyProps(data map[string]string, channelID string, userID string) (*model.ChannelMember, *model.AppError)
UpdateChannelMemberRoles(channelID string, userID string, newRoles string) (*model.ChannelMember, *model.AppError)
UpdateChannelMemberSchemeRoles(channelID string, userID string, isSchemeGuest bool, isSchemeUser bool, isSchemeAdmin bool) (*model.ChannelMember, *model.AppError)

42
app/channel.go
Просмотреть файл

@@ -2486,28 +2486,6 @@ func (a *App) SetActiveChannel(userID string, channelID string) *model.AppError
return nil
}
func (a *App) UpdateChannelLastViewedAt(channelIDs []string, userID string) *model.AppError {
if _, err := a.Srv().Store.Channel().UpdateLastViewedAt(channelIDs, userID, *a.Config().ServiceSettings.ThreadAutoFollow); err != nil {
var invErr *store.ErrInvalidInput
switch {
case errors.As(err, &invErr):
return model.NewAppError("UpdateChannelLastViewedAt", "app.channel.update_last_viewed_at.app_error", nil, invErr.Error(), http.StatusBadRequest)
default:
return model.NewAppError("UpdateChannelLastViewedAt", "app.channel.update_last_viewed_at.app_error", nil, err.Error(), http.StatusInternalServerError)
}
}
if *a.Config().ServiceSettings.EnableChannelViewedMessages {
for _, channelID := range channelIDs {
message := model.NewWebSocketEvent(model.WebsocketEventChannelViewed, "", "", userID, nil)
message.Add("channel_id", channelID)
a.Publish(message)
}
}
return nil
}
func (a *App) IsCRTEnabledForUser(userID string) bool {
if *a.Config().ServiceSettings.CollapsedThreads == model.CollapsedThreadsDisabled {
return false
@@ -2521,7 +2499,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) {
func (a *App) MarkChannelAsUnreadFromPost(postID string, userID string, collapsedThreadsSupported bool) (*model.ChannelUnreadAt, *model.AppError) {
if !collapsedThreadsSupported || !a.IsCRTEnabledForUser(userID) {
return a.markChannelAsUnreadFromPostCRTUnsupported(postID, userID)
}
@@ -2553,23 +2531,15 @@ func (a *App) MarkChannelAsUnreadFromPost(postID string, userID string, collapse
if storeErr != nil && !errors.As(storeErr, &nfErr) {
return nil, model.NewAppError("MarkChannelAsUnreadFromPost", "app.channel.update_last_viewed_at_post.app_error", nil, storeErr.Error(), http.StatusInternalServerError)
}
var opts store.ThreadMembershipOpts
// if this post was not followed before, create thread membership and update mention count
if threadMembership == nil {
opts = store.ThreadMembershipOpts{
Following: followThread,
opts := store.ThreadMembershipOpts{
Following: false,
IncrementMentions: false,
UpdateFollowing: true,
UpdateViewedTimestamp: true,
UpdateParticipants: false,
}
} else if !threadMembership.Following && followThread {
opts = store.ThreadMembershipOpts{
Following: true,
UpdateFollowing: true,
}
}
if opts.UpdateFollowing || threadMembership == nil {
threadMembership, storeErr = a.Srv().Store.Thread().MaintainMembership(user.Id, threadId, opts)
if storeErr != nil && !errors.As(storeErr, &nfErr) {
return nil, model.NewAppError("MarkChannelAsUnreadFromPost", "app.channel.update_last_viewed_at_post.app_error", nil, storeErr.Error(), http.StatusInternalServerError)
@@ -2608,7 +2578,7 @@ func (a *App) MarkChannelAsUnreadFromPost(postID string, userID string, collapse
}
}
channelUnread, nErr := a.Srv().Store.Channel().UpdateLastViewedAtPost(post, userID, unreadMentions, unreadMentionsRoot, *a.Config().ServiceSettings.ThreadAutoFollow, true)
channelUnread, nErr := a.Srv().Store.Channel().UpdateLastViewedAtPost(post, userID, unreadMentions, unreadMentionsRoot, true)
if nErr != nil {
return channelUnread, model.NewAppError("MarkChannelAsUnreadFromPost", "app.channel.update_last_viewed_at_post.app_error", nil, nErr.Error(), http.StatusInternalServerError)
}
@@ -2644,7 +2614,7 @@ func (a *App) markChannelAsUnreadFromPostCRTUnsupported(postID string, userID st
// In CRT Supported Client: badge on channel only sums mentions in root posts including and below the post that was marked.
// In CRT Unsupported Client: badge on channel sums mentions in all posts (root & replies) including and below the post that was marked unread.
if post.RootId == "" {
channelUnread, nErr := a.Srv().Store.Channel().UpdateLastViewedAtPost(post, userID, unreadMentions, unreadMentionsRoot, false, true)
channelUnread, nErr := a.Srv().Store.Channel().UpdateLastViewedAtPost(post, userID, unreadMentions, unreadMentionsRoot, true)
if nErr != nil {
return channelUnread, model.NewAppError("MarkChannelAsUnreadFromPost", "app.channel.update_last_viewed_at_post.app_error", nil, nErr.Error(), http.StatusInternalServerError)
}
@@ -2718,7 +2688,7 @@ func (a *App) markChannelAsUnreadFromPostCRTUnsupported(postID string, userID st
}
}
channelUnread, nErr := a.Srv().Store.Channel().UpdateLastViewedAtPost(post, userID, unreadMentions, 0, false, false)
channelUnread, nErr := a.Srv().Store.Channel().UpdateLastViewedAtPost(post, userID, unreadMentions, 0, false)
if nErr != nil {
return channelUnread, model.NewAppError("MarkChannelAsUnreadFromPost", "app.channel.update_last_viewed_at_post.app_error", nil, nErr.Error(), http.StatusInternalServerError)
}

84
app/channel_test.go
Просмотреть файл

@@ -1274,16 +1274,16 @@ func TestMarkChannelAsUnreadFromPost(t *testing.T) {
unread, err = th.App.GetChannelUnread(c1.Id, u2.Id)
require.Nil(t, err)
require.Equal(t, int64(4), unread.MsgCount)
err = th.App.UpdateChannelLastViewedAt([]string{c1.Id, pc1.Id}, u1.Id)
_, err = th.App.MarkChannelsAsViewed([]string{c1.Id, pc1.Id}, u1.Id, "", false)
require.Nil(t, err)
err = th.App.UpdateChannelLastViewedAt([]string{c1.Id, pc1.Id}, u2.Id)
_, err = th.App.MarkChannelsAsViewed([]string{c1.Id, pc1.Id}, u2.Id, "", false)
require.Nil(t, err)
unread, err = th.App.GetChannelUnread(c1.Id, u2.Id)
require.Nil(t, err)
require.Equal(t, int64(0), unread.MsgCount)
t.Run("Unread but last one", func(t *testing.T) {
response, err := th.App.MarkChannelAsUnreadFromPost(p2.Id, u1.Id, true, true)
response, err := th.App.MarkChannelAsUnreadFromPost(p2.Id, u1.Id, true)
require.Nil(t, err)
require.NotNil(t, response)
assert.Equal(t, int64(2), response.MsgCount)
@@ -1294,7 +1294,7 @@ func TestMarkChannelAsUnreadFromPost(t *testing.T) {
})
t.Run("Unread last one", func(t *testing.T) {
response, err := th.App.MarkChannelAsUnreadFromPost(p3.Id, u1.Id, true, true)
response, err := th.App.MarkChannelAsUnreadFromPost(p3.Id, u1.Id, true)
require.Nil(t, err)
require.NotNil(t, response)
assert.Equal(t, int64(3), response.MsgCount)
@@ -1305,7 +1305,7 @@ func TestMarkChannelAsUnreadFromPost(t *testing.T) {
})
t.Run("Unread first one", func(t *testing.T) {
response, err := th.App.MarkChannelAsUnreadFromPost(p1.Id, u1.Id, true, true)
response, err := th.App.MarkChannelAsUnreadFromPost(p1.Id, u1.Id, true)
require.Nil(t, err)
require.NotNil(t, response)
assert.Equal(t, int64(1), response.MsgCount)
@@ -1322,7 +1322,7 @@ func TestMarkChannelAsUnreadFromPost(t *testing.T) {
})
t.Run("Unread on a private channel", func(t *testing.T) {
response, err := th.App.MarkChannelAsUnreadFromPost(pp1.Id, u1.Id, true, true)
response, err := th.App.MarkChannelAsUnreadFromPost(pp1.Id, u1.Id, true)
require.Nil(t, err)
require.NotNil(t, response)
assert.Equal(t, int64(0), response.MsgCount)
@@ -1331,7 +1331,7 @@ func TestMarkChannelAsUnreadFromPost(t *testing.T) {
assert.Equal(t, int64(2), unread.MsgCount)
assert.Equal(t, pp1.CreateAt-1, response.LastViewedAt)
response, err = th.App.MarkChannelAsUnreadFromPost(pp2.Id, u1.Id, true, true)
response, err = th.App.MarkChannelAsUnreadFromPost(pp2.Id, u1.Id, true)
assert.Nil(t, err)
assert.Equal(t, int64(1), response.MsgCount)
unread, err = th.App.GetChannelUnread(pc1.Id, u1.Id)
@@ -1360,7 +1360,7 @@ func TestMarkChannelAsUnreadFromPost(t *testing.T) {
Message: "@" + u1.Username,
}, c2, false, true)
response, err := th.App.MarkChannelAsUnreadFromPost(p4.Id, u1.Id, true, true)
response, err := th.App.MarkChannelAsUnreadFromPost(p4.Id, u1.Id, true)
assert.Nil(t, err)
assert.Equal(t, int64(1), response.MsgCount)
assert.Equal(t, int64(2), response.MentionCount)
@@ -1383,7 +1383,7 @@ func TestMarkChannelAsUnreadFromPost(t *testing.T) {
_, err := th.App.CreatePost(th.Context, &model.Post{ChannelId: dc.Id, UserId: th.BasicUser.Id, Message: "testReply", RootId: dm1.Id}, dc, false, false)
assert.Nil(t, err)
response, err := th.App.MarkChannelAsUnreadFromPost(dm1.Id, u2.Id, true, true)
response, err := th.App.MarkChannelAsUnreadFromPost(dm1.Id, u2.Id, true)
assert.Nil(t, err)
assert.Equal(t, int64(0), response.MsgCount)
assert.Equal(t, int64(4), response.MentionCount)
@@ -1397,7 +1397,7 @@ func TestMarkChannelAsUnreadFromPost(t *testing.T) {
})
t.Run("Can't unread an imaginary post", func(t *testing.T) {
response, err := th.App.MarkChannelAsUnreadFromPost("invalid4ofngungryquinj976y", u1.Id, true, true)
response, err := th.App.MarkChannelAsUnreadFromPost("invalid4ofngungryquinj976y", u1.Id, true)
assert.NotNil(t, err)
assert.Nil(t, response)
})
@@ -2070,7 +2070,7 @@ func TestMarkChannelAsUnreadFromPostPanic(t *testing.T) {
})
require.NotPanics(t, func() {
th.App.MarkChannelAsUnreadFromPost("postID", "userID", true, true)
th.App.MarkChannelAsUnreadFromPost("postID", "userID", true)
}, "unexpected panic from MarkChannelAsUnreadFromPost")
}
@@ -2246,7 +2246,7 @@ func TestMarkChannelAsUnreadFromPostCollapsedThreadsTurnedOff(t *testing.T) {
require.Nil(t, appErr)
t.Run("Mark reply post as unread", func(t *testing.T) {
_, err := th.App.MarkChannelAsUnreadFromPost(replyPost1.Id, th.BasicUser.Id, true, true)
_, err := th.App.MarkChannelAsUnreadFromPost(replyPost1.Id, th.BasicUser.Id, true)
require.Nil(t, err)
// Get channel unreads
// Easier to reason with ChannelUnread now, than channelUnreadAt from the previous call
@@ -2270,7 +2270,7 @@ func TestMarkChannelAsUnreadFromPostCollapsedThreadsTurnedOff(t *testing.T) {
})
t.Run("Mark root post as unread", func(t *testing.T) {
_, err := th.App.MarkChannelAsUnreadFromPost(rootPost1.Id, th.BasicUser.Id, true, true)
_, err := th.App.MarkChannelAsUnreadFromPost(rootPost1.Id, th.BasicUser.Id, true)
require.Nil(t, err)
// Get channel unreads
// Easier to reason with ChannelUnread now, than channelUnreadAt from the previous call
@@ -2285,6 +2285,14 @@ func TestMarkChannelAsUnreadFromPostCollapsedThreadsTurnedOff(t *testing.T) {
})
}
// TestMarkUnreadWithThreads asserts the behaviour of App.MarkChannelAsUnreadFromPost, but was
// originally written when that API accepted a followThread parameter. While tested, that parameter
// was never actually called as true, resulting in deadcode.
//
// When removing the parameter, one of the following tests failed, as it covered behaviour that
// was unused and now unsupported. The test has since been updated to reflect the new reality,
// but a careful examination of MarkChannelAsUnreadFromPost should be conducted as it's unclear
// why that API tries to manipulate thread memberships at all. Fixing that is left to another task.
func TestMarkUnreadWithThreads(t *testing.T) {
os.Setenv("MM_FEATUREFLAGS_COLLAPSEDTHREADS", "true")
defer os.Unsetenv("MM_FEATUREFLAGS_COLLAPSEDTHREADS")
@@ -2295,47 +2303,11 @@ func TestMarkUnreadWithThreads(t *testing.T) {
*cfg.ServiceSettings.CollapsedThreads = model.CollapsedThreadsDefaultOn
})
t.Run("Follow threads only if specified", func(t *testing.T) {
rootPost, appErr := th.App.CreatePost(th.Context, &model.Post{UserId: th.BasicUser2.Id, CreateAt: model.GetMillis(), ChannelId: th.BasicChannel.Id, Message: "hi"}, th.BasicChannel, false, false)
require.Nil(t, appErr)
replyPost, appErr := th.App.CreatePost(th.Context, &model.Post{RootId: rootPost.Id, UserId: th.BasicUser2.Id, CreateAt: model.GetMillis(), ChannelId: th.BasicChannel.Id, Message: "hi"}, th.BasicChannel, false, false)
require.Nil(t, appErr)
threads, appErr := th.App.GetThreadsForUser(th.BasicUser.Id, th.BasicTeam.Id, model.GetUserThreadsOpts{})
require.Nil(t, appErr)
require.Zero(t, threads.Total)
_, appErr = th.App.MarkChannelAsUnreadFromPost(replyPost.Id, th.BasicUser.Id, true, true)
require.Nil(t, appErr)
threads, appErr = th.App.GetThreadsForUser(th.BasicUser.Id, th.BasicTeam.Id, model.GetUserThreadsOpts{})
require.Nil(t, appErr)
require.NotZero(t, threads.Total)
threadMembership, appErr := th.App.GetThreadMembershipForUser(th.BasicUser.Id, replyPost.RootId)
require.Nil(t, appErr)
require.NotNil(t, threadMembership)
assert.True(t, threadMembership.Following)
// Create a new thread
rootPost, appErr = th.App.CreatePost(th.Context, &model.Post{UserId: th.BasicUser2.Id, CreateAt: model.GetMillis(), ChannelId: th.BasicChannel.Id, Message: "hi2"}, th.BasicChannel, false, false)
require.Nil(t, appErr)
replyPost, appErr = th.App.CreatePost(th.Context, &model.Post{RootId: rootPost.Id, UserId: th.BasicUser2.Id, CreateAt: model.GetMillis(), ChannelId: th.BasicChannel.Id, Message: "hi2"}, th.BasicChannel, false, false)
require.Nil(t, appErr)
_, appErr = th.App.MarkChannelAsUnreadFromPost(replyPost.Id, th.BasicUser.Id, true, false)
require.Nil(t, appErr)
threadMembership, appErr = th.App.GetThreadMembershipForUser(th.BasicUser.Id, replyPost.RootId)
require.Nil(t, appErr)
require.NotNil(t, threadMembership)
assert.False(t, threadMembership.Following)
})
t.Run("Set unread mentions correctly", func(t *testing.T) {
t.Run("Never followed root post with no replies or mentions", func(t *testing.T) {
rootPost, appErr := th.App.CreatePost(th.Context, &model.Post{UserId: th.BasicUser2.Id, CreateAt: model.GetMillis(), ChannelId: th.BasicChannel.Id, Message: "hi"}, th.BasicChannel, false, false)
require.Nil(t, appErr)
_, appErr = th.App.MarkChannelAsUnreadFromPost(rootPost.Id, th.BasicUser.Id, true, true)
_, appErr = th.App.MarkChannelAsUnreadFromPost(rootPost.Id, th.BasicUser.Id, true)
require.Nil(t, appErr)
threadMembership, appErr := th.App.GetThreadMembershipForUser(th.BasicUser.Id, rootPost.Id)
@@ -2349,7 +2321,7 @@ func TestMarkUnreadWithThreads(t *testing.T) {
require.Nil(t, appErr)
_, appErr = th.App.CreatePost(th.Context, &model.Post{RootId: rootPost.Id, UserId: th.BasicUser2.Id, CreateAt: model.GetMillis(), ChannelId: th.BasicChannel.Id, Message: "hi"}, th.BasicChannel, false, false)
require.Nil(t, appErr)
_, appErr = th.App.MarkChannelAsUnreadFromPost(rootPost.Id, th.BasicUser.Id, true, true)
_, appErr = th.App.MarkChannelAsUnreadFromPost(rootPost.Id, th.BasicUser.Id, true)
require.Nil(t, appErr)
threadMembership, appErr := th.App.GetThreadMembershipForUser(th.BasicUser.Id, rootPost.Id)
@@ -2363,7 +2335,7 @@ func TestMarkUnreadWithThreads(t *testing.T) {
require.Nil(t, appErr)
_, appErr = th.App.CreatePost(th.Context, &model.Post{RootId: rootPost.Id, UserId: th.BasicUser2.Id, CreateAt: model.GetMillis(), ChannelId: th.BasicChannel.Id, Message: "hi @" + th.BasicUser.Username}, th.BasicChannel, false, false)
require.Nil(t, appErr)
_, appErr = th.App.MarkChannelAsUnreadFromPost(rootPost.Id, th.BasicUser.Id, true, true)
_, appErr = th.App.MarkChannelAsUnreadFromPost(rootPost.Id, th.BasicUser.Id, true)
require.Nil(t, appErr)
threadMembership, appErr := th.App.GetThreadMembershipForUser(th.BasicUser.Id, rootPost.Id)
@@ -2380,7 +2352,7 @@ func TestMarkUnreadWithThreads(t *testing.T) {
appErr = th.App.UpdateThreadFollowForUser(th.BasicUser.Id, th.BasicTeam.Id, rootPost.Id, false)
require.Nil(t, appErr)
_, appErr = th.App.MarkChannelAsUnreadFromPost(rootPost.Id, th.BasicUser.Id, true, true)
_, appErr = th.App.MarkChannelAsUnreadFromPost(rootPost.Id, th.BasicUser.Id, true)
require.Nil(t, appErr)
threadMembership, appErr := th.App.GetThreadMembershipForUser(th.BasicUser.Id, rootPost.Id)
@@ -2399,7 +2371,7 @@ func TestMarkUnreadWithThreads(t *testing.T) {
appErr = th.App.UpdateThreadFollowForUser(th.BasicUser.Id, th.BasicTeam.Id, rootPost.Id, false)
require.Nil(t, appErr)
_, appErr = th.App.MarkChannelAsUnreadFromPost(rootPost.Id, th.BasicUser.Id, true, true)
_, appErr = th.App.MarkChannelAsUnreadFromPost(rootPost.Id, th.BasicUser.Id, true)
require.Nil(t, appErr)
threadMembership, appErr := th.App.GetThreadMembershipForUser(th.BasicUser.Id, rootPost.Id)
@@ -2418,13 +2390,13 @@ func TestMarkUnreadWithThreads(t *testing.T) {
appErr = th.App.UpdateThreadFollowForUser(th.BasicUser.Id, th.BasicTeam.Id, rootPost.Id, false)
require.Nil(t, appErr)
_, appErr = th.App.MarkChannelAsUnreadFromPost(rootPost.Id, th.BasicUser.Id, true, true)
_, appErr = th.App.MarkChannelAsUnreadFromPost(rootPost.Id, th.BasicUser.Id, true)
require.Nil(t, appErr)
threadMembership, appErr := th.App.GetThreadMembershipForUser(th.BasicUser.Id, rootPost.Id)
require.Nil(t, appErr)
require.NotNil(t, threadMembership)
assert.Equal(t, int64(1), threadMembership.UnreadMentions)
assert.Zero(t, threadMembership.UnreadMentions)
})
})
}

2
app/notification.go
Просмотреть файл

@@ -294,7 +294,7 @@ func (a *App) SendNotifications(post *model.Post, team *model.Team, channel *mod
umc := make(chan *model.AppError, 1)
go func(userID string) {
defer close(umc)
nErr := a.Srv().Store.Channel().IncrementMentionCount(post.ChannelId, userID, *a.Config().ServiceSettings.ThreadAutoFollow, post.RootId == "")
nErr := a.Srv().Store.Channel().IncrementMentionCount(post.ChannelId, userID, post.RootId == "")
if nErr != nil {
umc <- model.NewAppError("SendNotifications", "app.channel.increment_mention_count.app_error", nil, nErr.Error(), http.StatusInternalServerError)
return

26
app/opentracing/opentracing_layer.go
Просмотреть файл

@@ -11579,7 +11579,7 @@ func (a *OpenTracingAppLayer) MakePermissionError(s *model.Session, permissions
return resultVar0
}
func (a *OpenTracingAppLayer) MarkChannelAsUnreadFromPost(postID string, userID string, collapsedThreadsSupported bool, followThread bool) (*model.ChannelUnreadAt, *model.AppError) {
func (a *OpenTracingAppLayer) MarkChannelAsUnreadFromPost(postID string, userID string, collapsedThreadsSupported bool) (*model.ChannelUnreadAt, *model.AppError) {
origCtx := a.ctx
span, newCtx := tracing.StartSpanWithParentByContext(a.ctx, "app.MarkChannelAsUnreadFromPost")
@@ -11591,7 +11591,7 @@ func (a *OpenTracingAppLayer) MarkChannelAsUnreadFromPost(postID string, userID
}()
defer span.Finish()
resultVar0, resultVar1 := a.app.MarkChannelAsUnreadFromPost(postID, userID, collapsedThreadsSupported, followThread)
resultVar0, resultVar1 := a.app.MarkChannelAsUnreadFromPost(postID, userID, collapsedThreadsSupported)
if resultVar1 != nil {
span.LogFields(spanlog.Error(resultVar1))
@@ -15973,28 +15973,6 @@ func (a *OpenTracingAppLayer) UpdateChannel(channel *model.Channel) (*model.Chan
return resultVar0, resultVar1
}
func (a *OpenTracingAppLayer) UpdateChannelLastViewedAt(channelIDs []string, userID string) *model.AppError {
origCtx := a.ctx
span, newCtx := tracing.StartSpanWithParentByContext(a.ctx, "app.UpdateChannelLastViewedAt")
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.UpdateChannelLastViewedAt(channelIDs, userID)
if resultVar0 != nil {
span.LogFields(spanlog.Error(resultVar0))
ext.Error.Set(span, true)
}
return resultVar0
}
func (a *OpenTracingAppLayer) UpdateChannelMemberNotifyProps(data map[string]string, channelID string, userID string) (*model.ChannelMember, *model.AppError) {
origCtx := a.ctx
span, newCtx := tracing.StartSpanWithParentByContext(a.ctx, "app.UpdateChannelMemberNotifyProps")

2
app/post_test.go
Просмотреть файл

@@ -2314,7 +2314,7 @@ func TestCollapsedThreadFetch(t *testing.T) {
thread, nErr := th.App.Srv().Store.Thread().Get(postRoot.Id)
require.NoError(t, nErr)
require.Len(t, thread.Participants, 1)
th.App.MarkChannelAsUnreadFromPost(postRoot.Id, user1.Id, true, true)
th.App.MarkChannelAsUnreadFromPost(postRoot.Id, user1.Id, true)
l, err := th.App.GetPostsForChannelAroundLastUnread(channel.Id, user1.Id, 10, 10, true, true, false)
require.Nil(t, err)
require.Len(t, l.Order, 1)

59
model/thread.go
Просмотреть файл

@@ -3,11 +3,24 @@
package model
// Thread tracks the metadata associated with a root post and its reply posts.
//
// Note that Thread metadata does not exist until the first reply to a root post.
type Thread struct {
PostId string `json:"id"`
ChannelId string `json:"channel_id"`
ReplyCount int64 `json:"reply_count"`
LastReplyAt int64 `json:"last_reply_at"`
// PostId is the root post of the thread.
PostId string `json:"id"`
// ChannelId is the channel in which the thread was posted.
ChannelId string `json:"channel_id"`
// ReplyCount is the number of replies to the thread (excluding deleted posts).
ReplyCount int64 `json:"reply_count"`
// LastReplyAt is the timestamp of the most recent post to the thread.
LastReplyAt int64 `json:"last_reply_at"`
// Participants is a list of user ids that have replied to the thread, sorted by the oldest
// to newest. Note that the root post author is not included in this list until they reply.
Participants StringArray `json:"participants"`
}
@@ -62,11 +75,37 @@ func (o *Thread) Etag() string {
return Etag(o.PostId, o.LastReplyAt)
}
// ThreadMembership models the relationship between a user and a thread of posts, with a similar
// data structure as ChannelMembership.
type ThreadMembership struct {
PostId string `json:"post_id"`
UserId string `json:"user_id"`
Following bool `json:"following"`
LastViewed int64 `json:"last_view_at"`
LastUpdated int64 `json:"last_update_at"`
UnreadMentions int64 `json:"unread_mentions"`
// PostId is the root post id of the thread in question.
PostId string `json:"post_id"`
// UserId is the user whose membership in the thread is being tracked.
UserId string `json:"user_id"`
// Following tracks whether the user is following the given thread. This defaults to true
// when a ThreadMembership record is created (a record doesn't exist until the user first
// starts following the thread), but the user can stop following or resume following at
// will.
Following bool `json:"following"`
// LastUpdated is either the creation time of the membership record, or the last time the
// membership record was changed (e.g. started/stopped following, viewed thread, mention
// count change).
//
// This field is used to constrain queries of thread memberships to those updated after
// a given timestamp (e.g. on websocket reconnect). It's also used as the time column for
// deletion decisions during any configured retention policy.
LastUpdated int64 `json:"last_update_at"`
// LastViewed is the last time the user viewed this thread. It is the thread analogue to
// the ChannelMembership's LastViewedAt and is used to decide when there are new replies
// for the user and where the user should start reading.
LastViewed int64 `json:"last_view_at"`
// UnreadMentions is the number of unseen at-mentions for the user in the given thread. It
// is the thread analogue to the ChannelMembership's MentionCount, and is used to highlight
// threads with the mention count.
UnreadMentions int64 `json:"unread_mentions"`
}

106
store/opentracinglayer/opentracinglayer.go
Просмотреть файл

@@ -1748,7 +1748,7 @@ func (s *OpenTracingLayerChannelStore) GroupSyncedChannelCount() (int64, error)
return result, err
}
func (s *OpenTracingLayerChannelStore) IncrementMentionCount(channelID string, userID string, updateThreads bool, isRoot bool) error {
func (s *OpenTracingLayerChannelStore) IncrementMentionCount(channelID string, userID string, isRoot bool) error {
origCtx := s.Root.Store.Context()
span, newCtx := tracing.StartSpanWithParentByContext(s.Root.Store.Context(), "ChannelStore.IncrementMentionCount")
s.Root.Store.SetContext(newCtx)
@@ -1757,7 +1757,7 @@ func (s *OpenTracingLayerChannelStore) IncrementMentionCount(channelID string, u
}()
defer span.Finish()
err := s.ChannelStore.IncrementMentionCount(channelID, userID, updateThreads, isRoot)
err := s.ChannelStore.IncrementMentionCount(channelID, userID, isRoot)
if err != nil {
span.LogFields(spanlog.Error(err))
ext.Error.Set(span, true)
@@ -2302,7 +2302,7 @@ func (s *OpenTracingLayerChannelStore) UpdateLastViewedAt(channelIds []string, u
return result, err
}
func (s *OpenTracingLayerChannelStore) UpdateLastViewedAtPost(unreadPost *model.Post, userID string, mentionCount int, mentionCountRoot int, updateThreads bool, setUnreadCountRoot bool) (*model.ChannelUnreadAt, error) {
func (s *OpenTracingLayerChannelStore) UpdateLastViewedAtPost(unreadPost *model.Post, userID string, mentionCount int, mentionCountRoot int, setUnreadCountRoot bool) (*model.ChannelUnreadAt, error) {
origCtx := s.Root.Store.Context()
span, newCtx := tracing.StartSpanWithParentByContext(s.Root.Store.Context(), "ChannelStore.UpdateLastViewedAtPost")
s.Root.Store.SetContext(newCtx)
@@ -2311,7 +2311,7 @@ func (s *OpenTracingLayerChannelStore) UpdateLastViewedAtPost(unreadPost *model.
}()
defer span.Finish()
result, err := s.ChannelStore.UpdateLastViewedAtPost(unreadPost, userID, mentionCount, mentionCountRoot, updateThreads, setUnreadCountRoot)
result, err := s.ChannelStore.UpdateLastViewedAtPost(unreadPost, userID, mentionCount, mentionCountRoot, setUnreadCountRoot)
if err != nil {
span.LogFields(spanlog.Error(err))
ext.Error.Set(span, true)
@@ -9231,24 +9231,6 @@ func (s *OpenTracingLayerThreadStore) CollectThreadsWithNewerReplies(userId stri
return result, err
}
func (s *OpenTracingLayerThreadStore) Delete(postID string) error {
origCtx := s.Root.Store.Context()
span, newCtx := tracing.StartSpanWithParentByContext(s.Root.Store.Context(), "ThreadStore.Delete")
s.Root.Store.SetContext(newCtx)
defer func() {
s.Root.Store.SetContext(origCtx)
}()
defer span.Finish()
err := s.ThreadStore.Delete(postID)
if err != nil {
span.LogFields(spanlog.Error(err))
ext.Error.Set(span, true)
}
return err
}
func (s *OpenTracingLayerThreadStore) DeleteMembershipForUser(userId string, postID string) error {
origCtx := s.Root.Store.Context()
span, newCtx := tracing.StartSpanWithParentByContext(s.Root.Store.Context(), "ThreadStore.DeleteMembershipForUser")
@@ -9555,76 +9537,22 @@ func (s *OpenTracingLayerThreadStore) PermanentDeleteBatchThreadMembershipsForRe
return result, resultVar1, err
}
func (s *OpenTracingLayerThreadStore) Save(thread *model.Thread) (*model.Thread, error) {
func (s *OpenTracingLayerThreadStore) UpdateLastViewedByThreadIds(userId string, threadIds []string, timestamp int64) error {
origCtx := s.Root.Store.Context()
span, newCtx := tracing.StartSpanWithParentByContext(s.Root.Store.Context(), "ThreadStore.Save")
span, newCtx := tracing.StartSpanWithParentByContext(s.Root.Store.Context(), "ThreadStore.UpdateLastViewedByThreadIds")
s.Root.Store.SetContext(newCtx)
defer func() {
s.Root.Store.SetContext(origCtx)
}()
defer span.Finish()
result, err := s.ThreadStore.Save(thread)
err := s.ThreadStore.UpdateLastViewedByThreadIds(userId, threadIds, timestamp)
if err != nil {
span.LogFields(spanlog.Error(err))
ext.Error.Set(span, true)
}
return result, err
}
func (s *OpenTracingLayerThreadStore) SaveMembership(membership *model.ThreadMembership) (*model.ThreadMembership, error) {
origCtx := s.Root.Store.Context()
span, newCtx := tracing.StartSpanWithParentByContext(s.Root.Store.Context(), "ThreadStore.SaveMembership")
s.Root.Store.SetContext(newCtx)
defer func() {
s.Root.Store.SetContext(origCtx)
}()
defer span.Finish()
result, err := s.ThreadStore.SaveMembership(membership)
if err != nil {
span.LogFields(spanlog.Error(err))
ext.Error.Set(span, true)
}
return result, err
}
func (s *OpenTracingLayerThreadStore) SaveMultiple(thread []*model.Thread) ([]*model.Thread, int, error) {
origCtx := s.Root.Store.Context()
span, newCtx := tracing.StartSpanWithParentByContext(s.Root.Store.Context(), "ThreadStore.SaveMultiple")
s.Root.Store.SetContext(newCtx)
defer func() {
s.Root.Store.SetContext(origCtx)
}()
defer span.Finish()
result, resultVar1, err := s.ThreadStore.SaveMultiple(thread)
if err != nil {
span.LogFields(spanlog.Error(err))
ext.Error.Set(span, true)
}
return result, resultVar1, err
}
func (s *OpenTracingLayerThreadStore) Update(thread *model.Thread) (*model.Thread, error) {
origCtx := s.Root.Store.Context()
span, newCtx := tracing.StartSpanWithParentByContext(s.Root.Store.Context(), "ThreadStore.Update")
s.Root.Store.SetContext(newCtx)
defer func() {
s.Root.Store.SetContext(origCtx)
}()
defer span.Finish()
result, err := s.ThreadStore.Update(thread)
if err != nil {
span.LogFields(spanlog.Error(err))
ext.Error.Set(span, true)
}
return result, err
return err
}
func (s *OpenTracingLayerThreadStore) UpdateMembership(membership *model.ThreadMembership) (*model.ThreadMembership, error) {
@@ -9645,24 +9573,6 @@ func (s *OpenTracingLayerThreadStore) UpdateMembership(membership *model.ThreadM
return result, err
}
func (s *OpenTracingLayerThreadStore) UpdateUnreadsByChannel(userId string, changedThreads []string, timestamp int64, updateViewedTimestamp bool) error {
origCtx := s.Root.Store.Context()
span, newCtx := tracing.StartSpanWithParentByContext(s.Root.Store.Context(), "ThreadStore.UpdateUnreadsByChannel")
s.Root.Store.SetContext(newCtx)
defer func() {
s.Root.Store.SetContext(origCtx)
}()
defer span.Finish()
err := s.ThreadStore.UpdateUnreadsByChannel(userId, changedThreads, timestamp, updateViewedTimestamp)
if err != nil {
span.LogFields(spanlog.Error(err))
ext.Error.Set(span, true)
}
return err
}
func (s *OpenTracingLayerTokenStore) Cleanup(expiryTime int64) {
origCtx := s.Root.Store.Context()
span, newCtx := tracing.StartSpanWithParentByContext(s.Root.Store.Context(), "TokenStore.Cleanup")

123
store/retrylayer/retrylayer.go
Просмотреть файл

@@ -1975,11 +1975,11 @@ func (s *RetryLayerChannelStore) GroupSyncedChannelCount() (int64, error) {
}
func (s *RetryLayerChannelStore) IncrementMentionCount(channelID string, userID string, updateThreads bool, isRoot bool) error {
func (s *RetryLayerChannelStore) IncrementMentionCount(channelID string, userID string, isRoot bool) error {
tries := 0
for {
err := s.ChannelStore.IncrementMentionCount(channelID, userID, updateThreads, isRoot)
err := s.ChannelStore.IncrementMentionCount(channelID, userID, isRoot)
if err == nil {
return nil
}
@@ -2548,11 +2548,11 @@ func (s *RetryLayerChannelStore) UpdateLastViewedAt(channelIds []string, userID
}
func (s *RetryLayerChannelStore) UpdateLastViewedAtPost(unreadPost *model.Post, userID string, mentionCount int, mentionCountRoot int, updateThreads bool, setUnreadCountRoot bool) (*model.ChannelUnreadAt, error) {
func (s *RetryLayerChannelStore) UpdateLastViewedAtPost(unreadPost *model.Post, userID string, mentionCount int, mentionCountRoot int, setUnreadCountRoot bool) (*model.ChannelUnreadAt, error) {
tries := 0
for {
result, err := s.ChannelStore.UpdateLastViewedAtPost(unreadPost, userID, mentionCount, mentionCountRoot, updateThreads, setUnreadCountRoot)
result, err := s.ChannelStore.UpdateLastViewedAtPost(unreadPost, userID, mentionCount, mentionCountRoot, setUnreadCountRoot)
if err == nil {
return result, nil
}
@@ -10540,27 +10540,6 @@ func (s *RetryLayerThreadStore) CollectThreadsWithNewerReplies(userId string, ch
}
func (s *RetryLayerThreadStore) Delete(postID string) error {
tries := 0
for {
err := s.ThreadStore.Delete(postID)
if err == nil {
return nil
}
if !isRepeatableError(err) {
return err
}
tries++
if tries >= 3 {
err = errors.Wrap(err, "giving up after 3 consecutive repeatable transaction failures")
return err
}
timepkg.Sleep(100 * timepkg.Millisecond)
}
}
func (s *RetryLayerThreadStore) DeleteMembershipForUser(userId string, postID string) error {
tries := 0
@@ -10918,84 +10897,21 @@ func (s *RetryLayerThreadStore) PermanentDeleteBatchThreadMembershipsForRetentio
}
func (s *RetryLayerThreadStore) Save(thread *model.Thread) (*model.Thread, error) {
func (s *RetryLayerThreadStore) UpdateLastViewedByThreadIds(userId string, threadIds []string, timestamp int64) error {
tries := 0
for {
result, err := s.ThreadStore.Save(thread)
err := s.ThreadStore.UpdateLastViewedByThreadIds(userId, threadIds, timestamp)
if err == nil {
return result, nil
return nil
}
if !isRepeatableError(err) {
return result, err
return err
}
tries++
if tries >= 3 {
err = errors.Wrap(err, "giving up after 3 consecutive repeatable transaction failures")
return result, err
}
timepkg.Sleep(100 * timepkg.Millisecond)
}
}
func (s *RetryLayerThreadStore) SaveMembership(membership *model.ThreadMembership) (*model.ThreadMembership, error) {
tries := 0
for {
result, err := s.ThreadStore.SaveMembership(membership)
if err == nil {
return result, nil
}
if !isRepeatableError(err) {
return result, err
}
tries++
if tries >= 3 {
err = errors.Wrap(err, "giving up after 3 consecutive repeatable transaction failures")
return result, err
}
timepkg.Sleep(100 * timepkg.Millisecond)
}
}
func (s *RetryLayerThreadStore) SaveMultiple(thread []*model.Thread) ([]*model.Thread, int, error) {
tries := 0
for {
result, resultVar1, err := s.ThreadStore.SaveMultiple(thread)
if err == nil {
return result, resultVar1, nil
}
if !isRepeatableError(err) {
return result, resultVar1, err
}
tries++
if tries >= 3 {
err = errors.Wrap(err, "giving up after 3 consecutive repeatable transaction failures")
return result, resultVar1, err
}
timepkg.Sleep(100 * timepkg.Millisecond)
}
}
func (s *RetryLayerThreadStore) Update(thread *model.Thread) (*model.Thread, error) {
tries := 0
for {
result, err := s.ThreadStore.Update(thread)
if err == nil {
return result, nil
}
if !isRepeatableError(err) {
return result, err
}
tries++
if tries >= 3 {
err = errors.Wrap(err, "giving up after 3 consecutive repeatable transaction failures")
return result, err
return err
}
timepkg.Sleep(100 * timepkg.Millisecond)
}
@@ -11023,27 +10939,6 @@ func (s *RetryLayerThreadStore) UpdateMembership(membership *model.ThreadMembers
}
func (s *RetryLayerThreadStore) UpdateUnreadsByChannel(userId string, changedThreads []string, timestamp int64, updateViewedTimestamp bool) error {
tries := 0
for {
err := s.ThreadStore.UpdateUnreadsByChannel(userId, changedThreads, timestamp, updateViewedTimestamp)
if err == nil {
return nil
}
if !isRepeatableError(err) {
return err
}
tries++
if tries >= 3 {
err = errors.Wrap(err, "giving up after 3 consecutive repeatable transaction failures")
return err
}
timepkg.Sleep(100 * timepkg.Millisecond)
}
}
func (s *RetryLayerTokenStore) Cleanup(expiryTime int64) {
s.TokenStore.Cleanup(expiryTime)

30
store/sqlstore/channel_store.go
Просмотреть файл

@@ -2380,7 +2380,7 @@ func (s SqlChannelStore) UpdateLastViewedAt(channelIds []string, userId string,
times[t.Id] = t.LastPostAt
}
if updateThreads {
s.Thread().UpdateUnreadsByChannel(userId, threadsToUpdate, now, true)
s.Thread().UpdateLastViewedByThreadIds(userId, threadsToUpdate, now)
}
return times, nil
}
@@ -2425,7 +2425,7 @@ func (s SqlChannelStore) UpdateLastViewedAt(channelIds []string, userId string,
}
if updateThreads {
s.Thread().UpdateUnreadsByChannel(userId, threadsToUpdate, now, true)
s.Thread().UpdateLastViewedByThreadIds(userId, threadsToUpdate, now)
}
return times, nil
}
@@ -2478,16 +2478,8 @@ func (s SqlChannelStore) CountPostsAfter(channelId string, timestamp int64, user
// UpdateLastViewedAtPost updates a ChannelMember as if the user last read the channel at the time of the given post.
// If the provided mentionCount is -1, the given post and all posts after it are considered to be mentions. Returns
// an updated model.ChannelUnreadAt that can be returned to the client.
func (s SqlChannelStore) UpdateLastViewedAtPost(unreadPost *model.Post, userID string, mentionCount, mentionCountRoot int, updateThreads bool, setUnreadCountRoot bool) (*model.ChannelUnreadAt, error) {
var threadsToUpdate []string
func (s SqlChannelStore) UpdateLastViewedAtPost(unreadPost *model.Post, userID string, mentionCount, mentionCountRoot int, setUnreadCountRoot bool) (*model.ChannelUnreadAt, error) {
unreadDate := unreadPost.CreateAt - 1
if updateThreads {
var err error
threadsToUpdate, err = s.Thread().CollectThreadsWithNewerReplies(userID, []string{unreadPost.ChannelId}, unreadDate)
if err != nil {
return nil, err
}
}
unread, unreadRoot, err := s.CountPostsAfter(unreadPost.ChannelId, unreadDate, "")
if err != nil {
@@ -2554,22 +2546,11 @@ func (s SqlChannelStore) UpdateLastViewedAtPost(unreadPost *model.Post, userID s
return nil, errors.Wrapf(err, "failed to get ChannelMember with channelId=%s", unreadPost.ChannelId)
}
if updateThreads {
s.Thread().UpdateUnreadsByChannel(userID, threadsToUpdate, unreadDate, false)
}
return result, nil
}
func (s SqlChannelStore) IncrementMentionCount(channelId string, userId string, updateThreads, isRoot bool) error {
func (s SqlChannelStore) IncrementMentionCount(channelId string, userId string, isRoot bool) error {
now := model.GetMillis()
var threadsToUpdate []string
if updateThreads {
var err error
threadsToUpdate, err = s.Thread().CollectThreadsWithNewerReplies(userId, []string{channelId}, now)
if err != nil {
return err
}
}
rootInc := 0
if isRoot {
rootInc = 1
@@ -2587,9 +2568,6 @@ func (s SqlChannelStore) IncrementMentionCount(channelId string, userId string,
if err != nil {
return errors.Wrapf(err, "failed to Update ChannelMembers with channelId=%s and userId=%s", channelId, userId)
}
if updateThreads {
s.Thread().UpdateUnreadsByChannel(userId, threadsToUpdate, now, false)
}
return nil
}

105
store/sqlstore/thread_store.go
Просмотреть файл

@@ -6,7 +6,6 @@ package sqlstore
import (
"context"
"database/sql"
"encoding/json"
"strconv"
"sync"
"time"
@@ -32,70 +31,6 @@ func newSqlThreadStore(sqlStore *SqlStore) store.ThreadStore {
}
}
func threadSliceColumns() []string {
return []string{"PostId", "ChannelId", "LastReplyAt", "ReplyCount", "Participants"}
}
func threadToSlice(thread *model.Thread) []interface{} {
return []interface{}{
thread.PostId,
thread.ChannelId,
thread.LastReplyAt,
thread.ReplyCount,
model.ArrayToJSON(thread.Participants),
}
}
func (s *SqlThreadStore) SaveMultiple(threads []*model.Thread) ([]*model.Thread, int, error) {
builder := s.getQueryBuilder().
Insert("Threads").
Columns(threadSliceColumns()...)
for _, thread := range threads {
builder = builder.Values(threadToSlice(thread)...)
}
query, args, err := builder.ToSql()
if err != nil {
return nil, -1, errors.Wrap(err, "thread_tosql")
}
if _, err := s.GetMasterX().Exec(query, args...); err != nil {
return nil, -1, errors.Wrap(err, "failed to save Post")
}
return threads, -1, nil
}
func (s *SqlThreadStore) Save(thread *model.Thread) (*model.Thread, error) {
threads, _, err := s.SaveMultiple([]*model.Thread{thread})
if err != nil {
return nil, err
}
return threads[0], nil
}
func (s *SqlThreadStore) Update(thread *model.Thread) (*model.Thread, error) {
jsonParticipants, err := json.Marshal(thread.Participants)
if err != nil {
return nil, errors.Wrap(err, "failed marshaling thread participants")
}
query, args, err := s.getQueryBuilder().
Update("Threads").
Set("ChannelId", thread.ChannelId).
Set("ReplyCount", thread.ReplyCount).
Set("LastReplyAt", thread.LastReplyAt).
Set("Participants", string(jsonParticipants)).
Where(sq.Eq{"PostId": thread.PostId}).
ToSql()
if err != nil {
return nil, errors.Wrap(err, "thread_tosql")
}
if _, err := s.GetMasterX().Exec(query, args...); err != nil {
return nil, errors.Wrapf(err, "failed to update thread with id=%s", thread.PostId)
}
return thread, nil
}
func (s *SqlThreadStore) Get(id string) (*model.Thread, error) {
var thread model.Thread
query, args, err := s.getQueryBuilder().
@@ -575,6 +510,9 @@ func (s *SqlThreadStore) GetThreadForUser(teamId string, threadMembership *model
return result, nil
}
// MarkAllAsReadInChannels marks all threads for the given user in the given channels as read from
// the current time.
func (s *SqlThreadStore) MarkAllAsReadInChannels(userID string, channelIDs []string) error {
threadIDs := []string{}
@@ -599,6 +537,7 @@ func (s *SqlThreadStore) MarkAllAsReadInChannels(userID string, channelIDs []str
Where(sq.Eq{"UserId": userID}).
Set("LastViewed", timestamp).
Set("UnreadMentions", 0).
Set("LastUpdated", model.GetMillis()).
ToSql()
if _, err := s.GetMasterX().Exec(query, args...); err != nil {
return errors.Wrapf(err, "failed to update thread read state for user id=%s", userID)
@@ -606,6 +545,9 @@ func (s *SqlThreadStore) MarkAllAsReadInChannels(userID string, channelIDs []str
return nil
}
// MarkAllAsRead marks all threads for the given user in the given team as read from the current
// time.
func (s *SqlThreadStore) MarkAllAsRead(userId, teamId string) error {
memberships, err := s.GetMembershipsForUser(userId, teamId)
if err != nil {
@@ -622,6 +564,7 @@ func (s *SqlThreadStore) MarkAllAsRead(userId, teamId string) error {
Where(sq.Eq{"UserId": userId}).
Set("LastViewed", timestamp).
Set("UnreadMentions", 0).
Set("LastUpdated", model.GetMillis()).
ToSql()
if _, err := s.GetMasterX().Exec(query, args...); err != nil {
return errors.Wrapf(err, "failed to update thread read state for user id=%s", userId)
@@ -629,12 +572,14 @@ func (s *SqlThreadStore) MarkAllAsRead(userId, teamId string) error {
return nil
}
// MarkAsRead marks the given thread for the given user as unread from the given timestamp.
func (s *SqlThreadStore) MarkAsRead(userId, threadId string, timestamp int64) error {
query, args, _ := s.getQueryBuilder().
Update("ThreadMemberships").
Where(sq.Eq{"UserId": userId}).
Where(sq.Eq{"PostId": threadId}).
Set("LastViewed", timestamp).
Set("LastUpdated", model.GetMillis()).
ToSql()
if _, err := s.GetMasterX().Exec(query, args...); err != nil {
return errors.Wrapf(err, "failed to update thread read state for user id=%s thread_id=%v", userId, threadId)
@@ -642,19 +587,6 @@ func (s *SqlThreadStore) MarkAsRead(userId, threadId string, timestamp int64) er
return nil
}
func (s *SqlThreadStore) Delete(threadId string) error {
query, args, _ := s.getQueryBuilder().Delete("Threads").Where(sq.Eq{"PostId": threadId}).ToSql()
if _, err := s.GetMasterX().Exec(query, args...); err != nil {
return errors.Wrap(err, "failed to update threads")
}
return nil
}
func (s *SqlThreadStore) SaveMembership(membership *model.ThreadMembership) (*model.ThreadMembership, error) {
return s.saveMembership(s.GetMasterX(), membership)
}
func (s *SqlThreadStore) saveMembership(ex sqlxExecutor, membership *model.ThreadMembership) (*model.ThreadMembership, error) {
query, args, err := s.getQueryBuilder().
Insert("ThreadMemberships").
@@ -875,19 +807,20 @@ func (s *SqlThreadStore) CollectThreadsWithNewerReplies(userId string, channelId
return changedThreads, nil
}
func (s *SqlThreadStore) UpdateUnreadsByChannel(userId string, changedThreads []string, timestamp int64, updateViewedTimestamp bool) error {
if len(changedThreads) == 0 {
// UpdateLastViewedByThreadIds marks the given threads as read up to the given timestamp. If there
// are no newer posts, it effectively marks the thread as read. If there are newer posts, say
// because the user explicitly marked a past post as unread, the thread will be considered unread
// past the given timestamp.
func (s *SqlThreadStore) UpdateLastViewedByThreadIds(userId string, threadIds []string, timestamp int64) error {
if len(threadIds) == 0 {
return nil
}
qb := s.getQueryBuilder().
Update("ThreadMemberships").
Where(sq.Eq{"UserId": userId, "PostId": changedThreads}).
Set("LastUpdated", timestamp)
if updateViewedTimestamp {
qb = qb.Set("LastViewed", timestamp)
}
Where(sq.Eq{"UserId": userId, "PostId": threadIds}).
Set("LastViewed", timestamp).
Set("LastUpdated", model.GetMillis())
updateQuery, updateArgs, _ := qb.ToSql()
if _, err := s.GetMasterX().Exec(updateQuery, updateArgs...); err != nil {

11
store/store.go
Просмотреть файл

@@ -224,9 +224,9 @@ type ChannelStore interface {
PermanentDeleteMembersByUser(userID string) error
PermanentDeleteMembersByChannel(channelID string) error
UpdateLastViewedAt(channelIds []string, userID string, updateThreads bool) (map[string]int64, error)
UpdateLastViewedAtPost(unreadPost *model.Post, userID string, mentionCount, mentionCountRoot int, updateThreads bool, setUnreadCountRoot bool) (*model.ChannelUnreadAt, error)
UpdateLastViewedAtPost(unreadPost *model.Post, userID string, mentionCount, mentionCountRoot int, setUnreadCountRoot bool) (*model.ChannelUnreadAt, error)
CountPostsAfter(channelID string, timestamp int64, userID string) (int, int, error)
IncrementMentionCount(channelID string, userID string, updateThreads, isRoot bool) error
IncrementMentionCount(channelID string, userID string, isRoot bool) error
AnalyticsTypeCount(teamID string, channelType model.ChannelType) (int64, error)
GetMembersForUser(teamID string, userID string) (model.ChannelMembers, error)
GetTeamMembersForChannel(channelID string) ([]string, error)
@@ -292,28 +292,23 @@ type ChannelMemberHistoryStore interface {
type ThreadStore interface {
GetThreadFollowers(threadID string, fetchOnlyActive bool) ([]string, error)
SaveMultiple(thread []*model.Thread) ([]*model.Thread, int, error)
Save(thread *model.Thread) (*model.Thread, error)
Update(thread *model.Thread) (*model.Thread, error)
Get(id string) (*model.Thread, error)
GetThreadsForUser(userId, teamID string, opts model.GetUserThreadsOpts) (*model.Threads, error)
GetThreadForUser(teamID string, threadMembership *model.ThreadMembership, extended bool) (*model.ThreadResponse, error)
GetTeamsUnreadForUser(userID string, teamIDs []string) (map[string]*model.TeamUnread, error)
Delete(postID string) error
GetPosts(threadID string, since int64) ([]*model.Post, error)
MarkAllAsRead(userID, teamID string) error
MarkAllAsReadInChannels(userID string, channelIDs []string) error
MarkAsRead(userID, threadID string, timestamp int64) error
SaveMembership(membership *model.ThreadMembership) (*model.ThreadMembership, error)
UpdateMembership(membership *model.ThreadMembership) (*model.ThreadMembership, error)
GetMembershipsForUser(userId, teamID string) ([]*model.ThreadMembership, error)
GetMembershipForUser(userId, postID string) (*model.ThreadMembership, error)
DeleteMembershipForUser(userId, postID string) error
MaintainMembership(userID, postID string, opts ThreadMembershipOpts) (*model.ThreadMembership, error)
CollectThreadsWithNewerReplies(userId string, channelIds []string, timestamp int64) ([]string, error)
UpdateUnreadsByChannel(userId string, changedThreads []string, timestamp int64, updateViewedTimestamp bool) error
UpdateLastViewedByThreadIds(userId string, threadIds []string, timestamp int64) error
PermanentDeleteBatchForRetentionPolicies(now, globalPolicyEndTime, limit int64, cursor model.RetentionPolicyCursor) (int64, model.RetentionPolicyCursor, error)
PermanentDeleteBatchThreadMembershipsForRetentionPolicies(now, globalPolicyEndTime, limit int64, cursor model.RetentionPolicyCursor) (int64, model.RetentionPolicyCursor, error)
DeleteOrphanedRows(limit int) (deleted int64, err error)

8
store/storetest/channel_store.go
Просмотреть файл

@@ -4790,16 +4790,16 @@ func testChannelStoreIncrementMentionCount(t *testing.T, ss store.Store) {
_, err := ss.Channel().SaveMember(&m1)
require.NoError(t, err)
err = ss.Channel().IncrementMentionCount(m1.ChannelId, m1.UserId, false, false)
err = ss.Channel().IncrementMentionCount(m1.ChannelId, m1.UserId, false)
require.NoError(t, err, "failed to update")
err = ss.Channel().IncrementMentionCount(m1.ChannelId, "missing id", false, false)
err = ss.Channel().IncrementMentionCount(m1.ChannelId, "missing id", false)
require.NoError(t, err, "failed to update")
err = ss.Channel().IncrementMentionCount("missing id", m1.UserId, false, false)
err = ss.Channel().IncrementMentionCount("missing id", m1.UserId, false)
require.NoError(t, err, "failed to update")
err = ss.Channel().IncrementMentionCount("missing id", "missing id", false, false)
err = ss.Channel().IncrementMentionCount("missing id", "missing id", false)
require.NoError(t, err, "failed to update")
}

24
store/storetest/mocks/ChannelStore.go
Просмотреть файл

@@ -1508,13 +1508,13 @@ func (_m *ChannelStore) GroupSyncedChannelCount() (int64, error) {
return r0, r1
}
// IncrementMentionCount provides a mock function with given fields: channelID, userID, updateThreads, isRoot
func (_m *ChannelStore) IncrementMentionCount(channelID string, userID string, updateThreads bool, isRoot bool) error {
ret := _m.Called(channelID, userID, updateThreads, isRoot)
// IncrementMentionCount provides a mock function with given fields: channelID, userID, isRoot
func (_m *ChannelStore) IncrementMentionCount(channelID string, userID string, isRoot bool) error {
ret := _m.Called(channelID, userID, isRoot)
var r0 error
if rf, ok := ret.Get(0).(func(string, string, bool, bool) error); ok {
r0 = rf(channelID, userID, updateThreads, isRoot)
if rf, ok := ret.Get(0).(func(string, string, bool) error); ok {
r0 = rf(channelID, userID, isRoot)
} else {
r0 = ret.Error(0)
}
@@ -2031,13 +2031,13 @@ func (_m *ChannelStore) UpdateLastViewedAt(channelIds []string, userID string, u
return r0, r1
}
// UpdateLastViewedAtPost provides a mock function with given fields: unreadPost, userID, mentionCount, mentionCountRoot, updateThreads, setUnreadCountRoot
func (_m *ChannelStore) UpdateLastViewedAtPost(unreadPost *model.Post, userID string, mentionCount int, mentionCountRoot int, updateThreads bool, setUnreadCountRoot bool) (*model.ChannelUnreadAt, error) {
ret := _m.Called(unreadPost, userID, mentionCount, mentionCountRoot, updateThreads, setUnreadCountRoot)
// UpdateLastViewedAtPost provides a mock function with given fields: unreadPost, userID, mentionCount, mentionCountRoot, setUnreadCountRoot
func (_m *ChannelStore) UpdateLastViewedAtPost(unreadPost *model.Post, userID string, mentionCount int, mentionCountRoot int, setUnreadCountRoot bool) (*model.ChannelUnreadAt, error) {
ret := _m.Called(unreadPost, userID, mentionCount, mentionCountRoot, setUnreadCountRoot)
var r0 *model.ChannelUnreadAt
if rf, ok := ret.Get(0).(func(*model.Post, string, int, int, bool, bool) *model.ChannelUnreadAt); ok {
r0 = rf(unreadPost, userID, mentionCount, mentionCountRoot, updateThreads, setUnreadCountRoot)
if rf, ok := ret.Get(0).(func(*model.Post, string, int, int, bool) *model.ChannelUnreadAt); ok {
r0 = rf(unreadPost, userID, mentionCount, mentionCountRoot, setUnreadCountRoot)
} else {
if ret.Get(0) != nil {
r0 = ret.Get(0).(*model.ChannelUnreadAt)
@@ -2045,8 +2045,8 @@ func (_m *ChannelStore) UpdateLastViewedAtPost(unreadPost *model.Post, userID st
}
var r1 error
if rf, ok := ret.Get(1).(func(*model.Post, string, int, int, bool, bool) error); ok {
r1 = rf(unreadPost, userID, mentionCount, mentionCountRoot, updateThreads, setUnreadCountRoot)
if rf, ok := ret.Get(1).(func(*model.Post, string, int, int, bool) error); ok {
r1 = rf(unreadPost, userID, mentionCount, mentionCountRoot, setUnreadCountRoot)
} else {
r1 = ret.Error(1)
}

129
store/storetest/mocks/ThreadStore.go
Просмотреть файл

@@ -38,20 +38,6 @@ func (_m *ThreadStore) CollectThreadsWithNewerReplies(userId string, channelIds
return r0, r1
}
// Delete provides a mock function with given fields: postID
func (_m *ThreadStore) Delete(postID string) error {
ret := _m.Called(postID)
var r0 error
if rf, ok := ret.Get(0).(func(string) error); ok {
r0 = rf(postID)
} else {
r0 = ret.Error(0)
}
return r0
}
// DeleteMembershipForUser provides a mock function with given fields: userId, postID
func (_m *ThreadStore) DeleteMembershipForUser(userId string, postID string) error {
ret := _m.Called(userId, postID)
@@ -413,103 +399,18 @@ func (_m *ThreadStore) PermanentDeleteBatchThreadMembershipsForRetentionPolicies
return r0, r1, r2
}
// Save provides a mock function with given fields: thread
func (_m *ThreadStore) Save(thread *model.Thread) (*model.Thread, error) {
ret := _m.Called(thread)
// UpdateLastViewedByThreadIds provides a mock function with given fields: userId, threadIds, timestamp
func (_m *ThreadStore) UpdateLastViewedByThreadIds(userId string, threadIds []string, timestamp int64) error {
ret := _m.Called(userId, threadIds, timestamp)
var r0 *model.Thread
if rf, ok := ret.Get(0).(func(*model.Thread) *model.Thread); ok {
r0 = rf(thread)
var r0 error
if rf, ok := ret.Get(0).(func(string, []string, int64) error); ok {
r0 = rf(userId, threadIds, timestamp)
} else {
if ret.Get(0) != nil {
r0 = ret.Get(0).(*model.Thread)
}
r0 = ret.Error(0)
}
var r1 error
if rf, ok := ret.Get(1).(func(*model.Thread) error); ok {
r1 = rf(thread)
} else {
r1 = ret.Error(1)
}
return r0, r1
}
// SaveMembership provides a mock function with given fields: membership
func (_m *ThreadStore) SaveMembership(membership *model.ThreadMembership) (*model.ThreadMembership, error) {
ret := _m.Called(membership)
var r0 *model.ThreadMembership
if rf, ok := ret.Get(0).(func(*model.ThreadMembership) *model.ThreadMembership); ok {
r0 = rf(membership)
} else {
if ret.Get(0) != nil {
r0 = ret.Get(0).(*model.ThreadMembership)
}
}
var r1 error
if rf, ok := ret.Get(1).(func(*model.ThreadMembership) error); ok {
r1 = rf(membership)
} else {
r1 = ret.Error(1)
}
return r0, r1
}
// SaveMultiple provides a mock function with given fields: thread
func (_m *ThreadStore) SaveMultiple(thread []*model.Thread) ([]*model.Thread, int, error) {
ret := _m.Called(thread)
var r0 []*model.Thread
if rf, ok := ret.Get(0).(func([]*model.Thread) []*model.Thread); ok {
r0 = rf(thread)
} else {
if ret.Get(0) != nil {
r0 = ret.Get(0).([]*model.Thread)
}
}
var r1 int
if rf, ok := ret.Get(1).(func([]*model.Thread) int); ok {
r1 = rf(thread)
} else {
r1 = ret.Get(1).(int)
}
var r2 error
if rf, ok := ret.Get(2).(func([]*model.Thread) error); ok {
r2 = rf(thread)
} else {
r2 = ret.Error(2)
}
return r0, r1, r2
}
// Update provides a mock function with given fields: thread
func (_m *ThreadStore) Update(thread *model.Thread) (*model.Thread, error) {
ret := _m.Called(thread)
var r0 *model.Thread
if rf, ok := ret.Get(0).(func(*model.Thread) *model.Thread); ok {
r0 = rf(thread)
} else {
if ret.Get(0) != nil {
r0 = ret.Get(0).(*model.Thread)
}
}
var r1 error
if rf, ok := ret.Get(1).(func(*model.Thread) error); ok {
r1 = rf(thread)
} else {
r1 = ret.Error(1)
}
return r0, r1
return r0
}
// UpdateMembership provides a mock function with given fields: membership
@@ -534,17 +435,3 @@ func (_m *ThreadStore) UpdateMembership(membership *model.ThreadMembership) (*mo
return r0, r1
}
// UpdateUnreadsByChannel provides a mock function with given fields: userId, changedThreads, timestamp, updateViewedTimestamp
func (_m *ThreadStore) UpdateUnreadsByChannel(userId string, changedThreads []string, timestamp int64, updateViewedTimestamp bool) error {
ret := _m.Called(userId, changedThreads, timestamp, updateViewedTimestamp)
var r0 error
if rf, ok := ret.Get(0).(func(string, []string, int64, bool) error); ok {
r0 = rf(userId, changedThreads, timestamp, updateViewedTimestamp)
} else {
r0 = ret.Error(0)
}
return r0
}

26
store/storetest/post_store.go
Просмотреть файл

@@ -510,15 +510,10 @@ func testPostStoreGetForThread(t *testing.T, ss store.Store) {
_, err = ss.Post().Save(&model.Post{ChannelId: o1.ChannelId, UserId: model.NewId(), Message: NewTestId(), RootId: o1.Id})
require.NoError(t, err)
threadMembership := &model.ThreadMembership{
PostId: o1.Id,
UserId: o1.UserId,
Following: true,
LastViewed: 0,
LastUpdated: 0,
UnreadMentions: 0,
}
_, err = ss.Thread().SaveMembership(threadMembership)
_, err = ss.Thread().MaintainMembership(o1.UserId, o1.Id, store.ThreadMembershipOpts{
Following: true,
UpdateFollowing: true,
})
require.NoError(t, err)
r1, err := ss.Post().Get(context.Background(), o1.Id, false, true, false, o1.UserId)
require.NoError(t, err)
@@ -533,15 +528,10 @@ func testPostStoreGetForThread(t *testing.T, ss store.Store) {
_, err = ss.Post().Save(&model.Post{ChannelId: o1.ChannelId, UserId: model.NewId(), Message: NewTestId(), RootId: o1.Id})
require.NoError(t, err)
threadMembership := &model.ThreadMembership{
PostId: o1.Id,
UserId: o1.UserId,
Following: false,
LastViewed: 0,
LastUpdated: 0,
UnreadMentions: 0,
}
_, err = ss.Thread().SaveMembership(threadMembership)
_, err = ss.Thread().MaintainMembership(o1.UserId, o1.Id, store.ThreadMembershipOpts{
Following: false,
UpdateFollowing: true,
})
require.NoError(t, err)
r1, err := ss.Post().Get(context.Background(), o1.Id, false, true, false, o1.UserId)
require.NoError(t, err)

107
store/storetest/thread_store.go
Просмотреть файл

@@ -16,13 +16,12 @@ import (
)
func TestThreadStore(t *testing.T, ss store.Store, s SqlStore) {
t.Run("ThreadSQLOperations", func(t *testing.T) { testThreadSQLOperations(t, ss, s) })
t.Run("ThreadStorePopulation", func(t *testing.T) { testThreadStorePopulation(t, ss) })
t.Run("ThreadStorePermanentDeleteBatchForRetentionPolicies", func(t *testing.T) {
testThreadStorePermanentDeleteBatchForRetentionPolicies(t, ss)
})
t.Run("ThreadStorePermanentDeleteBatchThreadMembershipsForRetentionPolicies", func(t *testing.T) {
testThreadStorePermanentDeleteBatchThreadMembershipsForRetentionPolicies(t, ss)
testThreadStorePermanentDeleteBatchThreadMembershipsForRetentionPolicies(t, ss, s)
})
t.Run("GetTeamsUnreadForUser", func(t *testing.T) { testGetTeamsUnreadForUser(t, ss) })
}
@@ -263,61 +262,6 @@ func testThreadStorePopulation(t *testing.T, ss store.Store) {
require.Nil(t, thread2)
})
t.Run("Thread last updated is changed when channel is updated after UpdateLastViewedAtPost", func(t *testing.T) {
newPosts := makeSomePosts()
opts := store.ThreadMembershipOpts{
Following: true,
IncrementMentions: false,
UpdateFollowing: true,
UpdateViewedTimestamp: false,
UpdateParticipants: false,
}
_, e := ss.Thread().MaintainMembership(newPosts[0].UserId, newPosts[0].Id, opts)
require.NoError(t, e)
m, err1 := ss.Thread().GetMembershipForUser(newPosts[0].UserId, newPosts[0].Id)
require.NoError(t, err1)
m.LastUpdated -= 1000
_, err := ss.Thread().UpdateMembership(m)
require.NoError(t, err)
_, err = ss.Channel().UpdateLastViewedAtPost(newPosts[0], newPosts[0].UserId, 0, 0, true, true)
require.NoError(t, err)
assert.Eventually(t, func() bool {
m2, err2 := ss.Thread().GetMembershipForUser(newPosts[0].UserId, newPosts[0].Id)
require.NoError(t, err2)
return m2.LastUpdated > m.LastUpdated
}, time.Second, 10*time.Millisecond)
})
t.Run("Thread last updated is changed when channel is updated after IncrementMentionCount", func(t *testing.T) {
newPosts := makeSomePosts()
opts := store.ThreadMembershipOpts{
Following: true,
IncrementMentions: false,
UpdateFollowing: true,
UpdateViewedTimestamp: false,
UpdateParticipants: false,
}
_, e := ss.Thread().MaintainMembership(newPosts[0].UserId, newPosts[0].Id, opts)
require.NoError(t, e)
m, err1 := ss.Thread().GetMembershipForUser(newPosts[0].UserId, newPosts[0].Id)
require.NoError(t, err1)
m.LastUpdated -= 1000
_, err := ss.Thread().UpdateMembership(m)
require.NoError(t, err)
err = ss.Channel().IncrementMentionCount(newPosts[0].ChannelId, newPosts[0].UserId, true, false)
require.NoError(t, err)
assert.Eventually(t, func() bool {
m2, err2 := ss.Thread().GetMembershipForUser(newPosts[0].UserId, newPosts[0].Id)
require.NoError(t, err2)
return m2.LastUpdated > m.LastUpdated
}, time.Second, 10*time.Millisecond)
})
t.Run("Thread last updated is changed when channel is updated after UpdateLastViewedAt", func(t *testing.T) {
newPosts := makeSomePosts()
opts := store.ThreadMembershipOpts{
@@ -394,33 +338,6 @@ func testThreadStorePopulation(t *testing.T, ss store.Store) {
require.NotEqual(t, int64(0), tm.LastViewed)
})
t.Run("Thread last updated is changed when channel is updated after UpdateLastViewedAtPost for mark unread", func(t *testing.T) {
newPosts := makeSomePosts()
opts := store.ThreadMembershipOpts{
Following: true,
IncrementMentions: false,
UpdateFollowing: true,
UpdateViewedTimestamp: false,
UpdateParticipants: false,
}
_, e := ss.Thread().MaintainMembership(newPosts[0].UserId, newPosts[0].Id, opts)
require.NoError(t, e)
m, err1 := ss.Thread().GetMembershipForUser(newPosts[0].UserId, newPosts[0].Id)
require.NoError(t, err1)
m.LastUpdated += 1000
_, err := ss.Thread().UpdateMembership(m)
require.NoError(t, err)
_, err = ss.Channel().UpdateLastViewedAtPost(newPosts[0], newPosts[0].UserId, 0, 0, true, true)
require.NoError(t, err)
assert.Eventually(t, func() bool {
m2, err2 := ss.Thread().GetMembershipForUser(newPosts[0].UserId, newPosts[0].Id)
require.NoError(t, err2)
return m2.LastUpdated < m.LastUpdated
}, time.Second, 10*time.Millisecond)
})
t.Run("Updating post does not make thread unread", func(t *testing.T) {
newPosts := makeSomePosts()
opts := store.ThreadMembershipOpts{
@@ -503,24 +420,6 @@ func testThreadStorePopulation(t *testing.T, ss store.Store) {
})
}
func testThreadSQLOperations(t *testing.T, ss store.Store, s SqlStore) {
t.Run("Save", func(t *testing.T) {
threadToSave := &model.Thread{
PostId: model.NewId(),
ChannelId: model.NewId(),
LastReplyAt: 10,
ReplyCount: 5,
Participants: model.StringArray{model.NewId(), model.NewId()},
}
_, err := ss.Thread().Save(threadToSave)
require.NoError(t, err)
th, err := ss.Thread().Get(threadToSave.PostId)
require.NoError(t, err)
require.Equal(t, threadToSave, th)
})
}
func threadStoreCreateReply(t *testing.T, ss store.Store, channelID, postID, userID string, createAt int64) *model.Post {
reply, err := ss.Post().Save(&model.Post{
ChannelId: channelID,
@@ -607,7 +506,7 @@ func testThreadStorePermanentDeleteBatchForRetentionPolicies(t *testing.T, ss st
assert.Nil(t, thread, "thread should have been deleted by team policy")
}
func testThreadStorePermanentDeleteBatchThreadMembershipsForRetentionPolicies(t *testing.T, ss store.Store) {
func testThreadStorePermanentDeleteBatchThreadMembershipsForRetentionPolicies(t *testing.T, ss store.Store, s SqlStore) {
const limit = 1000
userID := model.NewId()
createThreadMembership := func(userID, postID string) *model.ThreadMembership {
@@ -695,7 +594,7 @@ func testThreadStorePermanentDeleteBatchThreadMembershipsForRetentionPolicies(t
// Delete team policy and thread
err = ss.RetentionPolicy().Delete(teamPolicy.ID)
require.NoError(t, err)
err = ss.Thread().Delete(post.Id)
_, err = s.GetMasterX().Exec("DELETE FROM Threads WHERE PostId='" + post.Id + "'")
require.NoError(t, err)
deleted, err := ss.Thread().DeleteOrphanedRows(1000)

6
store/storetest/user_store.go
Просмотреть файл

@@ -2315,7 +2315,7 @@ func testUserUnreadCount(t *testing.T, ss store.Store) {
// Post one message with mention to open channel
_, nErr = ss.Post().Save(&p1)
require.NoError(t, nErr)
nErr = ss.Channel().IncrementMentionCount(c1.Id, u2.Id, false, false)
nErr = ss.Channel().IncrementMentionCount(c1.Id, u2.Id, false)
require.NoError(t, nErr)
// Post 2 messages without mention to direct channel
@@ -2326,7 +2326,7 @@ func testUserUnreadCount(t *testing.T, ss store.Store) {
_, nErr = ss.Post().Save(&p2)
require.NoError(t, nErr)
nErr = ss.Channel().IncrementMentionCount(c2.Id, u2.Id, false, false)
nErr = ss.Channel().IncrementMentionCount(c2.Id, u2.Id, false)
require.NoError(t, nErr)
p3 := model.Post{}
@@ -2336,7 +2336,7 @@ func testUserUnreadCount(t *testing.T, ss store.Store) {
_, nErr = ss.Post().Save(&p3)
require.NoError(t, nErr)
nErr = ss.Channel().IncrementMentionCount(c2.Id, u2.Id, false, false)
nErr = ss.Channel().IncrementMentionCount(c2.Id, u2.Id, false)
require.NoError(t, nErr)
badge, unreadCountErr := ss.User().GetUnreadCount(u2.Id)

96
store/timerlayer/timerlayer.go
Просмотреть файл

@@ -1605,10 +1605,10 @@ func (s *TimerLayerChannelStore) GroupSyncedChannelCount() (int64, error) {
return result, err
}
func (s *TimerLayerChannelStore) IncrementMentionCount(channelID string, userID string, updateThreads bool, isRoot bool) error {
func (s *TimerLayerChannelStore) IncrementMentionCount(channelID string, userID string, isRoot bool) error {
start := timemodule.Now()
err := s.ChannelStore.IncrementMentionCount(channelID, userID, updateThreads, isRoot)
err := s.ChannelStore.IncrementMentionCount(channelID, userID, isRoot)
elapsed := float64(timemodule.Since(start)) / float64(timemodule.Second)
if s.Root.Metrics != nil {
@@ -2126,10 +2126,10 @@ func (s *TimerLayerChannelStore) UpdateLastViewedAt(channelIds []string, userID
return result, err
}
func (s *TimerLayerChannelStore) UpdateLastViewedAtPost(unreadPost *model.Post, userID string, mentionCount int, mentionCountRoot int, updateThreads bool, setUnreadCountRoot bool) (*model.ChannelUnreadAt, error) {
func (s *TimerLayerChannelStore) UpdateLastViewedAtPost(unreadPost *model.Post, userID string, mentionCount int, mentionCountRoot int, setUnreadCountRoot bool) (*model.ChannelUnreadAt, error) {
start := timemodule.Now()
result, err := s.ChannelStore.UpdateLastViewedAtPost(unreadPost, userID, mentionCount, mentionCountRoot, updateThreads, setUnreadCountRoot)
result, err := s.ChannelStore.UpdateLastViewedAtPost(unreadPost, userID, mentionCount, mentionCountRoot, setUnreadCountRoot)
elapsed := float64(timemodule.Since(start)) / float64(timemodule.Second)
if s.Root.Metrics != nil {
@@ -8311,22 +8311,6 @@ func (s *TimerLayerThreadStore) CollectThreadsWithNewerReplies(userId string, ch
return result, err
}
func (s *TimerLayerThreadStore) Delete(postID string) error {
start := timemodule.Now()
err := s.ThreadStore.Delete(postID)
elapsed := float64(timemodule.Since(start)) / float64(timemodule.Second)
if s.Root.Metrics != nil {
success := "false"
if err == nil {
success = "true"
}
s.Root.Metrics.ObserveStoreMethodDuration("ThreadStore.Delete", success, elapsed)
}
return err
}
func (s *TimerLayerThreadStore) DeleteMembershipForUser(userId string, postID string) error {
start := timemodule.Now()
@@ -8599,10 +8583,10 @@ func (s *TimerLayerThreadStore) PermanentDeleteBatchThreadMembershipsForRetentio
return result, resultVar1, err
}
func (s *TimerLayerThreadStore) Save(thread *model.Thread) (*model.Thread, error) {
func (s *TimerLayerThreadStore) UpdateLastViewedByThreadIds(userId string, threadIds []string, timestamp int64) error {
start := timemodule.Now()
result, err := s.ThreadStore.Save(thread)
err := s.ThreadStore.UpdateLastViewedByThreadIds(userId, threadIds, timestamp)
elapsed := float64(timemodule.Since(start)) / float64(timemodule.Second)
if s.Root.Metrics != nil {
@@ -8610,57 +8594,9 @@ func (s *TimerLayerThreadStore) Save(thread *model.Thread) (*model.Thread, error
if err == nil {
success = "true"
}
s.Root.Metrics.ObserveStoreMethodDuration("ThreadStore.Save", success, elapsed)
s.Root.Metrics.ObserveStoreMethodDuration("ThreadStore.UpdateLastViewedByThreadIds", success, elapsed)
}
return result, err
}
func (s *TimerLayerThreadStore) SaveMembership(membership *model.ThreadMembership) (*model.ThreadMembership, error) {
start := timemodule.Now()
result, err := s.ThreadStore.SaveMembership(membership)
elapsed := float64(timemodule.Since(start)) / float64(timemodule.Second)
if s.Root.Metrics != nil {
success := "false"
if err == nil {
success = "true"
}
s.Root.Metrics.ObserveStoreMethodDuration("ThreadStore.SaveMembership", success, elapsed)
}
return result, err
}
func (s *TimerLayerThreadStore) SaveMultiple(thread []*model.Thread) ([]*model.Thread, int, error) {
start := timemodule.Now()
result, resultVar1, err := s.ThreadStore.SaveMultiple(thread)
elapsed := float64(timemodule.Since(start)) / float64(timemodule.Second)
if s.Root.Metrics != nil {
success := "false"
if err == nil {
success = "true"
}
s.Root.Metrics.ObserveStoreMethodDuration("ThreadStore.SaveMultiple", success, elapsed)
}
return result, resultVar1, err
}
func (s *TimerLayerThreadStore) Update(thread *model.Thread) (*model.Thread, error) {
start := timemodule.Now()
result, err := s.ThreadStore.Update(thread)
elapsed := float64(timemodule.Since(start)) / float64(timemodule.Second)
if s.Root.Metrics != nil {
success := "false"
if err == nil {
success = "true"
}
s.Root.Metrics.ObserveStoreMethodDuration("ThreadStore.Update", success, elapsed)
}
return result, err
return err
}
func (s *TimerLayerThreadStore) UpdateMembership(membership *model.ThreadMembership) (*model.ThreadMembership, error) {
@@ -8679,22 +8615,6 @@ func (s *TimerLayerThreadStore) UpdateMembership(membership *model.ThreadMembers
return result, err
}
func (s *TimerLayerThreadStore) UpdateUnreadsByChannel(userId string, changedThreads []string, timestamp int64, updateViewedTimestamp bool) error {
start := timemodule.Now()
err := s.ThreadStore.UpdateUnreadsByChannel(userId, changedThreads, timestamp, updateViewedTimestamp)
elapsed := float64(timemodule.Since(start)) / float64(timemodule.Second)
if s.Root.Metrics != nil {
success := "false"
if err == nil {
success = "true"
}
s.Root.Metrics.ObserveStoreMethodDuration("ThreadStore.UpdateUnreadsByChannel", success, elapsed)
}
return err
}
func (s *TimerLayerTokenStore) Cleanup(expiryTime int64) {
start := timemodule.Now()