From a0d88da4801a13b8d33246fc5654c2cf456f7dac Mon Sep 17 00:00:00 2001 From: Sumner Evans Date: Wed, 17 Jul 2024 16:44:53 -0600 Subject: [PATCH] metadata: add pagination config for members initial sync Signed-off-by: Sumner Evans --- pkg/connector/chatinfo.go | 286 ++++++++++++++++++++++++++++++++++++++ pkg/connector/client.go | 197 -------------------------- pkg/connector/config.go | 19 ++- 3 files changed, 300 insertions(+), 202 deletions(-) create mode 100644 pkg/connector/chatinfo.go diff --git a/pkg/connector/chatinfo.go b/pkg/connector/chatinfo.go new file mode 100644 index 00000000..dcbf014d --- /dev/null +++ b/pkg/connector/chatinfo.go @@ -0,0 +1,286 @@ +package connector + +import ( + "context" + "fmt" + "time" + + "github.com/gotd/td/tg" + "go.mau.fi/util/ptr" + "maunium.net/go/mautrix/bridgev2" + "maunium.net/go/mautrix/bridgev2/database" + + "go.mau.fi/mautrix-telegram/pkg/connector/ids" + "go.mau.fi/mautrix-telegram/pkg/connector/media" +) + +func (t *TelegramClient) getDMChatInfo(ctx context.Context, userID int64) (*bridgev2.ChatInfo, error) { + chatInfo := bridgev2.ChatInfo{ + Type: ptr.Ptr(database.RoomTypeDM), + Members: &bridgev2.ChatMemberList{IsFull: true}, + } + // TODO need access hash here + users, err := t.client.API().UsersGetUsers(ctx, []tg.InputUserClass{&tg.InputUser{UserID: userID}}) + if err != nil { + return nil, err + } else if len(users) == 0 { + return nil, fmt.Errorf("failed to get user info for user %d", userID) + } else if userInfo, err := t.getUserInfoFromTelegramUser(users[0]); err != nil { + return nil, err + } else if err = t.updateGhostWithUserInfo(ctx, userID, userInfo); err != nil { + return nil, err + } else { + chatInfo.Members.Members = []bridgev2.ChatMember{ + { + EventSender: bridgev2.EventSender{ + SenderLogin: ids.MakeUserLoginID(userID), + Sender: ids.MakeUserID(userID), + }, + UserInfo: userInfo, + }, + { + EventSender: bridgev2.EventSender{ + IsFromMe: true, + SenderLogin: t.loginID, + Sender: t.userID, + }, + }, + } + } + return &chatInfo, nil +} + +func (t *TelegramClient) getGroupChatInfo(ctx context.Context, fullChat *tg.MessagesChatFull, chatID int64) (*bridgev2.ChatInfo, error) { + if err := t.updateUsersFromResponse(ctx, fullChat); err != nil { + return nil, err + } + + chatInfo := bridgev2.ChatInfo{ + Type: ptr.Ptr(database.RoomTypeGroupDM), // TODO Is this correct for channels? + Members: &bridgev2.ChatMemberList{ + IsFull: true, + Members: []bridgev2.ChatMember{ + { + EventSender: bridgev2.EventSender{ + IsFromMe: true, + SenderLogin: t.loginID, + Sender: t.userID, + }, + }, + }, + }, + } + for _, c := range fullChat.GetChats() { + if c.GetID() == chatID { + switch chat := c.(type) { + case *tg.Chat: + chatInfo.Name = &chat.Title + case *tg.Channel: + chatInfo.Name = &chat.Title + } + break + } + } + + if ttl, ok := fullChat.FullChat.GetTTLPeriod(); ok { + chatInfo.Disappear = &database.DisappearingSetting{ + Type: database.DisappearingTypeAfterSend, + Timer: time.Duration(ttl) * time.Second, + } + } + + if about := fullChat.FullChat.GetAbout(); about != "" { + chatInfo.Topic = &about + } + + return &chatInfo, nil +} + +func (t *TelegramClient) avatarFromPhoto(photo tg.PhotoClass) *bridgev2.Avatar { + if photo == nil || photo.TypeID() != tg.PhotoTypeID { + return nil + } + return &bridgev2.Avatar{ + ID: ids.MakeAvatarID(photo.GetID()), + Get: func(ctx context.Context) (data []byte, err error) { + data, _, err = media.NewTransferer(t.client.API()).WithPhoto(photo).Download(ctx) + return + }, + } +} + +func (t *TelegramClient) filterChannelParticipants(chatParticipants []tg.ChannelParticipantClass, limit int) (members []bridgev2.ChatMember) { + for _, u := range chatParticipants { + userIDable, ok := u.(interface{ GetUserID() int64 }) + if !ok { + continue + } + + members = append(members, bridgev2.ChatMember{ + EventSender: bridgev2.EventSender{ + IsFromMe: userIDable.GetUserID() == t.telegramUserID, + SenderLogin: ids.MakeUserLoginID(userIDable.GetUserID()), + Sender: ids.MakeUserID(userIDable.GetUserID()), + }, + }) + + if len(members) >= limit { + break + } + } + return +} + +func (t *TelegramClient) GetChatInfo(ctx context.Context, portal *bridgev2.Portal) (*bridgev2.ChatInfo, error) { + // fmt.Printf("get chat info %+v\n", portal) + peerType, id, err := ids.ParsePortalID(portal.ID) + if err != nil { + return nil, err + } + + switch peerType { + case ids.PeerTypeUser: + return t.getDMChatInfo(ctx, id) + case ids.PeerTypeChat: + fullChat, err := t.client.API().MessagesGetFullChat(ctx, id) + if err != nil { + return nil, err + } + chatInfo, err := t.getGroupChatInfo(ctx, fullChat, id) + if err != nil { + return nil, err + } + + chatFull, ok := fullChat.FullChat.(*tg.ChatFull) + if !ok { + return nil, fmt.Errorf("full chat is %T not *tg.ChatFull", fullChat.FullChat) + } + chatInfo.Avatar = t.avatarFromPhoto(chatFull.ChatPhoto) + + if chatFull.Participants.TypeID() == tg.ChatParticipantsForbiddenTypeID { + chatInfo.Members.IsFull = false + return chatInfo, nil + } + chatParticipants := chatFull.Participants.(*tg.ChatParticipants) + + if !t.main.Config.ShouldBridge(len(chatParticipants.Participants)) { + // TODO change this to a better error whenever that is implemented in mautrix-go + return nil, fmt.Errorf("too many participants (%d) in chat %d", len(chatParticipants.Participants), id) + } + + for _, user := range chatParticipants.GetParticipants() { + if user.TypeID() == tg.ChannelParticipantBannedTypeID { + continue + } + + chatInfo.Members.Members = append(chatInfo.Members.Members, bridgev2.ChatMember{ + EventSender: bridgev2.EventSender{ + IsFromMe: user.GetUserID() == t.telegramUserID, + SenderLogin: ids.MakeUserLoginID(user.GetUserID()), + Sender: ids.MakeUserID(user.GetUserID()), + }, + }) + + if len(chatInfo.Members.Members) >= t.main.Config.MemberList.NormalizedMaxInitialSync() { + break + } + } + return chatInfo, nil + case ids.PeerTypeChannel: + accessHash, found, err := t.ScopedStore.GetChannelAccessHash(ctx, t.telegramUserID, id) + if err != nil { + return nil, fmt.Errorf("failed to get channel access hash: %w", err) + } else if !found { + return nil, fmt.Errorf("channel access hash not found for %d", id) + } + inputChannel := &tg.InputChannel{ChannelID: id, AccessHash: accessHash} + fullChat, err := t.client.API().ChannelsGetFullChannel(ctx, inputChannel) + if err != nil { + return nil, err + } + + chatInfo, err := t.getGroupChatInfo(ctx, fullChat, id) + if err != nil { + return nil, err + } + + channelFull, ok := fullChat.FullChat.(*tg.ChannelFull) + if !ok { + return nil, fmt.Errorf("full chat is %T not *tg.ChannelFull", fullChat.FullChat) + } + + if !t.main.Config.ShouldBridge(channelFull.ParticipantsCount) { + // TODO change this to a better error whenever that is implemented in mautrix-go + return nil, fmt.Errorf("too many participants (%d) in chat %d", channelFull.ParticipantsCount, id) + } + + chatInfo.Avatar = t.avatarFromPhoto(channelFull.ChatPhoto) + + // TODO save available reactions? + // TODO save reactions limit? + // TODO save emojiset? + + chatInfo.Members.IsFull = false + + // Just return the current user as a member if we can't view the + // participants or the max initial sync is 0 or if channel member sync + // is disabled. + if t.main.Config.MemberList.MaxInitialSync == 0 || !t.main.Config.MemberList.SyncChannels || !channelFull.CanViewParticipants || channelFull.ParticipantsHidden { + return chatInfo, nil + } + + limit := t.main.Config.MemberList.NormalizedMaxInitialSync() + if limit <= 200 { + p, err := t.client.API().ChannelsGetParticipants(ctx, &tg.ChannelsGetParticipantsRequest{ + Channel: inputChannel, + Filter: &tg.ChannelParticipantsRecent{}, + Limit: limit, + }) + if err != nil { + return nil, err + } + participants, ok := p.(*tg.ChannelsChannelParticipants) + if !ok { + return nil, fmt.Errorf("returned participants is %T not *tg.ChannelsChannelParticipants", p) + } + chatInfo.Members.IsFull = len(participants.Participants) < limit + if err := t.updateUsersFromResponse(ctx, participants); err != nil { + return nil, err + } + chatInfo.Members.Members = append(chatInfo.Members.Members, t.filterChannelParticipants(participants.Participants, limit)...) + } else { + remaining := t.main.Config.MemberList.NormalizedMaxInitialSync() + var offset int + for remaining > 0 { + p, err := t.client.API().ChannelsGetParticipants(ctx, &tg.ChannelsGetParticipantsRequest{ + Channel: inputChannel, + Filter: &tg.ChannelParticipantsSearch{}, + Limit: min(remaining, 200), + Offset: offset, + }) + if err != nil { + return nil, err + } + participants, ok := p.(*tg.ChannelsChannelParticipants) + if !ok { + return nil, fmt.Errorf("returned participants is %T not *tg.ChannelsChannelParticipants", p) + } + if len(participants.Participants) == 0 { + chatInfo.Members.IsFull = true + break + } + + if err := t.updateUsersFromResponse(ctx, participants); err != nil { + return nil, err + } + chatInfo.Members.Members = append(chatInfo.Members.Members, t.filterChannelParticipants(participants.Participants, limit)...) + + offset += len(participants.Participants) + remaining -= len(participants.Participants) + } + } + return chatInfo, nil + default: + panic(fmt.Sprintf("unsupported peer type %s", peerType)) + } +} diff --git a/pkg/connector/client.go b/pkg/connector/client.go index c416cb44..0b78ed8c 100644 --- a/pkg/connector/client.go +++ b/pkg/connector/client.go @@ -6,18 +6,15 @@ import ( "fmt" "strings" "sync" - "time" "github.com/gotd/td/telegram" "github.com/gotd/td/telegram/updates" "github.com/gotd/td/tg" "github.com/rs/zerolog" - "go.mau.fi/util/ptr" "go.mau.fi/zerozap" "go.uber.org/zap" "maunium.net/go/mautrix/bridge/status" "maunium.net/go/mautrix/bridgev2" - "maunium.net/go/mautrix/bridgev2/database" "maunium.net/go/mautrix/bridgev2/networkid" "go.mau.fi/mautrix-telegram/pkg/connector/ids" @@ -215,200 +212,6 @@ func (t *TelegramClient) updateUsersFromResponse(ctx context.Context, resp inter return nil } -func (t *TelegramClient) getDMChatInfo(ctx context.Context, userID int64) (*bridgev2.ChatInfo, error) { - chatInfo := bridgev2.ChatInfo{ - Type: ptr.Ptr(database.RoomTypeDM), - Members: &bridgev2.ChatMemberList{IsFull: true}, - } - // TODO need access hash here - users, err := t.client.API().UsersGetUsers(ctx, []tg.InputUserClass{&tg.InputUser{UserID: userID}}) - if err != nil { - return nil, err - } else if len(users) == 0 { - return nil, fmt.Errorf("failed to get user info for user %d", userID) - } else if userInfo, err := t.getUserInfoFromTelegramUser(users[0]); err != nil { - return nil, err - } else if err = t.updateGhostWithUserInfo(ctx, userID, userInfo); err != nil { - return nil, err - } else { - chatInfo.Members.Members = []bridgev2.ChatMember{ - { - EventSender: bridgev2.EventSender{ - SenderLogin: ids.MakeUserLoginID(userID), - Sender: ids.MakeUserID(userID), - }, - UserInfo: userInfo, - }, - { - EventSender: bridgev2.EventSender{ - IsFromMe: true, - SenderLogin: t.loginID, - Sender: t.userID, - }, - }, - } - } - return &chatInfo, nil -} - -func (t *TelegramClient) getGroupChatInfo(ctx context.Context, fullChat *tg.MessagesChatFull, chatID int64) (*bridgev2.ChatInfo, error) { - if err := t.updateUsersFromResponse(ctx, fullChat); err != nil { - return nil, err - } - - chatInfo := bridgev2.ChatInfo{ - Type: ptr.Ptr(database.RoomTypeGroupDM), // TODO Is this correct for channels? - Members: &bridgev2.ChatMemberList{IsFull: true}, - } - for _, c := range fullChat.GetChats() { - if c.GetID() == chatID { - switch chat := c.(type) { - case *tg.Chat: - chatInfo.Name = &chat.Title - case *tg.Channel: - chatInfo.Name = &chat.Title - } - break - } - } - - if ttl, ok := fullChat.FullChat.GetTTLPeriod(); ok { - chatInfo.Disappear = &database.DisappearingSetting{ - Type: database.DisappearingTypeAfterSend, - Timer: time.Duration(ttl) * time.Second, - } - } - - if about := fullChat.FullChat.GetAbout(); about != "" { - chatInfo.Topic = &about - } - - return &chatInfo, nil -} - -func (t *TelegramClient) avatarFromPhoto(ctx context.Context, photo tg.PhotoClass) *bridgev2.Avatar { - if photo == nil || photo.TypeID() != tg.PhotoTypeID { - return nil - } - return &bridgev2.Avatar{ - ID: ids.MakeAvatarID(photo.GetID()), - Get: func(ctx context.Context) (data []byte, err error) { - data, _, err = media.NewTransferer(t.client.API()).WithPhoto(photo).Download(ctx) - return - }, - } -} - -func (t *TelegramClient) GetChatInfo(ctx context.Context, portal *bridgev2.Portal) (*bridgev2.ChatInfo, error) { - // fmt.Printf("get chat info %+v\n", portal) - peerType, id, err := ids.ParsePortalID(portal.ID) - if err != nil { - return nil, err - } - - switch peerType { - case ids.PeerTypeUser: - return t.getDMChatInfo(ctx, id) - case ids.PeerTypeChat: - fullChat, err := t.client.API().MessagesGetFullChat(ctx, id) - if err != nil { - return nil, err - } - chatInfo, err := t.getGroupChatInfo(ctx, fullChat, id) - if err != nil { - return nil, err - } - - chatFull, ok := fullChat.FullChat.(*tg.ChatFull) - if !ok { - return nil, fmt.Errorf("full chat is %T not *tg.ChatFull", fullChat.FullChat) - } - chatInfo.Avatar = t.avatarFromPhoto(ctx, chatFull.ChatPhoto) - - participants, ok := chatFull.Participants.(*tg.ChatParticipants) - if !ok { - return nil, fmt.Errorf("full chat is %T not *tg.ChatParticipants", chatFull.Participants) - } - if !t.main.Config.ShouldBridge(len(participants.Participants)) { - return nil, fmt.Errorf("too many participants (%d) in chat %d", len(participants.Participants), id) - } - for _, user := range participants.Participants { - chatInfo.Members.Members = append(chatInfo.Members.Members, bridgev2.ChatMember{ - EventSender: bridgev2.EventSender{ - IsFromMe: user.GetUserID() == t.telegramUserID, - SenderLogin: ids.MakeUserLoginID(user.GetUserID()), - Sender: ids.MakeUserID(user.GetUserID()), - }, - }) - } - return chatInfo, nil - case ids.PeerTypeChannel: - accessHash, found, err := t.ScopedStore.GetChannelAccessHash(ctx, t.telegramUserID, id) - if err != nil { - return nil, fmt.Errorf("failed to get channel access hash: %w", err) - } else if !found { - return nil, fmt.Errorf("channel access hash not found for %d", id) - } - inputChannel := &tg.InputChannel{ChannelID: id, AccessHash: accessHash} - fullChat, err := t.client.API().ChannelsGetFullChannel(ctx, inputChannel) - if err != nil { - return nil, err - } - - chatInfo, err := t.getGroupChatInfo(ctx, fullChat, id) - if err != nil { - return nil, err - } - - channelFull, ok := fullChat.FullChat.(*tg.ChannelFull) - if !ok { - return nil, fmt.Errorf("full chat is %T not *tg.ChannelFull", fullChat.FullChat) - } - chatInfo.Avatar = t.avatarFromPhoto(ctx, channelFull.ChatPhoto) - - // TODO save available reactions? - // TODO save reactions limit? - // TODO save emojiset? - - chatInfo.Members.IsFull = false - - // TODO when do we have to make a request to get the participants? - - if channelFull.CanViewParticipants && !channelFull.ParticipantsHidden { - // TODO paginate - p, err := t.client.API().ChannelsGetParticipants(ctx, &tg.ChannelsGetParticipantsRequest{ - Channel: inputChannel, - Filter: &tg.ChannelParticipantsSearch{}, - Limit: 200, - }) - if err != nil { - return nil, err - } - participants, ok := p.(*tg.ChannelsChannelParticipants) - if !ok { - return nil, fmt.Errorf("returned participants is %T not *tg.ChannelsChannelParticipants", p) - } - for _, user := range participants.GetUsers() { - userInfo, err := t.getUserInfoFromTelegramUser(user) - if err != nil { - return nil, err - } - chatInfo.Members.Members = append(chatInfo.Members.Members, bridgev2.ChatMember{ - EventSender: bridgev2.EventSender{ - IsFromMe: user.GetID() == t.telegramUserID, - SenderLogin: ids.MakeUserLoginID(user.GetID()), - Sender: ids.MakeUserID(user.GetID()), - }, - UserInfo: userInfo, - }) - } - } - return chatInfo, nil - default: - panic(fmt.Sprintf("unsupported peer type %s", peerType)) - } -} - func (t *TelegramClient) GetUserInfo(ctx context.Context, ghost *bridgev2.Ghost) (*bridgev2.UserInfo, error) { id, err := ids.ParseUserID(ghost.ID) if err != nil { diff --git a/pkg/connector/config.go b/pkg/connector/config.go index ad3e77bf..fed7f4d5 100644 --- a/pkg/connector/config.go +++ b/pkg/connector/config.go @@ -14,17 +14,26 @@ import ( var _ bridgev2.ConfigValidatingNetwork = (*TelegramConnector)(nil) +type MemberListConfig struct { + MaxInitialSync int `yaml:"max_initial_sync"` + SyncChannels bool `yaml:"sync_channels"` + SkipDeleted bool `yaml:"skip_deleted"` +} + +func (c MemberListConfig) NormalizedMaxInitialSync() int { + if c.MaxInitialSync < 0 { + return 10000 + } + return c.MaxInitialSync +} + type TelegramConfig struct { AppID int `yaml:"app_id"` AppHash string `yaml:"app_hash"` AnimatedSticker media.AnimatedStickerConfig `yaml:"animated_sticker"` - MemberList struct { - MaxInitialSync int `yaml:"max_initial_sync"` - SyncChannels bool `yaml:"sync_channels"` - SkipDeleted bool `yaml:"skip_deleted"` - } `yaml:"member_list"` + MemberList MemberListConfig `yaml:"member_list"` MaxMemberCount int `yaml:"max_member_count"` }