PLT-7503: Create Message Export Scheduled Task and CLI Command (#7612)

* Created message export scheduled task

* Added CLI command to immediately kick off an export job

* Added email addresses for users joining and leaving the channel to the export

* Added support for both MySQL and PostgreSQL

* Fixing gofmt error

* Added a new ChannelMemberHistory store and associated tests

* Updating the ChannelMemberHistory channel as users create/join/leave channels

* Added user email to the message export object so it can be included in the actiance export xml

* Don't fail to log a leave event if a corresponding join event wasn't logged

* Adding copyright notices

* Adding message export settings to daily diagnostics report

* Added System Console integration for message export

* Cleaned up TODOs

* Made batch size configurable

* Added export from timestamp to CLI command

* Made ChannelMemberHistory table updates best effort

* Added a context-based timeout option to the message export CLI

* Minor PR updates/improvements

* Removed unnecessary fields from MessageExport object to reduce query overhead

* Removed JSON functions from the message export query in an effort to optimize performance

* Changed the way that channel member history queries and purges work to better account for edge cases

* Fixing a test I missed with the last refactor

* Added file copy functionality to file backend, improved config validation, added default config values

* Fixed file copy tests

* More concise use of the testing libraries

* Fixed context leak error

* Changed default export path to correctly place an 'export' directory under the 'data' directory

* Can't delete records from a read replica

* Fixed copy file tests

* Start job workers when license is applied, if configured to do so

* Suggestions from the PR

* Moar unit tests

* Fixed test imports

4 files changed, 178 insertions, 23 deletions

diff --git a/store/sqlstore/channel_member_history_store.go b/store/sqlstore/channel_member_history_store.go
new file mode 100644
index 000000000..20d0d3335
--- /dev/null
+++ b/store/sqlstore/channel_member_history_store.go
@@ -0,0 +1,102 @@
+// Copyright (c) 2017-present Mattermost, Inc. All Rights Reserved.
+// See License.txt for license information.
+
+package sqlstore
+
+import (
+	"net/http"
+
+	l4g "github.com/alecthomas/log4go"
+	"github.com/mattermost/mattermost-server/model"
+	"github.com/mattermost/mattermost-server/store"
+)
+
+type SqlChannelMemberHistoryStore struct {
+	SqlStore
+}
+
+func NewSqlChannelMemberHistoryStore(sqlStore SqlStore) store.ChannelMemberHistoryStore {
+	s := &SqlChannelMemberHistoryStore{
+		SqlStore: sqlStore,
+	}
+
+	for _, db := range sqlStore.GetAllConns() {
+		table := db.AddTableWithName(model.ChannelMemberHistory{}, "ChannelMemberHistory").SetKeys(false, "ChannelId", "UserId", "JoinTime")
+		table.ColMap("ChannelId").SetMaxSize(26)
+		table.ColMap("UserId").SetMaxSize(26)
+		table.ColMap("JoinTime").SetNotNull(true)
+	}
+
+	return s
+}
+
+func (s SqlChannelMemberHistoryStore) LogJoinEvent(userId string, channelId string, joinTime int64) store.StoreChannel {
+	return store.Do(func(result *store.StoreResult) {
+		channelMemberHistory := &model.ChannelMemberHistory{
+			UserId:    userId,
+			ChannelId: channelId,
+			JoinTime:  joinTime,
+		}
+
+		if err := s.GetMaster().Insert(channelMemberHistory); err != nil {
+			result.Err = model.NewAppError("SqlChannelMemberHistoryStore.LogJoinEvent", "store.sql_channel_member_history.log_join_event.app_error", map[string]interface{}{"ChannelMemberHistory": channelMemberHistory}, err.Error(), http.StatusInternalServerError)
+		}
+	})
+}
+
+func (s SqlChannelMemberHistoryStore) LogLeaveEvent(userId string, channelId string, leaveTime int64) store.StoreChannel {
+	return store.Do(func(result *store.StoreResult) {
+		query := `
+			UPDATE ChannelMemberHistory
+			SET LeaveTime = :LeaveTime
+			WHERE UserId = :UserId
+			AND ChannelId = :ChannelId
+			AND LeaveTime IS NULL`
+
+		params := map[string]interface{}{"UserId": userId, "ChannelId": channelId, "LeaveTime": leaveTime}
+		if sqlResult, err := s.GetMaster().Exec(query, params); err != nil {
+			result.Err = model.NewAppError("SqlChannelMemberHistoryStore.LogLeaveEvent", "store.sql_channel_member_history.log_leave_event.update_error", nil, err.Error(), http.StatusInternalServerError)
+		} else if rows, err := sqlResult.RowsAffected(); err == nil && rows != 1 {
+			// there was no join event to update
+			l4g.Warn("Channel join event for user %v and channel %v not found", userId, channelId)
+		}
+	})
+}
+
+func (s SqlChannelMemberHistoryStore) GetUsersInChannelDuring(startTime int64, endTime int64, channelId string) store.StoreChannel {
+	return store.Do(func(result *store.StoreResult) {
+		query := `
+			SELECT
+				cmh.*,
+				u.Email
+			FROM ChannelMemberHistory cmh
+			INNER JOIN Users u ON cmh.UserId = u.Id
+			WHERE cmh.ChannelId = :ChannelId
+			AND cmh.JoinTime <= :EndTime
+			AND (cmh.LeaveTime IS NULL OR cmh.LeaveTime >= :StartTime)
+			ORDER BY cmh.JoinTime ASC`
+
+		params := map[string]interface{}{"ChannelId": channelId, "StartTime": startTime, "EndTime": endTime}
+		var histories []*model.ChannelMemberHistory
+		if _, err := s.GetReplica().Select(&histories, query, params); err != nil {
+			result.Err = model.NewAppError("SqlChannelMemberHistoryStore.GetUsersInChannelAt", "store.sql_channel_member_history.get_users_in_channel_during.app_error", params, err.Error(), http.StatusInternalServerError)
+		} else {
+			result.Data = histories
+		}
+	})
+}
+
+func (s SqlChannelMemberHistoryStore) PurgeHistoryBefore(time int64, channelId string) store.StoreChannel {
+	return store.Do(func(result *store.StoreResult) {
+		query := `
+			DELETE FROM ChannelMemberHistory
+			WHERE ChannelId = :ChannelId
+			AND LeaveTime IS NOT NULL
+			AND LeaveTime <= :AtTime`
+
+		params := map[string]interface{}{"AtTime": time, "ChannelId": channelId}
+		if _, err := s.GetMaster().Exec(query, params); err != nil {
+			result.Err = model.NewAppError("SqlChannelMemberHistoryStore.PurgeHistoryBefore", "store.sql_channel_member_history.purge_history_before.app_error", params, err.Error(), http.StatusInternalServerError)
+		}
+	})
+}
diff --git a/store/sqlstore/channel_member_history_store_test.go b/store/sqlstore/channel_member_history_store_test.go
new file mode 100644
index 000000000..c1119d227
--- /dev/null
+++ b/store/sqlstore/channel_member_history_store_test.go
@@ -0,0 +1,14 @@
+// Copyright (c) 2015-present Mattermost, Inc. All Rights Reserved.
+// See License.txt for license information.
+
+package sqlstore
+
+import (
+	"testing"
+
+	"github.com/mattermost/mattermost-server/store/storetest"
+)
+
+func TestChannelMemberHistoryStore(t *testing.T) {
+	StoreTest(t, storetest.TestChannelMemberHistoryStore)
+}
diff --git a/store/sqlstore/compliance_store.go b/store/sqlstore/compliance_store.go
index 3d638b1fd..a25b01548 100644
--- a/store/sqlstore/compliance_store.go
+++ b/store/sqlstore/compliance_store.go
@@ -211,3 +211,36 @@ func (s SqlComplianceStore) ComplianceExport(job *model.Compliance) store.StoreC
 		}
 	})
 }
+
+func (s SqlComplianceStore) MessageExport(after int64, limit int) store.StoreChannel {
+	return store.Do(func(result *store.StoreResult) {
+		props := map[string]interface{}{"StartTime": after, "Limit": limit}
+		query :=
+			`SELECT
+				Posts.Id AS PostId,
+				Posts.CreateAt AS PostCreateAt,
+				Posts.Message AS PostMessage,
+				Posts.Type AS PostType,
+				Posts.FileIds AS PostFileIds,
+				Channels.Id AS ChannelId,
+				Channels.DisplayName AS ChannelDisplayName,
+				Users.Id AS UserId,
+				Users.Email AS UserEmail
+			FROM
+				Posts
+				LEFT OUTER JOIN Channels ON Posts.ChannelId = Channels.Id
+				LEFT OUTER JOIN Users ON Posts.UserId = Users.Id
+			WHERE
+				Posts.CreateAt > :StartTime AND
+				Posts.Type = ''
+			ORDER BY PostCreateAt
+			LIMIT :Limit`
+
+		var cposts []*model.MessageExport
+		if _, err := s.GetReplica().Select(&cposts, query, props); err != nil {
+			result.Err = model.NewAppError("SqlComplianceStore.MessageExport", "store.sql_compliance.message_export.app_error", nil, err.Error(), http.StatusInternalServerError)
+		} else {
+			result.Data = cposts
+		}
+	})
+}
diff --git a/store/sqlstore/supplier.go b/store/sqlstore/supplier.go
index dbe4aa92c..3b9528578 100644
--- a/store/sqlstore/supplier.go
+++ b/store/sqlstore/supplier.go
@@ -62,29 +62,30 @@ const (
 )
 
 type SqlSupplierOldStores struct {
-	team            store.TeamStore
-	channel         store.ChannelStore
-	post            store.PostStore
-	user            store.UserStore
-	audit           store.AuditStore
-	cluster         store.ClusterDiscoveryStore
-	compliance      store.ComplianceStore
-	session         store.SessionStore
-	oauth           store.OAuthStore
-	system          store.SystemStore
-	webhook         store.WebhookStore
-	command         store.CommandStore
-	commandWebhook  store.CommandWebhookStore
-	preference      store.PreferenceStore
-	license         store.LicenseStore
-	token           store.TokenStore
-	emoji           store.EmojiStore
-	status          store.StatusStore
-	fileInfo        store.FileInfoStore
-	reaction        store.ReactionStore
-	job             store.JobStore
-	userAccessToken store.UserAccessTokenStore
-	plugin          store.PluginStore
+	team                 store.TeamStore
+	channel              store.ChannelStore
+	post                 store.PostStore
+	user                 store.UserStore
+	audit                store.AuditStore
+	cluster              store.ClusterDiscoveryStore
+	compliance           store.ComplianceStore
+	session              store.SessionStore
+	oauth                store.OAuthStore
+	system               store.SystemStore
+	webhook              store.WebhookStore
+	command              store.CommandStore
+	commandWebhook       store.CommandWebhookStore
+	preference           store.PreferenceStore
+	license              store.LicenseStore
+	token                store.TokenStore
+	emoji                store.EmojiStore
+	status               store.StatusStore
+	fileInfo             store.FileInfoStore
+	reaction             store.ReactionStore
+	job                  store.JobStore
+	userAccessToken      store.UserAccessTokenStore
+	plugin               store.PluginStore
+	channelMemberHistory store.ChannelMemberHistoryStore
 }
 
 type SqlSupplier struct {
@@ -130,6 +131,7 @@ func NewSqlSupplier(settings model.SqlSettings, metrics einterfaces.MetricsInter
 	supplier.oldStores.fileInfo = NewSqlFileInfoStore(supplier, metrics)
 	supplier.oldStores.job = NewSqlJobStore(supplier)
 	supplier.oldStores.userAccessToken = NewSqlUserAccessTokenStore(supplier)
+	supplier.oldStores.channelMemberHistory = NewSqlChannelMemberHistoryStore(supplier)
 	supplier.oldStores.plugin = NewSqlPluginStore(supplier)
 
 	initSqlSupplierReactions(supplier)
@@ -801,6 +803,10 @@ func (ss *SqlSupplier) UserAccessToken() store.UserAccessTokenStore {
 	return ss.oldStores.userAccessToken
 }
 
+func (ss *SqlSupplier) ChannelMemberHistory() store.ChannelMemberHistoryStore {
+	return ss.oldStores.channelMemberHistory
+}
+
 func (ss *SqlSupplier) Plugin() store.PluginStore {
 	return ss.oldStores.plugin
 }