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

13 files changed, 646 insertions, 52 deletions

diff --git a/store/layered_store.go b/store/layered_store.go
index ecf02864c..65b4670c0 100644
--- a/store/layered_store.go
+++ b/store/layered_store.go
@@ -153,6 +153,10 @@ func (s *LayeredStore) UserAccessToken() UserAccessTokenStore {
 	return s.DatabaseLayer.UserAccessToken()
 }
 
+func (s *LayeredStore) ChannelMemberHistory() ChannelMemberHistoryStore {
+	return s.DatabaseLayer.ChannelMemberHistory()
+}
+
 func (s *LayeredStore) Plugin() PluginStore {
 	return s.DatabaseLayer.Plugin()
 }
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
 }
diff --git a/store/store.go b/store/store.go
index 3c950495d..c95888c22 100644
--- a/store/store.go
+++ b/store/store.go
@@ -63,6 +63,7 @@ type Store interface {
 	Reaction() ReactionStore
 	Job() JobStore
 	UserAccessToken() UserAccessTokenStore
+	ChannelMemberHistory() ChannelMemberHistoryStore
 	Plugin() PluginStore
 	MarkSystemRanUnitTests()
 	Close()
@@ -160,6 +161,13 @@ type ChannelStore interface {
 	GetChannelUnread(channelId, userId string) StoreChannel
 }
 
+type ChannelMemberHistoryStore interface {
+	LogJoinEvent(userId string, channelId string, joinTime int64) StoreChannel
+	LogLeaveEvent(userId string, channelId string, leaveTime int64) StoreChannel
+	GetUsersInChannelDuring(startTime int64, endTime int64, channelId string) StoreChannel
+	PurgeHistoryBefore(time int64, channelId string) StoreChannel
+}
+
 type PostStore interface {
 	Save(post *model.Post) StoreChannel
 	Update(newPost *model.Post, oldPost *model.Post) StoreChannel
@@ -276,6 +284,7 @@ type ComplianceStore interface {
 	Get(id string) StoreChannel
 	GetAll(offset, limit int) StoreChannel
 	ComplianceExport(compliance *model.Compliance) StoreChannel
+	MessageExport(after int64, limit int) StoreChannel
 }
 
 type OAuthStore interface {
diff --git a/store/storetest/channel_member_history_store.go b/store/storetest/channel_member_history_store.go
new file mode 100644
index 000000000..c73a25f65
--- /dev/null
+++ b/store/storetest/channel_member_history_store.go
@@ -0,0 +1,179 @@
+// Copyright (c) 2017-present Mattermost, Inc. All Rights Reserved.
+// See License.txt for license information.
+
+package storetest
+
+import (
+	"testing"
+
+	"github.com/mattermost/mattermost-server/model"
+	"github.com/mattermost/mattermost-server/store"
+	"github.com/stretchr/testify/assert"
+)
+
+func TestChannelMemberHistoryStore(t *testing.T, ss store.Store) {
+	t.Run("Log Join Event", func(t *testing.T) { testLogJoinEvent(t, ss) })
+	t.Run("Log Leave Event", func(t *testing.T) { testLogLeaveEvent(t, ss) })
+	t.Run("Get Users In Channel At Time", func(t *testing.T) { testGetUsersInChannelAt(t, ss) })
+	t.Run("Purge History", func(t *testing.T) { testPurgeHistoryBefore(t, ss) })
+}
+
+func testLogJoinEvent(t *testing.T, ss store.Store) {
+	// create a test channel
+	channel := model.Channel{
+		TeamId:      model.NewId(),
+		DisplayName: "Display " + model.NewId(),
+		Name:        "zz" + model.NewId() + "b",
+		Type:        model.CHANNEL_OPEN,
+	}
+	channel = *store.Must(ss.Channel().Save(&channel, -1)).(*model.Channel)
+
+	// and a test user
+	user := model.User{
+		Email:    model.NewId() + "@mattermost.com",
+		Nickname: model.NewId(),
+	}
+	user = *store.Must(ss.User().Save(&user)).(*model.User)
+
+	// log a join event
+	result := <-ss.ChannelMemberHistory().LogJoinEvent(user.Id, channel.Id, model.GetMillis())
+	assert.Nil(t, result.Err)
+}
+
+func testLogLeaveEvent(t *testing.T, ss store.Store) {
+	// create a test channel
+	channel := model.Channel{
+		TeamId:      model.NewId(),
+		DisplayName: "Display " + model.NewId(),
+		Name:        "zz" + model.NewId() + "b",
+		Type:        model.CHANNEL_OPEN,
+	}
+	channel = *store.Must(ss.Channel().Save(&channel, -1)).(*model.Channel)
+
+	// and a test user
+	user := model.User{
+		Email:    model.NewId() + "@mattermost.com",
+		Nickname: model.NewId(),
+	}
+	user = *store.Must(ss.User().Save(&user)).(*model.User)
+
+	// log a join event, followed by a leave event
+	result := <-ss.ChannelMemberHistory().LogJoinEvent(user.Id, channel.Id, model.GetMillis())
+	assert.Nil(t, result.Err)
+
+	result = <-ss.ChannelMemberHistory().LogLeaveEvent(user.Id, channel.Id, model.GetMillis())
+	assert.Nil(t, result.Err)
+}
+
+func testGetUsersInChannelAt(t *testing.T, ss store.Store) {
+	// create a test channel
+	channel := model.Channel{
+		TeamId:      model.NewId(),
+		DisplayName: "Display " + model.NewId(),
+		Name:        "zz" + model.NewId() + "b",
+		Type:        model.CHANNEL_OPEN,
+	}
+	channel = *store.Must(ss.Channel().Save(&channel, -1)).(*model.Channel)
+
+	// and a test user
+	user := model.User{
+		Email:    model.NewId() + "@mattermost.com",
+		Nickname: model.NewId(),
+	}
+	user = *store.Must(ss.User().Save(&user)).(*model.User)
+
+	// log a join event
+	leaveTime := model.GetMillis()
+	joinTime := leaveTime - 10000
+	store.Must(ss.ChannelMemberHistory().LogJoinEvent(user.Id, channel.Id, joinTime))
+
+	// case 1: both start and end before join time
+	channelMembers := store.Must(ss.ChannelMemberHistory().GetUsersInChannelDuring(joinTime-500, joinTime-100, channel.Id)).([]*model.ChannelMemberHistory)
+	assert.Len(t, channelMembers, 0)
+
+	// case 2: start before join time, no leave time
+	channelMembers = store.Must(ss.ChannelMemberHistory().GetUsersInChannelDuring(joinTime-100, joinTime+100, channel.Id)).([]*model.ChannelMemberHistory)
+	assert.Len(t, channelMembers, 1)
+	assert.Equal(t, channel.Id, channelMembers[0].ChannelId)
+	assert.Equal(t, user.Id, channelMembers[0].UserId)
+	assert.Equal(t, user.Email, channelMembers[0].UserEmail)
+	assert.Equal(t, joinTime, channelMembers[0].JoinTime)
+	assert.Nil(t, channelMembers[0].LeaveTime)
+
+	// case 3: start after join time, no leave time
+	channelMembers = store.Must(ss.ChannelMemberHistory().GetUsersInChannelDuring(joinTime+100, joinTime+500, channel.Id)).([]*model.ChannelMemberHistory)
+	assert.Len(t, channelMembers, 1)
+	assert.Equal(t, channel.Id, channelMembers[0].ChannelId)
+	assert.Equal(t, user.Id, channelMembers[0].UserId)
+	assert.Equal(t, user.Email, channelMembers[0].UserEmail)
+	assert.Equal(t, joinTime, channelMembers[0].JoinTime)
+	assert.Nil(t, channelMembers[0].LeaveTime)
+
+	// add a leave time for the user
+	store.Must(ss.ChannelMemberHistory().LogLeaveEvent(user.Id, channel.Id, leaveTime))
+
+	// case 4: start after join time, end before leave time
+	channelMembers = store.Must(ss.ChannelMemberHistory().GetUsersInChannelDuring(joinTime+100, leaveTime-100, channel.Id)).([]*model.ChannelMemberHistory)
+	assert.Len(t, channelMembers, 1)
+	assert.Equal(t, channel.Id, channelMembers[0].ChannelId)
+	assert.Equal(t, user.Id, channelMembers[0].UserId)
+	assert.Equal(t, user.Email, channelMembers[0].UserEmail)
+	assert.Equal(t, joinTime, channelMembers[0].JoinTime)
+	assert.Equal(t, leaveTime, *channelMembers[0].LeaveTime)
+
+	// case 5: start before join time, end after leave time
+	channelMembers = store.Must(ss.ChannelMemberHistory().GetUsersInChannelDuring(joinTime-100, leaveTime+100, channel.Id)).([]*model.ChannelMemberHistory)
+	assert.Len(t, channelMembers, 1)
+	assert.Equal(t, channel.Id, channelMembers[0].ChannelId)
+	assert.Equal(t, user.Id, channelMembers[0].UserId)
+	assert.Equal(t, user.Email, channelMembers[0].UserEmail)
+	assert.Equal(t, joinTime, channelMembers[0].JoinTime)
+	assert.Equal(t, leaveTime, *channelMembers[0].LeaveTime)
+
+	// case 6: start and end after leave time
+	channelMembers = store.Must(ss.ChannelMemberHistory().GetUsersInChannelDuring(leaveTime+100, leaveTime+200, channel.Id)).([]*model.ChannelMemberHistory)
+	assert.Len(t, channelMembers, 0)
+}
+
+func testPurgeHistoryBefore(t *testing.T, ss store.Store) {
+	// create a test channel
+	channel := model.Channel{
+		TeamId:      model.NewId(),
+		DisplayName: "Display " + model.NewId(),
+		Name:        "zz" + model.NewId() + "b",
+		Type:        model.CHANNEL_OPEN,
+	}
+	channel = *store.Must(ss.Channel().Save(&channel, -1)).(*model.Channel)
+
+	// and two test users
+	user := model.User{
+		Email:    model.NewId() + "@mattermost.com",
+		Nickname: model.NewId(),
+	}
+	user = *store.Must(ss.User().Save(&user)).(*model.User)
+
+	user2 := model.User{
+		Email:    model.NewId() + "@mattermost.com",
+		Nickname: model.NewId(),
+	}
+	user2 = *store.Must(ss.User().Save(&user2)).(*model.User)
+
+	// user1 joins and leaves the channel
+	leaveTime := model.GetMillis()
+	joinTime := leaveTime - 10000
+	store.Must(ss.ChannelMemberHistory().LogJoinEvent(user.Id, channel.Id, joinTime))
+	store.Must(ss.ChannelMemberHistory().LogLeaveEvent(user.Id, channel.Id, leaveTime))
+
+	// user2 joins the channel but never leaves
+	store.Must(ss.ChannelMemberHistory().LogJoinEvent(user2.Id, channel.Id, joinTime))
+
+	// in between the join time and the leave time, both users were members of the channel
+	channelMembers := store.Must(ss.ChannelMemberHistory().GetUsersInChannelDuring(joinTime+10, leaveTime-10, channel.Id)).([]*model.ChannelMemberHistory)
+	assert.Len(t, channelMembers, 2)
+
+	// but if we purge the old data, only the user that didn't leave is left
+	store.Must(ss.ChannelMemberHistory().PurgeHistoryBefore(leaveTime, channel.Id))
+	channelMembers = store.Must(ss.ChannelMemberHistory().GetUsersInChannelDuring(joinTime+10, leaveTime-10, channel.Id)).([]*model.ChannelMemberHistory)
+	assert.Len(t, channelMembers, 1)
+	assert.Equal(t, user2.Id, channelMembers[0].UserId)
+}
diff --git a/store/storetest/compliance_store.go b/store/storetest/compliance_store.go
index 514910f6f..c5bd60f05 100644
--- a/store/storetest/compliance_store.go
+++ b/store/storetest/compliance_store.go
@@ -9,12 +9,14 @@ import (
 
 	"github.com/mattermost/mattermost-server/model"
 	"github.com/mattermost/mattermost-server/store"
+	"github.com/stretchr/testify/assert"
 )
 
 func TestComplianceStore(t *testing.T, ss store.Store) {
 	t.Run("", func(t *testing.T) { testComplianceStore(t, ss) })
 	t.Run("ComplianceExport", func(t *testing.T) { testComplianceExport(t, ss) })
 	t.Run("ComplianceExportDirectMessages", func(t *testing.T) { testComplianceExportDirectMessages(t, ss) })
+	t.Run("MessageExport", func(t *testing.T) { testComplianceMessageExport(t, ss) })
 }
 
 func testComplianceStore(t *testing.T, ss store.Store) {
@@ -316,3 +318,118 @@ func testComplianceExportDirectMessages(t *testing.T, ss store.Store) {
 		}
 	}
 }
+
+func testComplianceMessageExport(t *testing.T, ss store.Store) {
+	// get the starting number of message export entries
+	startTime := model.GetMillis()
+	var numMessageExports = 0
+	if r1 := <-ss.Compliance().MessageExport(startTime-10, 10); r1.Err != nil {
+		t.Fatal(r1.Err)
+	} else {
+		messages := r1.Data.([]*model.MessageExport)
+		numMessageExports = len(messages)
+	}
+
+	// need a team
+	team := &model.Team{
+		DisplayName: "DisplayName",
+		Name:        "zz" + model.NewId() + "b",
+		Email:       model.NewId() + "@nowhere.com",
+		Type:        model.TEAM_OPEN,
+	}
+	team = store.Must(ss.Team().Save(team)).(*model.Team)
+
+	// and two users that are a part of that team
+	user1 := &model.User{
+		Email: model.NewId(),
+	}
+	user1 = store.Must(ss.User().Save(user1)).(*model.User)
+	store.Must(ss.Team().SaveMember(&model.TeamMember{
+		TeamId: team.Id,
+		UserId: user1.Id,
+	}, -1))
+
+	user2 := &model.User{
+		Email: model.NewId(),
+	}
+	user2 = store.Must(ss.User().Save(user2)).(*model.User)
+	store.Must(ss.Team().SaveMember(&model.TeamMember{
+		TeamId: team.Id,
+		UserId: user2.Id,
+	}, -1))
+
+	// need a public channel as well as a DM channel between the two users
+	channel := &model.Channel{
+		TeamId:      team.Id,
+		Name:        model.NewId(),
+		DisplayName: "Channel2",
+		Type:        model.CHANNEL_OPEN,
+	}
+	channel = store.Must(ss.Channel().Save(channel, -1)).(*model.Channel)
+	directMessageChannel := store.Must(ss.Channel().CreateDirectChannel(user1.Id, user2.Id)).(*model.Channel)
+
+	// user1 posts twice in the public channel
+	post1 := &model.Post{
+		ChannelId: channel.Id,
+		UserId:    user1.Id,
+		CreateAt:  startTime,
+		Message:   "zz" + model.NewId() + "a",
+	}
+	post1 = store.Must(ss.Post().Save(post1)).(*model.Post)
+
+	post2 := &model.Post{
+		ChannelId: channel.Id,
+		UserId:    user1.Id,
+		CreateAt:  startTime + 10,
+		Message:   "zz" + model.NewId() + "b",
+	}
+	post2 = store.Must(ss.Post().Save(post2)).(*model.Post)
+
+	// user1 also sends a DM to user2
+	post3 := &model.Post{
+		ChannelId: directMessageChannel.Id,
+		UserId:    user1.Id,
+		CreateAt:  startTime + 20,
+		Message:   "zz" + model.NewId() + "c",
+	}
+	post3 = store.Must(ss.Post().Save(post3)).(*model.Post)
+
+	// fetch the message exports for all three posts that user1 sent
+	messageExportMap := map[string]model.MessageExport{}
+	if r1 := <-ss.Compliance().MessageExport(startTime-10, 10); r1.Err != nil {
+		t.Fatal(r1.Err)
+	} else {
+		messages := r1.Data.([]*model.MessageExport)
+		assert.Equal(t, numMessageExports+3, len(messages))
+
+		for _, v := range messages {
+			messageExportMap[*v.PostId] = *v
+		}
+	}
+
+	// post1 was made by user1 in channel1 and team1
+	assert.Equal(t, post1.Id, *messageExportMap[post1.Id].PostId)
+	assert.Equal(t, post1.CreateAt, *messageExportMap[post1.Id].PostCreateAt)
+	assert.Equal(t, post1.Message, *messageExportMap[post1.Id].PostMessage)
+	assert.Equal(t, channel.Id, *messageExportMap[post1.Id].ChannelId)
+	assert.Equal(t, channel.DisplayName, *messageExportMap[post1.Id].ChannelDisplayName)
+	assert.Equal(t, user1.Id, *messageExportMap[post1.Id].UserId)
+	assert.Equal(t, user1.Email, *messageExportMap[post1.Id].UserEmail)
+
+	// post2 was made by user1 in channel1 and team1
+	assert.Equal(t, post2.Id, *messageExportMap[post2.Id].PostId)
+	assert.Equal(t, post2.CreateAt, *messageExportMap[post2.Id].PostCreateAt)
+	assert.Equal(t, post2.Message, *messageExportMap[post2.Id].PostMessage)
+	assert.Equal(t, channel.Id, *messageExportMap[post2.Id].ChannelId)
+	assert.Equal(t, channel.DisplayName, *messageExportMap[post2.Id].ChannelDisplayName)
+	assert.Equal(t, user1.Id, *messageExportMap[post2.Id].UserId)
+	assert.Equal(t, user1.Email, *messageExportMap[post2.Id].UserEmail)
+
+	// post3 is a DM between user1 and user2
+	assert.Equal(t, post3.Id, *messageExportMap[post3.Id].PostId)
+	assert.Equal(t, post3.CreateAt, *messageExportMap[post3.Id].PostCreateAt)
+	assert.Equal(t, post3.Message, *messageExportMap[post3.Id].PostMessage)
+	assert.Equal(t, directMessageChannel.Id, *messageExportMap[post3.Id].ChannelId)
+	assert.Equal(t, user1.Id, *messageExportMap[post3.Id].UserId)
+	assert.Equal(t, user1.Email, *messageExportMap[post3.Id].UserEmail)
+}
diff --git a/store/storetest/mocks/ChannelMemberHistoryStore.go b/store/storetest/mocks/ChannelMemberHistoryStore.go
new file mode 100644
index 000000000..4ac0967f9
--- /dev/null
+++ b/store/storetest/mocks/ChannelMemberHistoryStore.go
@@ -0,0 +1,77 @@
+// Code generated by mockery v1.0.0
+
+// Regenerate this file using `make store-mocks`.
+
+package mocks
+
+import mock "github.com/stretchr/testify/mock"
+import store "github.com/mattermost/mattermost-server/store"
+
+// ChannelMemberHistoryStore is an autogenerated mock type for the ChannelMemberHistoryStore type
+type ChannelMemberHistoryStore struct {
+	mock.Mock
+}
+
+// GetUsersInChannelDuring provides a mock function with given fields: startTime, endTime, channelId
+func (_m *ChannelMemberHistoryStore) GetUsersInChannelDuring(startTime int64, endTime int64, channelId string) store.StoreChannel {
+	ret := _m.Called(startTime, endTime, channelId)
+
+	var r0 store.StoreChannel
+	if rf, ok := ret.Get(0).(func(int64, int64, string) store.StoreChannel); ok {
+		r0 = rf(startTime, endTime, channelId)
+	} else {
+		if ret.Get(0) != nil {
+			r0 = ret.Get(0).(store.StoreChannel)
+		}
+	}
+
+	return r0
+}
+
+// LogJoinEvent provides a mock function with given fields: userId, channelId, joinTime
+func (_m *ChannelMemberHistoryStore) LogJoinEvent(userId string, channelId string, joinTime int64) store.StoreChannel {
+	ret := _m.Called(userId, channelId, joinTime)
+
+	var r0 store.StoreChannel
+	if rf, ok := ret.Get(0).(func(string, string, int64) store.StoreChannel); ok {
+		r0 = rf(userId, channelId, joinTime)
+	} else {
+		if ret.Get(0) != nil {
+			r0 = ret.Get(0).(store.StoreChannel)
+		}
+	}
+
+	return r0
+}
+
+// LogLeaveEvent provides a mock function with given fields: userId, channelId, leaveTime
+func (_m *ChannelMemberHistoryStore) LogLeaveEvent(userId string, channelId string, leaveTime int64) store.StoreChannel {
+	ret := _m.Called(userId, channelId, leaveTime)
+
+	var r0 store.StoreChannel
+	if rf, ok := ret.Get(0).(func(string, string, int64) store.StoreChannel); ok {
+		r0 = rf(userId, channelId, leaveTime)
+	} else {
+		if ret.Get(0) != nil {
+			r0 = ret.Get(0).(store.StoreChannel)
+		}
+	}
+
+	return r0
+}
+
+// PurgeHistoryBefore provides a mock function with given fields: time, channelId
+func (_m *ChannelMemberHistoryStore) PurgeHistoryBefore(time int64, channelId string) store.StoreChannel {
+	ret := _m.Called(time, channelId)
+
+	var r0 store.StoreChannel
+	if rf, ok := ret.Get(0).(func(int64, string) store.StoreChannel); ok {
+		r0 = rf(time, channelId)
+	} else {
+		if ret.Get(0) != nil {
+			r0 = ret.Get(0).(store.StoreChannel)
+		}
+	}
+
+	return r0
+}
diff --git a/store/storetest/mocks/ComplianceStore.go b/store/storetest/mocks/ComplianceStore.go
index b2208ead7..fb828cd4b 100644
--- a/store/storetest/mocks/ComplianceStore.go
+++ b/store/storetest/mocks/ComplianceStore.go
@@ -61,6 +61,22 @@ func (_m *ComplianceStore) GetAll(offset int, limit int) store.StoreChannel {
 	return r0
 }
 
+// MessageExport provides a mock function with given fields: after, limit
+func (_m *ComplianceStore) MessageExport(after int64, limit int) store.StoreChannel {
+	ret := _m.Called(after, limit)
+
+	var r0 store.StoreChannel
+	if rf, ok := ret.Get(0).(func(int64, int) store.StoreChannel); ok {
+		r0 = rf(after, limit)
+	} else {
+		if ret.Get(0) != nil {
+			r0 = ret.Get(0).(store.StoreChannel)
+		}
+	}
+
+	return r0
+}
+
 // Save provides a mock function with given fields: compliance
 func (_m *ComplianceStore) Save(compliance *model.Compliance) store.StoreChannel {
 	ret := _m.Called(compliance)
diff --git a/store/storetest/mocks/LayeredStoreDatabaseLayer.go b/store/storetest/mocks/LayeredStoreDatabaseLayer.go
index c3b8bbb60..9c66c4aac 100644
--- a/store/storetest/mocks/LayeredStoreDatabaseLayer.go
+++ b/store/storetest/mocks/LayeredStoreDatabaseLayer.go
@@ -46,6 +46,22 @@ func (_m *LayeredStoreDatabaseLayer) Channel() store.ChannelStore {
 	return r0
 }
 
+// ChannelMemberHistory provides a mock function with given fields:
+func (_m *LayeredStoreDatabaseLayer) ChannelMemberHistory() store.ChannelMemberHistoryStore {
+	ret := _m.Called()
+
+	var r0 store.ChannelMemberHistoryStore
+	if rf, ok := ret.Get(0).(func() store.ChannelMemberHistoryStore); ok {
+		r0 = rf()
+	} else {
+		if ret.Get(0) != nil {
+			r0 = ret.Get(0).(store.ChannelMemberHistoryStore)
+		}
+	}
+
+	return r0
+}
+
 // Close provides a mock function with given fields:
 func (_m *LayeredStoreDatabaseLayer) Close() {
 	_m.Called()
diff --git a/store/storetest/mocks/Store.go b/store/storetest/mocks/Store.go
index 85ed10d35..40b50a554 100644
--- a/store/storetest/mocks/Store.go
+++ b/store/storetest/mocks/Store.go
@@ -44,6 +44,22 @@ func (_m *Store) Channel() store.ChannelStore {
 	return r0
 }
 
+// ChannelMemberHistory provides a mock function with given fields:
+func (_m *Store) ChannelMemberHistory() store.ChannelMemberHistoryStore {
+	ret := _m.Called()
+
+	var r0 store.ChannelMemberHistoryStore
+	if rf, ok := ret.Get(0).(func() store.ChannelMemberHistoryStore); ok {
+		r0 = rf()
+	} else {
+		if ret.Get(0) != nil {
+			r0 = ret.Get(0).(store.ChannelMemberHistoryStore)
+		}
+	}
+
+	return r0
+}
+
 // Close provides a mock function with given fields:
 func (_m *Store) Close() {
 	_m.Called()
diff --git a/store/storetest/store.go b/store/storetest/store.go
index 55545decb..367c5f441 100644
--- a/store/storetest/store.go
+++ b/store/storetest/store.go
@@ -19,29 +19,30 @@ func NewStoreChannel(result store.StoreResult) store.StoreChannel {
 
 // Store can be used to provide mock stores for testing.
 type Store struct {
-	TeamStore             mocks.TeamStore
-	ChannelStore          mocks.ChannelStore
-	PostStore             mocks.PostStore
-	UserStore             mocks.UserStore
-	AuditStore            mocks.AuditStore
-	ClusterDiscoveryStore mocks.ClusterDiscoveryStore
-	ComplianceStore       mocks.ComplianceStore
-	SessionStore          mocks.SessionStore
-	OAuthStore            mocks.OAuthStore
-	SystemStore           mocks.SystemStore
-	WebhookStore          mocks.WebhookStore
-	CommandStore          mocks.CommandStore
-	CommandWebhookStore   mocks.CommandWebhookStore
-	PreferenceStore       mocks.PreferenceStore
-	LicenseStore          mocks.LicenseStore
-	TokenStore            mocks.TokenStore
-	EmojiStore            mocks.EmojiStore
-	StatusStore           mocks.StatusStore
-	FileInfoStore         mocks.FileInfoStore
-	ReactionStore         mocks.ReactionStore
-	JobStore              mocks.JobStore
-	UserAccessTokenStore  mocks.UserAccessTokenStore
-	PluginStore           mocks.PluginStore
+	TeamStore                 mocks.TeamStore
+	ChannelStore              mocks.ChannelStore
+	PostStore                 mocks.PostStore
+	UserStore                 mocks.UserStore
+	AuditStore                mocks.AuditStore
+	ClusterDiscoveryStore     mocks.ClusterDiscoveryStore
+	ComplianceStore           mocks.ComplianceStore
+	SessionStore              mocks.SessionStore
+	OAuthStore                mocks.OAuthStore
+	SystemStore               mocks.SystemStore
+	WebhookStore              mocks.WebhookStore
+	CommandStore              mocks.CommandStore
+	CommandWebhookStore       mocks.CommandWebhookStore
+	PreferenceStore           mocks.PreferenceStore
+	LicenseStore              mocks.LicenseStore
+	TokenStore                mocks.TokenStore
+	EmojiStore                mocks.EmojiStore
+	StatusStore               mocks.StatusStore
+	FileInfoStore             mocks.FileInfoStore
+	ReactionStore             mocks.ReactionStore
+	JobStore                  mocks.JobStore
+	UserAccessTokenStore      mocks.UserAccessTokenStore
+	PluginStore               mocks.PluginStore
+	ChannelMemberHistoryStore mocks.ChannelMemberHistoryStore
 }
 
 func (s *Store) Team() store.TeamStore                         { return &s.TeamStore }
@@ -67,12 +68,15 @@ func (s *Store) Reaction() store.ReactionStore                 { return &s.React
 func (s *Store) Job() store.JobStore                           { return &s.JobStore }
 func (s *Store) UserAccessToken() store.UserAccessTokenStore   { return &s.UserAccessTokenStore }
 func (s *Store) Plugin() store.PluginStore                     { return &s.PluginStore }
-func (s *Store) MarkSystemRanUnitTests()                       { /* do nothing */ }
-func (s *Store) Close()                                        { /* do nothing */ }
-func (s *Store) DropAllTables()                                { /* do nothing */ }
-func (s *Store) TotalMasterDbConnections() int                 { return 1 }
-func (s *Store) TotalReadDbConnections() int                   { return 1 }
-func (s *Store) TotalSearchDbConnections() int                 { return 1 }
+func (s *Store) ChannelMemberHistory() store.ChannelMemberHistoryStore {
+	return &s.ChannelMemberHistoryStore
+}
+func (s *Store) MarkSystemRanUnitTests()       { /* do nothing */ }
+func (s *Store) Close()                        { /* do nothing */ }
+func (s *Store) DropAllTables()                { /* do nothing */ }
+func (s *Store) TotalMasterDbConnections() int { return 1 }
+func (s *Store) TotalReadDbConnections() int   { return 1 }
+func (s *Store) TotalSearchDbConnections() int { return 1 }
 
 func (s *Store) AssertExpectations(t mock.TestingT) bool {
 	return mock.AssertExpectationsForObjects(t,
@@ -98,6 +102,7 @@ func (s *Store) AssertExpectations(t mock.TestingT) bool {
 		&s.ReactionStore,
 		&s.JobStore,
 		&s.UserAccessTokenStore,
+		&s.ChannelMemberHistoryStore,
 		&s.PluginStore,
 	)
 }