fix: code review

cmd/waku/server/rest/filter.go

@@ -10,6 +10,7 @@ import (
 
 	"github.com/go-chi/chi/v5"
 	"github.com/libp2p/go-libp2p/core/peer"
+	"github.com/waku-org/go-waku/logging"
 	"github.com/waku-org/go-waku/waku/v2/node"
 	"github.com/waku-org/go-waku/waku/v2/peermanager"
 	"github.com/waku-org/go-waku/waku/v2/protocol"
@@ -198,7 +199,11 @@ func (s *FilterService) unsubscribeGetMessage(ch <-chan filter.WakuFilterPushRes
 	var peerIds string
 	ind := 0
 	for entry := range ch {
-		s.log.Error("can't unsubscribe for ", zap.String("peer", entry.PeerID.String()), zap.Error(entry.Err))
+		if entry.Err == nil {
+			continue
+		}
+
+		s.log.Error("can't unsubscribe", logging.HostID("peer", entry.PeerID), zap.Error(entry.Err))
 		if ind != 0 {
 			peerIds += ", "
 		}

cmd/waku/server/rest/lightpush_rest_test.go

@@ -49,7 +49,6 @@ func TestLightpushMessagev1(t *testing.T) {
 		Message: &pb.WakuMessage{
 			Payload:      []byte{1, 2, 3},
 			ContentTopic: "abc",
-			Version:      0,
 			Timestamp:    utils.GetUnixEpoch(),
 		},
 	}

cmd/waku/server/rest/relay_test.go

@@ -18,6 +18,7 @@ import (
 	"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/utils"
+	"google.golang.org/protobuf/proto"
 )
 
 func makeRelayService(t *testing.T, mux *chi.Mux) *RelayService {
@@ -38,7 +39,6 @@ func TestPostV1Message(t *testing.T) {
 	msg := &pb.WakuMessage{
 		Payload:      []byte{1, 2, 3},
 		ContentTopic: "abc",
-		Version:      0,
 		Timestamp:    utils.GetUnixEpoch(),
 	}
 	msgJSONBytes, err := json.Marshal(msg)
@@ -73,17 +73,17 @@ func TestRelaySubscription(t *testing.T) {
 	require.Equal(t, "true", rr.Body.String())
 
 	// Test max messages in subscription
-	now := utils.GetUnixEpoch()
-	d.runner.broadcaster.Submit(protocol.NewEnvelope(tests.CreateWakuMessage("test", now+1), now, "test"))
-	d.runner.broadcaster.Submit(protocol.NewEnvelope(tests.CreateWakuMessage("test", now+2), now, "test"))
-	d.runner.broadcaster.Submit(protocol.NewEnvelope(tests.CreateWakuMessage("test", now+3), now, "test"))
+	now := *utils.GetUnixEpoch()
+	d.runner.broadcaster.Submit(protocol.NewEnvelope(tests.CreateWakuMessage("test", proto.Int64(now+1)), now, "test"))
+	d.runner.broadcaster.Submit(protocol.NewEnvelope(tests.CreateWakuMessage("test", proto.Int64(now+2)), now, "test"))
+	d.runner.broadcaster.Submit(protocol.NewEnvelope(tests.CreateWakuMessage("test", proto.Int64(now+3)), now, "test"))
 
 	// Wait for the messages to be processed
 	time.Sleep(500 * time.Millisecond)
 
 	require.Len(t, d.messages["test"], 3)
 
-	d.runner.broadcaster.Submit(protocol.NewEnvelope(tests.CreateWakuMessage("test", now+4), now+4, "test"))
+	d.runner.broadcaster.Submit(protocol.NewEnvelope(tests.CreateWakuMessage("test", proto.Int64(now+4)), now+4, "test"))
 
 	time.Sleep(500 * time.Millisecond)
 
@@ -139,7 +139,6 @@ func TestRelayGetV1Messages(t *testing.T) {
 	msg := &pb.WakuMessage{
 		Payload:      []byte{1, 2, 3},
 		ContentTopic: "test",
-		Version:      0,
 		Timestamp:    utils.GetUnixEpoch(),
 	}
 	msgJsonBytes, err := json.Marshal(msg)

cmd/waku/server/rest/store.go

@@ -39,7 +39,7 @@ type HistoryCursor struct {
 type StoreWakuMessage struct {
 	Payload      []byte `json:"payload"`
 	ContentTopic string `json:"content_topic"`
-	Version      int32  `json:"version"`
+	Version      uint32 `json:"version"`
 	Timestamp    int64  `json:"timestamp"`
 	Meta         []byte `json:"meta"`
 }
@@ -83,18 +83,20 @@ func getStoreParams(r *http.Request) (multiaddr.Multiaddr, *store.Query, []store
 
 	startTimeStr := r.URL.Query().Get("startTime")
 	if startTimeStr != "" {
-		query.StartTime, err = strconv.ParseInt(startTimeStr, 10, 64)
+		startTime, err := strconv.ParseInt(startTimeStr, 10, 64)
 		if err != nil {
 			return nil, nil, nil, err
 		}
+		query.StartTime = &startTime
 	}
 
 	endTimeStr := r.URL.Query().Get("endTime")
 	if endTimeStr != "" {
-		query.EndTime, err = strconv.ParseInt(endTimeStr, 10, 64)
+		endTime, err := strconv.ParseInt(endTimeStr, 10, 64)
 		if err != nil {
 			return nil, nil, nil, err
 		}
+		query.EndTime = &endTime
 	}
 
 	var cursor *pb.Index
@@ -178,8 +180,8 @@ func toStoreResponse(result *store.Result) StoreResponse {
 		response.Messages = append(response.Messages, StoreWakuMessage{
 			Payload:      m.Payload,
 			ContentTopic: m.ContentTopic,
-			Version:      int32(m.Version),
-			Timestamp:    m.Timestamp,
+			Version:      m.GetVersion(),
+			Timestamp:    m.GetTimestamp(),
 			Meta:         m.Meta,
 		})
 	}

cmd/waku/server/rest/store_test.go

@@ -17,6 +17,7 @@ import (
 	"github.com/waku-org/go-waku/waku/v2/node"
 	"github.com/waku-org/go-waku/waku/v2/protocol"
 	"github.com/waku-org/go-waku/waku/v2/utils"
+	"google.golang.org/protobuf/proto"
 )
 
 func TestGetMessages(t *testing.T) {
@@ -32,14 +33,14 @@ func TestGetMessages(t *testing.T) {
 	topic1 := "1"
 	pubsubTopic1 := "topic1"
 
-	now := utils.GetUnixEpoch()
-	msg1 := tests.CreateWakuMessage(topic1, now+1)
-	msg2 := tests.CreateWakuMessage(topic1, now+2)
-	msg3 := tests.CreateWakuMessage(topic1, now+3)
+	now := *utils.GetUnixEpoch()
+	msg1 := tests.CreateWakuMessage(topic1, proto.Int64(now+1))
+	msg2 := tests.CreateWakuMessage(topic1, proto.Int64(now+2))
+	msg3 := tests.CreateWakuMessage(topic1, proto.Int64(now+3))
 
-	node1.Broadcaster().Submit(protocol.NewEnvelope(msg1, utils.GetUnixEpoch(), pubsubTopic1))
-	node1.Broadcaster().Submit(protocol.NewEnvelope(msg2, utils.GetUnixEpoch(), pubsubTopic1))
-	node1.Broadcaster().Submit(protocol.NewEnvelope(msg3, utils.GetUnixEpoch(), pubsubTopic1))
+	node1.Broadcaster().Submit(protocol.NewEnvelope(msg1, *utils.GetUnixEpoch(), pubsubTopic1))
+	node1.Broadcaster().Submit(protocol.NewEnvelope(msg2, *utils.GetUnixEpoch(), pubsubTopic1))
+	node1.Broadcaster().Submit(protocol.NewEnvelope(msg3, *utils.GetUnixEpoch(), pubsubTopic1))
 
 	n1HostInfo, _ := multiaddr.NewMultiaddr(fmt.Sprintf("/p2p/%s", node1.Host().ID().Pretty()))
 	n1Addr := node1.ListenAddresses()[0].Encapsulate(n1HostInfo)

cmd/waku/server/rpc/admin_test.go

@@ -33,9 +33,12 @@ func TestV1Peers(t *testing.T) {
 	port, err := tests.FindFreePort(t, "", 5)
 	require.NoError(t, err)
 
+	broadcaster := relay.NewBroadcaster(10)
+	require.NoError(t, broadcaster.Start(context.Background()))
+
 	host, err := tests.MakeHost(context.Background(), port, rand.Reader)
 	require.NoError(t, err)
-	relay := relay.NewWakuRelay(nil, 0, timesource.NewDefaultClock(), prometheus.DefaultRegisterer, utils.Logger())
+	relay := relay.NewWakuRelay(broadcaster, 0, timesource.NewDefaultClock(), prometheus.DefaultRegisterer, utils.Logger())
 	relay.SetHost(host)
 	err = relay.Start(context.Background())
 	require.NoError(t, err)

cmd/waku/server/rpc/filter.go

@@ -125,11 +125,13 @@ func (f *FilterService) GetV1Messages(req *http.Request, args *ContentTopicArgs,
 	}
 
 	for i := range f.messages[args.ContentTopic] {
-		rpcMsg, err := ProtoToRPC(f.messages[args.ContentTopic][i])
+		msg := f.messages[args.ContentTopic][i]
+		rpcMsg, err := ProtoToRPC(msg)
 		if err != nil {
-			return err
+			f.log.Warn("could not include message in response", zap.Error(err))
+		} else {
+			*reply = append(*reply, rpcMsg)
 		}
-		*reply = append(*reply, rpcMsg)
 	}
 
 	f.messages[args.ContentTopic] = make([]*wpb.WakuMessage, 0)

cmd/waku/server/rpc/relay.go

@@ -7,6 +7,7 @@ import (
 	"sync"
 
 	"github.com/waku-org/go-waku/cmd/waku/server"
+	"github.com/waku-org/go-waku/logging"
 	"github.com/waku-org/go-waku/waku/v2/node"
 	"github.com/waku-org/go-waku/waku/v2/protocol"
 	"github.com/waku-org/go-waku/waku/v2/protocol/pb"
@@ -194,9 +195,10 @@ func (r *RelayService) GetV1AutoMessages(req *http.Request, args *TopicArgs, rep
 	case msg := <-sub.Ch:
 		rpcMsg, err := ProtoToRPC(msg.Message())
 		if err != nil {
-			return err
+			r.log.Warn("could not include message in response", logging.HexString("hash", msg.Hash()), zap.Error(err))
+		} else {
+			*reply = append(*reply, rpcMsg)
 		}
-		*reply = append(*reply, rpcMsg)
 	default:
 		break
 	}

cmd/waku/server/rpc/relay_test.go

@@ -31,7 +31,6 @@ func TestPostV1Message(t *testing.T) {
 	msg := &pb.WakuMessage{
 		Payload:      []byte{1, 2, 3},
 		ContentTopic: "abc",
-		Version:      0,
 		Timestamp:    utils.GetUnixEpoch(),
 	}
 

cmd/waku/server/rpc/store.go

@@ -25,11 +25,11 @@ type StorePagingOptions struct {
 }
 
 type StoreMessagesArgs struct {
-	Topic          string             `json:"pubsubTopic,omitempty"`
-	ContentFilters []string           `json:"contentFilters,omitempty"`
-	StartTime      int64              `json:"startTime,omitempty"`
-	EndTime        int64              `json:"endTime,omitempty"`
-	PagingOptions  StorePagingOptions `json:"pagingOptions,omitempty"`
+	Topic          string              `json:"pubsubTopic,omitempty"`
+	ContentFilters []string            `json:"contentFilters,omitempty"`
+	StartTime      *int64              `json:"startTime,omitempty"`
+	EndTime        *int64              `json:"endTime,omitempty"`
+	PagingOptions  *StorePagingOptions `json:"pagingOptions,omitempty"`
 }
 
 type StoreMessagesReply struct {

cmd/waku/server/rpc/utils.go

@@ -2,6 +2,8 @@ package rpc
 
 import (
 	"github.com/waku-org/go-waku/waku/v2/protocol/pb"
+	rlnpb "github.com/waku-org/go-waku/waku/v2/protocol/rln/pb"
+
 	"google.golang.org/protobuf/proto"
 )
 
@@ -32,18 +34,18 @@ func ProtoToRPC(input *pb.WakuMessage) (*RPCWakuMessage, error) {
 	rpcWakuMsg := &RPCWakuMessage{
 		Payload:      input.Payload,
 		ContentTopic: input.ContentTopic,
-		Version:      input.Version,
-		Timestamp:    input.Timestamp,
-		Ephemeral:    input.Ephemeral,
-	}
-
-	rateLimitProof := &pb.RateLimitProof{}
-	err := proto.Unmarshal(input.RateLimitProof, rateLimitProof)
-	if err != nil {
-		return nil, err
+		Version:      input.GetVersion(),
+		Timestamp:    input.GetTimestamp(),
+		Ephemeral:    input.GetEphemeral(),
 	}
 
 	if input.RateLimitProof != nil {
+		rateLimitProof := &rlnpb.RateLimitProof{}
+		err := proto.Unmarshal(input.RateLimitProof, rateLimitProof)
+		if err != nil {
+			return nil, err
+		}
+
 		rpcWakuMsg.RateLimitProof = &RateLimitProof{
 			Proof:         rateLimitProof.Proof,
 			MerkleRoot:    rateLimitProof.MerkleRoot,
@@ -66,13 +68,13 @@ func (r *RPCWakuMessage) toProto() (*pb.WakuMessage, error) {
 	msg := &pb.WakuMessage{
 		Payload:      r.Payload,
 		ContentTopic: r.ContentTopic,
-		Version:      r.Version,
-		Timestamp:    r.Timestamp,
-		Ephemeral:    r.Ephemeral,
+		Version:      proto.Uint32(r.Version),
+		Timestamp:    proto.Int64(r.Timestamp),
+		Ephemeral:    proto.Bool(r.Ephemeral),
 	}
 
 	if r.RateLimitProof != nil {
-		rateLimitProof := &pb.RateLimitProof{
+		rateLimitProof := &rlnpb.RateLimitProof{
 			Proof:         r.RateLimitProof.Proof,
 			MerkleRoot:    r.RateLimitProof.MerkleRoot,
 			Epoch:         r.RateLimitProof.Epoch,

examples/basic2/main.go

@@ -19,6 +19,7 @@ import (
 	"github.com/waku-org/go-waku/waku/v2/protocol/relay"
 	"github.com/waku-org/go-waku/waku/v2/utils"
 	"go.uber.org/zap"
+	"google.golang.org/protobuf/proto"
 )
 
 var log = utils.Logger().Named("basic2")
@@ -84,7 +85,6 @@ func randomHex(n int) (string, error) {
 
 func write(ctx context.Context, wakuNode *node.WakuNode, contentTopic string, msgContent string) {
 	var version uint32 = 0
-	var timestamp int64 = utils.GetUnixEpoch(wakuNode.Timesource())
 
 	p := new(payload.Payload)
 	p.Data = []byte(wakuNode.ID() + ": " + msgContent)
@@ -98,9 +98,9 @@ func write(ctx context.Context, wakuNode *node.WakuNode, contentTopic string, ms
 
 	msg := &pb.WakuMessage{
 		Payload:      payload,
-		Version:      version,
+		Version:      proto.Uint32(version),
 		ContentTopic: contentTopic,
-		Timestamp:    timestamp,
+		Timestamp:    utils.GetUnixEpoch(wakuNode.Timesource()),
 	}
 
 	_, err = wakuNode.Relay().Publish(ctx, msg, relay.WithDefaultPubsubTopic())

examples/chat2/chat.go

@@ -302,7 +302,7 @@ func (c *Chat) publish(ctx context.Context, message string) error {
 
 	wakuMsg := &wpb.WakuMessage{
 		Payload:      payload,
-		Version:      version,
+		Version:      proto.Uint32(version),
 		ContentTopic: options.ContentTopic,
 		Timestamp:    timestamp,
 	}
@@ -407,7 +407,7 @@ func (c *Chat) retrieveHistory(connectionWg *sync.WaitGroup) {
 			c.ui.InfoMessage("0 historic messages available")
 		} else {
 			for _, msg := range response.Messages {
-				c.C <- protocol.NewEnvelope(msg, msg.Timestamp, relay.DefaultWakuTopic)
+				c.C <- protocol.NewEnvelope(msg, msg.GetTimestamp(), relay.DefaultWakuTopic)
 			}
 		}
 	}

examples/filter2/main.go

@@ -21,6 +21,7 @@ import (
 	"github.com/waku-org/go-waku/waku/v2/protocol/pb"
 	"github.com/waku-org/go-waku/waku/v2/protocol/relay"
 	"github.com/waku-org/go-waku/waku/v2/utils"
+	"google.golang.org/protobuf/proto"
 )
 
 var log = logging.Logger("filter2")
@@ -144,7 +145,6 @@ func randomHex(n int) (string, error) {
 
 func write(ctx context.Context, wakuNode *node.WakuNode, msgContent string) {
 	var version uint32 = 0
-	var timestamp int64 = utils.GetUnixEpoch(wakuNode.Timesource())
 
 	p := new(payload.Payload)
 	p.Data = []byte(wakuNode.ID() + ": " + msgContent)
@@ -154,9 +154,9 @@ func write(ctx context.Context, wakuNode *node.WakuNode, msgContent string) {
 
 	msg := &pb.WakuMessage{
 		Payload:      payload,
-		Version:      version,
+		Version:      proto.Uint32(version),
 		ContentTopic: contentTopic,
-		Timestamp:    timestamp,
+		Timestamp:    utils.GetUnixEpoch(wakuNode.Timesource()),
 	}
 
 	_, err := wakuNode.Relay().Publish(ctx, msg, relay.WithPubSubTopic(pubSubTopic.String()))

examples/rln/main.go

@@ -20,6 +20,7 @@ import (
 	"github.com/waku-org/go-waku/waku/v2/protocol/relay"
 	"github.com/waku-org/go-waku/waku/v2/utils"
 	"go.uber.org/zap"
+	"google.golang.org/protobuf/proto"
 )
 
 var log = utils.Logger().Named("rln")
@@ -105,7 +106,6 @@ func randomHex(n int) (string, error) {
 
 func write(ctx context.Context, wakuNode *node.WakuNode, msgContent string) {
 	var version uint32 = 0
-	var timestamp int64 = utils.GetUnixEpoch(wakuNode.Timesource())
 
 	p := new(payload.Payload)
 	p.Data = []byte(wakuNode.ID() + ": " + msgContent)
@@ -119,9 +119,9 @@ func write(ctx context.Context, wakuNode *node.WakuNode, msgContent string) {
 
 	msg := &pb.WakuMessage{
 		Payload:      payload,
-		Version:      version,
+		Version:      proto.Uint32(version),
 		ContentTopic: contentTopic.String(),
-		Timestamp:    timestamp,
+		Timestamp:    utils.GetUnixEpoch(wakuNode.Timesource()),
 	}
 
 	err = wakuNode.RLNRelay().AppendRLNProof(msg, wakuNode.Timesource().Now())