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")
@@ -227,12 +227,13 @@ func (s *MessengerRawMessageResendTest) TestMessageResend() {
 	rawMessage, err := s.bobMessenger.RawMessageByID(ids[0])
 	s.Require().NoError(err)
 	s.Require().NotNil(rawMessage)
-	s.Require().NoError(s.bobMessenger.UpdateRawMessageSent(rawMessage.ID, false, 0))
+	s.Require().NoError(s.bobMessenger.UpdateRawMessageSent(rawMessage.ID, false))
+	s.Require().NoError(s.bobMessenger.UpdateRawMessageLastSent(rawMessage.ID, 0))
 	err = tt.RetryWithBackOff(func() error {
 		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/common/raw_messages_persistence.go

@@ -488,7 +488,12 @@ func (db *RawMessagesPersistence) RemoveMessageSegmentsCompletedOlderThan(timest
 	return err
 }
 
-func (db RawMessagesPersistence) UpdateRawMessageSent(id string, sent bool, lastSent uint64) error {
-	_, err := db.db.Exec("UPDATE raw_messages SET sent = ?, last_sent = ? WHERE id = ?", sent, lastSent, id)
+func (db RawMessagesPersistence) UpdateRawMessageSent(id string, sent bool) error {
+	_, err := db.db.Exec("UPDATE raw_messages SET sent = ? WHERE id = ?", sent, id)
+	return err
+}
+
+func (db RawMessagesPersistence) UpdateRawMessageLastSent(id string, lastSent uint64) error {
+	_, err := db.db.Exec("UPDATE raw_messages SET last_sent = ? WHERE id = ?", lastSent, id)
 	return err
 }

protocol/common/raw_messages_persistence_test.go

@@ -54,30 +54,59 @@ func TestUpdateRawMessageSent(t *testing.T) {
 	require.NoError(t, err)
 
 	rawMessageID := "1"
-	err = p.SaveRawMessage(&RawMessage{
-		ID:                    rawMessageID,
-		ResendType:            ResendTypeRawMessage,
-		LocalChatID:           "",
-		CommunityID:           []byte("c1"),
-		CommunityKeyExMsgType: KeyExMsgRekey,
-		Sender:                pk,
-		ResendMethod:          ResendMethodSendPrivate,
-		Recipients:            []*ecdsa.PublicKey{pk.Public().(*ecdsa.PublicKey)},
-		Sent:                  true,
-		LastSent:              uint64(time.Now().UnixNano() / int64(time.Millisecond)),
-	})
+	err = p.SaveRawMessage(buildRawMessage(rawMessageID, pk))
 	require.NoError(t, err)
 
 	rawMessage, err := p.RawMessageByID(rawMessageID)
 	require.NoError(t, err)
 	require.True(t, rawMessage.Sent)
 	require.Greater(t, rawMessage.LastSent, uint64(0))
 
-	err = p.UpdateRawMessageSent(rawMessageID, false, 0)
+	err = p.UpdateRawMessageSent(rawMessageID, false)
 	require.NoError(t, err)
 
 	m, err := p.RawMessageByID(rawMessageID)
 	require.NoError(t, err)
 	require.False(t, m.Sent)
+}
+
+func TestUpdateRawMessageLastSent(t *testing.T) {
+	db, err := helpers.SetupTestMemorySQLDB(appdatabase.DbInitializer{})
+	require.NoError(t, err)
+	require.NoError(t, sqlite.Migrate(db))
+	p := NewRawMessagesPersistence(db)
+
+	pk, err := crypto.GenerateKey()
+	require.NoError(t, err)
+
+	rawMessageID := "1"
+	err = p.SaveRawMessage(buildRawMessage(rawMessageID, pk))
+	require.NoError(t, err)
+
+	rawMessage, err := p.RawMessageByID(rawMessageID)
+	require.NoError(t, err)
+	require.True(t, rawMessage.Sent)
+	require.Greater(t, rawMessage.LastSent, uint64(0))
+
+	err = p.UpdateRawMessageLastSent(rawMessageID, 0)
+	require.NoError(t, err)
+
+	m, err := p.RawMessageByID(rawMessageID)
+	require.NoError(t, err)
 	require.Equal(t, m.LastSent, uint64(0))
 }
+
+func buildRawMessage(rawMessageID string, pk *ecdsa.PrivateKey) *RawMessage {
+	return &RawMessage{
+		ID:                    rawMessageID,
+		ResendType:            ResendTypeRawMessage,
+		LocalChatID:           "",
+		CommunityID:           []byte("c1"),
+		CommunityKeyExMsgType: KeyExMsgRekey,
+		Sender:                pk,
+		ResendMethod:          ResendMethodSendPrivate,
+		Recipients:            []*ecdsa.PublicKey{pk.Public().(*ecdsa.PublicKey)},
+		Sent:                  true,
+		LastSent:              uint64(time.Now().UnixNano() / int64(time.Millisecond)),
+	}
+}

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

@@ -38,11 +38,20 @@ func (m *Messenger) markDeliveredMessages(acks [][]byte) {
 		messageID := messageIDBytes.String()
 		//mark messages as delivered
 
+		m.logger.Debug("got datasync acknowledge for message", zap.String("ack", hex.EncodeToString(ack)), zap.String("messageID", messageID))
+
 		err = m.UpdateMessageOutgoingStatus(messageID, common.OutgoingStatusDelivered)
 		if err != nil {
 			m.logger.Debug("Can't set message status as delivered", zap.Error(err))
 		}
 
+		err = m.UpdateRawMessageSent(messageID, true)
+		if err != nil {
+			m.logger.Debug("can't set raw message as sent", 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/messenger_raw_message_resend.go

@@ -200,6 +200,10 @@ func (m *Messenger) RawMessageByID(id string) (*common.RawMessage, error) {
 	return m.persistence.RawMessageByID(id)
 }
 
-func (m *Messenger) UpdateRawMessageSent(id string, sent bool, lastSent uint64) error {
-	return m.persistence.UpdateRawMessageSent(id, sent, lastSent)
+func (m *Messenger) UpdateRawMessageSent(id string, sent bool) error {
+	return m.persistence.UpdateRawMessageSent(id, sent)
+}
+
+func (m *Messenger) UpdateRawMessageLastSent(id string, lastSent uint64) error {
+	return m.persistence.UpdateRawMessageLastSent(id, lastSent)
 }

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)