diff options
author | Jonathan <jonfritz@gmail.com> | 2017-11-30 09:07:04 -0500 |
---|---|---|
committer | GitHub <noreply@github.com> | 2017-11-30 09:07:04 -0500 |
commit | 375c0632fab03e3fb54865e320585888499c076d (patch) | |
tree | ce6cba679337a82370d5a730c428bef21a9964bf /store/sqlstore | |
parent | d0d9ba4a7e43301697d1c9f495930e0c0179fdc3 (diff) | |
download | chat-375c0632fab03e3fb54865e320585888499c076d.tar.gz chat-375c0632fab03e3fb54865e320585888499c076d.tar.bz2 chat-375c0632fab03e3fb54865e320585888499c076d.zip |
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
Diffstat (limited to 'store/sqlstore')
-rw-r--r-- | store/sqlstore/channel_member_history_store.go | 102 | ||||
-rw-r--r-- | store/sqlstore/channel_member_history_store_test.go | 14 | ||||
-rw-r--r-- | store/sqlstore/compliance_store.go | 33 | ||||
-rw-r--r-- | store/sqlstore/supplier.go | 52 |
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 } |