bridge/slack/users_channels.go
package bslack
import (
"context"
"fmt"
"strings"
"sync"
"time"
"github.com/42wim/matterbridge/bridge/config"
"github.com/sirupsen/logrus"
"github.com/slack-go/slack"
)
const minimumRefreshInterval = 10 * time.Second
type users struct {
log *logrus.Entry
sc *slack.Client
users map[string]*slack.User
usersMutex sync.RWMutex
usersSyncPoints map[string]chan struct{}
refreshInProgress bool
earliestRefresh time.Time
refreshMutex sync.Mutex
}
func newUserManager(log *logrus.Entry, sc *slack.Client) *users {
return &users{
log: log,
sc: sc,
users: make(map[string]*slack.User),
usersSyncPoints: make(map[string]chan struct{}),
earliestRefresh: time.Now(),
}
}
func (b *users) getUser(id string) *slack.User {
b.usersMutex.RLock()
user, ok := b.users[id]
b.usersMutex.RUnlock()
if ok {
return user
}
b.populateUser(id)
b.usersMutex.RLock()
defer b.usersMutex.RUnlock()
return b.users[id]
}
func (b *users) getUsername(id string) string {
if user := b.getUser(id); user != nil {
if user.Profile.DisplayName != "" {
return user.Profile.DisplayName
}
return user.Name
}
b.log.Warnf("Could not find user with ID '%s'", id)
return ""
}
func (b *users) getAvatar(id string) string {
if user := b.getUser(id); user != nil {
return user.Profile.Image48
}
return ""
}
func (b *users) populateUser(userID string) {
for {
b.usersMutex.Lock()
_, exists := b.users[userID]
if exists {
// already in cache
b.usersMutex.Unlock()
return
}
if syncPoint, ok := b.usersSyncPoints[userID]; ok {
// Another goroutine is already populating this user for us so wait on it to finish.
b.usersMutex.Unlock()
<-syncPoint
// We do not return and iterate again to check that the entry does indeed exist
// in case the previous query failed for some reason.
} else {
b.usersSyncPoints[userID] = make(chan struct{})
defer func() {
// Wake up any waiting goroutines and remove the synchronization point.
close(b.usersSyncPoints[userID])
delete(b.usersSyncPoints, userID)
}()
break
}
}
// Do not hold the lock while fetching information from Slack
// as this might take an unbounded amount of time.
b.usersMutex.Unlock()
user, err := b.sc.GetUserInfo(userID)
if err != nil {
b.log.Debugf("GetUserInfo failed for %v: %v", userID, err)
return
}
b.usersMutex.Lock()
defer b.usersMutex.Unlock()
// Register user information.
b.users[userID] = user
}
func (b *users) invalidateUser(userID string) {
b.usersMutex.Lock()
defer b.usersMutex.Unlock()
delete(b.users, userID)
}
func (b *users) populateUsers(wait bool) {
b.refreshMutex.Lock()
if !wait && (time.Now().Before(b.earliestRefresh) || b.refreshInProgress) {
b.log.Debugf("Not refreshing user list as it was done less than %v ago.", minimumRefreshInterval)
b.refreshMutex.Unlock()
return
}
for b.refreshInProgress {
b.refreshMutex.Unlock()
time.Sleep(time.Second)
b.refreshMutex.Lock()
}
b.refreshInProgress = true
b.refreshMutex.Unlock()
newUsers := map[string]*slack.User{}
pagination := b.sc.GetUsersPaginated(slack.GetUsersOptionLimit(200))
count := 0
for {
var err error
pagination, err = pagination.Next(context.Background())
time.Sleep(time.Second)
if err != nil {
if pagination.Done(err) {
break
}
if err = handleRateLimit(b.log, err); err != nil {
b.log.Errorf("Could not retrieve users: %#v", err)
return
}
continue
}
for i := range pagination.Users {
newUsers[pagination.Users[i].ID] = &pagination.Users[i]
}
b.log.Debugf("getting %d users", len(pagination.Users))
count++
// more > 2000 users, slack will complain and ratelimit. break
if count > 10 {
b.log.Info("Large slack detected > 2000 users, skipping loading complete userlist.")
break
}
}
b.usersMutex.Lock()
defer b.usersMutex.Unlock()
b.users = newUsers
b.refreshMutex.Lock()
defer b.refreshMutex.Unlock()
b.earliestRefresh = time.Now().Add(minimumRefreshInterval)
b.refreshInProgress = false
}
type channels struct {
log *logrus.Entry
sc *slack.Client
channelsByID map[string]*slack.Channel
channelsByName map[string]*slack.Channel
channelsMutex sync.RWMutex
channelMembers map[string][]string
channelMembersMutex sync.RWMutex
refreshInProgress bool
earliestRefresh time.Time
refreshMutex sync.Mutex
}
func newChannelManager(log *logrus.Entry, sc *slack.Client) *channels {
return &channels{
log: log,
sc: sc,
channelsByID: make(map[string]*slack.Channel),
channelsByName: make(map[string]*slack.Channel),
earliestRefresh: time.Now(),
}
}
func (b *channels) getChannel(channel string) (*slack.Channel, error) {
if strings.HasPrefix(channel, "ID:") {
return b.getChannelByID(strings.TrimPrefix(channel, "ID:"))
}
return b.getChannelByName(channel)
}
func (b *channels) getChannelByName(name string) (*slack.Channel, error) {
return b.getChannelBy(name, b.channelsByName)
}
func (b *channels) getChannelByID(id string) (*slack.Channel, error) {
return b.getChannelBy(id, b.channelsByID)
}
func (b *channels) getChannelBy(lookupKey string, lookupMap map[string]*slack.Channel) (*slack.Channel, error) {
b.channelsMutex.RLock()
defer b.channelsMutex.RUnlock()
if channel, ok := lookupMap[lookupKey]; ok {
return channel, nil
}
return nil, fmt.Errorf("channel %s not found", lookupKey)
}
func (b *channels) getChannelMembers(users *users) config.ChannelMembers {
b.channelMembersMutex.RLock()
defer b.channelMembersMutex.RUnlock()
membersInfo := config.ChannelMembers{}
for channelID, members := range b.channelMembers {
for _, member := range members {
channelName := ""
userName := ""
userNick := ""
user := users.getUser(member)
if user != nil {
userName = user.Name
userNick = user.Profile.DisplayName
}
channel, _ := b.getChannelByID(channelID)
if channel != nil {
channelName = channel.Name
}
memberInfo := config.ChannelMember{
Username: userName,
Nick: userNick,
UserID: member,
ChannelID: channelID,
ChannelName: channelName,
}
membersInfo = append(membersInfo, memberInfo)
}
}
return membersInfo
}
func (b *channels) registerChannel(channel slack.Channel) {
b.channelsMutex.Lock()
defer b.channelsMutex.Unlock()
b.channelsByID[channel.ID] = &channel
b.channelsByName[channel.Name] = &channel
}
func (b *channels) populateChannels(wait bool) {
b.refreshMutex.Lock()
if !wait && (time.Now().Before(b.earliestRefresh) || b.refreshInProgress) {
b.log.Debugf("Not refreshing channel list as it was done less than %v seconds ago.", minimumRefreshInterval)
b.refreshMutex.Unlock()
return
}
for b.refreshInProgress {
b.refreshMutex.Unlock()
time.Sleep(time.Second)
b.refreshMutex.Lock()
}
b.refreshInProgress = true
b.refreshMutex.Unlock()
newChannelsByID := map[string]*slack.Channel{}
newChannelsByName := map[string]*slack.Channel{}
newChannelMembers := make(map[string][]string)
// We only retrieve public and private channels, not IMs
// and MPIMs as those do not have a channel name.
queryParams := &slack.GetConversationsParameters{
ExcludeArchived: true,
Types: []string{"public_channel,private_channel"},
Limit: 1000,
}
for {
channels, nextCursor, err := b.sc.GetConversations(queryParams)
if err != nil {
if err = handleRateLimit(b.log, err); err != nil {
b.log.Errorf("Could not retrieve channels: %#v", err)
return
}
continue
}
for i := range channels {
newChannelsByID[channels[i].ID] = &channels[i]
newChannelsByName[channels[i].Name] = &channels[i]
// also find all the members in every channel
// comment for now, issues on big slacks
/*
members, err := b.getUsersInConversation(channels[i].ID)
if err != nil {
if err = b.handleRateLimit(err); err != nil {
b.Log.Errorf("Could not retrieve channel members: %#v", err)
return
}
continue
}
newChannelMembers[channels[i].ID] = members
*/
}
if nextCursor == "" {
break
}
queryParams.Cursor = nextCursor
}
b.channelsMutex.Lock()
defer b.channelsMutex.Unlock()
b.channelsByID = newChannelsByID
b.channelsByName = newChannelsByName
b.channelMembersMutex.Lock()
defer b.channelMembersMutex.Unlock()
b.channelMembers = newChannelMembers
b.refreshMutex.Lock()
defer b.refreshMutex.Unlock()
b.earliestRefresh = time.Now().Add(minimumRefreshInterval)
b.refreshInProgress = false
}