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

MM-29987 Implement new collapsed threads API (#16091)

Просмотр исходного кода
Этот коммит содержится в:
Eli Yukelzon
2020-11-08 10:36:46 +02:00
коммит произвёл GitHub
родитель 483441cea2
Коммит 45e340b5be
22 изменённых файлов: 1155 добавлений и 49 удалений
Скачать Patch файл Скачать Diff файл Показать все файлы Свернуть все файлы

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

@@ -7648,7 +7648,7 @@ func (s *OpenTracingLayerThreadStore) CollectThreadsWithNewerReplies(userId stri
return result, err
}
func (s *OpenTracingLayerThreadStore) CreateMembershipIfNeeded(userId string, postId string) error {
func (s *OpenTracingLayerThreadStore) CreateMembershipIfNeeded(userId string, postId string, following bool) error {
origCtx := s.Root.Store.Context()
span, newCtx := tracing.StartSpanWithParentByContext(s.Root.Store.Context(), "ThreadStore.CreateMembershipIfNeeded")
s.Root.Store.SetContext(newCtx)
@@ -7657,7 +7657,7 @@ func (s *OpenTracingLayerThreadStore) CreateMembershipIfNeeded(userId string, po
}()
defer span.Finish()
err := s.ThreadStore.CreateMembershipIfNeeded(userId, postId)
err := s.ThreadStore.CreateMembershipIfNeeded(userId, postId, following)
if err != nil {
span.LogFields(spanlog.Error(err))
ext.Error.Set(span, true)
@@ -7756,6 +7756,60 @@ func (s *OpenTracingLayerThreadStore) GetMembershipsForUser(userId string) ([]*m
return result, err
}
func (s *OpenTracingLayerThreadStore) GetThreadsForUser(userId string, opts model.GetUserThreadsOpts) (*model.Threads, error) {
origCtx := s.Root.Store.Context()
span, newCtx := tracing.StartSpanWithParentByContext(s.Root.Store.Context(), "ThreadStore.GetThreadsForUser")
s.Root.Store.SetContext(newCtx)
defer func() {
s.Root.Store.SetContext(origCtx)
}()
defer span.Finish()
result, err := s.ThreadStore.GetThreadsForUser(userId, opts)
if err != nil {
span.LogFields(spanlog.Error(err))
ext.Error.Set(span, true)
}
return result, err
}
func (s *OpenTracingLayerThreadStore) MarkAllAsRead(userId string, timestamp int64) error {
origCtx := s.Root.Store.Context()
span, newCtx := tracing.StartSpanWithParentByContext(s.Root.Store.Context(), "ThreadStore.MarkAllAsRead")
s.Root.Store.SetContext(newCtx)
defer func() {
s.Root.Store.SetContext(origCtx)
}()
defer span.Finish()
err := s.ThreadStore.MarkAllAsRead(userId, timestamp)
if err != nil {
span.LogFields(spanlog.Error(err))
ext.Error.Set(span, true)
}
return err
}
func (s *OpenTracingLayerThreadStore) MarkAsRead(userId string, threadId string, timestamp int64) error {
origCtx := s.Root.Store.Context()
span, newCtx := tracing.StartSpanWithParentByContext(s.Root.Store.Context(), "ThreadStore.MarkAsRead")
s.Root.Store.SetContext(newCtx)
defer func() {
s.Root.Store.SetContext(origCtx)
}()
defer span.Finish()
err := s.ThreadStore.MarkAsRead(userId, threadId, timestamp)
if err != nil {
span.LogFields(spanlog.Error(err))
ext.Error.Set(span, true)
}
return err
}
func (s *OpenTracingLayerThreadStore) Save(thread *model.Thread) (*model.Thread, error) {
origCtx := s.Root.Store.Context()
span, newCtx := tracing.StartSpanWithParentByContext(s.Root.Store.Context(), "ThreadStore.Save")

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

@@ -7682,11 +7682,11 @@ func (s *RetryLayerThreadStore) CollectThreadsWithNewerReplies(userId string, ch
}
func (s *RetryLayerThreadStore) CreateMembershipIfNeeded(userId string, postId string) error {
func (s *RetryLayerThreadStore) CreateMembershipIfNeeded(userId string, postId string, following bool) error {
tries := 0
for {
err := s.ThreadStore.CreateMembershipIfNeeded(userId, postId)
err := s.ThreadStore.CreateMembershipIfNeeded(userId, postId, following)
if err == nil {
return nil
}
@@ -7802,6 +7802,66 @@ func (s *RetryLayerThreadStore) GetMembershipsForUser(userId string) ([]*model.T
}
func (s *RetryLayerThreadStore) GetThreadsForUser(userId string, opts model.GetUserThreadsOpts) (*model.Threads, error) {
tries := 0
for {
result, err := s.ThreadStore.GetThreadsForUser(userId, opts)
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
}
}
}
func (s *RetryLayerThreadStore) MarkAllAsRead(userId string, timestamp int64) error {
tries := 0
for {
err := s.ThreadStore.MarkAllAsRead(userId, timestamp)
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
}
}
}
func (s *RetryLayerThreadStore) MarkAsRead(userId string, threadId string, timestamp int64) error {
tries := 0
for {
err := s.ThreadStore.MarkAsRead(userId, threadId, timestamp)
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
}
}
}
func (s *RetryLayerThreadStore) Save(thread *model.Thread) (*model.Thread, error) {
tries := 0

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

@@ -515,20 +515,17 @@ func (s *SqlPostStore) Delete(postId string, time int64, deleteByID string) erro
return errors.Wrap(err, "failed to update Posts")
}
return s.cleanupThreads(post.Id, post.RootId, post.UserId)
return s.cleanupThreads(post.Id, post.RootId, post.UserId, false)
}
func (s *SqlPostStore) permanentDelete(postId string) error {
var post model.Post
err := s.GetReplica().SelectOne(&post, "SELECT * FROM Posts WHERE Id = :Id AND DeleteAt = 0", map[string]interface{}{"Id": postId})
if err != nil && err != sql.ErrNoRows {
if err != sql.ErrNoRows {
return errors.Wrapf(err, "failed to get Post with id=%s", postId)
}
if err = s.cleanupThreads(post.Id, post.RootId, post.UserId); err != nil {
return errors.Wrapf(err, "failed to cleanup threads for Post with id=%s", postId)
}
return errors.Wrapf(err, "failed to get Post with id=%s", postId)
}
if err = s.cleanupThreads(post.Id, post.RootId, post.UserId, true); err != nil {
return errors.Wrapf(err, "failed to cleanup threads for Post with id=%s", postId)
}
if _, err = s.GetMaster().Exec("DELETE FROM Posts WHERE Id = :Id OR RootId = :RootId", map[string]interface{}{"Id": postId, "RootId": postId}); err != nil {
@@ -552,7 +549,7 @@ func (s *SqlPostStore) permanentDeleteAllCommentByUser(userId string) error {
}
for _, ids := range results {
if err = s.cleanupThreads(ids.Id, ids.RootId, userId); err != nil {
if err = s.cleanupThreads(ids.Id, ids.RootId, userId, true); err != nil {
return err
}
}
@@ -608,7 +605,7 @@ func (s *SqlPostStore) PermanentDeleteByChannel(channelId string) error {
}
for _, ids := range results {
if err = s.cleanupThreads(ids.Id, ids.RootId, ids.UserId); err != nil {
if err = s.cleanupThreads(ids.Id, ids.RootId, ids.UserId, true); err != nil {
return err
}
}
@@ -1921,7 +1918,16 @@ func (s *SqlPostStore) GetOldestEntityCreationTime() (int64, error) {
return oldest, nil
}
func (s *SqlPostStore) cleanupThreads(postId, rootId, userId string) error {
func (s *SqlPostStore) cleanupThreads(postId, rootId, userId string, permanent bool) error {
if permanent {
if _, err := s.GetMaster().Exec("DELETE FROM Threads WHERE PostId = :Id", map[string]interface{}{"Id": postId}); err != nil {
return errors.Wrap(err, "failed to delete Threads")
}
if _, err := s.GetMaster().Exec("DELETE FROM ThreadMemberships WHERE PostId = :Id", map[string]interface{}{"Id": postId}); err != nil {
return errors.Wrap(err, "failed to delete ThreadMemberships")
}
return nil
}
if len(rootId) > 0 {
thread, err := s.Thread().Get(rootId)
if err != nil {
@@ -1937,12 +1943,6 @@ func (s *SqlPostStore) cleanupThreads(postId, rootId, userId string) error {
}
}
}
if _, err := s.GetMaster().Exec("DELETE FROM Threads WHERE PostId = :Id", map[string]interface{}{"Id": postId}); err != nil {
return errors.Wrap(err, "failed to delete Threads")
}
if _, err := s.GetMaster().Exec("DELETE FROM ThreadMemberships WHERE PostId = :Id", map[string]interface{}{"Id": postId}); err != nil {
return errors.Wrap(err, "failed to delete ThreadMemberships")
}
return nil
}

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

@@ -108,6 +108,117 @@ func (s *SqlThreadStore) Get(id string) (*model.Thread, error) {
return &thread, nil
}
func (s *SqlThreadStore) GetThreadsForUser(userId string, opts model.GetUserThreadsOpts) (*model.Threads, error) {
type JoinedThread struct {
PostId string
ReplyCount int64
LastReplyAt int64
LastViewedAt int64
Participants model.StringArray
model.Post
}
var threads []*JoinedThread
fetchConditions := sq.And{
sq.Eq{"Posts.UserId": userId},
sq.Eq{"ThreadMemberships.Following": true},
}
if !opts.Deleted {
fetchConditions = sq.And{fetchConditions, sq.Eq{"Posts.DeleteAt": 0}}
}
if opts.Since > 0 {
fetchConditions = sq.And{fetchConditions, sq.GtOrEq{"Threads.LastReplyAt": opts.Since}}
}
pageSize := uint64(30)
if opts.PageSize == 0 {
pageSize = opts.PageSize
}
query, args, _ := s.getQueryBuilder().
Select("Threads.*, Posts.*, ThreadMemberships.LastViewed as LastViewedAt").
From("Threads").
LeftJoin("Posts ON Posts.Id = Threads.PostId").
LeftJoin("ThreadMemberships ON ThreadMemberships.PostId = Threads.PostId").
OrderBy("Threads.LastReplyAt DESC").
Offset(pageSize * opts.Page).
Limit(pageSize).
Where(fetchConditions).ToSql()
_, err := s.GetReplica().Select(&threads, query, args...)
if err != nil {
return nil, errors.Wrapf(err, "failed to get threads for user id=%s", userId)
}
var userIds []string
userIdMap := map[string]bool{}
for _, thread := range threads {
for _, participantId := range thread.Participants {
if _, ok := userIdMap[participantId]; !ok {
userIdMap[participantId] = true
userIds = append(userIds, participantId)
}
}
}
var users []*model.User
if opts.Extended {
query, args, _ = s.getQueryBuilder().Select("*").From("Users").Where(sq.Eq{"Id": userIds}).ToSql()
_, err = s.GetReplica().Select(&users, query, args...)
if err != nil {
return nil, errors.Wrapf(err, "failed to get threads for user id=%s", userId)
}
} else {
for _, userId := range userIds {
users = append(users, &model.User{Id: userId})
}
}
result := &model.Threads{
Total: 0,
Threads: nil,
}
for _, thread := range threads {
var participants []*model.User
for _, participantId := range thread.Participants {
var participant *model.User
for _, u := range users {
if u.Id == participantId {
participant = u
break
}
}
if participant == nil {
return nil, errors.New("cannot find thread participant with id=" + participantId)
}
participants = append(participants, participant)
}
result.Threads = append(result.Threads, &model.ThreadResponse{
PostId: thread.PostId,
ReplyCount: thread.ReplyCount,
LastReplyAt: thread.LastReplyAt,
LastViewedAt: thread.LastViewedAt,
Participants: participants,
Post: &thread.Post,
})
}
return result, nil
}
func (s *SqlThreadStore) MarkAllAsRead(userId string, timestamp int64) error {
query, args, _ := s.getQueryBuilder().Update("ThreadMemberships").Where(sq.Eq{"UserId": userId}).Set("LastViewed", timestamp).ToSql()
if _, err := s.GetMaster().Exec(query, args...); err != nil {
return errors.Wrapf(err, "failed to update thread read state for user id=%s", userId)
}
return nil
}
func (s *SqlThreadStore) MarkAsRead(userId, threadId string, timestamp int64) error {
query, args, _ := s.getQueryBuilder().Update("ThreadMemberships").Where(sq.Eq{"UserId": userId}, sq.Eq{"PostId": threadId}).Set("LastViewed", timestamp).ToSql()
if _, err := s.GetMaster().Exec(query, args...); err != nil {
return errors.Wrapf(err, "failed to update thread read state for user id=%s thread_id=%v", userId, threadId)
}
return nil
}
func (s *SqlThreadStore) Delete(threadId string) error {
query, args, _ := s.getQueryBuilder().Delete("Threads").Where(sq.Eq{"PostId": threadId}).ToSql()
if _, err := s.GetMaster().Exec(query, args...); err != nil {
@@ -162,12 +273,12 @@ func (s *SqlThreadStore) DeleteMembershipForUser(userId string, postId string) e
return nil
}
func (s *SqlThreadStore) CreateMembershipIfNeeded(userId, postId string) error {
func (s *SqlThreadStore) CreateMembershipIfNeeded(userId, postId string, following bool) error {
membership, err := s.GetMembershipForUser(userId, postId)
now := utils.MillisFromTime(time.Now())
if err == nil {
if !membership.Following {
membership.Following = true
if !membership.Following || membership.Following != following {
membership.Following = following
membership.LastUpdated = now
_, err = s.UpdateMembership(membership)
}
@@ -182,7 +293,7 @@ func (s *SqlThreadStore) CreateMembershipIfNeeded(userId, postId string) error {
_, err = s.SaveMembership(&model.ThreadMembership{
PostId: postId,
UserId: userId,
Following: true,
Following: following,
LastViewed: 0,
LastUpdated: now,
})

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

@@ -251,14 +251,18 @@ type ThreadStore interface {
Save(thread *model.Thread) (*model.Thread, error)
Update(thread *model.Thread) (*model.Thread, error)
Get(id string) (*model.Thread, error)
GetThreadsForUser(userId string, opts model.GetUserThreadsOpts) (*model.Threads, error)
Delete(postId string) error
MarkAllAsRead(userId string, timestamp int64) error
MarkAsRead(userId, threadId string, timestamp int64) error
SaveMembership(membership *model.ThreadMembership) (*model.ThreadMembership, error)
UpdateMembership(membership *model.ThreadMembership) (*model.ThreadMembership, error)
GetMembershipsForUser(userId string) ([]*model.ThreadMembership, error)
GetMembershipForUser(userId, postId string) (*model.ThreadMembership, error)
DeleteMembershipForUser(userId, postId string) error
CreateMembershipIfNeeded(userId, postId string) error
CreateMembershipIfNeeded(userId, postId string, following bool) error
CollectThreadsWithNewerReplies(userId string, channelIds []string, timestamp int64) ([]string, error)
UpdateUnreadsByChannel(userId string, changedThreads []string, timestamp int64) error
}

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

@@ -37,13 +37,13 @@ func (_m *ThreadStore) CollectThreadsWithNewerReplies(userId string, channelIds
return r0, r1
}
// CreateMembershipIfNeeded provides a mock function with given fields: userId, postId
func (_m *ThreadStore) CreateMembershipIfNeeded(userId string, postId string) error {
ret := _m.Called(userId, postId)
// CreateMembershipIfNeeded provides a mock function with given fields: userId, postId, following
func (_m *ThreadStore) CreateMembershipIfNeeded(userId string, postId string, following bool) error {
ret := _m.Called(userId, postId, following)
var r0 error
if rf, ok := ret.Get(0).(func(string, string) error); ok {
r0 = rf(userId, postId)
if rf, ok := ret.Get(0).(func(string, string, bool) error); ok {
r0 = rf(userId, postId, following)
} else {
r0 = ret.Error(0)
}
@@ -148,6 +148,57 @@ func (_m *ThreadStore) GetMembershipsForUser(userId string) ([]*model.ThreadMemb
return r0, r1
}
// GetThreadsForUser provides a mock function with given fields: userId, opts
func (_m *ThreadStore) GetThreadsForUser(userId string, opts model.GetUserThreadsOpts) (*model.Threads, error) {
ret := _m.Called(userId, opts)
var r0 *model.Threads
if rf, ok := ret.Get(0).(func(string, model.GetUserThreadsOpts) *model.Threads); ok {
r0 = rf(userId, opts)
} else {
if ret.Get(0) != nil {
r0 = ret.Get(0).(*model.Threads)
}
}
var r1 error
if rf, ok := ret.Get(1).(func(string, model.GetUserThreadsOpts) error); ok {
r1 = rf(userId, opts)
} else {
r1 = ret.Error(1)
}
return r0, r1
}
// MarkAllAsRead provides a mock function with given fields: userId, timestamp
func (_m *ThreadStore) MarkAllAsRead(userId string, timestamp int64) error {
ret := _m.Called(userId, timestamp)
var r0 error
if rf, ok := ret.Get(0).(func(string, int64) error); ok {
r0 = rf(userId, timestamp)
} else {
r0 = ret.Error(0)
}
return r0
}
// MarkAsRead provides a mock function with given fields: userId, threadId, timestamp
func (_m *ThreadStore) MarkAsRead(userId string, threadId string, timestamp int64) error {
ret := _m.Called(userId, threadId, timestamp)
var r0 error
if rf, ok := ret.Get(0).(func(string, string, int64) error); ok {
r0 = rf(userId, threadId, timestamp)
} else {
r0 = ret.Error(0)
}
return r0
}
// Save provides a mock function with given fields: thread
func (_m *ThreadStore) Save(thread *model.Thread) (*model.Thread, error) {
ret := _m.Called(thread)

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

@@ -223,7 +223,7 @@ func testThreadStorePopulation(t *testing.T, ss store.Store) {
require.EqualValues(t, thread1.ReplyCount, 1)
require.Len(t, thread1.Participants, 1)
err = ss.Post().Delete(rootPost.Id, 123, model.NewId())
err = ss.Post().PermanentDeleteByUser(rootPost.UserId)
require.Nil(t, err)
thread2, _ := ss.Thread().Get(rootPost.Id)
@@ -233,7 +233,7 @@ func testThreadStorePopulation(t *testing.T, ss store.Store) {
t.Run("Thread last updated is changed when channel is updated after UpdateLastViewedAtPost", func(t *testing.T) {
newPosts := makeSomePosts()
require.Nil(t, ss.Thread().CreateMembershipIfNeeded(newPosts[0].UserId, newPosts[0].Id))
require.Nil(t, ss.Thread().CreateMembershipIfNeeded(newPosts[0].UserId, newPosts[0].Id, true))
m, err1 := ss.Thread().GetMembershipForUser(newPosts[0].UserId, newPosts[0].Id)
require.Nil(t, err1)
m.LastUpdated -= 1000
@@ -253,7 +253,7 @@ func testThreadStorePopulation(t *testing.T, ss store.Store) {
t.Run("Thread last updated is changed when channel is updated after IncrementMentionCount", func(t *testing.T) {
newPosts := makeSomePosts()
require.Nil(t, ss.Thread().CreateMembershipIfNeeded(newPosts[0].UserId, newPosts[0].Id))
require.Nil(t, ss.Thread().CreateMembershipIfNeeded(newPosts[0].UserId, newPosts[0].Id, true))
m, err1 := ss.Thread().GetMembershipForUser(newPosts[0].UserId, newPosts[0].Id)
require.Nil(t, err1)
m.LastUpdated -= 1000
@@ -273,7 +273,7 @@ func testThreadStorePopulation(t *testing.T, ss store.Store) {
t.Run("Thread last updated is changed when channel is updated after UpdateLastViewedAt", func(t *testing.T) {
newPosts := makeSomePosts()
require.Nil(t, ss.Thread().CreateMembershipIfNeeded(newPosts[0].UserId, newPosts[0].Id))
require.Nil(t, ss.Thread().CreateMembershipIfNeeded(newPosts[0].UserId, newPosts[0].Id, true))
m, err1 := ss.Thread().GetMembershipForUser(newPosts[0].UserId, newPosts[0].Id)
require.Nil(t, err1)
m.LastUpdated -= 1000
@@ -293,7 +293,7 @@ func testThreadStorePopulation(t *testing.T, ss store.Store) {
t.Run("Thread last updated is changed when channel is updated after UpdateLastViewedAtPost for mark unread", func(t *testing.T) {
newPosts := makeSomePosts()
require.Nil(t, ss.Thread().CreateMembershipIfNeeded(newPosts[0].UserId, newPosts[0].Id))
require.Nil(t, ss.Thread().CreateMembershipIfNeeded(newPosts[0].UserId, newPosts[0].Id, true))
m, err1 := ss.Thread().GetMembershipForUser(newPosts[0].UserId, newPosts[0].Id)
require.Nil(t, err1)
m.LastUpdated += 1000

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

@@ -6904,10 +6904,10 @@ func (s *TimerLayerThreadStore) CollectThreadsWithNewerReplies(userId string, ch
return result, err
}
func (s *TimerLayerThreadStore) CreateMembershipIfNeeded(userId string, postId string) error {
func (s *TimerLayerThreadStore) CreateMembershipIfNeeded(userId string, postId string, following bool) error {
start := timemodule.Now()
err := s.ThreadStore.CreateMembershipIfNeeded(userId, postId)
err := s.ThreadStore.CreateMembershipIfNeeded(userId, postId, following)
elapsed := float64(timemodule.Since(start)) / float64(timemodule.Second)
if s.Root.Metrics != nil {
@@ -7000,6 +7000,54 @@ func (s *TimerLayerThreadStore) GetMembershipsForUser(userId string) ([]*model.T
return result, err
}
func (s *TimerLayerThreadStore) GetThreadsForUser(userId string, opts model.GetUserThreadsOpts) (*model.Threads, error) {
start := timemodule.Now()
result, err := s.ThreadStore.GetThreadsForUser(userId, opts)
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.GetThreadsForUser", success, elapsed)
}
return result, err
}
func (s *TimerLayerThreadStore) MarkAllAsRead(userId string, timestamp int64) error {
start := timemodule.Now()
err := s.ThreadStore.MarkAllAsRead(userId, timestamp)
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.MarkAllAsRead", success, elapsed)
}
return err
}
func (s *TimerLayerThreadStore) MarkAsRead(userId string, threadId string, timestamp int64) error {
start := timemodule.Now()
err := s.ThreadStore.MarkAsRead(userId, threadId, timestamp)
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.MarkAsRead", success, elapsed)
}
return err
}
func (s *TimerLayerThreadStore) Save(thread *model.Thread) (*model.Thread, error) {
start := timemodule.Now()