diff --git a/bridge/mattermost/mattermost.go b/bridge/mattermost/mattermost.go index 6714fcb2..97a3181a 100644 --- a/bridge/mattermost/mattermost.go +++ b/bridge/mattermost/mattermost.go @@ -530,34 +530,14 @@ func (m *Mattermost) GetChannelName(channelID string) string { } func (m *Mattermost) GetChannelUsers(channelID string) ([]*bridge.UserInfo, error) { - var ( - mmusersPaged []*model.User - err error - resp *model.Response - ) - - idx := 0 - const batchSize = 200 - users := make([]*bridge.UserInfo, 0, batchSize) - - for { - mmusersPaged, resp, err = m.mc.Client.GetUsersInChannel(channelID, idx, batchSize, "") - if err != nil { - if rlErr := m.mc.HandleRatelimit("GetUsersInChannel", resp); rlErr != nil { - return nil, rlErr - } - continue - } - - for _, mmuser := range mmusersPaged { - users = append(users, m.createUser(mmuser)) - } - - if len(mmusersPaged) < batchSize { - break - } + mmUsers, err := m.mc.GetChannelUsers(channelID) + if err != nil { + return nil, err + } - idx++ + users := make([]*bridge.UserInfo, 0, len(mmUsers)) + for _, mmuser := range mmUsers { + users = append(users, m.createUser(mmuser)) } return users, nil diff --git a/vendor/github.com/matterbridge/matterclient/channels.go b/vendor/github.com/matterbridge/matterclient/channels.go index 86bd5ee5..a881874d 100644 --- a/vendor/github.com/matterbridge/matterclient/channels.go +++ b/vendor/github.com/matterbridge/matterclient/channels.go @@ -150,6 +150,56 @@ func (m *Client) GetChannelTeamID(id string) string { return "" } +func (m *Client) GetChannelUsers(channelID string) ([]*model.User, error) { + m.Users.mu.RLock() + if userIDs, exists := m.Users.channels[channelID]; exists { + users := make([]*model.User, 0, len(userIDs)) + for uid := range userIDs { + if user, ok := m.Users.users[uid]; ok { + users = append(users, user) + } + } + m.Users.mu.RUnlock() + return users, nil + } + m.Users.mu.RUnlock() + + var allUsers []*model.User + idx := 0 + const batchSize = 200 + + for { + mmusersPaged, resp, err := m.Client.GetUsersInChannel(channelID, idx, batchSize, "") + if err != nil { + if rlErr := m.HandleRatelimit("GetUsersInChannel", resp); rlErr != nil { + return nil, rlErr + } + continue + } + + allUsers = append(allUsers, mmusersPaged...) + + if len(mmusersPaged) < batchSize { + break + } + idx++ + } + + m.Users.mu.Lock() + defer m.Users.mu.Unlock() + + if m.Users.channels[channelID] == nil { + m.Users.channels[channelID] = make(map[string]struct{}) + } + + for _, u := range allUsers { + m.Users.users[u.Id] = u + m.Users.channels[channelID][u.Id] = struct{}{} + } + + return allUsers, nil +} + func (m *Client) GetLastViewedAt(channelID string) int64 { m.RLock() defer m.RUnlock() @@ -326,6 +376,28 @@ func (m *Client) UpdateChannelHeader(channelID string, header string) { } } +func (m *Client) UpdateChannelUsersCache(channelID string, user *model.User) { + m.Users.mu.Lock() + defer m.Users.mu.Unlock() + + m.Users.users[user.Id] = user + + if channelID != "" { + if m.Users.channels[channelID] != nil { + m.Users.channels[channelID][user.Id] = struct{}{} + } + } +} + +func (m *Client) UpdateChannelUsersCacheRemove(channelID string, userID string) { + m.Users.mu.Lock() + defer m.Users.mu.Unlock() + + if m.Users.channels != nil && m.Users.channels[channelID] != nil { + delete(m.Users.channels[channelID], userID) + } +} + func (m *Client) UpdateLastViewed(channelID string) error { m.logger.Debugf("posting lastview %#v", channelID) diff --git a/vendor/github.com/matterbridge/matterclient/matterclient.go b/vendor/github.com/matterbridge/matterclient/matterclient.go index 74fb7c35..ff58b546 100644 --- a/vendor/github.com/matterbridge/matterclient/matterclient.go +++ b/vendor/github.com/matterbridge/matterclient/matterclient.go @@ -3,6 +3,7 @@ package matterclient import ( "context" "crypto/tls" + "encoding/json" "errors" "fmt" "net" @@ -36,12 +37,21 @@ type Credentials struct { MFAToken string } +type UsersCache struct { + mu sync.RWMutex + users map[string]*model.User + channels map[string]map[string]struct{} + teams map[string]map[string]struct{} + statuses map[string]string + + lastUpdated atomic.Int64 +} + type Team struct { Team *model.Team ID string Channels []*model.Channel MoreChannels []*model.Channel - Users map[string]*model.User LastUserSync time.Time LastChannelSync time.Time @@ -66,7 +76,7 @@ type Client struct { OtherTeams map[string]*Team Client *model.Client4 User *model.User - Users map[string]*model.User + Users *UsersCache MessageChan chan *Message WsClient *model.WebSocketClient AntiIdle bool @@ -86,9 +96,6 @@ type Client struct { lastWsActivity atomic.Int64 connectedAt atomic.Int64 - - UserStatusMutex sync.RWMutex - UserStatuses map[string]string } var Matterircd bool @@ -114,8 +121,12 @@ func New(login string, pass string, team string, server string, mfatoken string) return &Client{ Credentials: cred, MessageChan: make(chan *Message, 100), - Users: make(map[string]*model.User), - UserStatuses: make(map[string]string), + Users: &UsersCache{ + users: make(map[string]*model.User), + channels: make(map[string]map[string]struct{}), + teams: make(map[string]map[string]struct{}), + statuses: make(map[string]string), + }, rootLogger: rootLogger, lruCache: cache, logger: rootLogger.WithFields(logrus.Fields{"prefix": "matterclient"}), @@ -131,6 +142,23 @@ func (m *Client) Login() error { firstConnection = false } + if !firstConnection { + lastUpdatedUnix := m.Users.lastUpdated.Load() + timeOffline := time.Since(time.Unix(lastUpdatedUnix, 0)) + + if timeOffline > 15*time.Minute { + m.logger.Info("reconnect: flushing channel user cache to ensure state consistency") + + m.Users.mu.Lock() + m.Users.channels = make(map[string]map[string]struct{}) + m.Users.mu.Unlock() + + m.Users.lastUpdated.Store(time.Now().Unix()) + } else { + m.logger.Debugf("reconnect: preserving channel user cache (offline for only %v)", timeOffline.Round(time.Second)) + } + } + m.WsConnected = false if m.WsQuit { return nil @@ -355,28 +383,27 @@ func (m *Client) initUser() error { m.Unlock() if exists && time.Since(existingTeam.LastUserSync) < 15*time.Minute { - m.logger.Debugf("skipping user fetch for team %s: cache is only %v old", team.Name, time.Since(existingTeam.LastUserSync).Round(time.Second)) - m.Lock() - if team.Name == m.Credentials.Team { - m.Team = existingTeam - } - m.Unlock() - continue - } + m.logger.Debugf("skipping user fetch for team %s: cache is only %v old", team.Name, time.Since(existingTeam.LastUserSync).Round(time.Second)) + m.Lock() + if team.Name == m.Credentials.Team { + m.Team = existingTeam + } + m.Unlock() + continue + } m.logger.Debugf("fetching users for team %s (cache expired or missing)", team.Name) idx := 0 - usermap := make(map[string]*model.User) + var teamUsers []*model.User + for { mmusers, _, err := m.Client.GetUsersInTeam(team.Id, idx, batchSize, "") if err != nil { return err } - for _, user := range mmusers { - usermap[user.Id] = user - } + teamUsers = append(teamUsers, mmusers...) if len(mmusers) < batchSize { break @@ -386,22 +413,31 @@ func (m *Client) initUser() error { time.Sleep(time.Millisecond * 200) } - m.logger.Debugf("found %d users in team %s", len(usermap), team.Name) + m.logger.Debugf("found %d users in team %s", len(teamUsers), team.Name) + + m.Users.mu.Lock() + if m.Users.teams == nil { + m.Users.teams = make(map[string]map[string]struct{}) + } + if m.Users.teams[team.Id] == nil { + m.Users.teams[team.Id] = make(map[string]struct{}) + } + + for _, u := range teamUsers { + m.Users.users[u.Id] = u + m.Users.teams[team.Id][u.Id] = struct{}{} + m.Users.lastUpdated.Store(time.Now().Unix()) + } + m.Users.mu.Unlock() t := &Team{ Team: team, - Users: usermap, ID: team.Id, LastUserSync: time.Now(), } m.Lock() - // add all users - for k, v := range usermap { - m.Users[k] = v - } - m.OtherTeams[team.Id] = t if team.Name == m.Credentials.Team { @@ -704,11 +740,7 @@ func (m *Client) WsReceiver(ctx context.Context) { m.lastWsActivity.Store(time.Now().Unix()) - eventType := event.EventType() - - if eventType == model.WebsocketEventNewUser || eventType == model.WebsocketEventUserUpdated { - go m.syncSingleUser(event) - } + go m.maintainUsersCache(event) msg := &Message{ Raw: event, @@ -852,35 +884,105 @@ func (m *Client) antiIdle(ctx context.Context, channelID string, interval int) { } } +func (m *Client) UpdateTeamUsersCache(teamID string, user *model.User) { + m.Users.mu.Lock() + defer m.Users.mu.Unlock() + + m.Users.users[user.Id] = user + + if teamID != "" { + if m.Users.teams == nil { + m.Users.teams = make(map[string]map[string]struct{}) + } + if m.Users.teams[teamID] == nil { + m.Users.teams[teamID] = make(map[string]struct{}) + } + + m.Users.teams[teamID][user.Id] = struct{}{} + } + + m.Users.lastUpdated.Store(time.Now().Unix()) +} + func (m *Client) syncSingleUser(event *model.WebSocketEvent) { - userID, ok := event.GetData()["user_id"].(string) - if !ok { - return - } - - user, _, err := m.Client.GetUser(userID, "") - if err != nil { - m.logger.Errorf("syncSingleUser failed to get user %s: %v", userID, err) - return - } - - m.logger.Debugf("dynamically caching updated/new user: %s", user.Username) - - m.Lock() - defer m.Unlock() - - if m.Users == nil { - m.Users = make(map[string]*model.User) - } - m.Users[user.Id] = user - - if teamID, hasTeam := event.GetData()["team_id"].(string); hasTeam { - if team, exists := m.OtherTeams[teamID]; exists { - if team.Users == nil { - team.Users = make(map[string]*model.User) - } - team.Users[user.Id] = user - m.logger.Debugf("added/updated user %s in map for team %s", user.Username, team.Team.Name) - } - } + userID, ok := event.GetData()["user_id"].(string) + if !ok { + return + } + + user, _, err := m.Client.GetUser(userID, "") + if err != nil { + m.logger.Errorf("syncSingleUser failed to get user %s: %v", userID, err) + return + } + + m.logger.Debugf("dynamically caching updated/new user: %s", user.Username) + + teamID, _ := event.GetData()["team_id"].(string) + + m.UpdateTeamUsersCache(teamID, user) +} + +func (m *Client) maintainUsersCache(event *model.WebSocketEvent) { + switch event.EventType() { + case model.WebsocketEventNewUser: + if userID, ok := event.GetData()["user_id"].(string); ok { + if user := m.GetUser(userID); user != nil { + if teamID, hasTeam := event.GetData()["team_id"].(string); hasTeam && teamID != "" { + m.UpdateTeamUsersCache(teamID, user) + } + } + } + + case model.WebsocketEventUserUpdated: + if userStr, ok := event.GetData()["user"].(string); ok { + user := &model.User{} + if err := json.Unmarshal([]byte(userStr), user); err == nil { + if teamID, hasTeam := event.GetData()["team_id"].(string); hasTeam && teamID != "" { + m.UpdateTeamUsersCache(teamID, user) + } else { + m.UpdateUser(user) + } + } + } + + case model.WebsocketEventUserAdded: + channelID := event.GetBroadcast().ChannelId + if userID, ok := event.GetData()["user_id"].(string); ok && channelID != "" { + if user := m.GetUser(userID); user != nil { + m.UpdateChannelUsersCache(channelID, user) + m.Users.lastUpdated.Store(time.Now().Unix()) + } + } + + case model.WebsocketEventUserRemoved: + channelID := event.GetBroadcast().ChannelId + if userID, ok := event.GetData()["user_id"].(string); ok && channelID != "" { + m.UpdateChannelUsersCacheRemove(channelID, userID) + m.Users.lastUpdated.Store(time.Now().Unix()) + } + + case model.WebsocketEventPosted: + channelID := event.GetBroadcast().ChannelId + if postStr, ok := event.GetData()["post"].(string); ok && channelID != "" { + post := &model.Post{} + if err := json.Unmarshal([]byte(postStr), post); err == nil { + + m.Users.mu.RLock() + _, channelIsCached := m.Users.channels[channelID] + _, userIsCached := m.Users.channels[channelID][post.UserId] + m.Users.mu.RUnlock() + + // If we are actively caching this channel, but the user isn't in our list, our cache is stale! + if channelIsCached && !userIsCached { + m.logger.Warnf("Unrecognized user %s spoke in %s. Invalidating channel cache.", post.UserId, channelID) + + m.Users.mu.Lock() + delete(m.Users.channels, channelID) + m.Users.mu.Unlock() + m.Users.lastUpdated.Store(time.Now().Unix()) + } + } + } + } } diff --git a/vendor/github.com/matterbridge/matterclient/messages.go b/vendor/github.com/matterbridge/matterclient/messages.go index 077d2935..37244a8c 100644 --- a/vendor/github.com/matterbridge/matterclient/messages.go +++ b/vendor/github.com/matterbridge/matterclient/messages.go @@ -288,7 +288,7 @@ func (m *Client) parseMessage(rmsg *Message) { m.parseActionPost(rmsg) case "user_updated": if user, ok := rmsg.Raw.GetData()["user"].(*model.User); ok { - m.UpdateUser(user.Id) + m.UpdateUser(user) } case "group_added": if err := m.UpdateChannels(); err != nil { diff --git a/vendor/github.com/matterbridge/matterclient/users.go b/vendor/github.com/matterbridge/matterclient/users.go index 7bc259c9..65d27598 100644 --- a/vendor/github.com/matterbridge/matterclient/users.go +++ b/vendor/github.com/matterbridge/matterclient/users.go @@ -1,6 +1,9 @@ package matterclient import ( + "fmt" + "time" + "github.com/mattermost/mattermost-server/v6/model" ) @@ -13,9 +16,9 @@ func (m *Client) GetNickName(userID string) string { } func (m *Client) GetStatus(userID string) string { - m.UserStatusMutex.RLock() - status, ok := m.UserStatuses[userID] - m.UserStatusMutex.RUnlock() + m.Users.mu.RLock() + status, ok := m.Users.statuses[userID] + m.Users.mu.RUnlock() if ok { return status } @@ -29,24 +32,25 @@ func (m *Client) GetStatus(userID string) string { } func (m *Client) GetStatuses() map[string]string { - statuses := make(map[string]string, len(m.Users)) + statuses := make(map[string]string, len(m.Users.users)) var missingIDs []string - m.UserStatusMutex.RLock() - for id := range m.Users { - if status, ok := m.UserStatuses[id]; ok { + m.Users.mu.RLock() + for id := range m.Users.users { + if status, ok := m.Users.statuses[id]; ok { statuses[id] = status } else { missingIDs = append(missingIDs, id) } } - m.UserStatusMutex.RUnlock() + m.Users.mu.RUnlock() if len(missingIDs) == 0 { return statuses } const batchSize = 5000 + for i := 0; i < len(missingIDs); i += batchSize { end := i + batchSize if end > len(missingIDs) { @@ -92,20 +96,23 @@ func (m *Client) GetTeamName(teamID string) string { } func (m *Client) GetUser(userID string) *model.User { - m.Lock() - defer m.Unlock() + m.Users.mu.RLock() + user, exists := m.Users.users[userID] + m.Users.mu.RUnlock() - _, ok := m.Users[userID] - if !ok { - res, _, err := m.Client.GetUser(userID, "") - if err != nil { - return nil - } + if exists { + return user + } - m.Users[userID] = res + res, _, err := m.Client.GetUser(userID, "") + if err != nil { + m.logger.Debugf("GetUser failed to fetch missing user %s: %s", userID, err) + return nil } - return m.Users[userID] + m.UpdateUser(res) + + return res } func (m *Client) GetUserName(userID string) string { @@ -117,12 +124,12 @@ func (m *Client) GetUserName(userID string) string { } func (m *Client) GetUsers() map[string]*model.User { - users := make(map[string]*model.User) + users := make(map[string]*model.User, len(m.Users.users)) - m.RLock() - defer m.RUnlock() + m.Users.mu.RLock() + defer m.Users.mu.RUnlock() - for k, v := range m.Users { + for k, v := range m.Users.users { users[k] = v } @@ -138,12 +145,11 @@ func (m *Client) SetUserStatus(userID string, rawStatus string) string { statusStr = "away" } - m.UserStatusMutex.Lock() - if m.UserStatuses == nil { - m.UserStatuses = make(map[string]string) - } - m.UserStatuses[userID] = statusStr - m.UserStatusMutex.Unlock() + m.Users.mu.Lock() + defer m.Users.mu.Unlock() + + m.Users.statuses[userID] = statusStr + m.Users.lastUpdated.Store(time.Now().Unix()) return statusStr } @@ -167,11 +173,12 @@ func (m *Client) UpdateUsers() error { continue } - m.Lock() + m.Users.mu.Lock() for _, user := range mmusers { - m.Users[user.Id] = user + m.Users.users[user.Id] = user + m.Users.lastUpdated.Store(time.Now().Unix()) } - m.Unlock() + m.Users.mu.Unlock() if len(mmusers) < batchSize { break @@ -184,14 +191,25 @@ func (m *Client) UpdateUsers() error { } func (m *Client) UpdateUserNick(nick string) error { - user := m.User - user.Nickname = nick + m.RLock() + if m.User == nil { + m.RUnlock() + return fmt.Errorf("current user profile is not loaded") + } + userClone := *m.User + m.RUnlock() + userClone.Nickname = nick - _, _, err := m.Client.UpdateUser(user) + updatedUser, _, err := m.Client.UpdateUser(&userClone) if err != nil { return err } + m.Lock() + m.User = updatedUser + m.Unlock() + m.UpdateUser(updatedUser) + return nil } @@ -219,17 +237,15 @@ func (m *Client) UpdateStatus(userID string, status string) error { return err } + m.SetUserStatus(userID, status) + return nil } -func (m *Client) UpdateUser(userID string) { - m.Lock() - defer m.Unlock() - - res, _, err := m.Client.GetUser(userID, "") - if err != nil { - return - } +func (m *Client) UpdateUser(user *model.User) { + m.Users.mu.Lock() + defer m.Users.mu.Unlock() - m.Users[userID] = res + m.Users.users[user.Id] = user + m.Users.lastUpdated.Store(time.Now().Unix()) }