diff --git a/logos_delivery/waku/node/waku_node/store.nim b/logos_delivery/waku/node/waku_node/store.nim index 92d236a37..dca89d5df 100644 --- a/logos_delivery/waku/node/waku_node/store.nim +++ b/logos_delivery/waku/node/waku_node/store.nim @@ -114,6 +114,19 @@ proc mountStore*( node.switch.mount(node.wakuStore, protocolMatcher(store_common.WakuStoreCodec)) +proc queryArchive*( + node: WakuNode, request: StoreQueryRequest +): Future[StoreQueryResult] {.async.} = + ## Serves a store query directly from the local archive, for nodes that + ## carry an archive without mounting the store protocol (e.g. full nodes + ## running store sync standalone). Same conversion path as mountStore's + ## request handler, so pagination/cursor semantics are identical. + if node.wakuArchive.isNil(): + return err(StoreError.new(300, "waku archive is not available")) + + let response = await node.wakuArchive.findMessages(request.toArchiveQuery()) + return response.toStoreResult() + proc mountStoreClient*(node: WakuNode) = info "mounting store client" diff --git a/logos_delivery/waku/rest_api/endpoint/store/handlers.nim b/logos_delivery/waku/rest_api/endpoint/store/handlers.nim index aa0136922..deee32072 100644 --- a/logos_delivery/waku/rest_api/endpoint/store/handlers.nim +++ b/logos_delivery/waku/rest_api/endpoint/store/handlers.nim @@ -179,6 +179,23 @@ proc retrieveMsgsFromSelfNode( return resp +proc retrieveMsgsFromArchive( + self: WakuNode, storeQuery: StoreQueryRequest +): Future[RestApiResponse] {.async.} = + ## Serves a local store query from the node's archive when the store + ## protocol is not mounted (e.g. store-sync standalone full nodes). + + let storeResp = (await self.queryArchive(storeQuery)).valueOr: + return RestApiResponse.internalServerError($error) + + let resp = RestApiResponse.jsonResponse(storeResp.toHex(), status = Http200).valueOr: + const msg = "Error building the json respose" + let e = $error + error msg, error = e + return RestApiResponse.internalServerError(fmt("{msg} [{e}]")) + + return resp + # Subscribes the rest handler to attend "/store/v1/messages" requests proc installStoreApiHandlers*( router: var RestRouter, @@ -226,6 +243,13 @@ proc installStoreApiHandlers*( ## the local/self store node. return await node.retrieveMsgsFromSelfNode(storeQuery) + if peer.isNone() and not node.wakuArchive.isNil(): + ## No peer given and no store protocol, but the node carries an archive + ## (store-sync standalone): serve the query from the local archive, like + ## a store node serving its own. Only the archive's retention window is + ## returned; pass peerAddr to query a remote store for deeper history. + return await node.retrieveMsgsFromArchive(storeQuery) + # Parse the peer address parameter let parsedPeerAddr = parseUrlPeerAddr(peer).valueOr: return RestApiResponse.badRequest(error) diff --git a/tests/wakunode_rest/test_rest_store.nim b/tests/wakunode_rest/test_rest_store.nim index 74847bf9b..f7534be7b 100644 --- a/tests/wakunode_rest/test_rest_store.nim +++ b/tests/wakunode_rest/test_rest_store.nim @@ -673,6 +673,82 @@ procSuite "Waku Rest API - Store v3": $response.contentType == $MIMETYPE_JSON response.data.messages.len == 0 + asyncTest "retrieve messages from an archive-only node (no store protocol)": + ## Store-sync standalone nodes carry an archive without mounting the + ## store protocol; REST self-queries must be served from the archive. + + # 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() + + # Archive only, deliberately no mountStore + let driver: ArchiveDriver = QueueDriver.new() + let mountArchiveRes = node.mountArchive(driver) + assert mountArchiveRes.isOk(), mountArchiveRes.error + + require node.wakuStore.isNil() + + let msgList = @[ + fakeWakuMessage(@[byte 0], ts = 0), + fakeWakuMessage(@[byte 1], ts = 1), + fakeWakuMessage(@[byte 2], ts = 2), + fakeWakuMessage(@[byte 3], ts = 3), + fakeWakuMessage(@[byte 4], ts = 4), + ] + for msg in msgList: + require (await driver.put(DefaultPubsubTopic, msg)).isOk() + + let client = newRestHttpClient(initTAddress(restAddress, restPort)) + + # Self-query without a peer address is served from the archive + let response = await client.getStoreMessagesV3( + includeData = "true", pubsubTopic = encodeUrl(DefaultPubsubTopic) + ) + + check: + response.status == 200 + $response.contentType == $MIMETYPE_JSON + response.data.messages.len == 5 + + # Cursor pagination works through the archive fallback + var received = newSeq[WakuMessage]() + var reqCursor = "" + for page in 0 ..< 3: + let resp = await client.getStoreMessagesV3( + includeData = "true", + pubsubTopic = encodeUrl(DefaultPubsubTopic), + cursor = reqCursor, + ascending = "true", + pageSize = "2", + ) + + check: + resp.status == 200 + $resp.contentType == $MIMETYPE_JSON + + for element in resp.data.messages: + if element.message.isSome(): + received.add(element.message.get()) + + if resp.data.paginationCursor.isSome(): + reqCursor = resp.data.paginationCursor.get() + else: + break + + check received == msgList + + await restServer.stop() + await restServer.closeWait() + await node.stop() + asyncTest "correct message fields are returned": # Given let node = testWakuNode()