Merge branch 'master' into MM-47853-true-up-review-telemetry-off-non-air-gapped

2022-12-23 23:12:20 +02:00
--- a/store/sqlstore/channel_store.go
+++ b/store/sqlstore/channel_store.go
@@ -3850,7 +3850,31 @@ func (s SqlChannelStore) MigrateChannelMembers(fromChannelId string, fromUserId
 	defer finalizeTransactionX(transaction, &err)

 	channelMembers := []channelMember{}
-	if err := transaction.Select(&channelMembers, "SELECT * from ChannelMembers WHERE (ChannelId, UserId) > (?, ?) ORDER BY ChannelId, UserId LIMIT 100", fromChannelId, fromUserId); err != nil {
+	query := `
+		SELECT
+			ChannelId,
+			UserId,
+			Roles,
+			LastViewedAt,
+			MsgCount,
+			MentionCount,
+			MentionCountRoot,
+			COALESCE(UrgentMentionCount, 0) AS UrgentMentionCount,
+			MsgCountRoot,
+			NotifyProps,
+			LastUpdateAt,
+			SchemeUser,
+			SchemeAdmin,
+			SchemeGuest
+		FROM
+			ChannelMembers
+		WHERE
+			(ChannelId, UserId) > (?, ?)
+		ORDER BY ChannelId, UserId
+		LIMIT 100
+	`
+
+	if err := transaction.Select(&channelMembers, query, fromChannelId, fromUserId); err != nil {
 		return nil, errors.Wrap(err, "failed to find ChannelMembers")
 	}

@@ -3954,7 +3978,31 @@ func (s SqlChannelStore) ClearAllCustomRoleAssignments() (err error) {
 		}

 		channelMembers := []*channelMember{}
-		if err = transaction.Select(&channelMembers, "SELECT * from ChannelMembers WHERE (ChannelId, UserId) > (?, ?) ORDER BY ChannelId, UserId LIMIT 1000", lastChannelId, lastUserId); err != nil {
+		query := `
+			SELECT
+				ChannelId,
+				UserId,
+				Roles,
+				LastViewedAt,
+				MsgCount,
+				MentionCount,
+				MentionCountRoot,
+				COALESCE(UrgentMentionCount, 0) AS UrgentMentionCount,
+				MsgCountRoot,
+				NotifyProps,
+				LastUpdateAt,
+				SchemeUser,
+				SchemeAdmin,
+				SchemeGuest
+			FROM
+				ChannelMembers
+			WHERE
+				(ChannelId, UserId) > (?, ?)
+			ORDER BY ChannelId, UserId
+			LIMIT 1000
+		`
+
+		if err = transaction.Select(&channelMembers, query, lastChannelId, lastUserId); err != nil {
 			finalizeTransactionX(transaction, &err)
 			return errors.Wrap(err, "failed to find ChannelMembers")
 		}
--- a/store/sqlstore/compliance_store.go
+++ b/store/sqlstore/compliance_store.go
@@ -4,6 +4,7 @@
 package sqlstore

 import (
+	"context"
 	"database/sql"
 	"fmt"
 	"strings"
@@ -270,7 +271,7 @@ func (s SqlComplianceStore) ComplianceExport(job *model.Compliance, cursor model
 	return append(channelPosts, directMessagePosts...), cursor, nil
 }

-func (s SqlComplianceStore) MessageExport(cursor model.MessageExportCursor, limit int) ([]*model.MessageExport, model.MessageExportCursor, error) {
+func (s SqlComplianceStore) MessageExport(ctx context.Context, cursor model.MessageExportCursor, limit int) ([]*model.MessageExport, model.MessageExportCursor, error) {
 	var args []any
 	args = append(args, model.ChannelTypeDirect, model.ChannelTypeGroup, cursor.LastPostUpdateAt, cursor.LastPostUpdateAt, cursor.LastPostId, limit)
 	query :=
@@ -317,7 +318,7 @@ func (s SqlComplianceStore) MessageExport(cursor model.MessageExportCursor, limi
 		LIMIT ?`

 	cposts := []*model.MessageExport{}
-	if err := s.GetReplicaX().Select(&cposts, query, args...); err != nil {
+	if err := s.GetReplicaX().SelectCtx(ctx, &cposts, query, args...); err != nil {
 		return nil, cursor, errors.Wrap(err, "unable to export messages")
 	}
 	if len(cposts) > 0 {
--- a/store/sqlstore/sqlx_wrapper.go
+++ b/store/sqlstore/sqlx_wrapper.go
@@ -224,8 +224,12 @@ func (w *sqlxDBWrapper) QueryX(query string, args ...any) (*sqlx.Rows, error) {
 }

 func (w *sqlxDBWrapper) Select(dest any, query string, args ...any) error {
+	return w.SelectCtx(context.Background(), dest, query, args...)
+}
+
+func (w *sqlxDBWrapper) SelectCtx(ctx context.Context, dest any, query string, args ...any) error {
 	query = w.DB.Rebind(query)
-	ctx, cancel := context.WithTimeout(context.Background(), w.queryTimeout)
+	ctx, cancel := context.WithTimeout(ctx, w.queryTimeout)
 	defer cancel()

 	if w.trace {
--- a/store/sqlstore/store.go
+++ b/store/sqlstore/store.go
@@ -9,7 +9,7 @@ import (
 	dbsql "database/sql"
 	"fmt"
 	"log"
-	"path/filepath"
+	"path"
 	"strconv"
 	"strings"
 	"sync"
@@ -228,7 +228,9 @@ func New(settings model.SqlSettings, metrics einterfaces.MetricsInterface) *SqlS
 	return store
 }

-func setupConnection(connType string, dataSource string, settings *model.SqlSettings) *dbsql.DB {
+// SetupConnection sets up the connection to the database and pings it to make sure it's alive.
+// It also applies any database configuration settings that are required.
+func SetupConnection(connType string, dataSource string, settings *model.SqlSettings) *dbsql.DB {
 	db, err := dbsql.Open(*settings.DriverName, dataSource)
 	if err != nil {
 		mlog.Fatal("Failed to open SQL connection to err.", mlog.Err(err))
@@ -294,7 +296,7 @@ func (ss *SqlStore) initConnection() {
 		}
 	}

-	handle := setupConnection("master", dataSource, ss.settings)
+	handle := SetupConnection("master", dataSource, ss.settings)
 	ss.masterX = newSqlxDBWrapper(sqlx.NewDb(handle, ss.DriverName()),
 		time.Duration(*ss.settings.QueryTimeout)*time.Second,
 		*ss.settings.Trace)
@@ -305,7 +307,7 @@ func (ss *SqlStore) initConnection() {
 	if len(ss.settings.DataSourceReplicas) > 0 {
 		ss.ReplicaXs = make([]*sqlxDBWrapper, len(ss.settings.DataSourceReplicas))
 		for i, replica := range ss.settings.DataSourceReplicas {
-			handle := setupConnection(fmt.Sprintf("replica-%v", i), replica, ss.settings)
+			handle := SetupConnection(fmt.Sprintf("replica-%v", i), replica, ss.settings)
 			ss.ReplicaXs[i] = newSqlxDBWrapper(sqlx.NewDb(handle, ss.DriverName()),
 				time.Duration(*ss.settings.QueryTimeout)*time.Second,
 				*ss.settings.Trace)
@@ -318,7 +320,7 @@ func (ss *SqlStore) initConnection() {
 	if len(ss.settings.DataSourceSearchReplicas) > 0 {
 		ss.searchReplicaXs = make([]*sqlxDBWrapper, len(ss.settings.DataSourceSearchReplicas))
 		for i, replica := range ss.settings.DataSourceSearchReplicas {
-			handle := setupConnection(fmt.Sprintf("search-replica-%v", i), replica, ss.settings)
+			handle := SetupConnection(fmt.Sprintf("search-replica-%v", i), replica, ss.settings)
 			ss.searchReplicaXs[i] = newSqlxDBWrapper(sqlx.NewDb(handle, ss.DriverName()),
 				time.Duration(*ss.settings.QueryTimeout)*time.Second,
 				*ss.settings.Trace)
@@ -334,7 +336,7 @@ func (ss *SqlStore) initConnection() {
 			if src.DataSource == nil {
 				continue
 			}
-			ss.replicaLagHandles[i] = setupConnection(fmt.Sprintf(replicaLagPrefix+"-%d", i), *src.DataSource, ss.settings)
+			ss.replicaLagHandles[i] = SetupConnection(fmt.Sprintf(replicaLagPrefix+"-%d", i), *src.DataSource, ss.settings)
 		}
 	}
 }
@@ -1047,7 +1049,7 @@ func (ss *SqlStore) hasLicense() bool {
 func (ss *SqlStore) migrate(direction migrationDirection) error {
 	assets := db.Assets()

-	assetsList, err := assets.ReadDir(filepath.Join("migrations", ss.DriverName()))
+	assetsList, err := assets.ReadDir(path.Join("migrations", ss.DriverName()))
 	if err != nil {
 		return err
 	}
@@ -1060,7 +1062,7 @@ func (ss *SqlStore) migrate(direction migrationDirection) error {
 	src, err := mbindata.WithInstance(&mbindata.AssetSource{
 		Names: assetNamesForDriver,
 		AssetFunc: func(name string) ([]byte, error) {
-			return assets.ReadFile(filepath.Join("migrations", ss.DriverName(), name))
+			return assets.ReadFile(path.Join("migrations", ss.DriverName(), name))
 		},
 	})
 	if err != nil {
@@ -1079,7 +1081,7 @@ func (ss *SqlStore) migrate(direction migrationDirection) error {
 		if err != nil {
 			return err
 		}
-		db := setupConnection("master", dataSource, ss.settings)
+		db := SetupConnection("master", dataSource, ss.settings)
 		driver, err = ms.WithInstance(db)
 		defer db.Close()
 	case model.DatabaseDriverPostgres: