nwaku/waku/v2/protocol/waku_lightpush/protocol.nim

{.push raises: [Defect].}

import
  std/options,
  stew/results,
  chronicles,
  chronos,
  metrics,
  bearssl/rand
import
  ../waku_message,
  ../waku_relay,
  ../../node/peer_manager/peer_manager,
  ../../utils/requests,
  ./rpc,
  ./rpc_codec,
  ./protocol_metrics


logScope:
  topics = "wakulightpush"


const WakuLightPushCodec* = "/vac/waku/lightpush/2.0.0-beta1"


type
  WakuLightPushResult*[T] = Result[T, string]
  
  PushMessageHandler* = proc(peer: PeerId, pubsubTopic: string, message: WakuMessage): Future[WakuLightPushResult[void]] {.gcsafe, closure.}

  WakuLightPush* = ref object of LPProtocol
    rng*: ref rand.HmacDrbgContext
    peerManager*: PeerManager
    pushHandler*: PushMessageHandler

proc initProtocolHandler*(wl: WakuLightPush) =
  proc handle(conn: Connection, proto: string) {.async, gcsafe, closure.} =
    let buffer = await conn.readLp(MaxRpcSize.int)
    let reqDecodeRes = PushRPC.init(buffer)
    if reqDecodeRes.isErr():
      error "failed to decode rpc"
      waku_lightpush_errors.inc(labelValues = [decodeRpcFailure])
      return

    let req = reqDecodeRes.get()
    if req.request == PushRequest():
      error "invalid lightpush rpc received", error=emptyRequestBodyFailure
      waku_lightpush_errors.inc(labelValues = [emptyRequestBodyFailure])
      return

    waku_lightpush_messages.inc(labelValues = ["PushRequest"])
    let
      pubSubTopic = req.request.pubSubTopic
      message = req.request.message
    debug "push request", peerId=conn.peerId, requestId=req.requestId, pubsubTopic=pubsubTopic

    var response: PushResponse
    let handleRes = await wl.pushHandler(conn.peerId, pubsubTopic, message)
    if handleRes.isOk():
      response = PushResponse(is_success: true, info: "OK")
    else:
      response = PushResponse(is_success: false, info: handleRes.error)
      waku_lightpush_errors.inc(labelValues = [messagePushFailure])
      error "pushed message handling failed", error=handleRes.error

    let rpc = PushRPC(requestId: req.requestId, response: response)
    await conn.writeLp(rpc.encode().buffer)

  wl.handler = handle
  wl.codec = WakuLightPushCodec

proc new*(T: type WakuLightPush, 
          peerManager: PeerManager, 
          rng: ref rand.HmacDrbgContext,
          pushHandler: PushMessageHandler): T = 
  let wl = WakuLightPush(rng: rng, peerManager: peerManager, pushHandler: pushHandler)
  wl.initProtocolHandler()
  return wl


proc setPeer*(wlp: WakuLightPush, peer: RemotePeerInfo) {.
  deprecated: "Use 'WakuLightPushClient.setPeer()' instead" .} = 
  wlp.peerManager.addPeer(peer, WakuLightPushCodec)
  waku_lightpush_peers.inc()

proc request(wl: WakuLightPush, req: PushRequest, peer: RemotePeerInfo): Future[WakuLightPushResult[PushResponse]] {.async, gcsafe,
  deprecated: "Use 'WakuLightPushClient.request()' instead" .} = 
  let connOpt = await wl.peerManager.dialPeer(peer, WakuLightPushCodec)
  if connOpt.isNone():
    waku_lightpush_errors.inc(labelValues = [dialFailure])
    return err(dialFailure)

  let connection = connOpt.get()

  let rpc = PushRPC(requestId: generateRequestId(wl.rng), request: req)
  await connection.writeLP(rpc.encode().buffer)

  var message = await connection.readLp(MaxRpcSize.int)
  let res = PushRPC.init(message)

  if res.isErr():
    waku_lightpush_errors.inc(labelValues = [decodeRpcFailure])
    return err(decodeRpcFailure)

  let rpcRes = res.get()
  if rpcRes.response == PushResponse():
    return err("empty response body")

  return ok(rpcRes.response)

proc request*(wl: WakuLightPush, req: PushRequest): Future[WakuLightPushResult[PushResponse]] {.async, gcsafe,
  deprecated: "Use 'WakuLightPushClient.request()' instead" .} = 
  let peerOpt = wl.peerManager.selectPeer(WakuLightPushCodec)
  if peerOpt.isNone():
    waku_lightpush_errors.inc(labelValues = [dialFailure])
    return err(dialFailure)

  return await wl.request(req, peerOpt.get())