From a6d4e2abc64789a013373267d3a3f08a2f7c4b59 Mon Sep 17 00:00:00 2001 From: Rustam Gilyazov <16064414+rusq@users.noreply.github.com> Date: Sat, 20 May 2023 09:36:50 +1000 Subject: [PATCH 1/2] fetch channel members --- .gitignore | 1 + channels.go | 27 +++++++++++++ clienter_mock_test.go | 16 ++++++++ export/export.go | 50 +++++++++++++++++++++++- export/future.go | 3 ++ export/mock_dumper_test.go | 15 +++++++ go.mod | 1 + go.sum | 1 + internal/mocks/mock_fsadapter/mock_fs.go | 2 +- internal/network/limiter.go | 2 +- options.go | 10 ++++- slackdump.go | 1 + slackdump_test.go | 11 +++--- 13 files changed, 128 insertions(+), 12 deletions(-) diff --git a/.gitignore b/.gitignore index d300d292..6a9630e8 100644 --- a/.gitignore +++ b/.gitignore @@ -52,3 +52,4 @@ dist/ # sundry junk used for testing and other fuckery /tmp *.dot +*.gz diff --git a/channels.go b/channels.go index 7feda839..95a7ef64 100644 --- a/channels.go +++ b/channels.go @@ -98,3 +98,30 @@ func (sd *Session) getChannels(ctx context.Context, chanTypes []string, cb func( } return nil } + +// GetChannelMembers returns a list of all members in a channel. +func (sd *Session) GetChannelMembers(ctx context.Context, channelID string) ([]string, error) { + var ids []string + var cursor string + for { + var uu []string + var next string + if err := network.WithRetry(ctx, sd.limiter(network.Tier4), sd.options.Tier4Retries, func() error { + var err error + uu, next, err = sd.Client().GetUsersInConversationContext(ctx, &slack.GetUsersInConversationParameters{ + ChannelID: channelID, + Cursor: cursor, + }) + return err + }); err != nil { + return nil, err + } + ids = append(ids, uu...) + + if next == "" { + break + } + cursor = next + } + return ids, nil +} diff --git a/clienter_mock_test.go b/clienter_mock_test.go index 5ea85bce..32da7a8f 100644 --- a/clienter_mock_test.go +++ b/clienter_mock_test.go @@ -162,3 +162,19 @@ func (mr *mockClienterMockRecorder) GetUsersContext(ctx interface{}, options ... varargs := append([]interface{}{ctx}, options...) return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetUsersContext", reflect.TypeOf((*mockClienter)(nil).GetUsersContext), varargs...) } + +// GetUsersInConversationContext mocks base method. +func (m *mockClienter) GetUsersInConversationContext(ctx context.Context, params *slack.GetUsersInConversationParameters) ([]string, string, error) { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "GetUsersInConversationContext", ctx, params) + ret0, _ := ret[0].([]string) + ret1, _ := ret[1].(string) + ret2, _ := ret[2].(error) + return ret0, ret1, ret2 +} + +// GetUsersInConversationContext indicates an expected call of GetUsersInConversationContext. +func (mr *mockClienterMockRecorder) GetUsersInConversationContext(ctx, params interface{}) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetUsersInConversationContext", reflect.TypeOf((*mockClienter)(nil).GetUsersInConversationContext), ctx, params) +} diff --git a/export/export.go b/export/export.go index 458cebbb..dca9d970 100644 --- a/export/export.go +++ b/export/export.go @@ -11,6 +11,7 @@ import ( "github.com/rusq/slackdump/v2/fsadapter" "github.com/slack-go/slack" + "golang.org/x/sync/errgroup" "github.com/rusq/slackdump/v2" "github.com/rusq/slackdump/v2/internal/network" @@ -128,10 +129,34 @@ func (se *Export) exclusiveExport(ctx context.Context, uidx structures.UserIndex se.lg.Printf("skipping: %s", ch.ID) return nil } - if err := se.exportConversation(ctx, uidx, ch); err != nil { + + var eg errgroup.Group + + // 1. get members + var members []string + eg.Go(func() error { + var err error + members, err = se.sd.GetChannelMembers(ctx, ch.ID) + if err != nil { + return fmt.Errorf("error getting info for %s: %w", ch.ID, err) + } + return nil + }) + + // 2. export conversation + eg.Go(func() error { + if err := se.exportConversation(ctx, uidx, ch); err != nil { + return fmt.Errorf("error exporting conversation %s: %w", ch.ID, err) + } + return nil + }) + + // wait for both to finish + if err := eg.Wait(); err != nil { return err } + ch.Members = members chans = append(chans, ch) return nil @@ -174,10 +199,31 @@ func (se *Export) inclusiveExport(ctx context.Context, uidx structures.UserIndex return nil, fmt.Errorf("error getting info for %s: %w", sl, err) } - if err := se.exportConversation(ctx, uidx, *ch); err != nil { + var eg errgroup.Group + + var members []string + eg.Go(func() error { + var err error + members, err = se.sd.GetChannelMembers(ctx, ch.ID) + if err != nil { + return fmt.Errorf("error getting members for %s: %w", sl, err) + } + return nil + }) + + eg.Go(func() error { + if err := se.exportConversation(ctx, uidx, *ch); err != nil { + return fmt.Errorf("error exporting convesation %s: %w", ch.ID, err) + } + return nil + }) + + if err := eg.Wait(); err != nil { return nil, err } + ch.Members = members + chans = append(chans, *ch) } diff --git a/export/future.go b/export/future.go index 56e771e5..7a7fd30d 100644 --- a/export/future.go +++ b/export/future.go @@ -32,4 +32,7 @@ type dumper interface { // DumpRaw gets data from the Slack API and returns a Conversation object. DumpRaw(ctx context.Context, link string, oldest time.Time, latest time.Time, processFn ...slackdump.ProcessFunc) (*types.Conversation, error) + + // GetChannelMembers gets the list of members for a channel. + GetChannelMembers(ctx context.Context, channelID string) ([]string, error) } diff --git a/export/mock_dumper_test.go b/export/mock_dumper_test.go index 6d61fd0e..2938ae77 100644 --- a/export/mock_dumper_test.go +++ b/export/mock_dumper_test.go @@ -86,6 +86,21 @@ func (mr *MockdumperMockRecorder) DumpRaw(ctx, link, oldest, latest interface{}, return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "DumpRaw", reflect.TypeOf((*Mockdumper)(nil).DumpRaw), varargs...) } +// GetChannelMembers mocks base method. +func (m *Mockdumper) GetChannelMembers(ctx context.Context, channelID string) ([]string, error) { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "GetChannelMembers", ctx, channelID) + ret0, _ := ret[0].([]string) + ret1, _ := ret[1].(error) + return ret0, ret1 +} + +// GetChannelMembers indicates an expected call of GetChannelMembers. +func (mr *MockdumperMockRecorder) GetChannelMembers(ctx, channelID interface{}) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetChannelMembers", reflect.TypeOf((*Mockdumper)(nil).GetChannelMembers), ctx, channelID) +} + // GetUsers mocks base method. func (m *Mockdumper) GetUsers(ctx context.Context) (types.Users, error) { m.ctrl.T.Helper() diff --git a/go.mod b/go.mod index 785e0be0..3f905f72 100644 --- a/go.mod +++ b/go.mod @@ -18,6 +18,7 @@ require ( github.com/schollz/progressbar/v3 v3.13.0 github.com/slack-go/slack v0.12.1 github.com/stretchr/testify v1.8.2 + golang.org/x/sync v0.0.0-20210220032951-036812b2e83c golang.org/x/time v0.3.0 ) diff --git a/go.sum b/go.sum index a1a741dc..62d8c021 100644 --- a/go.sum +++ b/go.sum @@ -94,6 +94,7 @@ golang.org/x/net v0.0.0-20210405180319-a5a99cb37ef4/go.mod h1:p54w0d4576C0XHj96b golang.org/x/net v0.8.0 h1:Zrh2ngAOFYneWTAIAPethzeaQLuHwhuBkuV6ZiRnUaQ= golang.org/x/net v0.8.0/go.mod h1:QVkue5JL9kW//ek3r6jTKnTFis1tRmNAW2P1shuFdJc= golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= +golang.org/x/sync v0.0.0-20210220032951-036812b2e83c h1:5KslGYwFpkhGh+Q16bwMP3cOontH8FOep7tGV86Y7SQ= golang.org/x/sync v0.0.0-20210220032951-036812b2e83c/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= golang.org/x/sys v0.0.0-20190222072716-a9d3bda3a223/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= diff --git a/internal/mocks/mock_fsadapter/mock_fs.go b/internal/mocks/mock_fsadapter/mock_fs.go index 59aadb34..1c5889c9 100644 --- a/internal/mocks/mock_fsadapter/mock_fs.go +++ b/internal/mocks/mock_fsadapter/mock_fs.go @@ -1,5 +1,5 @@ // Code generated by MockGen. DO NOT EDIT. -// Source: github.com/rusq/fsadapter (interfaces: FS) +// Source: github.com/rusq/slackdump/v2/fsadapter (interfaces: FS) // Package mock_fsadapter is a generated GoMock package. package mock_fsadapter diff --git a/internal/network/limiter.go b/internal/network/limiter.go index 27511b2c..0c3b4325 100644 --- a/internal/network/limiter.go +++ b/internal/network/limiter.go @@ -13,7 +13,7 @@ const ( // Tier1 Tier = 1 Tier2 Tier = 20 Tier3 Tier = 50 - // Tier4 Tier = 100 + Tier4 Tier = 100 // secPerMin is the number of seconds in a minute, it is here to allow easy // modification of the program, should this value change. diff --git a/options.go b/options.go index b2132426..98eedfa9 100644 --- a/options.go +++ b/options.go @@ -22,6 +22,9 @@ type Options struct { Tier3Boost uint // Tier-3 limiter boost allows to increase or decrease the slack Tier req/min rate. Affects all tiers. Tier3Burst uint // Tier-3 limiter burst allows to set the limiter burst in req/sec. Default of 1 is safe. Tier3Retries int // number of retries to do when getting 429 on conversation fetch + Tier4Boost uint // Tier-4 limiter boost allows to increase or decrease the slack Tier req/min rate. Affects all tiers. + Tier4Burst uint // Tier-4 limiter burst allows to set the limiter burst in req/sec. Default of 1 is safe. + Tier4Retries int // number of retries to do when getting 429 on conversation fetch ConversationsPerReq int // number of messages we get per 1 API request. bigger the number, less requests, but they become more beefy. ChannelsPerReq int // number of channels to fetch per 1 API request. RepliesPerReq int // number of thread replies per request (slack default: 1000) @@ -43,6 +46,9 @@ var DefOptions = Options{ Tier3Boost: 120, // playing safe there, but generally value of 120 is fine. Tier3Burst: 1, // safe value, who would ever want to modify it? I don't know. Tier3Retries: 3, // on Tier 3 this was never a problem, even with limiter-boost=120 + Tier4Boost: 1, + Tier4Burst: 1, + Tier4Retries: 3, ConversationsPerReq: 200, // this is the recommended value by Slack. But who listens to them anyway. ChannelsPerReq: 100, // channels are Tier2 rate limited. Slack is greedy and never returns more than 100 per call. RepliesPerReq: 200, // the API-default is 1000 (see conversations.replies), but on large threads it may fail (see #54) @@ -86,7 +92,7 @@ func RetryDownloads(attempts int) Option { // base slack Tier limits. The resulting // events per minute will be calculated like this: // -// events_per_sec = ( + ) / 60.0 +// events_per_sec = ( + ) / 60.0 func Tier3Boost(eventsPerMin uint) Option { return func(options *Options) { options.Tier3Boost = eventsPerMin @@ -104,7 +110,7 @@ func Tier3Burst(eventsPerSec uint) Option { // base slack Tier limits. The resulting // events per minute will be calculated like this: // -// events_per_sec = ( + ) / 60.0 +// events_per_sec = ( + ) / 60.0 func Tier2Boost(eventsPerMin uint) Option { return func(options *Options) { options.Tier2Boost = eventsPerMin diff --git a/slackdump.go b/slackdump.go index 3ec4d5b4..8cb316df 100644 --- a/slackdump.go +++ b/slackdump.go @@ -53,6 +53,7 @@ type clienter interface { GetTeamInfo() (*slack.TeamInfo, error) GetUsersContext(ctx context.Context, options ...slack.GetUsersOption) ([]slack.User, error) GetEmojiContext(ctx context.Context) (map[string]string, error) + GetUsersInConversationContext(ctx context.Context, params *slack.GetUsersInConversationParameters) ([]string, string, error) } var ( diff --git a/slackdump_test.go b/slackdump_test.go index 6a875d54..c7da1513 100644 --- a/slackdump_test.go +++ b/slackdump_test.go @@ -163,18 +163,19 @@ func Test_newLimiter(t *testing.T) { } for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { - t.Parallel() + tst := t + tst.Parallel() got := network.NewLimiter(tt.args.t, tt.args.burst, tt.args.boost) - assert.NoError(t, got.Wait(context.Background())) // prime + assert.NoError(tst, got.Wait(context.Background())) // prime start := time.Now() err := got.Wait(context.Background()) stop := time.Now() - assert.NoError(t, err) - assert.WithinDurationf(t, start.Add(tt.wantDelay), stop, 10*time.Millisecond, "delayed for: %s, expected: %s", stop.Sub(start), tt.wantDelay) + assert.NoError(tst, err) + assert.WithinDurationf(tst, start.Add(tt.wantDelay), stop, 10*time.Millisecond, "delayed for: %s, expected: %s", stop.Sub(start), tt.wantDelay) }) } } @@ -229,7 +230,6 @@ func TestSession_Me(t *testing.T) { Users types.Users UserIndex structures.UserIndex options Options - cacheDir string } tests := []struct { name string @@ -290,7 +290,6 @@ func TestSession_l(t *testing.T) { Users types.Users UserIndex structures.UserIndex options Options - cacheDir string } tests := []struct { name string From 24caf7c67f8fbc4e2eb48a5cffc3b71fb185b1f0 Mon Sep 17 00:00:00 2001 From: Rustam Gilyazov <16064414+rusq@users.noreply.github.com> Date: Sat, 20 May 2023 09:51:10 +1000 Subject: [PATCH 2/2] tests for GetChannelMembers --- channels.go | 2 +- channels_test.go | 100 +++++++++++++++++++++++++++++++++++++++++++++-- 2 files changed, 98 insertions(+), 4 deletions(-) diff --git a/channels.go b/channels.go index 95a7ef64..e9bfb9af 100644 --- a/channels.go +++ b/channels.go @@ -108,7 +108,7 @@ func (sd *Session) GetChannelMembers(ctx context.Context, channelID string) ([]s var next string if err := network.WithRetry(ctx, sd.limiter(network.Tier4), sd.options.Tier4Retries, func() error { var err error - uu, next, err = sd.Client().GetUsersInConversationContext(ctx, &slack.GetUsersInConversationParameters{ + uu, next, err = sd.client.GetUsersInConversationContext(ctx, &slack.GetUsersInConversationParameters{ ChannelID: channelID, Cursor: cursor, }) diff --git a/channels_test.go b/channels_test.go index c42a2484..846482be 100644 --- a/channels_test.go +++ b/channels_test.go @@ -7,11 +7,11 @@ import ( "testing" "github.com/golang/mock/gomock" - "github.com/slack-go/slack" - "github.com/stretchr/testify/assert" - + "github.com/rusq/slackdump/v2/fsadapter" "github.com/rusq/slackdump/v2/internal/structures" "github.com/rusq/slackdump/v2/types" + "github.com/slack-go/slack" + "github.com/stretchr/testify/assert" ) func TestSession_getChannels(t *testing.T) { @@ -142,3 +142,97 @@ func TestSession_GetChannels(t *testing.T) { }) } } + +func TestSession_GetChannelMembers(t *testing.T) { + type fields struct { + wspInfo *slack.AuthTestResponse + fs fsadapter.FS + Users types.Users + UserIndex structures.UserIndex + options Options + } + type args struct { + ctx context.Context + channelID string + } + tests := []struct { + name string + fields fields + args args + expect func(mc *mockClienter) + want []string + wantErr bool + }{ + { + "ok, single call", + fields{options: DefOptions}, + args{ + context.Background(), + "chanID", + }, + func(mc *mockClienter) { + mc.EXPECT().GetUsersInConversationContext(gomock.Any(), &slack.GetUsersInConversationParameters{ + ChannelID: "chanID", + }).Return([]string{"user1", "user2"}, "", nil) + }, + []string{"user1", "user2"}, + false, + }, + { + "ok, two calls", + fields{options: DefOptions}, + args{ + context.Background(), + "chanID", + }, + func(mc *mockClienter) { + first := mc.EXPECT().GetUsersInConversationContext(gomock.Any(), &slack.GetUsersInConversationParameters{ + ChannelID: "chanID", + }).Return([]string{"user1", "user2"}, "cursor", nil).Times(1) + _ = mc.EXPECT().GetUsersInConversationContext(gomock.Any(), &slack.GetUsersInConversationParameters{ + ChannelID: "chanID", + Cursor: "cursor", + }).Return([]string{"user3"}, "", nil).After(first).Times(1) + }, + []string{"user1", "user2", "user3"}, + false, + }, + { + "error", + fields{options: DefOptions}, + args{ + context.Background(), + "chanID", + }, + func(mc *mockClienter) { + mc.EXPECT().GetUsersInConversationContext(gomock.Any(), &slack.GetUsersInConversationParameters{ + ChannelID: "chanID", + }).Return([]string{}, "", errors.New("error fornicating corrugations")) + }, + nil, + true, + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + mc := newmockClienter(gomock.NewController(t)) + tt.expect(mc) + sd := &Session{ + client: mc, + wspInfo: tt.fields.wspInfo, + fs: tt.fields.fs, + Users: tt.fields.Users, + UserIndex: tt.fields.UserIndex, + options: tt.fields.options, + } + got, err := sd.GetChannelMembers(tt.args.ctx, tt.args.channelID) + if (err != nil) != tt.wantErr { + t.Errorf("Session.GetChannelMembers() error = %v, wantErr %v", err, tt.wantErr) + return + } + if !reflect.DeepEqual(got, tt.want) { + t.Errorf("Session.GetChannelMembers() = %v, want %v", got, tt.want) + } + }) + } +}