From 103b415144497eac21fe2cbf145aff455fad4f98 Mon Sep 17 00:00:00 2001 From: Prem Chaitanya Prathi Date: Fri, 1 Nov 2024 17:13:40 +0530 Subject: [PATCH] feat_: phase-1 of use single content-topic to send messages for all community chats --- eth-node/types/rpc.go | 27 ++++++++------- protocol/common/message_sender.go | 12 ++++--- protocol/common/raw_message.go | 1 + protocol/communities/community.go | 6 ++++ protocol/communities/manager.go | 16 +++++++++ protocol/communities/manager_archive.go | 4 +++ ...nities_messenger_token_permissions_test.go | 3 +- protocol/messenger.go | 12 ++++++- protocol/messenger_communities.go | 11 ++++++ protocol/transport/filters_manager.go | 34 +++++++++++++------ protocol/transport/transport.go | 13 ++++--- 11 files changed, 103 insertions(+), 36 deletions(-) diff --git a/eth-node/types/rpc.go b/eth-node/types/rpc.go index 6105f3b01..4af4b967a 100644 --- a/eth-node/types/rpc.go +++ b/eth-node/types/rpc.go @@ -6,19 +6,20 @@ import ( // NewMessage represents a new whisper message that is posted through the RPC. type NewMessage struct { - SymKeyID string `json:"symKeyID"` - PublicKey []byte `json:"pubKey"` - SigID string `json:"sig"` - TTL uint32 `json:"ttl"` - PubsubTopic string `json:"pubsubTopic"` - Topic TopicType `json:"topic"` - Payload []byte `json:"payload"` - Padding []byte `json:"padding"` - PowTime uint32 `json:"powTime"` - PowTarget float64 `json:"powTarget"` - TargetPeer string `json:"targetPeer"` - Ephemeral bool `json:"ephemeral"` - Priority *int `json:"priority"` + SymKeyID string `json:"symKeyID"` + PublicKey []byte `json:"pubKey"` + SigID string `json:"sig"` + TTL uint32 `json:"ttl"` + PubsubTopic string `json:"pubsubTopic"` + Topic TopicType `json:"topic"` + Payload []byte `json:"payload"` + Padding []byte `json:"padding"` + PowTime uint32 `json:"powTime"` + PowTarget float64 `json:"powTarget"` + TargetPeer string `json:"targetPeer"` + Ephemeral bool `json:"ephemeral"` + Priority *int `json:"priority"` + ContentTopicOverride string `json:"contentTopicOverride"` } // Message is the RPC representation of a whisper message. diff --git a/protocol/common/message_sender.go b/protocol/common/message_sender.go index 20e9ab544..9662258e7 100644 --- a/protocol/common/message_sender.go +++ b/protocol/common/message_sender.go @@ -660,11 +660,12 @@ func (s *MessageSender) dispatchCommunityChatMessage(ctx context.Context, rawMes } newMessage := &types.NewMessage{ - TTL: whisperTTL, - Payload: payload, - PowTarget: calculatePoW(payload), - PowTime: whisperPoWTime, - PubsubTopic: rawMessage.PubsubTopic, + TTL: whisperTTL, + Payload: payload, + PowTarget: calculatePoW(payload), + PowTime: whisperPoWTime, + PubsubTopic: rawMessage.PubsubTopic, + ContentTopicOverride: rawMessage.ContentTopicOverride, } if rawMessage.BeforeDispatch != nil { @@ -765,6 +766,7 @@ func (s *MessageSender) SendPublic( newMessage.Ephemeral = rawMessage.Ephemeral newMessage.PubsubTopic = rawMessage.PubsubTopic newMessage.Priority = rawMessage.Priority + newMessage.ContentTopicOverride = rawMessage.ContentTopicOverride messageID := v1protocol.MessageID(&rawMessage.Sender.PublicKey, wrappedMessage) diff --git a/protocol/common/raw_message.go b/protocol/common/raw_message.go index 687d3b78a..9c69801d4 100644 --- a/protocol/common/raw_message.go +++ b/protocol/common/raw_message.go @@ -84,4 +84,5 @@ type RawMessage struct { ResendType ResendType ResendMethod ResendMethod Priority *MessagePriority + ContentTopicOverride string } diff --git a/protocol/communities/community.go b/protocol/communities/community.go index 566be3336..0f46c2159 100644 --- a/protocol/communities/community.go +++ b/protocol/communities/community.go @@ -1567,6 +1567,12 @@ func (o *Community) setPrivateKey(pk *ecdsa.PrivateKey) { o.config.PrivateKey = pk } } +func (o *Community) UniversalChatID() string { + // Using Member updates channelID as chatID to act as a universal content-topic for all chats in the community as explained here https://forum.vac.dev/t/status-communities-review-and-proposed-usage-of-waku-content-topics/335 + // This is to match filter criteria of community with the content-topic usage. + // This specific topic is chosen as existing users before the change are already subscribed to this and will not get affected by it. + return o.MemberUpdateChannelID() +} func (o *Community) SetResendAccountsClock(clock uint64) { o.config.CommunityDescription.ResendAccountsClock = clock diff --git a/protocol/communities/manager.go b/protocol/communities/manager.go index 56b179f0a..e3461f168 100644 --- a/protocol/communities/manager.go +++ b/protocol/communities/manager.go @@ -4194,6 +4194,22 @@ func (m *Manager) GetOwnedCommunitiesChatIDs() (map[string]bool, error) { return chatIDs, nil } +func (m *Manager) GetOwnedCommunitiesUniversalChatIDs() (map[string]bool, error) { + ownedCommunities, err := m.Controlled() + if err != nil { + return nil, err + } + + chatIDs := make(map[string]bool) + for _, c := range ownedCommunities { + if c.Joined() { + chatIDs[c.UniversalChatID()] = true + } + } + return chatIDs, nil + +} + func (m *Manager) StoreWakuMessage(message *types.Message) error { return m.persistence.SaveWakuMessage(message) } diff --git a/protocol/communities/manager_archive.go b/protocol/communities/manager_archive.go index 4b62c99ed..98713f1bb 100644 --- a/protocol/communities/manager_archive.go +++ b/protocol/communities/manager_archive.go @@ -354,6 +354,10 @@ func (m *ArchiveManager) StartHistoryArchiveTasksInterval(community *Community, m.logger.Error("failed to get community chat topics ", zap.Error(err)) continue } + // adding the content-topic used for all community chats as all chat messages will be published on this topic. + // since member updates would not be too frequent i.e only addition/deletion would add a new message, + // this shouldn't cause too much increase in size of archive generated. + topics = append(topics, m.transport.FilterByChatID(community.UniversalChatID()).ContentTopic) ts := time.Now().Unix() to := time.Unix(ts, 0) diff --git a/protocol/communities_messenger_token_permissions_test.go b/protocol/communities_messenger_token_permissions_test.go index bbf05984c..9a226603b 100644 --- a/protocol/communities_messenger_token_permissions_test.go +++ b/protocol/communities_messenger_token_permissions_test.go @@ -2152,7 +2152,8 @@ func (s *MessengerCommunitiesTokenPermissionsSuite) TestImportDecryptedArchiveMe startDate := messageDate.Add(-time.Minute) endDate := messageDate.Add(time.Minute) topic := types.BytesToTopic(transport.ToTopic(chat.ID)) - topics := []types.TopicType{topic} + communityCommonTopic := types.BytesToTopic(transport.ToTopic(community.UniversalChatID())) + topics := []types.TopicType{topic, communityCommonTopic} torrentConfig := params.TorrentConfig{ Enabled: true, diff --git a/protocol/messenger.go b/protocol/messenger.go index 7d9c455f8..e4c07146a 100644 --- a/protocol/messenger.go +++ b/protocol/messenger.go @@ -2268,7 +2268,8 @@ func (m *Messenger) dispatchMessage(ctx context.Context, rawMessage common.RawMe ) return rawMessage, fmt.Errorf("can't post message type '%d' on chat '%s'", rawMessage.MessageType, chat.ID) } - + //setting content-topic over-ride for community messages to use memberUpdatesChannelID + rawMessage.ContentTopicOverride = community.UniversalChatID() logger.Debug("sending community chat message", zap.String("chatName", chat.Name)) isCommunityEncrypted, err := m.communitiesManager.IsEncrypted(chat.CommunityID) if err != nil { @@ -3589,6 +3590,15 @@ func (m *Messenger) handleRetrievedMessages(chatWithMessages map[transport.Filte if err != nil { logger.Info("failed to retrieve admin communities", zap.Error(err)) } + //fetch universal chatIDs as well. + controlledCommunitiesUniversalChatIDs, err := m.communitiesManager.GetOwnedCommunitiesUniversalChatIDs() + if err != nil { + logger.Info("failed to retrieve admin communities", zap.Error(err)) + } + + for chatID, flag := range controlledCommunitiesUniversalChatIDs { + controlledCommunitiesChatIDs[chatID] = flag + } iterator := m.retrievedMessagesIteratorFactory(chatWithMessages) for iterator.HasNext() { diff --git a/protocol/messenger_communities.go b/protocol/messenger_communities.go index ad7038858..1321b7a32 100644 --- a/protocol/messenger_communities.go +++ b/protocol/messenger_communities.go @@ -2714,6 +2714,13 @@ func (m *Messenger) UpdateCommunityFilters(community *communities.Community) err publicFiltersToInit = append(publicFiltersToInit, defaultFilters...) + for _, filter := range defaultFilters { + _, err := m.transport.RemoveFilterByChatID(filter.ChatID) + if err != nil { + return err + } + } + for chatID := range community.Chats() { communityChatID := community.IDString() + chatID _, err := m.transport.RemoveFilterByChatID(communityChatID) @@ -3949,6 +3956,10 @@ func (m *Messenger) InitHistoryArchiveTasks(communities []*communities.Community for _, filter := range filters { topics = append(topics, filter.ContentTopic) } + // adding the content-topic used for member updates. + // since member updates would not be too frequent i.e only addition/deletion would add a new message, + // this shouldn't cause too much increase in size of archive generated. + filters = append(filters, m.transport.FilterByChatID(c.UniversalChatID())) // First we need to know the timestamp of the latest waku message // we've received for this community, so we can request messages we've diff --git a/protocol/transport/filters_manager.go b/protocol/transport/filters_manager.go index 5393d63bf..3ac990508 100644 --- a/protocol/transport/filters_manager.go +++ b/protocol/transport/filters_manager.go @@ -99,7 +99,7 @@ func (f *FiltersManager) Init( // Add public, one-to-one and negotiated filters. for _, fi := range filtersToInit { - _, err := f.LoadPublic(fi.ChatID, fi.PubsubTopic) + _, err := f.LoadPublic(fi.ChatID, fi.PubsubTopic, fi.ContentTopicOverrideID) if err != nil { return nil, err } @@ -123,15 +123,16 @@ func (f *FiltersManager) Init( } type FiltersToInitialize struct { - ChatID string - PubsubTopic string + ChatID string + PubsubTopic string + ContentTopicOverrideID string //litte hacky but this is used to override content-topic in filtersManager. } func (f *FiltersManager) InitPublicFilters(publicFiltersToInit []FiltersToInitialize) ([]*Filter, error) { var filters []*Filter // Add public, one-to-one and negotiated filters. for _, pf := range publicFiltersToInit { - f, err := f.LoadPublic(pf.ChatID, pf.PubsubTopic) + f, err := f.LoadPublic(pf.ChatID, pf.PubsubTopic, pf.ContentTopicOverrideID) if err != nil { return nil, err } @@ -455,7 +456,7 @@ func (f *FiltersManager) LoadNegotiated(secret types.NegotiatedSecret) (*Filter, } keyString := hex.EncodeToString(secret.Key) - filter, err := f.addSymmetric(keyString, "") + filter, err := f.addSymmetric(keyString, "", "") if err != nil { f.logger.Debug("could not register negotiated topic", zap.Error(err)) return nil, err @@ -534,11 +535,16 @@ func (f *FiltersManager) PersonalTopicFilter() *Filter { } // LoadPublic adds a filter for a public chat. -func (f *FiltersManager) LoadPublic(chatID string, pubsubTopic string) (*Filter, error) { +func (f *FiltersManager) LoadPublic(chatID string, pubsubTopic string, contentTopicID string) (*Filter, error) { f.mutex.Lock() defer f.mutex.Unlock() - if chat, ok := f.filters[chatID]; ok { + chatIDToLoad := chatID + if contentTopicID != "" { + chatIDToLoad = contentTopicID + } + + if chat, ok := f.filters[chatIDToLoad]; ok { if chat.PubsubTopic != pubsubTopic { f.logger.Debug("updating pubsub topic for filter", zap.String("chatID", chatID), @@ -547,13 +553,13 @@ func (f *FiltersManager) LoadPublic(chatID string, pubsubTopic string) (*Filter, zap.String("newTopic", pubsubTopic), ) chat.PubsubTopic = pubsubTopic - f.filters[chatID] = chat + f.filters[chatIDToLoad] = chat //TODO: Do we need to update watchers as well on modification? } return chat, nil } - filterAndTopic, err := f.addSymmetric(chatID, pubsubTopic) + filterAndTopic, err := f.addSymmetric(chatID, pubsubTopic, contentTopicID) if err != nil { f.logger.Debug("could not register public chat topic", zap.String("chatID", chatID), zap.Error(err)) return nil, err @@ -592,7 +598,7 @@ func (f *FiltersManager) LoadContactCode(pubKey *ecdsa.PublicKey) (*Filter, erro return f.filters[chatID], nil } - contactCodeFilter, err := f.addSymmetric(chatID, "") + contactCodeFilter, err := f.addSymmetric(chatID, "", "") if err != nil { f.logger.Debug("could not register contact code topic", zap.String("chatID", chatID), zap.Error(err)) return nil, err @@ -615,7 +621,7 @@ func (f *FiltersManager) LoadContactCode(pubKey *ecdsa.PublicKey) (*Filter, erro } // addSymmetric adds a symmetric key filter -func (f *FiltersManager) addSymmetric(chatID string, pubsubTopic string) (*RawFilter, error) { +func (f *FiltersManager) addSymmetric(chatID string, pubsubTopic string, contentTopicID string) (*RawFilter, error) { var symKeyID string var err error @@ -644,6 +650,12 @@ func (f *FiltersManager) addSymmetric(chatID string, pubsubTopic string) (*RawFi } } + if contentTopicID != "" { + //add receive filter for the single default contentTopic for all community chats + topic = ToTopic(contentTopicID) + topics = append(topics, topic) + } + id, err := f.service.Subscribe(&types.SubscriptionOptions{ SymKeyID: symKeyID, PoW: minPow, diff --git a/protocol/transport/transport.go b/protocol/transport/transport.go index d72540457..6b7d43f71 100644 --- a/protocol/transport/transport.go +++ b/protocol/transport/transport.go @@ -191,7 +191,7 @@ func (t *Transport) ProcessNegotiatedSecret(secret types.NegotiatedSecret) (*Fil } func (t *Transport) JoinPublic(chatID string) (*Filter, error) { - return t.filters.LoadPublic(chatID, "") + return t.filters.LoadPublic(chatID, "", "") } func (t *Transport) LeavePublic(chatID string) error { @@ -223,6 +223,8 @@ func (t *Transport) GetStats() types.StatsSummary { return t.waku.GetStats() } +// With change in filter used for communities, messages are indexed here with common filter and not their own chatID filter. +// The caller should not use chatID from the filter to determine chatID of the message, rather should dervice it from messaage itself. func (t *Transport) RetrieveRawAll() (map[Filter][]*types.Message, error) { result := make(map[Filter][]*types.Message) logger := t.logger.With(zap.String("site", "retrieveRawAll")) @@ -276,16 +278,16 @@ func (t *Transport) RetrieveRawAll() (map[Filter][]*types.Message, error) { // SendPublic sends a new message using the Whisper service. // For public filters, chat name is used as an ID as well as // a topic. +// In case of communities a single topic is used to send all messages. func (t *Transport) SendPublic(ctx context.Context, newMessage *types.NewMessage, chatName string) ([]byte, error) { if err := t.addSig(newMessage); err != nil { return nil, err } - - filter, err := t.filters.LoadPublic(chatName, newMessage.PubsubTopic) + //passing content-topic override, it will be used if set. otherwise chatName will be used to load filter. + filter, err := t.filters.LoadPublic(chatName, newMessage.PubsubTopic, newMessage.ContentTopicOverride) if err != nil { return nil, err } - newMessage.SymKeyID = filter.SymKeyID newMessage.Topic = filter.ContentTopic newMessage.PubsubTopic = filter.PubsubTopic @@ -362,7 +364,8 @@ func (t *Transport) SendCommunityMessage(ctx context.Context, newMessage *types. } // We load the filter to make sure we can post on it - filter, err := t.filters.LoadPublic(PubkeyToHex(publicKey)[2:], newMessage.PubsubTopic) + //passing content-topic override, it will be used if set. otherwise chatName will be used to load filter. + filter, err := t.filters.LoadPublic(PubkeyToHex(publicKey)[2:], newMessage.PubsubTopic, newMessage.ContentTopicOverride) if err != nil { return nil, err }