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

MM-28247 Threads metadata table (#15571)

Просмотр исходного кода
Этот коммит содержится в:
Eli Yukelzon
2020-10-01 18:48:38 +03:00
коммит произвёл GitHub
родитель 10961f9500
Коммит 595248b10e
17 изменённых файлов: 1007 добавлений и 9 удалений
Скачать Patch файл Скачать Diff файл Показать все файлы Свернуть все файлы

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

@@ -11,15 +11,15 @@ import (
"strings"
"sync"
"github.com/mattermost/mattermost-server/v5/store/searchlayer"
sq "github.com/Masterminds/squirrel"
"github.com/mattermost/gorp"
"github.com/pkg/errors"
"github.com/mattermost/mattermost-server/v5/einterfaces"
"github.com/mattermost/mattermost-server/v5/mlog"
"github.com/mattermost/mattermost-server/v5/model"
"github.com/mattermost/mattermost-server/v5/store"
"github.com/mattermost/mattermost-server/v5/store/searchlayer"
"github.com/mattermost/mattermost-server/v5/utils"
)
@@ -159,18 +159,34 @@ func (s *SqlPostStore) SaveMultiple(posts []*model.Post) ([]*model.Post, int, er
return nil, -1, errors.Wrap(err, "post_tosql")
}
if _, err := s.GetMaster().Exec(query, args...); err != nil {
transaction, err := s.GetMaster().Begin()
if err != nil {
return posts, -1, errors.Wrap(err, "begin_transaction")
}
defer finalizeTransaction(transaction)
if _, err = transaction.Exec(query, args...); err != nil {
return nil, -1, errors.Wrap(err, "failed to save Post")
}
if err = s.updateThreadsFromPosts(transaction, posts); err != nil {
mlog.Error("Error updating posts, thread update failed", mlog.Err(err))
}
if err = transaction.Commit(); err != nil {
// don't need to rollback here since the transaction is already closed
return posts, -1, errors.Wrap(err, "commit_transaction")
}
for channelId, count := range channelNewPosts {
if _, err := s.GetMaster().Exec("UPDATE Channels SET LastPostAt = GREATEST(:LastPostAt, LastPostAt), TotalMsgCount = TotalMsgCount + :Count WHERE Id = :ChannelId", map[string]interface{}{"LastPostAt": maxDateNewPosts[channelId], "ChannelId": channelId, "Count": count}); err != nil {
if _, err = s.GetMaster().Exec("UPDATE Channels SET LastPostAt = GREATEST(:LastPostAt, LastPostAt), TotalMsgCount = TotalMsgCount + :Count WHERE Id = :ChannelId", map[string]interface{}{"LastPostAt": maxDateNewPosts[channelId], "ChannelId": channelId, "Count": count}); err != nil {
mlog.Error("Error updating Channel LastPostAt.", mlog.Err(err))
}
}
for rootId := range rootIds {
if _, err := s.GetMaster().Exec("UPDATE Posts SET UpdateAt = :UpdateAt WHERE Id = :RootId", map[string]interface{}{"UpdateAt": maxDateRootIds[rootId], "RootId": rootId}); err != nil {
if _, err = s.GetMaster().Exec("UPDATE Posts SET UpdateAt = :UpdateAt WHERE Id = :RootId", map[string]interface{}{"UpdateAt": maxDateRootIds[rootId], "RootId": rootId}); err != nil {
mlog.Error("Error updating Post UpdateAt.", mlog.Err(err))
}
}
@@ -265,6 +281,7 @@ func (s *SqlPostStore) Update(newPost *model.Post, oldPost *model.Post) (*model.
if len(newPost.RootId) > 0 {
s.GetMaster().Exec("UPDATE Posts SET UpdateAt = :UpdateAt WHERE Id = :RootId AND UpdateAt < :UpdateAt", map[string]interface{}{"UpdateAt": time, "RootId": newPost.RootId})
s.GetMaster().Exec("UPDATE Threads SET LastReplyAt = :UpdateAt WHERE PostId = :RootId", map[string]interface{}{"UpdateAt": time, "RootId": newPost.RootId})
}
// mark the old post as deleted
@@ -296,6 +313,9 @@ func (s *SqlPostStore) OverwriteMultiple(posts []*model.Post) ([]*model.Post, in
return nil, idx, errors.Wrap(err, "failed to update Post")
}
if len(post.RootId) > 0 {
tx.Exec("UPDATE Threads SET LastReplyAt = :UpdateAt WHERE PostId = :RootId", map[string]interface{}{"UpdateAt": updateAt, "RootId": post.Id})
}
}
err = tx.Commit()
if err != nil {
@@ -495,19 +515,49 @@ func (s *SqlPostStore) Delete(postId string, time int64, deleteByID string) erro
return errors.Wrap(err, "failed to update Posts")
}
return nil
return s.cleanupThreads(post.Id, post.RootId, post.UserId)
}
func (s *SqlPostStore) permanentDelete(postId string) error {
_, err := s.GetMaster().Exec("DELETE FROM Posts WHERE Id = :Id OR RootId = :RootId", map[string]interface{}{"Id": postId, "RootId": postId})
if err != nil {
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)
}
}
if _, err = s.GetMaster().Exec("DELETE FROM Posts WHERE Id = :Id OR RootId = :RootId", map[string]interface{}{"Id": postId, "RootId": postId}); err != nil {
return errors.Wrapf(err, "failed to delete Post with id=%s", postId)
}
return nil
}
type postIds struct {
Id string
RootId string
UserId string
}
func (s *SqlPostStore) permanentDeleteAllCommentByUser(userId string) error {
_, err := s.GetMaster().Exec("DELETE FROM Posts WHERE UserId = :UserId AND RootId != ''", map[string]interface{}{"UserId": userId})
results := []postIds{}
_, err := s.GetMaster().Select(&results, "Select Id, RootId FROM Posts WHERE UserId = :UserId AND RootId != ''", map[string]interface{}{"UserId": userId})
if err != nil {
return errors.Wrapf(err, "failed to fetch Posts with userId=%s", userId)
}
for _, ids := range results {
if err = s.cleanupThreads(ids.Id, ids.RootId, userId); err != nil {
return err
}
}
_, err = s.GetMaster().Exec("DELETE FROM Posts WHERE UserId = :UserId AND RootId != ''", map[string]interface{}{"UserId": userId})
if err != nil {
return errors.Wrapf(err, "failed to delete Posts with userId=%s", userId)
}
@@ -551,6 +601,18 @@ func (s *SqlPostStore) PermanentDeleteByUser(userId string) error {
}
func (s *SqlPostStore) PermanentDeleteByChannel(channelId string) error {
results := []postIds{}
_, err := s.GetMaster().Select(&results, "SELECT Id, RootId, UserId FROM Posts WHERE ChannelId = :ChannelId", map[string]interface{}{"ChannelId": channelId})
if err != nil {
return errors.Wrapf(err, "failed to fetch Posts with channelId=%s", channelId)
}
for _, ids := range results {
if err = s.cleanupThreads(ids.Id, ids.RootId, ids.UserId); err != nil {
return err
}
}
if _, err := s.GetMaster().Exec("DELETE FROM Posts WHERE ChannelId = :ChannelId", map[string]interface{}{"ChannelId": channelId}); err != nil {
return errors.Wrapf(err, "failed to delete Posts with channelId=%s", channelId)
}
@@ -1858,3 +1920,90 @@ func (s *SqlPostStore) GetOldestEntityCreationTime() (int64, error) {
}
return oldest, nil
}
func (s *SqlPostStore) cleanupThreads(postId, rootId, userId string) error {
if len(rootId) > 0 {
thread, err := s.Thread().Get(rootId)
if err != nil {
if err != sql.ErrNoRows {
return errors.Wrap(err, "failed to get a thread")
}
}
if thread != nil {
thread.ReplyCount -= 1
thread.Participants = thread.Participants.Remove(userId)
if _, err = s.Thread().Update(thread); err != nil {
return errors.Wrap(err, "failed to update thread")
}
}
}
_, err := s.GetMaster().Exec("DELETE FROM Threads WHERE PostId = :Id", map[string]interface{}{"Id": postId})
if err != nil {
return errors.Wrap(err, "failed to update Threads")
}
return nil
}
func (s *SqlPostStore) updateThreadsFromPosts(transaction *gorp.Transaction, posts []*model.Post) error {
postsByRoot := map[string][]*model.Post{}
var rootIds []string
for _, post := range posts {
// skip if post is not a part of a thread
if len(post.RootId) == 0 {
continue
}
rootIds = append(rootIds, post.RootId)
postsByRoot[post.RootId] = append(postsByRoot[post.RootId], post)
}
if len(rootIds) == 0 {
return nil
}
now := model.GetMillis()
threadsByRootsSql, threadsByRootsArgs, _ := s.getQueryBuilder().Select("*").From("Threads").Where(sq.Eq{"PostId": rootIds}).ToSql()
var threadsByRoots []*model.Thread
if _, err := transaction.Select(&threadsByRoots, threadsByRootsSql, threadsByRootsArgs...); err != nil {
return err
}
threadByRoot := map[string]*model.Thread{}
for _, thread := range threadsByRoots {
threadByRoot[thread.PostId] = thread
}
for rootId, posts := range postsByRoot {
if thread, found := threadByRoot[rootId]; !found {
// calculate participants
var participants model.StringArray
if _, err := transaction.Select(&participants, "SELECT DISTINCT UserId FROM Posts WHERE RootId=:RootId", map[string]interface{}{"RootId": rootId}); err != nil {
return err
}
// calculate reply count
count, err := transaction.SelectInt("SELECT COUNT(Id) FROM Posts WHERE RootId=:RootId", map[string]interface{}{"RootId": rootId})
if err != nil {
return err
}
// no metadata entry, create one
if err := transaction.Insert(&model.Thread{
PostId: rootId,
ReplyCount: count,
LastReplyAt: now,
Participants: participants,
}); err != nil {
return err
}
} else {
// metadata exists, update it
thread.LastReplyAt = now
for _, post := range posts {
thread.ReplyCount += 1
if !thread.Participants.Contains(post.UserId) {
thread.Participants = append(thread.Participants, post.UserId)
}
}
if _, err := transaction.Update(thread); err != nil {
return err
}
}
}
return nil
}

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

@@ -75,6 +75,7 @@ type SqlStore interface {
Team() store.TeamStore
Channel() store.ChannelStore
Post() store.PostStore
Thread() store.ThreadStore
User() store.UserStore
Bot() store.BotStore
Audit() store.AuditStore

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

@@ -72,6 +72,7 @@ type SqlSupplierStores struct {
team store.TeamStore
channel store.ChannelStore
post store.PostStore
thread store.ThreadStore
user store.UserStore
bot store.BotStore
audit store.AuditStore
@@ -160,6 +161,7 @@ func NewSqlSupplier(settings model.SqlSettings, metrics einterfaces.MetricsInter
supplier.stores.status = newSqlStatusStore(supplier)
supplier.stores.fileInfo = newSqlFileInfoStore(supplier, metrics)
supplier.stores.uploadSession = newSqlUploadSessionStore(supplier)
supplier.stores.thread = newSqlThreadStore(supplier)
supplier.stores.job = newSqlJobStore(supplier)
supplier.stores.userAccessToken = newSqlUserAccessTokenStore(supplier)
supplier.stores.channelMemberHistory = newSqlChannelMemberHistoryStore(supplier)
@@ -189,6 +191,7 @@ func NewSqlSupplier(settings model.SqlSettings, metrics einterfaces.MetricsInter
supplier.stores.team.(*SqlTeamStore).createIndexesIfNotExists()
supplier.stores.channel.(*SqlChannelStore).createIndexesIfNotExists()
supplier.stores.post.(*SqlPostStore).createIndexesIfNotExists()
supplier.stores.thread.(*SqlThreadStore).createIndexesIfNotExists()
supplier.stores.user.(*SqlUserStore).createIndexesIfNotExists()
supplier.stores.bot.(*SqlBotStore).createIndexesIfNotExists()
supplier.stores.audit.(*SqlAuditStore).createIndexesIfNotExists()
@@ -1163,6 +1166,10 @@ func (ss *SqlSupplier) Plugin() store.PluginStore {
return ss.stores.plugin
}
func (ss *SqlSupplier) Thread() store.ThreadStore {
return ss.stores.thread
}
func (ss *SqlSupplier) Role() store.RoleStore {
return ss.stores.role
}

108
store/sqlstore/thread_store.go Обычный файл
Просмотреть файл

@@ -0,0 +1,108 @@
// Copyright (c) 2015-present Mattermost, Inc. All Rights Reserved.
// See LICENSE.txt for license information.
package sqlstore
import (
"database/sql"
"github.com/mattermost/mattermost-server/v5/model"
"github.com/mattermost/mattermost-server/v5/store"
"github.com/pkg/errors"
sq "github.com/Masterminds/squirrel"
)
type SqlThreadStore struct {
SqlStore
}
func (s *SqlThreadStore) ClearCaches() {
}
func newSqlThreadStore(sqlStore SqlStore) store.ThreadStore {
s := &SqlThreadStore{
SqlStore: sqlStore,
}
for _, db := range sqlStore.GetAllConns() {
table := db.AddTableWithName(model.Thread{}, "Threads").SetKeys(false, "PostId")
table.ColMap("PostId").SetMaxSize(26)
table.ColMap("Participants").SetMaxSize(0)
}
return s
}
func threadSliceColumns() []string {
return []string{"PostId", "LastReplyAt", "ReplyCount", "Participants"}
}
func threadToSlice(thread *model.Thread) []interface{} {
return []interface{}{
thread.PostId,
thread.LastReplyAt,
thread.ReplyCount,
thread.Participants,
}
}
func (s *SqlThreadStore) createIndexesIfNotExists() {
s.CreateIndexIfNotExists("idx_threads_last_reply_at", "Threads", "LastReplyAt")
s.CreateIndexIfNotExists("idx_threads_post_id", "Threads", "PostId")
}
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.GetMaster().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) {
if _, err := s.GetMaster().Update(thread); 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, _ := s.getQueryBuilder().Select("*").From("Threads").Where(sq.Eq{"PostId": id}).ToSql()
err := s.GetReplica().SelectOne(&thread, query, args...)
if err != nil {
if err == sql.ErrNoRows {
return nil, store.NewErrNotFound("Thread", id)
}
return nil, errors.Wrapf(err, "failed to get thread with id=%s", id)
}
return &thread, 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 {
return errors.Wrap(err, "failed to update threads")
}
return nil
}

14
store/sqlstore/thread_store_test.go Обычный файл
Просмотреть файл

@@ -0,0 +1,14 @@
// Copyright (c) 2015-present Mattermost, Inc. All Rights Reserved.
// See LICENSE.txt for license information.
package sqlstore
import (
"testing"
"github.com/mattermost/mattermost-server/v5/store/storetest"
)
func TestThreadStore(t *testing.T) {
StoreTestWithSqlSupplier(t, storetest.TestThreadStore)
}