Extend peersyncing to sync 1-to-1 messages

status-im · Mar 21, 2024 · ab73cf8 · ab73cf8
1 parent 0aed93f
commit ab73cf8
Show file tree

Hide file tree

Showing 8 changed files with 32 additions and 24 deletions.
diff --git a/protocol/common/message_sender.go b/protocol/common/message_sender.go
@@ -748,6 +748,7 @@ func (s *MessageSender) SendPublic(
 
 // unwrapDatasyncMessage tries to unwrap message as datasync one and in case of success
 // returns cloned messages with replaced payloads
+// TODO(alwx): peer syncing
 func (s *MessageSender) unwrapDatasyncMessage(m *v1protocol.StatusMessage, response *handleMessageResponse) error {
 
 	datasyncMessage, err := s.datasync.Unwrap(
@@ -763,7 +764,7 @@ func (s *MessageSender) unwrapDatasyncMessage(m *v1protocol.StatusMessage, respo
 	response.DatasyncRequests = append(response.DatasyncRequests, datasyncMessage.Requests...)
 	for _, o := range datasyncMessage.GroupOffers {
 		for _, mID := range o.MessageIds {
-			response.DatasyncOffers = append(response.DatasyncOffers, DatasyncOffer{GroupID: o.GroupId, MessageID: mID})
+			response.DatasyncOffers = append(response.DatasyncOffers, DatasyncOffer{ChatID: o.GroupId, MessageID: mID})
 		}
 	}
 
@@ -837,7 +838,7 @@ func (s *MessageSender) HandleMessages(wakuMessage *types.Message) (*HandleMessa
 }
 
 type DatasyncOffer struct {
-	GroupID   []byte
+	ChatID    []byte
 	MessageID []byte
 }
 

diff --git a/protocol/messenger.go b/protocol/messenger.go
@@ -2529,7 +2529,7 @@ func (m *Messenger) sendChatMessage(ctx context.Context, message *common.Message
 		syncMessage := peersyncing.SyncMessage{
 			Type:      syncMessageType,
 			ID:        types.Hex2Bytes(rawMessage.ID),
-			GroupID:   []byte(chat.ID),
+			ChatID:    []byte(chat.ID),
 			Payload:   wrappedMessage,
 			Timestamp: m.transport.GetCurrentTime() / 1000,
 		}

diff --git a/protocol/messenger_handler.go b/protocol/messenger_handler.go
@@ -2442,7 +2442,7 @@ func (m *Messenger) addPeersyncingMessage(chat *Chat, msg *v1protocol.StatusMess
 	syncMessage := peersyncing.SyncMessage{
 		Type:      syncMessageType,
 		ID:        msg.ApplicationLayer.ID,
-		GroupID:   []byte(chat.ID),
+		ChatID:    []byte(chat.ID),
 		Payload:   msg.EncryptionLayer.Payload,
 		Timestamp: uint64(msg.TransportLayer.Message.Timestamp),
 	}

diff --git a/protocol/messenger_peersyncing.go b/protocol/messenger_peersyncing.go
@@ -139,7 +139,7 @@ func (m *Messenger) sendDatasyncOffers() error {
 		}
 		availableMessagesMap := make(map[string][][]byte)
 		for _, m := range availableMessages {
-			groupID := types.Bytes2Hex(m.GroupID)
+			groupID := types.Bytes2Hex(m.ChatID)
 			availableMessagesMap[groupID] = append(availableMessagesMap[groupID], m.ID)
 		}
 
@@ -188,7 +188,7 @@ func (m *Messenger) OnDatasyncOffer(response *common.HandleMessageResponse) erro
 	var offeredMessages []peersyncing.SyncMessage
 
 	for _, o := range offers {
-		offeredMessages = append(offeredMessages, peersyncing.SyncMessage{GroupID: o.GroupID, ID: o.MessageID})
+		offeredMessages = append(offeredMessages, peersyncing.SyncMessage{ChatID: o.ChatID, ID: o.MessageID})
 	}
 
 	messagesToFetch, err := m.peersyncing.OnOffer(offeredMessages)
@@ -231,11 +231,12 @@ func (m *Messenger) OnDatasyncOffer(response *common.HandleMessageResponse) erro
 	return nil
 }
 
+// TODO(alwx): peer syncing
 // canSyncMessageWith checks the permission of a message
 func (m *Messenger) canSyncMessageWith(message peersyncing.SyncMessage, peer *ecdsa.PublicKey) (bool, error) {
 	switch message.Type {
 	case peersyncing.SyncMessageCommunityType:
-		chat, ok := m.allChats.Load(string(message.GroupID))
+		chat, ok := m.allChats.Load(string(message.ChatID))
 		if !ok {
 			return false, nil
 		}
@@ -245,7 +246,13 @@ func (m *Messenger) canSyncMessageWith(message peersyncing.SyncMessage, peer *ec
 		}
 
 		return m.canSyncCommunityMessageWith(chat, community, peer)
-
+	case peersyncing.SyncMessageOneToOneType:
+		chat, ok := m.allChats.Load(string(message.ChatID))
+		if !ok {
+			return false, nil
+		}
+		key := common.PubkeyToHex(peer)
+		return chat.HasMember(key), nil
 	default:
 		return false, nil
 	}

diff --git a/protocol/messenger_peersyncing_test.go b/protocol/messenger_peersyncing_test.go
@@ -243,7 +243,7 @@ func (s *MessengerPeersyncingSuite) TestCanSyncMessageWith() {
 
 	syncMessage := peersyncing.SyncMessage{
 		ID:        []byte("test-id"),
-		GroupID:   []byte(chat.ID),
+		ChatID:    []byte(chat.ID),
 		Type:      peersyncing.SyncMessageCommunityType,
 		Payload:   []byte("some-payload"),
 		Timestamp: 1,

diff --git a/protocol/peersyncing/peersyncing_test.go b/protocol/peersyncing/peersyncing_test.go
@@ -35,7 +35,7 @@ func (s *PeerSyncingSuite) TestBasic() {
 
 	syncMessage := SyncMessage{
 		ID:        []byte("test-id"),
-		GroupID:   testGroupID,
+		ChatID:    testGroupID,
 		Type:      SyncMessageCommunityType,
 		Payload:   []byte("test"),
 		Timestamp: 1,
@@ -48,7 +48,7 @@ func (s *PeerSyncingSuite) TestBasic() {
 	s.Require().NoError(err)
 	s.Require().Len(allMessages, 1)
 
-	byGroupID, err := s.p.AvailableMessagesByGroupID(syncMessage.GroupID, 10)
+	byGroupID, err := s.p.AvailableMessagesByGroupID(syncMessage.ChatID, 10)
 
 	s.Require().NoError(err)
 	s.Require().Len(byGroupID, 1)
@@ -60,7 +60,7 @@ func (s *PeerSyncingSuite) TestBasic() {
 
 	newSyncMessage := SyncMessage{
 		ID:        []byte("test-id-2"),
-		GroupID:   testGroupID,
+		ChatID:    testGroupID,
 		Type:      SyncMessageCommunityType,
 		Payload:   []byte("test-2"),
 		Timestamp: 2,
@@ -77,31 +77,31 @@ func (s *PeerSyncingSuite) TestOrderAndLimit() {
 
 	syncMessage1 := SyncMessage{
 		ID:        []byte("test-id-1"),
-		GroupID:   testGroupID,
+		ChatID:    testGroupID,
 		Type:      SyncMessageCommunityType,
 		Payload:   []byte("test"),
 		Timestamp: 1,
 	}
 
 	syncMessage2 := SyncMessage{
 		ID:        []byte("test-id-2"),
-		GroupID:   testGroupID,
+		ChatID:    testGroupID,
 		Type:      SyncMessageCommunityType,
 		Payload:   []byte("test"),
 		Timestamp: 2,
 	}
 
 	syncMessage3 := SyncMessage{
 		ID:        []byte("test-id-3"),
-		GroupID:   testGroupID,
+		ChatID:    testGroupID,
 		Type:      SyncMessageCommunityType,
 		Payload:   []byte("test"),
 		Timestamp: 3,
 	}
 
 	syncMessage4 := SyncMessage{
 		ID:        []byte("test-id-4"),
-		GroupID:   testGroupID,
+		ChatID:    testGroupID,
 		Type:      SyncMessageCommunityType,
 		Payload:   []byte("test"),
 		Timestamp: 4,

diff --git a/protocol/peersyncing/sync_message.go b/protocol/peersyncing/sync_message.go
@@ -7,15 +7,15 @@ type SyncMessageType int
 type SyncMessage struct {
 	ID        []byte
 	Type      SyncMessageType
-	GroupID   []byte
+	ChatID    []byte
 	Payload   []byte
 	Timestamp uint64
 }
 
 var ErrSyncMessageNotValid = errors.New("sync message not valid")
 
 func (s *SyncMessage) Valid() error {
-	valid := len(s.ID) != 0 && s.Type != SyncMessageNoType && len(s.GroupID) != 0 && len(s.Payload) != 0 && s.Timestamp != 0
+	valid := len(s.ID) != 0 && s.Type != SyncMessageNoType && len(s.ChatID) != 0 && len(s.Payload) != 0 && s.Timestamp != 0
 	if !valid {
 		return ErrSyncMessageNotValid
 	}

diff --git a/protocol/peersyncing/sync_message_persistence.go b/protocol/peersyncing/sync_message_persistence.go
@@ -28,7 +28,7 @@ func (p *SyncMessageSQLitePersistence) Add(message SyncMessage) error {
 	if err := message.Valid(); err != nil {
 		return err
 	}
-	_, err := p.db.Exec(`INSERT INTO peersyncing_messages (id, type, group_id, payload, timestamp) VALUES (?, ?, ?, ?, ?)`, message.ID, message.Type, message.GroupID, message.Payload, message.Timestamp)
+	_, err := p.db.Exec(`INSERT INTO peersyncing_messages (id, type, group_id, payload, timestamp) VALUES (?, ?, ?, ?, ?)`, message.ID, message.Type, message.ChatID, message.Payload, message.Timestamp)
 	return err
 }
 
@@ -44,7 +44,7 @@ func (p *SyncMessageSQLitePersistence) All() ([]SyncMessage, error) {
 	for rows.Next() {
 		var m SyncMessage
 
-		err := rows.Scan(&m.ID, &m.Type, &m.GroupID, &m.Payload, &m.Timestamp)
+		err := rows.Scan(&m.ID, &m.Type, &m.ChatID, &m.Payload, &m.Timestamp)
 		if err != nil {
 			return nil, err
 		}
@@ -66,7 +66,7 @@ func (p *SyncMessageSQLitePersistence) ByGroupID(groupID []byte, limit int) ([]S
 	for rows.Next() {
 		var m SyncMessage
 
-		err := rows.Scan(&m.ID, &m.Type, &m.GroupID, &m.Payload, &m.Timestamp)
+		err := rows.Scan(&m.ID, &m.Type, &m.ChatID, &m.Payload, &m.Timestamp)
 		if err != nil {
 			return nil, err
 		}
@@ -100,7 +100,7 @@ func (p *SyncMessageSQLitePersistence) Complement(messages []SyncMessage) ([]Syn
 	for rows.Next() {
 		var m SyncMessage
 
-		err := rows.Scan(&m.ID, &m.Type, &m.GroupID, &m.Payload, &m.Timestamp)
+		err := rows.Scan(&m.ID, &m.Type, &m.ChatID, &m.Payload, &m.Timestamp)
 		if err != nil {
 			return nil, err
 		}
@@ -145,7 +145,7 @@ func (p *SyncMessageSQLitePersistence) ByGroupIDs(ids [][]byte, limit int) ([]Sy
 	for rows.Next() {
 		var m SyncMessage
 
-		err := rows.Scan(&m.ID, &m.Type, &m.GroupID, &m.Payload, &m.Timestamp)
+		err := rows.Scan(&m.ID, &m.Type, &m.ChatID, &m.Payload, &m.Timestamp)
 		if err != nil {
 			return nil, err
 		}
@@ -180,7 +180,7 @@ func (p *SyncMessageSQLitePersistence) ByMessageIDs(ids [][]byte) ([]SyncMessage
 	for rows.Next() {
 		var m SyncMessage
 
-		err := rows.Scan(&m.ID, &m.Type, &m.GroupID, &m.Payload, &m.Timestamp)
+		err := rows.Scan(&m.ID, &m.Type, &m.ChatID, &m.Payload, &m.Timestamp)
 		if err != nil {
 			return nil, err
 		}