package library import ( "C" "encoding/json" "github.com/waku-org/go-waku/waku/v2/protocol/legacy_store/pb" wpb "github.com/waku-org/go-waku/waku/v2/protocol/pb" ) import ( "context" "time" "github.com/libp2p/go-libp2p/core/peer" "github.com/waku-org/go-waku/waku/v2/protocol/legacy_store" ) type storePagingOptions struct { PageSize uint64 `json:"pageSize,omitempty"` Cursor *pb.Index `json:"cursor,omitempty"` Forward bool `json:"forward,omitempty"` } type storeMessagesArgs struct { Topic string `json:"pubsubTopic,omitempty"` ContentTopics []string `json:"contentTopics,omitempty"` StartTime *int64 `json:"startTime,omitempty"` EndTime *int64 `json:"endTime,omitempty"` PagingOptions *storePagingOptions `json:"pagingOptions,omitempty"` } type storeMessagesReply struct { Messages []*wpb.WakuMessage `json:"messages,omitempty"` PagingInfo storePagingOptions `json:"pagingInfo,omitempty"` Error string `json:"error,omitempty"` } func queryResponse(ctx context.Context, instance *WakuInstance, args storeMessagesArgs, options []legacy_store.HistoryRequestOption) (string, error) { res, err := instance.node.LegacyStore().Query( ctx, legacy_store.Query{ PubsubTopic: args.Topic, ContentTopics: args.ContentTopics, StartTime: args.StartTime, EndTime: args.EndTime, }, options..., ) reply := storeMessagesReply{} if err != nil { reply.Error = err.Error() return marshalJSON(reply) } reply.Messages = res.Messages reply.PagingInfo = storePagingOptions{ PageSize: args.PagingOptions.PageSize, Cursor: res.Cursor(), Forward: args.PagingOptions.Forward, } return marshalJSON(reply) } // StoreQuery is used to retrieve historic messages using waku store protocol. func StoreQuery(instance *WakuInstance, queryJSON string, peerID string, ms int) (string, error) { if err := validateInstance(instance, MustBeStarted); err != nil { return "", err } var args storeMessagesArgs err := json.Unmarshal([]byte(queryJSON), &args) if err != nil { return "", err } options := []legacy_store.HistoryRequestOption{ legacy_store.WithAutomaticRequestID(), legacy_store.WithPaging(args.PagingOptions.Forward, args.PagingOptions.PageSize), legacy_store.WithCursor(args.PagingOptions.Cursor), } if peerID != "" { p, err := peer.Decode(peerID) if err != nil { return "", err } options = append(options, legacy_store.WithPeer(p)) } else { options = append(options, legacy_store.WithAutomaticPeerSelection()) } var ctx context.Context var cancel context.CancelFunc if ms > 0 { ctx, cancel = context.WithTimeout(instance.ctx, time.Duration(int(ms))*time.Millisecond) defer cancel() } else { ctx = instance.ctx } return queryResponse(ctx, instance, args, options) } // StoreLocalQuery is used to retrieve historic messages stored in the localDB using waku store protocol. func StoreLocalQuery(instance *WakuInstance, queryJSON string) (string, error) { if err := validateInstance(instance, MustBeStarted); err != nil { return "", err } var args storeMessagesArgs err := json.Unmarshal([]byte(queryJSON), &args) if err != nil { return "", err } options := []legacy_store.HistoryRequestOption{ legacy_store.WithAutomaticRequestID(), legacy_store.WithPaging(args.PagingOptions.Forward, args.PagingOptions.PageSize), legacy_store.WithCursor(args.PagingOptions.Cursor), legacy_store.WithLocalQuery(), } return queryResponse(instance.ctx, instance, args, options) }