status-im
/
status-go
mirror of https://github.com/status-im/status-go.git
2
0
Fork 0
Code Issues Projects Releases Wiki Activity

refactor_: start using nwaku

This commit is contained in:
Ivan Folgueira Bande 2024-08-27 15:14:36 +02:00
parent 97f4c6e3df
commit fb1b1a8605
No known key found for this signature in database
GPG Key ID: 3C117481F89E24A7
30 changed files with 2578 additions and 2114 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
eth-node/bridge/geth/node.go
View File

@ -19,10 +19,10 @@ import (
type gethNodeWrapper struct { type gethNodeWrapper struct {
stack *node.Node stack *node.Node
waku1 *waku.Waku waku1 *waku.Waku
waku2 *wakuv2.Waku waku2 *wakuv2.NWaku
} }
func NewNodeBridge(stack *node.Node, waku1 *waku.Waku, waku2 *wakuv2.Waku) types.Node { func NewNodeBridge(stack *node.Node, waku1 *waku.Waku, waku2 *wakuv2.NWaku) types.Node {
return &gethNodeWrapper{stack: stack, waku1: waku1, waku2: waku2} return &gethNodeWrapper{stack: stack, waku1: waku1, waku2: waku2}
} }
@ -38,7 +38,7 @@ func (w *gethNodeWrapper) SetWaku1(waku *waku.Waku) {
w.waku1 = waku w.waku1 = waku
} }
func (w *gethNodeWrapper) SetWaku2(waku *wakuv2.Waku) { func (w *gethNodeWrapper) SetWaku2(waku *wakuv2.NWaku) {
w.waku2 = waku w.waku2 = waku
} }

8
eth-node/bridge/geth/wakuv2.go
View File

@ -22,11 +22,11 @@ import (
) )
type gethWakuV2Wrapper struct { type gethWakuV2Wrapper struct {
waku *wakuv2.Waku waku *wakuv2.NWaku
} }
// NewGethWakuWrapper returns an object that wraps Geth's Waku in a types interface // NewGethWakuWrapper returns an object that wraps Geth's Waku in a types interface
func NewGethWakuV2Wrapper(w *wakuv2.Waku) types.Waku { func NewGethWakuV2Wrapper(w *wakuv2.NWaku) types.Waku {
if w == nil { if w == nil {
panic("waku cannot be nil") panic("waku cannot be nil")
} }
@ -37,7 +37,7 @@ func NewGethWakuV2Wrapper(w *wakuv2.Waku) types.Waku {
} }
// GetGethWhisperFrom retrieves the underlying whisper Whisper struct from a wrapped Whisper interface // GetGethWhisperFrom retrieves the underlying whisper Whisper struct from a wrapped Whisper interface
func GetGethWakuV2From(m types.Waku) *wakuv2.Waku { func GetGethWakuV2From(m types.Waku) *wakuv2.NWaku {
return m.(*gethWakuV2Wrapper).waku return m.(*gethWakuV2Wrapper).waku
} }
@ -275,7 +275,7 @@ func (w *gethWakuV2Wrapper) DialPeerByID(peerID peer.ID) error {
} }
func (w *gethWakuV2Wrapper) ListenAddresses() ([]multiaddr.Multiaddr, error) { func (w *gethWakuV2Wrapper) ListenAddresses() ([]multiaddr.Multiaddr, error) {
return w.waku.ListenAddresses(), nil return w.waku.ListenAddresses()
} }
func (w *gethWakuV2Wrapper) RelayPeersByTopic(topic string) (*types.PeerList, error) { func (w *gethWakuV2Wrapper) RelayPeersByTopic(topic string) (*types.PeerList, error) {

25
node/get_status_node.go
View File

@ -119,18 +119,19 @@ type StatusNode struct {
localNotificationsSrvc *localnotifications.Service localNotificationsSrvc *localnotifications.Service
personalSrvc *personal.Service personalSrvc *personal.Service
timeSourceSrvc *timesource.NTPTimeSource timeSourceSrvc *timesource.NTPTimeSource
wakuSrvc *waku.Waku // nwakuSrvc *
wakuExtSrvc *wakuext.Service wakuSrvc *waku.Waku
wakuV2Srvc *wakuv2.Waku wakuExtSrvc *wakuext.Service
wakuV2ExtSrvc *wakuv2ext.Service wakuV2Srvc *wakuv2.NWaku
ensSrvc *ens.Service wakuV2ExtSrvc *wakuv2ext.Service
communityTokensSrvc *communitytokens.Service ensSrvc *ens.Service
gifSrvc *gif.Service communityTokensSrvc *communitytokens.Service
stickersSrvc *stickers.Service gifSrvc *gif.Service
chatSrvc *chat.Service stickersSrvc *stickers.Service
updatesSrvc *updates.Service chatSrvc *chat.Service
pendingTracker *transactions.PendingTxTracker updatesSrvc *updates.Service
connectorSrvc *connector.Service pendingTracker *transactions.PendingTxTracker
connectorSrvc *connector.Service
walletFeed event.Feed walletFeed event.Feed
} }

7
node/status_node_services.go
View File

@ -10,7 +10,6 @@ import (
"reflect" "reflect"
"time" "time"
"github.com/status-im/status-go/protocol/common/shard"
"github.com/status-im/status-go/server" "github.com/status-im/status-go/server"
"github.com/status-im/status-go/signal" "github.com/status-im/status-go/signal"
"github.com/status-im/status-go/transactions" "github.com/status-im/status-go/transactions"
@ -264,7 +263,7 @@ func (b *StatusNode) WakuExtService() *wakuext.Service {
func (b *StatusNode) WakuV2ExtService() *wakuv2ext.Service { func (b *StatusNode) WakuV2ExtService() *wakuv2ext.Service {
return b.wakuV2ExtSrvc return b.wakuV2ExtSrvc
} }
func (b *StatusNode) WakuV2Service() *wakuv2.Waku { func (b *StatusNode) WakuV2Service() *wakuv2.NWaku {
return b.wakuV2Srvc return b.wakuV2Srvc
} }
@ -316,7 +315,7 @@ func (b *StatusNode) wakuService(wakuCfg *params.WakuConfig, clusterCfg *params.
} }
func (b *StatusNode) wakuV2Service(nodeConfig *params.NodeConfig) (*wakuv2.Waku, error) { func (b *StatusNode) wakuV2Service(nodeConfig *params.NodeConfig) (*wakuv2.NWaku, error) {
if b.wakuV2Srvc == nil { if b.wakuV2Srvc == nil {
cfg := &wakuv2.Config{ cfg := &wakuv2.Config{
MaxMessageSize: wakucommon.DefaultMaxMessageSize, MaxMessageSize: wakucommon.DefaultMaxMessageSize,
@ -333,7 +332,7 @@ func (b *StatusNode) wakuV2Service(nodeConfig *params.NodeConfig) (*wakuv2.Waku,
Nameserver: nodeConfig.WakuV2Config.Nameserver, Nameserver: nodeConfig.WakuV2Config.Nameserver,
UDPPort: nodeConfig.WakuV2Config.UDPPort, UDPPort: nodeConfig.WakuV2Config.UDPPort,
AutoUpdate: nodeConfig.WakuV2Config.AutoUpdate, AutoUpdate: nodeConfig.WakuV2Config.AutoUpdate,
DefaultShardPubsubTopic: shard.DefaultShardPubsubTopic(), DefaultShardPubsubTopic: wakuv2.DefaultShardPubsubTopic(),
TelemetryServerURL: nodeConfig.WakuV2Config.TelemetryServerURL, TelemetryServerURL: nodeConfig.WakuV2Config.TelemetryServerURL,
ClusterID: nodeConfig.ClusterConfig.ClusterID, ClusterID: nodeConfig.ClusterConfig.ClusterID,
EnableMissingMessageVerification: nodeConfig.WakuV2Config.EnableMissingMessageVerification, EnableMissingMessageVerification: nodeConfig.WakuV2Config.EnableMissingMessageVerification,

59
protocol/common/shard/shard.go
View File

@ -1,59 +0,0 @@
package shard
import (
wakuproto "github.com/waku-org/go-waku/waku/v2/protocol"
"github.com/status-im/status-go/protocol/protobuf"
)
type Shard struct {
Cluster uint16 `json:"cluster"`
Index uint16 `json:"index"`
}
func FromProtobuff(p *protobuf.Shard) *Shard {
if p == nil {
return nil
}
return &Shard{
Cluster: uint16(p.Cluster),
Index: uint16(p.Index),
}
}
func (s *Shard) Protobuffer() *protobuf.Shard {
if s == nil {
return nil
}
return &protobuf.Shard{
Cluster: int32(s.Cluster),
Index: int32(s.Index),
}
}
func (s *Shard) PubsubTopic() string {
if s != nil {
return wakuproto.NewStaticShardingPubsubTopic(s.Cluster, s.Index).String()
}
return ""
}
const MainStatusShardCluster = 16
const DefaultShardIndex = 32
const NonProtectedShardIndex = 64
func DefaultShardPubsubTopic() string {
return wakuproto.NewStaticShardingPubsubTopic(MainStatusShardCluster, DefaultShardIndex).String()
}
func DefaultNonProtectedShard() *Shard {
return &Shard{
Cluster: MainStatusShardCluster,
Index: NonProtectedShardIndex,
}
}
func DefaultNonProtectedPubsubTopic() string {
return DefaultNonProtectedShard().PubsubTopic()
}

10
protocol/communities/community.go
View File

@ -23,12 +23,12 @@ import (
"github.com/status-im/status-go/eth-node/types" "github.com/status-im/status-go/eth-node/types"
"github.com/status-im/status-go/images" "github.com/status-im/status-go/images"
"github.com/status-im/status-go/protocol/common" "github.com/status-im/status-go/protocol/common"
"github.com/status-im/status-go/protocol/common/shard"
community_token "github.com/status-im/status-go/protocol/communities/token" community_token "github.com/status-im/status-go/protocol/communities/token"
"github.com/status-im/status-go/protocol/protobuf" "github.com/status-im/status-go/protocol/protobuf"
"github.com/status-im/status-go/protocol/requests" "github.com/status-im/status-go/protocol/requests"
"github.com/status-im/status-go/protocol/v1" "github.com/status-im/status-go/protocol/v1"
"github.com/status-im/status-go/server" "github.com/status-im/status-go/server"
"github.com/status-im/status-go/wakuv2"
) )
const signatureLength = 65 const signatureLength = 65
@ -55,7 +55,7 @@ type Config struct {
RequestsToJoin []*RequestToJoin RequestsToJoin []*RequestToJoin
MemberIdentity *ecdsa.PrivateKey MemberIdentity *ecdsa.PrivateKey
EventsData *EventsData EventsData *EventsData
Shard *shard.Shard Shard *wakuv2.Shard
PubsubTopicPrivateKey *ecdsa.PrivateKey PubsubTopicPrivateKey *ecdsa.PrivateKey
LastOpenedAt int64 LastOpenedAt int64
} }
@ -172,7 +172,7 @@ func (o *Community) MarshalPublicAPIJSON() ([]byte, error) {
ActiveMembersCount uint64 `json:"activeMembersCount"` ActiveMembersCount uint64 `json:"activeMembersCount"`
PubsubTopic string `json:"pubsubTopic"` PubsubTopic string `json:"pubsubTopic"`
PubsubTopicKey string `json:"pubsubTopicKey"` PubsubTopicKey string `json:"pubsubTopicKey"`
Shard *shard.Shard `json:"shard"` Shard *wakuv2.Shard `json:"shard"`
}{ }{
ID: o.ID(), ID: o.ID(),
Verified: o.config.Verified, Verified: o.config.Verified,
@ -308,7 +308,7 @@ func (o *Community) MarshalJSON() ([]byte, error) {
ActiveMembersCount uint64 `json:"activeMembersCount"` ActiveMembersCount uint64 `json:"activeMembersCount"`
PubsubTopic string `json:"pubsubTopic"` PubsubTopic string `json:"pubsubTopic"`
PubsubTopicKey string `json:"pubsubTopicKey"` PubsubTopicKey string `json:"pubsubTopicKey"`
Shard *shard.Shard `json:"shard"` Shard *wakuv2.Shard `json:"shard"`
LastOpenedAt int64 `json:"lastOpenedAt"` LastOpenedAt int64 `json:"lastOpenedAt"`
Clock uint64 `json:"clock"` Clock uint64 `json:"clock"`
}{ }{
@ -461,7 +461,7 @@ func (o *Community) DescriptionText() string {
return "" return ""
} }
func (o *Community) Shard() *shard.Shard { func (o *Community) Shard() *wakuv2.Shard {
if o != nil && o.config != nil { if o != nil && o.config != nil {
return o.config.Shard return o.config.Shard
} }

22
protocol/communities/manager.go
View File

@ -30,7 +30,6 @@ import (
multiaccountscommon "github.com/status-im/status-go/multiaccounts/common" multiaccountscommon "github.com/status-im/status-go/multiaccounts/common"
"github.com/status-im/status-go/params" "github.com/status-im/status-go/params"
"github.com/status-im/status-go/protocol/common" "github.com/status-im/status-go/protocol/common"
"github.com/status-im/status-go/protocol/common/shard"
community_token "github.com/status-im/status-go/protocol/communities/token" community_token "github.com/status-im/status-go/protocol/communities/token"
"github.com/status-im/status-go/protocol/encryption" "github.com/status-im/status-go/protocol/encryption"
"github.com/status-im/status-go/protocol/ens" "github.com/status-im/status-go/protocol/ens"
@ -45,6 +44,7 @@ import (
"github.com/status-im/status-go/services/wallet/token" "github.com/status-im/status-go/services/wallet/token"
"github.com/status-im/status-go/signal" "github.com/status-im/status-go/signal"
"github.com/status-im/status-go/transactions" "github.com/status-im/status-go/transactions"
"github.com/status-im/status-go/wakuv2"
) )
type Publisher interface { type Publisher interface {
@ -737,8 +737,8 @@ func (m *Manager) All() ([]*Community, error) {
} }
type CommunityShard struct { type CommunityShard struct {
CommunityID string `json:"communityID"` CommunityID string `json:"communityID"`
Shard *shard.Shard `json:"shard"` Shard *wakuv2.Shard `json:"shard"`
} }
type CuratedCommunities struct { type CuratedCommunities struct {
@ -1546,7 +1546,7 @@ func (m *Manager) DeleteCommunity(id types.HexBytes) error {
return m.persistence.DeleteCommunitySettings(id) return m.persistence.DeleteCommunitySettings(id)
} }
func (m *Manager) updateShard(community *Community, shard *shard.Shard, clock uint64) error { func (m *Manager) updateShard(community *Community, shard *wakuv2.Shard, clock uint64) error {
community.config.Shard = shard community.config.Shard = shard
if shard == nil { if shard == nil {
return m.persistence.DeleteCommunityShard(community.ID()) return m.persistence.DeleteCommunityShard(community.ID())
@ -1555,7 +1555,7 @@ func (m *Manager) updateShard(community *Community, shard *shard.Shard, clock ui
return m.persistence.SaveCommunityShard(community.ID(), shard, clock) return m.persistence.SaveCommunityShard(community.ID(), shard, clock)
} }
func (m *Manager) UpdateShard(community *Community, shard *shard.Shard, clock uint64) error { func (m *Manager) UpdateShard(community *Community, shard *wakuv2.Shard, clock uint64) error {
m.communityLock.Lock(community.ID()) m.communityLock.Lock(community.ID())
defer m.communityLock.Unlock(community.ID()) defer m.communityLock.Unlock(community.ID())
@ -1563,7 +1563,7 @@ func (m *Manager) UpdateShard(community *Community, shard *shard.Shard, clock ui
} }
// SetShard assigns a shard to a community // SetShard assigns a shard to a community
func (m *Manager) SetShard(communityID types.HexBytes, shard *shard.Shard) (*Community, error) { func (m *Manager) SetShard(communityID types.HexBytes, shard *wakuv2.Shard) (*Community, error) {
m.communityLock.Lock(communityID) m.communityLock.Lock(communityID)
defer m.communityLock.Unlock(communityID) defer m.communityLock.Unlock(communityID)
@ -2155,11 +2155,11 @@ func (m *Manager) HandleCommunityDescriptionMessage(signer *ecdsa.PublicKey, des
if err != nil { if err != nil {
return nil, err return nil, err
} }
var cShard *shard.Shard var cShard *wakuv2.Shard
if communityShard == nil { if communityShard == nil {
cShard = &shard.Shard{Cluster: shard.MainStatusShardCluster, Index: shard.DefaultShardIndex} cShard = &wakuv2.Shard{Cluster: wakuv2.MainStatusShardCluster, Index: wakuv2.DefaultShardIndex}
} else { } else {
cShard = shard.FromProtobuff(communityShard) cShard = wakuv2.FromProtobuff(communityShard)
} }
config := Config{ config := Config{
CommunityDescription: processedDescription, CommunityDescription: processedDescription,
@ -3972,11 +3972,11 @@ func (m *Manager) GetByIDString(idString string) (*Community, error) {
return m.GetByID(id) return m.GetByID(id)
} }
func (m *Manager) GetCommunityShard(communityID types.HexBytes) (*shard.Shard, error) { func (m *Manager) GetCommunityShard(communityID types.HexBytes) (*wakuv2.Shard, error) {
return m.persistence.GetCommunityShard(communityID) return m.persistence.GetCommunityShard(communityID)
} }
func (m *Manager) SaveCommunityShard(communityID types.HexBytes, shard *shard.Shard, clock uint64) error { func (m *Manager) SaveCommunityShard(communityID types.HexBytes, shard *wakuv2.Shard, clock uint64) error {
m.communityLock.Lock(communityID) m.communityLock.Lock(communityID)
defer m.communityLock.Unlock(communityID) defer m.communityLock.Unlock(communityID)

8
protocol/communities/persistence.go
View File

@ -16,11 +16,11 @@ import (
"github.com/status-im/status-go/eth-node/crypto" "github.com/status-im/status-go/eth-node/crypto"
"github.com/status-im/status-go/eth-node/types" "github.com/status-im/status-go/eth-node/types"
"github.com/status-im/status-go/protocol/common" "github.com/status-im/status-go/protocol/common"
"github.com/status-im/status-go/protocol/common/shard"
"github.com/status-im/status-go/protocol/communities/token" "github.com/status-im/status-go/protocol/communities/token"
"github.com/status-im/status-go/protocol/encryption" "github.com/status-im/status-go/protocol/encryption"
"github.com/status-im/status-go/protocol/protobuf" "github.com/status-im/status-go/protocol/protobuf"
"github.com/status-im/status-go/services/wallet/bigint" "github.com/status-im/status-go/services/wallet/bigint"
"github.com/status-im/status-go/wakuv2"
) )
type Persistence struct { type Persistence struct {
@ -1766,7 +1766,7 @@ func (p *Persistence) AllNonApprovedCommunitiesRequestsToJoin() ([]*RequestToJoi
return nonApprovedRequestsToJoin, nil return nonApprovedRequestsToJoin, nil
} }
func (p *Persistence) SaveCommunityShard(communityID types.HexBytes, shard *shard.Shard, clock uint64) error { func (p *Persistence) SaveCommunityShard(communityID types.HexBytes, shard *wakuv2.Shard, clock uint64) error {
var cluster, index *uint16 var cluster, index *uint16
if shard != nil { if shard != nil {
@ -1801,7 +1801,7 @@ func (p *Persistence) SaveCommunityShard(communityID types.HexBytes, shard *shar
} }
// if data will not be found, will return sql.ErrNoRows. Must be handled on the caller side // if data will not be found, will return sql.ErrNoRows. Must be handled on the caller side
func (p *Persistence) GetCommunityShard(communityID types.HexBytes) (*shard.Shard, error) { func (p *Persistence) GetCommunityShard(communityID types.HexBytes) (*wakuv2.Shard, error) {
var cluster sql.NullInt64 var cluster sql.NullInt64
var index sql.NullInt64 var index sql.NullInt64
err := p.db.QueryRow(`SELECT shard_cluster, shard_index FROM communities_shards WHERE community_id = ?`, err := p.db.QueryRow(`SELECT shard_cluster, shard_index FROM communities_shards WHERE community_id = ?`,
@ -1815,7 +1815,7 @@ func (p *Persistence) GetCommunityShard(communityID types.HexBytes) (*shard.Shar
return nil, nil return nil, nil
} }
return &shard.Shard{ return &wakuv2.Shard{
Cluster: uint16(cluster.Int64), Cluster: uint16(cluster.Int64),
Index: uint16(index.Int64), Index: uint16(index.Int64),
}, nil }, nil

6
protocol/communities/persistence_mapping.go
View File

@ -7,8 +7,8 @@ import (
"github.com/status-im/status-go/eth-node/crypto" "github.com/status-im/status-go/eth-node/crypto"
"github.com/status-im/status-go/protocol/common" "github.com/status-im/status-go/protocol/common"
"github.com/status-im/status-go/protocol/common/shard"
"github.com/status-im/status-go/server" "github.com/status-im/status-go/server"
"github.com/status-im/status-go/wakuv2"
) )
func communityToRecord(community *Community) (*CommunityRecord, error) { func communityToRecord(community *Community) (*CommunityRecord, error) {
@ -118,9 +118,9 @@ func recordBundleToCommunity(
} }
} }
var s *shard.Shard = nil var s *wakuv2.Shard = nil
if r.community.shardCluster != nil && r.community.shardIndex != nil { if r.community.shardCluster != nil && r.community.shardIndex != nil {
s = &shard.Shard{ s = &wakuv2.Shard{
Cluster: uint16(*r.community.shardCluster), Cluster: uint16(*r.community.shardCluster),
Index: uint16(*r.community.shardIndex), Index: uint16(*r.community.shardIndex),
} }

6
protocol/linkpreview_unfurler_status.go
View File

@ -8,8 +8,8 @@ import (
"github.com/status-im/status-go/api/multiformat" "github.com/status-im/status-go/api/multiformat"
"github.com/status-im/status-go/images" "github.com/status-im/status-go/images"
"github.com/status-im/status-go/protocol/common" "github.com/status-im/status-go/protocol/common"
"github.com/status-im/status-go/protocol/common/shard"
"github.com/status-im/status-go/protocol/communities" "github.com/status-im/status-go/protocol/communities"
"github.com/status-im/status-go/wakuv2"
) )
type StatusUnfurler struct { type StatusUnfurler struct {
@ -83,7 +83,7 @@ func (u *StatusUnfurler) buildContactData(publicKey string) (*common.StatusConta
return c, nil return c, nil
} }
func (u *StatusUnfurler) buildCommunityData(communityID string, shard *shard.Shard) (*communities.Community, *common.StatusCommunityLinkPreview, error) { func (u *StatusUnfurler) buildCommunityData(communityID string, shard *wakuv2.Shard) (*communities.Community, *common.StatusCommunityLinkPreview, error) {
// This automatically checks the database // This automatically checks the database
community, err := u.m.FetchCommunity(&FetchCommunityRequest{ community, err := u.m.FetchCommunity(&FetchCommunityRequest{
CommunityKey: communityID, CommunityKey: communityID,
@ -108,7 +108,7 @@ func (u *StatusUnfurler) buildCommunityData(communityID string, shard *shard.Sha
return community, statusCommunityLinkPreviews, nil return community, statusCommunityLinkPreviews, nil
} }
func (u *StatusUnfurler) buildChannelData(channelUUID string, communityID string, communityShard *shard.Shard) (*common.StatusCommunityChannelLinkPreview, error) { func (u *StatusUnfurler) buildChannelData(channelUUID string, communityID string, communityShard *wakuv2.Shard) (*common.StatusCommunityChannelLinkPreview, error) {
community, communityData, err := u.buildCommunityData(communityID, communityShard) community, communityData, err := u.buildCommunityData(communityID, communityShard)
if err != nil { if err != nil {
return nil, fmt.Errorf("failed to build channel community data: %w", err) return nil, fmt.Errorf("failed to build channel community data: %w", err)

4
protocol/messenger.go
View File

@ -38,13 +38,13 @@ import (
"github.com/status-im/status-go/eth-node/types" "github.com/status-im/status-go/eth-node/types"
"github.com/status-im/status-go/images" "github.com/status-im/status-go/images"
multiaccountscommon "github.com/status-im/status-go/multiaccounts/common" multiaccountscommon "github.com/status-im/status-go/multiaccounts/common"
"github.com/status-im/status-go/wakuv2"
"github.com/status-im/status-go/multiaccounts" "github.com/status-im/status-go/multiaccounts"
"github.com/status-im/status-go/multiaccounts/accounts" "github.com/status-im/status-go/multiaccounts/accounts"
"github.com/status-im/status-go/multiaccounts/settings" "github.com/status-im/status-go/multiaccounts/settings"
"github.com/status-im/status-go/protocol/anonmetrics" "github.com/status-im/status-go/protocol/anonmetrics"
"github.com/status-im/status-go/protocol/common" "github.com/status-im/status-go/protocol/common"
"github.com/status-im/status-go/protocol/common/shard"
"github.com/status-im/status-go/protocol/communities" "github.com/status-im/status-go/protocol/communities"
"github.com/status-im/status-go/protocol/encryption" "github.com/status-im/status-go/protocol/encryption"
"github.com/status-im/status-go/protocol/encryption/multidevice" "github.com/status-im/status-go/protocol/encryption/multidevice"
@ -1738,7 +1738,7 @@ func (m *Messenger) InitFilters() error {
logger := m.logger.With(zap.String("site", "Init")) logger := m.logger.With(zap.String("site", "Init"))
// Community requests will arrive in this pubsub topic // Community requests will arrive in this pubsub topic
err := m.SubscribeToPubsubTopic(shard.DefaultNonProtectedPubsubTopic(), nil) err := m.SubscribeToPubsubTopic(wakuv2.DefaultNonProtectedPubsubTopic(), nil)
if err != nil { if err != nil {
return err return err
} }

32
protocol/messenger_communities.go
View File

@ -23,6 +23,7 @@ import (
"go.uber.org/zap" "go.uber.org/zap"
utils "github.com/status-im/status-go/common" utils "github.com/status-im/status-go/common"
"github.com/status-im/status-go/wakuv2"
"github.com/status-im/status-go/account" "github.com/status-im/status-go/account"
multiaccountscommon "github.com/status-im/status-go/multiaccounts/common" multiaccountscommon "github.com/status-im/status-go/multiaccounts/common"
@ -32,7 +33,6 @@ import (
"github.com/status-im/status-go/images" "github.com/status-im/status-go/images"
"github.com/status-im/status-go/multiaccounts/accounts" "github.com/status-im/status-go/multiaccounts/accounts"
"github.com/status-im/status-go/protocol/common" "github.com/status-im/status-go/protocol/common"
"github.com/status-im/status-go/protocol/common/shard"
"github.com/status-im/status-go/protocol/communities" "github.com/status-im/status-go/protocol/communities"
"github.com/status-im/status-go/protocol/communities/token" "github.com/status-im/status-go/protocol/communities/token"
"github.com/status-im/status-go/protocol/discord" "github.com/status-im/status-go/protocol/discord"
@ -86,10 +86,10 @@ const (
type FetchCommunityRequest struct { type FetchCommunityRequest struct {
// CommunityKey should be either a public or a private community key // CommunityKey should be either a public or a private community key
CommunityKey string `json:"communityKey"` CommunityKey string `json:"communityKey"`
Shard *shard.Shard `json:"shard"` Shard *wakuv2.Shard `json:"shard"`
TryDatabase bool `json:"tryDatabase"` TryDatabase bool `json:"tryDatabase"`
WaitForResponse bool `json:"waitForResponse"` WaitForResponse bool `json:"waitForResponse"`
} }
func (r *FetchCommunityRequest) Validate() error { func (r *FetchCommunityRequest) Validate() error {
@ -342,7 +342,7 @@ func (m *Messenger) handleCommunitiesSubscription(c chan *communities.Subscripti
Sender: community.PrivateKey(), Sender: community.PrivateKey(),
SkipEncryptionLayer: true, SkipEncryptionLayer: true,
MessageType: protobuf.ApplicationMetadataMessage_COMMUNITY_USER_KICKED, MessageType: protobuf.ApplicationMetadataMessage_COMMUNITY_USER_KICKED,
PubsubTopic: shard.DefaultNonProtectedPubsubTopic(), PubsubTopic: wakuv2.DefaultNonProtectedPubsubTopic(),
} }
_, err = m.sender.SendPrivate(context.Background(), pk, rawMessage) _, err = m.sender.SendPrivate(context.Background(), pk, rawMessage)
@ -675,7 +675,7 @@ func (m *Messenger) handleCommunitySharedAddressesRequest(state *ReceivedMessage
CommunityID: community.ID(), CommunityID: community.ID(),
SkipEncryptionLayer: true, SkipEncryptionLayer: true,
MessageType: protobuf.ApplicationMetadataMessage_COMMUNITY_SHARED_ADDRESSES_RESPONSE, MessageType: protobuf.ApplicationMetadataMessage_COMMUNITY_SHARED_ADDRESSES_RESPONSE,
PubsubTopic: shard.DefaultNonProtectedPubsubTopic(), PubsubTopic: wakuv2.DefaultNonProtectedPubsubTopic(),
ResendType: common.ResendTypeRawMessage, ResendType: common.ResendTypeRawMessage,
ResendMethod: common.ResendMethodSendPrivate, ResendMethod: common.ResendMethodSendPrivate,
Recipients: []*ecdsa.PublicKey{signer}, Recipients: []*ecdsa.PublicKey{signer},
@ -1041,7 +1041,7 @@ func (m *Messenger) JoinCommunity(ctx context.Context, communityID types.HexByte
return mr, nil return mr, nil
} }
func (m *Messenger) subscribeToCommunityShard(communityID []byte, shard *shard.Shard) error { func (m *Messenger) subscribeToCommunityShard(communityID []byte, shard *wakuv2.Shard) error {
if m.transport.WakuVersion() != 2 { if m.transport.WakuVersion() != 2 {
return nil return nil
} }
@ -1062,7 +1062,7 @@ func (m *Messenger) subscribeToCommunityShard(communityID []byte, shard *shard.S
return m.transport.SubscribeToPubsubTopic(pubsubTopic, pubK) return m.transport.SubscribeToPubsubTopic(pubsubTopic, pubK)
} }
func (m *Messenger) unsubscribeFromShard(shard *shard.Shard) error { func (m *Messenger) unsubscribeFromShard(shard *wakuv2.Shard) error {
if m.transport.WakuVersion() != 2 { if m.transport.WakuVersion() != 2 {
return nil return nil
} }
@ -1489,7 +1489,7 @@ func (m *Messenger) RequestToJoinCommunity(request *requests.RequestToJoinCommun
ResendType: common.ResendTypeRawMessage, ResendType: common.ResendTypeRawMessage,
SkipEncryptionLayer: true, SkipEncryptionLayer: true,
MessageType: protobuf.ApplicationMetadataMessage_COMMUNITY_REQUEST_TO_JOIN, MessageType: protobuf.ApplicationMetadataMessage_COMMUNITY_REQUEST_TO_JOIN,
PubsubTopic: shard.DefaultNonProtectedPubsubTopic(), PubsubTopic: wakuv2.DefaultNonProtectedPubsubTopic(),
Priority: &common.HighPriority, Priority: &common.HighPriority,
} }
@ -1866,7 +1866,7 @@ func (m *Messenger) CancelRequestToJoinCommunity(ctx context.Context, request *r
CommunityID: community.ID(), CommunityID: community.ID(),
SkipEncryptionLayer: true, SkipEncryptionLayer: true,
MessageType: protobuf.ApplicationMetadataMessage_COMMUNITY_CANCEL_REQUEST_TO_JOIN, MessageType: protobuf.ApplicationMetadataMessage_COMMUNITY_CANCEL_REQUEST_TO_JOIN,
PubsubTopic: shard.DefaultNonProtectedPubsubTopic(), PubsubTopic: wakuv2.DefaultNonProtectedPubsubTopic(),
ResendType: common.ResendTypeRawMessage, ResendType: common.ResendTypeRawMessage,
Priority: &common.HighPriority, Priority: &common.HighPriority,
} }
@ -2012,7 +2012,7 @@ func (m *Messenger) acceptRequestToJoinCommunity(requestToJoin *communities.Requ
CommunityID: community.ID(), CommunityID: community.ID(),
SkipEncryptionLayer: true, SkipEncryptionLayer: true,
MessageType: protobuf.ApplicationMetadataMessage_COMMUNITY_REQUEST_TO_JOIN_RESPONSE, MessageType: protobuf.ApplicationMetadataMessage_COMMUNITY_REQUEST_TO_JOIN_RESPONSE,
PubsubTopic: shard.DefaultNonProtectedPubsubTopic(), PubsubTopic: wakuv2.DefaultNonProtectedPubsubTopic(),
ResendType: common.ResendTypeRawMessage, ResendType: common.ResendTypeRawMessage,
ResendMethod: common.ResendMethodSendPrivate, ResendMethod: common.ResendMethodSendPrivate,
Recipients: []*ecdsa.PublicKey{pk}, Recipients: []*ecdsa.PublicKey{pk},
@ -2475,7 +2475,7 @@ func (m *Messenger) DefaultFilters(o *communities.Community) []transport.Filters
{ChatID: updatesChannelID, PubsubTopic: communityPubsubTopic}, {ChatID: updatesChannelID, PubsubTopic: communityPubsubTopic},
{ChatID: mlChannelID, PubsubTopic: communityPubsubTopic}, {ChatID: mlChannelID, PubsubTopic: communityPubsubTopic},
{ChatID: memberUpdateChannelID, PubsubTopic: communityPubsubTopic}, {ChatID: memberUpdateChannelID, PubsubTopic: communityPubsubTopic},
{ChatID: uncompressedPubKey, PubsubTopic: shard.DefaultNonProtectedPubsubTopic()}, {ChatID: uncompressedPubKey, PubsubTopic: wakuv2.DefaultNonProtectedPubsubTopic()},
} }
return filters return filters
@ -3534,7 +3534,7 @@ func (m *Messenger) HandleCommunityShardKey(state *ReceivedMessageState, message
} }
func (m *Messenger) handleCommunityShardAndFiltersFromProto(community *communities.Community, message *protobuf.CommunityShardKey) error { func (m *Messenger) handleCommunityShardAndFiltersFromProto(community *communities.Community, message *protobuf.CommunityShardKey) error {
err := m.communitiesManager.UpdateShard(community, shard.FromProtobuff(message.Shard), message.Clock) err := m.communitiesManager.UpdateShard(community, wakuv2.FromProtobuff(message.Shard), message.Clock)
if err != nil { if err != nil {
return err return err
} }
@ -3556,7 +3556,7 @@ func (m *Messenger) handleCommunityShardAndFiltersFromProto(community *communiti
} }
// Unsubscribing from existing shard // Unsubscribing from existing shard
if community.Shard() != nil && community.Shard() != shard.FromProtobuff(message.GetShard()) { if community.Shard() != nil && community.Shard() != wakuv2.FromProtobuff(message.GetShard()) {
err := m.unsubscribeFromShard(community.Shard()) err := m.unsubscribeFromShard(community.Shard())
if err != nil { if err != nil {
return err return err
@ -3570,7 +3570,7 @@ func (m *Messenger) handleCommunityShardAndFiltersFromProto(community *communiti
return err return err
} }
// Update community filters in case of change of shard // Update community filters in case of change of shard
if community.Shard() != shard.FromProtobuff(message.GetShard()) { if community.Shard() != wakuv2.FromProtobuff(message.GetShard()) {
err = m.UpdateCommunityFilters(community) err = m.UpdateCommunityFilters(community)
if err != nil { if err != nil {
return err return err

6
protocol/messenger_community_shard.go
View File

@ -12,11 +12,11 @@ import (
"github.com/status-im/status-go/eth-node/crypto" "github.com/status-im/status-go/eth-node/crypto"
"github.com/status-im/status-go/eth-node/types" "github.com/status-im/status-go/eth-node/types"
"github.com/status-im/status-go/protocol/common" "github.com/status-im/status-go/protocol/common"
"github.com/status-im/status-go/protocol/common/shard"
"github.com/status-im/status-go/protocol/communities" "github.com/status-im/status-go/protocol/communities"
"github.com/status-im/status-go/protocol/protobuf" "github.com/status-im/status-go/protocol/protobuf"
"github.com/status-im/status-go/protocol/transport" "github.com/status-im/status-go/protocol/transport"
v1protocol "github.com/status-im/status-go/protocol/v1" v1protocol "github.com/status-im/status-go/protocol/v1"
"github.com/status-im/status-go/wakuv2"
) )
func (m *Messenger) sendPublicCommunityShardInfo(community *communities.Community) error { func (m *Messenger) sendPublicCommunityShardInfo(community *communities.Community) error {
@ -57,7 +57,7 @@ func (m *Messenger) sendPublicCommunityShardInfo(community *communities.Communit
// we don't want to wrap in an encryption layer message // we don't want to wrap in an encryption layer message
SkipEncryptionLayer: true, SkipEncryptionLayer: true,
MessageType: protobuf.ApplicationMetadataMessage_COMMUNITY_PUBLIC_SHARD_INFO, MessageType: protobuf.ApplicationMetadataMessage_COMMUNITY_PUBLIC_SHARD_INFO,
PubsubTopic: shard.DefaultNonProtectedPubsubTopic(), // it must be sent always to default shard pubsub topic PubsubTopic: wakuv2.DefaultNonProtectedPubsubTopic(), // it must be sent always to default shard pubsub topic
Priority: &common.HighPriority, Priority: &common.HighPriority,
} }
@ -89,7 +89,7 @@ func (m *Messenger) HandleCommunityPublicShardInfo(state *ReceivedMessageState,
return err return err
} }
err = m.communitiesManager.SaveCommunityShard(publicShardInfo.CommunityId, shard.FromProtobuff(publicShardInfo.Shard), publicShardInfo.Clock) err = m.communitiesManager.SaveCommunityShard(publicShardInfo.CommunityId, wakuv2.FromProtobuff(publicShardInfo.Shard), publicShardInfo.Clock)
if err != nil && err != communities.ErrOldShardInfo { if err != nil && err != communities.ErrOldShardInfo {
logError(err) logError(err)
return err return err

4
protocol/messenger_config.go
View File

@ -114,7 +114,7 @@ type config struct {
telemetryServerURL string telemetryServerURL string
telemetrySendPeriod time.Duration telemetrySendPeriod time.Duration
wakuService *wakuv2.Waku wakuService *wakuv2.NWaku
messageResendMinDelay time.Duration messageResendMinDelay time.Duration
messageResendMaxCount int messageResendMaxCount int
@ -387,7 +387,7 @@ func WithCommunityTokensService(s communities.CommunityTokensServiceInterface) O
} }
} }
func WithWakuService(s *wakuv2.Waku) Option { func WithWakuService(s *wakuv2.NWaku) Option {
return func(c *config) error { return func(c *config) error {
c.wakuService = s c.wakuService = s
return nil return nil

8
protocol/messenger_share_urls.go
View File

@ -11,12 +11,12 @@ import (
"github.com/status-im/status-go/eth-node/crypto" "github.com/status-im/status-go/eth-node/crypto"
"github.com/status-im/status-go/eth-node/types" "github.com/status-im/status-go/eth-node/types"
"github.com/status-im/status-go/protocol/common" "github.com/status-im/status-go/protocol/common"
"github.com/status-im/status-go/protocol/common/shard"
"github.com/status-im/status-go/protocol/communities" "github.com/status-im/status-go/protocol/communities"
"github.com/status-im/status-go/protocol/protobuf" "github.com/status-im/status-go/protocol/protobuf"
"github.com/status-im/status-go/protocol/requests" "github.com/status-im/status-go/protocol/requests"
"github.com/status-im/status-go/protocol/urls" "github.com/status-im/status-go/protocol/urls"
"github.com/status-im/status-go/services/utils" "github.com/status-im/status-go/services/utils"
"github.com/status-im/status-go/wakuv2"
) )
type CommunityURLData struct { type CommunityURLData struct {
@ -46,7 +46,7 @@ type URLDataResponse struct {
Community *CommunityURLData `json:"community"` Community *CommunityURLData `json:"community"`
Channel *CommunityChannelURLData `json:"channel"` Channel *CommunityChannelURLData `json:"channel"`
Contact *ContactURLData `json:"contact"` Contact *ContactURLData `json:"contact"`
Shard *shard.Shard `json:"shard,omitempty"` Shard *wakuv2.Shard `json:"shard,omitempty"`
} }
const baseShareURL = "https://status.app" const baseShareURL = "https://status.app"
@ -201,7 +201,7 @@ func parseCommunityURLWithData(data string, chatKey string) (*URLDataResponse, e
TagIndices: tagIndices, TagIndices: tagIndices,
CommunityID: types.EncodeHex(communityID), CommunityID: types.EncodeHex(communityID),
}, },
Shard: shard.FromProtobuff(urlDataProto.Shard), Shard: wakuv2.FromProtobuff(urlDataProto.Shard),
}, nil }, nil
} }
@ -377,7 +377,7 @@ func parseCommunityChannelURLWithData(data string, chatKey string) (*URLDataResp
Color: channelProto.Color, Color: channelProto.Color,
ChannelUUID: channelProto.Uuid, ChannelUUID: channelProto.Uuid,
}, },
Shard: shard.FromProtobuff(urlDataProto.Shard), Shard: wakuv2.FromProtobuff(urlDataProto.Shard),
}, nil }, nil
} }

12
protocol/messenger_store_node_request_manager.go
View File

@ -8,7 +8,6 @@ import (
"time" "time"
"github.com/status-im/status-go/eth-node/crypto" "github.com/status-im/status-go/eth-node/crypto"
"github.com/status-im/status-go/protocol/common/shard"
"go.uber.org/zap" "go.uber.org/zap"
@ -16,6 +15,7 @@ import (
"github.com/status-im/status-go/protocol/communities" "github.com/status-im/status-go/protocol/communities"
"github.com/status-im/status-go/protocol/transport" "github.com/status-im/status-go/protocol/transport"
"github.com/status-im/status-go/services/mailservers" "github.com/status-im/status-go/services/mailservers"
"github.com/status-im/status-go/wakuv2"
) )
const ( const (
@ -81,7 +81,7 @@ func (m *StoreNodeRequestManager) FetchCommunity(community communities.Community
zap.Any("community", community), zap.Any("community", community),
zap.Any("config", cfg)) zap.Any("config", cfg))
requestCommunity := func(communityID string, shard *shard.Shard) (*communities.Community, StoreNodeRequestStats, error) { requestCommunity := func(communityID string, shard *wakuv2.Shard) (*communities.Community, StoreNodeRequestStats, error) {
channel, err := m.subscribeToRequest(storeNodeCommunityRequest, communityID, shard, cfg) channel, err := m.subscribeToRequest(storeNodeCommunityRequest, communityID, shard, cfg)
if err != nil { if err != nil {
return nil, StoreNodeRequestStats{}, fmt.Errorf("failed to create a request for community: %w", err) return nil, StoreNodeRequestStats{}, fmt.Errorf("failed to create a request for community: %w", err)
@ -99,7 +99,7 @@ func (m *StoreNodeRequestManager) FetchCommunity(community communities.Community
communityShard := community.Shard communityShard := community.Shard
if communityShard == nil { if communityShard == nil {
id := transport.CommunityShardInfoTopic(community.CommunityID) id := transport.CommunityShardInfoTopic(community.CommunityID)
fetchedShard, err := m.subscribeToRequest(storeNodeShardRequest, id, shard.DefaultNonProtectedShard(), cfg) fetchedShard, err := m.subscribeToRequest(storeNodeShardRequest, id, wakuv2.DefaultNonProtectedShard(), cfg)
if err != nil { if err != nil {
return nil, StoreNodeRequestStats{}, fmt.Errorf("failed to create a shard info request: %w", err) return nil, StoreNodeRequestStats{}, fmt.Errorf("failed to create a shard info request: %w", err)
} }
@ -176,7 +176,7 @@ func (m *StoreNodeRequestManager) FetchContact(contactID string, opts []StoreNod
// subscribeToRequest checks if a request for given community/contact is already in progress, creates and installs // subscribeToRequest checks if a request for given community/contact is already in progress, creates and installs
// a new one if not found, and returns a subscription to the result of the found/started request. // a new one if not found, and returns a subscription to the result of the found/started request.
// The subscription can then be used to get the result of the request, this could be either a community/contact or an error. // The subscription can then be used to get the result of the request, this could be either a community/contact or an error.
func (m *StoreNodeRequestManager) subscribeToRequest(requestType storeNodeRequestType, dataID string, shard *shard.Shard, cfg StoreNodeRequestConfig) (storeNodeResponseSubscription, error) { func (m *StoreNodeRequestManager) subscribeToRequest(requestType storeNodeRequestType, dataID string, shard *wakuv2.Shard, cfg StoreNodeRequestConfig) (storeNodeResponseSubscription, error) {
// It's important to unlock only after getting the subscription channel. // It's important to unlock only after getting the subscription channel.
// We also lock `activeRequestsLock` during finalizing the requests. This ensures that the subscription // We also lock `activeRequestsLock` during finalizing the requests. This ensures that the subscription
// created in this function will get the result even if the requests proceeds faster than this function ends. // created in this function will get the result even if the requests proceeds faster than this function ends.
@ -230,7 +230,7 @@ func (m *StoreNodeRequestManager) newStoreNodeRequest() *storeNodeRequest {
// getFilter checks if a filter for a given community is already created and creates one of not found. // getFilter checks if a filter for a given community is already created and creates one of not found.
// Returns the found/created filter, a flag if the filter was created by the function and an error. // Returns the found/created filter, a flag if the filter was created by the function and an error.
func (m *StoreNodeRequestManager) getFilter(requestType storeNodeRequestType, dataID string, shard *shard.Shard) (*transport.Filter, bool, error) { func (m *StoreNodeRequestManager) getFilter(requestType storeNodeRequestType, dataID string, shard *wakuv2.Shard) (*transport.Filter, bool, error) {
// First check if such filter already exists. // First check if such filter already exists.
filter := m.messenger.transport.FilterByChatID(dataID) filter := m.messenger.transport.FilterByChatID(dataID)
if filter != nil { if filter != nil {
@ -332,7 +332,7 @@ type storeNodeRequestResult struct {
// One of data fields (community or contact) will be present depending on request type // One of data fields (community or contact) will be present depending on request type
community *communities.Community community *communities.Community
contact *Contact contact *Contact
shard *shard.Shard shard *wakuv2.Shard
} }
type storeNodeResponseSubscription = chan storeNodeRequestResult type storeNodeResponseSubscription = chan storeNodeRequestResult

6
protocol/messenger_testing_utils.go
View File

@ -13,11 +13,11 @@ import (
"github.com/libp2p/go-libp2p/core/peer" "github.com/libp2p/go-libp2p/core/peer"
"github.com/status-im/status-go/protocol/wakusync" "github.com/status-im/status-go/protocol/wakusync"
"github.com/status-im/status-go/wakuv2"
"github.com/status-im/status-go/protocol/identity" "github.com/status-im/status-go/protocol/identity"
"github.com/status-im/status-go/eth-node/types" "github.com/status-im/status-go/eth-node/types"
waku2 "github.com/status-im/status-go/wakuv2"
"github.com/stretchr/testify/suite" "github.com/stretchr/testify/suite"
@ -205,7 +205,7 @@ func WaitOnSignaledCommunityFound(m *Messenger, action func(), condition func(co
} }
} }
func WaitForConnectionStatus(s *suite.Suite, waku *waku2.Waku, action func() bool) { func WaitForConnectionStatus(s *suite.Suite, waku *wakuv2.NWaku, action func() bool) {
subscription := waku.SubscribeToConnStatusChanges() subscription := waku.SubscribeToConnStatusChanges()
defer subscription.Unsubscribe() defer subscription.Unsubscribe()
@ -237,7 +237,7 @@ func hasAllPeers(m map[peer.ID]types.WakuV2Peer, checkSlice peer.IDSlice) bool {
return true return true
} }
func WaitForPeersConnected(s *suite.Suite, waku *waku2.Waku, action func() peer.IDSlice) { func WaitForPeersConnected(s *suite.Suite, waku *wakuv2.NWaku, action func() peer.IDSlice) {
subscription := waku.SubscribeToConnStatusChanges() subscription := waku.SubscribeToConnStatusChanges()
defer subscription.Unsubscribe() defer subscription.Unsubscribe()

6
protocol/requests/set_community_shard.go
View File

@ -4,12 +4,12 @@ import (
"errors" "errors"
"github.com/status-im/status-go/eth-node/types" "github.com/status-im/status-go/eth-node/types"
"github.com/status-im/status-go/protocol/common/shard" "github.com/status-im/status-go/wakuv2"
) )
type SetCommunityShard struct { type SetCommunityShard struct {
CommunityID types.HexBytes `json:"communityId"` CommunityID types.HexBytes `json:"communityId"`
Shard *shard.Shard `json:"shard,omitempty"` Shard *wakuv2.Shard `json:"shard,omitempty"`
PrivateKey *types.HexBytes `json:"privateKey,omitempty"` PrivateKey *types.HexBytes `json:"privateKey,omitempty"`
} }
@ -19,7 +19,7 @@ func (s *SetCommunityShard) Validate() error {
} }
if s.Shard != nil { if s.Shard != nil {
// TODO: for now only MainStatusShard(16) is accepted // TODO: for now only MainStatusShard(16) is accepted
if s.Shard.Cluster != shard.MainStatusShardCluster { if s.Shard.Cluster != wakuv2.MainStatusShardCluster {
return errors.New("invalid shard cluster") return errors.New("invalid shard cluster")
} }
if s.Shard.Index > 1023 { if s.Shard.Index > 1023 {

6
protocol/transport/filters_manager.go
View File

@ -11,7 +11,7 @@ import (
"go.uber.org/zap" "go.uber.org/zap"
"github.com/status-im/status-go/eth-node/types" "github.com/status-im/status-go/eth-node/types"
"github.com/status-im/status-go/protocol/common/shard" "github.com/status-im/status-go/wakuv2"
) )
const ( const (
@ -141,7 +141,7 @@ func (f *FiltersManager) InitPublicFilters(publicFiltersToInit []FiltersToInitia
} }
type CommunityFilterToInitialize struct { type CommunityFilterToInitialize struct {
Shard *shard.Shard Shard *wakuv2.Shard
PrivKey *ecdsa.PrivateKey PrivKey *ecdsa.PrivateKey
} }
@ -158,7 +158,7 @@ func (f *FiltersManager) InitCommunityFilters(communityFiltersToInitialize []Com
} }
topics := make([]string, 0) topics := make([]string, 0)
topics = append(topics, shard.DefaultNonProtectedPubsubTopic()) topics = append(topics, wakuv2.DefaultNonProtectedPubsubTopic())
topics = append(topics, communityFilter.Shard.PubsubTopic()) topics = append(topics, communityFilter.Shard.PubsubTopic())
for _, pubsubTopic := range topics { for _, pubsubTopic := range topics {

6
services/ext/api.go
View File

@ -16,6 +16,7 @@ import (
"github.com/status-im/status-go/services/browsers" "github.com/status-im/status-go/services/browsers"
"github.com/status-im/status-go/services/wallet" "github.com/status-im/status-go/services/wallet"
"github.com/status-im/status-go/services/wallet/bigint" "github.com/status-im/status-go/services/wallet/bigint"
"github.com/status-im/status-go/wakuv2"
"github.com/ethereum/go-ethereum/common/hexutil" "github.com/ethereum/go-ethereum/common/hexutil"
"github.com/ethereum/go-ethereum/log" "github.com/ethereum/go-ethereum/log"
@ -32,7 +33,6 @@ import (
"github.com/status-im/status-go/multiaccounts/settings" "github.com/status-im/status-go/multiaccounts/settings"
"github.com/status-im/status-go/protocol" "github.com/status-im/status-go/protocol"
"github.com/status-im/status-go/protocol/common" "github.com/status-im/status-go/protocol/common"
"github.com/status-im/status-go/protocol/common/shard"
"github.com/status-im/status-go/protocol/communities" "github.com/status-im/status-go/protocol/communities"
"github.com/status-im/status-go/protocol/communities/token" "github.com/status-im/status-go/protocol/communities/token"
"github.com/status-im/status-go/protocol/discord" "github.com/status-im/status-go/protocol/discord"
@ -1311,7 +1311,7 @@ func (api *PublicAPI) RequestCommunityInfoFromMailserver(communityID string) (*c
// Deprecated: RequestCommunityInfoFromMailserverWithShard is deprecated in favor of // Deprecated: RequestCommunityInfoFromMailserverWithShard is deprecated in favor of
// configurable FetchCommunity. // configurable FetchCommunity.
func (api *PublicAPI) RequestCommunityInfoFromMailserverWithShard(communityID string, shard *shard.Shard) (*communities.Community, error) { func (api *PublicAPI) RequestCommunityInfoFromMailserverWithShard(communityID string, shard *wakuv2.Shard) (*communities.Community, error) {
request := &protocol.FetchCommunityRequest{ request := &protocol.FetchCommunityRequest{
CommunityKey: communityID, CommunityKey: communityID,
Shard: shard, Shard: shard,
@ -1336,7 +1336,7 @@ func (api *PublicAPI) RequestCommunityInfoFromMailserverAsync(communityID string
// Deprecated: RequestCommunityInfoFromMailserverAsyncWithShard is deprecated in favor of // Deprecated: RequestCommunityInfoFromMailserverAsyncWithShard is deprecated in favor of
// configurable FetchCommunity. // configurable FetchCommunity.
func (api *PublicAPI) RequestCommunityInfoFromMailserverAsyncWithShard(communityID string, shard *shard.Shard) error { func (api *PublicAPI) RequestCommunityInfoFromMailserverAsyncWithShard(communityID string, shard *wakuv2.Shard) error {
request := &protocol.FetchCommunityRequest{ request := &protocol.FetchCommunityRequest{
CommunityKey: communityID, CommunityKey: communityID,
Shard: shard, Shard: shard,

4
services/ext/service.go
View File

@ -123,7 +123,7 @@ func (s *Service) GetPeer(rawURL string) (*enode.Node, error) {
return enode.ParseV4(rawURL) return enode.ParseV4(rawURL)
} }
func (s *Service) InitProtocol(nodeName string, identity *ecdsa.PrivateKey, appDb, walletDb *sql.DB, httpServer *server.MediaServer, multiAccountDb *multiaccounts.Database, acc *multiaccounts.Account, accountManager *account.GethManager, rpcClient *rpc.Client, walletService *wallet.Service, communityTokensService *communitytokens.Service, wakuService *wakuv2.Waku, logger *zap.Logger) error { func (s *Service) InitProtocol(nodeName string, identity *ecdsa.PrivateKey, appDb, walletDb *sql.DB, httpServer *server.MediaServer, multiAccountDb *multiaccounts.Database, acc *multiaccounts.Account, accountManager *account.GethManager, rpcClient *rpc.Client, walletService *wallet.Service, communityTokensService *communitytokens.Service, wakuService *wakuv2.NWaku, logger *zap.Logger) error {
var err error var err error
if !s.config.ShhextConfig.PFSEnabled { if !s.config.ShhextConfig.PFSEnabled {
return nil return nil
@ -393,7 +393,7 @@ func buildMessengerOptions(
accountsDB *accounts.Database, accountsDB *accounts.Database,
walletService *wallet.Service, walletService *wallet.Service,
communityTokensService *communitytokens.Service, communityTokensService *communitytokens.Service,
wakuService *wakuv2.Waku, wakuService *wakuv2.NWaku,
logger *zap.Logger, logger *zap.Logger,
messengerSignalsHandler protocol.MessengerSignalsHandler, messengerSignalsHandler protocol.MessengerSignalsHandler,
accountManager account.Manager, accountManager account.Manager,

4
services/status/service.go
View File

@ -10,7 +10,7 @@ import (
"github.com/status-im/status-go/eth-node/types" "github.com/status-im/status-go/eth-node/types"
"github.com/status-im/status-go/protocol" "github.com/status-im/status-go/protocol"
"github.com/status-im/status-go/protocol/common/shard" "github.com/status-im/status-go/wakuv2"
) )
// Make sure that Service implements node.Lifecycle interface. // Make sure that Service implements node.Lifecycle interface.
@ -70,7 +70,7 @@ type PublicAPI struct {
service *Service service *Service
} }
func (p *PublicAPI) CommunityInfo(communityID types.HexBytes, shard *shard.Shard) (json.RawMessage, error) { func (p *PublicAPI) CommunityInfo(communityID types.HexBytes, shard *wakuv2.Shard) (json.RawMessage, error) {
if p.service.messenger == nil { if p.service.messenger == nil {
return nil, ErrNotInitialized return nil, ErrNotInitialized
} }

32
vendor/github.com/waku-org/go-waku/waku/v2/protocol/store/client.go generated vendored
View File

@ -113,40 +113,40 @@ func (s *WakuStore) Request(ctx context.Context, criteria Criteria, opts ...Requ
} }
//Add Peer to peerstore. //Add Peer to peerstore.
if s.pm != nil && params.peerAddr != nil { if s.pm != nil && params.PeerAddr != nil {
pData, err := s.pm.AddPeer(params.peerAddr, peerstore.Static, pubsubTopics, StoreQueryID_v300) pData, err := s.pm.AddPeer(params.PeerAddr, peerstore.Static, pubsubTopics, StoreQueryID_v300)
if err != nil { if err != nil {
return nil, err return nil, err
} }
s.pm.Connect(pData) s.pm.Connect(pData)
params.selectedPeer = pData.AddrInfo.ID params.SelectedPeer = pData.AddrInfo.ID
} }
if s.pm != nil && params.selectedPeer == "" { if s.pm != nil && params.SelectedPeer == "" {
if isFilterCriteria { if isFilterCriteria {
selectedPeers, err := s.pm.SelectPeers( selectedPeers, err := s.pm.SelectPeers(
peermanager.PeerSelectionCriteria{ peermanager.PeerSelectionCriteria{
SelectionType: params.peerSelectionType, SelectionType: params.PeerSelectionType,
Proto: StoreQueryID_v300, Proto: StoreQueryID_v300,
PubsubTopics: []string{filterCriteria.PubsubTopic}, PubsubTopics: []string{filterCriteria.PubsubTopic},
SpecificPeers: params.preferredPeers, SpecificPeers: params.PreferredPeers,
Ctx: ctx, Ctx: ctx,
}, },
) )
if err != nil { if err != nil {
return nil, err return nil, err
} }
params.selectedPeer = selectedPeers[0] params.SelectedPeer = selectedPeers[0]
} else { } else {
return nil, ErrMustSelectPeer return nil, ErrMustSelectPeer
} }
} }
if params.selectedPeer == "" { if params.SelectedPeer == "" {
return nil, ErrNoPeersAvailable return nil, ErrNoPeersAvailable
} }
pageLimit := params.pageLimit pageLimit := params.PageLimit
if pageLimit == 0 { if pageLimit == 0 {
pageLimit = DefaultPageSize pageLimit = DefaultPageSize
} else if pageLimit > uint64(MaxPageSize) { } else if pageLimit > uint64(MaxPageSize) {
@ -154,16 +154,16 @@ func (s *WakuStore) Request(ctx context.Context, criteria Criteria, opts ...Requ
} }
storeRequest := &pb.StoreQueryRequest{ storeRequest := &pb.StoreQueryRequest{
RequestId: hex.EncodeToString(params.requestID), RequestId: hex.EncodeToString(params.RequestID),
IncludeData: params.includeData, IncludeData: params.IncludeData,
PaginationForward: params.forward, PaginationForward: params.Forward,
PaginationLimit: proto.Uint64(pageLimit), PaginationLimit: proto.Uint64(pageLimit),
} }
criteria.PopulateStoreRequest(storeRequest) criteria.PopulateStoreRequest(storeRequest)
if params.cursor != nil { if params.Cursor != nil {
storeRequest.PaginationCursor = params.cursor storeRequest.PaginationCursor = params.Cursor
} }
err := storeRequest.Validate() err := storeRequest.Validate()
@ -171,7 +171,7 @@ func (s *WakuStore) Request(ctx context.Context, criteria Criteria, opts ...Requ
return nil, err return nil, err
} }
response, err := s.queryFrom(ctx, storeRequest, params.selectedPeer) response, err := s.queryFrom(ctx, storeRequest, params.SelectedPeer)
if err != nil { if err != nil {
return nil, err return nil, err
} }
@ -181,7 +181,7 @@ func (s *WakuStore) Request(ctx context.Context, criteria Criteria, opts ...Requ
messages: response.Messages, messages: response.Messages,
storeRequest: storeRequest, storeRequest: storeRequest,
storeResponse: response, storeResponse: response,
peerID: params.selectedPeer, peerID: params.SelectedPeer,
cursor: response.PaginationCursor, cursor: response.PaginationCursor,
} }

44
vendor/github.com/waku-org/go-waku/waku/v2/protocol/store/options.go generated vendored
View File

@ -10,15 +10,15 @@ import (
) )
type Parameters struct { type Parameters struct {
selectedPeer peer.ID SelectedPeer peer.ID
peerAddr multiaddr.Multiaddr PeerAddr multiaddr.Multiaddr
peerSelectionType peermanager.PeerSelection PeerSelectionType peermanager.PeerSelection
preferredPeers peer.IDSlice PreferredPeers peer.IDSlice
requestID []byte RequestID []byte
cursor []byte Cursor []byte
pageLimit uint64 PageLimit uint64
forward bool Forward bool
includeData bool IncludeData bool
} }
type RequestOption func(*Parameters) error type RequestOption func(*Parameters) error
@ -27,8 +27,8 @@ type RequestOption func(*Parameters) error
// Note that this option is mutually exclusive to WithPeerAddr, only one of them can be used. // Note that this option is mutually exclusive to WithPeerAddr, only one of them can be used.
func WithPeer(p peer.ID) RequestOption { func WithPeer(p peer.ID) RequestOption {
return func(params *Parameters) error { return func(params *Parameters) error {
params.selectedPeer = p params.SelectedPeer = p
if params.peerAddr != nil { if params.PeerAddr != nil {
return errors.New("WithPeer and WithPeerAddr options are mutually exclusive") return errors.New("WithPeer and WithPeerAddr options are mutually exclusive")
} }
return nil return nil
@ -40,8 +40,8 @@ func WithPeer(p peer.ID) RequestOption {
// Note that this option is mutually exclusive to WithPeerAddr, only one of them can be used. // Note that this option is mutually exclusive to WithPeerAddr, only one of them can be used.
func WithPeerAddr(pAddr multiaddr.Multiaddr) RequestOption { func WithPeerAddr(pAddr multiaddr.Multiaddr) RequestOption {
return func(params *Parameters) error { return func(params *Parameters) error {
params.peerAddr = pAddr params.PeerAddr = pAddr
if params.selectedPeer != "" { if params.SelectedPeer != "" {
return errors.New("WithPeerAddr and WithPeer options are mutually exclusive") return errors.New("WithPeerAddr and WithPeer options are mutually exclusive")
} }
return nil return nil
@ -55,8 +55,8 @@ func WithPeerAddr(pAddr multiaddr.Multiaddr) RequestOption {
// Note: This option is avaiable only with peerManager // Note: This option is avaiable only with peerManager
func WithAutomaticPeerSelection(fromThesePeers ...peer.ID) RequestOption { func WithAutomaticPeerSelection(fromThesePeers ...peer.ID) RequestOption {
return func(params *Parameters) error { return func(params *Parameters) error {
params.peerSelectionType = peermanager.Automatic params.PeerSelectionType = peermanager.Automatic
params.preferredPeers = fromThesePeers params.PreferredPeers = fromThesePeers
return nil return nil
} }
} }
@ -68,7 +68,7 @@ func WithAutomaticPeerSelection(fromThesePeers ...peer.ID) RequestOption {
// Note: This option is avaiable only with peerManager // Note: This option is avaiable only with peerManager
func WithFastestPeerSelection(fromThesePeers ...peer.ID) RequestOption { func WithFastestPeerSelection(fromThesePeers ...peer.ID) RequestOption {
return func(params *Parameters) error { return func(params *Parameters) error {
params.peerSelectionType = peermanager.LowestRTT params.PeerSelectionType = peermanager.LowestRTT
return nil return nil
} }
} }
@ -77,7 +77,7 @@ func WithFastestPeerSelection(fromThesePeers ...peer.ID) RequestOption {
// creating a store request // creating a store request
func WithRequestID(requestID []byte) RequestOption { func WithRequestID(requestID []byte) RequestOption {
return func(params *Parameters) error { return func(params *Parameters) error {
params.requestID = requestID params.RequestID = requestID
return nil return nil
} }
} }
@ -86,14 +86,14 @@ func WithRequestID(requestID []byte) RequestOption {
// when creating a store request // when creating a store request
func WithAutomaticRequestID() RequestOption { func WithAutomaticRequestID() RequestOption {
return func(params *Parameters) error { return func(params *Parameters) error {
params.requestID = protocol.GenerateRequestID() params.RequestID = protocol.GenerateRequestID()
return nil return nil
} }
} }
func WithCursor(cursor []byte) RequestOption { func WithCursor(cursor []byte) RequestOption {
return func(params *Parameters) error { return func(params *Parameters) error {
params.cursor = cursor params.Cursor = cursor
return nil return nil
} }
} }
@ -101,8 +101,8 @@ func WithCursor(cursor []byte) RequestOption {
// WithPaging is an option used to specify the order and maximum number of records to return // WithPaging is an option used to specify the order and maximum number of records to return
func WithPaging(forward bool, limit uint64) RequestOption { func WithPaging(forward bool, limit uint64) RequestOption {
return func(params *Parameters) error { return func(params *Parameters) error {
params.forward = forward params.Forward = forward
params.pageLimit = limit params.PageLimit = limit
return nil return nil
} }
} }
@ -110,7 +110,7 @@ func WithPaging(forward bool, limit uint64) RequestOption {
// IncludeData is an option used to indicate whether you want to return the message content or not // IncludeData is an option used to indicate whether you want to return the message content or not
func IncludeData(v bool) RequestOption { func IncludeData(v bool) RequestOption {
return func(params *Parameters) error { return func(params *Parameters) error {
params.includeData = v params.IncludeData = v
return nil return nil
} }
} }

16
wakuv2/api.go
View File

@ -1,17 +1,17 @@
// Copyright 2019 The Waku Library Authors. // Copyright 2019 The NWaku Library Authors.
// //
// The Waku library is free software: you can redistribute it and/or modify // The NWaku library is free software: you can redistribute it and/or modify
// it under the terms of the GNU Lesser General Public License as published by // it under the terms of the GNU Lesser General Public License as published by
// the Free Software Foundation, either version 3 of the License, or // the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version. // (at your option) any later version.
// //
// The Waku library is distributed in the hope that it will be useful, // The NWaku library is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty off // but WITHOUT ANY WARRANTY; without even the implied warranty off
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the // MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU Lesser General Public License for more details. // GNU Lesser General Public License for more details.
// //
// You should have received a copy of the GNU Lesser General Public License // You should have received a copy of the GNU Lesser General Public License
// along with the Waku library. If not, see <http://www.gnu.org/licenses/>. // along with the NWaku library. If not, see <http://www.gnu.org/licenses/>.
// //
// This software uses the go-ethereum library, which is licensed // This software uses the go-ethereum library, which is licensed
// under the GNU Lesser General Public Library, version 3 or any later. // under the GNU Lesser General Public Library, version 3 or any later.
@ -52,14 +52,14 @@ var (
// PublicWakuAPI provides the waku RPC service that can be // PublicWakuAPI provides the waku RPC service that can be
// use publicly without security implications. // use publicly without security implications.
type PublicWakuAPI struct { type PublicWakuAPI struct {
w *Waku w *NWaku
mu sync.Mutex mu sync.Mutex
lastUsed map[string]time.Time // keeps track when a filter was polled for the last time. lastUsed map[string]time.Time // keeps track when a filter was polled for the last time.
} }
// NewPublicWakuAPI create a new RPC waku service. // NewPublicWakuAPI create a new RPC waku service.
func NewPublicWakuAPI(w *Waku) *PublicWakuAPI { func NewPublicWakuAPI(w *NWaku) *PublicWakuAPI {
api := &PublicWakuAPI{ api := &PublicWakuAPI{
w: w, w: w,
lastUsed: make(map[string]time.Time), lastUsed: make(map[string]time.Time),
@ -185,7 +185,7 @@ type NewMessage struct {
Priority *int `json:"priority"` Priority *int `json:"priority"`
} }
// Post posts a message on the Waku network. // Post posts a message on the NWaku network.
// returns the hash of the message in case of success. // returns the hash of the message in case of success.
func (api *PublicWakuAPI) Post(ctx context.Context, req NewMessage) (hexutil.Bytes, error) { func (api *PublicWakuAPI) Post(ctx context.Context, req NewMessage) (hexutil.Bytes, error) {
var ( var (
@ -252,7 +252,7 @@ func (api *PublicWakuAPI) Post(ctx context.Context, req NewMessage) (hexutil.Byt
Version: &version, Version: &version,
ContentTopic: req.ContentTopic.ContentTopic(), ContentTopic: req.ContentTopic.ContentTopic(),
Timestamp: proto.Int64(api.w.timestamp()), Timestamp: proto.Int64(api.w.timestamp()),
Meta: []byte{}, // TODO: empty for now. Once we use Waku Archive v2, we should deprecate the timestamp and use an ULID here Meta: []byte{}, // TODO: empty for now. Once we use NWaku Archive v2, we should deprecate the timestamp and use an ULID here
Ephemeral: &req.Ephemeral, Ephemeral: &req.Ephemeral,
} }

8
wakuv2/config.go
View File

@ -23,8 +23,6 @@ import (
"go.uber.org/zap" "go.uber.org/zap"
"github.com/status-im/status-go/protocol/common/shard"
ethdisc "github.com/ethereum/go-ethereum/p2p/dnsdisc" ethdisc "github.com/ethereum/go-ethereum/p2p/dnsdisc"
"github.com/status-im/status-go/wakuv2/common" "github.com/status-im/status-go/wakuv2/common"
@ -117,10 +115,10 @@ func setDefaults(cfg *Config) *Config {
} }
if cfg.DefaultShardPubsubTopic == "" { if cfg.DefaultShardPubsubTopic == "" {
cfg.DefaultShardPubsubTopic = shard.DefaultShardPubsubTopic() cfg.DefaultShardPubsubTopic = DefaultShardPubsubTopic()
//For now populating with both used shards, but this can be populated from user subscribed communities etc once community sharding is implemented //For now populating with both used shards, but this can be populated from user subscribed communities etc once community sharding is implemented
cfg.DefaultShardedPubsubTopics = append(cfg.DefaultShardedPubsubTopics, shard.DefaultShardPubsubTopic()) cfg.DefaultShardedPubsubTopics = append(cfg.DefaultShardedPubsubTopics, DefaultShardPubsubTopic())
cfg.DefaultShardedPubsubTopics = append(cfg.DefaultShardedPubsubTopics, shard.DefaultNonProtectedPubsubTopic()) cfg.DefaultShardedPubsubTopics = append(cfg.DefaultShardedPubsubTopics, DefaultNonProtectedPubsubTopic())
} }
return cfg return cfg

29
wakuv2/message_publishing.go
View File

@ -1,13 +1,13 @@
package wakuv2 package wakuv2
import ( import (
"encoding/json"
"errors" "errors"
"go.uber.org/zap" "go.uber.org/zap"
"github.com/waku-org/go-waku/waku/v2/api/publish" "github.com/waku-org/go-waku/waku/v2/api/publish"
"github.com/waku-org/go-waku/waku/v2/protocol" "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/pb"
"github.com/waku-org/go-waku/waku/v2/protocol/relay" "github.com/waku-org/go-waku/waku/v2/protocol/relay"
@ -35,7 +35,7 @@ func (pm PublishMethod) String() string {
// Send injects a message into the waku send queue, to be distributed in the // Send injects a message into the waku send queue, to be distributed in the
// network in the coming cycles. // network in the coming cycles.
func (w *Waku) Send(pubsubTopic string, msg *pb.WakuMessage, priority *int) ([]byte, error) { func (w *NWaku) Send(pubsubTopic string, msg *pb.WakuMessage, priority *int) ([]byte, error) {
pubsubTopic = w.GetPubsubTopic(pubsubTopic) pubsubTopic = w.GetPubsubTopic(pubsubTopic)
if w.protectedTopicStore != nil { if w.protectedTopicStore != nil {
privKey, err := w.protectedTopicStore.FetchPrivateKey(pubsubTopic) privKey, err := w.protectedTopicStore.FetchPrivateKey(pubsubTopic)
@ -77,7 +77,7 @@ func (w *Waku) Send(pubsubTopic string, msg *pb.WakuMessage, priority *int) ([]b
return envelope.Hash().Bytes(), nil return envelope.Hash().Bytes(), nil
} }
func (w *Waku) broadcast() { func (w *NWaku) broadcast() {
for { for {
var envelope *protocol.Envelope var envelope *protocol.Envelope
@ -103,15 +103,30 @@ func (w *Waku) broadcast() {
publishMethod = LightPush publishMethod = LightPush
fn = func(env *protocol.Envelope, logger *zap.Logger) error { fn = func(env *protocol.Envelope, logger *zap.Logger) error {
logger.Info("publishing message via lightpush") logger.Info("publishing message via lightpush")
_, err := w.node.Lightpush().Publish(w.ctx, env.Message(), lightpush.WithPubSubTopic(env.PubsubTopic()), lightpush.WithMaxPeers(peersToPublishForLightpush)) jsonMsg, err := json.Marshal(env.Message())
if err != nil {
return err
}
_, err = w.WakuLightpushPublish(string(jsonMsg), env.PubsubTopic())
return err return err
} }
} else { } else {
publishMethod = Relay publishMethod = Relay
fn = func(env *protocol.Envelope, logger *zap.Logger) error { fn = func(env *protocol.Envelope, logger *zap.Logger) error {
peerCnt := len(w.node.Relay().PubSub().ListPeers(env.PubsubTopic())) peerCnt, err := w.ListPeersInMesh(env.PubsubTopic())
if err != nil {
return err
}
logger.Info("publishing message via relay", zap.Int("peerCnt", peerCnt)) logger.Info("publishing message via relay", zap.Int("peerCnt", peerCnt))
_, err := w.node.Relay().Publish(w.ctx, env.Message(), relay.WithPubSubTopic(env.PubsubTopic())) timeoutMs := 1000
msg, err := json.Marshal(env.Message())
if err != nil {
return err
}
_, err = w.WakuRelayPublish(env.PubsubTopic(), string(msg), timeoutMs)
return err return err
} }
} }
@ -138,7 +153,7 @@ func (w *Waku) broadcast() {
} }
} }
func (w *Waku) publishEnvelope(envelope *protocol.Envelope, publishFn publish.PublishFn, logger *zap.Logger) { func (w *NWaku) publishEnvelope(envelope *protocol.Envelope, publishFn publish.PublishFn, logger *zap.Logger) {
defer w.wg.Done() defer w.wg.Done()
if err := publishFn(envelope, logger); err != nil { if err := publishFn(envelope, logger); err != nil {

2448
wakuv2/nwaku.go
View File

File diff suppressed because it is too large Load Diff

2
wakuv2/tracer.go
View File

@ -11,7 +11,7 @@ import (
// Trace implements EventTracer interface. // Trace implements EventTracer interface.
// We use custom logging, because we want to base58-encode the peerIDs. And also make the messageIDs readable. // We use custom logging, because we want to base58-encode the peerIDs. And also make the messageIDs readable.
func (w *Waku) Trace(evt *pubsub_pb.TraceEvent) { func (w *NWaku) Trace(evt *pubsub_pb.TraceEvent) {
f := []zap.Field{ f := []zap.Field{
zap.String("type", evt.Type.String()), zap.String("type", evt.Type.String()),

1858
wakuv2/waku.go
View File

File diff suppressed because it is too large Load Diff