[MM-36268] Fix replication lag error on post reply (#17752)
* Fix replication lag error on post reply * Improve further by using a db transaction
Этот коммит содержится в:
Claudio Costa
коммит произвёл
GitHub
GitHub
родитель
d093e102a4
Коммит
6483abd263
@@ -9,6 +9,7 @@ import (
|
||||
"time"
|
||||
|
||||
sq "github.com/Masterminds/squirrel"
|
||||
"github.com/mattermost/gorp"
|
||||
"github.com/pkg/errors"
|
||||
|
||||
"github.com/mattermost/mattermost-server/v5/model"
|
||||
@@ -88,7 +89,11 @@ func (s *SqlThreadStore) Save(thread *model.Thread) (*model.Thread, error) {
|
||||
}
|
||||
|
||||
func (s *SqlThreadStore) Update(thread *model.Thread) (*model.Thread, error) {
|
||||
if _, err := s.GetMaster().Update(thread); err != nil {
|
||||
return s.update(s.GetMaster(), thread)
|
||||
}
|
||||
|
||||
func (s *SqlThreadStore) update(ex gorp.SqlExecutor, thread *model.Thread) (*model.Thread, error) {
|
||||
if _, err := ex.Update(thread); err != nil {
|
||||
return nil, errors.Wrapf(err, "failed to update thread with id=%s", thread.PostId)
|
||||
}
|
||||
|
||||
@@ -96,9 +101,13 @@ func (s *SqlThreadStore) Update(thread *model.Thread) (*model.Thread, error) {
|
||||
}
|
||||
|
||||
func (s *SqlThreadStore) Get(id string) (*model.Thread, error) {
|
||||
return s.get(s.GetReplica(), id)
|
||||
}
|
||||
|
||||
func (s *SqlThreadStore) get(ex gorp.SqlExecutor, 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...)
|
||||
err := ex.SelectOne(&thread, query, args...)
|
||||
if err != nil {
|
||||
if err == sql.ErrNoRows {
|
||||
return nil, store.NewErrNotFound("Thread", id)
|
||||
@@ -474,7 +483,11 @@ func (s *SqlThreadStore) Delete(threadId string) error {
|
||||
}
|
||||
|
||||
func (s *SqlThreadStore) SaveMembership(membership *model.ThreadMembership) (*model.ThreadMembership, error) {
|
||||
if err := s.GetMaster().Insert(membership); err != nil {
|
||||
return s.saveMembership(s.GetMaster(), membership)
|
||||
}
|
||||
|
||||
func (s *SqlThreadStore) saveMembership(ex gorp.SqlExecutor, membership *model.ThreadMembership) (*model.ThreadMembership, error) {
|
||||
if err := ex.Insert(membership); err != nil {
|
||||
return nil, errors.Wrapf(err, "failed to save thread membership with postid=%s userid=%s", membership.PostId, membership.UserId)
|
||||
}
|
||||
|
||||
@@ -482,7 +495,11 @@ func (s *SqlThreadStore) SaveMembership(membership *model.ThreadMembership) (*mo
|
||||
}
|
||||
|
||||
func (s *SqlThreadStore) UpdateMembership(membership *model.ThreadMembership) (*model.ThreadMembership, error) {
|
||||
if _, err := s.GetMaster().Update(membership); err != nil {
|
||||
return s.updateMembership(s.GetMaster(), membership)
|
||||
}
|
||||
|
||||
func (s *SqlThreadStore) updateMembership(ex gorp.SqlExecutor, membership *model.ThreadMembership) (*model.ThreadMembership, error) {
|
||||
if _, err := ex.Update(membership); err != nil {
|
||||
return nil, errors.Wrapf(err, "failed to update thread membership with postid=%s userid=%s", membership.PostId, membership.UserId)
|
||||
}
|
||||
|
||||
@@ -509,8 +526,12 @@ func (s *SqlThreadStore) GetMembershipsForUser(userId, teamId string) ([]*model.
|
||||
}
|
||||
|
||||
func (s *SqlThreadStore) GetMembershipForUser(userId, postId string) (*model.ThreadMembership, error) {
|
||||
return s.getMembershipForUser(s.GetReplica(), userId, postId)
|
||||
}
|
||||
|
||||
func (s *SqlThreadStore) getMembershipForUser(ex gorp.SqlExecutor, userId, postId string) (*model.ThreadMembership, error) {
|
||||
var membership model.ThreadMembership
|
||||
err := s.GetReplica().SelectOne(&membership, "SELECT * from ThreadMemberships WHERE UserId = :UserId AND PostId = :PostId", map[string]interface{}{"UserId": userId, "PostId": postId})
|
||||
err := ex.SelectOne(&membership, "SELECT * from ThreadMemberships WHERE UserId = :UserId AND PostId = :PostId", map[string]interface{}{"UserId": userId, "PostId": postId})
|
||||
if err != nil {
|
||||
if err == sql.ErrNoRows {
|
||||
return nil, store.NewErrNotFound("Thread", postId)
|
||||
@@ -528,66 +549,89 @@ func (s *SqlThreadStore) DeleteMembershipForUser(userId string, postId string) e
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *SqlThreadStore) MaintainMembership(userId, postId string, following, incrementMentions, updateFollowing, updateViewedTimestamp, updateParticipants bool) (*model.ThreadMembership, error) {
|
||||
membership, err := s.GetMembershipForUser(userId, postId)
|
||||
// MaintainMembership creates or updates a thread membership for the given user
|
||||
// and post. This method is used to update the state of a membership in response
|
||||
// to some events like:
|
||||
// - post creation (mentions handling)
|
||||
// - channel marked unread
|
||||
// - user explicitly following a thread
|
||||
func (s *SqlThreadStore) MaintainMembership(userId, postId string, opts store.ThreadMembershipOpts) (*model.ThreadMembership, error) {
|
||||
trx, err := s.GetMaster().Begin()
|
||||
if err != nil {
|
||||
return nil, errors.Wrap(err, "begin_transaction")
|
||||
}
|
||||
defer finalizeTransaction(trx)
|
||||
|
||||
membership, err := s.getMembershipForUser(trx, userId, postId)
|
||||
now := utils.MillisFromTime(time.Now())
|
||||
// if memebership exists, update it if:
|
||||
// a. user started/stopped following a thread
|
||||
// b. mention count changed
|
||||
// c. user viewed a thread
|
||||
if err == nil {
|
||||
followingNeedsUpdate := (updateFollowing && !membership.Following || membership.Following != following)
|
||||
if followingNeedsUpdate || incrementMentions || updateViewedTimestamp {
|
||||
followingNeedsUpdate := (opts.UpdateFollowing && !membership.Following || membership.Following != opts.Following)
|
||||
if followingNeedsUpdate || opts.IncrementMentions || opts.UpdateViewedTimestamp {
|
||||
if followingNeedsUpdate {
|
||||
membership.Following = following
|
||||
membership.Following = opts.Following
|
||||
}
|
||||
if updateViewedTimestamp {
|
||||
if opts.UpdateViewedTimestamp {
|
||||
membership.LastViewed = now
|
||||
}
|
||||
membership.LastUpdated = now
|
||||
if incrementMentions {
|
||||
if opts.IncrementMentions {
|
||||
membership.UnreadMentions += 1
|
||||
}
|
||||
_, err = s.UpdateMembership(membership)
|
||||
if _, err = s.updateMembership(trx, membership); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
}
|
||||
return nil, err
|
||||
|
||||
if err = trx.Commit(); err != nil {
|
||||
return nil, errors.Wrap(err, "commit_transaction")
|
||||
}
|
||||
|
||||
return membership, err
|
||||
}
|
||||
|
||||
var nfErr *store.ErrNotFound
|
||||
|
||||
if !errors.As(err, &nfErr) {
|
||||
return nil, errors.Wrap(err, "failed to get thread membership")
|
||||
}
|
||||
mentions := 0
|
||||
if incrementMentions {
|
||||
mentions = 1
|
||||
|
||||
membership = &model.ThreadMembership{
|
||||
PostId: postId,
|
||||
UserId: userId,
|
||||
Following: opts.Following,
|
||||
LastUpdated: now,
|
||||
}
|
||||
var lastViewed int64
|
||||
if updateViewedTimestamp {
|
||||
lastViewed = now
|
||||
if opts.IncrementMentions {
|
||||
membership.UnreadMentions = 1
|
||||
}
|
||||
membership, err = s.SaveMembership(&model.ThreadMembership{
|
||||
PostId: postId,
|
||||
UserId: userId,
|
||||
Following: following,
|
||||
LastViewed: lastViewed,
|
||||
LastUpdated: now,
|
||||
UnreadMentions: int64(mentions),
|
||||
})
|
||||
if opts.UpdateViewedTimestamp {
|
||||
membership.LastViewed = now
|
||||
}
|
||||
membership, err = s.saveMembership(trx, membership)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
if updateParticipants {
|
||||
thread, err2 := s.Get(postId)
|
||||
if err2 != nil {
|
||||
return nil, err2
|
||||
if opts.UpdateParticipants {
|
||||
thread, getErr := s.get(trx, postId)
|
||||
if getErr != nil {
|
||||
return nil, getErr
|
||||
}
|
||||
if !thread.Participants.Contains(userId) {
|
||||
thread.Participants = append(thread.Participants, userId)
|
||||
_, err = s.Update(thread)
|
||||
if _, err = s.update(trx, thread); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
if err = trx.Commit(); err != nil {
|
||||
return nil, errors.Wrap(err, "commit_transaction")
|
||||
}
|
||||
|
||||
return membership, err
|
||||
}
|
||||
|
||||
|
||||
Ссылка в новой задаче
Block a user