[MM-39629] - Migrate from gorp to sqlx in store/sqlstore/thread_store.go (#19094)

Automatic Merge

store/sqlstore/sqlx_wrapper.go

@@ -18,6 +18,19 @@ import (
 	"github.com/mattermost/mattermost-server/v6/shared/mlog"
 )
 
+// sqlxExecutor exposes sqlx operations. It is used to enable some internal store methods to
+// accept both transactions (*sqlxTxWrapper) and common db handlers (*sqlxDbWrapper).
+type sqlxExecutor interface {
+	Get(dest interface{}, query string, args ...interface{}) error
+	NamedExec(query string, arg interface{}) (sql.Result, error)
+	Exec(query string, args ...interface{}) (sql.Result, error)
+	ExecRaw(query string, args ...interface{}) (sql.Result, error)
+	NamedQuery(query string, arg interface{}) (*sqlx.Rows, error)
+	QueryRowX(query string, args ...interface{}) *sqlx.Row
+	QueryX(query string, args ...interface{}) (*sqlx.Rows, error)
+	Select(dest interface{}, query string, args ...interface{}) error
+}
+
 // namedParamRegex is used to capture all named parameters and convert them
 // to lowercase. This is necessary to be able to use a single query for both
 // Postgres and MySQL.
@@ -90,6 +103,12 @@ func (w *sqlxDBWrapper) NamedExec(query string, arg interface{}) (sql.Result, er
 func (w *sqlxDBWrapper) Exec(query string, args ...interface{}) (sql.Result, error) {
 	query = w.DB.Rebind(query)
 
+	return w.ExecRaw(query, args...)
+}
+
+// ExecRaw is like Exec but without any rebinding of params. You need to pass
+// the exact param types of your target database.
+func (w *sqlxDBWrapper) ExecRaw(query string, args ...interface{}) (sql.Result, error) {
 	ctx, cancel := context.WithTimeout(context.Background(), w.queryTimeout)
 	defer cancel()
 
@@ -191,6 +210,12 @@ func (w *sqlxTxWrapper) Get(dest interface{}, query string, args ...interface{})
 func (w *sqlxTxWrapper) Exec(query string, args ...interface{}) (sql.Result, error) {
 	query = w.Tx.Rebind(query)
 
+	return w.ExecRaw(query, args...)
+}
+
+// ExecRaw is like Exec but without any rebinding of params. You need to pass
+// the exact param types of your target database.
+func (w *sqlxTxWrapper) ExecRaw(query string, args ...interface{}) (sql.Result, error) {
 	ctx, cancel := context.WithTimeout(context.Background(), w.queryTimeout)
 	defer cancel()
 

store/sqlstore/thread_store.go

@@ -6,11 +6,11 @@ package sqlstore
 import (
 	"context"
 	"database/sql"
+	"encoding/json"
 	"strconv"
 	"time"
 
 	sq "github.com/Masterminds/squirrel"
-	"github.com/mattermost/gorp"
 	"github.com/pkg/errors"
 
 	"github.com/mattermost/mattermost-server/v6/model"
@@ -58,7 +58,9 @@ func threadToSlice(thread *model.Thread) []interface{} {
 }
 
 func (s *SqlThreadStore) SaveMultiple(threads []*model.Thread) ([]*model.Thread, int, error) {
-	builder := s.getQueryBuilder().Insert("Threads").Columns(threadSliceColumns()...)
+	builder := s.getQueryBuilder().
+		Insert("Threads").
+		Columns(threadSliceColumns()...)
 	for _, thread := range threads {
 		builder = builder.Values(threadToSlice(thread)...)
 	}
@@ -67,7 +69,7 @@ func (s *SqlThreadStore) SaveMultiple(threads []*model.Thread) ([]*model.Thread,
 		return nil, -1, errors.Wrap(err, "thread_tosql")
 	}
 
-	if _, err := s.GetMaster().Exec(query, args...); err != nil {
+	if _, err := s.GetMasterX().Exec(query, args...); err != nil {
 		return nil, -1, errors.Wrap(err, "failed to save Post")
 	}
 
@@ -83,11 +85,22 @@ func (s *SqlThreadStore) Save(thread *model.Thread) (*model.Thread, error) {
 }
 
 func (s *SqlThreadStore) Update(thread *model.Thread) (*model.Thread, error) {
-	return s.update(s.GetMaster(), thread)
+	jsonParticipants, err := json.Marshal(thread.Participants)
-}
+	if err != nil {
-
+		return nil, errors.Wrap(err, "failed marshaling thread participants")
-func (s *SqlThreadStore) update(ex gorp.SqlExecutor, thread *model.Thread) (*model.Thread, error) {
+	}
-	if _, err := ex.Update(thread); err != nil {
+	query, args, err := s.getQueryBuilder().
+		Update("Threads").
+		Set("ChannelId", thread.ChannelId).
+		Set("ReplyCount", thread.ReplyCount).
+		Set("LastReplyAt", thread.LastReplyAt).
+		Set("Participants", string(jsonParticipants)).
+		Where(sq.Eq{"PostId": thread.PostId}).
+		ToSql()
+	if err != nil {
+		return nil, errors.Wrap(err, "thread_tosql")
+	}
+	if _, err := s.GetMasterX().Exec(query, args...); err != nil {
 		return nil, errors.Wrapf(err, "failed to update thread with id=%s", thread.PostId)
 	}
 
@@ -95,13 +108,16 @@ func (s *SqlThreadStore) update(ex gorp.SqlExecutor, thread *model.Thread) (*mod
 }
 
 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()
+	query, args, err := s.getQueryBuilder().
-	err := ex.SelectOne(&thread, query, args...)
+		Select("*").
+		From("Threads").
+		Where(sq.Eq{"PostId": id}).
+		ToSql()
+	if err != nil {
+		return nil, errors.Wrap(err, "thread_tosql")
+	}
+	err = s.GetMasterX().Get(&thread, query, args...)
 	if err != nil {
 		if err == sql.ErrNoRows {
 			return nil, nil
@@ -168,7 +184,8 @@ func (s *SqlThreadStore) GetThreadsForUser(userId, teamId string, opts model.Get
 			Where(fetchConditions).
 			Where("Posts.CreateAt > ThreadMemberships.LastViewed").ToSql()
 
-		totalUnreadThreads, err := s.GetMaster().SelectInt(repliesQuery, repliesQueryArgs...)
+		var totalUnreadThreads int64
+		err := s.GetMasterX().Get(&totalUnreadThreads, repliesQuery, repliesQueryArgs...)
 		totalUnreadThreadsChan <- store.StoreResult{Data: totalUnreadThreads, NErr: errors.Wrapf(err, "failed to get count unread on threads for user id=%s", userId)}
 		close(totalUnreadThreadsChan)
 	}()
@@ -187,7 +204,8 @@ func (s *SqlThreadStore) GetThreadsForUser(userId, teamId string, opts model.Get
 			From("ThreadMemberships").
 			Where(newFetchConditions).ToSql()
 
-		totalCount, err := s.GetMaster().SelectInt(threadsQuery, threadsQueryArgs...)
+		var totalCount int64
+		err := s.GetMasterX().Get(&totalCount, threadsQuery, threadsQueryArgs...)
 		totalCountChan <- store.StoreResult{Data: totalCount, NErr: err}
 		close(totalCountChan)
 	}()
@@ -199,7 +217,9 @@ func (s *SqlThreadStore) GetThreadsForUser(userId, teamId string, opts model.Get
 			LeftJoin("Posts ON Posts.Id = ThreadMemberships.PostId").
 			LeftJoin("Channels ON Threads.ChannelId = Channels.Id").
 			Where(fetchConditions).ToSql()
-		totalUnreadMentions, err := s.GetMaster().SelectInt(mentionsQuery, mentionsQueryArgs...)
+
+		var totalUnreadMentions int64
+		err := s.GetMasterX().Get(&totalUnreadMentions, mentionsQuery, mentionsQueryArgs...)
 		totalUnreadMentionsChan <- store.StoreResult{Data: totalUnreadMentions, NErr: err}
 		close(totalUnreadMentionsChan)
 	}()
@@ -245,7 +265,7 @@ func (s *SqlThreadStore) GetThreadsForUser(userId, teamId string, opts model.Get
 				Where(unreadRepliesFetchConditions).
 				MustSql()
 
-			var threads []*JoinedThread
+			threads := []*JoinedThread{}
 			query, args, _ := s.getQueryBuilder().
 				Select(`Threads.*,
 				` + postSliceCoalesceQuery() + `,
@@ -260,7 +280,7 @@ func (s *SqlThreadStore) GetThreadsForUser(userId, teamId string, opts model.Get
 				OrderBy("Threads.LastReplyAt " + order).
 				Limit(pageSize).ToSql()
 
-			_, err := s.GetReplica().Select(&threads, query, args...)
+			err := s.GetReplicaX().Select(&threads, query, args...)
 			threadsChan <- store.StoreResult{Data: threads, NErr: err}
 			close(threadsChan)
 		}()
@@ -285,7 +305,7 @@ func (s *SqlThreadStore) GetThreadsForUser(userId, teamId string, opts model.Get
 	totalUnreadThreads := totalUnreadThreadsResult.Data.(int64)
 
 	// userIds is the de-duped list of participant ids from all threads.
-	var userIds []string
+	userIds := []string{}
 	// userIdMap is the map of participant ids from all threads.
 	// Used to generate userIds
 	userIdMap := map[string]bool{}
@@ -356,7 +376,7 @@ func (s *SqlThreadStore) GetThreadsForUser(userId, teamId string, opts model.Get
 }
 
 func (s *SqlThreadStore) GetThreadFollowers(threadID string, fetchOnlyActive bool) ([]string, error) {
-	var users []string
+	users := []string{}
 
 	fetchConditions := sq.And{
 		sq.Eq{"PostId": threadID},
@@ -372,8 +392,9 @@ func (s *SqlThreadStore) GetThreadFollowers(threadID string, fetchOnlyActive boo
 	query, args, _ := s.getQueryBuilder().
 		Select("ThreadMemberships.UserId").
 		From("ThreadMemberships").
-		Where(fetchConditions).ToSql()
+		Where(fetchConditions).
-	_, err := s.GetReplica().Select(&users, query, args...)
+		ToSql()
+	err := s.GetReplicaX().Select(&users, query, args...)
 
 	if err != nil {
 		return nil, err
@@ -423,7 +444,7 @@ func (s *SqlThreadStore) GetThreadForUser(teamId string, threadMembership *model
 
 	args := append(unreadRepliesArgs, threadArgs...)
 
-	err := s.GetReplica().SelectOne(&thread, query, args...)
+	err := s.GetReplicaX().Get(&thread, query, args...)
 	if err != nil {
 		if err == sql.ErrNoRows {
 			return nil, store.NewErrNotFound("Thread", threadMembership.PostId)
@@ -434,7 +455,7 @@ func (s *SqlThreadStore) GetThreadForUser(teamId string, threadMembership *model
 	thread.LastViewedAt = threadMembership.LastViewed
 	thread.UnreadMentions = threadMembership.UnreadMentions
 
-	var users []*model.User
+	users := []*model.User{}
 	if extended {
 		var err error
 		users, err = s.User().GetProfileByIds(context.Background(), thread.Participants, &store.UserGetByIdsOpts{}, true)
@@ -447,7 +468,7 @@ func (s *SqlThreadStore) GetThreadForUser(teamId string, threadMembership *model
 		}
 	}
 
-	var participants []*model.User
+	participants := []*model.User{}
 	for _, participantId := range thread.Participants {
 		var participant *model.User
 		for _, u := range users {
@@ -475,7 +496,7 @@ func (s *SqlThreadStore) GetThreadForUser(teamId string, threadMembership *model
 	return result, nil
 }
 func (s *SqlThreadStore) MarkAllAsReadInChannels(userID string, channelIDs []string) error {
-	var threadIDs []string
+	threadIDs := []string{}
 
 	query, args, _ := s.getQueryBuilder().
 		Select("ThreadMemberships.PostId").
@@ -486,7 +507,7 @@ func (s *SqlThreadStore) MarkAllAsReadInChannels(userID string, channelIDs []str
 		Where(sq.Eq{"ThreadMemberships.UserId": userID}).
 		ToSql()
 
-	_, err := s.GetReplica().Select(&threadIDs, query, args...)
+	err := s.GetReplicaX().Select(&threadIDs, query, args...)
 	if err != nil {
 		return errors.Wrapf(err, "failed to get thread membership with userid=%s", userID)
 	}
@@ -499,7 +520,7 @@ func (s *SqlThreadStore) MarkAllAsReadInChannels(userID string, channelIDs []str
 		Set("LastViewed", timestamp).
 		Set("UnreadMentions", 0).
 		ToSql()
-	if _, err := s.GetMaster().Exec(query, args...); err != nil {
+	if _, err := s.GetMasterX().Exec(query, args...); err != nil {
 		return errors.Wrapf(err, "failed to update thread read state for user id=%s", userID)
 	}
 	return nil
@@ -510,7 +531,7 @@ func (s *SqlThreadStore) MarkAllAsRead(userId, teamId string) error {
 	if err != nil {
 		return err
 	}
-	var membershipIds []string
+	membershipIds := []string{}
 	for _, m := range memberships {
 		membershipIds = append(membershipIds, m.PostId)
 	}
@@ -522,7 +543,7 @@ func (s *SqlThreadStore) MarkAllAsRead(userId, teamId string) error {
 		Set("LastViewed", timestamp).
 		Set("UnreadMentions", 0).
 		ToSql()
-	if _, err := s.GetMaster().Exec(query, args...); err != nil {
+	if _, err := s.GetMasterX().Exec(query, args...); err != nil {
 		return errors.Wrapf(err, "failed to update thread read state for user id=%s", userId)
 	}
 	return nil
@@ -535,7 +556,7 @@ func (s *SqlThreadStore) MarkAsRead(userId, threadId string, timestamp int64) er
 		Where(sq.Eq{"PostId": threadId}).
 		Set("LastViewed", timestamp).
 		ToSql()
-	if _, err := s.GetMaster().Exec(query, args...); err != nil {
+	if _, err := s.GetMasterX().Exec(query, args...); err != nil {
 		return errors.Wrapf(err, "failed to update thread read state for user id=%s thread_id=%v", userId, threadId)
 	}
 	return nil
@@ -543,7 +564,7 @@ func (s *SqlThreadStore) MarkAsRead(userId, threadId string, timestamp int64) er
 
 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 {
+	if _, err := s.GetMasterX().Exec(query, args...); err != nil {
 		return errors.Wrap(err, "failed to update threads")
 	}
 
@@ -551,11 +572,19 @@ func (s *SqlThreadStore) Delete(threadId string) error {
 }
 
 func (s *SqlThreadStore) SaveMembership(membership *model.ThreadMembership) (*model.ThreadMembership, error) {
-	return s.saveMembership(s.GetMaster(), membership)
+	return s.saveMembership(s.GetMasterX(), membership)
 }
 
-func (s *SqlThreadStore) saveMembership(ex gorp.SqlExecutor, membership *model.ThreadMembership) (*model.ThreadMembership, error) {
+func (s *SqlThreadStore) saveMembership(ex sqlxExecutor, membership *model.ThreadMembership) (*model.ThreadMembership, error) {
-	if err := ex.Insert(membership); err != nil {
+	query, args, err := s.getQueryBuilder().
+		Insert("ThreadMemberships").
+		Columns("PostId", "UserId", "Following", "LastViewed", "LastUpdated", "UnreadMentions").
+		Values(membership.PostId, membership.UserId, membership.Following, membership.LastViewed, membership.LastUpdated, membership.UnreadMentions).
+		ToSql()
+	if err != nil {
+		return nil, errors.Wrap(err, "threadmembership_tosql")
+	}
+	if _, err := ex.Exec(query, args...); err != nil {
 		return nil, errors.Wrapf(err, "failed to save thread membership with postid=%s userid=%s", membership.PostId, membership.UserId)
 	}
 
@@ -563,11 +592,25 @@ func (s *SqlThreadStore) saveMembership(ex gorp.SqlExecutor, membership *model.T
 }
 
 func (s *SqlThreadStore) UpdateMembership(membership *model.ThreadMembership) (*model.ThreadMembership, error) {
-	return s.updateMembership(s.GetMaster(), membership)
+	return s.updateMembership(s.GetMasterX(), membership)
 }
 
-func (s *SqlThreadStore) updateMembership(ex gorp.SqlExecutor, membership *model.ThreadMembership) (*model.ThreadMembership, error) {
+func (s *SqlThreadStore) updateMembership(ex sqlxExecutor, membership *model.ThreadMembership) (*model.ThreadMembership, error) {
-	if _, err := ex.Update(membership); err != nil {
+	query, args, err := s.getQueryBuilder().
+		Update("ThreadMemberships").
+		Set("Following", membership.Following).
+		Set("LastViewed", membership.LastViewed).
+		Set("LastUpdated", membership.LastUpdated).
+		Set("UnreadMentions", membership.UnreadMentions).
+		Where(sq.And{
+			sq.Eq{"PostId": membership.PostId},
+			sq.Eq{"UserId": membership.UserId},
+		}).
+		ToSql()
+	if err != nil {
+		return nil, errors.Wrap(err, "threadmembership_tosql")
+	}
+	if _, err := ex.Exec(query, args...); err != nil {
 		return nil, errors.Wrapf(err, "failed to update thread membership with postid=%s userid=%s", membership.PostId, membership.UserId)
 	}
 
@@ -575,7 +618,7 @@ func (s *SqlThreadStore) updateMembership(ex gorp.SqlExecutor, membership *model
 }
 
 func (s *SqlThreadStore) GetMembershipsForUser(userId, teamId string) ([]*model.ThreadMembership, error) {
-	var memberships []*model.ThreadMembership
+	memberships := []*model.ThreadMembership{}
 
 	query, args, _ := s.getQueryBuilder().
 		Select("ThreadMemberships.*").
@@ -586,7 +629,7 @@ func (s *SqlThreadStore) GetMembershipsForUser(userId, teamId string) ([]*model.
 		Where(sq.Eq{"ThreadMemberships.UserId": userId}).
 		ToSql()
 
-	_, err := s.GetReplica().Select(&memberships, query, args...)
+	err := s.GetReplicaX().Select(&memberships, query, args...)
 	if err != nil {
 		return nil, errors.Wrapf(err, "failed to get thread membership with userid=%s", userId)
 	}
@@ -594,12 +637,23 @@ 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)
+	return s.getMembershipForUser(s.GetReplicaX(), userId, postId)
 }
 
-func (s *SqlThreadStore) getMembershipForUser(ex gorp.SqlExecutor, userId, postId string) (*model.ThreadMembership, error) {
+func (s *SqlThreadStore) getMembershipForUser(ex sqlxExecutor, userId, postId string) (*model.ThreadMembership, error) {
 	var membership model.ThreadMembership
-	err := ex.SelectOne(&membership, "SELECT * from ThreadMemberships WHERE UserId = :UserId AND PostId = :PostId", map[string]interface{}{"UserId": userId, "PostId": postId})
+	query, args, err := s.getQueryBuilder().
+		Select("*").
+		From("ThreadMemberships").
+		Where(sq.And{
+			sq.Eq{"PostId": postId},
+			sq.Eq{"UserId": userId},
+		}).
+		ToSql()
+	if err != nil {
+		return nil, errors.Wrap(err, "threadmembership_tosql")
+	}
+	err = ex.Get(&membership, query, args...)
 	if err != nil {
 		if err == sql.ErrNoRows {
 			return nil, store.NewErrNotFound("Thread", postId)
@@ -610,8 +664,18 @@ func (s *SqlThreadStore) getMembershipForUser(ex gorp.SqlExecutor, userId, postI
 }
 
 func (s *SqlThreadStore) DeleteMembershipForUser(userId string, postId string) error {
-	if _, err := s.GetMaster().Exec("DELETE FROM ThreadMemberships Where PostId = :PostId AND UserId = :UserId", map[string]interface{}{"PostId": postId, "UserId": userId}); err != nil {
+	query, args, err := s.getQueryBuilder().
-		return errors.Wrap(err, "failed to update thread membership")
+		Delete("ThreadMemberships").
+		Where(sq.And{
+			sq.Eq{"PostId": postId},
+			sq.Eq{"UserId": userId},
+		}).
+		ToSql()
+	if err != nil {
+		return errors.Wrap(err, "threadmembership_tosql")
+	}
+	if _, err := s.GetMasterX().Exec(query, args...); err != nil {
+		return errors.Wrap(err, "failed to delete thread membership")
 	}
 
 	return nil
@@ -624,11 +688,11 @@ func (s *SqlThreadStore) DeleteMembershipForUser(userId string, postId string) e
 // - 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()
+	trx, err := s.GetMasterX().Beginx()
 	if err != nil {
 		return nil, errors.Wrap(err, "begin_transaction")
 	}
-	defer finalizeTransaction(trx)
+	defer finalizeTransactionX(trx)
 
 	membership, err := s.getMembershipForUser(trx, userId, postId)
 	now := utils.MillisFromTime(time.Now())
@@ -685,10 +749,10 @@ func (s *SqlThreadStore) MaintainMembership(userId, postId string, opts store.Th
 
 	if opts.UpdateParticipants {
 		if s.DriverName() == model.DatabaseDriverPostgres {
-			if _, err2 := trx.Exec(`UPDATE Threads
+			if _, err2 := trx.ExecRaw(`UPDATE Threads
-				SET participants = participants || $1::jsonb
+                        SET participants = participants || $1::jsonb
-				WHERE postid=$2
+                        WHERE postid=$2
-				AND NOT participants ? $3`, jsonArray([]string{userId}), postId, userId); err2 != nil {
+                        AND NOT participants ? $3`, jsonArray([]string{userId}), postId, userId); err2 != nil {
 				return nil, err2
 			}
 		} else {
@@ -711,7 +775,7 @@ func (s *SqlThreadStore) MaintainMembership(userId, postId string, opts store.Th
 }
 
 func (s *SqlThreadStore) CollectThreadsWithNewerReplies(userId string, channelIds []string, timestamp int64) ([]string, error) {
-	var changedThreads []string
+	changedThreads := []string{}
 	query, args, _ := s.getQueryBuilder().
 		Select("Threads.PostId").
 		From("Threads").
@@ -725,7 +789,7 @@ func (s *SqlThreadStore) CollectThreadsWithNewerReplies(userId string, channelId
 			},
 		}).
 		ToSql()
-	if _, err := s.GetReplica().Select(&changedThreads, query, args...); err != nil {
+	if err := s.GetReplicaX().Select(&changedThreads, query, args...); err != nil {
 		return nil, errors.Wrap(err, "failed to fetch threads")
 	}
 	return changedThreads, nil
@@ -746,7 +810,7 @@ func (s *SqlThreadStore) UpdateUnreadsByChannel(userId string, changedThreads []
 	}
 	updateQuery, updateArgs, _ := qb.ToSql()
 
-	if _, err := s.GetMaster().Exec(updateQuery, updateArgs...); err != nil {
+	if _, err := s.GetMasterX().Exec(updateQuery, updateArgs...); err != nil {
 		return errors.Wrap(err, "failed to update thread membership")
 	}
 
@@ -760,8 +824,8 @@ func (s *SqlThreadStore) GetPosts(threadId string, since int64) ([]*model.Post,
 		Where(sq.Eq{"RootId": threadId}).
 		Where(sq.Eq{"DeleteAt": 0}).
 		Where(sq.GtOrEq{"UpdateAt": since}).ToSql()
-	var result []*model.Post
+	result := []*model.Post{}
-	if _, err := s.GetReplica().Select(&result, query, args...); err != nil {
+	if err := s.GetReplicaX().Select(&result, query, args...); err != nil {
 		return nil, errors.Wrap(err, "failed to fetch thread posts")
 	}
 	return result, nil
@@ -815,7 +879,7 @@ func (s *SqlThreadStore) DeleteOrphanedRows(limit int) (deleted int64, err error
 			SELECT Threads.PostId FROM Threads
 			LEFT JOIN Channels ON Threads.ChannelId = Channels.Id
 			WHERE Channels.Id IS NULL
-			LIMIT :Limit
+			LIMIT ?
 		) AS A
 	)`
 	// We only delete a thread membership if the entire thread no longer exists,
@@ -826,11 +890,10 @@ func (s *SqlThreadStore) DeleteOrphanedRows(limit int) (deleted int64, err error
 			SELECT ThreadMemberships.PostId FROM ThreadMemberships
 			LEFT JOIN Threads ON ThreadMemberships.PostId = Threads.PostId
 			WHERE Threads.PostId IS NULL
-			LIMIT :Limit
+			LIMIT ?
 		) AS A
 	)`
-	props := map[string]interface{}{"Limit": limit}
+	result, err := s.GetMasterX().Exec(threadsQuery, limit)
-	result, err := s.GetMaster().Exec(threadsQuery, props)
 	if err != nil {
 		return
 	}
@@ -838,7 +901,7 @@ func (s *SqlThreadStore) DeleteOrphanedRows(limit int) (deleted int64, err error
 	if err != nil {
 		return
 	}
-	result, err = s.GetMaster().Exec(threadMembershipsQuery, props)
+	result, err = s.GetMasterX().Exec(threadMembershipsQuery, limit)
 	if err != nil {
 		return
 	}

store/storetest/mocks/sqlxExecutor.go

@@ -0,0 +1,195 @@
+// Code generated by mockery v1.0.0. DO NOT EDIT.
+
+// Regenerate this file using `make store-mocks`.
+
+package mocks
+
+import (
+	sql "database/sql"
+
+	mock "github.com/stretchr/testify/mock"
+
+	sqlx "github.com/jmoiron/sqlx"
+)
+
+// sqlxExecutor is an autogenerated mock type for the sqlxExecutor type
+type sqlxExecutor struct {
+	mock.Mock
+}
+
+// Exec provides a mock function with given fields: query, args
+func (_m *sqlxExecutor) Exec(query string, args ...interface{}) (sql.Result, error) {
+	var _ca []interface{}
+	_ca = append(_ca, query)
+	_ca = append(_ca, args...)
+	ret := _m.Called(_ca...)
+
+	var r0 sql.Result
+	if rf, ok := ret.Get(0).(func(string, ...interface{}) sql.Result); ok {
+		r0 = rf(query, args...)
+	} else {
+		if ret.Get(0) != nil {
+			r0 = ret.Get(0).(sql.Result)
+		}
+	}
+
+	var r1 error
+	if rf, ok := ret.Get(1).(func(string, ...interface{}) error); ok {
+		r1 = rf(query, args...)
+	} else {
+		r1 = ret.Error(1)
+	}
+
+	return r0, r1
+}
+
+// ExecRaw provides a mock function with given fields: query, args
+func (_m *sqlxExecutor) ExecRaw(query string, args ...interface{}) (sql.Result, error) {
+	var _ca []interface{}
+	_ca = append(_ca, query)
+	_ca = append(_ca, args...)
+	ret := _m.Called(_ca...)
+
+	var r0 sql.Result
+	if rf, ok := ret.Get(0).(func(string, ...interface{}) sql.Result); ok {
+		r0 = rf(query, args...)
+	} else {
+		if ret.Get(0) != nil {
+			r0 = ret.Get(0).(sql.Result)
+		}
+	}
+
+	var r1 error
+	if rf, ok := ret.Get(1).(func(string, ...interface{}) error); ok {
+		r1 = rf(query, args...)
+	} else {
+		r1 = ret.Error(1)
+	}
+
+	return r0, r1
+}
+
+// Get provides a mock function with given fields: dest, query, args
+func (_m *sqlxExecutor) Get(dest interface{}, query string, args ...interface{}) error {
+	var _ca []interface{}
+	_ca = append(_ca, dest, query)
+	_ca = append(_ca, args...)
+	ret := _m.Called(_ca...)
+
+	var r0 error
+	if rf, ok := ret.Get(0).(func(interface{}, string, ...interface{}) error); ok {
+		r0 = rf(dest, query, args...)
+	} else {
+		r0 = ret.Error(0)
+	}
+
+	return r0
+}
+
+// NamedExec provides a mock function with given fields: query, arg
+func (_m *sqlxExecutor) NamedExec(query string, arg interface{}) (sql.Result, error) {
+	ret := _m.Called(query, arg)
+
+	var r0 sql.Result
+	if rf, ok := ret.Get(0).(func(string, interface{}) sql.Result); ok {
+		r0 = rf(query, arg)
+	} else {
+		if ret.Get(0) != nil {
+			r0 = ret.Get(0).(sql.Result)
+		}
+	}
+
+	var r1 error
+	if rf, ok := ret.Get(1).(func(string, interface{}) error); ok {
+		r1 = rf(query, arg)
+	} else {
+		r1 = ret.Error(1)
+	}
+
+	return r0, r1
+}
+
+// NamedQuery provides a mock function with given fields: query, arg
+func (_m *sqlxExecutor) NamedQuery(query string, arg interface{}) (*sqlx.Rows, error) {
+	ret := _m.Called(query, arg)
+
+	var r0 *sqlx.Rows
+	if rf, ok := ret.Get(0).(func(string, interface{}) *sqlx.Rows); ok {
+		r0 = rf(query, arg)
+	} else {
+		if ret.Get(0) != nil {
+			r0 = ret.Get(0).(*sqlx.Rows)
+		}
+	}
+
+	var r1 error
+	if rf, ok := ret.Get(1).(func(string, interface{}) error); ok {
+		r1 = rf(query, arg)
+	} else {
+		r1 = ret.Error(1)
+	}
+
+	return r0, r1
+}
+
+// QueryRowX provides a mock function with given fields: query, args
+func (_m *sqlxExecutor) QueryRowX(query string, args ...interface{}) *sqlx.Row {
+	var _ca []interface{}
+	_ca = append(_ca, query)
+	_ca = append(_ca, args...)
+	ret := _m.Called(_ca...)
+
+	var r0 *sqlx.Row
+	if rf, ok := ret.Get(0).(func(string, ...interface{}) *sqlx.Row); ok {
+		r0 = rf(query, args...)
+	} else {
+		if ret.Get(0) != nil {
+			r0 = ret.Get(0).(*sqlx.Row)
+		}
+	}
+
+	return r0
+}
+
+// QueryX provides a mock function with given fields: query, args
+func (_m *sqlxExecutor) QueryX(query string, args ...interface{}) (*sqlx.Rows, error) {
+	var _ca []interface{}
+	_ca = append(_ca, query)
+	_ca = append(_ca, args...)
+	ret := _m.Called(_ca...)
+
+	var r0 *sqlx.Rows
+	if rf, ok := ret.Get(0).(func(string, ...interface{}) *sqlx.Rows); ok {
+		r0 = rf(query, args...)
+	} else {
+		if ret.Get(0) != nil {
+			r0 = ret.Get(0).(*sqlx.Rows)
+		}
+	}
+
+	var r1 error
+	if rf, ok := ret.Get(1).(func(string, ...interface{}) error); ok {
+		r1 = rf(query, args...)
+	} else {
+		r1 = ret.Error(1)
+	}
+
+	return r0, r1
+}
+
+// Select provides a mock function with given fields: dest, query, args
+func (_m *sqlxExecutor) Select(dest interface{}, query string, args ...interface{}) error {
+	var _ca []interface{}
+	_ca = append(_ca, dest, query)
+	_ca = append(_ca, args...)
+	ret := _m.Called(_ca...)
+
+	var r0 error
+	if rf, ok := ret.Get(0).(func(interface{}, string, ...interface{}) error); ok {
+		r0 = rf(dest, query, args...)
+	} else {
+		r0 = ret.Error(0)
+	}
+
+	return r0
+}