From b32eb63669054b8f01d829997dea2f2567dc1b5e Mon Sep 17 00:00:00 2001 From: Prem Chaitanya Prathi Date: Thu, 17 Oct 2024 13:11:11 +0530 Subject: [PATCH 1/3] feat_: phase-1 of use single content-topic for all community chats --- eth-node/types/rpc.go | 27 ++++++++------- protocol/common/message_sender.go | 12 ++++--- protocol/common/raw_message.go | 1 + protocol/communities/manager.go | 1 + protocol/communities/manager_archive.go | 4 +++ protocol/communities_key_distributor.go | 1 + ...nities_messenger_token_permissions_test.go | 3 +- protocol/messenger.go | 3 +- protocol/messenger_communities.go | 11 +++++- protocol/transport/filters_manager.go | 34 +++++++++++++------ protocol/transport/transport.go | 13 ++++--- 11 files changed, 73 insertions(+), 37 deletions(-) diff --git a/eth-node/types/rpc.go b/eth-node/types/rpc.go index 6105f3b0160..4af4b967a18 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 20e9ab54461..9662258e7e4 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 687d3b78a9b..9c69801d4fe 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/manager.go b/protocol/communities/manager.go index 56b179f0ae4..4ecabd07501 100644 --- a/protocol/communities/manager.go +++ b/protocol/communities/manager.go @@ -4189,6 +4189,7 @@ func (m *Manager) GetOwnedCommunitiesChatIDs() (map[string]bool, error) { for _, id := range c.ChatIDs() { chatIDs[id] = true } + chatIDs[c.MemberUpdateChannelID()] = true // TODO: for now including this chatID as controlled so that archiving works without any issues. } } return chatIDs, nil diff --git a/protocol/communities/manager_archive.go b/protocol/communities/manager_archive.go index 4b62c99edc2..311aca2cb39 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 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. + topics = append(topics, m.transport.FilterByChatID(community.MemberUpdateChannelID()).ContentTopic) ts := time.Now().Unix() to := time.Unix(ts, 0) diff --git a/protocol/communities_key_distributor.go b/protocol/communities_key_distributor.go index c0c4d125d1e..4a7ebea6ebc 100644 --- a/protocol/communities_key_distributor.go +++ b/protocol/communities_key_distributor.go @@ -96,6 +96,7 @@ func (ckd *CommunitiesKeyDistributorImpl) sendKeyExchangeMessage(community *comm MessageType: protobuf.ApplicationMetadataMessage_CHAT_MESSAGE, HashRatchetGroupID: hashRatchetGroupID, PubsubTopic: community.PubsubTopic(), // TODO: confirm if it should be sent in community pubsub topic + //ContentTopicOverride: community.MemberUpdateChannelID(), //TODO: Confirm if this is correct, could not figure out where LocalChatID is set in this flow } _, err := ckd.sender.SendCommunityMessage(context.Background(), &rawMessage) diff --git a/protocol/communities_messenger_token_permissions_test.go b/protocol/communities_messenger_token_permissions_test.go index bbf05984c67..2e15dd6eae3 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.MemberUpdateChannelID())) + topics := []types.TopicType{topic, communityCommonTopic} torrentConfig := params.TorrentConfig{ Enabled: true, diff --git a/protocol/messenger.go b/protocol/messenger.go index 7d9c455f8c0..88cd29e81fb 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.MemberUpdateChannelID() logger.Debug("sending community chat message", zap.String("chatName", chat.Name)) isCommunityEncrypted, err := m.communitiesManager.IsEncrypted(chat.CommunityID) if err != nil { diff --git a/protocol/messenger_communities.go b/protocol/messenger_communities.go index ad70388589a..392286c74bb 100644 --- a/protocol/messenger_communities.go +++ b/protocol/messenger_communities.go @@ -2713,7 +2713,12 @@ func (m *Messenger) UpdateCommunityFilters(community *communities.Community) err publicFiltersToInit := make([]transport.FiltersToInitialize, 0, len(defaultFilters)+len(community.Chats())) publicFiltersToInit = append(publicFiltersToInit, defaultFilters...) - + for _, filter := range publicFiltersToInit { + _, 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 +3954,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.MemberUpdateChannelID())) // 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 5393d63bf78..3ac990508b0 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 d7254045712..7d86bcfda42 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 { @@ -258,6 +258,8 @@ func (t *Transport) RetrieveRawAll() (map[Filter][]*types.Message, error) { } for i := range msgs { + // TODO: find the filter for the msg based on chatID in the message and map it properly. or better this is done in filter layer itself? + // something like t.FilterByChatID() // Exclude anything that is a cache hit if !hits[types.EncodeHex(msgs[i].Hash)] { result[*filter] = append(result[*filter], msgs[i]) @@ -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 } From d61d20042aa8956ce8634da0ee5f603df379d6dc Mon Sep 17 00:00:00 2001 From: Prem Chaitanya Prathi Date: Thu, 17 Oct 2024 14:40:50 +0530 Subject: [PATCH 2/3] fix_: change communities test --- protocol/communities/manager_test.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/protocol/communities/manager_test.go b/protocol/communities/manager_test.go index 6a9ccd108de..a0699a26d62 100644 --- a/protocol/communities/manager_test.go +++ b/protocol/communities/manager_test.go @@ -452,7 +452,7 @@ func (s *ManagerSuite) TestGetControlledCommunitiesChatIDs() { controlledChatIDs, err := s.manager.GetOwnedCommunitiesChatIDs() s.Require().NoError(err) - s.Require().Len(controlledChatIDs, 1) + s.Require().Len(controlledChatIDs, 2) } func (s *ManagerSuite) TestStartAndStopTorrentClient() { From 54b1cd7d7f5cb329e59a440b90027a11e15b9fb6 Mon Sep 17 00:00:00 2001 From: Prem Chaitanya Prathi Date: Tue, 29 Oct 2024 17:49:10 +0530 Subject: [PATCH 3/3] feat_: use content-topic override for all community filters --- protocol/communities_messenger_test.go | 2 +- protocol/messenger_communities.go | 17 +++++++++-------- protocol/transport/filters_manager.go | 15 +++++++-------- 3 files changed, 17 insertions(+), 17 deletions(-) diff --git a/protocol/communities_messenger_test.go b/protocol/communities_messenger_test.go index fab89c9c8a5..56e8c060d58 100644 --- a/protocol/communities_messenger_test.go +++ b/protocol/communities_messenger_test.go @@ -3630,7 +3630,7 @@ func (s *MessengerCommunitiesSuite) TestHandleImport() { message.Sig = crypto.FromECDSAPub(&s.owner.identity.PublicKey) message.Payload = wrappedPayload - filter := s.alice.transport.FilterByChatID(chat.ID) + filter := s.alice.transport.FilterByChatID(community.MemberUpdateChannelID()) importedMessages := make(map[transport.Filter][]*types.Message, 0) importedMessages[*filter] = append(importedMessages[*filter], message) diff --git a/protocol/messenger_communities.go b/protocol/messenger_communities.go index 392286c74bb..2a46717d2d0 100644 --- a/protocol/messenger_communities.go +++ b/protocol/messenger_communities.go @@ -952,7 +952,7 @@ func (m *Messenger) initCommunityChats(community *communities.Community) ([]*Cha chats := CreateCommunityChats(community, m.getTimesource()) for _, chat := range chats { - publicFiltersToInit = append(publicFiltersToInit, transport.FiltersToInitialize{ChatID: chat.ID, PubsubTopic: community.PubsubTopic()}) + publicFiltersToInit = append(publicFiltersToInit, transport.FiltersToInitialize{ChatID: chat.ID, PubsubTopic: community.PubsubTopic(), ContentTopicOverrideID: community.MemberUpdateChannelID()}) } @@ -2402,7 +2402,7 @@ func (m *Messenger) CreateCommunityChat(communityID types.HexBytes, c *protobuf. for chatID, chat := range changes.ChatsAdded { c := CreateCommunityChat(changes.Community.IDString(), chatID, chat, m.getTimesource()) chats = append(chats, c) - publicFiltersToInit = append(publicFiltersToInit, transport.FiltersToInitialize{ChatID: c.ID, PubsubTopic: changes.Community.PubsubTopic()}) + publicFiltersToInit = append(publicFiltersToInit, transport.FiltersToInitialize{ChatID: c.ID, PubsubTopic: changes.Community.PubsubTopic(), ContentTopicOverrideID: changes.Community.MemberUpdateChannelID()}) response.AddChat(c) } @@ -2444,7 +2444,7 @@ func (m *Messenger) EditCommunityChat(communityID types.HexBytes, chatID string, for chatID, change := range changes.ChatsModified { c := CreateCommunityChat(community.IDString(), chatID, change.ChatModified, m.getTimesource()) chats = append(chats, c) - publicFiltersToInit = append(publicFiltersToInit, transport.FiltersToInitialize{ChatID: c.ID, PubsubTopic: community.PubsubTopic()}) + publicFiltersToInit = append(publicFiltersToInit, transport.FiltersToInitialize{ChatID: c.ID, PubsubTopic: community.PubsubTopic(), ContentTopicOverrideID: community.MemberUpdateChannelID()}) response.AddChat(c) } @@ -2498,8 +2498,8 @@ func (m *Messenger) DefaultFilters(o *communities.Community) []transport.Filters filters := []transport.FiltersToInitialize{ {ChatID: cID, PubsubTopic: communityPubsubTopic}, - {ChatID: updatesChannelID, PubsubTopic: communityPubsubTopic}, - {ChatID: mlChannelID, PubsubTopic: communityPubsubTopic}, + {ChatID: updatesChannelID, PubsubTopic: communityPubsubTopic, ContentTopicOverrideID: memberUpdateChannelID}, + {ChatID: mlChannelID, PubsubTopic: communityPubsubTopic, ContentTopicOverrideID: memberUpdateChannelID}, {ChatID: memberUpdateChannelID, PubsubTopic: communityPubsubTopic}, {ChatID: uncompressedPubKey, PubsubTopic: shard.DefaultNonProtectedPubsubTopic()}, } @@ -2725,7 +2725,7 @@ func (m *Messenger) UpdateCommunityFilters(community *communities.Community) err if err != nil { return err } - publicFiltersToInit = append(publicFiltersToInit, transport.FiltersToInitialize{ChatID: communityChatID, PubsubTopic: community.PubsubTopic()}) + publicFiltersToInit = append(publicFiltersToInit, transport.FiltersToInitialize{ChatID: communityChatID, PubsubTopic: community.PubsubTopic(), ContentTopicOverrideID: community.MemberUpdateChannelID()}) } _, err := m.transport.InitPublicFilters(publicFiltersToInit) @@ -3419,8 +3419,9 @@ func (m *Messenger) handleCommunityResponse(state *ReceivedMessageState, communi state.Response.AddChat(chat) publicFiltersToInit = append(publicFiltersToInit, transport.FiltersToInitialize{ - ChatID: chat.ID, - PubsubTopic: community.PubsubTopic(), + ChatID: chat.ID, + PubsubTopic: community.PubsubTopic(), + ContentTopicOverrideID: community.MemberUpdateChannelID(), }) // Update name, currently is the only field is mutable } else if oldChat.Name != chat.Name || diff --git a/protocol/transport/filters_manager.go b/protocol/transport/filters_manager.go index 3ac990508b0..fb00be0def6 100644 --- a/protocol/transport/filters_manager.go +++ b/protocol/transport/filters_manager.go @@ -624,8 +624,13 @@ func (f *FiltersManager) LoadContactCode(pubKey *ecdsa.PublicKey) (*Filter, erro func (f *FiltersManager) addSymmetric(chatID string, pubsubTopic string, contentTopicID string) (*RawFilter, error) { var symKeyID string var err error - - topic := ToTopic(chatID) + var topic []byte + if contentTopicID != "" { + //use override contentTopicID to generate contentTopic + topic = ToTopic(contentTopicID) + } else { + topic = ToTopic(chatID) + } topics := [][]byte{topic} symKey, ok := f.keys[chatID] @@ -650,12 +655,6 @@ func (f *FiltersManager) addSymmetric(chatID string, pubsubTopic string, content } } - 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,