nim-codex/dagger/blockexchange/network.nim

## Nim-Dagger
## Copyright (c) 2021 Status Research & Development GmbH
## Licensed under either of
##  * Apache License, version 2.0, ([LICENSE-APACHE](LICENSE-APACHE))
##  * MIT license ([LICENSE-MIT](LICENSE-MIT))
## at your option.
## This file may not be copied, modified, or distributed except according to
## those terms.

import std/tables
import std/sequtils

import pkg/chronicles
import pkg/chronos

import pkg/libp2p
import pkg/questionable
import pkg/questionable/results

import ../blocktype as bt
import ./protobuf/blockexc as pb
import ./protobuf/payments

import ./networkpeer

export networkpeer, payments

logScope:
  topics = "dagger blockexc network"

const Codec* = "/dagger/blockexc/1.0.0"

type
  WantListHandler* = proc(peer: PeerID, wantList: WantList): Future[void] {.gcsafe.}
  BlocksHandler* = proc(peer: PeerID, blocks: seq[bt.Block]): Future[void] {.gcsafe.}
  BlockPresenceHandler* = proc(peer: PeerID, precense: seq[BlockPresence]): Future[void] {.gcsafe.}
  AccountHandler* = proc(peer: PeerID, account: Account): Future[void] {.gcsafe.}
  PaymentHandler* = proc(peer: PeerID, payment: SignedState): Future[void] {.gcsafe.}

  BlockExcHandlers* = object
    onWantList*: WantListHandler
    onBlocks*: BlocksHandler
    onPresence*: BlockPresenceHandler
    onAccount*: AccountHandler
    onPayment*: PaymentHandler

  WantListBroadcaster* = proc(
    id: PeerID,
    cids: seq[Cid],
    priority: int32 = 0,
    cancel: bool = false,
    wantType: WantType = WantType.wantHave,
    full: bool = false,
    sendDontHave: bool = false) {.gcsafe.}

  BlocksBroadcaster* = proc(peer: PeerID, presence: seq[bt.Block]) {.gcsafe.}
  PresenceBroadcaster* = proc(peer: PeerID, presence: seq[BlockPresence]) {.gcsafe.}
  AccountBroadcaster* = proc(peer: PeerID, account: Account) {.gcsafe.}
  PaymentBroadcaster* = proc(peer: PeerID, payment: SignedState) {.gcsafe.}

  BlockExcRequest* = object
    sendWantList*: WantListBroadcaster
    sendBlocks*: BlocksBroadcaster
    sendPresence*: PresenceBroadcaster
    sendAccount*: AccountBroadcaster
    sendPayment*: PaymentBroadcaster

  BlockExcNetwork* = ref object of LPProtocol
    peers*: Table[PeerID, NetworkPeer]
    switch*: Switch
    handlers*: BlockExcHandlers
    request*: BlockExcRequest
    getConn: ConnProvider

proc handleWantList(
  b: BlockExcNetwork,
  peer: NetworkPeer,
  list: WantList): Future[void] =
  ## Handle incoming want list
  ##

  if isNil(b.handlers.onWantList):
    return

  trace "Handling want list for peer", peer = peer.id
  b.handlers.onWantList(peer.id, list)

# TODO: make into a template
proc makeWantList*(
  cids: seq[Cid],
  priority: int = 0,
  cancel: bool = false,
  wantType: WantType = WantType.wantHave,
  full: bool = false,
  sendDontHave: bool = false): WantList =
  var entries: seq[Entry]
  for cid in cids:
    entries.add(Entry(
      `block`: cid.data.buffer,
      priority: priority.int32,
      cancel: cancel,
      wantType: wantType,
      sendDontHave: sendDontHave))

  WantList(entries: entries, full: full)

proc broadcastWantList*(
  b: BlockExcNetwork,
  id: PeerID,
  cids: seq[Cid],
  priority: int32 = 0,
  cancel: bool = false,
  wantType: WantType = WantType.wantHave,
  full: bool = false,
  sendDontHave: bool = false) =
  ## send a want message to peer
  ##

  if id notin b.peers:
    return

  trace "Sending want list to peer", peer = id, `type` = $wantType, len = cids.len

  let
    wantList = makeWantList(
      cids,
      priority,
      cancel,
      wantType,
      full,
      sendDontHave)
  b.peers.withValue(id, peer):
    peer[].broadcast(Message(wantlist: wantList))

proc handleBlocks(
  b: BlockExcNetwork,
  peer: NetworkPeer,
  blocks: seq[pb.Block]): Future[void] =
  ## Handle incoming blocks
  ##

  if isNil(b.handlers.onBlocks):
    return

  trace "Handling blocks for peer", peer = peer.id

  var blks: seq[bt.Block]
  for blob in blocks:
    without cid =? Cid.init(blob.prefix):
      trace "Unable to initialize Cid from protobuf message"

    without blk =? bt.Block.new(cid, blob.data, verify = true):
      trace "Unable to initialize Block from data"

    blks.add(blk)

  b.handlers.onBlocks(peer.id, blks)

template makeBlocks*(blocks: seq[bt.Block]): seq[pb.Block] =
  var blks: seq[pb.Block]
  for blk in blocks:
    blks.add(pb.Block(
      prefix: blk.cid.data.buffer,
      data: blk.data
    ))

  blks

proc broadcastBlocks*(
  b: BlockExcNetwork,
  id: PeerID,
  blocks: seq[bt.Block]) =
  ## Send blocks to remote
  ##

  if id notin b.peers:
    trace "Unable to send blocks, peer disconnected", peer = id
    return

  b.peers.withValue(id, peer):
    trace "Sending blocks to peer", peer = id, len = blocks.len
    peer[].broadcast(pb.Message(payload: makeBlocks(blocks)))

proc handleBlockPresence(
  b: BlockExcNetwork,
  peer: NetworkPeer,
  presence: seq[BlockPresence]): Future[void] =
  ## Handle block presence
  ##

  if isNil(b.handlers.onPresence):
    return

  trace "Handling block presence for peer", peer = peer.id
  b.handlers.onPresence(peer.id, presence)

proc broadcastBlockPresence*(
  b: BlockExcNetwork,
  id: PeerID,
  presence: seq[BlockPresence]) =
  ## Send presence to remote
  ##

  if id notin b.peers:
    return

  trace "Sending presence to peer", peer = id
  b.peers.withValue(id, peer):
    peer[].broadcast(Message(blockPresences: @presence))

proc handleAccount(network: BlockExcNetwork,
                   peer: NetworkPeer,
                   account: Account): Future[void] =
  if network.handlers.onAccount.isNil:
    return
  network.handlers.onAccount(peer.id, account)

proc broadcastAccount*(network: BlockExcNetwork,
                       id: PeerId,
                       account: Account) =
  if id notin network.peers:
    return

  let message = Message(account: AccountMessage.init(account))
  network.peers.withValue(id, peer):
    peer[].broadcast(message)

proc broadcastPayment*(network: BlockExcNetwork,
                       id: PeerId,
                       payment: SignedState) =
  if id notin network.peers:
    return

  let message = Message(payment: StateChannelUpdate.init(payment))
  network.peers.withValue(id, peer):
    peer[].broadcast(message)

proc handlePayment(network: BlockExcNetwork,
                   peer: NetworkPeer,
                   payment: SignedState): Future[void] =
  if network.handlers.onPayment.isNil:
    return
  network.handlers.onPayment(peer.id, payment)

proc rpcHandler(b: BlockExcNetwork, peer: NetworkPeer, msg: Message) {.async.} =
  try:
    if msg.wantlist.entries.len > 0:
      asyncSpawn b.handleWantList(peer, msg.wantlist)

    if msg.payload.len > 0:
      asyncSpawn b.handleBlocks(peer, msg.payload)

    if msg.blockPresences.len > 0:
      asyncSpawn b.handleBlockPresence(peer, msg.blockPresences)

    if account =? Account.init(msg.account):
      asyncSpawn b.handleAccount(peer, account)

    if payment =? SignedState.init(msg.payment):
      asyncSpawn b.handlePayment(peer, payment)

  except CatchableError as exc:
    trace "Exception in blockexc rpc handler", exc = exc.msg

proc getOrCreatePeer(b: BlockExcNetwork, peer: PeerID): NetworkPeer =
  ## Creates or retrieves a BlockExcNetwork Peer
  ##

  if peer in b.peers:
    return b.peers.getOrDefault(peer, nil)

  var getConn = proc(): Future[Connection] {.async.} =
    try:
      return await b.switch.dial(peer, Codec)
    except CatchableError as exc:
      trace "Unable to connect to blockexc peer", exc = exc.msg

  if not isNil(b.getConn):
    getConn = b.getConn

  let rpcHandler = proc (p: NetworkPeer, msg: Message): Future[void] =
    b.rpcHandler(p, msg)

  # create new pubsub peer
  let blockExcPeer = NetworkPeer.new(peer, getConn, rpcHandler)
  debug "Created new blockexc peer", peer

  b.peers[peer] = blockExcPeer

  return blockExcPeer

proc setupPeer*(b: BlockExcNetwork, peer: PeerID) =
  ## Perform initial setup, such as want
  ## list exchange
  ##

  discard b.getOrCreatePeer(peer)

proc dialPeer*(b: BlockExcNetwork, peer: PeerRecord) {.async.} =
  try:
    await b.switch.connect(peer.peerId, peer.addresses.mapIt(it.address))
  except CatchableError as exc:
    debug "Failed to connect to peer", error = exc.msg, peer

proc dropPeer*(b: BlockExcNetwork, peer: PeerID) =
  ## Cleanup disconnected peer
  ##

  b.peers.del(peer)

method init*(b: BlockExcNetwork) =
  ## Perform protocol initialization
  ##

  proc peerEventHandler(peerId: PeerID, event: PeerEvent) {.async.} =
    if event.kind == PeerEventKind.Joined:
      b.setupPeer(peerId)
    else:
      b.dropPeer(peerId)

  b.switch.addPeerEventHandler(peerEventHandler, PeerEventKind.Joined)
  b.switch.addPeerEventHandler(peerEventHandler, PeerEventKind.Left)

  proc handle(conn: Connection, proto: string) {.async, gcsafe, closure.} =
    let peerId = conn.peerId
    let blockexcPeer = b.getOrCreatePeer(peerId)
    await blockexcPeer.readLoop(conn)  # attach read loop

  b.handler = handle
  b.codec = Codec

proc new*(
  T: type BlockExcNetwork,
  switch: Switch,
  connProvider: ConnProvider = nil): T =
  ## Create a new BlockExcNetwork instance
  ##

  let b = BlockExcNetwork(
    switch: switch,
    getConn: connProvider)

  proc sendWantList(
    id: PeerID,
    cids: seq[Cid],
    priority: int32 = 0,
    cancel: bool = false,
    wantType: WantType = WantType.wantHave,
    full: bool = false,
    sendDontHave: bool = false) {.gcsafe.} =
    b.broadcastWantList(
      id, cids, priority, cancel,
      wantType, full, sendDontHave)

  proc sendBlocks(id: PeerID, blocks: seq[bt.Block]) {.gcsafe.} =
    b.broadcastBlocks(id, blocks)

  proc sendPresence(id: PeerID, presence: seq[BlockPresence]) {.gcsafe.} =
    b.broadcastBlockPresence(id, presence)

  proc sendAccount(id: PeerID, account: Account) =
    b.broadcastAccount(id, account)

  proc sendPayment(id: PeerID, payment: SignedState) =
    b.broadcastPayment(id, payment)

  b.request = BlockExcRequest(
    sendWantList: sendWantList,
    sendBlocks: sendBlocks,
    sendPresence: sendPresence,
    sendAccount: sendAccount,
    sendPayment: sendPayment)

  b.init()
  return b