feat_: hash based query for outgoing messages.

api/messenger_raw_message_resend_test.go

@@ -211,7 +211,7 @@ func (s *MessengerRawMessageResendTest) TestMessageSent() {
 		rawMessage, err := s.bobMessenger.RawMessageByID(ids[0])
 		s.Require().NoError(err)
 		s.Require().NotNil(rawMessage)
-		if rawMessage.Sent {
+		if rawMessage.SendCount > 0 {
 			return nil
 		}
 		return errors.New("raw message should be sent finally")
@@ -232,7 +232,7 @@ func (s *MessengerRawMessageResendTest) TestMessageResend() {
 		rawMessage, err := s.bobMessenger.RawMessageByID(ids[0])
 		s.Require().NoError(err)
 		s.Require().NotNil(rawMessage)
-		if !rawMessage.Sent {
+		if rawMessage.SendCount < 2 {
 			return errors.New("message ApplicationMetadataMessage_COMMUNITY_REQUEST_TO_JOIN was not resent yet")
 		}
 		return nil

eth-node/bridge/geth/waku.go

@@ -314,3 +314,9 @@ func GetWakuFilterFrom(f types.Filter) *wakucommon.Filter {
 func (w *wakuFilterWrapper) ID() string {
 	return w.id
 }
+
+func (w *GethWakuWrapper) ConfirmMessageDelivered(hashes []common.Hash) {
+}
+
+func (w *GethWakuWrapper) SetStorePeerID(peerID peer.ID) {
+}

eth-node/bridge/geth/wakuv2.go

@@ -333,3 +333,11 @@ func GetWakuV2FilterFrom(f types.Filter) *wakucommon.Filter {
 func (w *wakuV2FilterWrapper) ID() string {
 	return w.id
 }
+
+func (w *gethWakuV2Wrapper) ConfirmMessageDelivered(hashes []common.Hash) {
+	w.waku.ConfirmMessageDelivered(hashes)
+}
+
+func (w *gethWakuV2Wrapper) SetStorePeerID(peerID peer.ID) {
+	w.waku.SetStorePeerID(peerID)
+}

eth-node/types/waku.go

@@ -178,4 +178,10 @@ type Waku interface {
 
 	// ClearEnvelopesCache clears waku envelopes cache
 	ClearEnvelopesCache()
+
+	// ConfirmMessageDelivered updates a message has been delivered in waku
+	ConfirmMessageDelivered(hash []common.Hash)
+
+	// SetStorePeerID updates the peer id of store node
+	SetStorePeerID(peerID peer.ID)
 }

protocol/messenger_mailserver_cycle.go

@@ -426,6 +426,11 @@ func (m *Messenger) connectToMailserver(ms mailservers.Mailserver) error {
 			m.logger.Info("mailserver available", zap.String("address", m.mailserverCycle.activeMailserver.UniqueID()))
 			m.EmitMailserverAvailable()
 			signal.SendMailserverAvailable(m.mailserverCycle.activeMailserver.Address, m.mailserverCycle.activeMailserver.ID)
+			peerID, err := m.mailserverCycle.activeMailserver.PeerID()
+			if err != nil {
+				m.logger.Error("could not decode the peer id of mailserver", zap.Error(err))
+			}
+			m.transport.SetStorePeerID(peerID)
 
 			// Query mailserver
 			if m.config.codeControlFlags.AutoRequestHistoricMessages {

protocol/messenger_messages_tracking_test.go

@@ -172,7 +172,7 @@ func (s *MessengerMessagesTrackingSuite) testMessageMarkedAsSent(textSize int) {
 	// Message should be marked as sent eventually
 	err = tt.RetryWithBackOff(func() error {
 		rawMessage, err = s.bob.persistence.RawMessageByID(inputMessage.ID)
-		if err != nil || !rawMessage.Sent {
+		if err != nil || rawMessage.SendCount < 1 {
 			return errors.New("message not marked as sent")
 		}
 		return nil

protocol/messenger_peers.go

@@ -13,7 +13,7 @@ func (m *Messenger) AddStorePeer(address string) (peer.ID, error) {
 }
 
 func (m *Messenger) AddRelayPeer(address string) (peer.ID, error) {
-	return m.transport.AddStorePeer(address)
+	return m.transport.AddRelayPeer(address)
 }
 
 func (m *Messenger) DialPeer(address string) error {

protocol/messenger_peersyncing.go

@@ -43,6 +43,8 @@ func (m *Messenger) markDeliveredMessages(acks [][]byte) {
 			m.logger.Debug("Can't set message status as delivered", zap.Error(err))
 		}
 
+		m.transport.ConfirmMessageDelivered(messageID)
+
 		//send signal to client that message status updated
 		if m.config.messengerSignalsHandler != nil {
 			message, err := m.persistence.MessageByID(messageID)

protocol/transport/envelopes_monitor.go

@@ -136,7 +136,7 @@ func (m *EnvelopesMonitor) Add(messageIDs [][]byte, envelopeHashes []types.Hash,
 	defer m.mu.Unlock()
 
 	for _, messageID := range messageIDs {
-		m.messageEnvelopeHashes[string(messageID)] = envelopeHashes
+		m.messageEnvelopeHashes[types.HexBytes(messageID).String()] = envelopeHashes
 	}
 
 	for i, envelopeHash := range envelopeHashes {
@@ -399,7 +399,7 @@ func (m *EnvelopesMonitor) processMessageIDs(messageIDs [][]byte) {
 	sentMessageIDs := make([][]byte, 0, len(messageIDs))
 
 	for _, messageID := range messageIDs {
-		hashes, ok := m.messageEnvelopeHashes[string(messageID)]
+		hashes, ok := m.messageEnvelopeHashes[types.HexBytes(messageID).String()]
 		if !ok {
 			continue
 		}
@@ -432,6 +432,6 @@ func (m *EnvelopesMonitor) clearMessageState(envelopeID types.Hash) {
 	}
 	delete(m.envelopes, envelopeID)
 	for _, messageID := range envelope.messageIDs {
-		delete(m.messageEnvelopeHashes, string(messageID))
+		delete(m.messageEnvelopeHashes, types.HexBytes(messageID).String())
 	}
 }

protocol/transport/transport.go

@@ -717,3 +717,22 @@ func (t *Transport) RemovePubsubTopicKey(topic string) error {
 	}
 	return nil
 }
+
+func (t *Transport) ConfirmMessageDelivered(messageID string) {
+	if t.envelopesMonitor == nil {
+		return
+	}
+	hashes, ok := t.envelopesMonitor.messageEnvelopeHashes[messageID]
+	if !ok {
+		return
+	}
+	commHashes := make([]common.Hash, len(hashes))
+	for i, h := range hashes {
+		commHashes[i] = common.BytesToHash(h[:])
+	}
+	t.waku.ConfirmMessageDelivered(commHashes)
+}
+
+func (t *Transport) SetStorePeerID(peerID peer.ID) {
+	t.waku.SetStorePeerID(peerID)
+}

wakuv2/common/message.go

@@ -22,6 +22,7 @@ type MessageType int
 const (
 	RelayedMessageType MessageType = iota
 	StoreMessageType
+	SendMessageType
 )
 
 // MessageParams specifies the exact way a message should be wrapped
@@ -46,7 +47,7 @@ type ReceivedMessage struct {
 	Padding   []byte
 	Signature []byte
 
-	Sent uint32           // Time when the message was posted into the network
+	Sent uint32           // Time when the message was posted into the network in seconds
 	Src  *ecdsa.PublicKey // Message recipient (identity used to decode the message)
 	Dst  *ecdsa.PublicKey // Message recipient (identity used to decode the message)