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

[MM-39636] Migrate from gorp to sqlx in store/sqlstore/retention_policy_store.go (#18994)

Просмотр исходного кода 
* Migrate Save from gorp to sqlx

* Migrate checkTeamsExist from gorp to sqlx

* Migrate checkChannelsExist from gorp to sqlx

* Migrate Patch from gorp to sqlx

* Migrate buildGetPoliciesQuery from gorp to sqlx

* Migrate Get from gorp to sqlx

* Change buildGetPolicyQuery return values

* Migrate GetAll from gorp to sqlx

* Migrate GetCount from gorp to sqlx

* Migrate Delete from gorp to sqlx

* Migrate GetChannels from gorp to sqlx

* Migrate GetChannelsCount from gorp to sqlx

* Migrate AddChannels from gorp to sqlx

* Migrate RemoveChannels from gorp to sqlx

* Migrate GetTeams from gorp to sqlx

* Migrate GetTeamsCount from gorp to sqlx

* Migrate AddTeams from gorp to sqlx

* Migrate RemoveTeams from gorp to sqlx

* Migrate DeleteOrphanedRows from gorp to sqlx

* Migrate GetTeamPoliciesForUser from gorp to sqlx

* Migrate GetTeamPoliciesCountForUser from gorp to sqlx

* Migrate GetChannelPoliciesForUser from gorp to sqlx

* Migrate GetChannelPoliciesCountForUser from gorp to sqlx

* Migrate Delete from gorp to sqlx

* Add mapper tag function before saving a record

Summary:
Since there is a db tag in this model for the ID column,
I set a mapper in this transaction to lowercase
all mapping columns before saving the record.

* Use quoted identifiers when selecting Id

* Address PR comments, replace variable declarations for their short hand syntax

* Enhance subquery implementation

Co-authored-by: Mattermod <mattermod@users.noreply.github.com>
Этот коммит содержится в:
Yair Facio
2021-12-10 03:27:24 -06:00
коммит произвёл GitHub
родитель fa38202d98
Коммит 62d4d8166d
1 изменённых файлов: 387 добавлений и 198 удалений
Скачать Patch файл Скачать Diff файл Показать все файлы Свернуть все файлы

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

@@ -5,13 +5,13 @@ package sqlstore
import ( import (
"database/sql" "database/sql"
"fmt"
"strconv" "strconv"
"strings" "strings"
sq "github.com/Masterminds/squirrel" sq "github.com/Masterminds/squirrel"
"github.com/go-sql-driver/mysql" "github.com/go-sql-driver/mysql"
"github.com/lib/pq" "github.com/lib/pq"
"github.com/mattermost/gorp"
"github.com/mattermost/mattermost-server/v6/einterfaces" "github.com/mattermost/mattermost-server/v6/einterfaces"
"github.com/mattermost/mattermost-server/v6/model" "github.com/mattermost/mattermost-server/v6/model"
"github.com/mattermost/mattermost-server/v6/store" "github.com/mattermost/mattermost-server/v6/store"
@@ -59,7 +59,7 @@ func (s *SqlRetentionPolicyStore) createIndexesIfNotExists() {
// executePossiblyEmptyQuery only executes the query if it is non-empty. This helps avoid // executePossiblyEmptyQuery only executes the query if it is non-empty. This helps avoid
// having to check for MySQL, which, unlike Postgres, does not allow empty queries. // having to check for MySQL, which, unlike Postgres, does not allow empty queries.
func executePossiblyEmptyQuery(txn *gorp.Transaction, query string, args ...interface{}) (sql.Result, error) { func executePossiblyEmptyQuery(txn *sqlxTxWrapper, query string, args ...interface{}) (sql.Result, error) {
if query == "" { if query == "" {
return nil, nil return nil, nil
} }
@@ -100,13 +100,16 @@ func (s *SqlRetentionPolicyStore) Save(policy *model.RetentionPolicyWithTeamAndC
return nil, err return nil, err
} }
policySelectQuery, policySelectProps := s.buildGetPolicyQuery(policy.ID) queryString, args, err := s.buildGetPolicyQuery(policy.ID)
txn, err := s.GetMaster().Begin()
if err != nil { if err != nil {
return nil, err return nil, err
} }
defer finalizeTransaction(txn)
txn, err := s.GetMasterX().Beginx()
if err != nil {
return nil, err
}
defer finalizeTransactionX(txn)
// Create a new policy in RetentionPolicies // Create a new policy in RetentionPolicies
if _, err = txn.Exec(policyInsertQuery, policyInsertArgs...); err != nil { if _, err = txn.Exec(policyInsertQuery, policyInsertArgs...); err != nil {
return nil, err return nil, err
@@ -121,7 +124,8 @@ func (s *SqlRetentionPolicyStore) Save(policy *model.RetentionPolicyWithTeamAndC
} }
// Select the new policy (with team/channel counts) which we just created // Select the new policy (with team/channel counts) which we just created
var newPolicy model.RetentionPolicyWithTeamAndChannelCounts var newPolicy model.RetentionPolicyWithTeamAndChannelCounts
if err = txn.SelectOne(&newPolicy, policySelectQuery, policySelectProps); err != nil {
if err = txn.Get(&newPolicy, queryString, args...); err != nil {
return nil, err return nil, err
} }
if err = txn.Commit(); err != nil { if err = txn.Commit(); err != nil {
@@ -140,8 +144,8 @@ func (s *SqlRetentionPolicyStore) checkTeamsExist(teamIDs []string) error {
if err != nil { if err != nil {
return err return err
} }
var rows []*string rows := []*string{}
_, err = s.GetReplica().Select(&rows, teamsSelectQuery, teamsSelectArgs...) err = s.GetReplicaX().Select(&rows, teamsSelectQuery, teamsSelectArgs...)
if err != nil { if err != nil {
return err return err
} }
@@ -171,8 +175,8 @@ func (s *SqlRetentionPolicyStore) checkChannelsExist(channelIDs []string) error
if err != nil { if err != nil {
return err return err
} }
var rows []*string rows := []*string{}
_, err = s.GetReplica().Select(&rows, channelsSelectQuery, channelsSelectArgs...) err = s.GetReplicaX().Select(&rows, channelsSelectQuery, channelsSelectArgs...)
if err != nil { if err != nil {
return err return err
} }
@@ -291,13 +295,16 @@ func (s *SqlRetentionPolicyStore) Patch(patch *model.RetentionPolicyWithTeamAndC
} }
} }
policySelectQuery, policySelectProps := s.buildGetPolicyQuery(patch.ID) queryString, args, err := s.buildGetPolicyQuery(patch.ID)
txn, err := s.GetMaster().Begin()
if err != nil { if err != nil {
return nil, err return nil, err
} }
defer finalizeTransaction(txn)
txn, err := s.GetMasterX().Beginx()
if err != nil {
return nil, err
}
defer finalizeTransactionX(txn)
// Update the fields of the policy in RetentionPolicies // Update the fields of the policy in RetentionPolicies
if _, err = executePossiblyEmptyQuery(txn, policyUpdateQuery, policyUpdateArgs...); err != nil { if _, err = executePossiblyEmptyQuery(txn, policyUpdateQuery, policyUpdateArgs...); err != nil {
return nil, err return nil, err
@@ -320,7 +327,7 @@ func (s *SqlRetentionPolicyStore) Patch(patch *model.RetentionPolicyWithTeamAndC
} }
// Select the policy which we just updated // Select the policy which we just updated
var newPolicy model.RetentionPolicyWithTeamAndChannelCounts var newPolicy model.RetentionPolicyWithTeamAndChannelCounts
if err = txn.SelectOne(&newPolicy, policySelectQuery, policySelectProps); err != nil { if err = txn.Get(&newPolicy, queryString, args...); err != nil {
return nil, err return nil, err
} }
if err = txn.Commit(); err != nil { if err = txn.Commit(); err != nil {
@@ -329,118 +336,179 @@ func (s *SqlRetentionPolicyStore) Patch(patch *model.RetentionPolicyWithTeamAndC
return &newPolicy, nil return &newPolicy, nil
} }
func (s *SqlRetentionPolicyStore) buildGetPolicyQuery(id string) (query string, props map[string]interface{}) { func (s *SqlRetentionPolicyStore) buildGetPolicyQuery(id string) (string, []interface{}, error) {
return s.buildGetPoliciesQuery(id, 0, 1) return s.buildGetPoliciesQuery(id, 0, 1)
} }
// buildGetPoliciesQuery builds a query to select information for the policy with the specified // buildGetPoliciesQuery builds a query to select information for the policy with the specified
// ID, or, if `id` is the empty string, from all policies. The results returned will be sorted by // ID, or, if `id` is the empty string, from all policies. The results returned will be sorted by
// policy display name and ID. // policy display name and ID.
func (s *SqlRetentionPolicyStore) buildGetPoliciesQuery(id string, offset, limit int) (query string, props map[string]interface{}) { func (s *SqlRetentionPolicyStore) buildGetPoliciesQuery(id string, offset, limit int) (string, []interface{}, error) {
props = map[string]interface{}{"Offset": offset, "Limit": limit} rpcSubQuery := s.getQueryBuilder().
whereIdEqualsPolicyId := "" Select("RetentionPolicies.Id, COUNT(RetentionPoliciesChannels.ChannelId) AS Count").
From("RetentionPolicies").
LeftJoin("RetentionPoliciesChannels ON RetentionPolicies.Id = RetentionPoliciesChannels.PolicyId").
GroupBy("RetentionPolicies.Id").
OrderBy("RetentionPolicies.DisplayName, RetentionPolicies.Id").
Limit(uint64(limit)).
Offset(uint64(offset))
if id != "" { if id != "" {
whereIdEqualsPolicyId = "WHERE RetentionPolicies.Id = :PolicyId" rpcSubQuery = rpcSubQuery.Where(sq.Eq{"RetentionPolicies.Id": id})
props["PolicyId"] = id
} }
query = `
SELECT RetentionPolicies.Id, rpcSubQueryString, args, err := rpcSubQuery.ToSql()
RetentionPolicies.DisplayName, if err != nil {
RetentionPolicies.PostDuration, return "", nil, errors.Wrap(err, "retention_policies_tosql")
A.Count AS ChannelCount, }
B.Count AS TeamCount
FROM RetentionPolicies rptSubQuery := s.getQueryBuilder().
INNER JOIN ( Select("RetentionPolicies.Id, COUNT(RetentionPoliciesTeams.TeamId) AS Count").
SELECT RetentionPolicies.Id, From("RetentionPolicies").
COUNT(RetentionPoliciesChannels.ChannelId) AS Count LeftJoin("RetentionPoliciesTeams ON RetentionPolicies.Id = RetentionPoliciesTeams.PolicyId").
FROM RetentionPolicies GroupBy("RetentionPolicies.Id").
LEFT JOIN RetentionPoliciesChannels ON RetentionPolicies.Id = RetentionPoliciesChannels.PolicyId OrderBy("RetentionPolicies.DisplayName, RetentionPolicies.Id").
` + whereIdEqualsPolicyId + ` Limit(uint64(limit)).
GROUP BY RetentionPolicies.Id Offset(uint64(offset))
ORDER BY RetentionPolicies.DisplayName, RetentionPolicies.Id
LIMIT :Limit if id != "" {
OFFSET :Offset rptSubQuery = rptSubQuery.Where(sq.Eq{"RetentionPolicies.Id": id})
) AS A ON RetentionPolicies.Id = A.Id }
INNER JOIN (
SELECT RetentionPolicies.Id, rptSubQueryString, _, err := rptSubQuery.ToSql()
COUNT(RetentionPoliciesTeams.TeamId) AS Count if err != nil {
FROM RetentionPolicies return "", nil, errors.Wrap(err, "retention_policies_tosql")
LEFT JOIN RetentionPoliciesTeams ON RetentionPolicies.Id = RetentionPoliciesTeams.PolicyId }
` + whereIdEqualsPolicyId + `
GROUP BY RetentionPolicies.Id query := s.getQueryBuilder().
ORDER BY RetentionPolicies.DisplayName, RetentionPolicies.Id Select(`
LIMIT :Limit RetentionPolicies.Id as "Id",
OFFSET :Offset RetentionPolicies.DisplayName,
) AS B ON RetentionPolicies.Id = B.Id RetentionPolicies.PostDuration,
ORDER BY RetentionPolicies.DisplayName, RetentionPolicies.Id` A.Count AS ChannelCount,
return B.Count AS TeamCount
`).
From("RetentionPolicies").
InnerJoin(`(` + rpcSubQueryString + `) AS A ON RetentionPolicies.Id = A.Id`).
InnerJoin(`(` + rptSubQueryString + `) AS B ON RetentionPolicies.Id = B.Id`).
OrderBy("RetentionPolicies.DisplayName, RetentionPolicies.Id")
queryString, _, err := query.ToSql()
if err != nil {
return "", nil, errors.Wrap(err, "retention_policies_tosql")
}
// MySQL does not support positional params, so we add one param for each WHERE clause.
if s.DriverName() == model.DatabaseDriverMysql {
args = append(args, args...)
}
return queryString, args, nil
} }
func (s *SqlRetentionPolicyStore) Get(id string) (*model.RetentionPolicyWithTeamAndChannelCounts, error) { func (s *SqlRetentionPolicyStore) Get(id string) (*model.RetentionPolicyWithTeamAndChannelCounts, error) {
query, props := s.buildGetPolicyQuery(id) queryString, args, err := s.buildGetPolicyQuery(id)
if err != nil {
return nil, err
}
var policy model.RetentionPolicyWithTeamAndChannelCounts var policy model.RetentionPolicyWithTeamAndChannelCounts
if err := s.GetReplica().SelectOne(&policy, query, props); err != nil { if err := s.GetReplicaX().Get(&policy, queryString, args...); err != nil {
return nil, err return nil, err
} }
return &policy, nil return &policy, nil
} }
func (s *SqlRetentionPolicyStore) GetAll(offset, limit int) (policies []*model.RetentionPolicyWithTeamAndChannelCounts, err error) { func (s *SqlRetentionPolicyStore) GetAll(offset, limit int) (policies []*model.RetentionPolicyWithTeamAndChannelCounts, err error) {
query, props := s.buildGetPoliciesQuery("", offset, limit) queryString, args, err := s.buildGetPoliciesQuery("", offset, limit)
_, err = s.GetReplica().Select(&policies, query, props) if err != nil {
return
}
err = s.GetReplicaX().Select(&policies, queryString, args...)
return return
} }
func (s *SqlRetentionPolicyStore) GetCount() (int64, error) { func (s *SqlRetentionPolicyStore) GetCount() (int64, error) {
return s.GetReplica().SelectInt("SELECT COUNT(*) FROM RetentionPolicies") var count int64
err := s.GetReplicaX().Get(&count, "SELECT COUNT(*) FROM RetentionPolicies")
if err != nil {
return count, err
}
return count, nil
} }
func (s *SqlRetentionPolicyStore) Delete(id string) error { func (s *SqlRetentionPolicyStore) Delete(id string) error {
builder := s.getQueryBuilder(). query := s.getQueryBuilder().
Delete("RetentionPolicies"). Delete("RetentionPolicies").
Where(sq.Eq{"Id": id}) Where(sq.Eq{"Id": id})
result, err := builder.RunWith(s.GetMaster()).Exec()
queryString, args, err := query.ToSql()
if err != nil { if err != nil {
return err return errors.Wrap(err, "retention_policies_tosql")
} }
numRowsAffected, err := result.RowsAffected()
sqlResult, err := s.GetMasterX().Exec(queryString, args...)
if err != nil { if err != nil {
return err return errors.Wrapf(err, "failed to permanent delete retention policy with id=%s", id)
}
numRowsAffected, err := sqlResult.RowsAffected()
if err != nil {
return errors.Wrap(err, "unable to get rows affected")
} else if numRowsAffected == 0 { } else if numRowsAffected == 0 {
return errors.New("policy not found") return errors.New("policy not found")
} }
return nil return nil
} }
func (s *SqlRetentionPolicyStore) GetChannels(policyId string, offset, limit int) (channels model.ChannelListWithTeamData, err error) { func (s *SqlRetentionPolicyStore) GetChannels(policyId string, offset, limit int) (model.ChannelListWithTeamData, error) {
const query = ` query := s.getQueryBuilder().Select(`Channels.*, Teams.DisplayName AS TeamDisplayName,
SELECT Channels.*, Teams.Name AS TeamName,Teams.UpdateAt AS TeamUpdateAt`).
Teams.DisplayName AS TeamDisplayName, From("RetentionPoliciesChannels").
Teams.Name AS TeamName, InnerJoin("Channels ON RetentionPoliciesChannels.ChannelId = Channels.Id").
Teams.UpdateAt AS TeamUpdateAt InnerJoin("Teams ON Channels.TeamId = Teams.Id").
FROM RetentionPoliciesChannels Where(sq.Eq{"RetentionPoliciesChannels.PolicyId": policyId}).
INNER JOIN Channels ON RetentionPoliciesChannels.ChannelId = Channels.Id OrderBy("Channels.DisplayName, Channels.Id").
INNER JOIN Teams ON Channels.TeamId = Teams.Id Limit(uint64(limit)).
WHERE RetentionPoliciesChannels.PolicyId = :PolicyId Offset(uint64(offset))
ORDER BY Channels.DisplayName, Channels.Id
LIMIT :Limit queryString, args, err := query.ToSql()
OFFSET :Offset` if err != nil {
props := map[string]interface{}{"PolicyId": policyId, "Limit": limit, "Offset": offset} return nil, errors.Wrap(err, "retention_policies_channels_tosql")
_, err = s.GetReplica().Select(&channels, query, props) }
channels := model.ChannelListWithTeamData{}
if err := s.GetReplicaX().Select(&channels, queryString, args...); err != nil {
return channels, errors.Wrap(err, "failed to find RetentionPoliciesChannels")
}
for _, channel := range channels { for _, channel := range channels {
channel.PolicyID = model.NewString(policyId) channel.PolicyID = model.NewString(policyId)
} }
return
return channels, nil
} }
func (s *SqlRetentionPolicyStore) GetChannelsCount(policyId string) (int64, error) { func (s *SqlRetentionPolicyStore) GetChannelsCount(policyId string) (int64, error) {
const query = ` query := s.getQueryBuilder().
SELECT COUNT(*) Select("Count(*)").
FROM RetentionPolicies From("RetentionPolicies").
INNER JOIN RetentionPoliciesChannels ON RetentionPolicies.Id = RetentionPoliciesChannels.PolicyId InnerJoin("RetentionPoliciesChannels ON RetentionPolicies.Id = RetentionPoliciesChannels.PolicyId").
WHERE RetentionPolicies.Id = :PolicyId` Where(sq.Eq{"RetentionPolicies.Id": policyId})
props := map[string]interface{}{"PolicyId": policyId}
return s.GetReplica().SelectInt(query, props) queryString, args, err := query.ToSql()
if err != nil {
return 0, errors.Wrap(err, "retention_policies_tosql")
}
var count int64
if err := s.GetReplicaX().Get(&count, queryString, args...); err != nil {
return 0, errors.Wrap(err, "failed to count RetentionPolicies")
}
return count, nil
} }
func (s *SqlRetentionPolicyStore) AddChannels(policyId string, channelIds []string) error { func (s *SqlRetentionPolicyStore) AddChannels(policyId string, channelIds []string) error {
@@ -450,13 +518,20 @@ func (s *SqlRetentionPolicyStore) AddChannels(policyId string, channelIds []stri
if err := s.checkChannelsExist(channelIds); err != nil { if err := s.checkChannelsExist(channelIds); err != nil {
return err return err
} }
builder := s.getQueryBuilder(). query := s.getQueryBuilder().
Insert("RetentionPoliciesChannels"). Insert("RetentionPoliciesChannels").
Columns("policyId", "channelId") Columns("policyId", "channelId")
for _, channelId := range channelIds { for _, channelId := range channelIds {
builder = builder.Values(policyId, channelId) query = query.Values(policyId, channelId)
} }
_, err := builder.RunWith(s.GetMaster()).Exec()
queryString, args, err := query.ToSql()
if err != nil {
return errors.Wrap(err, "retention_policies_channels_tosql")
}
_, err = s.GetMasterX().Exec(queryString, args...)
if err != nil { if err != nil {
switch dbErr := err.(type) { switch dbErr := err.(type) {
case *pq.Error: case *pq.Error:
@@ -469,47 +544,74 @@ func (s *SqlRetentionPolicyStore) AddChannels(policyId string, channelIds []stri
} }
} }
} }
return err
return nil
} }
func (s *SqlRetentionPolicyStore) RemoveChannels(policyId string, channelIds []string) error { func (s *SqlRetentionPolicyStore) RemoveChannels(policyId string, channelIds []string) error {
if len(channelIds) == 0 { if len(channelIds) == 0 {
return nil return nil
} }
builder := s.getQueryBuilder(). query := s.getQueryBuilder().
Delete("RetentionPoliciesChannels"). Delete("RetentionPoliciesChannels").
Where(sq.And{ Where(sq.And{
sq.Eq{"PolicyId": policyId}, sq.Eq{"PolicyId": policyId},
sq.Eq{"ChannelId": channelIds}, sq.Eq{"ChannelId": channelIds},
}) })
_, err := builder.RunWith(s.GetMaster()).Exec()
return err queryString, args, err := query.ToSql()
if err != nil {
return errors.Wrap(err, "retention_policies_channels_tosql")
}
if _, err := s.GetMasterX().Exec(queryString, args...); err != nil {
return errors.Wrapf(err, "failed to permanent delete retention policy channels with policyid=%s", policyId)
}
return nil
} }
func (s *SqlRetentionPolicyStore) GetTeams(policyId string, offset, limit int) (teams []*model.Team, err error) { func (s *SqlRetentionPolicyStore) GetTeams(policyId string, offset, limit int) ([]*model.Team, error) {
const query = ` query := s.getQueryBuilder().
SELECT Teams.* FROM RetentionPoliciesTeams Select("Teams.*").
INNER JOIN Teams ON RetentionPoliciesTeams.TeamId = Teams.Id From("RetentionPoliciesTeams").
WHERE RetentionPoliciesTeams.PolicyId = :PolicyId InnerJoin("Teams ON RetentionPoliciesTeams.TeamId = Teams.Id").
ORDER BY Teams.DisplayName, Teams.Id Where(sq.Eq{"RetentionPoliciesTeams.PolicyId": policyId}).
LIMIT :Limit OrderBy("Teams.DisplayName, Teams.Id").
OFFSET :Offset` Limit(uint64(limit)).
props := map[string]interface{}{"PolicyId": policyId, "Limit": limit, "Offset": offset} Offset(uint64(offset))
_, err = s.GetReplica().Select(&teams, query, props)
for _, team := range teams { queryString, args, err := query.ToSql()
team.PolicyID = &policyId if err != nil {
return nil, errors.Wrap(err, "retention_policies_teams_tosql")
} }
return
teams := []*model.Team{}
if err = s.GetReplicaX().Select(&teams, queryString, args...); err != nil {
return teams, errors.Wrap(err, "failed to find Teams")
}
return teams, nil
} }
func (s *SqlRetentionPolicyStore) GetTeamsCount(policyId string) (int64, error) { func (s *SqlRetentionPolicyStore) GetTeamsCount(policyId string) (int64, error) {
const query = ` query := s.getQueryBuilder().
SELECT COUNT(*) Select("Count(*)").
FROM RetentionPolicies From("RetentionPolicies").
INNER JOIN RetentionPoliciesTeams ON RetentionPolicies.Id = RetentionPoliciesTeams.PolicyId InnerJoin("RetentionPoliciesTeams ON RetentionPolicies.Id = RetentionPoliciesTeams.PolicyId").
WHERE RetentionPolicies.Id = :PolicyId` Where(sq.Eq{"RetentionPolicies.Id": policyId})
props := map[string]interface{}{"PolicyId": policyId}
return s.GetReplica().SelectInt(query, props) queryString, args, err := query.ToSql()
if err != nil {
return 0, errors.Wrap(err, "retention_policies_tosql")
}
var count int64
if err := s.GetReplicaX().Get(&count, queryString, args...); err != nil {
return 0, errors.Wrap(err, "failed to count RetentionPolicies")
}
return count, nil
} }
func (s *SqlRetentionPolicyStore) AddTeams(policyId string, teamIds []string) error { func (s *SqlRetentionPolicyStore) AddTeams(policyId string, teamIds []string) error {
@@ -519,54 +621,87 @@ func (s *SqlRetentionPolicyStore) AddTeams(policyId string, teamIds []string) er
if err := s.checkTeamsExist(teamIds); err != nil { if err := s.checkTeamsExist(teamIds); err != nil {
return err return err
} }
builder := s.getQueryBuilder(). query := s.getQueryBuilder().
Insert("RetentionPoliciesTeams"). Insert("RetentionPoliciesTeams").
Columns("PolicyId", "TeamId") Columns("PolicyId", "TeamId")
for _, teamId := range teamIds { for _, teamId := range teamIds {
builder = builder.Values(policyId, teamId) query = query.Values(policyId, teamId)
} }
_, err := builder.RunWith(s.GetMaster()).Exec()
return err queryString, args, err := query.ToSql()
if err != nil {
return errors.Wrap(err, "retention_policies_teams_tosql")
}
if _, err := s.GetMasterX().Exec(queryString, args...); err != nil {
return errors.Wrap(err, "failed to insert retention policies teams")
}
return nil
} }
func (s *SqlRetentionPolicyStore) RemoveTeams(policyId string, teamIds []string) error { func (s *SqlRetentionPolicyStore) RemoveTeams(policyId string, teamIds []string) error {
if len(teamIds) == 0 { if len(teamIds) == 0 {
return nil return nil
} }
builder := s.getQueryBuilder(). query := s.getQueryBuilder().
Delete("RetentionPoliciesTeams"). Delete("RetentionPoliciesTeams").
Where(sq.And{ Where(sq.And{
sq.Eq{"PolicyId": policyId}, sq.Eq{"PolicyId": policyId},
sq.Eq{"TeamId": teamIds}, sq.Eq{"TeamId": teamIds},
}) })
_, err := builder.RunWith(s.GetMaster()).Exec()
return err queryString, args, err := query.ToSql()
if err != nil {
return errors.Wrap(err, "retention_policies_teams_tosql")
}
if _, err := s.GetMasterX().Exec(queryString, args...); err != nil {
return errors.Wrapf(err, "unable to permanent delete retention policies teams with policyid=%s", policyId)
}
return nil
}
func subQueryIN(property string, query sq.SelectBuilder) sq.Sqlizer {
queryString, args, _ := query.ToSql()
subQuery := fmt.Sprintf("%s IN (SELECT * FROM (%s) AS A)", property, queryString)
return sq.Expr(subQuery, args...)
} }
// DeleteOrphanedRows removes entries from RetentionPoliciesChannels and RetentionPoliciesTeams // DeleteOrphanedRows removes entries from RetentionPoliciesChannels and RetentionPoliciesTeams
// where a channel or team no longer exists. // where a channel or team no longer exists.
func (s *SqlRetentionPolicyStore) DeleteOrphanedRows(limit int) (deleted int64, err error) { func (s *SqlRetentionPolicyStore) DeleteOrphanedRows(limit int) (deleted int64, err error) {
// We need the extra level of nesting to deal with MySQL's locking // We need the extra level of nesting to deal with MySQL's locking
const rpcDeleteQuery = ` rpcSubQuery := sq.Select("ChannelId").
DELETE FROM RetentionPoliciesChannels WHERE ChannelId IN ( From("RetentionPoliciesChannels").
SELECT * FROM ( LeftJoin("Channels ON RetentionPoliciesChannels.ChannelId = Channels.Id").
SELECT ChannelId FROM RetentionPoliciesChannels Where("Channels.Id IS NULL").
LEFT JOIN Channels ON RetentionPoliciesChannels.ChannelId = Channels.Id Limit(uint64(limit))
WHERE Channels.Id IS NULL
LIMIT :Limit rpcDeleteQuery, rpcArgs, err := s.getQueryBuilder().
) AS A Delete("RetentionPoliciesChannels").
)` Where(subQueryIN("ChannelId", rpcSubQuery)).
const rptDeleteQuery = ` ToSql()
DELETE FROM RetentionPoliciesTeams WHERE TeamId IN ( if err != nil {
SELECT * FROM ( return int64(0), errors.Wrap(err, "retention_policies_channels_tosql")
SELECT TeamId FROM RetentionPoliciesTeams }
LEFT JOIN Teams ON RetentionPoliciesTeams.TeamId = Teams.Id
WHERE Teams.Id IS NULL rptSubQuery := sq.Select("TeamId").
LIMIT :Limit From("RetentionPoliciesTeams").
) AS A LeftJoin("Teams ON RetentionPoliciesTeams.TeamId = Teams.Id").
)` Where("Teams.Id IS NULL").
props := map[string]interface{}{"Limit": limit} Limit(uint64(limit))
result, err := s.GetMaster().Exec(rpcDeleteQuery, props)
rptDeleteQuery, rptArgs, err := s.getQueryBuilder().
Delete("RetentionPoliciesTeams").
Where(subQueryIN("TeamId", rptSubQuery)).
ToSql()
if err != nil {
return int64(0), errors.Wrap(err, "retention_policies_teams_tosql")
}
result, err := s.GetMasterX().Exec(rpcDeleteQuery, rpcArgs...)
if err != nil { if err != nil {
return return
} }
@@ -574,7 +709,7 @@ func (s *SqlRetentionPolicyStore) DeleteOrphanedRows(limit int) (deleted int64,
if err != nil { if err != nil {
return return
} }
result, err = s.GetMaster().Exec(rptDeleteQuery, props) result, err = s.GetMasterX().Exec(rptDeleteQuery, rptArgs...)
if err != nil { if err != nil {
return return
} }
@@ -586,70 +721,124 @@ func (s *SqlRetentionPolicyStore) DeleteOrphanedRows(limit int) (deleted int64,
return return
} }
func (s *SqlRetentionPolicyStore) GetTeamPoliciesForUser(userID string, offset, limit int) (policies []*model.RetentionPolicyForTeam, err error) { func (s *SqlRetentionPolicyStore) GetTeamPoliciesForUser(userID string, offset, limit int) ([]*model.RetentionPolicyForTeam, error) {
const query = ` query := s.getQueryBuilder().
SELECT Teams.Id, RetentionPolicies.PostDuration Select(`Teams.Id AS "Id", RetentionPolicies.PostDuration`).
FROM Users From("Users").
INNER JOIN TeamMembers ON Users.Id = TeamMembers.UserId InnerJoin("TeamMembers ON Users.Id = TeamMembers.UserId").
INNER JOIN Teams ON TeamMembers.TeamId = Teams.Id InnerJoin("Teams ON TeamMembers.TeamId = Teams.Id").
INNER JOIN RetentionPoliciesTeams ON Teams.Id = RetentionPoliciesTeams.TeamId InnerJoin("RetentionPoliciesTeams ON Teams.Id = RetentionPoliciesTeams.TeamId").
INNER JOIN RetentionPolicies ON RetentionPoliciesTeams.PolicyId = RetentionPolicies.Id InnerJoin("RetentionPolicies ON RetentionPoliciesTeams.PolicyId = RetentionPolicies.Id").
WHERE Users.Id = :UserId Where(
AND TeamMembers.DeleteAt = 0 sq.And{
AND Teams.DeleteAt = 0 sq.Eq{"Users.Id": userID},
ORDER BY Teams.Id sq.Eq{"TeamMembers.DeleteAt": 0},
LIMIT :Limit sq.Eq{"Teams.DeleteAt": 0},
OFFSET :Offset` },
props := map[string]interface{}{"UserId": userID, "Limit": limit, "Offset": offset} ).
_, err = s.GetReplica().Select(&policies, query, props) OrderBy("Teams.Id").
return Limit(uint64(limit)).
Offset(uint64(offset))
queryString, args, err := query.ToSql()
if err != nil {
return nil, errors.Wrap(err, "team_policies_for_user_tosql")
}
policies := []*model.RetentionPolicyForTeam{}
if err := s.GetReplicaX().Select(&policies, queryString, args...); err != nil {
return policies, errors.Wrap(err, "failed to find Users")
}
return policies, nil
} }
func (s *SqlRetentionPolicyStore) GetTeamPoliciesCountForUser(userID string) (int64, error) { func (s *SqlRetentionPolicyStore) GetTeamPoliciesCountForUser(userID string) (int64, error) {
const query = ` query := s.getQueryBuilder().
SELECT COUNT(*) Select("Count(*)").
FROM Users From("Users").
INNER JOIN TeamMembers ON Users.Id = TeamMembers.UserId InnerJoin("TeamMembers ON Users.Id = TeamMembers.UserId").
INNER JOIN Teams ON TeamMembers.TeamId = Teams.Id InnerJoin("Teams ON TeamMembers.TeamId = Teams.Id").
INNER JOIN RetentionPoliciesTeams ON Teams.Id = RetentionPoliciesTeams.TeamId InnerJoin("RetentionPoliciesTeams ON Teams.Id = RetentionPoliciesTeams.TeamId").
INNER JOIN RetentionPolicies ON RetentionPoliciesTeams.PolicyId = RetentionPolicies.Id InnerJoin("RetentionPolicies ON RetentionPoliciesTeams.PolicyId = RetentionPolicies.Id").
WHERE Users.Id = :UserId Where(
AND TeamMembers.DeleteAt = 0 sq.And{
AND Teams.DeleteAt = 0` sq.Eq{"Users.Id": userID},
props := map[string]interface{}{"UserId": userID} sq.Eq{"TeamMembers.DeleteAt": 0},
return s.GetReplica().SelectInt(query, props) sq.Eq{"Teams.DeleteAt": 0},
},
)
queryString, args, err := query.ToSql()
if err != nil {
return 0, errors.Wrap(err, "team_policies_count_for_user_tosql")
}
var count int64
if err := s.GetReplicaX().Get(&count, queryString, args...); err != nil {
return 0, errors.Wrap(err, "failed to count TeamPoliciesCountForUser")
}
return count, nil
} }
func (s *SqlRetentionPolicyStore) GetChannelPoliciesForUser(userID string, offset, limit int) (policies []*model.RetentionPolicyForChannel, err error) { func (s *SqlRetentionPolicyStore) GetChannelPoliciesForUser(userID string, offset, limit int) ([]*model.RetentionPolicyForChannel, error) {
const query = ` query := s.getQueryBuilder().
SELECT Channels.Id, RetentionPolicies.PostDuration Select(`Channels.Id as "Id", RetentionPolicies.PostDuration`).
FROM Users From("Users").
INNER JOIN ChannelMembers ON Users.Id = ChannelMembers.UserId InnerJoin("ChannelMembers ON Users.Id = ChannelMembers.UserId").
INNER JOIN Channels ON ChannelMembers.ChannelId = Channels.Id InnerJoin("Channels ON ChannelMembers.ChannelId = Channels.Id").
INNER JOIN RetentionPoliciesChannels ON Channels.Id = RetentionPoliciesChannels.ChannelId InnerJoin("RetentionPoliciesChannels ON Channels.Id = RetentionPoliciesChannels.ChannelId").
INNER JOIN RetentionPolicies ON RetentionPoliciesChannels.PolicyId = RetentionPolicies.Id InnerJoin("RetentionPolicies ON RetentionPoliciesChannels.PolicyId = RetentionPolicies.Id").
WHERE Users.Id = :UserId Where(
AND Channels.DeleteAt = 0 sq.And{
ORDER BY Channels.Id sq.Eq{"Users.Id": userID},
LIMIT :Limit sq.Eq{"Channels.DeleteAt": 0},
OFFSET :Offset` },
props := map[string]interface{}{"UserId": userID, "Limit": limit, "Offset": offset} ).
_, err = s.GetReplica().Select(&policies, query, props) OrderBy("Channels.Id").
return Limit(uint64(limit)).
Offset(uint64(offset))
queryString, args, err := query.ToSql()
if err != nil {
return nil, errors.Wrap(err, "channel_policies_for_user_tosql")
}
policies := []*model.RetentionPolicyForChannel{}
if err := s.GetReplicaX().Select(&policies, queryString, args...); err != nil {
return nil, errors.Wrap(err, "failed to find Users")
}
return policies, nil
} }
func (s *SqlRetentionPolicyStore) GetChannelPoliciesCountForUser(userID string) (int64, error) { func (s *SqlRetentionPolicyStore) GetChannelPoliciesCountForUser(userID string) (int64, error) {
const query = ` query := s.getQueryBuilder().
SELECT COUNT(*) Select("Count(*)").
FROM Users From("Users").
INNER JOIN ChannelMembers ON Users.Id = ChannelMembers.UserId InnerJoin("ChannelMembers ON Users.Id = ChannelMembers.UserId").
INNER JOIN Channels ON ChannelMembers.ChannelId = Channels.Id InnerJoin("Channels ON ChannelMembers.ChannelId = Channels.Id").
INNER JOIN RetentionPoliciesChannels ON Channels.Id = RetentionPoliciesChannels.ChannelId InnerJoin("RetentionPoliciesChannels ON Channels.Id = RetentionPoliciesChannels.ChannelId").
INNER JOIN RetentionPolicies ON RetentionPoliciesChannels.PolicyId = RetentionPolicies.Id InnerJoin("RetentionPolicies ON RetentionPoliciesChannels.PolicyId = RetentionPolicies.Id").
WHERE Users.Id = :UserId Where(
AND Channels.DeleteAt = 0` sq.And{
props := map[string]interface{}{"UserId": userID} sq.Eq{"Users.Id": userID},
return s.GetReplica().SelectInt(query, props) sq.Eq{"Channels.DeleteAt": 0},
},
)
queryString, args, err := query.ToSql()
if err != nil {
return 0, errors.Wrap(err, "channel_policies_count_users_tosql")
}
var count int64
if err := s.GetReplicaX().Get(&count, queryString, args...); err != nil {
return 0, errors.Wrap(err, "failed to count ChannelPoliciesCountForUser")
}
return count, nil
} }
// RetentionPolicyBatchDeletionInfo gives information on how to delete records // RetentionPolicyBatchDeletionInfo gives information on how to delete records
@@ -811,7 +1000,7 @@ func genericRetentionPoliciesDeletion(
` + query + ` ` + query + `
) AS A ON ` + joinClause ) AS A ON ` + joinClause
} }
result, err := s.GetMaster().Exec(query, args...) result, err := s.GetMasterX().Exec(query, args...)
if err != nil { if err != nil {
return 0, errors.Wrap(err, "failed to delete "+r.Table) return 0, errors.Wrap(err, "failed to delete "+r.Table)
} }