refactor: decouple API from go-waku (#1239)

waku/v2/api/common/result.go

@@ -0,0 +1,17 @@
+package common
+
+import (
+	"context"
+
+	"github.com/libp2p/go-libp2p/core/peer"
+	"github.com/waku-org/go-waku/waku/v2/protocol/store"
+	"github.com/waku-org/go-waku/waku/v2/protocol/store/pb"
+)
+
+type StoreRequestResult interface {
+	Cursor() []byte
+	IsComplete() bool
+	PeerID() peer.ID
+	Next(ctx context.Context, opts ...store.RequestOption) error // TODO: see how to decouple store.RequestOption
+	Messages() []*pb.WakuMessageKeyValue
+}

waku/v2/api/missing/default_requestor.go

@@ -0,0 +1,33 @@
+package missing
+
+import (
+	"context"
+
+	"github.com/libp2p/go-libp2p/core/peer"
+	"github.com/waku-org/go-waku/waku/v2/api/common"
+	"github.com/waku-org/go-waku/waku/v2/protocol"
+	"github.com/waku-org/go-waku/waku/v2/protocol/pb"
+	"github.com/waku-org/go-waku/waku/v2/protocol/store"
+)
+
+func NewDefaultStorenodeRequestor(store *store.WakuStore) StorenodeRequestor {
+	return &defaultStorenodeRequestor{
+		store: store,
+	}
+}
+
+type defaultStorenodeRequestor struct {
+	store *store.WakuStore
+}
+
+func (d *defaultStorenodeRequestor) GetMessagesByHash(ctx context.Context, peerID peer.ID, pageSize uint64, messageHashes []pb.MessageHash) (common.StoreRequestResult, error) {
+	return d.store.QueryByHash(ctx, messageHashes, store.WithPeer(peerID), store.WithPaging(false, pageSize))
+}
+
+func (d *defaultStorenodeRequestor) QueryWithCriteria(ctx context.Context, peerID peer.ID, pageSize uint64, pubsubTopic string, contentTopics []string, from *int64, to *int64) (common.StoreRequestResult, error) {
+	return d.store.Query(ctx, store.FilterCriteria{
+		ContentFilter: protocol.NewContentFilter(pubsubTopic, contentTopics...),
+		TimeStart:     from,
+		TimeEnd:       to,
+	}, store.WithPeer(peerID), store.WithPaging(false, pageSize), store.IncludeData(false))
+}

waku/v2/api/missing/missing_messages.go

@@ -11,9 +11,9 @@ import (
 
 	"github.com/libp2p/go-libp2p/core/peer"
 	"github.com/waku-org/go-waku/logging"
+	"github.com/waku-org/go-waku/waku/v2/api/common"
 	"github.com/waku-org/go-waku/waku/v2/protocol"
 	"github.com/waku-org/go-waku/waku/v2/protocol/pb"
-	"github.com/waku-org/go-waku/waku/v2/protocol/store"
 	"github.com/waku-org/go-waku/waku/v2/timesource"
 	"github.com/waku-org/go-waku/waku/v2/utils"
 	"go.uber.org/zap"
@@ -22,6 +22,7 @@ import (
 
 const maxContentTopicsPerRequest = 10
 const maxMsgHashesPerRequest = 50
+const messageFetchPageSize = 100
 
 // MessageTracker should keep track of messages it has seen before and
 // provide a way to determine whether a message exists or not. This
@@ -30,11 +31,17 @@ type MessageTracker interface {
 	MessageExists(pb.MessageHash) (bool, error)
 }
 
+type StorenodeRequestor interface {
+	GetMessagesByHash(ctx context.Context, peerID peer.ID, pageSize uint64, messageHashes []pb.MessageHash) (common.StoreRequestResult, error)
+	QueryWithCriteria(ctx context.Context, peerID peer.ID, pageSize uint64, pubsubTopic string, contentTopics []string, from *int64, to *int64) (common.StoreRequestResult, error)
+}
+
 // MissingMessageVerifier is used to periodically retrieve missing messages from store nodes that have some specific criteria
 type MissingMessageVerifier struct {
 	ctx    context.Context
 	params missingMessageVerifierParams
 
+	storenodeRequestor StorenodeRequestor
 	messageTracker     MessageTracker
 
 	criteriaInterest   map[string]criteriaInterest // Track message verification requests and when was the last time a pubsub topic was verified for missing messages
@@ -42,13 +49,12 @@ type MissingMessageVerifier struct {
 
 	C <-chan *protocol.Envelope
 
-	store      *store.WakuStore
 	timesource timesource.Timesource
 	logger     *zap.Logger
 }
 
 // NewMissingMessageVerifier creates an instance of a MissingMessageVerifier
-func NewMissingMessageVerifier(store *store.WakuStore, messageTracker MessageTracker, timesource timesource.Timesource, logger *zap.Logger, options ...MissingMessageVerifierOption) *MissingMessageVerifier {
+func NewMissingMessageVerifier(storenodeRequester StorenodeRequestor, messageTracker MessageTracker, timesource timesource.Timesource, logger *zap.Logger, options ...MissingMessageVerifierOption) *MissingMessageVerifier {
 	options = append(defaultMissingMessagesVerifierOptions, options...)
 	params := missingMessageVerifierParams{}
 	for _, opt := range options {
@@ -56,7 +62,7 @@ func NewMissingMessageVerifier(store *store.WakuStore, messageTracker MessageTra
 	}
 
 	return &MissingMessageVerifier{
-		store:          store,
+		storenodeRequestor: storenodeRequester,
 		timesource:         timesource,
 		messageTracker:     messageTracker,
 		logger:             logger.Named("missing-msg-verifier"),
@@ -178,7 +184,7 @@ func (m *MissingMessageVerifier) fetchHistory(c chan<- *protocol.Envelope, inter
 	}
 }
 
-func (m *MissingMessageVerifier) storeQueryWithRetry(ctx context.Context, queryFunc func(ctx context.Context) (store.Result, error), logger *zap.Logger, logMsg string) (store.Result, error) {
+func (m *MissingMessageVerifier) storeQueryWithRetry(ctx context.Context, queryFunc func(ctx context.Context) (common.StoreRequestResult, error), logger *zap.Logger, logMsg string) (common.StoreRequestResult, error) {
 	retry := true
 	count := 1
 	for retry && count <= m.params.maxAttemptsToRetrieveHistory {
@@ -212,12 +218,16 @@ func (m *MissingMessageVerifier) fetchMessagesBatch(c chan<- *protocol.Envelope,
 		logging.Epoch("to", now),
 	)
 
-	result, err := m.storeQueryWithRetry(interest.ctx, func(ctx context.Context) (store.Result, error) {
+	result, err := m.storeQueryWithRetry(interest.ctx, func(ctx context.Context) (common.StoreRequestResult, error) {
-		return m.store.Query(ctx, store.FilterCriteria{
+		return m.storenodeRequestor.QueryWithCriteria(
-			ContentFilter: protocol.NewContentFilter(interest.contentFilter.PubsubTopic, contentTopics[batchFrom:batchTo]...),
+			ctx,
-			TimeStart:     proto.Int64(interest.lastChecked.Add(-m.params.delay).UnixNano()),
+			interest.peerID,
-			TimeEnd:       proto.Int64(now.Add(-m.params.delay).UnixNano()),
+			messageFetchPageSize,
-		}, store.WithPeer(interest.peerID), store.WithPaging(false, 100), store.IncludeData(false))
+			interest.contentFilter.PubsubTopic,
+			contentTopics[batchFrom:batchTo],
+			proto.Int64(interest.lastChecked.Add(-m.params.delay).UnixNano()),
+			proto.Int64(now.Add(-m.params.delay).UnixNano()),
+		)
 	}, logger, "retrieving history to check for missing messages")
 	if err != nil {
 		if !errors.Is(err, context.Canceled) {
@@ -243,7 +253,7 @@ func (m *MissingMessageVerifier) fetchMessagesBatch(c chan<- *protocol.Envelope,
 			missingHashes = append(missingHashes, hash)
 		}
 
-		result, err = m.storeQueryWithRetry(interest.ctx, func(ctx context.Context) (store.Result, error) {
+		result, err = m.storeQueryWithRetry(interest.ctx, func(ctx context.Context) (common.StoreRequestResult, error) {
 			if err = result.Next(ctx); err != nil {
 				return nil, err
 			}
@@ -282,10 +292,10 @@ func (m *MissingMessageVerifier) fetchMessagesBatch(c chan<- *protocol.Envelope,
 			defer utils.LogOnPanic()
 			defer wg.Wait()
 
-			result, err := m.storeQueryWithRetry(interest.ctx, func(ctx context.Context) (store.Result, error) {
+			result, err := m.storeQueryWithRetry(interest.ctx, func(ctx context.Context) (common.StoreRequestResult, error) {
 				queryCtx, cancel := context.WithTimeout(ctx, m.params.storeQueryTimeout)
 				defer cancel()
-				return m.store.QueryByHash(queryCtx, messageHashes, store.WithPeer(interest.peerID), store.WithPaging(false, maxMsgHashesPerRequest))
+				return m.storenodeRequestor.GetMessagesByHash(queryCtx, interest.peerID, maxMsgHashesPerRequest, messageHashes)
 			}, logger, "retrieving missing messages")
 			if err != nil {
 				if !errors.Is(err, context.Canceled) {
@@ -303,7 +313,7 @@ func (m *MissingMessageVerifier) fetchMessagesBatch(c chan<- *protocol.Envelope,
 					}
 				}
 
-				result, err = m.storeQueryWithRetry(interest.ctx, func(ctx context.Context) (store.Result, error) {
+				result, err = m.storeQueryWithRetry(interest.ctx, func(ctx context.Context) (common.StoreRequestResult, error) {
 					if err = result.Next(ctx); err != nil {
 						return nil, err
 					}

waku/v2/api/publish/default_publisher.go

@@ -0,0 +1,50 @@
+package publish
+
+import (
+	"context"
+	"errors"
+
+	"github.com/libp2p/go-libp2p/core/peer"
+	"github.com/waku-org/go-waku/waku/v2/protocol/lightpush"
+	"github.com/waku-org/go-waku/waku/v2/protocol/pb"
+	"github.com/waku-org/go-waku/waku/v2/protocol/relay"
+)
+
+var ErrRelayNotAvailable = errors.New("relay is not available")
+var ErrLightpushNotAvailable = errors.New("lightpush is not available")
+
+func NewDefaultPublisher(lightpush *lightpush.WakuLightPush, relay *relay.WakuRelay) Publisher {
+	return &defaultPublisher{
+		lightpush: lightpush,
+		relay:     relay,
+	}
+}
+
+type defaultPublisher struct {
+	lightpush *lightpush.WakuLightPush
+	relay     *relay.WakuRelay
+}
+
+func (d *defaultPublisher) RelayListPeers(pubsubTopic string) ([]peer.ID, error) {
+	if d.relay == nil {
+		return nil, ErrRelayNotAvailable
+	}
+
+	return d.relay.PubSub().ListPeers(pubsubTopic), nil
+}
+
+func (d *defaultPublisher) RelayPublish(ctx context.Context, message *pb.WakuMessage, pubsubTopic string) (pb.MessageHash, error) {
+	if d.relay == nil {
+		return pb.MessageHash{}, ErrRelayNotAvailable
+	}
+
+	return d.relay.Publish(ctx, message, relay.WithPubSubTopic(pubsubTopic))
+}
+
+func (d *defaultPublisher) LightpushPublish(ctx context.Context, message *pb.WakuMessage, pubsubTopic string, maxPeers int) (pb.MessageHash, error) {
+	if d.lightpush == nil {
+		return pb.MessageHash{}, ErrLightpushNotAvailable
+	}
+
+	return d.lightpush.Publish(ctx, message, lightpush.WithPubSubTopic(pubsubTopic), lightpush.WithMaxPeers(maxPeers))
+}

waku/v2/api/publish/default_verifier.go

@@ -0,0 +1,39 @@
+package publish
+
+import (
+	"context"
+
+	"github.com/libp2p/go-libp2p/core/peer"
+	"github.com/waku-org/go-waku/waku/v2/protocol/pb"
+	"github.com/waku-org/go-waku/waku/v2/protocol/store"
+)
+
+func NewDefaultStorenodeMessageVerifier(store *store.WakuStore) StorenodeMessageVerifier {
+	return &defaultStorenodeMessageVerifier{
+		store: store,
+	}
+}
+
+type defaultStorenodeMessageVerifier struct {
+	store *store.WakuStore
+}
+
+func (d *defaultStorenodeMessageVerifier) MessageHashesExist(ctx context.Context, requestID []byte, peerID peer.ID, pageSize uint64, messageHashes []pb.MessageHash) ([]pb.MessageHash, error) {
+	var opts []store.RequestOption
+	opts = append(opts, store.WithRequestID(requestID))
+	opts = append(opts, store.WithPeer(peerID))
+	opts = append(opts, store.WithPaging(false, pageSize))
+	opts = append(opts, store.IncludeData(false))
+
+	response, err := d.store.QueryByHash(ctx, messageHashes, opts...)
+	if err != nil {
+		return nil, err
+	}
+
+	result := make([]pb.MessageHash, len(response.Messages()))
+	for i, msg := range response.Messages() {
+		result[i] = msg.WakuMessageHash()
+	}
+
+	return result, nil
+}

waku/v2/api/publish/message_check.go

@@ -8,11 +8,11 @@ import (
 
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/common/hexutil"
+	"github.com/libp2p/go-libp2p/core/peer"
 	apicommon "github.com/waku-org/go-waku/waku/v2/api/common"
 	"github.com/waku-org/go-waku/waku/v2/api/history"
 	"github.com/waku-org/go-waku/waku/v2/protocol"
 	"github.com/waku-org/go-waku/waku/v2/protocol/pb"
-	"github.com/waku-org/go-waku/waku/v2/protocol/store"
 	"github.com/waku-org/go-waku/waku/v2/timesource"
 	"github.com/waku-org/go-waku/waku/v2/utils"
 	"go.uber.org/zap"
@@ -31,6 +31,11 @@ type ISentCheck interface {
 	DeleteByMessageIDs(messageIDs []common.Hash)
 }
 
+type StorenodeMessageVerifier interface {
+	// MessagesExist returns a list of the messages it found from a list of message hashes
+	MessageHashesExist(ctx context.Context, requestID []byte, peerID peer.ID, pageSize uint64, messageHashes []pb.MessageHash) ([]pb.MessageHash, error)
+}
+
 // MessageSentCheck tracks the outgoing messages and check against store node
 // if the message sent time has passed the `messageSentPeriod`, the message id will be includes for the next query
 // if the message keeps missing after `messageExpiredPerid`, the message id will be expired
@@ -40,7 +45,7 @@ type MessageSentCheck struct {
 	messageStoredChan   chan common.Hash
 	messageExpiredChan  chan common.Hash
 	ctx                 context.Context
-	store               *store.WakuStore
+	messageVerifier     StorenodeMessageVerifier
 	storenodeCycle      *history.StorenodeCycle
 	timesource          timesource.Timesource
 	logger              *zap.Logger
@@ -52,14 +57,14 @@ type MessageSentCheck struct {
 }
 
 // NewMessageSentCheck creates a new instance of MessageSentCheck with default parameters
-func NewMessageSentCheck(ctx context.Context, store *store.WakuStore, cycle *history.StorenodeCycle, timesource timesource.Timesource, msgStoredChan chan common.Hash, msgExpiredChan chan common.Hash, logger *zap.Logger) *MessageSentCheck {
+func NewMessageSentCheck(ctx context.Context, messageVerifier StorenodeMessageVerifier, cycle *history.StorenodeCycle, timesource timesource.Timesource, msgStoredChan chan common.Hash, msgExpiredChan chan common.Hash, logger *zap.Logger) *MessageSentCheck {
 	return &MessageSentCheck{
 		messageIDs:          make(map[string]map[common.Hash]uint32),
 		messageIDsMu:        sync.RWMutex{},
 		messageStoredChan:   msgStoredChan,
 		messageExpiredChan:  msgExpiredChan,
 		ctx:                 ctx,
-		store:               store,
+		messageVerifier:     messageVerifier,
 		storenodeCycle:      cycle,
 		timesource:          timesource,
 		logger:              logger,
@@ -212,12 +217,7 @@ func (m *MessageSentCheck) messageHashBasedQuery(ctx context.Context, hashes []c
 		return []common.Hash{}
 	}
 
-	var opts []store.RequestOption
 	requestID := protocol.GenerateRequestID()
-	opts = append(opts, store.WithRequestID(requestID))
-	opts = append(opts, store.WithPeer(selectedPeer))
-	opts = append(opts, store.WithPaging(false, m.maxHashQueryLength))
-	opts = append(opts, store.IncludeData(false))
 
 	messageHashes := make([]pb.MessageHash, len(hashes))
 	for i, hash := range hashes {
@@ -228,20 +228,20 @@ func (m *MessageSentCheck) messageHashBasedQuery(ctx context.Context, hashes []c
 
 	queryCtx, cancel := context.WithTimeout(ctx, m.storeQueryTimeout)
 	defer cancel()
-	result, err := m.store.QueryByHash(queryCtx, messageHashes, opts...)
+	result, err := m.messageVerifier.MessageHashesExist(queryCtx, requestID, selectedPeer, m.maxHashQueryLength, messageHashes)
 	if err != nil {
 		m.logger.Error("store.queryByHash failed", zap.String("requestID", hexutil.Encode(requestID)), zap.Stringer("peerID", selectedPeer), zap.Error(err))
 		return []common.Hash{}
 	}
 
-	m.logger.Debug("store.queryByHash result", zap.String("requestID", hexutil.Encode(requestID)), zap.Int("messages", len(result.Messages())))
+	m.logger.Debug("store.queryByHash result", zap.String("requestID", hexutil.Encode(requestID)), zap.Int("messages", len(result)))
 
 	var ackHashes []common.Hash
 	var missedHashes []common.Hash
 	for i, hash := range hashes {
 		found := false
-		for _, msg := range result.Messages() {
+		for _, msgHash := range result {
-			if bytes.Equal(msg.GetMessageHash(), hash.Bytes()) {
+			if bytes.Equal(msgHash.Bytes(), hash.Bytes()) {
 				found = true
 				break
 			}

waku/v2/api/publish/message_sender.go

@@ -6,9 +6,9 @@ import (
 	"time"
 
 	"github.com/ethereum/go-ethereum/common"
+	"github.com/libp2p/go-libp2p/core/peer"
 	"github.com/waku-org/go-waku/waku/v2/protocol"
-	"github.com/waku-org/go-waku/waku/v2/protocol/lightpush"
+	"github.com/waku-org/go-waku/waku/v2/protocol/pb"
-	"github.com/waku-org/go-waku/waku/v2/protocol/relay"
 	"go.uber.org/zap"
 	"golang.org/x/time/rate"
 )
@@ -36,10 +36,20 @@ func (pm PublishMethod) String() string {
 	}
 }
 
+type Publisher interface {
+	// RelayListPeers returns the list of peers for a pubsub topic
+	RelayListPeers(pubsubTopic string) ([]peer.ID, error)
+
+	// RelayPublish publishes a message via WakuRelay
+	RelayPublish(ctx context.Context, message *pb.WakuMessage, pubsubTopic string) (pb.MessageHash, error)
+
+	// LightpushPublish publishes a message via WakuLightPush
+	LightpushPublish(ctx context.Context, message *pb.WakuMessage, pubsubTopic string, maxPeers int) (pb.MessageHash, error)
+}
+
 type MessageSender struct {
 	publishMethod    PublishMethod
-	lightPush        *lightpush.WakuLightPush
+	publisher        Publisher
-	relay            *relay.WakuRelay
 	messageSentCheck ISentCheck
 	rateLimiter      *PublishRateLimiter
 	logger           *zap.Logger
@@ -64,14 +74,13 @@ func (r *Request) WithPublishMethod(publishMethod PublishMethod) *Request {
 	return r
 }
 
-func NewMessageSender(publishMethod PublishMethod, lightPush *lightpush.WakuLightPush, relay *relay.WakuRelay, logger *zap.Logger) (*MessageSender, error) {
+func NewMessageSender(publishMethod PublishMethod, publisher Publisher, logger *zap.Logger) (*MessageSender, error) {
 	if publishMethod == UnknownMethod {
 		return nil, errors.New("publish method is required")
 	}
 	return &MessageSender{
 		publishMethod: publishMethod,
-		lightPush:     lightPush,
+		publisher:     publisher,
-		relay:         relay,
 		rateLimiter:   NewPublishRateLimiter(DefaultPublishingLimiterRate, DefaultPublishingLimitBurst),
 		logger:        logger,
 	}, nil
@@ -108,26 +117,23 @@ func (ms *MessageSender) Send(req *Request) error {
 
 	switch publishMethod {
 	case LightPush:
-		if ms.lightPush == nil {
-			return errors.New("lightpush is not available")
-		}
 		logger.Info("publishing message via lightpush")
-		_, err := ms.lightPush.Publish(
+		_, err := ms.publisher.LightpushPublish(
 			req.ctx,
 			req.envelope.Message(),
-			lightpush.WithPubSubTopic(req.envelope.PubsubTopic()),
+			req.envelope.PubsubTopic(),
-			lightpush.WithMaxPeers(DefaultPeersToPublishForLightpush),
+			DefaultPeersToPublishForLightpush,
 		)
 		if err != nil {
 			return err
 		}
 	case Relay:
-		if ms.relay == nil {
+		peers, err := ms.publisher.RelayListPeers(req.envelope.PubsubTopic())
-			return errors.New("relay is not available")
+		if err != nil {
+			return err
 		}
-		peerCnt := len(ms.relay.PubSub().ListPeers(req.envelope.PubsubTopic()))
+		logger.Info("publishing message via relay", zap.Int("peerCnt", len(peers)))
-		logger.Info("publishing message via relay", zap.Int("peerCnt", peerCnt))
+		_, err = ms.publisher.RelayPublish(req.ctx, req.envelope.Message(), req.envelope.PubsubTopic())
-		_, err := ms.relay.Publish(req.ctx, req.envelope.Message(), relay.WithPubSubTopic(req.envelope.PubsubTopic()))
 		if err != nil {
 			return err
 		}

waku/v2/api/publish/message_sender_test.go

@@ -40,7 +40,7 @@ func (m *MockMessageSentCheck) Start() {
 }
 
 func TestNewSenderWithUnknownMethod(t *testing.T) {
-	sender, err := NewMessageSender(UnknownMethod, nil, nil, nil)
+	sender, err := NewMessageSender(UnknownMethod, nil, nil)
 	require.NotNil(t, err)
 	require.Nil(t, sender)
 }
@@ -53,7 +53,8 @@ func TestNewSenderWithRelay(t *testing.T) {
 
 	_, err = relayNode.Subscribe(context.Background(), protocol.NewContentFilter("test-pubsub-topic"))
 	require.Nil(t, err)
-	sender, err := NewMessageSender(Relay, nil, relayNode, utils.Logger())
+	publisher := NewDefaultPublisher(nil, relayNode)
+	sender, err := NewMessageSender(Relay, publisher, utils.Logger())
 	require.Nil(t, err)
 	require.NotNil(t, sender)
 	require.Nil(t, sender.messageSentCheck)
@@ -78,7 +79,8 @@ func TestNewSenderWithRelayAndMessageSentCheck(t *testing.T) {
 
 	_, err = relayNode.Subscribe(context.Background(), protocol.NewContentFilter("test-pubsub-topic"))
 	require.Nil(t, err)
-	sender, err := NewMessageSender(Relay, nil, relayNode, utils.Logger())
+	publisher := NewDefaultPublisher(nil, relayNode)
+	sender, err := NewMessageSender(Relay, publisher, utils.Logger())
 
 	check := &MockMessageSentCheck{Messages: make(map[string]map[common.Hash]uint32)}
 	sender.WithMessageSentCheck(check)
@@ -108,7 +110,7 @@ func TestNewSenderWithRelayAndMessageSentCheck(t *testing.T) {
 }
 
 func TestNewSenderWithLightPush(t *testing.T) {
-	sender, err := NewMessageSender(LightPush, nil, nil, nil)
+	sender, err := NewMessageSender(LightPush, nil, nil)
 	require.Nil(t, err)
 	require.NotNil(t, sender)
 	require.Equal(t, LightPush, sender.publishMethod)