nwaku/apps/wakunode2/wakunode2.nim

{.push raises: [Defect].}

import
  std/[options, tables, strutils, sequtils, os],
  stew/shims/net as stewNet,
  chronicles, 
  chronos,
  metrics,
  confutils, 
  toml_serialization,
  system/ansi_c,
  eth/keys,
  eth/p2p/discoveryv5/enr,
  libp2p/[builders, multihash],
  libp2p/crypto/crypto,
  libp2p/protocols/ping,
  libp2p/protocols/pubsub/[gossipsub, rpc/messages],
  libp2p/transports/[transport, wstransport],
  libp2p/nameresolving/dnsresolver
import
  ../../waku/v2/protocol/waku_store,
  ../../waku/v2/protocol/waku_filter,
  ../../waku/v2/protocol/waku_peer_exchange,
  ../../waku/v2/node/peer_manager/peer_manager,
  ../../waku/v2/node/dnsdisc/waku_dnsdisc,
  ../../waku/v2/node/discv5/waku_discv5,
  ../../waku/v2/node/storage/sqlite,
  ../../waku/v2/node/storage/peer/waku_peer_storage,
  ../../waku/v2/node/storage/message/waku_store_queue,
  ../../waku/v2/node/storage/message/dual_message_store,
  ../../waku/v2/node/storage/message/sqlite_store,
  ../../waku/v2/node/storage/message/message_retention_policy_capacity,
  ../../waku/v2/node/storage/message/message_retention_policy_time,
  ../../waku/v2/node/[wakuswitch, waku_node, waku_metrics],
  ../../waku/v2/utils/[peers, wakuenr],
  ../../waku/common/utils/nat,
  ./wakunode2_setup_rest,
  ./wakunode2_setup_rpc,
  ./wakunode2_setup_sql_migrations,
  ./config

when defined(rln) or defined(rlnzerokit):
  import
    ../../waku/v2/protocol/waku_rln_relay/waku_rln_relay_types,
    ../../waku/v2/protocol/waku_rln_relay/waku_rln_relay_utils


logScope:
  topics = "wakunode.setup"


type SetupResult[T] = Result[T, string]


proc setupStorage(conf: WakuNodeConf):
  SetupResult[tuple[pStorage: WakuPeerStorage, mStorage: MessageStore]] =

  ## Setup a SQLite Database for a wakunode based on a supplied
  ## configuration file and perform all necessary migration.
  ## 
  ## If config allows, return peer storage and message store
  ## for use elsewhere.
  
  var
    sqliteDatabase: SqliteDatabase
    storeTuple: tuple[pStorage: WakuPeerStorage, mStorage: MessageStore]

  # Setup database connection
  if conf.dbPath != "":
    let dbRes = SqliteDatabase.init(conf.dbPath)
    if dbRes.isErr():
      warn "failed to init database connection", err = dbRes.error
      waku_node_errors.inc(labelValues = ["init_db_failure"])
      return err("failed to init database connection")
    else:
      sqliteDatabase = dbRes.value


  if not sqliteDatabase.isNil():
    ## Database vacuuming
    # TODO: Wrap and move this logic to the appropriate module
    let 
      pageSize = ?sqliteDatabase.getPageSize()
      pageCount = ?sqliteDatabase.getPageCount()
      freelistCount = ?sqliteDatabase.getFreelistCount()

    debug "sqlite database page stats", pageSize=pageSize, pages=pageCount, freePages=freelistCount

    # TODO: Run vacuuming conditionally based on database page stats
    if conf.dbVacuum and (pageCount > 0 and freelistCount > 0):
      debug "starting sqlite database vacuuming"

      let resVacuum = sqliteDatabase.vacuum()
      if resVacuum.isErr():
        return err("failed to execute vacuum: " & resVacuum.error())

      debug "finished sqlite database vacuuming"

    sqliteDatabase.runMigrations(conf) 


  if conf.persistPeers:
    let res = WakuPeerStorage.new(sqliteDatabase)
    if res.isErr():
      warn "failed to init peer store", err = res.error
      waku_node_errors.inc(labelValues = ["init_store_failure"])
    else:
      storeTuple.pStorage = res.value

  if conf.persistMessages:
    if conf.sqliteStore:
      debug "setting up sqlite-only store"
      let res = SqliteStore.init(sqliteDatabase)
      if res.isErr():
        warn "failed to init message store", err = res.error
        waku_node_errors.inc(labelValues = ["init_store_failure"])
      else:
        storeTuple.mStorage = res.value
    elif not sqliteDatabase.isNil():
      debug "setting up dual message store"
      let res = DualMessageStore.init(sqliteDatabase, conf.storeCapacity)
      if res.isErr():
        warn "failed to init message store", err = res.error
        waku_node_errors.inc(labelValues = ["init_store_failure"])
      else:
        storeTuple.mStorage = res.value
    else:
      debug "setting up in-memory store"
      storeTuple.mStorage = StoreQueueRef.new(conf.storeCapacity)

  ok(storeTuple)

proc retrieveDynamicBootstrapNodes(conf: WakuNodeConf): SetupResult[seq[RemotePeerInfo]] =
  
  if conf.dnsDiscovery and conf.dnsDiscoveryUrl != "":
    # DNS discovery
    debug "Discovering nodes using Waku DNS discovery", url=conf.dnsDiscoveryUrl

    var nameServers: seq[TransportAddress]
    for ip in conf.dnsDiscoveryNameServers:
      nameServers.add(initTAddress(ip, Port(53))) # Assume all servers use port 53

    let dnsResolver = DnsResolver.new(nameServers)

    proc resolver(domain: string): Future[string] {.async, gcsafe.} =
      trace "resolving", domain=domain
      let resolved = await dnsResolver.resolveTxt(domain)
      return resolved[0] # Use only first answer

    var wakuDnsDiscovery = WakuDnsDiscovery.init(conf.dnsDiscoveryUrl,
                                                  resolver)
    if wakuDnsDiscovery.isOk:
      return wakuDnsDiscovery.get().findPeers()
        .mapErr(proc (e: cstring): string = $e)
    else:
      warn "Failed to init Waku DNS discovery"

  debug "No method for retrieving dynamic bootstrap nodes specified."
  ok(newSeq[RemotePeerInfo]()) # Return an empty seq by default

proc initNode(conf: WakuNodeConf,
              pStorage: WakuPeerStorage = nil,
              dynamicBootstrapNodes: openArray[RemotePeerInfo] = @[]): SetupResult[WakuNode] =
  
  ## Setup a basic Waku v2 node based on a supplied configuration
  ## file. Optionally include persistent peer storage.
  ## No protocols are mounted yet.

  var dnsResolver: DnsResolver
  if conf.dnsAddrs:
    # Support for DNS multiaddrs
    var nameServers: seq[TransportAddress]
    for ip in conf.dnsAddrsNameServers:
      nameServers.add(initTAddress(ip, Port(53))) # Assume all servers use port 53
    
    dnsResolver = DnsResolver.new(nameServers)
  
  let 
    ## `udpPort` is only supplied to satisfy underlying APIs but is not
    ## actually a supported transport for libp2p traffic.
    udpPort = conf.tcpPort
    (extIp, extTcpPort, extUdpPort) = setupNat(conf.nat,
                                              clientId,
                                              Port(uint16(conf.tcpPort) + conf.portsShift),
                                              Port(uint16(udpPort) + conf.portsShift))

    dns4DomainName = if conf.dns4DomainName != "": some(conf.dns4DomainName)
                      else: none(string)
    
    discv5UdpPort = if conf.discv5Discovery: some(Port(uint16(conf.discv5UdpPort) + conf.portsShift))
                    else: none(Port)

    ## @TODO: the NAT setup assumes a manual port mapping configuration if extIp config is set. This probably
    ## implies adding manual config item for extPort as well. The following heuristic assumes that, in absence of manual
    ## config, the external port is the same as the bind port.
    extPort = if (extIp.isSome() or dns4DomainName.isSome()) and extTcpPort.isNone():
                some(Port(uint16(conf.tcpPort) + conf.portsShift))
              else:
                extTcpPort
    
    wakuFlags = initWakuFlags(conf.lightpush,
                              conf.filter,
                              conf.store,
                              conf.relay)

  var node: WakuNode
  try:
    node = WakuNode.new(conf.nodekey,
                        conf.listenAddress, Port(uint16(conf.tcpPort) + conf.portsShift), 
                        extIp, extPort,
                        pStorage,
                        conf.maxConnections.int,
                        Port(uint16(conf.websocketPort) + conf.portsShift),
                        conf.websocketSupport,
                        conf.websocketSecureSupport,
                        conf.websocketSecureKeyPath,
                        conf.websocketSecureCertPath,
                        some(wakuFlags),
                        dnsResolver,
                        conf.relayPeerExchange, # We send our own signed peer record when peer exchange enabled
                        dns4DomainName,
                        discv5UdpPort
                        )
  except:
    return err("failed to create waku node instance: " & getCurrentExceptionMsg())
  
  if conf.discv5Discovery:
    let
      discoveryConfig = DiscoveryConfig.init(
        conf.discv5TableIpLimit, conf.discv5BucketIpLimit, conf.discv5BitsPerHop)

    # select dynamic bootstrap nodes that have an ENR containing a udp port.
    # Discv5 only supports UDP https://github.com/ethereum/devp2p/blob/master/discv5/discv5-theory.md)
    var discv5BootstrapEnrs: seq[enr.Record]
    for n in dynamicBootstrapNodes:
      if n.enr.isSome():
        let
          enr = n.enr.get()
          tenrRes = enr.toTypedRecord()
        if tenrRes.isOk() and (tenrRes.get().udp.isSome() or tenrRes.get().udp6.isSome()):
          discv5BootstrapEnrs.add(enr)
  
    # parse enrURIs from the configuration and add the resulting ENRs to the discv5BootstrapEnrs seq
    for enrUri in conf.discv5BootstrapNodes:
      addBootstrapNode(enrUri, discv5BootstrapEnrs)

    node.wakuDiscv5 = WakuDiscoveryV5.new(
      extIP, extPort, discv5UdpPort,
      conf.listenAddress,
      discv5UdpPort.get(),
      discv5BootstrapEnrs,
      conf.discv5EnrAutoUpdate,
      keys.PrivateKey(conf.nodekey.skkey),
      wakuFlags,
      [], # Empty enr fields, for now
      node.rng,
      discoveryConfig
    )
  
  ok(node)

proc setupProtocols(node: WakuNode, conf: WakuNodeConf, mStorage: MessageStore): Future[SetupResult[void]] {.async.} =
  ## Setup configured protocols on an existing Waku v2 node.
  ## Optionally include persistent message storage.
  ## No protocols are started yet.
  
  # Mount relay on all nodes
  var peerExchangeHandler = none(RoutingRecordsHandler)
  if conf.relayPeerExchange:
    proc handlePeerExchange(peer: PeerId, topic: string,
                            peers: seq[RoutingRecordsPair]) {.gcsafe, raises: [Defect].} =
      ## Handle peers received via gossipsub peer exchange
      # TODO: Only consider peers on pubsub topics we subscribe to
      let exchangedPeers = peers.filterIt(it.record.isSome()) # only peers with populated records
                                .mapIt(toRemotePeerInfo(it.record.get()))
      
      debug "connecting to exchanged peers", src=peer, topic=topic, numPeers=exchangedPeers.len

      # asyncSpawn, as we don't want to block here
      asyncSpawn node.connectToNodes(exchangedPeers, "peer exchange")
  
    peerExchangeHandler = some(handlePeerExchange)

  if conf.relay:
    try:
      let pubsubTopics = conf.topics.split(" ")
      await mountRelay(node, pubsubTopics, peerExchangeHandler = peerExchangeHandler) 
    except:
      return err("failed to mount waku relay protocol: " & getCurrentExceptionMsg())

  
  # Keepalive mounted on all nodes
  try:
    await mountLibp2pPing(node)
  except:
    return err("failed to mount libp2p ping protocol: " & getCurrentExceptionMsg())
  
  when defined(rln) or defined(rlnzerokit): 
    if conf.rlnRelay:
      try: 
        let res = node.mountRlnRelay(conf)
        if res.isErr():
          return err("failed to mount waku RLN relay protocol: " & res.error)
      except:
        return err("failed to mount waku RLN relay protocol: " & getCurrentExceptionMsg())
  
  if conf.swap:
    try:
      await mountSwap(node)
      # TODO: Set swap peer, for now should be same as store peer
    except:
      return err("failed to mount waku swap protocol: " & getCurrentExceptionMsg())

  # Store setup
  if (conf.storenode != "") or (conf.store):
    let retentionPolicy = if conf.sqliteStore: TimeRetentionPolicy.init(conf.sqliteRetentionTime)
                          else: CapacityRetentionPolicy.init(conf.storeCapacity)

    try:
      await mountStore(node, mStorage, retentionPolicy=some(retentionPolicy))
    except:
      return err("failed to mount waku store protocol: " & getCurrentExceptionMsg())

    executeMessageRetentionPolicy(node)
    startMessageRetentionPolicyPeriodicTask(node, interval=MessageStoreDefaultRetentionPolicyInterval)

    if conf.storenode != "":
      try:
        setStorePeer(node, conf.storenode)
      except:
        return err("failed to set node waku store peer: " & getCurrentExceptionMsg())

  # NOTE Must be mounted after relay
  if (conf.lightpushnode != "") or (conf.lightpush):
    try:
      await mountLightPush(node)
    except:
      return err("failed to mount waku lightpush protocol: " & getCurrentExceptionMsg())

    if conf.lightpushnode != "":
      try:
        setLightPushPeer(node, conf.lightpushnode)
      except:
        return err("failed to set node waku lightpush peer: " & getCurrentExceptionMsg())
  
  # Filter setup. NOTE Must be mounted after relay
  if (conf.filternode != "") or (conf.filter):
    try:
      await mountFilter(node, filterTimeout = chronos.seconds(conf.filterTimeout))
    except:
      return err("failed to mount waku filter protocol: " & getCurrentExceptionMsg())

    if conf.filternode != "":
      try:
        setFilterPeer(node, conf.filternode)
      except:
        return err("failed to set node waku filter peer: " & getCurrentExceptionMsg())
  
  # waku peer exchange setup
  if (conf.peerExchangeNode != "") or (conf.peerExchange):
    try:
      await mountWakuPeerExchange(node)
    except:
      return err("failed to mount waku peer-exchange protocol: " & getCurrentExceptionMsg())

    if conf.peerExchangeNode != "":
      try:
        setPeerExchangePeer(node, conf.peerExchangeNode)
      except:
        return err("failed to set node waku lightpush peer: " & getCurrentExceptionMsg())

  return ok()

proc startNode(node: WakuNode, conf: WakuNodeConf,
  dynamicBootstrapNodes: seq[RemotePeerInfo] = @[]): Future[SetupResult[void]] {.async.} =
  ## Start a configured node and all mounted protocols.
  ## Resume history, connect to static nodes and start
  ## keep-alive, if configured.
  
  # Start Waku v2 node
  try:
    await node.start()
  except:
    return err("failed to start waku node: " & getCurrentExceptionMsg())

  # Start discv5 and connect to discovered nodes
  if conf.discv5Discovery:
    try:
      if not await node.startDiscv5():
        error "could not start Discovery v5"
    except:
      return err("failed to start waku discovery v5: " & getCurrentExceptionMsg())


  # Resume historical messages, this has to be called after the node has been started
  if conf.store and conf.persistMessages:
    try:
      await node.resume()
    except:
      return err("failed to resume messages history: " & getCurrentExceptionMsg())
  
  # Connect to configured static nodes
  if conf.staticnodes.len > 0:
    try:
      await connectToNodes(node, conf.staticnodes, "static")
    except:
      return err("failed to connect to static nodes: " & getCurrentExceptionMsg())
  
  if dynamicBootstrapNodes.len > 0:
    info "Connecting to dynamic bootstrap peers"
    try:
      await connectToNodes(node, dynamicBootstrapNodes, "dynamic bootstrap")
    except:
      return err("failed to connect to dynamic bootstrap nodes: " & getCurrentExceptionMsg())
    

  # retrieve and connect to peer exchange peers
  if conf.peerExchangeNode != "":
    info "Retrieving peer info via peer exchange protocol"
    let desiredOutDegree = node.wakuRelay.parameters.d.uint64()
    try:
      discard await node.wakuPeerExchange.request(desiredOutDegree)
    except:
      return err("failed to retrieve peer info via peer exchange protocol: " & getCurrentExceptionMsg())

  # Start keepalive, if enabled
  if conf.keepAlive:
    node.startKeepalive()
  
  return ok()

proc startExternal(node: WakuNode, conf: WakuNodeConf): SetupResult[void] =
  ## Start configured external interfaces and monitoring tools
  ## on a Waku v2 node, including the RPC API, REST API and metrics
  ## monitoring ports.
  
  if conf.rpc:
    try:
      startRpcServer(node, conf.rpcAddress, Port(conf.rpcPort + conf.portsShift), conf)
    except:
      return err("failed to start the json-rpc server: " & getCurrentExceptionMsg())
  
  if conf.rest:
    startRestServer(node, conf.restAddress, Port(conf.restPort + conf.portsShift), conf)

  if conf.metricsLogging:
    startMetricsLog()

  if conf.metricsServer:
    startMetricsServer(conf.metricsServerAddress,
      Port(conf.metricsServerPort + conf.portsShift))
  
  ok()


{.pop.} # @TODO confutils.nim(775, 17) Error: can raise an unlisted exception: ref IOError
when isMainModule:
  ## Node setup happens in 6 phases:
  ## 1. Set up storage
  ## 2. Initialize node
  ## 3. Mount and initialize configured protocols
  ## 4. Start node and mounted protocols
  ## 5. Start monitoring tools and external interfaces
  ## 6. Setup graceful shutdown hooks
  
  {.push warning[ProveInit]: off.}
  let conf = try:
    WakuNodeConf.load(
      secondarySources = proc (conf: WakuNodeConf, sources: auto) =
        if conf.configFile.isSome:
          sources.addConfigFile(Toml, conf.configFile.get)
    )
  except CatchableError:
    error "Failure while loading the configuration: \n", error=getCurrentExceptionMsg()
    quit 1 # if we don't leave here, the initialization of conf does not work in the success case
  {.pop.}

  # if called with --version, print the version and quit
  if conf.version:
    echo "version / git commit hash: ", git_version
    quit(QuitSuccess)

  # set log level
  if conf.logLevel != LogLevel.NONE:
    setLogLevel(conf.logLevel)
  
  var
    node: WakuNode  # This is the node we're going to setup using the conf

  ##############
  # Node setup #
  ##############
  
  debug "1/7 Setting up storage"
  
  var
    pStorage: WakuPeerStorage
    mStorage: MessageStore
  
  let setupStorageRes = setupStorage(conf)
  if setupStorageRes.isErr():
    error "1/7 Setting up storage failed. Continuing without storage.", error=setupStorageRes.error
  else:
    (pStorage, mStorage) = setupStorageRes.get()

  debug "2/7 Retrieve dynamic bootstrap nodes"
  
  var dynamicBootstrapNodes: seq[RemotePeerInfo]
  let dynamicBootstrapNodesRes = retrieveDynamicBootstrapNodes(conf)
  if dynamicBootstrapNodesRes.isErr():
    warn "2/7 Retrieving dynamic bootstrap nodes failed. Continuing without dynamic bootstrap nodes.", error=dynamicBootstrapNodesRes.error
  else:
    dynamicBootstrapNodes = dynamicBootstrapNodesRes.get()

  debug "3/7 Initializing node"

  let initNodeRes = initNode(conf, pStorage, dynamicBootstrapNodes)
  if initNodeRes.isErr():
    error "3/7 Initializing node failed. Quitting.", error=initNodeRes.error
    quit(QuitFailure)
  else:
    node = initNodeRes.get()

  debug "4/7 Mounting protocols"

  let setupProtocolsRes = waitFor setupProtocols(node, conf, mStorage)
  if setupProtocolsRes.isErr():
    error "4/7 Mounting protocols failed. Continuing in current state.", error=setupProtocolsRes.error

  debug "5/7 Starting node and mounted protocols"
  
  let startNodeRes = waitFor startNode(node, conf, dynamicBootstrapNodes)
  if startNodeRes.isErr():
    error "5/7 Starting node and mounted protocols failed. Continuing in current state.", error=startNodeRes.error

  debug "6/7 Starting monitoring and external interfaces"

  let startExternalRes = startExternal(node, conf)
  if startExternalRes.isErr():
    error "6/7 Starting monitoring and external interfaces failed. Continuing in current state.", error=startExternalRes.error

  debug "7/7 Setting up shutdown hooks"

  # 7/7 Setup graceful shutdown hooks
  ## Setup shutdown hooks for this process.
  ## Stop node gracefully on shutdown.
  
  # Handle Ctrl-C SIGINT
  proc handleCtrlC() {.noconv.} =
    when defined(windows):
      # workaround for https://github.com/nim-lang/Nim/issues/4057
      setupForeignThreadGc()
    info "Shutting down after receiving SIGINT"
    waitFor node.stop()
    quit(QuitSuccess)
  
  setControlCHook(handleCtrlC)

  # Handle SIGTERM
  when defined(posix):
    proc handleSigterm(signal: cint) {.noconv.} =
      info "Shutting down after receiving SIGTERM"
      waitFor node.stop()
      quit(QuitSuccess)
    
    c_signal(ansi_c.SIGTERM, handleSigterm)
  
  info "Node setup complete"

  runForever()