mirror of
https://github.com/logos-messaging/message-finder.git
synced 2026-01-02 14:03:11 +00:00
94 lines
2.6 KiB
Go
94 lines
2.6 KiB
Go
package main
|
|
|
|
import (
|
|
"bytes"
|
|
"context"
|
|
"encoding/hex"
|
|
"fmt"
|
|
"time"
|
|
|
|
"github.com/libp2p/go-libp2p/core/peer"
|
|
"github.com/multiformats/go-multiaddr"
|
|
"github.com/waku-org/go-waku/logging"
|
|
"github.com/waku-org/go-waku/waku/v2/node"
|
|
"github.com/waku-org/go-waku/waku/v2/peerstore"
|
|
"github.com/waku-org/go-waku/waku/v2/protocol/store"
|
|
"github.com/waku-org/go-waku/waku/v2/utils"
|
|
"go.uber.org/zap"
|
|
|
|
"google.golang.org/protobuf/proto"
|
|
)
|
|
|
|
var log = utils.Logger().Named("message-finder")
|
|
|
|
func addNodes(ctx context.Context, node *node.WakuNode) {
|
|
for _, addr := range nodeList {
|
|
ma, err := multiaddr.NewMultiaddr(addr)
|
|
if err != nil {
|
|
log.Error("invalid multiaddress", zap.Error(err), zap.String("addr", addr))
|
|
continue
|
|
}
|
|
|
|
_ = ma
|
|
|
|
_, err = node.AddPeer(ma, peerstore.Static, []string{string(store.StoreID_v20beta4)})
|
|
if err != nil {
|
|
log.Error("could not add peer", zap.Error(err), zap.Stringer("addr", ma))
|
|
continue
|
|
}
|
|
}
|
|
}
|
|
|
|
func queryNode(ctx context.Context, node *node.WakuNode, addr string, pubsubTopic string, contentTopics []string, startTime time.Time, endTime time.Time, envelopeHash []byte) (int, error) {
|
|
p, err := multiaddr.NewMultiaddr(addr)
|
|
if err != nil {
|
|
return -1, err
|
|
}
|
|
|
|
info, err := peer.AddrInfoFromP2pAddr(p)
|
|
if err != nil {
|
|
return -1, err
|
|
}
|
|
|
|
cnt := 0
|
|
cursorIterations := 0
|
|
|
|
result, err := node.Store().Query(ctx, store.Query{
|
|
PubsubTopic: pubsubTopic,
|
|
ContentTopics: contentTopics,
|
|
StartTime: proto.Int64(startTime.UnixNano()),
|
|
EndTime: proto.Int64(endTime.UnixNano()),
|
|
}, store.WithPeer(info.ID), store.WithPaging(false, 20), store.WithRequestID([]byte{1, 2, 3, 4, 5, 6, 7, 8}))
|
|
if err != nil {
|
|
return -1, err
|
|
}
|
|
|
|
for {
|
|
hasNext, err := result.Next(ctx)
|
|
if err != nil {
|
|
return -1, err
|
|
}
|
|
|
|
if !hasNext { // No more messages available
|
|
break
|
|
}
|
|
cursorIterations += 1
|
|
|
|
// uncomment to find message by ID
|
|
for _, m := range result.GetMessages() {
|
|
if len(envelopeHash) != 0 && bytes.Equal(m.Hash(pubsubTopic), envelopeHash) {
|
|
log.Info("⚠️⚠️⚠️ MESSAGE FOUND ⚠️⚠️⚠️", logging.HexBytes("envelopeHash", envelopeHash), logging.HostID("storeNode", info.ID))
|
|
return 0, nil
|
|
} else {
|
|
log.Info("", zap.String("envelopeHash", hex.EncodeToString(m.Hash(pubsubTopic))), zap.String("contentTopic", m.ContentTopic), zap.String("timestamp", fmt.Sprintf("%d", m.GetTimestamp())), logging.HostID("storeNode", info.ID), zap.Int("page", cursorIterations))
|
|
}
|
|
}
|
|
|
|
cnt += len(result.GetMessages())
|
|
}
|
|
|
|
log.Info(fmt.Sprintf("%d messages found in %s (Used cursor %d times)\n", cnt, info.ID, cursorIterations))
|
|
|
|
return cnt, nil
|
|
}
|