1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
|
// Copyright (c) 2017-present Mattermost, Inc. All Rights Reserved.
// See License.txt for license information.
package sqlstore
import (
"fmt"
"net/http"
"database/sql"
"github.com/mattermost/mattermost-server/mlog"
"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", nil, 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", params, err.Error(), http.StatusInternalServerError)
} else if rows, err := sqlResult.RowsAffected(); err == nil && rows != 1 {
// there was no join event to update - this is best effort, so no need to raise an error
mlog.Warn(fmt.Sprintf("Channel join event for user %v and channel %v not found", userId, channelId), mlog.String("user_id", userId))
}
})
}
func (s SqlChannelMemberHistoryStore) GetUsersInChannelDuring(startTime int64, endTime int64, channelId string) store.StoreChannel {
return store.Do(func(result *store.StoreResult) {
if useChannelMemberHistory, err := s.hasDataAtOrBefore(startTime); err != nil {
result.Err = model.NewAppError("SqlChannelMemberHistoryStore.GetUsersInChannelAt", "store.sql_channel_member_history.get_users_in_channel_during.app_error", nil, err.Error(), http.StatusInternalServerError)
} else if useChannelMemberHistory {
// the export period starts after the ChannelMemberHistory table was first introduced, so we can use the
// data from it for our export
if channelMemberHistories, err := s.getFromChannelMemberHistoryTable(startTime, endTime, channelId); err != nil {
result.Err = model.NewAppError("SqlChannelMemberHistoryStore.GetUsersInChannelAt", "store.sql_channel_member_history.get_users_in_channel_during.app_error", nil, err.Error(), http.StatusInternalServerError)
} else {
result.Data = channelMemberHistories
}
} else {
// the export period starts before the ChannelMemberHistory table was introduced, so we need to fake the
// data by assuming that anybody who has ever joined the channel in question was present during the export period.
// this may not always be true, but it's better than saying that somebody wasn't there when they were
if channelMemberHistories, err := s.getFromChannelMembersTable(startTime, endTime, channelId); err != nil {
result.Err = model.NewAppError("SqlChannelMemberHistoryStore.GetUsersInChannelAt", "store.sql_channel_member_history.get_users_in_channel_during.app_error", nil, err.Error(), http.StatusInternalServerError)
} else {
result.Data = channelMemberHistories
}
}
})
}
func (s SqlChannelMemberHistoryStore) hasDataAtOrBefore(time int64) (bool, error) {
type NullableCountResult struct {
Min sql.NullInt64
}
var result NullableCountResult
query := "SELECT MIN(JoinTime) AS Min FROM ChannelMemberHistory"
if err := s.GetReplica().SelectOne(&result, query); err != nil {
return false, err
} else if result.Min.Valid {
return result.Min.Int64 <= time, nil
} else {
// if the result was null, there are no rows in the table, so there is no data from before
return false, nil
}
}
func (s SqlChannelMemberHistoryStore) getFromChannelMemberHistoryTable(startTime int64, endTime int64, channelId string) ([]*model.ChannelMemberHistoryResult, error) {
query := `
SELECT
cmh.*,
u.Email,
u.Username
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.ChannelMemberHistoryResult
if _, err := s.GetReplica().Select(&histories, query, params); err != nil {
return nil, err
} else {
return histories, nil
}
}
func (s SqlChannelMemberHistoryStore) getFromChannelMembersTable(startTime int64, endTime int64, channelId string) ([]*model.ChannelMemberHistoryResult, error) {
query := `
SELECT DISTINCT
ch.ChannelId,
ch.UserId,
u.Email,
u.Username
FROM ChannelMembers AS ch
INNER JOIN Users AS u ON ch.UserId = u.id
WHERE ch.ChannelId = :ChannelId`
params := map[string]interface{}{"ChannelId": channelId}
var histories []*model.ChannelMemberHistoryResult
if _, err := s.GetReplica().Select(&histories, query, params); err != nil {
return nil, err
} else {
// we have to fill in the join/leave times, because that data doesn't exist in the channel members table
for _, channelMemberHistory := range histories {
channelMemberHistory.JoinTime = startTime
channelMemberHistory.LeaveTime = model.NewInt64(endTime)
}
return histories, nil
}
}
func (s SqlChannelMemberHistoryStore) PermanentDeleteBatch(endTime int64, limit int64) store.StoreChannel {
return store.Do(func(result *store.StoreResult) {
var query string
if s.DriverName() == model.DATABASE_DRIVER_POSTGRES {
query =
`DELETE FROM ChannelMemberHistory
WHERE ctid IN (
SELECT ctid FROM ChannelMemberHistory
WHERE LeaveTime IS NOT NULL
AND LeaveTime <= :EndTime
LIMIT :Limit
);`
} else {
query =
`DELETE FROM ChannelMemberHistory
WHERE LeaveTime IS NOT NULL
AND LeaveTime <= :EndTime
LIMIT :Limit`
}
params := map[string]interface{}{"EndTime": endTime, "Limit": limit}
if sqlResult, err := s.GetMaster().Exec(query, params); err != nil {
result.Err = model.NewAppError("SqlChannelMemberHistoryStore.PermanentDeleteBatchForChannel", "store.sql_channel_member_history.permanent_delete_batch.app_error", params, err.Error(), http.StatusInternalServerError)
} else {
if rowsAffected, err1 := sqlResult.RowsAffected(); err1 != nil {
result.Err = model.NewAppError("SqlChannelMemberHistoryStore.PermanentDeleteBatchForChannel", "store.sql_channel_member_history.permanent_delete_batch.app_error", params, err.Error(), http.StatusInternalServerError)
} else {
result.Data = rowsAffected
}
}
})
}
|
