From cfcdce9ae7909c57d3dddfd7ae4f0c2b537ca2ff Mon Sep 17 00:00:00 2001 From: Agniva De Sarker Date: Mon, 3 Jan 2022 10:40:16 +0530 Subject: [PATCH] Migrate post_store to sqlx (#19241) * Migrate post_store to sqlx There are several queries which should improve further if re-written with squirrel. HW tickets will be opened for those. https://community-daily.mattermost.com/boards/workspace/zyoahc9uapdn3xdptac6jb69ic/285b80a3-257d-41f6-8cf4-ed80ca9d92e5/495cdb4d-c13a-4992-8eb9-80cfee2819a4/ctnryw9mga7fu9kc16xsap4phuw ```release-note NONE ``` * Remove struct ```release-note NONE ``` * fix tests ```release-note NONE ``` --- model/utils.go | 5 + store/sqlstore/post_store.go | 602 +++++++++++++++++++++------------- store/storetest/post_store.go | 2 +- 3 files changed, 385 insertions(+), 224 deletions(-) diff --git a/model/utils.go b/model/utils.go index e917058865..36527ce5cf 100644 --- a/model/utils.go +++ b/model/utils.go @@ -135,6 +135,11 @@ func (si *StringInterface) Scan(value interface{}) error { return errors.New("received value is neither a byte slice nor string") } +// Value converts StringInterface to database value +func (si StringInterface) Value() (driver.Value, error) { + return json.Marshal(si) +} + var translateFunc i18n.TranslateFunc var translateFuncOnce sync.Once diff --git a/store/sqlstore/post_store.go b/store/sqlstore/post_store.go index 6047613cad..e246290858 100644 --- a/store/sqlstore/post_store.go +++ b/store/sqlstore/post_store.go @@ -14,7 +14,6 @@ import ( "sync" sq "github.com/Masterminds/squirrel" - "github.com/mattermost/gorp" "github.com/pkg/errors" "github.com/mattermost/mattermost-server/v6/einterfaces" @@ -36,7 +35,77 @@ type postWithExtra struct { ThreadReplyCount int64 IsFollowing *bool ThreadParticipants model.StringArray - model.Post + postInternal +} + +// postInternal is an internal struct without the `db:"-"` tags +// for sqlx to be able to scan them. This should go away once +// we have morph for migrations. +// It is named like this to avoid clashes with the variable name "post" +type postInternal struct { + Id string + CreateAt int64 + UpdateAt int64 + EditAt int64 + DeleteAt int64 + IsPinned bool + UserId string + ChannelId string + RootId string + OriginalId string + Message string + MessageSource string + Type string + Props model.StringInterface + Hashtags string + Filenames model.StringArray + FileIds model.StringArray + PendingPostId string + HasReactions bool + RemoteId *string + ReplyCount int64 + LastReplyAt int64 + Participants []*model.User + IsFollowing *bool + Metadata *model.PostMetadata +} + +func (p *postInternal) ToModel() *model.Post { + return &model.Post{ + Id: p.Id, + CreateAt: p.CreateAt, + UpdateAt: p.UpdateAt, + EditAt: p.EditAt, + DeleteAt: p.DeleteAt, + IsPinned: p.IsPinned, + UserId: p.UserId, + ChannelId: p.ChannelId, + RootId: p.RootId, + OriginalId: p.OriginalId, + Message: p.Message, + MessageSource: p.MessageSource, + Type: p.Type, + Props: p.Props, + Hashtags: p.Hashtags, + Filenames: p.Filenames, + FileIds: p.FileIds, + PendingPostId: p.PendingPostId, + HasReactions: p.HasReactions, + RemoteId: p.RemoteId, + ReplyCount: p.ReplyCount, + LastReplyAt: p.LastReplyAt, + Participants: p.Participants, + IsFollowing: p.IsFollowing, + Metadata: p.Metadata, + } +} + +func sliceToModel(posts []*postInternal) []*model.Post { + res := make([]*model.Post, 0, len(posts)) + for _, p := range posts { + res = append(res, p.ToModel()) + } + return res } func (s *SqlPostStore) ClearCaches() { @@ -238,12 +307,12 @@ func (s *SqlPostStore) SaveMultiple(posts []*model.Post) ([]*model.Post, int, er return nil, -1, errors.Wrap(err, "post_tosql") } - transaction, err := s.GetMaster().Begin() + transaction, err := s.GetMasterX().Beginx() if err != nil { return posts, -1, errors.Wrap(err, "begin_transaction") } - defer finalizeTransaction(transaction) + defer finalizeTransactionX(transaction) if _, err = transaction.Exec(query, args...); err != nil { return nil, -1, errors.Wrap(err, "failed to save Post") @@ -261,13 +330,24 @@ func (s *SqlPostStore) SaveMultiple(posts []*model.Post) ([]*model.Post, int, er for channelId, count := range channelNewPosts { countRoot := channelNewRootPosts[channelId] - if _, err = s.GetMaster().Exec("UPDATE Channels SET LastPostAt = GREATEST(:LastPostAt, LastPostAt), LastRootPostAt = GREATEST(:LastRootPostAt, LastRootPostAt), TotalMsgCount = TotalMsgCount + :Count, TotalMsgCountRoot = TotalMsgCountRoot + :CountRoot WHERE Id = :ChannelId", map[string]interface{}{"LastPostAt": maxDateNewPosts[channelId], "LastRootPostAt": maxDateNewRootPosts[channelId], "ChannelId": channelId, "Count": count, "CountRoot": countRoot}); err != nil { + if _, err = s.GetMasterX().NamedExec(`UPDATE Channels + SET LastPostAt = GREATEST(:lastpostat, LastPostAt), + LastRootPostAt = GREATEST(:lastrootpostat, LastRootPostAt), + TotalMsgCount = TotalMsgCount + :count, + TotalMsgCountRoot = TotalMsgCountRoot + :countroot + WHERE Id = :channelid`, map[string]interface{}{ + "lastpostat": maxDateNewPosts[channelId], + "lastrootpostat": maxDateNewRootPosts[channelId], + "channelid": channelId, + "count": count, + "countroot": countRoot, + }); err != nil { mlog.Warn("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.GetMasterX().Exec("UPDATE Posts SET UpdateAt = ? WHERE Id = ?", maxDateRootIds[rootId], rootId); err != nil { mlog.Warn("Error updating Post UpdateAt.", mlog.Err(err)) } } @@ -310,13 +390,18 @@ func (s *SqlPostStore) populateReplyCount(posts []*model.Post) error { RootId string Count int64 }{} - query := s.getQueryBuilder().Select("RootId, COUNT(Id) AS Count").From("Posts").Where(sq.Eq{"RootId": rootIds}).Where(sq.Eq{"DeleteAt": 0}).GroupBy("RootId") + query := s.getQueryBuilder(). + Select("RootId, COUNT(Id) AS Count"). + From("Posts"). + Where(sq.Eq{"RootId": rootIds}). + Where(sq.Eq{"DeleteAt": 0}). + GroupBy("RootId") queryString, args, err := query.ToSql() if err != nil { return errors.Wrap(err, "post_tosql") } - _, err = s.GetMaster().Select(&countList, queryString, args...) + err = s.GetMasterX().Select(&countList, queryString, args...) if err != nil { return errors.Wrap(err, "failed to count Posts") } @@ -353,19 +438,54 @@ func (s *SqlPostStore) Update(newPost *model.Post, oldPost *model.Post) (*model. return nil, err } - if _, err := s.GetMaster().Update(newPost); err != nil { + if _, err := s.GetMasterX().NamedExec(`UPDATE Posts + SET CreateAt=:CreateAt, + UpdateAt=:UpdateAt, + EditAt=:EditAt, + DeleteAt=:DeleteAt, + IsPinned=:IsPinned, + UserId=:UserId, + ChannelId=:ChannelId, + RootId=:RootId, + OriginalId=:OriginalId, + Message=:Message, + Type=:Type, + Props=:Props, + Hashtags=:Hashtags, + Filenames=:Filenames, + FileIds=:FileIds, + HasReactions=:HasReactions, + RemoteId=:RemoteId + WHERE + Id=:Id + `, newPost); err != nil { return nil, errors.Wrapf(err, "failed to update Post with id=%s", newPost.Id) } time := model.GetMillis() - s.GetMaster().Exec("UPDATE Channels SET LastPostAt = :LastPostAt WHERE Id = :ChannelId AND LastPostAt < :LastPostAt", map[string]interface{}{"LastPostAt": time, "ChannelId": newPost.ChannelId}) + if _, err := s.GetMasterX().Exec("UPDATE Channels SET LastPostAt = ? WHERE Id = ? AND LastPostAt < ?", time, newPost.ChannelId, time); err != nil { + return nil, errors.Wrap(err, "failed to update lastpostat of channels") + } if newPost.RootId != "" { - s.GetMaster().Exec("UPDATE Posts SET UpdateAt = :UpdateAt WHERE Id = :RootId AND UpdateAt < :UpdateAt", map[string]interface{}{"UpdateAt": time, "RootId": newPost.RootId}) + if _, err := s.GetMasterX().Exec("UPDATE Posts SET UpdateAt = ? WHERE Id = ? AND UpdateAt < ?", time, newPost.RootId, time); err != nil { + return nil, errors.Wrap(err, "failed to update updateAt of posts") + } } // mark the old post as deleted - s.GetMaster().Insert(oldPost) + builder := s.getQueryBuilder(). + Insert("Posts"). + Columns(postSliceColumns()...). + Values(postToSlice(oldPost)...) + query, args, err := builder.ToSql() + if err != nil { + return nil, errors.Wrap(err, "post_tosql") + } + _, err = s.GetMasterX().Exec(query, args...) + if err != nil { + return nil, errors.Wrap(err, "failed to insert the old post") + } return newPost, nil } @@ -380,21 +500,40 @@ func (s *SqlPostStore) OverwriteMultiple(posts []*model.Post) ([]*model.Post, in } } - tx, err := s.GetMaster().Begin() + tx, err := s.GetMasterX().Beginx() if err != nil { return nil, -1, errors.Wrap(err, "begin_transaction") } - for idx, post := range posts { - if _, err = tx.Update(post); err != nil { - txErr := tx.Rollback() - if txErr != nil { - return nil, idx, errors.Wrap(txErr, "rollback_transaction") - } + defer finalizeTransactionX(tx) - return nil, idx, errors.Wrap(err, "failed to update Post") + for idx, post := range posts { + if _, err2 := tx.NamedExec(`UPDATE Posts + SET CreateAt=:CreateAt, + UpdateAt=:UpdateAt, + EditAt=:EditAt, + DeleteAt=:DeleteAt, + IsPinned=:IsPinned, + UserId=:UserId, + ChannelId=:ChannelId, + RootId=:RootId, + OriginalId=:OriginalId, + Message=:Message, + Type=:Type, + Props=:Props, + Hashtags=:Hashtags, + Filenames=:Filenames, + FileIds=:FileIds, + HasReactions=:HasReactions, + RemoteId=:RemoteId + WHERE + Id=:Id + `, post); err2 != nil { + return nil, idx, errors.Wrapf(err2, "failed to update Post with id=%s", post.Id) } if post.RootId != "" { - tx.Exec("UPDATE Threads SET LastReplyAt = :UpdateAt WHERE PostId = :RootId", map[string]interface{}{"UpdateAt": updateAt, "RootId": post.Id}) + if _, err2 := tx.Exec("UPDATE Threads SET LastReplyAt = ? WHERE PostId = ?", updateAt, post.Id); err2 != nil { + return nil, idx, errors.Wrapf(err2, "failed to update Threads with postid=%s", post.Id) + } } } err = tx.Commit() @@ -426,10 +565,11 @@ func (s *SqlPostStore) GetFlaggedPostsForChannel(userId, channelId string, offse return s.getFlaggedPosts(userId, channelId, "", offset, limit) } +// TODO: convert to squirrel HW func (s *SqlPostStore) getFlaggedPosts(userId, channelId, teamId string, offset int, limit int) (*model.PostList, error) { pl := model.NewPostList() - var posts []*model.Post + posts := []*postInternal{} query := ` SELECT A.*, (SELECT count(Posts.Id) FROM Posts WHERE Posts.RootId = (CASE WHEN A.RootId = '' THEN A.Id ELSE A.RootId END) AND Posts.DeleteAt = 0) as ReplyCount @@ -447,8 +587,8 @@ func (s *SqlPostStore) getFlaggedPosts(userId, channelId, teamId string, offset FROM Preferences WHERE - UserId = :UserId - AND Category = :Category + UserId = ? + AND Category = ? ) CHANNEL_FILTER AND DeleteAt = 0 @@ -464,55 +604,51 @@ func (s *SqlPostStore) getFlaggedPosts(userId, channelId, teamId string, offset ChannelMembers WHERE Id = ChannelId - AND UserId = :UserId + AND UserId = ? ) TEAM_FILTER ORDER BY CreateAt DESC - LIMIT :Limit OFFSET :Offset` + LIMIT ? OFFSET ?` - queryParams := map[string]interface{}{ - "UserId": userId, - "Category": model.PreferenceCategoryFlaggedPost, - "Offset": offset, - "Limit": limit, - } + queryParams := []interface{}{userId, model.PreferenceCategoryFlaggedPost} - teamClause, queryParams := s.buildFlaggedPostTeamFilterClause(teamId, queryParams) - query = strings.Replace(query, "TEAM_FILTER", teamClause, 1) - - channelClause, queryParams := s.buildFlaggedPostChannelFilterClause(channelId, queryParams) + var channelClause, teamClause string + channelClause, queryParams = s.buildFlaggedPostChannelFilterClause(channelId, queryParams) query = strings.Replace(query, "CHANNEL_FILTER", channelClause, 1) - if _, err := s.GetReplica().Select(&posts, query, queryParams); err != nil { + queryParams = append(queryParams, userId) + + teamClause, queryParams = s.buildFlaggedPostTeamFilterClause(teamId, queryParams) + query = strings.Replace(query, "TEAM_FILTER", teamClause, 1) + + queryParams = append(queryParams, limit, offset) + + if err := s.GetReplicaX().Select(&posts, query, queryParams...); err != nil { return nil, errors.Wrap(err, "failed to find Posts") } for _, post := range posts { - pl.AddPost(post) + pl.AddPost(post.ToModel()) pl.AddOrder(post.Id) } return pl, nil } -func (s *SqlPostStore) buildFlaggedPostTeamFilterClause(teamId string, queryParams map[string]interface{}) (string, map[string]interface{}) { +func (s *SqlPostStore) buildFlaggedPostTeamFilterClause(teamId string, queryParams []interface{}) (string, []interface{}) { if teamId == "" { return "", queryParams } - queryParams["TeamId"] = teamId - - return "AND B.TeamId = :TeamId OR B.TeamId = ''", queryParams + return "AND B.TeamId = ? OR B.TeamId = ''", append(queryParams, teamId) } -func (s *SqlPostStore) buildFlaggedPostChannelFilterClause(channelId string, queryParams map[string]interface{}) (string, map[string]interface{}) { +func (s *SqlPostStore) buildFlaggedPostChannelFilterClause(channelId string, queryParams []interface{}) (string, []interface{}) { if channelId == "" { return "", queryParams } - queryParams["ChannelId"] = channelId - - return "AND ChannelId = :ChannelId", queryParams + return "AND ChannelId = ?", append(queryParams, channelId) } func (s *SqlPostStore) getPostWithCollapsedThreads(id, userID string, extended bool) (*model.PostList, error) { @@ -540,7 +676,7 @@ func (s *SqlPostStore) getPostWithCollapsedThreads(id, userID string, extended b Where(sq.Eq{"DeleteAt": 0}). Where(sq.Eq{"Id": id}).ToSql() - err := s.GetReplica().SelectOne(&post, postFetchQuery, args...) + err := s.GetReplicaX().Get(&post, postFetchQuery, args...) if err != nil { if err == sql.ErrNoRows { return nil, store.NewErrNotFound("Post", id) @@ -549,8 +685,8 @@ func (s *SqlPostStore) getPostWithCollapsedThreads(id, userID string, extended b return nil, errors.Wrapf(err, "failed to get Post with id=%s", id) } - var posts []*model.Post - _, err = s.GetReplica().Select(&posts, "SELECT * FROM Posts WHERE Posts.RootId = :RootId AND DeleteAt = 0", map[string]interface{}{"RootId": id}) + posts := []*model.Post{} + err = s.GetReplicaX().Select(&posts, "SELECT * FROM Posts WHERE Posts.RootId = ? AND DeleteAt = 0", id) if err != nil { return nil, errors.Wrapf(err, "failed to find Posts for thread %s", id) } @@ -576,9 +712,9 @@ func (s *SqlPostStore) Get(ctx context.Context, id string, skipFetchThreads, col return nil, store.NewErrInvalidInput("Post", "id", id) } - var post model.Post - postFetchQuery := "SELECT p.*, (SELECT count(Posts.Id) FROM Posts WHERE Posts.RootId = (CASE WHEN p.RootId = '' THEN p.Id ELSE p.RootId END) AND Posts.DeleteAt = 0) as ReplyCount FROM Posts p WHERE p.Id = :Id AND p.DeleteAt = 0" - err := s.DBFromContext(ctx).SelectOne(&post, postFetchQuery, map[string]interface{}{"Id": id}) + var post postInternal + postFetchQuery := "SELECT p.*, (SELECT count(Posts.Id) FROM Posts WHERE Posts.RootId = (CASE WHEN p.RootId = '' THEN p.Id ELSE p.RootId END) AND Posts.DeleteAt = 0) as ReplyCount FROM Posts p WHERE p.Id = ? AND p.DeleteAt = 0" + err := s.DBXFromContext(ctx).Get(&post, postFetchQuery, id) if err != nil { if err == sql.ErrNoRows { return nil, store.NewErrNotFound("Post", id) @@ -586,7 +722,7 @@ func (s *SqlPostStore) Get(ctx context.Context, id string, skipFetchThreads, col return nil, errors.Wrapf(err, "failed to get Post with id=%s", id) } - pl.AddPost(&post) + pl.AddPost(post.ToModel()) pl.AddOrder(id) if !skipFetchThreads { rootId := post.RootId @@ -599,14 +735,14 @@ func (s *SqlPostStore) Get(ctx context.Context, id string, skipFetchThreads, col return nil, errors.Wrapf(err, "invalid rootId with value=%s", rootId) } - var posts []*model.Post - _, err = s.GetReplica().Select(&posts, "SELECT *, (SELECT count(Id) FROM Posts WHERE Posts.RootId = (CASE WHEN p.RootId = '' THEN p.Id ELSE p.RootId END) AND Posts.DeleteAt = 0) as ReplyCount FROM Posts p WHERE (Id = :Id OR RootId = :RootId) AND DeleteAt = 0", map[string]interface{}{"Id": rootId, "RootId": rootId}) + posts := []*postInternal{} + err = s.GetReplicaX().Select(&posts, "SELECT *, (SELECT count(Id) FROM Posts WHERE Posts.RootId = (CASE WHEN p.RootId = '' THEN p.Id ELSE p.RootId END) AND Posts.DeleteAt = 0) as ReplyCount FROM Posts p WHERE (Id = ? OR RootId = ?) AND DeleteAt = 0", rootId, rootId) if err != nil { return nil, errors.Wrap(err, "failed to find Posts") } for _, p := range posts { - pl.AddPost(p) + pl.AddPost(p.ToModel()) pl.AddOrder(p.Id) } } @@ -634,8 +770,8 @@ func (s *SqlPostStore) GetSingle(id string, inclDeleted bool) (*model.Post, erro return nil, errors.Wrap(err, "getsingleincldeleted_tosql") } - var post model.Post - err = s.GetReplica().SelectOne(&post, queryString, args...) + var post postInternal + err = s.GetReplicaX().Get(&post, queryString, args...) if err != nil { if err == sql.ErrNoRows { return nil, store.NewErrNotFound("Post", id) @@ -643,7 +779,7 @@ func (s *SqlPostStore) GetSingle(id string, inclDeleted bool) (*model.Post, erro return nil, errors.Wrapf(err, "failed to get Post with id=%s", id) } - return &post, nil + return post.ToModel(), nil } type etagPosts struct { @@ -664,7 +800,7 @@ func (s *SqlPostStore) GetEtag(channelId string, allowFromCache, collapsedThread sql, args, _ := q.ToSql() var et etagPosts - err := s.GetReplica().SelectOne(&et, sql, args...) + err := s.GetReplicaX().Get(&et, sql, args...) var result string if err != nil { result = fmt.Sprintf("%v.%v", model.CurrentVersion, model.GetMillis()) @@ -813,7 +949,7 @@ func (s *SqlPostStore) PermanentDeleteByUser(userId string) error { for found { var ids []string - _, err := s.GetMaster().Select(&ids, "SELECT Id FROM Posts WHERE UserId = :UserId LIMIT 1000", map[string]interface{}{"UserId": userId}) + err := s.GetMasterX().Select(&ids, "SELECT Id FROM Posts WHERE UserId = ? LIMIT 1000", userId) if err != nil { return errors.Wrapf(err, "failed to find Posts with userId=%s", userId) } @@ -898,16 +1034,16 @@ func (s *SqlPostStore) prepareThreadedResponse(posts []*postWithExtra, extended, } processPost := func(p *postWithExtra) error { - p.Post.ReplyCount = p.ThreadReplyCount + p.postInternal.ReplyCount = p.ThreadReplyCount if p.IsFollowing != nil { - p.Post.IsFollowing = model.NewBool(*p.IsFollowing) + p.postInternal.IsFollowing = model.NewBool(*p.IsFollowing) } for _, userID := range p.ThreadParticipants { participant, ok := usersMap[userID] if !ok { return errors.New("cannot find thread participant with id=" + userID) } - p.Post.Participants = append(p.Post.Participants, participant) + p.postInternal.Participants = append(p.postInternal.Participants, participant) } return nil } @@ -923,7 +1059,8 @@ func (s *SqlPostStore) prepareThreadedResponse(posts []*postWithExtra, extended, if err := processPost(posts[idx]); err != nil { return nil, err } - list.AddPost(&posts[idx].Post) + post := &posts[idx].postInternal + list.AddPost(post.ToModel()) list.AddOrder(posts[idx].Id) } @@ -956,8 +1093,7 @@ func (s *SqlPostStore) getPostsCollapsedThreads(options model.GetPostsOptions) ( Offset(uint64(offset)). OrderBy("CreateAt DESC").ToSql() - _, err := s.GetReplica().Select(&posts, postFetchQuery, args...) - + err := s.GetReplicaX().Select(&posts, postFetchQuery, args...) if err != nil { return nil, errors.Wrapf(err, "failed to find Posts with channelId=%s", options.ChannelId) } @@ -1040,8 +1176,7 @@ func (s *SqlPostStore) getPostsSinceCollapsedThreads(options model.GetPostsSince Where(sq.Eq{"RootId": ""}). OrderBy("CreateAt DESC").ToSql() - _, err := s.GetReplica().Select(&posts, postFetchQuery, args...) - + err := s.GetReplicaX().Select(&posts, postFetchQuery, args...) if err != nil { return nil, errors.Wrapf(err, "failed to find Posts with channelId=%s", options.ChannelId) } @@ -1054,7 +1189,7 @@ func (s *SqlPostStore) GetPostsSince(options model.GetPostsSinceOptions, allowFr return s.getPostsSinceCollapsedThreads(options) } - var posts []*model.Post + posts := []*postInternal{} order := "DESC" if options.SortAscending { @@ -1068,6 +1203,7 @@ func (s *SqlPostStore) GetPostsSince(options model.GetPostsSinceOptions, allowFr replyCountQuery2 = `, (SELECT COUNT(Posts.Id) FROM Posts WHERE Posts.RootId = (CASE WHEN cte.RootId = '' THEN cte.Id ELSE cte.RootId END) AND Posts.DeleteAt = 0) as ReplyCount` } var query string + var params []interface{} // union of IDs and then join to get full posts is faster in mysql if s.DriverName() == model.DatabaseDriverMysql { @@ -1077,8 +1213,8 @@ func (s *SqlPostStore) GetPostsSince(options model.GetPostsSinceOptions, allowFr FROM Posts p2 WHERE - (UpdateAt > :Time - AND ChannelId = :ChannelId) + (UpdateAt > ? + AND ChannelId = ?) LIMIT 1000) UNION (SELECT @@ -1093,26 +1229,29 @@ func (s *SqlPostStore) GetPostsSince(options model.GetPostsSinceOptions, allowFr FROM Posts WHERE - UpdateAt > :Time - AND ChannelId = :ChannelId + UpdateAt > ? + AND ChannelId = ? LIMIT 1000) temp_tab)) ) j ON p1.Id = j.Id ORDER BY CreateAt ` + order + + params = []interface{}{options.Time, options.ChannelId, options.Time, options.ChannelId} } else if s.DriverName() == model.DatabaseDriverPostgres { query = `WITH cte AS (SELECT * FROM Posts WHERE - UpdateAt > :Time AND ChannelId = :ChannelId + UpdateAt > ? AND ChannelId = ? LIMIT 1000) (SELECT *` + replyCountQuery2 + ` FROM cte) UNION (SELECT *` + replyCountQuery1 + ` FROM Posts p1 WHERE id in (SELECT rootid FROM cte)) ORDER BY CreateAt ` + order - } - _, err := s.GetReplica().Select(&posts, query, map[string]interface{}{"ChannelId": options.ChannelId, "Time": options.Time}) + params = []interface{}{options.Time, options.ChannelId} + } + err := s.GetReplicaX().Select(&posts, query, params...) if err != nil { return nil, errors.Wrapf(err, "failed to find Posts with channelId=%s", options.ChannelId) } @@ -1120,7 +1259,7 @@ func (s *SqlPostStore) GetPostsSince(options model.GetPostsSinceOptions, allowFr list := model.NewPostList() for _, p := range posts { - list.AddPost(p) + list.AddPost(p.ToModel()) if p.UpdateAt > options.Time { list.AddOrder(p.Id) } @@ -1135,23 +1274,17 @@ func (s *SqlPostStore) HasAutoResponsePostByUserSince(options model.GetPostsSinc FROM Posts WHERE - UpdateAt >= :Time + UpdateAt >= ? AND - ChannelId = :ChannelId + ChannelId = ? AND - UserId = :UserId + UserId = ? AND - Type = :Type + Type = ? LIMIT 1)` var exist bool - err := s.GetReplica().SelectOne(&exist, query, map[string]interface{}{ - "ChannelId": options.ChannelId, - "Time": options.Time, - "UserId": userId, - "Type": model.PostTypeAutoResponder, - }) - + err := s.GetReplicaX().Get(&exist, query, options.Time, options.ChannelId, userId, model.PostTypeAutoResponder) if err != nil { return false, errors.Wrapf(err, "failed to check if autoresponse posts in channelId=%s for userId=%s since %s", options.ChannelId, userId, model.GetTimeForMillis(options.Time)) @@ -1185,8 +1318,8 @@ func (s *SqlPostStore) GetPostsSinceForSync(options model.GetPostsSinceForSyncOp return nil, cursor, errors.Wrap(err, "getpostssinceforsync_tosql") } - var posts []*model.Post - _, err = s.GetReplica().Select(&posts, queryString, args...) + posts := []*model.Post{} + err = s.GetReplicaX().Select(&posts, queryString, args...) if err != nil { return nil, cursor, errors.Wrapf(err, "error getting Posts with channelId=%s", options.ChannelId) } @@ -1216,8 +1349,8 @@ func (s *SqlPostStore) getPostsAround(before bool, options model.GetPostsOptions } offset := options.Page * options.PerPage - var posts []*postWithExtra - var parents []*model.Post + posts := []*postWithExtra{} + parents := []*postInternal{} var direction string var sort string @@ -1271,7 +1404,7 @@ func (s *SqlPostStore) getPostsAround(before bool, options model.GetPostsOptions if err != nil { return nil, errors.Wrap(err, "post_tosql") } - _, err = s.GetMaster().Select(&posts, queryString, args...) + err = s.GetMasterX().Select(&posts, queryString, args...) if err != nil { return nil, errors.Wrapf(err, "failed to find Posts with channelId=%s", options.ChannelId) } @@ -1306,7 +1439,7 @@ func (s *SqlPostStore) getPostsAround(before bool, options model.GetPostsOptions if nErr != nil { return nil, errors.Wrap(nErr, "post_tosql") } - _, nErr = s.GetMaster().Select(&parents, rootQueryString, rootArgs...) + nErr = s.GetMasterX().Select(&parents, rootQueryString, rootArgs...) if nErr != nil { return nil, errors.Wrapf(nErr, "failed to find Posts with channelId=%s", options.ChannelId) } @@ -1318,7 +1451,7 @@ func (s *SqlPostStore) getPostsAround(before bool, options model.GetPostsOptions } for _, p := range parents { - list.AddPost(p) + list.AddPost(p.ToModel()) } return list, nil @@ -1375,7 +1508,7 @@ func (s *SqlPostStore) getPostIdAroundTime(channelId string, time int64, before } var postId string - if err := s.GetMaster().SelectOne(&postId, queryString, args...); err != nil { + if err := s.GetMasterX().Get(&postId, queryString, args...); err != nil { if err != sql.ErrNoRows { return "", errors.Wrapf(err, "failed to get Post id with channelId=%s", channelId) } @@ -1415,29 +1548,29 @@ func (s *SqlPostStore) GetPostAfterTime(channelId string, time int64, collapsedT return nil, errors.Wrap(err, "post_tosql") } - var post *model.Post - if err := s.GetMaster().SelectOne(&post, queryString, args...); err != nil { + var post model.Post + if err := s.GetMasterX().Get(&post, queryString, args...); err != nil { if err != sql.ErrNoRows { return nil, errors.Wrapf(err, "failed to get Post with channelId=%s", channelId) } } - return post, nil + return &post, nil } func (s *SqlPostStore) getRootPosts(channelId string, offset int, limit int, skipFetchThreads bool) ([]*model.Post, error) { - var posts []*model.Post + posts := []*postInternal{} var fetchQuery string if skipFetchThreads { - fetchQuery = "SELECT p.*, (SELECT COUNT(Posts.Id) FROM Posts WHERE Posts.RootId = (CASE WHEN p.RootId = '' THEN p.Id ELSE p.RootId END) AND Posts.DeleteAt = 0) as ReplyCount FROM Posts p WHERE ChannelId = :ChannelId AND DeleteAt = 0 ORDER BY CreateAt DESC LIMIT :Limit OFFSET :Offset" + fetchQuery = "SELECT p.*, (SELECT COUNT(Posts.Id) FROM Posts WHERE Posts.RootId = (CASE WHEN p.RootId = '' THEN p.Id ELSE p.RootId END) AND Posts.DeleteAt = 0) as ReplyCount FROM Posts p WHERE ChannelId = ? AND DeleteAt = 0 ORDER BY CreateAt DESC LIMIT ? OFFSET ?" } else { - fetchQuery = "SELECT * FROM Posts WHERE ChannelId = :ChannelId AND DeleteAt = 0 ORDER BY CreateAt DESC LIMIT :Limit OFFSET :Offset" + fetchQuery = "SELECT * FROM Posts WHERE ChannelId = ? AND DeleteAt = 0 ORDER BY CreateAt DESC LIMIT ? OFFSET ?" } - _, err := s.GetReplica().Select(&posts, fetchQuery, map[string]interface{}{"ChannelId": channelId, "Offset": offset, "Limit": limit}) + err := s.GetReplicaX().Select(&posts, fetchQuery, channelId, limit, offset) if err != nil { return nil, errors.Wrap(err, "failed to find Posts") } - return posts, nil + return sliceToModel(posts), nil } func (s *SqlPostStore) getParentsPosts(channelId string, offset int, limit int, skipFetchThreads bool) ([]*model.Post, error) { @@ -1446,9 +1579,7 @@ func (s *SqlPostStore) getParentsPosts(channelId string, offset int, limit int, } // query parent Ids first - var roots []*struct { - RootId string - } + roots := []string{} rootQuery := ` SELECT DISTINCT q.RootId @@ -1458,24 +1589,25 @@ func (s *SqlPostStore) getParentsPosts(channelId string, offset int, limit int, FROM Posts WHERE - ChannelId = :ChannelId + ChannelId = ? AND DeleteAt = 0 ORDER BY CreateAt DESC - LIMIT :Limit OFFSET :Offset) q + LIMIT ? OFFSET ?) q WHERE q.RootId != ''` - _, err := s.GetReplica().Select(&roots, rootQuery, map[string]interface{}{"ChannelId": channelId, "Offset": offset, "Limit": limit}) + err := s.GetReplicaX().Select(&roots, rootQuery, channelId, limit, offset) if err != nil { return nil, errors.Wrap(err, "failed to find Posts") } if len(roots) == 0 { return nil, nil } + // TODO: convert to squirrel HW params := make(map[string]interface{}) placeholders := make([]string, len(roots)) for idx, r := range roots { key := fmt.Sprintf(":Root%v", idx) - params[key[1:]] = r.RootId + params[key[1:]] = r placeholders[idx] = key } placeholderString := strings.Join(placeholders, ", ") @@ -1505,7 +1637,7 @@ func (s *SqlPostStore) getParentsPosts(channelId string, offset int, limit int, } func (s *SqlPostStore) getParentsPostsPostgreSQL(channelId string, offset int, limit int, skipFetchThreads bool) ([]*model.Post, error) { - var posts []*model.Post + posts := []*postInternal{} replyCountQuery := "" onStatement := "q1.RootId = q2.Id" if skipFetchThreads { @@ -1513,7 +1645,7 @@ func (s *SqlPostStore) getParentsPostsPostgreSQL(channelId string, offset int, l } else { onStatement += " OR q1.RootId = q2.RootId" } - _, err := s.GetReplica().Select(&posts, + err := s.GetReplicaX().Select(&posts, `SELECT q2.*`+replyCountQuery+` FROM Posts q2 @@ -1526,21 +1658,20 @@ func (s *SqlPostStore) getParentsPostsPostgreSQL(channelId string, offset int, l FROM Posts WHERE - ChannelId = :ChannelId1 + ChannelId = ? AND DeleteAt = 0 ORDER BY CreateAt DESC - LIMIT :Limit OFFSET :Offset) q3 + LIMIT ? OFFSET ?) q3 WHERE q3.RootId != '') q1 ON `+onStatement+` WHERE - ChannelId = :ChannelId2 - AND DeleteAt = 0 - ORDER BY CreateAt`, - map[string]interface{}{"ChannelId1": channelId, "Offset": offset, "Limit": limit, "ChannelId2": channelId}) + ChannelId = ? + AND DeleteAt = 0 + ORDER BY CreateAt`, channelId, limit, offset, channelId) if err != nil { return nil, errors.Wrapf(err, "failed to find Posts with channelId=%s", channelId) } - return posts, nil + return sliceToModel(posts), nil } var specialSearchChar = []string{ @@ -1697,6 +1828,7 @@ func (s *SqlPostStore) Search(teamId string, userId string, params *model.Search return s.search(teamId, userId, params, true, true) } +// TODO: convert to squirrel func (s *SqlPostStore) search(teamId string, userId string, params *model.SearchParams, channelsByName bool, userByUsername bool) (*model.PostList, error) { queryParams := map[string]interface{}{ "UserId": userId, @@ -1871,7 +2003,9 @@ func removeMysqlStopWordsFromTerms(terms string) (string, error) { return strings.Join(newTerms, " "), nil } +// TODO: convert to squirrel HW func (s *SqlPostStore) AnalyticsUserCountsWithPostsByDay(teamId string) (model.AnalyticsRows, error) { + var args []interface{} query := `SELECT DISTINCT DATE(FROM_UNIXTIME(Posts.CreateAt / 1000)) AS Name, @@ -1879,12 +2013,13 @@ func (s *SqlPostStore) AnalyticsUserCountsWithPostsByDay(teamId string) (model.A FROM Posts` if teamId != "" { - query += " INNER JOIN Channels ON Posts.ChannelId = Channels.Id AND Channels.TeamId = :TeamId AND" + query += " INNER JOIN Channels ON Posts.ChannelId = Channels.Id AND Channels.TeamId = ? AND" + args = []interface{}{teamId} } else { query += " WHERE" } - query += ` Posts.CreateAt >= :StartTime AND Posts.CreateAt <= :EndTime + query += ` Posts.CreateAt >= ? AND Posts.CreateAt <= ? GROUP BY DATE(FROM_UNIXTIME(Posts.CreateAt / 1000)) ORDER BY Name DESC LIMIT 30` @@ -1896,12 +2031,13 @@ func (s *SqlPostStore) AnalyticsUserCountsWithPostsByDay(teamId string) (model.A FROM Posts` if teamId != "" { - query += " INNER JOIN Channels ON Posts.ChannelId = Channels.Id AND Channels.TeamId = :TeamId AND" + query += " INNER JOIN Channels ON Posts.ChannelId = Channels.Id AND Channels.TeamId = ? AND" + args = []interface{}{teamId} } else { query += " WHERE" } - query += ` Posts.CreateAt >= :StartTime AND Posts.CreateAt <= :EndTime + query += ` Posts.CreateAt >= ? AND Posts.CreateAt <= ? GROUP BY DATE(TO_TIMESTAMP(Posts.CreateAt / 1000)) ORDER BY Name DESC LIMIT 30` @@ -1909,20 +2045,23 @@ func (s *SqlPostStore) AnalyticsUserCountsWithPostsByDay(teamId string) (model.A end := utils.MillisFromTime(utils.EndOfDay(utils.Yesterday())) start := utils.MillisFromTime(utils.StartOfDay(utils.Yesterday().AddDate(0, 0, -31))) + args = append(args, start, end) - var rows model.AnalyticsRows - _, err := s.GetReplica().Select( + rows := model.AnalyticsRows{} + err := s.GetReplicaX().Select( &rows, query, - map[string]interface{}{"TeamId": teamId, "StartTime": start, "EndTime": end}) + args...) if err != nil { return nil, errors.Wrapf(err, "failed to find Posts with teamId=%s", teamId) } return rows, nil } +// TODO: convert to squirrel HW func (s *SqlPostStore) AnalyticsPostCountsByDay(options *model.AnalyticsPostCountsOptions) (model.AnalyticsRows, error) { + var args []interface{} query := `SELECT DATE(FROM_UNIXTIME(Posts.CreateAt / 1000)) AS Name, @@ -1934,13 +2073,14 @@ func (s *SqlPostStore) AnalyticsPostCountsByDay(options *model.AnalyticsPostCoun } if options.TeamId != "" { - query += " INNER JOIN Channels ON Posts.ChannelId = Channels.Id AND Channels.TeamId = :TeamId AND" + query += " INNER JOIN Channels ON Posts.ChannelId = Channels.Id AND Channels.TeamId = ? AND" + args = []interface{}{options.TeamId} } else { query += " WHERE" } - query += ` Posts.CreateAt <= :EndTime - AND Posts.CreateAt >= :StartTime + query += ` Posts.CreateAt <= ? + AND Posts.CreateAt >= ? GROUP BY DATE(FROM_UNIXTIME(Posts.CreateAt / 1000)) ORDER BY Name DESC LIMIT 30` @@ -1956,13 +2096,14 @@ func (s *SqlPostStore) AnalyticsPostCountsByDay(options *model.AnalyticsPostCoun } if options.TeamId != "" { - query += " INNER JOIN Channels ON Posts.ChannelId = Channels.Id AND Channels.TeamId = :TeamId AND" + query += " INNER JOIN Channels ON Posts.ChannelId = Channels.Id AND Channels.TeamId = ? AND" + args = []interface{}{options.TeamId} } else { query += " WHERE" } - query += ` Posts.CreateAt <= :EndTime - AND Posts.CreateAt >= :StartTime + query += ` Posts.CreateAt <= ? + AND Posts.CreateAt >= ? GROUP BY DATE(TO_TIMESTAMP(Posts.CreateAt / 1000)) ORDER BY Name DESC LIMIT 30` @@ -1973,12 +2114,13 @@ func (s *SqlPostStore) AnalyticsPostCountsByDay(options *model.AnalyticsPostCoun if options.YesterdayOnly { start = utils.MillisFromTime(utils.StartOfDay(utils.Yesterday().AddDate(0, 0, -1))) } + args = append(args, end, start) - var rows model.AnalyticsRows - _, err := s.GetReplica().Select( + rows := model.AnalyticsRows{} + err := s.GetReplicaX().Select( &rows, query, - map[string]interface{}{"TeamId": options.TeamId, "StartTime": start, "EndTime": end}) + args...) if err != nil { return nil, errors.Wrapf(err, "failed to find Posts with teamId=%s", options.TeamId) } @@ -2009,7 +2151,8 @@ func (s *SqlPostStore) AnalyticsPostCount(teamId string, mustHaveFile bool, must return 0, errors.Wrap(err, "post_tosql") } - v, err := s.GetReplica().SelectInt(queryString, args...) + var v int64 + err = s.GetReplicaX().Get(&v, queryString, args...) if err != nil { return 0, errors.Wrap(err, "failed to count Posts") } @@ -2018,11 +2161,10 @@ func (s *SqlPostStore) AnalyticsPostCount(teamId string, mustHaveFile bool, must } func (s *SqlPostStore) GetPostsCreatedAt(channelId string, time int64) ([]*model.Post, error) { - query := `SELECT * FROM Posts WHERE CreateAt = :CreateAt AND ChannelId = :ChannelId` - - var posts []*model.Post - _, err := s.GetReplica().Select(&posts, query, map[string]interface{}{"CreateAt": time, "ChannelId": channelId}) + query := `SELECT * FROM Posts WHERE CreateAt = ? AND ChannelId = ?` + posts := []*model.Post{} + err := s.GetReplicaX().Select(&posts, query, time, channelId) if err != nil { return nil, errors.Wrapf(err, "failed to find Posts with channelId=%s", channelId) } @@ -2030,25 +2172,30 @@ func (s *SqlPostStore) GetPostsCreatedAt(channelId string, time int64) ([]*model } func (s *SqlPostStore) GetPostsByIds(postIds []string) ([]*model.Post, error) { - keys, params := MapStringsToQueryParams(postIds, "Post") + baseQuery := s.getQueryBuilder().Select("p.*, (SELECT count(Posts.Id) FROM Posts WHERE Posts.RootId = (CASE WHEN p.RootId = '' THEN p.Id ELSE p.RootId END) AND Posts.DeleteAt = 0) as ReplyCount"). + From("Posts p"). + Where(sq.Eq{"p.Id": postIds}). + OrderBy("CreateAt DESC") - query := `SELECT p.*, (SELECT count(Posts.Id) FROM Posts WHERE Posts.RootId = (CASE WHEN p.RootId = '' THEN p.Id ELSE p.RootId END) AND Posts.DeleteAt = 0) as ReplyCount FROM Posts p WHERE p.Id IN ` + keys + ` ORDER BY CreateAt DESC` - - var posts []*model.Post - _, err := s.GetReplica().Select(&posts, query, params) + query, args, err := baseQuery.ToSql() + if err != nil { + return nil, errors.Wrap(err, "getPostsByIds_tosql") + } + posts := []*postInternal{} + err = s.GetReplicaX().Select(&posts, query, args...) if err != nil { return nil, errors.Wrap(err, "failed to find Posts") } if len(posts) == 0 { return nil, store.NewErrNotFound("Post", fmt.Sprintf("postIds=%v", postIds)) } - return posts, nil + return sliceToModel(posts), nil } func (s *SqlPostStore) GetPostsBatchForIndexing(startTime int64, endTime int64, limit int) ([]*model.PostForIndexing, error) { - var posts []*model.PostForIndexing - _, err := s.GetSearchReplica().Select(&posts, + posts := []*model.PostForIndexing{} + err := s.GetSearchReplicaX().Select(&posts, `SELECT PostsQuery.*, Channels.TeamId, ParentPosts.CreateAt ParentCreateAt FROM ( @@ -2057,13 +2204,13 @@ func (s *SqlPostStore) GetPostsBatchForIndexing(startTime int64, endTime int64, FROM Posts WHERE - Posts.CreateAt >= :StartTime + Posts.CreateAt >= ? AND - Posts.CreateAt < :EndTime + Posts.CreateAt < ? ORDER BY CreateAt ASC LIMIT - 1000 + ? ) AS PostsQuery @@ -2075,7 +2222,7 @@ func (s *SqlPostStore) GetPostsBatchForIndexing(startTime int64, endTime int64, Posts ParentPosts ON PostsQuery.RootId = ParentPosts.Id`, - map[string]interface{}{"StartTime": startTime, "EndTime": endTime, "NumPosts": limit}) + startTime, endTime, limit) if err != nil { return nil, errors.Wrap(err, "failed to find Posts") @@ -2111,11 +2258,10 @@ func (s *SqlPostStore) DeleteOrphanedRows(limit int) (deleted int64, err error) SELECT Posts.Id FROM Posts LEFT JOIN Channels ON Posts.ChannelId = Channels.Id WHERE Channels.Id IS NULL - LIMIT :Limit + LIMIT ? ) AS A )` - props := map[string]interface{}{"Limit": limit} - result, err := s.GetMaster().Exec(query, props) + result, err := s.GetMasterX().Exec(query, limit) if err != nil { return } @@ -2126,12 +2272,12 @@ func (s *SqlPostStore) DeleteOrphanedRows(limit int) (deleted int64, err error) func (s *SqlPostStore) PermanentDeleteBatch(endTime int64, limit int64) (int64, error) { var query string if s.DriverName() == "postgres" { - query = "DELETE from Posts WHERE Id = any (array (SELECT Id FROM Posts WHERE CreateAt < :EndTime LIMIT :Limit))" + query = "DELETE from Posts WHERE Id = any (array (SELECT Id FROM Posts WHERE CreateAt < ? LIMIT ?))" } else { - query = "DELETE from Posts WHERE CreateAt < :EndTime LIMIT :Limit" + query = "DELETE from Posts WHERE CreateAt < ? LIMIT ?" } - sqlResult, err := s.GetMaster().Exec(query, map[string]interface{}{"EndTime": endTime, "Limit": limit}) + sqlResult, err := s.GetMasterX().Exec(query, endTime, limit) if err != nil { return 0, errors.Wrap(err, "failed to delete Posts") } @@ -2145,7 +2291,7 @@ func (s *SqlPostStore) PermanentDeleteBatch(endTime int64, limit int64) (int64, func (s *SqlPostStore) GetOldest() (*model.Post, error) { var post model.Post - err := s.GetReplica().SelectOne(&post, "SELECT * FROM Posts ORDER BY CreateAt LIMIT 1") + err := s.GetReplicaX().Get(&post, "SELECT * FROM Posts ORDER BY CreateAt LIMIT 1") if err != nil { if err == sql.ErrNoRows { return nil, store.NewErrNotFound("Post", "none") @@ -2163,7 +2309,7 @@ func (s *SqlPostStore) determineMaxPostSize() int { if s.DriverName() == model.DatabaseDriverPostgres { // The Post.Message column in Postgres has historically been VARCHAR(4000), but // may be manually enlarged to support longer posts. - if err := s.GetReplica().SelectOne(&maxPostSizeBytes, ` + if err := s.GetReplicaX().Get(&maxPostSizeBytes, ` SELECT COALESCE(character_maximum_length, 0) FROM @@ -2177,7 +2323,7 @@ func (s *SqlPostStore) determineMaxPostSize() int { } else if s.DriverName() == model.DatabaseDriverMysql { // The Post.Message column in MySQL has historically been TEXT, with a maximum // limit of 65535. - if err := s.GetReplica().SelectOne(&maxPostSizeBytes, ` + if err := s.GetReplicaX().Get(&maxPostSizeBytes, ` SELECT COALESCE(CHARACTER_MAXIMUM_LENGTH, 0) FROM @@ -2219,49 +2365,46 @@ func (s *SqlPostStore) GetMaxPostSize() int { func (s *SqlPostStore) GetParentsForExportAfter(limit int, afterId string) ([]*model.PostForExport, error) { for { - var rootIds []string - _, err := s.GetReplica().Select(&rootIds, + rootIds := []string{} + err := s.GetReplicaX().Select(&rootIds, `SELECT Id FROM Posts WHERE - Id > :AfterId + Id > ? AND RootId = '' AND DeleteAt = 0 ORDER BY Id - LIMIT :Limit`, - map[string]interface{}{"Limit": limit, "AfterId": afterId}) + LIMIT ?`, + afterId, limit) if err != nil { return nil, errors.Wrap(err, "failed to find Posts") } - var postsForExport []*model.PostForExport + postsForExport := []*model.PostForExport{} if len(rootIds) == 0 { return postsForExport, nil } - keys, params := MapStringsToQueryParams(rootIds, "PostId") - _, err = s.GetSearchReplica().Select(&postsForExport, ` - SELECT - p1.*, - Users.Username as Username, - Teams.Name as TeamName, - Channels.Name as ChannelName - FROM - (Select * FROM Posts WHERE Id IN `+keys+`) p1 - INNER JOIN - Channels ON p1.ChannelId = Channels.Id - INNER JOIN - Teams ON Channels.TeamId = Teams.Id - INNER JOIN - Users ON p1.UserId = Users.Id - WHERE - Channels.DeleteAt = 0 - AND Teams.DeleteAt = 0 - ORDER BY - p1.Id`, - params) + builder := s.getQueryBuilder(). + Select("p1.*, Users.Username as Username, Teams.Name as TeamName, Channels.Name as ChannelName"). + FromSelect(sq.Select("*").From("Posts").Where(sq.Eq{"Id": rootIds}), "p1"). + InnerJoin("Channels ON p1.ChannelId = Channels.Id"). + InnerJoin("Teams ON Channels.TeamId = Teams.Id"). + InnerJoin("Users ON p1.UserId = Users.Id"). + Where(sq.And{ + sq.Eq{"Channels.DeleteAt": 0}, + sq.Eq{"Teams.DeleteAt": 0}, + }). + OrderBy("p1.Id") + + query, args, err := builder.ToSql() + if err != nil { + return nil, errors.Wrap(err, "postsForExport_toSql") + } + + err = s.GetSearchReplicaX().Select(&postsForExport, query, args...) if err != nil { return nil, errors.Wrap(err, "failed to find Posts") } @@ -2278,8 +2421,8 @@ func (s *SqlPostStore) GetParentsForExportAfter(limit int, afterId string) ([]*m } func (s *SqlPostStore) GetRepliesForExport(rootId string) ([]*model.ReplyForExport, error) { - var posts []*model.ReplyForExport - _, err := s.GetSearchReplica().Select(&posts, ` + posts := []*model.ReplyForExport{} + err := s.GetSearchReplicaX().Select(&posts, ` SELECT Posts.*, Users.Username as Username @@ -2288,12 +2431,10 @@ func (s *SqlPostStore) GetRepliesForExport(rootId string) ([]*model.ReplyForExpo INNER JOIN Users ON Posts.UserId = Users.Id WHERE - Posts.RootId = :RootId + Posts.RootId = ? AND Posts.DeleteAt = 0 ORDER BY - Posts.Id`, - map[string]interface{}{"RootId": rootId}) - + Posts.Id`, rootId) if err != nil { return nil, errors.Wrap(err, "failed to find Posts") } @@ -2309,10 +2450,10 @@ func (s *SqlPostStore) GetDirectPostParentsForExportAfter(limit int, afterId str Join("Users ON p.UserId = Users.Id"). Where(sq.And{ sq.Gt{"p.Id": afterId}, - sq.Eq{"p.RootId": string("")}, - sq.Eq{"p.DeleteAt": int(0)}, - sq.Eq{"Channels.DeleteAt": int(0)}, - sq.Eq{"Users.DeleteAt": int(0)}, + sq.Eq{"p.RootId": ""}, + sq.Eq{"p.DeleteAt": 0}, + sq.Eq{"Channels.DeleteAt": 0}, + sq.Eq{"Users.DeleteAt": 0}, sq.Eq{"Channels.Type": []string{"D", "G"}}, }). OrderBy("p.Id"). @@ -2323,9 +2464,9 @@ func (s *SqlPostStore) GetDirectPostParentsForExportAfter(limit int, afterId str return nil, errors.Wrap(err, "post_tosql") } - var posts []*model.DirectPostForExport - if _, err = s.GetReplica().Select(&posts, queryString, args...); err != nil { - return nil, errors.Wrap(err, "failed to find Posts") + posts := []*model.DirectPostForExport{} + if err2 := s.GetReplicaX().Select(&posts, queryString, args...); err2 != nil { + return nil, errors.Wrap(err2, "failed to find Posts") } var channelIds []string for _, post := range posts { @@ -2344,8 +2485,8 @@ func (s *SqlPostStore) GetDirectPostParentsForExportAfter(limit int, afterId str return nil, errors.Wrap(err, "post_tosql") } - var channelMembers []*model.ChannelMemberForExport - if _, err := s.GetReplica().Select(&channelMembers, queryString, args...); err != nil { + channelMembers := []*model.ChannelMemberForExport{} + if err := s.GetReplicaX().Select(&channelMembers, queryString, args...); err != nil { return nil, errors.Wrap(err, "failed to find ChannelMembers") } @@ -2427,13 +2568,14 @@ func (s *SqlPostStore) GetOldestEntityCreationTime() (int64, error) { UNION (SELECT MIN(createat) min_createat FROM Channels) ) entities`) - queryString, _, err := query.ToSql() + queryString, args, err := query.ToSql() if err != nil { return -1, errors.Wrap(err, "post_tosql") } - row := s.GetReplica().Db.QueryRow(queryString) + var oldest int64 - if err := row.Scan(&oldest); err != nil { + err = s.GetReplicaX().Get(&oldest, queryString, args...) + if err != nil { return -1, errors.Wrap(err, "unable to scan oldest entity creation time") } return oldest, nil @@ -2512,7 +2654,7 @@ func (s *SqlPostStore) cleanupThreadComments(transaction *sqlxTxWrapper, postId, return nil } -func (s *SqlPostStore) updateThreadsFromPosts(transaction *gorp.Transaction, posts []*model.Post) error { +func (s *SqlPostStore) updateThreadsFromPosts(transaction *sqlxTxWrapper, posts []*model.Post) error { postsByRoot := map[string][]*model.Post{} var rootIds []string for _, post := range posts { @@ -2526,9 +2668,13 @@ func (s *SqlPostStore) updateThreadsFromPosts(transaction *gorp.Transaction, pos if len(rootIds) == 0 { return nil } - 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 { + threadsByRootsSql, threadsByRootsArgs, _ := s.getQueryBuilder(). + Select("*"). + From("Threads"). + Where(sq.Eq{"PostId": rootIds}). + ToSql() + threadsByRoots := []*model.Thread{} + if err := transaction.Select(&threadsByRoots, threadsByRootsSql, threadsByRootsArgs...); err != nil { return err } @@ -2539,13 +2685,13 @@ func (s *SqlPostStore) updateThreadsFromPosts(transaction *gorp.Transaction, pos for rootId, posts := range postsByRoot { if thread, found := threadByRoot[rootId]; !found { - var data []struct { - UserId string `db:"userid"` - RepliedAt int64 `db:"mc"` - } + data := []struct { + UserId string + RepliedAt int64 + }{} // calculate participants - if _, err := transaction.Select(&data, "SELECT UserId, MAX(CreateAt) as mc FROM Posts WHERE RootId=:RootId AND DeleteAt=0 GROUP BY UserId ORDER BY mc ASC", map[string]interface{}{"RootId": rootId}); err != nil { + if err := transaction.Select(&data, "SELECT UserId, MAX(CreateAt) as RepliedAt FROM Posts WHERE RootId=? AND DeleteAt=0 GROUP BY UserId ORDER BY RepliedAt ASC", rootId); err != nil { return err } @@ -2555,17 +2701,22 @@ func (s *SqlPostStore) updateThreadsFromPosts(transaction *gorp.Transaction, pos } // calculate reply count - count, err := transaction.SelectInt("SELECT COUNT(Id) FROM Posts WHERE RootId=:RootId And DeleteAt=0", map[string]interface{}{"RootId": rootId}) + var count int64 + err := transaction.Get(&count, "SELECT COUNT(Id) FROM Posts WHERE RootId=? And DeleteAt=0", rootId) if err != nil { return err } // calculate last reply at - lastReplyAt, err := transaction.SelectInt("SELECT COALESCE(MAX(Posts.CreateAt), 0) FROM Posts WHERE RootID=:RootId and DeleteAt=0", map[string]interface{}{"RootId": rootId}) + var lastReplyAt int64 + err = transaction.Get(&lastReplyAt, "SELECT COALESCE(MAX(Posts.CreateAt), 0) FROM Posts WHERE RootID=? and DeleteAt=0", rootId) if err != nil { return err } // no metadata entry, create one - if err := transaction.Insert(&model.Thread{ + if _, err := transaction.NamedExec(`INSERT INTO Threads + (PostId, ChannelId, ReplyCount, LastReplyAt, Participants) + VALUES + (:PostId, :ChannelId, :ReplyCount, :LastReplyAt, :Participants)`, &model.Thread{ PostId: rootId, ChannelId: posts[0].ChannelId, ReplyCount: count, @@ -2586,7 +2737,12 @@ func (s *SqlPostStore) updateThreadsFromPosts(transaction *gorp.Transaction, pos thread.LastReplyAt = post.CreateAt } } - if _, err := transaction.Update(thread); err != nil { + if _, err := transaction.NamedExec(`UPDATE Threads + SET ChannelId = :ChannelId, + ReplyCount = :ReplyCount, + LastReplyAt = :LastReplyAt, + Participants = :Participants + WHERE PostId=:PostId`, thread); err != nil { return err } } @@ -2607,8 +2763,8 @@ func (s *SqlPostStore) GetUniquePostTypesSince(channelId string, timestamp int64 if err != nil { return nil, err } - var types []string - if _, err := s.GetReplica().Select(&types, query, args...); err != nil { + types := []string{} + if err := s.GetReplicaX().Select(&types, query, args...); err != nil { return nil, err } return types, nil diff --git a/store/storetest/post_store.go b/store/storetest/post_store.go index cd8feab799..596606dd25 100644 --- a/store/storetest/post_store.go +++ b/store/storetest/post_store.go @@ -1818,7 +1818,7 @@ func testPostStoreGetPostBeforeAfter(t *testing.T, ss store.Store) { require.NoError(t, err) rPost3, err := ss.Post().GetPostAfterTime(channelId, o2a.CreateAt, false) - require.Empty(t, rPost3, "should return no post") + require.Empty(t, rPost3.Id, "should return no post") require.NoError(t, err) }