mirror of
https://github.com/waku-org/nwaku.git
synced 2025-02-16 00:47:24 +00:00
f54ba10bc7
* queue driver refactor (#2753) * chore(archive): archive refactor (#2752) * chore(archive): sqlite driver refactor (#2754) * chore(archive): postgres driver refactor (#2755) * chore(archive): renaming & copies (#2751) * posgres legacy: stop using the storedAt field * migration script 6: we still need the id column The id column is needed because it contains the message digest which is used in store v2, and we need to keep support to store v2 for a while * legacy archive: set target migration version to 6 * waku_node: try to use wakuLegacyArchive if wakuArchive is nil * node_factory, waku_node: mount legacy and future store simultaneously We want the nwaku node to simultaneously support store-v2 requests and store-v3 requests. Only the legacy archive is in charge of archiving messages, and the archived information is suitable to fulfill both store-v2 and store-v3 needs. * postgres_driver: adding temporary code until store-v2 is removed --------- Co-authored-by: Ivan FB <128452529+Ivansete-status@users.noreply.github.com> Co-authored-by: gabrielmer <101006718+gabrielmer@users.noreply.github.com> Co-authored-by: Ivan Folgueira Bande <ivansete@status.im>
873 lines
28 KiB
Nim
873 lines
28 KiB
Nim
{.used.}
|
|
|
|
import
|
|
std/[options, sugar],
|
|
stew/shims/net as stewNet,
|
|
chronicles,
|
|
chronos/timer,
|
|
testutils/unittests,
|
|
eth/keys,
|
|
presto,
|
|
presto/client as presto_client,
|
|
libp2p/crypto/crypto
|
|
import
|
|
waku/[
|
|
waku_core/message,
|
|
waku_core/message/digest,
|
|
waku_core/topics,
|
|
waku_core/time,
|
|
waku_node,
|
|
node/peer_manager,
|
|
waku_api/rest/server,
|
|
waku_api/rest/client,
|
|
waku_api/rest/responses,
|
|
waku_api/rest/store/handlers as store_api,
|
|
waku_api/rest/store/client as store_api_client,
|
|
waku_api/rest/store/types,
|
|
waku_archive,
|
|
waku_archive/driver/queue_driver,
|
|
waku_archive/driver/sqlite_driver,
|
|
common/databases/db_sqlite,
|
|
waku_archive/driver/postgres_driver,
|
|
waku_store as waku_store,
|
|
],
|
|
../testlib/wakucore,
|
|
../testlib/wakunode
|
|
|
|
logScope:
|
|
topics = "waku node rest store_api test"
|
|
|
|
proc put(
|
|
store: ArchiveDriver, pubsubTopic: PubsubTopic, message: WakuMessage
|
|
): Future[Result[void, string]] =
|
|
let msgHash = computeMessageHash(pubsubTopic, message)
|
|
|
|
store.put(msgHash, pubsubTopic, message)
|
|
|
|
# Creates a new WakuNode
|
|
proc testWakuNode(): WakuNode =
|
|
let
|
|
privkey = generateSecp256k1Key()
|
|
bindIp = parseIpAddress("0.0.0.0")
|
|
extIp = parseIpAddress("127.0.0.1")
|
|
port = Port(0)
|
|
|
|
return newTestWakuNode(privkey, bindIp, port, some(extIp), some(port))
|
|
|
|
################################################################################
|
|
# Beginning of the tests
|
|
################################################################################
|
|
procSuite "Waku Rest API - Store v3":
|
|
asyncTest "MessageHash <-> string conversions":
|
|
# Validate MessageHash conversion from a WakuMessage obj
|
|
let wakuMsg = WakuMessage(
|
|
contentTopic: "Test content topic", payload: @[byte('H'), byte('i'), byte('!')]
|
|
)
|
|
|
|
let messageHash = computeMessageHash(DefaultPubsubTopic, wakuMsg)
|
|
let restMsgHash = some(messageHash.toRestStringWakuMessageHash())
|
|
|
|
let parsedMsgHashRes = parseHash(restMsgHash)
|
|
assert parsedMsgHashRes.isOk(), $parsedMsgHashRes.error
|
|
|
|
check:
|
|
messageHash == parsedMsgHashRes.get().get()
|
|
|
|
# Random validation. Obtained the raw values manually
|
|
let expected = some("f6za9OzG1xSiEZagZc2b3litRbkd3zRl61rezDd3pgQ%3D")
|
|
|
|
let msgHashRes = parseHash(expected)
|
|
assert msgHashRes.isOk(), $msgHashRes.error
|
|
|
|
check:
|
|
expected.get() == msgHashRes.get().get().toRestStringWakuMessageHash()
|
|
|
|
asyncTest "invalid cursor":
|
|
let node = testWakuNode()
|
|
await node.start()
|
|
await node.mountRelay()
|
|
|
|
var restPort = Port(0)
|
|
let restAddress = parseIpAddress("0.0.0.0")
|
|
let restServer = WakuRestServerRef.init(restAddress, restPort).tryGet()
|
|
restPort = restServer.httpServer.address.port # update with bound port for client use
|
|
|
|
installStoreApiHandlers(restServer.router, node)
|
|
restServer.start()
|
|
|
|
# WakuStore setup
|
|
let db: SqliteDatabase =
|
|
SqliteDatabase.new(string.none().get(":memory:")).expect("valid DB")
|
|
let driver: ArchiveDriver = SqliteDriver.new(db).expect("valid driver")
|
|
let mountArchiveRes = node.mountArchive(driver)
|
|
assert mountArchiveRes.isOk(), mountArchiveRes.error
|
|
|
|
await node.mountStore()
|
|
node.mountStoreClient()
|
|
|
|
let key = generateEcdsaKey()
|
|
var peerSwitch = newStandardSwitch(some(key))
|
|
await peerSwitch.start()
|
|
|
|
peerSwitch.mount(node.wakuStore)
|
|
|
|
await sleepAsync(1.seconds())
|
|
|
|
# Now prime it with some history before tests
|
|
let msgList =
|
|
@[
|
|
fakeWakuMessage(@[byte 0], contentTopic = ContentTopic("ct1"), ts = 0),
|
|
fakeWakuMessage(@[byte 1], ts = 1),
|
|
fakeWakuMessage(@[byte 1, byte 2], ts = 2),
|
|
fakeWakuMessage(@[byte 1], ts = 3),
|
|
fakeWakuMessage(@[byte 1], ts = 4),
|
|
fakeWakuMessage(@[byte 1], ts = 5),
|
|
fakeWakuMessage(@[byte 1], ts = 6),
|
|
fakeWakuMessage(@[byte 9], contentTopic = ContentTopic("c2"), ts = 9),
|
|
]
|
|
for msg in msgList:
|
|
require (await driver.put(DefaultPubsubTopic, msg)).isOk()
|
|
|
|
let client = newRestHttpClient(initTAddress(restAddress, restPort))
|
|
|
|
let remotePeerInfo = peerSwitch.peerInfo.toRemotePeerInfo()
|
|
let fullAddr = $remotePeerInfo.addrs[0] & "/p2p/" & $remotePeerInfo.peerId
|
|
|
|
await sleepAsync(1.seconds())
|
|
|
|
let fakeCursor = computeMessageHash(DefaultPubsubTopic, fakeWakuMessage())
|
|
let encodedCursor = fakeCursor.toRestStringWakuMessageHash()
|
|
|
|
# Apply filter by start and end timestamps
|
|
var response = await client.getStoreMessagesV3(
|
|
encodeUrl(fullAddr),
|
|
"true", # include data
|
|
"", # pubsub topic
|
|
"ct1,c2", # empty content topics.
|
|
"", # start time
|
|
"", # end time
|
|
"", # hashes
|
|
encodedCursor, # base64-encoded hash
|
|
"true", # ascending
|
|
"5", # empty implies default page size
|
|
)
|
|
|
|
check:
|
|
response.status == 200
|
|
$response.contentType == $MIMETYPE_JSON
|
|
response.data.messages.len == 0
|
|
|
|
await restServer.stop()
|
|
await restServer.closeWait()
|
|
await node.stop()
|
|
|
|
asyncTest "Filter by start and end time":
|
|
let node = testWakuNode()
|
|
await node.start()
|
|
await node.mountRelay()
|
|
|
|
var restPort = Port(0)
|
|
let restAddress = parseIpAddress("0.0.0.0")
|
|
let restServer = WakuRestServerRef.init(restAddress, restPort).tryGet()
|
|
restPort = restServer.httpServer.address.port # update with bound port for client use
|
|
|
|
installStoreApiHandlers(restServer.router, node)
|
|
restServer.start()
|
|
|
|
# WakuStore setup
|
|
let driver: ArchiveDriver = QueueDriver.new()
|
|
let mountArchiveRes = node.mountArchive(driver)
|
|
assert mountArchiveRes.isOk(), mountArchiveRes.error
|
|
|
|
await node.mountStore()
|
|
node.mountStoreClient()
|
|
|
|
let key = generateEcdsaKey()
|
|
var peerSwitch = newStandardSwitch(some(key))
|
|
await peerSwitch.start()
|
|
|
|
peerSwitch.mount(node.wakuStore)
|
|
|
|
# Now prime it with some history before tests
|
|
let msgList =
|
|
@[
|
|
fakeWakuMessage(@[byte 0], contentTopic = ContentTopic("ct1"), ts = 0),
|
|
fakeWakuMessage(@[byte 1], ts = 1),
|
|
fakeWakuMessage(@[byte 1, byte 2], ts = 2),
|
|
fakeWakuMessage(@[byte 1], ts = 3),
|
|
fakeWakuMessage(@[byte 1], ts = 4),
|
|
fakeWakuMessage(@[byte 1], ts = 5),
|
|
fakeWakuMessage(@[byte 1], ts = 6),
|
|
fakeWakuMessage(@[byte 9], contentTopic = ContentTopic("c2"), ts = 9),
|
|
]
|
|
for msg in msgList:
|
|
require (await driver.put(DefaultPubsubTopic, msg)).isOk()
|
|
|
|
let client = newRestHttpClient(initTAddress(restAddress, restPort))
|
|
|
|
let remotePeerInfo = peerSwitch.peerInfo.toRemotePeerInfo()
|
|
let fullAddr = $remotePeerInfo.addrs[0] & "/p2p/" & $remotePeerInfo.peerId
|
|
|
|
# Apply filter by start and end timestamps
|
|
var response = await client.getStoreMessagesV3(
|
|
encodeUrl(fullAddr),
|
|
"true", # include data
|
|
encodeUrl(DefaultPubsubTopic),
|
|
"", # empty content topics. Don't filter by this field
|
|
"3", # start time
|
|
"6", # end time
|
|
"", # hashes
|
|
"", # base64-encoded hash
|
|
"true", # ascending
|
|
"", # empty implies default page size
|
|
)
|
|
|
|
check:
|
|
response.status == 200
|
|
$response.contentType == $MIMETYPE_JSON
|
|
response.data.messages.len == 4
|
|
|
|
await restServer.stop()
|
|
await restServer.closeWait()
|
|
await node.stop()
|
|
|
|
asyncTest "Store node history response - forward pagination":
|
|
# Test adapted from the analogous present at waku_store/test_wakunode_store.nim
|
|
let node = testWakuNode()
|
|
await node.start()
|
|
|
|
var restPort = Port(0)
|
|
let restAddress = parseIpAddress("0.0.0.0")
|
|
let restServer = WakuRestServerRef.init(restAddress, restPort).tryGet()
|
|
restPort = restServer.httpServer.address.port # update with bound port for client use
|
|
|
|
installStoreApiHandlers(restServer.router, node)
|
|
restServer.start()
|
|
|
|
# WakuStore setup
|
|
let driver: ArchiveDriver = QueueDriver.new()
|
|
let mountArchiveRes = node.mountArchive(driver)
|
|
assert mountArchiveRes.isOk(), mountArchiveRes.error
|
|
|
|
await node.mountStore()
|
|
node.mountStoreClient()
|
|
|
|
let key = generateEcdsaKey()
|
|
var peerSwitch = newStandardSwitch(some(key))
|
|
await peerSwitch.start()
|
|
|
|
peerSwitch.mount(node.wakuStore)
|
|
|
|
# Now prime it with some history before tests
|
|
let timeOrigin = wakucore.now()
|
|
let msgList =
|
|
@[
|
|
fakeWakuMessage(@[byte 00], ts = ts(00, timeOrigin)),
|
|
fakeWakuMessage(@[byte 01], ts = ts(10, timeOrigin)),
|
|
fakeWakuMessage(@[byte 02], ts = ts(20, timeOrigin)),
|
|
fakeWakuMessage(@[byte 03], ts = ts(30, timeOrigin)),
|
|
fakeWakuMessage(@[byte 04], ts = ts(40, timeOrigin)),
|
|
fakeWakuMessage(@[byte 05], ts = ts(50, timeOrigin)),
|
|
fakeWakuMessage(@[byte 06], ts = ts(60, timeOrigin)),
|
|
fakeWakuMessage(@[byte 07], ts = ts(70, timeOrigin)),
|
|
fakeWakuMessage(@[byte 08], ts = ts(80, timeOrigin)),
|
|
fakeWakuMessage(@[byte 09], ts = ts(90, timeOrigin)),
|
|
]
|
|
for msg in msgList:
|
|
require (await driver.put(DefaultPubsubTopic, msg)).isOk()
|
|
|
|
let client = newRestHttpClient(initTAddress(restAddress, restPort))
|
|
|
|
let remotePeerInfo = peerSwitch.peerInfo.toRemotePeerInfo()
|
|
let fullAddr = $remotePeerInfo.addrs[0] & "/p2p/" & $remotePeerInfo.peerId
|
|
|
|
var pages = newSeq[seq[WakuMessage]](2)
|
|
|
|
var reqHash = none(WakuMessageHash)
|
|
|
|
for i in 0 ..< 2:
|
|
let response = await client.getStoreMessagesV3(
|
|
encodeUrl(fullAddr),
|
|
"true", # include data
|
|
encodeUrl(DefaultPubsubTopic),
|
|
"", # content topics. Empty ignores the field.
|
|
"", # start time. Empty ignores the field.
|
|
"", # end time. Empty ignores the field.
|
|
"", # hashes
|
|
if reqHash.isSome():
|
|
reqHash.get().toRestStringWakuMessageHash()
|
|
else:
|
|
""
|
|
, # base64-encoded digest. Empty ignores the field.
|
|
"true", # ascending
|
|
"7", # page size. Empty implies default page size.
|
|
)
|
|
|
|
let wakuMessages = collect(newSeq):
|
|
for element in response.data.messages:
|
|
if element.message.isSome():
|
|
element.message.get()
|
|
|
|
pages[i] = wakuMessages
|
|
|
|
# populate the cursor for next page
|
|
if response.data.paginationCursor.isSome():
|
|
reqHash = some(response.data.paginationCursor.get())
|
|
|
|
check:
|
|
response.status == 200
|
|
$response.contentType == $MIMETYPE_JSON
|
|
|
|
check:
|
|
pages[0] == msgList[0 .. 6]
|
|
pages[1] == msgList[7 .. 9]
|
|
|
|
await restServer.stop()
|
|
await restServer.closeWait()
|
|
await node.stop()
|
|
|
|
asyncTest "query a node and retrieve historical messages filtered by pubsub topic":
|
|
# Given
|
|
let node = testWakuNode()
|
|
await node.start()
|
|
await node.mountRelay()
|
|
|
|
var restPort = Port(0)
|
|
let restAddress = parseIpAddress("0.0.0.0")
|
|
let restServer = WakuRestServerRef.init(restAddress, restPort).tryGet()
|
|
restPort = restServer.httpServer.address.port # update with bound port for client use
|
|
|
|
installStoreApiHandlers(restServer.router, node)
|
|
restServer.start()
|
|
|
|
# WakuStore setup
|
|
let driver: ArchiveDriver = QueueDriver.new()
|
|
let mountArchiveRes = node.mountArchive(driver)
|
|
assert mountArchiveRes.isOk(), mountArchiveRes.error
|
|
|
|
await node.mountStore()
|
|
node.mountStoreClient()
|
|
|
|
let key = generateEcdsaKey()
|
|
var peerSwitch = newStandardSwitch(some(key))
|
|
await peerSwitch.start()
|
|
|
|
peerSwitch.mount(node.wakuStore)
|
|
|
|
# Now prime it with some history before tests
|
|
let msgList =
|
|
@[
|
|
fakeWakuMessage(@[byte 0], contentTopic = ContentTopic("2"), ts = 0),
|
|
fakeWakuMessage(@[byte 1], ts = 1),
|
|
fakeWakuMessage(@[byte 9], contentTopic = ContentTopic("2"), ts = 9),
|
|
]
|
|
for msg in msgList:
|
|
require (await driver.put(DefaultPubsubTopic, msg)).isOk()
|
|
|
|
let client = newRestHttpClient(initTAddress(restAddress, restPort))
|
|
|
|
let remotePeerInfo = peerSwitch.peerInfo.toRemotePeerInfo()
|
|
let fullAddr = $remotePeerInfo.addrs[0] & "/p2p/" & $remotePeerInfo.peerId
|
|
|
|
# Filtering by a known pubsub topic
|
|
var response = await client.getStoreMessagesV3(
|
|
encodeUrl($fullAddr), "true", encodeUrl(DefaultPubsubTopic)
|
|
)
|
|
|
|
check:
|
|
response.status == 200
|
|
$response.contentType == $MIMETYPE_JSON
|
|
response.data.messages.len == 3
|
|
|
|
# Get all the messages by specifying an empty pubsub topic
|
|
response = await client.getStoreMessagesV3(encodeUrl($fullAddr), "true")
|
|
check:
|
|
response.status == 200
|
|
$response.contentType == $MIMETYPE_JSON
|
|
response.data.messages.len == 3
|
|
|
|
# Receiving no messages by filtering with a random pubsub topic
|
|
response = await client.getStoreMessagesV3(
|
|
encodeUrl($fullAddr), "true", encodeUrl("random pubsub topic")
|
|
)
|
|
check:
|
|
response.status == 200
|
|
$response.contentType == $MIMETYPE_JSON
|
|
response.data.messages.len == 0
|
|
|
|
await restServer.stop()
|
|
await restServer.closeWait()
|
|
await node.stop()
|
|
|
|
asyncTest "retrieve historical messages from a provided store node address":
|
|
# Given
|
|
let node = testWakuNode()
|
|
await node.start()
|
|
await node.mountRelay()
|
|
|
|
var restPort = Port(0)
|
|
let restAddress = parseIpAddress("0.0.0.0")
|
|
let restServer = WakuRestServerRef.init(restAddress, restPort).tryGet()
|
|
restPort = restServer.httpServer.address.port # update with bound port for client use
|
|
|
|
installStoreApiHandlers(restServer.router, node)
|
|
restServer.start()
|
|
|
|
# WakuStore setup
|
|
let driver: ArchiveDriver = QueueDriver.new()
|
|
let mountArchiveRes = node.mountArchive(driver)
|
|
assert mountArchiveRes.isOk(), mountArchiveRes.error
|
|
|
|
await node.mountStore()
|
|
node.mountStoreClient()
|
|
|
|
let key = generateEcdsaKey()
|
|
var peerSwitch = newStandardSwitch(some(key))
|
|
await peerSwitch.start()
|
|
|
|
peerSwitch.mount(node.wakuStore)
|
|
|
|
# Now prime it with some history before tests
|
|
let msgList =
|
|
@[
|
|
fakeWakuMessage(@[byte 0], contentTopic = ContentTopic("ct1"), ts = 0),
|
|
fakeWakuMessage(@[byte 1], ts = 1),
|
|
fakeWakuMessage(@[byte 9], contentTopic = ContentTopic("ct2"), ts = 9),
|
|
]
|
|
for msg in msgList:
|
|
require (await driver.put(DefaultPubsubTopic, msg)).isOk()
|
|
|
|
let client = newRestHttpClient(initTAddress(restAddress, restPort))
|
|
|
|
let remotePeerInfo = peerSwitch.peerInfo.toRemotePeerInfo()
|
|
let fullAddr = $remotePeerInfo.addrs[0] & "/p2p/" & $remotePeerInfo.peerId
|
|
|
|
# Filtering by a known pubsub topic.
|
|
# We also pass the store-node address in the request.
|
|
var response = await client.getStoreMessagesV3(
|
|
encodeUrl(fullAddr), "true", encodeUrl(DefaultPubsubTopic)
|
|
)
|
|
check:
|
|
response.status == 200
|
|
$response.contentType == $MIMETYPE_JSON
|
|
response.data.messages.len == 3
|
|
|
|
# Get all the messages by specifying an empty pubsub topic
|
|
# We also pass the store-node address in the request.
|
|
response =
|
|
await client.getStoreMessagesV3(encodeUrl(fullAddr), "true", encodeUrl(""))
|
|
check:
|
|
response.status == 200
|
|
$response.contentType == $MIMETYPE_JSON
|
|
response.data.messages.len == 3
|
|
|
|
# Receiving no messages by filtering with a random pubsub topic
|
|
# We also pass the store-node address in the request.
|
|
response = await client.getStoreMessagesV3(
|
|
encodeUrl(fullAddr), "true", encodeUrl("random pubsub topic")
|
|
)
|
|
check:
|
|
response.status == 200
|
|
$response.contentType == $MIMETYPE_JSON
|
|
response.data.messages.len == 0
|
|
|
|
# Receiving 400 response if setting wrong store-node address
|
|
response = await client.getStoreMessagesV3(
|
|
encodeUrl("incorrect multi address format"),
|
|
"true",
|
|
encodeUrl("random pubsub topic"),
|
|
)
|
|
check:
|
|
response.status == 400
|
|
$response.contentType == $MIMETYPE_TEXT
|
|
response.data.messages.len == 0
|
|
response.data.statusDesc ==
|
|
"Failed parsing remote peer info [MultiAddress.init [multiaddress: Invalid MultiAddress, must start with `/`]]"
|
|
|
|
await restServer.stop()
|
|
await restServer.closeWait()
|
|
await node.stop()
|
|
|
|
asyncTest "filter historical messages by content topic":
|
|
# Given
|
|
let node = testWakuNode()
|
|
await node.start()
|
|
await node.mountRelay()
|
|
|
|
var restPort = Port(0)
|
|
let restAddress = parseIpAddress("0.0.0.0")
|
|
let restServer = WakuRestServerRef.init(restAddress, restPort).tryGet()
|
|
restPort = restServer.httpServer.address.port # update with bound port for client use
|
|
|
|
installStoreApiHandlers(restServer.router, node)
|
|
restServer.start()
|
|
|
|
# WakuStore setup
|
|
let driver: ArchiveDriver = QueueDriver.new()
|
|
let mountArchiveRes = node.mountArchive(driver)
|
|
assert mountArchiveRes.isOk(), mountArchiveRes.error
|
|
|
|
await node.mountStore()
|
|
node.mountStoreClient()
|
|
|
|
let key = generateEcdsaKey()
|
|
var peerSwitch = newStandardSwitch(some(key))
|
|
await peerSwitch.start()
|
|
|
|
peerSwitch.mount(node.wakuStore)
|
|
|
|
# Now prime it with some history before tests
|
|
let msgList =
|
|
@[
|
|
fakeWakuMessage(@[byte 0], contentTopic = ContentTopic("ct1"), ts = 0),
|
|
fakeWakuMessage(@[byte 1], ts = 1),
|
|
fakeWakuMessage(@[byte 9], contentTopic = ContentTopic("ct2"), ts = 9),
|
|
]
|
|
for msg in msgList:
|
|
require (await driver.put(DefaultPubsubTopic, msg)).isOk()
|
|
|
|
let client = newRestHttpClient(initTAddress(restAddress, restPort))
|
|
|
|
let remotePeerInfo = peerSwitch.peerInfo.toRemotePeerInfo()
|
|
let fullAddr = $remotePeerInfo.addrs[0] & "/p2p/" & $remotePeerInfo.peerId
|
|
|
|
# Filtering by content topic
|
|
let response = await client.getStoreMessagesV3(
|
|
encodeUrl(fullAddr), "true", encodeUrl(DefaultPubsubTopic), encodeUrl("ct1,ct2")
|
|
)
|
|
check:
|
|
response.status == 200
|
|
$response.contentType == $MIMETYPE_JSON
|
|
response.data.messages.len == 2
|
|
|
|
await restServer.stop()
|
|
await restServer.closeWait()
|
|
await node.stop()
|
|
|
|
asyncTest "precondition failed":
|
|
# Given
|
|
let node = testWakuNode()
|
|
await node.start()
|
|
await node.mountRelay()
|
|
|
|
var restPort = Port(0)
|
|
let restAddress = parseIpAddress("0.0.0.0")
|
|
let restServer = WakuRestServerRef.init(restAddress, restPort).tryGet()
|
|
restPort = restServer.httpServer.address.port # update with bound port for client use
|
|
|
|
installStoreApiHandlers(restServer.router, node)
|
|
restServer.start()
|
|
|
|
node.mountStoreClient()
|
|
|
|
let key = generateEcdsaKey()
|
|
var peerSwitch = newStandardSwitch(some(key))
|
|
await peerSwitch.start()
|
|
|
|
let client = newRestHttpClient(initTAddress(restAddress, restPort))
|
|
|
|
let remotePeerInfo = peerSwitch.peerInfo.toRemotePeerInfo()
|
|
|
|
# Sending no peer-store node address
|
|
var response = await client.getStoreMessagesV3(
|
|
encodeUrl(""), "true", encodeUrl(DefaultPubsubTopic)
|
|
)
|
|
check:
|
|
response.status == 412
|
|
$response.contentType == $MIMETYPE_TEXT
|
|
response.data.messages.len == 0
|
|
response.data.statusDesc == NoPeerNoDiscError.errobj.message
|
|
|
|
# Now add the storenode from "config"
|
|
node.peerManager.addServicePeer(remotePeerInfo, WakuStoreCodec)
|
|
|
|
# WakuStore setup
|
|
let driver: ArchiveDriver = QueueDriver.new()
|
|
let mountArchiveRes = node.mountArchive(driver)
|
|
assert mountArchiveRes.isOk(), mountArchiveRes.error
|
|
|
|
await node.mountStore()
|
|
|
|
# Now prime it with some history before tests
|
|
let msgList =
|
|
@[
|
|
fakeWakuMessage(@[byte 0], contentTopic = ContentTopic("ct1"), ts = 0),
|
|
fakeWakuMessage(@[byte 1], ts = 1),
|
|
fakeWakuMessage(@[byte 9], contentTopic = ContentTopic("ct2"), ts = 9),
|
|
]
|
|
for msg in msgList:
|
|
require (await driver.put(DefaultPubsubTopic, msg)).isOk()
|
|
|
|
# Sending no peer-store node address
|
|
response = await client.getStoreMessagesV3(
|
|
encodeUrl(""), "true", encodeUrl(DefaultPubsubTopic)
|
|
)
|
|
check:
|
|
response.status == 200
|
|
$response.contentType == $MIMETYPE_JSON
|
|
response.data.messages.len == 3
|
|
|
|
await restServer.stop()
|
|
await restServer.closeWait()
|
|
await node.stop()
|
|
|
|
asyncTest "retrieve historical messages from a self-store-node":
|
|
## This test aims to validate the correct message retrieval for a store-node which exposes
|
|
## a REST server.
|
|
|
|
# Given
|
|
let node = testWakuNode()
|
|
await node.start()
|
|
|
|
var restPort = Port(0)
|
|
let restAddress = parseIpAddress("0.0.0.0")
|
|
let restServer = WakuRestServerRef.init(restAddress, restPort).tryGet()
|
|
restPort = restServer.httpServer.address.port # update with bound port for client use
|
|
|
|
installStoreApiHandlers(restServer.router, node)
|
|
restServer.start()
|
|
|
|
# WakuStore setup
|
|
let driver: ArchiveDriver = QueueDriver.new()
|
|
let mountArchiveRes = node.mountArchive(driver)
|
|
assert mountArchiveRes.isOk(), mountArchiveRes.error
|
|
|
|
await node.mountStore()
|
|
|
|
# Now prime it with some history before tests
|
|
let msgList =
|
|
@[
|
|
fakeWakuMessage(
|
|
@[byte 0], contentTopic = ContentTopic("ct1"), ts = 0, meta = (@[byte 8])
|
|
),
|
|
fakeWakuMessage(@[byte 1], ts = 1),
|
|
fakeWakuMessage(@[byte 9], contentTopic = ContentTopic("ct2"), ts = 9),
|
|
]
|
|
for msg in msgList:
|
|
require (await driver.put(DefaultPubsubTopic, msg)).isOk()
|
|
|
|
let client = newRestHttpClient(initTAddress(restAddress, restPort))
|
|
|
|
# Filtering by a known pubsub topic.
|
|
var response = await client.getStoreMessagesV3(
|
|
includeData = "true", pubsubTopic = encodeUrl(DefaultPubsubTopic)
|
|
)
|
|
|
|
check:
|
|
response.status == 200
|
|
$response.contentType == $MIMETYPE_JSON
|
|
response.data.messages.len == 3
|
|
|
|
# Get all the messages by specifying an empty pubsub topic
|
|
response =
|
|
await client.getStoreMessagesV3(includeData = "true", pubsubTopic = encodeUrl(""))
|
|
check:
|
|
response.status == 200
|
|
$response.contentType == $MIMETYPE_JSON
|
|
response.data.messages.len == 3
|
|
|
|
# Receiving no messages by filtering with a random pubsub topic
|
|
response = await client.getStoreMessagesV3(
|
|
includeData = "true", pubsubTopic = encodeUrl("random pubsub topic")
|
|
)
|
|
check:
|
|
response.status == 200
|
|
$response.contentType == $MIMETYPE_JSON
|
|
response.data.messages.len == 0
|
|
|
|
asyncTest "correct message fields are returned":
|
|
# Given
|
|
let node = testWakuNode()
|
|
await node.start()
|
|
|
|
var restPort = Port(0)
|
|
let restAddress = parseIpAddress("0.0.0.0")
|
|
let restServer = WakuRestServerRef.init(restAddress, restPort).tryGet()
|
|
restPort = restServer.httpServer.address.port # update with bound port for client use
|
|
|
|
installStoreApiHandlers(restServer.router, node)
|
|
restServer.start()
|
|
|
|
# WakuStore setup
|
|
let driver: ArchiveDriver = QueueDriver.new()
|
|
let mountArchiveRes = node.mountArchive(driver)
|
|
assert mountArchiveRes.isOk(), mountArchiveRes.error
|
|
|
|
await node.mountStore()
|
|
|
|
# Now prime it with some history before tests
|
|
let msg = fakeWakuMessage(
|
|
@[byte 0], contentTopic = ContentTopic("ct1"), ts = 0, meta = (@[byte 8])
|
|
)
|
|
require (await driver.put(DefaultPubsubTopic, msg)).isOk()
|
|
|
|
let client = newRestHttpClient(initTAddress(restAddress, restPort))
|
|
|
|
# Filtering by a known pubsub topic.
|
|
var response = await client.getStoreMessagesV3(
|
|
includeData = "true", pubsubTopic = encodeUrl(DefaultPubsubTopic)
|
|
)
|
|
|
|
check:
|
|
response.status == 200
|
|
$response.contentType == $MIMETYPE_JSON
|
|
response.data.messages.len == 1
|
|
|
|
let storeMessage = response.data.messages[0].message.get()
|
|
|
|
check:
|
|
storeMessage.payload == msg.payload
|
|
storeMessage.contentTopic == msg.contentTopic
|
|
storeMessage.version == msg.version
|
|
storeMessage.timestamp == msg.timestamp
|
|
storeMessage.ephemeral == msg.ephemeral
|
|
storeMessage.meta == msg.meta
|
|
|
|
asyncTest "Rate limit store node store query":
|
|
# Test adapted from the analogous present at waku_store/test_wakunode_store.nim
|
|
let node = testWakuNode()
|
|
await node.start()
|
|
|
|
var restPort = Port(0)
|
|
let restAddress = parseIpAddress("0.0.0.0")
|
|
let restServer = WakuRestServerRef.init(restAddress, restPort).tryGet()
|
|
restPort = restServer.httpServer.address.port # update with bound port for client use
|
|
|
|
installStoreApiHandlers(restServer.router, node)
|
|
restServer.start()
|
|
|
|
# WakuStore setup
|
|
let driver: ArchiveDriver = QueueDriver.new()
|
|
let mountArchiveRes = node.mountArchive(driver)
|
|
assert mountArchiveRes.isOk(), mountArchiveRes.error
|
|
|
|
await node.mountStore((2, 500.millis))
|
|
node.mountStoreClient()
|
|
|
|
let key = generateEcdsaKey()
|
|
var peerSwitch = newStandardSwitch(some(key))
|
|
await peerSwitch.start()
|
|
|
|
peerSwitch.mount(node.wakuStore)
|
|
|
|
# Now prime it with some history before tests
|
|
let timeOrigin = wakucore.now()
|
|
let msgList =
|
|
@[
|
|
fakeWakuMessage(@[byte 00], ts = ts(00, timeOrigin)),
|
|
fakeWakuMessage(@[byte 01], ts = ts(10, timeOrigin)),
|
|
fakeWakuMessage(@[byte 02], ts = ts(20, timeOrigin)),
|
|
fakeWakuMessage(@[byte 03], ts = ts(30, timeOrigin)),
|
|
fakeWakuMessage(@[byte 04], ts = ts(40, timeOrigin)),
|
|
fakeWakuMessage(@[byte 05], ts = ts(50, timeOrigin)),
|
|
fakeWakuMessage(@[byte 06], ts = ts(60, timeOrigin)),
|
|
fakeWakuMessage(@[byte 07], ts = ts(70, timeOrigin)),
|
|
fakeWakuMessage(@[byte 08], ts = ts(80, timeOrigin)),
|
|
fakeWakuMessage(@[byte 09], ts = ts(90, timeOrigin)),
|
|
]
|
|
for msg in msgList:
|
|
require (await driver.put(DefaultPubsubTopic, msg)).isOk()
|
|
|
|
let client = newRestHttpClient(initTAddress(restAddress, restPort))
|
|
|
|
let remotePeerInfo = peerSwitch.peerInfo.toRemotePeerInfo()
|
|
let fullAddr = $remotePeerInfo.addrs[0] & "/p2p/" & $remotePeerInfo.peerId
|
|
|
|
var pages = newSeq[seq[WakuMessage]](2)
|
|
|
|
var reqPubsubTopic = DefaultPubsubTopic
|
|
var reqHash = none(WakuMessageHash)
|
|
|
|
for i in 0 ..< 2:
|
|
let response = await client.getStoreMessagesV3(
|
|
encodeUrl(fullAddr),
|
|
"true", # include data
|
|
encodeUrl(reqPubsubTopic),
|
|
"", # content topics. Empty ignores the field.
|
|
"", # start time. Empty ignores the field.
|
|
"", # end time. Empty ignores the field.
|
|
"", # hashes
|
|
if reqHash.isSome():
|
|
reqHash.get().toRestStringWakuMessageHash()
|
|
else:
|
|
""
|
|
, # base64-encoded digest. Empty ignores the field.
|
|
"true", # ascending
|
|
"3", # page size. Empty implies default page size.
|
|
)
|
|
|
|
let wakuMessages = collect(newSeq):
|
|
for element in response.data.messages:
|
|
if element.message.isSome():
|
|
element.message.get()
|
|
|
|
pages[i] = wakuMessages
|
|
|
|
# populate the cursor for next page
|
|
if response.data.paginationCursor.isSome():
|
|
reqHash = response.data.paginationCursor
|
|
|
|
check:
|
|
response.status == 200
|
|
$response.contentType == $MIMETYPE_JSON
|
|
|
|
check:
|
|
pages[0] == msgList[0 .. 2]
|
|
pages[1] == msgList[3 .. 5]
|
|
|
|
# request last third will lead to rate limit rejection
|
|
var response = await client.getStoreMessagesV3(
|
|
encodeUrl(fullAddr),
|
|
"true", # include data
|
|
encodeUrl(reqPubsubTopic),
|
|
"", # content topics. Empty ignores the field.
|
|
"", # start time. Empty ignores the field.
|
|
"", # end time. Empty ignores the field.
|
|
"", # hashes
|
|
if reqHash.isSome():
|
|
reqHash.get().toRestStringWakuMessageHash()
|
|
else:
|
|
""
|
|
, # base64-encoded digest. Empty ignores the field.
|
|
)
|
|
|
|
check:
|
|
response.status == 429
|
|
$response.contentType == $MIMETYPE_TEXT
|
|
response.data.statusDesc == "Request rate limit reached"
|
|
|
|
await sleepAsync(500.millis)
|
|
|
|
# retry after respective amount of time shall succeed
|
|
response = await client.getStoreMessagesV3(
|
|
encodeUrl(fullAddr),
|
|
"true", # include data
|
|
encodeUrl(reqPubsubTopic),
|
|
"", # content topics. Empty ignores the field.
|
|
"", # start time. Empty ignores the field.
|
|
"", # end time. Empty ignores the field.
|
|
"", # hashes
|
|
if reqHash.isSome():
|
|
reqHash.get().toRestStringWakuMessageHash()
|
|
else:
|
|
""
|
|
, # base64-encoded digest. Empty ignores the field.
|
|
"true", # ascending
|
|
"5", # page size. Empty implies default page size.
|
|
)
|
|
|
|
check:
|
|
response.status == 200
|
|
$response.contentType == $MIMETYPE_JSON
|
|
|
|
let wakuMessages = collect(newSeq):
|
|
for element in response.data.messages:
|
|
if element.message.isSome():
|
|
element.message.get()
|
|
|
|
check wakuMessages == msgList[6 .. 9]
|
|
|
|
await restServer.stop()
|
|
await restServer.closeWait()
|
|
await node.stop()
|