feat: Add incentivization PoC with on-chain txid check (closes #2847)

tests/incentivization/test_all.nim

@@ -1,2 +1,3 @@
 import
-  ./test_rpc_codec
+  ./test_rpc_codec,
+  ./test_poc

tests/incentivization/test_poc.nim

@@ -0,0 +1,99 @@
+{.used.}
+
+import
+  std/[options, strscans],
+  testutils/unittests,
+  chronicles,
+  chronos,
+  libp2p/crypto/crypto,
+  web3
+
+import
+  ../../../waku/[
+    node/peer_manager,
+    waku_core,
+  ],
+  ../testlib/[assertions, wakucore, testasync, futures, testutils],
+  ../../../waku/incentivization/[
+    rpc,
+    rpc_codec,
+    common,
+    client,
+    protocol,
+    txid_proof
+  ]
+
+
+# a random confirmed txis (Sepolia)
+const TxHashExisting* = TxHash.fromHex(
+  "0xc1be5f442d3688a8d3e4b5980a73f15e4351358e0f16e2fdd99c2517c9cf6270"
+  )
+const TxHashNonExisting* = TxHash.fromHex(
+  "0x0000000000000000000000000000000000000000000000000000000000000000"
+  )
+
+const EthClient = "https://sepolia.infura.io/v3/470c2e9a16f24057aee6660081729fb9"
+
+proc newTestDummyProtocolNode*(
+  switch: Switch, 
+  handler: DummyHandler,
+  ethClient: string
+  ): Future[DummyProtocol] {.async.} =
+  let
+    peerManager = PeerManager.new(switch)
+    dummyProtocol = DummyProtocol.new(peerManager, handler, ethClient)
+
+  await dummyProtocol.start()
+  switch.mount(dummyProtocol)
+
+  return dummyProtocol
+
+
+suite "Waku Incentivization PoC Dummy Protocol":
+
+  var
+    serverSwitch {.threadvar.}: Switch
+    serverRemotePeerInfo {.threadvar.}: RemotePeerInfo
+    handlerFuture {.threadvar.}: Future[DummyRequest]
+    handler {.threadvar.}: DummyHandler
+    server {.threadvar.}: DummyProtocol
+    clientSwitch {.threadvar.}: Switch
+    client {.threadvar.}: WakuDummyClient
+    clientPeerId {.threadvar.}: PeerId
+
+  asyncSetup:
+
+    # setting up a server
+    serverSwitch = newTestSwitch()
+    handlerFuture = newFuture[DummyRequest]()
+    handler = proc(
+        peer: PeerId, dummyRequest: DummyRequest
+    ): Future[DummyResult[void]] {.async.} =
+      handlerFuture.complete(dummyRequest)
+      return ok()
+    server = await newTestDummyProtocolNode(serverSwitch, handler, EthClient)
+
+    # setting up a client
+    clientSwitch = newTestSwitch()
+    let peerManager = PeerManager.new(clientSwitch)
+    client = WakuDummyClient.new(peerManager)
+
+    await allFutures(serverSwitch.start(), clientSwitch.start())
+    
+    clientPeerId = clientSwitch.peerInfo.peerId
+    serverRemotePeerInfo = serverSwitch.peerInfo.toRemotePeerInfo()
+
+  asyncTeardown:
+    await allFutures(clientSwitch.stop(), serverSwitch.stop())
+
+  asyncTest "incentivization PoC: dummy protocol with a valid txid eligibility proof":
+    let request = genDummyRequestWithTxIdEligibilityProof(@(TxHashExisting.bytes()))
+    let response = await client.sendRequest(request, serverRemotePeerInfo)
+    check:
+      response.isOk()
+  
+  asyncTest "incentivization PoC: dummy protocol client with an invalid txid eligibility proof":
+    let request = genDummyRequestWithTxIdEligibilityProof(@(TxHashNonExisting.bytes()))
+    let response = await client.sendRequest(request, serverRemotePeerInfo)
+    check:
+      response.isErr()

tests/incentivization/test_rpc_codec.nim

@@ -1,32 +1,58 @@
 import
   std/options,
-  std/strscans,
   testutils/unittests,
-  chronicles,
   chronos,
-  libp2p/crypto/crypto
+  libp2p/crypto/crypto,
+  web3
 
 import
-  ../../../waku/incentivization/rpc,
-  ../../../waku/incentivization/rpc_codec
+  ../../../waku/incentivization/[
+    rpc,
+    rpc_codec,
+    common,
+    txid_proof,
+    eligibility
+  ]
 
+let txHash = TxHash.fromHex(
+  "0x0000000000000000000000000000000000000000000000000000000000000000"
+  )
+let txHashAsBytes = @(txHash.bytes())
 
 suite "Waku Incentivization Eligibility Codec":
 
     asyncTest "encode eligibility proof":
-      var byteSequence: seq[byte] = @[1, 2, 3, 4, 5, 6, 7, 8]
-      let epRpc = EligibilityProof(proofOfPayment: some(byteSequence))
-      let encoded = encode(epRpc)
+      let txHash = TxHash.fromHex(
+        "0x0000000000000000000000000000000000000000000000000000000000000000")
+      let txHashAsBytes = @(txHash.bytes())
+      let eligibilityProof = EligibilityProof(proofOfPayment: some(txHashAsBytes))
+      let encoded = encode(eligibilityProof)
       let decoded = EligibilityProof.decode(encoded.buffer).get()
       check:
-          epRpc == decoded
+          eligibilityProof == decoded
     
     asyncTest "encode eligibility status":
-      let esRpc = EligibilityStatus(
-        statusCode: uint32(200),
-        statusDesc: some("OK")
-      )
-      let encoded = encode(esRpc)
+      let eligibilityStatus = genEligibilityStatus(true)
+      let encoded = encode(eligibilityStatus)
       let decoded = EligibilityStatus.decode(encoded.buffer).get()
       check:
-        esRpc == decoded
+        eligibilityStatus == decoded
+
+    asyncTest "encode dummy request":
+      let txHash = TxHash.fromHex(
+        "0x0000000000000000000000000000000000000000000000000000000000000000")
+      let txHashAsBytes = @(txHash.bytes())
+      let dummyRequest = genDummyRequestWithTxIdEligibilityProof(txHashAsBytes)
+      let encoded = encode(dummyRequest)
+      let decoded = DummyRequest.decode(encoded.buffer).get()
+      check:
+          dummyRequest == decoded
+
+    asyncTest "encode dummy response":
+      var dummyResponse = genDummyResponseWithEligibilityStatus(true)
+      let encoded = encode(dummyResponse)
+      let decoded = DummyResponse.decode(encoded.buffer).get()
+      check:
+          dummyResponse == decoded
+
+    

waku/incentivization/client.nim

@@ -0,0 +1,55 @@
+import std/options, chronicles, chronos, libp2p/protocols/protocol
+import
+  ../node/peer_manager, ../waku_core, ./common, ./rpc_codec, ./rpc
+
+logScope:
+  topics = "waku incentivization PoC client"
+
+type WakuDummyClient* = ref object
+  peerManager*: PeerManager
+
+proc new*(
+    T: type WakuDummyClient, peerManager: PeerManager): T =
+  WakuDummyClient(peerManager: peerManager)
+
+proc sendDummyRequest(
+    dummyClient: WakuDummyClient, dummyRequest: DummyRequest, peer: PeerId | RemotePeerInfo
+): Future[DummyResult[void]] {.async, gcsafe.} =
+  let connOpt = await dummyClient.peerManager.dialPeer(peer, DummyCodec)
+  if connOpt.isNone():
+    return err("dialFailure")
+  let connection = connOpt.get()
+  await connection.writeLP(dummyRequest.encode().buffer)
+
+  var buffer: seq[byte]
+  try:
+    buffer = await connection.readLp(DefaultMaxRpcSize.int)
+  except LPStreamRemoteClosedError:
+    return err("Exception reading: " & getCurrentExceptionMsg())
+
+  let decodeRespRes = DummyResponse.decode(buffer)
+  if decodeRespRes.isErr():
+    return err("decodeRpcFailure")
+
+  let dummyResponse = decodeRespRes.get()
+
+  let requestId = dummyResponse.requestId
+  let eligibilityStatus = dummyResponse.eligibilityStatus
+  let statusCode = eligibilityStatus.statusCode
+  # status description is optional
+  var statusDesc = ""
+  let statusDescRes = eligibilityStatus.statusDesc
+  if statusDescRes.isSome():
+    statusDesc = statusDescRes.get()
+
+  if statusCode == 200:
+    return ok()
+  else:
+    return err(statusDesc)
+
+proc sendRequest*(
+    dummyClient: WakuDummyClient,
+    dummyRequest: DummyRequest,
+    peer: PeerId | RemotePeerInfo,
+): Future[DummyResult[void]] {.async, gcsafe.} =
+  return await dummyClient.sendDummyRequest(dummyRequest, peer)

waku/incentivization/common.nim

@@ -0,0 +1,53 @@
+import
+  std/options,
+  std/strscans,
+  std/sequtils,
+  testutils/unittests,
+  chronicles,
+  chronos,
+  libp2p/crypto/crypto
+
+import stew/results, libp2p/peerid
+
+import
+  ../../../waku/incentivization/rpc
+
+const DummyCodec* = "/vac/waku/dummy/0.0.1"
+
+type DummyResult*[T] = Result[T, string]
+
+type DummyHandler* = proc(
+  peer: PeerId,
+  dummyRequest: DummyRequest
+): Future[DummyResult[void]] {.async.}
+
+type
+  DummyProtocolErrorKind* {.pure.} = enum
+    UNKNOWN = uint32(000)
+    BAD_RESPONSE = uint32(300)
+    BAD_REQUEST = uint32(400)
+    PAYMENT_REQUIRED = uint(402)  # error type specific for incentivization
+    NOT_FOUND = uint32(404)
+    SERVICE_UNAVAILABLE = uint32(503)
+    PEER_DIAL_FAILURE = uint32(504)
+
+  DummyProtocolError* = object
+    case kind*: DummyProtocolErrorKind
+    of PEER_DIAL_FAILURE:
+      address*: string
+    of BAD_RESPONSE, BAD_REQUEST, NOT_FOUND, SERVICE_UNAVAILABLE, PAYMENT_REQUIRED:
+      cause*: string
+    else:
+      discard
+
+  DummyProtocolResult* = Result[void, DummyProtocolError]
+
+proc genEligibilityStatus*(isEligible: bool): EligibilityStatus = 
+  if isEligible:
+    EligibilityStatus(
+      statusCode: uint32(200),
+      statusDesc: some("OK"))
+  else:
+    EligibilityStatus(
+      statusCode: uint32(402),
+      statusDesc: some("Payment Required"))

waku/incentivization/eligibility.nim

@@ -0,0 +1,26 @@
+import
+  std/options,
+  std/strscans,
+  std/sequtils,
+  testutils/unittests,
+  chronicles,
+  chronos,
+  libp2p/crypto/crypto
+
+import stew/results, libp2p/peerid
+
+import
+  ../../../waku/incentivization/rpc,
+  ../../../waku/incentivization/rpc_codec,
+  ../../../waku/incentivization/common,
+  ../../../waku/incentivization/txid_proof
+
+
+proc isEligible*(eligibilityProof: EligibilityProof, ethClient: string): Future[bool] {.async.} =
+  result = await txidEligiblityCriteriaMet(eligibilityProof, ethClient)
+
+proc genDummyResponseWithEligibilityStatus*(proofValid: bool, requestId: string = ""): DummyResponse = 
+  let eligibilityStatus = genEligibilityStatus(proofValid)
+  result.requestId = requestId
+  result.eligibilityStatus = eligibilityStatus
+

waku/incentivization/protocol.nim

@@ -0,0 +1,60 @@
+import
+  std/[options, sequtils, sets, strutils, tables],
+  stew/byteutils,
+  chronicles,
+  chronos,
+  libp2p/peerid,
+  libp2p/protocols/protocol
+import
+  ../node/peer_manager,
+  ../waku_core,
+  ./common,
+  ./rpc_codec,
+  ./rpc,
+  ./eligibility
+
+logScope:
+  topics = "waku incentivization PoC"
+
+type DummyProtocol* = ref object of LPProtocol
+  peerManager*: PeerManager
+  dummyHandler*: DummyHandler
+  ethClient*: string
+
+proc handleRequest*(
+    dummyProtocol: DummyProtocol, peerId: PeerId, buffer: seq[byte]
+    ): Future[DummyResponse] {.async.} =
+  let reqDecodeRes = DummyRequest.decode(buffer)
+  var isProofValid = false
+  var requestId = ""
+  if reqDecodeRes.isOk():
+    let dummyRequest = reqDecodeRes.get()
+    let eligibilityProof = dummyRequest.eligibilityProof
+    requestId = dummyRequest.requestId
+    isProofValid = await isEligible(eligibilityProof, dummyProtocol.ethClient)
+  let response = genDummyResponseWithEligibilityStatus(isProofValid, requestId)
+  return response
+
+proc initProtocolHandler(dummyProtocol: DummyProtocol) =
+  proc handle(conn: Connection, proto: string) {.async.} =
+    let buffer = await conn.readLp(DefaultMaxRpcSize)
+    var dummyResponse = await handleRequest(dummyProtocol, conn.peerId, buffer)
+    await conn.writeLp(dummyResponse.encode().buffer)
+
+  dummyProtocol.handler = handle
+  dummyProtocol.codec = DummyCodec
+
+proc new*(
+    T: type DummyProtocol,
+    peerManager: PeerManager,
+    dummyHandler: DummyHandler,
+    ethClient: string,
+  ): T =
+  let dummyProtocol = DummyProtocol(
+    peerManager: peerManager,
+    dummyHandler: dummyHandler,
+    ethClient: ethClient
+  )
+  dummyProtocol.initProtocolHandler()
+  return dummyProtocol
+

waku/incentivization/rpc.nim

@@ -15,3 +15,13 @@ type
   EligibilityStatus* = object
     statusCode*: uint32
     statusDesc*: Option[string]
+
+  DummyRequest* = object
+    requestId*: string
+    # request content goes here
+    eligibilityProof*: EligibilityProof
+
+  DummyResponse* = object
+    requestId*: string
+    # response content goes here
+    eligibilityStatus*: EligibilityStatus

waku/incentivization/rpc_codec.nim

@@ -5,6 +5,7 @@ import
   ../waku_core,
   ./rpc
 
+const DefaultMaxRpcSize* = -1
 
 # Codec for EligibilityProof
 
@@ -28,7 +29,6 @@ proc decode*(T: type EligibilityProof, buffer: seq[byte]): ProtobufResult[T] =
     epRpc.proofOfPayment = some(proofOfPayment)
   ok(epRpc)
 
-
 # Codec for EligibilityStatus
 
 proc encode*(esRpc: EligibilityStatus): ProtoBuffer = 
@@ -57,3 +57,47 @@ proc decode*(T: type EligibilityStatus, buffer: seq[byte]): ProtobufResult[T] =
   ok(esRpc)
   
 
+# Codec for DummyRequest
+
+proc encode*(request: DummyRequest): ProtoBuffer = 
+  var pb = initProtoBuffer()
+  pb.write3(1, request.requestId)
+  pb.write3(10, request.eligibilityProof.encode())
+  pb
+
+proc decode*(T: type DummyRequest, buffer: seq[byte]): ProtobufResult[T] = 
+  let pb = initProtoBuffer(buffer)
+  var request = DummyRequest()
+
+  if not ?pb.getField(1,request.requestId):
+    return err(ProtobufError.missingRequiredField("requestId"))
+
+  var eligibilityProofBytes: seq[byte]
+  if not ?pb.getField(10, eligibilityProofBytes):
+    return err(ProtobufError.missingRequiredField("eligibilityProof"))
+  else:
+    request.eligibilityProof = ?EligibilityProof.decode(eligibilityProofBytes)
+  ok(request)
+
+
+# Codec for DummyResponse
+
+proc encode*(response: DummyResponse): ProtoBuffer = 
+  var pb = initProtoBuffer()
+  pb.write3(1, response.requestId)
+  pb.write3(5, response.eligibilityStatus.encode())
+  pb
+
+proc decode*(T: type DummyResponse, buffer: seq[byte]): ProtobufResult[T] = 
+  let pb = initProtoBuffer(buffer)
+  var response = DummyResponse()
+
+  if not ?pb.getField(1,response.requestId):
+    return err(ProtobufError.missingRequiredField("requestId"))
+
+  var eligibilityStatusBytes: seq[byte]
+  if not ?pb.getField(5, eligibilityStatusBytes):
+    return err(ProtobufError.missingRequiredField("eligibilityStatus"))
+  else:
+    response.eligibilityStatus = ?EligibilityStatus.decode(eligibilityStatusBytes)
+  ok(response)

waku/incentivization/txid_proof.nim

@@ -0,0 +1,28 @@
+import
+  std/options,
+  chronos,
+  web3,
+  stew/byteutils
+
+import ../../../waku/incentivization/rpc
+
+proc genDummyRequestWithTxIdEligibilityProof*(txHashAsBytes: seq[byte], requestId: string = ""): DummyRequest =
+  result.requestId = requestId
+  result.eligibilityProof = EligibilityProof(proofOfPayment: some(txHashAsBytes))
+
+proc checkTxIdExists(txHash: TxHash, ethClient: string): Future[bool] {.async.} = 
+  let web3 = await newWeb3(ethClient)
+  try:
+    discard await web3.provider.eth_getTransactionByHash(txHash)
+    result = true
+  except ValueError as e:
+    result = false
+  await web3.close()
+  result
+
+proc txidEligiblityCriteriaMet*(eligibilityProof: EligibilityProof, ethClient: string): Future[bool] {.async.} =
+  if eligibilityProof.proofOfPayment.isNone():
+    return false
+  let txHash = TxHash.fromHex(byteutils.toHex(eligibilityProof.proofOfPayment.get()))
+  let txExists = await checkTxIdExists(txHash, ethClient)
+  return txExists