Refactor wakunode2 setup (#675)

* Refactor wakunode2 setup

Author: jm-clius

CHANGELOG.md

@@ -15,6 +15,7 @@ This release contains the following:
   The filed numbers of the `HistoryResponse` are shifted up by one to match up the [13/WAKU2-STORE](https://rfc.vac.dev/spec/13/) specs. 
 - Adds optional `timestamp` to `WakuRelayMessage`.
 #### General refactoring
+- `wakunode2` setup refactored into 6 distinct phases with improved logging and error handling
 #### Docs
 - Adds the database migration tutorial.
 #### Schema

waku/v2/node/jsonrpc/filter_api.nim

@@ -2,8 +2,8 @@
 
 import
   std/[tables,sequtils],
+  chronicles,
   json_rpc/rpcserver,
-  eth/[common, rlp, keys, p2p],
   ../wakunode2,
   ./jsonrpc_types
 

waku/v2/node/jsonrpc/private_api.nim

@@ -2,11 +2,13 @@
 
 import
   std/[tables,sequtils],
+  chronicles,
   json_rpc/rpcserver,
   nimcrypto/sysrand,
-  eth/[common, rlp, keys, p2p],
-  ../wakunode2, ../waku_payload,
-  ./jsonrpc_types, ./jsonrpc_utils
+  ../wakunode2,
+  ../waku_payload,
+  ./jsonrpc_types,
+  ./jsonrpc_utils
 
 export waku_payload, jsonrpc_types
 
@@ -37,7 +39,7 @@ proc installPrivateApiHandlers*(node: WakuNode, rpcsrv: RpcServer, rng: ref BrHm
 
     let msg = message.toWakuMessage(version = 1,
                                     rng = rng,
-                                    pubKey = none(keys.PublicKey),
+                                    pubKey = none(waku_payload.PublicKey),
                                     symkey =  some(symkey.toSymKey()))
 
     if (await node.publish(topic, msg).withTimeout(futTimeout)):
@@ -60,7 +62,7 @@ proc installPrivateApiHandlers*(node: WakuNode, rpcsrv: RpcServer, rng: ref BrHm
       # Clear cache before next call
       topicCache[topic] = @[]
       return msgs.mapIt(it.toWakuRelayMessage(symkey = some(symkey.toSymKey()),
-                                              privateKey = none(keys.PrivateKey)))
+                                              privateKey = none(waku_payload.PrivateKey)))
     else:
       # Not subscribed to this topic
       raise newException(ValueError, "Not subscribed to topic: " & topic)
@@ -71,7 +73,7 @@ proc installPrivateApiHandlers*(node: WakuNode, rpcsrv: RpcServer, rng: ref BrHm
     ## Generates and returns a public/private key pair for asymmetric message encryption and decryption.
     debug "get_waku_v2_private_v1_asymmetric_keypair"
 
-    let privKey = keys.PrivateKey.random(rng[])
+    let privKey = waku_payload.PrivateKey.random(rng[])
 
     return WakuKeyPair(seckey: privKey, pubkey: privKey.toPublicKey())
 

waku/v2/node/jsonrpc/relay_api.nim

@@ -2,11 +2,12 @@
 
 import
   std/[tables,sequtils],
+  chronicles,
   json_rpc/rpcserver,
   libp2p/protocols/pubsub/pubsub,
-  eth/[common, rlp, keys, p2p],
   ../wakunode2,
-  ./jsonrpc_types, ./jsonrpc_utils
+  ./jsonrpc_types,
+  ./jsonrpc_utils
 
 export jsonrpc_types
 

waku/v2/node/waku_setup.nim

@@ -0,0 +1,115 @@
+{.push raises: [Defect].}
+
+## Collection of utilities commonly used
+## during the setup phase of a Waku v2 node
+
+import
+  std/tables,
+  chronos,
+  chronicles,
+  json_rpc/rpcserver,
+  metrics,
+  metrics/chronos_httpserver,
+  stew/results,
+  stew/shims/net,
+  ./storage/sqlite,
+  ./storage/migration/migration_types,
+  ./jsonrpc/[admin_api,
+             debug_api,
+             filter_api,
+             relay_api,
+             store_api,
+             private_api,
+             debug_api],
+  ./config,
+  ./wakunode2
+
+logScope:
+  topics = "wakunode.setup"
+
+type
+  SetupResult*[T] = Result[T, string]
+
+##########################
+# Setup helper functions #
+##########################
+
+proc startRpc*(node: WakuNode, rpcIp: ValidIpAddress, rpcPort: Port, conf: WakuNodeConf)
+  {.raises: [Defect, RpcBindError, CatchableError].} =
+  # @TODO: API handlers still raise CatchableError
+
+  let
+    ta = initTAddress(rpcIp, rpcPort)
+    rpcServer = newRpcHttpServer([ta])
+  installDebugApiHandlers(node, rpcServer)
+
+  # Install enabled API handlers:
+  if conf.relay:
+    let topicCache = newTable[string, seq[WakuMessage]]()
+    installRelayApiHandlers(node, rpcServer, topicCache)
+    if conf.rpcPrivate:
+      # Private API access allows WakuRelay functionality that 
+      # is backwards compatible with Waku v1.
+      installPrivateApiHandlers(node, rpcServer, node.rng, topicCache)
+  
+  if conf.filter:
+    let messageCache = newTable[ContentTopic, seq[WakuMessage]]()
+    installFilterApiHandlers(node, rpcServer, messageCache)
+  
+  if conf.store:
+    installStoreApiHandlers(node, rpcServer)
+  
+  if conf.rpcAdmin:
+    installAdminApiHandlers(node, rpcServer)
+  
+  rpcServer.start()
+  info "RPC Server started", ta
+
+proc startMetricsServer*(serverIp: ValidIpAddress, serverPort: Port) =
+    info "Starting metrics HTTP server", serverIp, serverPort
+    
+    try:
+      startMetricsHttpServer($serverIp, serverPort)
+    except Exception as e:
+      raiseAssert("Exception while starting metrics HTTP server: " & e.msg)
+
+    info "Metrics HTTP server started", serverIp, serverPort
+
+proc startMetricsLog*() =
+  # https://github.com/nim-lang/Nim/issues/17369
+  var logMetrics: proc(udata: pointer) {.gcsafe, raises: [Defect].}
+  logMetrics = proc(udata: pointer) =
+    {.gcsafe.}:
+      # TODO: libp2p_pubsub_peers is not public, so we need to make this either
+      # public in libp2p or do our own peer counting after all.
+      var
+        totalMessages = 0.float64
+
+      for key in waku_node_messages.metrics.keys():
+        try:
+          totalMessages = totalMessages + waku_node_messages.value(key)
+        except KeyError:
+          discard
+
+    info "Node metrics", totalMessages
+    discard setTimer(Moment.fromNow(2.seconds), logMetrics)
+  discard setTimer(Moment.fromNow(2.seconds), logMetrics)
+  
+proc runMigrations*(sqliteDatabase: SqliteDatabase, conf: WakuNodeConf) =
+  # Run migration scripts on persistent storage
+
+  var migrationPath: string
+  if conf.persistPeers and conf.persistMessages:
+    migrationPath = migration_types.ALL_STORE_MIGRATION_PATH
+  elif conf.persistPeers:
+    migrationPath = migration_types.PEER_STORE_MIGRATION_PATH
+  elif conf.persistMessages:
+    migrationPath = migration_types.MESSAGE_STORE_MIGRATION_PATH
+
+  # run migration 
+  info "running migration ... "
+  let migrationResult = sqliteDatabase.migrate(migrationPath)
+  if migrationResult.isErr:
+    warn "migration failed"
+  else:
+    info "migration is done"