NodeManager.ts

import type { DB, DBTransaction, LevelPath } from '@matrixai/db';
import type { ContextTimed, ContextTimedInput } from '@matrixai/contexts';
import type { PromiseCancellable } from '@matrixai/async-cancellable';
import type { ResourceAcquire } from '@matrixai/resources';
import type KeyRing from '../keys/KeyRing.js';
import type Sigchain from '../sigchain/Sigchain.js';
import type TaskManager from '../tasks/TaskManager.js';
import type GestaltGraph from '../gestalts/GestaltGraph.js';
import type {
  Task,
  TaskHandler,
  TaskHandlerId,
  TaskInfo,
} from '../tasks/types.js';
import type { SignedTokenEncoded } from '../tokens/types.js';
import type { Host, Port } from '../network/types.js';
import type {
  Claim,
  ClaimId,
  ClaimIdEncoded,
  SignedClaim,
} from '../claims/types.js';
import type {
  ClaimLinkNode,
  ClaimNetworkAccess,
} from '../claims/payloads/index.js';
import type NodeConnection from '../nodes/NodeConnection.js';
import type {
  AgentClaimMessage,
  AgentRPCRequestParams,
  AgentRPCResponseResult,
} from './agent/types.js';
import type {
  NodeAddress,
  NodeBucket,
  NodeBucketIndex,
  NodeContact,
  NodeContactAddressData,
  NodeId,
  NodeIdEncoded,
} from './types.js';
import type NodeConnectionManager from './NodeConnectionManager.js';
import type NodeGraph from './NodeGraph.js';
import type { ServicePOJO } from '@matrixai/mdns';
import type { AgentClientManifestNodeManager } from './agent/callers/index.js';
import type { ClaimNetworkAuthority } from '../claims/payloads/claimNetworkAuthority.js';
import { withF } from '@matrixai/resources';
import { events as mdnsEvents, MDNS, utils as mdnsUtils } from '@matrixai/mdns';
import Logger from '@matrixai/logger';
import { startStop } from '@matrixai/async-init';
import { Lock, LockBox, Semaphore } from '@matrixai/async-locks';
import { IdInternal } from '@matrixai/id';
import { decorators } from '@matrixai/contexts';
import * as nodesUtils from './utils.js';
import * as nodesEvents from './events.js';
import * as nodesErrors from './errors.js';
import NodeConnectionQueue from './NodeConnectionQueue.js';
import config from '../config.js';
import * as claimNetworkAuthorityUtils from '../claims/payloads/claimNetworkAuthority.js';
import * as claimNetworkAccessUtils from '../claims/payloads/claimNetworkAccess.js';
import Token from '../tokens/Token.js';
import * as keysUtils from '../keys/utils/index.js';
import * as tasksErrors from '../tasks/errors.js';
import * as claimsUtils from '../claims/utils.js';
import * as claimsErrors from '../claims/errors.js';
import * as utils from '../utils/utils.js';
import * as networkUtils from '../network/utils.js';

const abortEphemeralTaskReason = Symbol('abort ephemeral task reason');
const abortSingletonTaskReason = Symbol('abort singleton task reason');
const abortPendingConnectionsReason = Symbol(
  'abort pending connections reason',
);

/**
 * NodeManager manages all operations involving nodes.
 * It encapsulates mutations to the NodeGraph.
 * It listens to the NodeConnectionManager events.
 */
// eslint-disable-next-line
interface NodeManager<Manifest extends AgentClientManifestNodeManager>
  extends startStop.StartStop {}
@startStop.StartStop({
  eventStart: nodesEvents.EventNodeManagerStart,
  eventStarted: nodesEvents.EventNodeManagerStarted,
  eventStop: nodesEvents.EventNodeManagerStop,
  eventStopped: nodesEvents.EventNodeManagerStopped,
})
class NodeManager<Manifest extends AgentClientManifestNodeManager> {
  /**
   * Time used to establish `NodeConnection`
   */
  public readonly connectionConnectTimeoutTime: number;

  public readonly refreshBucketDelayTime: number;
  public readonly refreshBucketDelayJitter: number;
  /**
   * Interval used to reestablish connections to maintain network health.
   * Will trigger a refreshBucket for bucket 255 if it is missing connections.
   * Will always trigger a findNode(this.keyRing.getNodeId()).
   */
  public readonly retryConnectionsDelayTime: number;
  /**
   * Timeout for finding a connection via MDNS
   */
  public readonly connectionFindMDNSTimeoutTime: number;
  public readonly tasksPath = 'NodeManager';

  protected db: DB;
  protected logger: Logger;
  protected keyRing: KeyRing;
  protected sigchain: Sigchain;
  protected gestaltGraph: GestaltGraph;
  protected taskManager: TaskManager;
  protected nodeGraph: NodeGraph;
  protected nodeConnectionManager: NodeConnectionManager<Manifest>;
  protected mdnsOptions:
    | {
        groups: Array<Host>;
        port: Port;
      }
    | undefined;
  protected mdns: MDNS | undefined;

  protected pendingNodes: Map<
    number,
    Map<string, [NodeAddress, NodeContactAddressData]>
  > = new Map();
  protected concurrencyLimit = 3;
  protected dnsServers: Array<string> | undefined = undefined;

  /**
   * Used to track locks for authentication failure and acquiring connections
   */
  protected connectionLockBox: LockBox<Lock> = new LockBox();

  /**
   * If this node is acting as a network authority then the claim is stored here and used as needed.
   * If the node is not acting as an authority then the lack of claim here should indicate that.
   * If the claim is missing then any request that requires it should reject with an error.
   */
  protected claimNetworkAuthority: Token<ClaimNetworkAuthority> | undefined =
    undefined;
  /**
   * If a node has joined a network then it's `ClaimNetworkAccess` is tracked here
   */
  protected claimNetworkAccess: Token<ClaimNetworkAccess> | undefined =
    undefined;

  /**
   * These are the level paths for mapping the ClaimNetworkAccess and ClaimNetworkAuthority claims for each network it has joined.
   * Used to look up and switch between networks as needed.
   */
  protected nodeManagerDbPath: LevelPath = [this.constructor.name];
  protected nodeManagerClaimNetworkAuthorityPath: LevelPath = [
    ...this.nodeManagerDbPath,
    'claimNetworkAuthority',
  ];
  protected nodeManagerClaimNetworkAccessPath: LevelPath = [
    ...this.nodeManagerDbPath,
    'claimNetworkAccess',
  ];

  protected refreshBucketHandler: TaskHandler = async (
    ctx,
    _taskInfo,
    bucketIndex: NodeBucketIndex,
  ) => {
    await this.refreshBucket(bucketIndex, undefined, ctx);
    // When completed reschedule the task
    // if refreshBucketDelay is 0 then it's considered disabled
    if (this.refreshBucketDelayTime > 0) {
      const jitter = nodesUtils.refreshBucketsDelayJitter(
        this.refreshBucketDelayTime,
        this.refreshBucketDelayJitter,
      );
      await this.taskManager.scheduleTask({
        delay: this.refreshBucketDelayTime + jitter,
        handlerId: this.refreshBucketHandlerId,
        lazy: true,
        parameters: [bucketIndex],
        path: [this.tasksPath, this.refreshBucketHandlerId, `${bucketIndex}`],
        priority: 0,
      });
    }
  };

  public readonly refreshBucketHandlerId =
    `${this.tasksPath}.refreshBucketHandler` as TaskHandlerId;

  protected gcBucketHandler: TaskHandler = async (
    ctx,
    _taskInfo,
    bucketIndex: number,
  ) => {
    await this.garbageCollectBucket(
      bucketIndex,
      this.connectionConnectTimeoutTime,
      ctx,
    );
    // Checking for any new pending tasks
    const pendingNodesRemaining = this.pendingNodes.get(bucketIndex);
    if (pendingNodesRemaining == null || pendingNodesRemaining.size === 0) {
      return;
    }
    // Re-schedule the task
    await this.setupGCTask(bucketIndex);
  };

  public readonly gcBucketHandlerId =
    `${this.tasksPath}.gcBucketHandler` as TaskHandlerId;

  protected checkConnectionsHandler: TaskHandler = async (ctx, taskInfo) => {
    this.logger.debug('Checking connections');
    let connectionCount = 0;
    for (const connection of this.nodeConnectionManager.listConnections()) {
      if (connection.primary && connection.authenticated) {
        const [bucketId] = this.nodeGraph.bucketIndex(connection.nodeId);
        if (bucketId === 255) connectionCount++;
      }
    }
    if (connectionCount > 0) {
      this.logger.debug('triggering bucket refresh for bucket 255');
      await this.updateRefreshBucketDelay(255, 0, undefined, undefined, ctx);
    }
    try {
      this.logger.debug(
        'triggering findNode for self to populate closest nodes',
      );
      await this.findNode(
        {
          nodeId: this.keyRing.getNodeId(),
        },
        ctx,
      );
    } finally {
      this.logger.debug('Checked connections');
      // Re-schedule this task
      await this.taskManager.scheduleTask({
        delay: taskInfo.delay,
        deadline: taskInfo.deadline,
        handlerId: this.checkConnectionsHandlerId,
        lazy: true,
        path: [this.tasksPath, this.checkConnectionsHandlerId],
        priority: taskInfo.priority,
      });
    }
  };

  public readonly checkConnectionsHandlerId: TaskHandlerId =
    `${this.tasksPath}.checkConnectionsHandler` as TaskHandlerId;

  protected syncNodeGraphHandler = async (
    ctx: ContextTimed,
    _taskInfo: TaskInfo | undefined,
    network: string | undefined,
    initialNodes: Array<[NodeIdEncoded, NodeAddress]>,
    connectionConnectTimeoutTime: number | undefined,
  ) => {
    // Establishing connections to the initial nodes
    const connectionResults = await Promise.allSettled(
      initialNodes.map(async ([nodeIdEncoded, nodeAddress]) => {
        const resolvedHosts = await networkUtils.resolveHostnames(
          [nodeAddress],
          undefined,
          this.dnsServers,
          ctx,
        );
        if (resolvedHosts.length === 0) {
          throw new nodesErrors.ErrorNodeManagerResolveNodeFailed(
            `Failed to resolve '${nodeAddress[0]}'`,
          );
        }
        const nodeId = nodesUtils.decodeNodeId(nodeIdEncoded);
        if (nodeId == null) {
          utils.never(`failed to decode NodeId "${nodeIdEncoded}"`);
        }
        return this.nodeConnectionManager.createConnectionMultiple(
          [nodeId],
          resolvedHosts,
          { timer: connectionConnectTimeoutTime, signal: ctx.signal },
        );
      }),
    );
    const successfulConnections = connectionResults.filter(
      (r) => r.status === 'fulfilled',
    ) as Array<PromiseFulfilledResult<NodeConnection<Manifest>>>;
    if (successfulConnections.length === 0) {
      const failedConnectionErrors = connectionResults
        .filter((r) => r.status === 'rejected')
        .map((v) => {
          if (v.status === 'rejected') return v.reason;
        });
      throw new nodesErrors.ErrorNodeManagerSyncNodeGraphFailed(
        `Failed to establish any connections with the following errors '[${failedConnectionErrors}]'`,
        {
          cause: new AggregateError(failedConnectionErrors),
        },
      );
    }
    if (ctx.signal.aborted) return;

    if (network != null) {
      if ((await this.getClaimNetworkAccess(network)) == null) {
        await this.claimNetwork(successfulConnections[0].value.nodeId, network);
      } else {
        await this.switchNetwork(network);
      }
    }

    // Attempt a findNode operation looking for ourselves
    await this.findNode(
      {
        nodeId: this.keyRing.getNodeId(),
      },
      ctx,
    );
    if (ctx.signal.aborted) return;

    // Getting the closest node from the `NodeGraph`
    let bucketIndex: number | undefined;
    for await (const bucket of this.nodeGraph.getBuckets('distance', 'asc')) {
      if (ctx.signal.aborted) return;
      bucketIndex = bucket[0];
    }
    // If no buckets then end here
    if (bucketIndex == null) return;
    // Trigger refreshBucket operations for all buckets above bucketIndex
    const refreshBuckets: Array<Promise<any>> = [];
    for (let i = bucketIndex; i < this.nodeGraph.nodeIdBits; i++) {
      const task = await this.updateRefreshBucketDelay(i, 0, false);
      refreshBuckets.push(task.promise());
    }
    const signalP = utils.signalPromise(ctx.signal);
    await Promise.race([Promise.all(refreshBuckets), signalP]).finally(
      async () => {
        // Clean up signal promise when done
        signalP.cancel();
        await signalP;
      },
    );
  };

  public readonly syncNodeGraphHandlerId: TaskHandlerId =
    `${this.tasksPath}.syncNodeGraphHandler` as TaskHandlerId;

  protected handleEventNodeConnectionManagerConnectionAuthenticated = async (
    e: nodesEvents.EventNodeConnectionManagerConnectionAuthenticated,
  ) => {
    await this.setNode(
      e.detail.remoteNodeId,
      [e.detail.remoteHost, e.detail.remotePort],
      // FIXME: We want to distinguish punched, direct and local connections
      {
        mode: 'direct',
        connectedTime: Date.now(),
        scopes: ['global'],
      },
      false,
      false,
    );
  };

  constructor({
    db,
    keyRing,
    sigchain,
    gestaltGraph,
    taskManager,
    nodeGraph,
    nodeConnectionManager,
    mdnsOptions,
    connectionConnectTimeoutTime = config.defaultsSystem
      .nodesConnectionConnectTimeoutTime,
    refreshBucketDelayTime = config.defaultsSystem
      .nodesRefreshBucketIntervalTime,
    refreshBucketDelayJitter = config.defaultsSystem
      .nodesRefreshBucketIntervalTimeJitter,
    retryConnectionsDelayTime = 45_000, // 45 seconds
    nodesConnectionFindLocalTimeoutTime = config.defaultsSystem
      .nodesConnectionFindLocalTimeoutTime,
    dnsServers,
    logger,
  }: {
    db: DB;
    keyRing: KeyRing;
    sigchain: Sigchain;
    gestaltGraph: GestaltGraph;
    taskManager: TaskManager;
    nodeGraph: NodeGraph;
    mdnsOptions?: {
      groups: Array<Host>;
      port: Port;
    };
    nodeConnectionManager: NodeConnectionManager<Manifest>;
    connectionConnectTimeoutTime?: number;
    refreshBucketDelayTime?: number;
    refreshBucketDelayJitter?: number;
    retryConnectionsDelayTime?: number;
    nodesConnectionFindLocalTimeoutTime?: number;
    dnsServers?: Array<string>;
    logger?: Logger;
  }) {
    this.logger = logger ?? new Logger(this.constructor.name);
    this.db = db;
    this.keyRing = keyRing;
    this.sigchain = sigchain;
    this.nodeConnectionManager = nodeConnectionManager;
    this.nodeGraph = nodeGraph;
    this.taskManager = taskManager;
    this.gestaltGraph = gestaltGraph;
    this.mdnsOptions = mdnsOptions;
    if (mdnsOptions != null) {
      this.mdns = new MDNS({ logger: this.logger.getChild(MDNS.name) });
    }
    this.connectionConnectTimeoutTime = connectionConnectTimeoutTime;
    this.refreshBucketDelayTime = refreshBucketDelayTime;
    this.refreshBucketDelayJitter = Math.max(0, refreshBucketDelayJitter);
    this.retryConnectionsDelayTime = retryConnectionsDelayTime;
    this.connectionFindMDNSTimeoutTime = nodesConnectionFindLocalTimeoutTime;
    if (dnsServers != null) {
      this.logger.info(
        `Overriding DNS resolution servers with ${JSON.stringify(dnsServers)}`,
      );
    }
    this.dnsServers = dnsServers;
  }

  public async start() {
    this.logger.info(`Starting ${this.constructor.name}`);
    this.taskManager.registerHandler(
      this.refreshBucketHandlerId,
      this.refreshBucketHandler,
    );
    this.taskManager.registerHandler(
      this.gcBucketHandlerId,
      this.gcBucketHandler,
    );
    this.taskManager.registerHandler(
      this.checkConnectionsHandlerId,
      this.checkConnectionsHandler,
    );
    this.taskManager.registerHandler(
      this.syncNodeGraphHandlerId,
      this.syncNodeGraphHandler,
    );
    // This will clean up tasks that were not properly cleaned up during an ungracefully shutdown of the process
    await this.stopTasks();
    await this.setupRefreshBucketTasks();
    // Can be disabled with 0 delay, only use for testing
    if (this.retryConnectionsDelayTime > 0) {
      await this.taskManager.scheduleTask({
        delay: this.retryConnectionsDelayTime,
        handlerId: this.checkConnectionsHandlerId,
        lazy: true,
        path: [this.tasksPath, this.checkConnectionsHandlerId],
      });
    }
    // Starting MDNS
    if (this.mdns != null) {
      const nodeId = this.keyRing.getNodeId();
      await this.mdns.start({
        id: nodeId.toBuffer().readUint16BE(),
        hostname: nodesUtils.encodeNodeId(nodeId),
        groups: this.mdnsOptions!.groups,
        port: this.mdnsOptions!.port,
      });
      this.mdns.registerService({
        name: nodesUtils.encodeNodeId(this.keyRing.getNodeId()),
        port: this.nodeConnectionManager.port,
        type: 'polykey',
        protocol: 'udp',
      });
    }
    // Add handling for connections
    this.nodeConnectionManager.addEventListener(
      nodesEvents.EventNodeConnectionManagerConnectionAuthenticated.name,
      this.handleEventNodeConnectionManagerConnectionAuthenticated,
    );
    this.logger.info(`Started ${this.constructor.name}`);
  }

  public async stop() {
    this.logger.info(`Stopping ${this.constructor.name}`);
    // Remove handling for connections
    this.nodeConnectionManager.removeEventListener(
      nodesEvents.EventNodeConnectionManagerConnectionAuthenticated.name,
      this.handleEventNodeConnectionManagerConnectionAuthenticated,
    );
    await this.mdns?.stop();
    await this.stopTasks();
    this.taskManager.deregisterHandler(this.refreshBucketHandlerId);
    this.taskManager.deregisterHandler(this.gcBucketHandlerId);
    this.taskManager.deregisterHandler(this.checkConnectionsHandlerId);
    this.taskManager.deregisterHandler(this.syncNodeGraphHandlerId);
    this.logger.info(`Stopped ${this.constructor.name}`);
  }

  /**
   * For usage with withF, to acquire a connection
   * This unique acquire function structure of returning the ResourceAcquire
   * itself is such that we can pass targetNodeId as a parameter (as opposed to
   * an acquire function with no parameters).
   * @param nodeId Id of target node to communicate with
   * @param ctx
   * @returns ResourceAcquire Resource API for use in with contexts
   */
  public acquireConnection(
    nodeId: NodeId,
    ctx: ContextTimed,
  ): ResourceAcquire<NodeConnection<Manifest>> {
    if (this.keyRing.getNodeId().equals(nodeId)) {
      throw new nodesErrors.ErrorNodeManagerNodeIdOwn();
    }
    return async () => {
      return await this.connectionLockBox.withF(
        [['acquireConnection', nodeId.toString()].join('.'), Lock],
        async () => {
          // Checking if connection already exists
          if (!this.nodeConnectionManager.hasConnection(nodeId)) {
            // Establish the connection
            const result = await this.findNode({ nodeId: nodeId }, ctx);
            if (result == null) {
              throw new nodesErrors.ErrorNodeManagerConnectionFailed();
            }
          }
          // Initiate authentication and await
          return await this.nodeConnectionManager.acquireConnection(
            nodeId,
            ctx,
          )();
        },
      );
    };
  }

  /**
   * Perform some function on another node over the network with a connection.
   * Will either retrieve an existing connection, or create a new one if it
   * doesn't exist.
   * For use with normal arrow function
   * @param nodeId Id of target node to communicate with
   * @param f Function to handle communication
   * @param ctx
   */
  public async withConnF<T>(
    nodeId: NodeId,
    ctx: Partial<ContextTimedInput> | undefined,
    f: (conn: NodeConnection<Manifest>) => Promise<T>,
  ): Promise<T>;
  @startStop.ready(new nodesErrors.ErrorNodeManagerNotRunning())
  public async withConnF<T>(
    nodeId: NodeId,
    @decorators.context ctx: ContextTimed,
    f: (conn: NodeConnection<Manifest>) => Promise<T>,
  ): Promise<T> {
    return await withF(
      [this.acquireConnection(nodeId, ctx)],
      async ([conn]) => {
        return await f(conn);
      },
    );
  }

  /**
   * Perform some function on another node over the network with a connection.
   * Will either retrieve an existing connection, or create a new one if it
   * doesn't exist.
   * for use with a generator function
   * @param nodeId Id of target node to communicate with
   * @param g Generator function to handle communication
   * @param ctx
   */
  public withConnG<T, TReturn, TNext>(
    nodeId: NodeId,
    ctx: Partial<ContextTimedInput> | undefined,
    g: (conn: NodeConnection<Manifest>) => AsyncGenerator<T, TReturn, TNext>,
  ): AsyncGenerator<T, TReturn, TNext>;
  @startStop.ready(new nodesErrors.ErrorNodeManagerNotRunning())
  @decorators.timed()
  public async *withConnG<T, TReturn, TNext>(
    nodeId: NodeId,
    @decorators.context ctx: ContextTimed,
    g: (conn: NodeConnection<Manifest>) => AsyncGenerator<T, TReturn, TNext>,
  ): AsyncGenerator<T, TReturn, TNext> {
    const acquire = this.acquireConnection(nodeId, ctx);
    const [release, conn] = await acquire();
    let caughtError: Error | undefined;
    try {
      if (conn == null) utils.never('NodeConnection should exist');
      return yield* g(conn);
    } catch (e) {
      caughtError = e;
      throw e;
    } finally {
      await release(caughtError);
    }
  }

  /**
   * Will attempt to find a node within the network using a hybrid strategy of
   * attempting signalled connections, direct connections and checking MDNS.
   *
   * Will attempt to fix regardless of existing connection.
   * @param nodeId - NodeId of target to find.
   * @param connectionConnectTimeoutTime - timeout time for each individual connection.
   * @param connectionFindLocalTimeoutTime
   * @param concurrencyLimit - Limit the number of concurrent connections.
   * @param limit - Limit the number of total connections to be made before giving up.
   * @param ctx
   */
  public findNode(
    {
      nodeId,
      connectionConnectTimeoutTime,
      connectionFindMDNSTimeoutTime,
      concurrencyLimit,
      limit,
    }: {
      nodeId: NodeId;
      connectionConnectTimeoutTime?: number;
      connectionFindMDNSTimeoutTime?: number;
      concurrencyLimit?: number;
      limit?: number;
    },
    ctx?: Partial<ContextTimedInput>,
  ): PromiseCancellable<[NodeAddress, NodeContactAddressData] | undefined>;
  @decorators.timedCancellable(true)
  public async findNode(
    {
      nodeId,
      connectionConnectTimeoutTime = this.connectionConnectTimeoutTime,
      connectionFindMDNSTimeoutTime = this.connectionFindMDNSTimeoutTime,
      concurrencyLimit = this.concurrencyLimit,
      limit = this.nodeGraph.nodeBucketLimit,
    }: {
      nodeId: NodeId;
      connectionConnectTimeoutTime: number;
      connectionFindMDNSTimeoutTime: number;
      concurrencyLimit: number;
      limit: number;
    },
    @decorators.context ctx: ContextTimed,
  ): Promise<[NodeAddress, NodeContactAddressData] | undefined> {
    // Setting up intermediate signal
    const abortController = new AbortController();
    const newCtx = {
      timer: ctx.timer,
      signal: abortController.signal,
    };
    const handleAbort = () => {
      abortController.abort(ctx.signal.reason);
    };
    if (ctx.signal.aborted) {
      handleAbort();
    } else {
      ctx.signal.addEventListener('abort', handleAbort, { once: true });
    }

    const rateLimit = new Semaphore(concurrencyLimit);
    const connectionsQueue = new NodeConnectionQueue(
      this.keyRing.getNodeId(),
      nodeId,
      limit,
      rateLimit,
      rateLimit,
    );

    // Starting discovery strategies
    const findBySignal = this.findNodeBySignal(
      nodeId,
      connectionsQueue,
      connectionConnectTimeoutTime,
      newCtx,
    );
    const findByDirect = this.findNodeByDirect(
      nodeId,
      connectionsQueue,
      connectionConnectTimeoutTime,
      newCtx,
    );
    const findByMDNS = this.findNodeByMDNS(nodeId, {
      timer: connectionFindMDNSTimeoutTime,
      signal: newCtx.signal,
    });

    try {
      return await Promise.any([findBySignal, findByDirect, findByMDNS]);
    } catch (e) {
      if (e instanceof AggregateError) {
        for (const error of e.errors) {
          // Checking if each error is an expected error
          if (!(error instanceof nodesErrors.ErrorNodeManagerFindNodeFailed)) {
            throw e;
          }
        }
      } else if (!(e instanceof nodesErrors.ErrorNodeManagerFindNodeFailed)) {
        throw e;
      }
      return;
    } finally {
      abortController.abort(abortPendingConnectionsReason);
      await Promise.allSettled([findBySignal, findByDirect, findByMDNS]);
      ctx.signal.removeEventListener('abort', handleAbort);
    }
  }

  /**
   * Will try to make a connection to the node using hole punched connections only
   *
   * @param nodeId - NodeId of the target to find.
   * @param nodeConnectionsQueue - shared nodeConnectionQueue helper class.
   * @param connectionConnectTimeoutTime - timeout time for each individual connection.
   * @param ctx
   */
  public findNodeBySignal(
    nodeId: NodeId,
    nodeConnectionsQueue: NodeConnectionQueue,
    connectionConnectTimeoutTime?: number,
    ctx?: Partial<ContextTimedInput>,
  ): PromiseCancellable<[[Host, Port], NodeContactAddressData]>;
  @decorators.timedCancellable(true)
  public async findNodeBySignal(
    nodeId: NodeId,
    nodeConnectionsQueue: NodeConnectionQueue,
    connectionConnectTimeoutTime: number = this.connectionConnectTimeoutTime,
    @decorators.context ctx: ContextTimed,
  ): Promise<[[Host, Port], NodeContactAddressData]> {
    // Setting up intermediate signal
    const abortController = new AbortController();
    utils.setMaxListeners(abortController.signal);
    const newCtx = {
      timer: ctx.timer,
      signal: abortController.signal,
    };
    const handleAbort = () => {
      abortController.abort(ctx.signal.reason);
    };
    if (ctx.signal.aborted) {
      handleAbort();
    } else {
      ctx.signal.addEventListener('abort', handleAbort, { once: true });
    }

    const chain: Map<string, string | undefined> = new Map();
    let connectionMade: [Host, Port] | undefined;

    // Seed the initial queue
    for (const {
      nodeId: nodeIdConnected,
    } of this.nodeConnectionManager.getClosestConnections(nodeId)) {
      nodeConnectionsQueue.queueNodeSignal(nodeIdConnected, undefined);
    }

    while (true) {
      const isDone = await nodeConnectionsQueue.withNodeSignal(
        async (nodeIdTarget, nodeIdSignaller) => {
          let nodeConnection: NodeConnection<Manifest> | undefined;
          if (
            !this.nodeConnectionManager.hasConnection(nodeIdTarget) &&
            nodeIdSignaller != null
          ) {
            this.logger.debug(
              `attempting connection to ${nodesUtils.encodeNodeId(
                nodeIdTarget,
              )} via ${nodesUtils.encodeNodeId(nodeIdSignaller)}`,
            );
            nodeConnection =
              await this.nodeConnectionManager.createConnectionPunch(
                nodeIdTarget,
                nodeIdSignaller,
                {
                  timer: connectionConnectTimeoutTime,
                  signal: newCtx.signal,
                },
              );
            // If connection succeeds add it to the chain
            chain.set(nodeIdTarget.toString(), nodeIdSignaller?.toString());
          }
          nodeConnectionsQueue.contactedNode(nodeIdTarget);
          // If connection was our target then we're done
          if (nodeId.toString() === nodeIdTarget.toString()) {
            nodeConnection =
              nodeConnection ??
              this.nodeConnectionManager.getConnection(nodeIdTarget)
                ?.connection;
            if (nodeConnection == null) utils.never('connection should exist');
            connectionMade = [nodeConnection.host, nodeConnection.port];
            return true;
          }
          await this.queueDataFromRequest(
            nodeIdTarget,
            nodeId,
            nodeConnectionsQueue,
            newCtx,
          );
          this.logger.debug(
            `connection attempt to ${nodesUtils.encodeNodeId(
              nodeIdTarget,
            )} succeeded`,
          );
          return false;
        },
        newCtx,
      );
      if (isDone) break;
    }
    // After queue is done we want to signal and await clean up
    abortController.abort(abortPendingConnectionsReason);
    ctx.signal.removeEventListener('abort', handleAbort);
    // Wait for pending attempts to finish
    for (const pendingP of nodeConnectionsQueue.nodesRunningSignal) {
      await pendingP.catch((e) => {
        if (e instanceof nodesErrors.ErrorNodeConnectionTimeout) return;
        throw e;
      });
    }

    // Connection was not made
    if (connectionMade == null) {
      throw new nodesErrors.ErrorNodeManagerFindNodeFailed(
        'failed to find node via signal',
      );
    }
    // We can get the path taken with this code
    // const path: Array<NodeId> = [];
    // let current: string | undefined = nodeId.toString();
    // while (current != null) {
    //   const nodeId = IdInternal.fromString<NodeId>(current);
    //   path.unshift(nodeId);
    //   current = chain.get(current);
    // }
    // console.log(path);
    return [
      connectionMade,
      {
        mode: 'signal',
        connectedTime: Date.now(),
        scopes: ['global'],
      },
    ] as [[Host, Port], NodeContactAddressData];
  }

  /**
   * Will try to make a connection to the node using direct connections only
   *
   * @param nodeId - NodeId of the target to find.
   * @param nodeConnectionsQueue - shared nodeConnectionQueue helper class.
   * @param connectionConnectTimeoutTime - timeout time for each individual connection.
   * @param ctx
   */
  public findNodeByDirect(
    nodeId: NodeId,
    nodeConnectionsQueue: NodeConnectionQueue,
    connectionConnectTimeoutTime?: number,
    ctx?: Partial<ContextTimedInput>,
  ): PromiseCancellable<[[Host, Port], NodeContactAddressData]>;
  @decorators.timedCancellable(true)
  public async findNodeByDirect(
    nodeId: NodeId,
    nodeConnectionsQueue: NodeConnectionQueue,
    connectionConnectTimeoutTime: number = this.connectionConnectTimeoutTime,
    @decorators.context ctx: ContextTimed,
  ): Promise<[[Host, Port], NodeContactAddressData]> {
    // Setting up intermediate signal
    const abortController = new AbortController();
    utils.setMaxListeners(abortController.signal);
    const newCtx = {
      timer: ctx.timer,
      signal: abortController.signal,
    };
    const handleAbort = () => {
      abortController.abort(ctx.signal.reason);
    };
    if (ctx.signal.aborted) {
      handleAbort();
    } else {
      ctx.signal.addEventListener('abort', handleAbort, { once: true });
    }

    let connectionMade = false;

    // Seed the initial queue
    for (const [
      nodeIdTarget,
      nodeContact,
    ] of await this.nodeGraph.getClosestNodes(
      nodeId,
      this.nodeGraph.nodeBucketLimit,
      undefined,
      ctx,
    )) {
      nodeConnectionsQueue.queueNodeDirect(nodeIdTarget, nodeContact);
    }

    while (true) {
      const isDone = await nodeConnectionsQueue.withNodeDirect(
        async (nodeIdTarget, nodeContact) => {
          if (!this.nodeConnectionManager.hasConnection(nodeIdTarget)) {
            this.logger.debug(
              `attempting connection to ${nodesUtils.encodeNodeId(
                nodeIdTarget,
              )} via direct connection`,
            );

            // Attempt all direct
            const addresses: Array<NodeAddress> = [];
            for (const [
              nodeContactAddress,
              nodeContactAddressData,
            ] of Object.entries(nodeContact)) {
              const [host, port] =
                nodesUtils.parseNodeContactAddress(nodeContactAddress);
              if (nodeContactAddressData.mode === 'direct') {
                addresses.push([host, port]);
              }
            }
            const resolvedHosts = await networkUtils.resolveHostnames(
              addresses,
              undefined,
              this.dnsServers,
              ctx,
            );

            try {
              await this.nodeConnectionManager.createConnectionMultiple(
                [nodeIdTarget],
                resolvedHosts,
                { timer: connectionConnectTimeoutTime, signal: newCtx.signal },
              );
            } catch (e) {
              if (e instanceof nodesErrors.ErrorNodeConnectionTimeout) {
                return false;
              }
              throw e;
            }
          }
          nodeConnectionsQueue.contactedNode(nodeIdTarget);
          // If connection was our target then we're done
          if (nodeId.toString() === nodeIdTarget.toString()) {
            connectionMade = true;
            return true;
          }
          await this.queueDataFromRequest(
            nodeIdTarget,
            nodeId,
            nodeConnectionsQueue,
            newCtx,
          );
          this.logger.debug(
            `connection attempt to ${nodesUtils.encodeNodeId(
              nodeIdTarget,
            )} succeeded`,
          );
          return false;
        },
        newCtx,
      );
      if (isDone) break;
    }
    // After queue is done we want to signal and await clean up
    abortController.abort(abortPendingConnectionsReason);
    ctx.signal.removeEventListener('abort', handleAbort);
    // Wait for pending attempts to finish
    for (const pendingP of nodeConnectionsQueue.nodesRunningDirect) {
      await pendingP.catch((e) => {
        if (e instanceof nodesErrors.ErrorNodeConnectionTimeout) return;
        throw e;
      });
    }
    if (!connectionMade) {
      throw new nodesErrors.ErrorNodeManagerFindNodeFailed(
        'failed to find node via direct',
      );
    }
    const conAndTimer = this.nodeConnectionManager.getConnection(nodeId);
    if (conAndTimer == null) {
      utils.never('connection should have been established');
    }
    return [
      [conAndTimer.connection.host, conAndTimer.connection.port],
      {
        mode: 'direct',
        connectedTime: Date.now(),
        scopes: ['global'],
      },
    ] as [[Host, Port], NodeContactAddressData];
  }

  /**
   * Will query via MDNS for running Polykey nodes with the matching NodeId
   *
   * @param nodeId
   * @param ctx
   */
  public queryMDNS(
    nodeId: NodeId,
    ctx?: Partial<ContextTimedInput>,
  ): PromiseCancellable<Array<[Host, Port]>>;
  @decorators.timedCancellable(
    true,
    (nodeManager: NodeManager<Manifest>) =>
      nodeManager.connectionConnectTimeoutTime,
  )
  public async queryMDNS(
    nodeId: NodeId,
    @decorators.context ctx: ContextTimed,
  ): Promise<Array<[Host, Port]>> {
    const addresses: Array<[Host, Port]> = [];
    if (this.mdns == null) return addresses;
    const encodedNodeId = nodesUtils.encodeNodeId(nodeId);
    // First check if we already have an existing MDNS Service
    const mdnsOptions = { type: 'polykey', protocol: 'udp' } as const;
    let service: ServicePOJO | void = this.mdns.networkServices.get(
      mdnsUtils.toFqdn({ name: encodedNodeId, ...mdnsOptions }),
    );
    if (service == null) {
      // Setup promises
      const { p: endedP, resolveP: resolveEndedP } = utils.promise<void>();
      const {
        p: serviceP,
        resolveP: resolveServiceP,
        rejectP: rejectServiceP,
      } = utils.promise<ServicePOJO>();
      const abortHandler = () => {
        resolveEndedP();
        rejectServiceP();
      };
      ctx.signal.addEventListener('abort', abortHandler, { once: true });
      ctx.timer.catch(() => {}).finally(() => abortHandler());
      const handleEventMDNSService = (evt: mdnsEvents.EventMDNSService) => {
        if (evt.detail.name === encodedNodeId) {
          resolveServiceP(evt.detail);
        }
      };
      this.mdns.addEventListener(
        mdnsEvents.EventMDNSService.name,
        handleEventMDNSService,
        { once: true },
      );
      // Abort and restart query in case already running
      this.mdns.stopQuery(mdnsOptions);
      this.mdns.startQuery(mdnsOptions);
      // Race promises to find node or timeout
      service = await Promise.race([serviceP, endedP]);
      this.mdns.removeEventListener(
        mdnsEvents.EventMDNSService.name,
        handleEventMDNSService,
      );
      this.mdns.stopQuery(mdnsOptions);
      ctx.signal.removeEventListener('abort', abortHandler);
    }
    // If the service is not found, just return no addresses
    if (service == null) {
      return addresses;
    }
    for (const host_ of service.hosts) {
      let host: string;
      switch (this.nodeConnectionManager.type) {
        case 'ipv4':
          if (networkUtils.isIPv4(host_)) {
            host = host_;
          } else if (networkUtils.isIPv4MappedIPv6(host_)) {
            host = networkUtils.fromIPv4MappedIPv6(host_);
          } else {
            continue;
          }
          break;
        case 'ipv6':
          if (networkUtils.isIPv6(host_)) host = host_;
          else continue;
          break;
        case 'ipv4&ipv6':
          host = host_;
          break;
        default:
          continue;
      }
      addresses.push([host as Host, service.port as Port]);
      this.logger.debug(
        `found address for ${nodesUtils.encodeNodeId(nodeId)} at ${host}:${
          service.port
        }`,
      );
    }
    return addresses;
  }

  /**
   * Will query MDNS for local nodes and attempt a connection.
   * @param nodeId - NodeId of the target to find.
   * @param ctx
   */
  public findNodeByMDNS(
    nodeId: NodeId,
    ctx?: Partial<ContextTimedInput>,
  ): PromiseCancellable<[[Host, Port], NodeContactAddressData]>;
  @decorators.timedCancellable(
    true,
    (nodeManager: NodeManager<Manifest>) =>
      nodeManager.connectionFindMDNSTimeoutTime,
  )
  public async findNodeByMDNS(
    nodeId: NodeId,
    @decorators.context ctx: ContextTimed,
  ): Promise<[[Host, Port], NodeContactAddressData]> {
    try {
      if (this.mdns == null) {
        throw new nodesErrors.ErrorNodeManagerFindNodeFailed(
          'MDNS not running',
        );
      }
      // First get the address data
      const addresses = await this.queryMDNS(nodeId, ctx);
      if (addresses.length === 0) {
        throw new nodesErrors.ErrorNodeManagerFindNodeFailed(
          'query resulted in no addresses found',
        );
      }
      // Then make the connection
      const nodeConnection =
        await this.nodeConnectionManager.createConnectionMultiple(
          [nodeId],
          addresses,
          ctx,
        );
      return [
        [nodeConnection.host, nodeConnection.port],
        {
          mode: 'direct',
          connectedTime: Date.now(),
          scopes: ['local'],
        },
      ] as [[Host, Port], NodeContactAddressData];
    } catch (e) {
      throw new nodesErrors.ErrorNodeManagerFindNodeFailed(
        'failed to find node via MDNS',
        { cause: e },
      );
    }
  }

  /**
   * Will ask the target node about the closest nodes to the `node`
   * and add them to the `nodeConnectionsQueue`.
   *
   * @param nodeId - node to find the closest nodes to
   * @param nodeIdTarget - node to make RPC requests to
   * @param nodeConnectionsQueue
   * @param ctx
   */
  protected async queueDataFromRequest(
    nodeId: NodeId,
    nodeIdTarget: NodeId,
    nodeConnectionsQueue: NodeConnectionQueue,
    ctx: ContextTimed,
  ) {
    const nodeIdEncoded = nodesUtils.encodeNodeId(nodeIdTarget);
    const closestConnectionsRequestP = (async () => {
      const data = await this.nodeConnectionManager.withConnF(
        nodeId,
        ctx,
        async (conn) => {
          const resultStream =
            await conn.rpcClient.methods.nodesClosestActiveConnectionsGet(
              {
                nodeIdEncoded: nodeIdEncoded,
              },
              ctx,
            );
          const connections: Array<NodeId> = [];
          // Collecting results
          for await (const result of resultStream) {
            ctx.signal.throwIfAborted();
            const nodeIdNew = nodesUtils.decodeNodeId(result.nodeId);
            if (nodeIdNew == null) {
              utils.never(`failed to decode NodeId "${result.nodeId}"`);
            }
            connections.push(nodeIdNew);
          }
          return connections;
        },
      );
      for (const nodeIdNew of data) {
        nodeConnectionsQueue.queueNodeSignal(nodeIdNew, nodeId);
      }
    })();
    const closestNodesRequestP = (async () => {
      const data = await this.nodeConnectionManager.withConnF(
        nodeId,
        ctx,
        async (conn) => {
          const resultStream =
            await conn.rpcClient.methods.nodesClosestLocalNodesGet(
              {
                nodeIdEncoded: nodeIdEncoded,
              },
              ctx,
            );
          const data: Array<[NodeId, NodeContact]> = [];
          for await (const { nodeIdEncoded, nodeContact } of resultStream) {
            ctx.signal.throwIfAborted();
            const nodeId = nodesUtils.decodeNodeId(nodeIdEncoded);
            if (nodeId == null) {
              utils.never(`failed to decode NodeId "${nodeIdEncoded}"`);
            }
            data.push([nodeId, nodeContact]);
          }
          return data;
        },
      );
      for (const [nodeId, nodeContact] of data) {
        nodeConnectionsQueue.queueNodeDirect(nodeId, nodeContact);
      }
    })();
    await Promise.allSettled([
      closestConnectionsRequestP,
      closestNodesRequestP,
    ]);
  }

  /**
   * Will attempt to establish connection using `findNode` or use existing connection.
   * Will return true if connection was established or already exists, false otherwise.
   */
  public pingNode(
    nodeId: NodeId,
    ctx?: Partial<ContextTimedInput>,
  ): PromiseCancellable<[NodeAddress, NodeContactAddressData] | undefined>;
  @startStop.ready(new nodesErrors.ErrorNodeConnectionManagerNotRunning())
  @decorators.timedCancellable(
    true,
    (nodeConnectionManager: NodeConnectionManager<Manifest>) =>
      nodeConnectionManager.connectionConnectTimeoutTime,
  )
  public async pingNode(
    nodeId: NodeId,
    @decorators.context ctx: ContextTimed,
  ): Promise<[NodeAddress, NodeContactAddressData] | undefined> {
    return await this.findNode(
      {
        nodeId: nodeId,
      },
      ctx,
    );
  }

  /**
   * Will attempt to make a direct connection without ICE.
   * This will only succeed due to these conditions
   * 1. connection already exists to target.
   * 2. Nat already allows port due to already being punched.
   * 3. Port is publicly accessible due to nat configuration .
   * Will return true if connection was established or already exists, false otherwise.
   */
  public pingNodeAddress(
    nodeId: NodeId,
    host: Host,
    port: Port,
    ctx?: Partial<ContextTimedInput>,
  ): PromiseCancellable<boolean>;
  @startStop.ready(new nodesErrors.ErrorNodeConnectionManagerNotRunning())
  @decorators.timedCancellable(
    true,
    (nodeConnectionManager: NodeConnectionManager<Manifest>) =>
      nodeConnectionManager.connectionConnectTimeoutTime,
  )
  public async pingNodeAddress(
    nodeId: NodeId,
    host: Host,
    port: Port,
    @decorators.context ctx: ContextTimed,
  ): Promise<boolean> {
    if (this.nodeConnectionManager.hasConnection(nodeId)) return true;
    try {
      await this.nodeConnectionManager.createConnection(
        [nodeId],
        host,
        port,
        ctx,
      );
      return true;
    } catch (e) {
      if (!nodesUtils.isConnectionError(e)) throw e;
      return false;
    }
  }

  /**
   * Will attempt to make a direct connection without ICE.
   * This will only succeed due to these conditions
   * 1. connection already exists to target.
   * 2. Nat already allows port due to already being punched.
   * 3. Port is publicly accessible due to nat configuration .
   * Will return true if connection was established or already exists, false otherwise.
   */
  public pingNodeAddressMultiple(
    nodeId: NodeId,
    addresses: Array<[Host, Port]>,
    ctx?: Partial<ContextTimedInput>,
  ): PromiseCancellable<boolean>;
  @startStop.ready(new nodesErrors.ErrorNodeConnectionManagerNotRunning())
  @decorators.timedCancellable(
    true,
    (nodeConnectionManager: NodeConnectionManager<Manifest>) =>
      nodeConnectionManager.connectionConnectTimeoutTime,
  )
  public async pingNodeAddressMultiple(
    nodeId: NodeId,
    addresses: Array<[Host, Port]>,
    @decorators.context ctx: ContextTimed,
  ): Promise<boolean> {
    if (this.nodeConnectionManager.hasConnection(nodeId)) return true;
    try {
      await this.nodeConnectionManager.createConnectionMultiple(
        [nodeId],
        addresses,
        ctx,
      );
      return true;
    } catch (e) {
      if (!nodesUtils.isConnectionError(e)) throw e;
      return false;
    }
  }

  /**
   * Connects to the target node, and retrieves its sigchain data.
   * Verifies and returns the decoded chain as ChainData. Note: this will drop
   * any unverifiable claims.
   * For node1 -> node2 claims, the verification process also involves connecting
   * to node2 to verify the claim (to retrieve its signing public key).
   * @param targetNodeId Id of the node to connect request the chain data of.
   * @param claimId If set then we get the claims newer that this claim ID.
   * @param ctx
   */
  public requestChainData(
    targetNodeId: NodeId,
    claimId?: ClaimId,
    ctx?: Partial<ContextTimed>,
  ): PromiseCancellable<Record<ClaimId, SignedClaim>>;
  @decorators.timedCancellable(true)
  public async requestChainData(
    targetNodeId: NodeId,
    claimId: ClaimId | undefined,
    @decorators.context ctx: ContextTimed,
  ): Promise<Record<ClaimId, SignedClaim>> {
    // Verify the node's chain with its own public key
    return await this.withConnF(targetNodeId, ctx, async (connection) => {
      const claims: Record<ClaimId, SignedClaim> = {};
      const client = connection.getClient();

      const claimIdEncoded: ClaimIdEncoded | undefined =
        claimId != null ? claimsUtils.encodeClaimId(claimId) : undefined;
      for await (const agentClaim of await client.methods.nodesClaimsGet(
        {
          seek: claimIdEncoded,
        },
        ctx,
      )) {
        ctx.signal.throwIfAborted();
        // Need to re-construct each claim
        const claimId: ClaimId = claimsUtils.decodeClaimId(
          agentClaim.claimIdEncoded,
        )!;
        const signedClaimEncoded = agentClaim.signedTokenEncoded;
        const signedClaim = claimsUtils.parseSignedClaim(signedClaimEncoded);
        // Verifying the claim
        const issPublicKey = keysUtils.publicKeyFromNodeId(
          nodesUtils.decodeNodeId(signedClaim.payload.iss)!,
        );
        const subPublicKey =
          signedClaim.payload.typ === 'node'
            ? keysUtils.publicKeyFromNodeId(
                nodesUtils.decodeNodeId(signedClaim.payload.iss)!,
              )
            : null;
        const token = Token.fromSigned(signedClaim);
        if (!token.verifyWithPublicKey(issPublicKey)) {
          this.logger.warn('Failed to verify issuing node');
          continue;
        }
        if (subPublicKey != null && !token.verifyWithPublicKey(subPublicKey)) {
          this.logger.warn('Failed to verify subject node');
          continue;
        }
        claims[claimId] = signedClaim;
      }
      return claims;
    });
  }

  /**
   * Call this function upon receiving a "claim node request" notification from
   * another node.
   */
  public claimNode(
    targetNodeId: NodeId,
    tran?: DBTransaction,
    ctx?: Partial<ContextTimedInput>,
  ): PromiseCancellable<void>;
  @startStop.ready(new nodesErrors.ErrorNodeManagerNotRunning())
  public async claimNode(
    targetNodeId: NodeId,
    tran: DBTransaction | undefined,
    @decorators.context ctx: ContextTimed,
  ): Promise<void> {
    if (tran == null) {
      return await this.db.withTransactionF(async (tran) => {
        return await this.claimNode(targetNodeId, tran);
      });
    }
    const [, claim] = await this.sigchain.addClaim(
      {
        typ: 'ClaimLinkNode',
        iss: nodesUtils.encodeNodeId(this.keyRing.getNodeId()),
        sub: nodesUtils.encodeNodeId(targetNodeId),
      },
      undefined,
      async (token) => {
        return this.withConnF(targetNodeId, ctx, async (conn) => {
          // 2. create the agentClaim message to send
          const halfSignedClaim = token.toSigned();
          const halfSignedClaimEncoded =
            claimsUtils.generateSignedClaim(halfSignedClaim);
          const client = conn.getClient();
          const stream = await client.methods.nodesCrossSignClaim();
          const writer = stream.writable.getWriter();
          const reader = stream.readable.getReader();
          let fullySignedToken: Token<Claim>;
          try {
            await writer.write({
              signedTokenEncoded: halfSignedClaimEncoded,
            });
            // 3. We expect to receive the doubly signed claim
            const readStatus = await reader.read();
            if (readStatus.done) {
              throw new claimsErrors.ErrorEmptyStream();
            }
            const receivedClaim = readStatus.value;
            // We need to re-construct the token from the message
            const signedClaim = claimsUtils.parseSignedClaim(
              receivedClaim.signedTokenEncoded,
            );
            fullySignedToken = Token.fromSigned(signedClaim);
            // Check that the signatures are correct
            const targetNodePublicKey =
              keysUtils.publicKeyFromNodeId(targetNodeId);
            if (
              !fullySignedToken.verifyWithPublicKey(
                this.keyRing.keyPair.publicKey,
              ) ||
              !fullySignedToken.verifyWithPublicKey(targetNodePublicKey)
            ) {
              throw new claimsErrors.ErrorDoublySignedClaimVerificationFailed();
            }

            // Next stage is to process the claim for the other node
            const readStatus2 = await reader.read();
            if (readStatus2.done) {
              throw new claimsErrors.ErrorEmptyStream();
            }
            const receivedClaimRemote = readStatus2.value;
            // We need to re-construct the token from the message
            const signedClaimRemote = claimsUtils.parseSignedClaim(
              receivedClaimRemote.signedTokenEncoded,
            );
            // This is a singly signed claim,
            // we want to verify it before signing and sending back
            const signedTokenRemote = Token.fromSigned(signedClaimRemote);
            if (!signedTokenRemote.verifyWithPublicKey(targetNodePublicKey)) {
              throw new claimsErrors.ErrorSinglySignedClaimVerificationFailed();
            }
            signedTokenRemote.signWithPrivateKey(this.keyRing.keyPair);
            // 4. X <- responds with double signing the X signed claim <- Y
            const agentClaimedMessageRemote = claimsUtils.generateSignedClaim(
              signedTokenRemote.toSigned(),
            );
            await writer.write({
              signedTokenEncoded: agentClaimedMessageRemote,
            });

            // Check the stream is closed (should be closed by other side)
            const finalResponse = await reader.read();
            if (finalResponse.done != null) {
              await writer.close();
            }
          } catch (e) {
            await writer.abort(e);
            throw e;
          }
          return fullySignedToken;
        });
      },
      tran,
    );
    // With the claim created we want to add it to the gestalt graph
    const issNodeInfo = {
      nodeId: this.keyRing.getNodeId(),
    };
    const subNodeInfo = {
      nodeId: targetNodeId,
    };
    await this.gestaltGraph.linkNodeAndNode(issNodeInfo, subNodeInfo, {
      claim: claim as SignedClaim<ClaimLinkNode>,
      meta: {},
    });
  }

  public async *handleClaimNode(
    requestingNodeId: NodeId,
    input: AsyncIterableIterator<AgentRPCRequestParams<AgentClaimMessage>>,
    tran?: DBTransaction,
  ): AsyncGenerator<AgentRPCResponseResult<AgentClaimMessage>> {
    if (tran == null) {
      return yield* this.db.withTransactionG((tran) =>
        this.handleClaimNode(requestingNodeId, input, tran),
      );
    }
    const readStatus = await input.next();
    // If nothing to read, end and destroy
    if (readStatus.done) {
      throw new claimsErrors.ErrorEmptyStream();
    }
    const receivedMessage = readStatus.value;
    const signedClaim = claimsUtils.parseSignedClaim(
      receivedMessage.signedTokenEncoded,
    );
    const token = Token.fromSigned(signedClaim);
    // Verify if the token is signed
    if (
      !token.verifyWithPublicKey(
        keysUtils.publicKeyFromNodeId(requestingNodeId),
      )
    ) {
      throw new claimsErrors.ErrorSinglySignedClaimVerificationFailed();
    }
    // If verified, add your own signature to the received claim
    token.signWithPrivateKey(this.keyRing.keyPair);
    // Return the signed claim
    const doublySignedClaim = token.toSigned();
    const halfSignedClaimEncoded =
      claimsUtils.generateSignedClaim(doublySignedClaim);
    yield {
      signedTokenEncoded: halfSignedClaimEncoded,
    };

    // Now we want to send our own claim signed
    const { p: halfSignedClaimP, resolveP: halfSignedClaimResolveP } =
      utils.promise<SignedTokenEncoded>();
    const claimP = this.sigchain.addClaim(
      {
        typ: 'ClaimLinkNode',
        iss: nodesUtils.encodeNodeId(requestingNodeId),
        sub: nodesUtils.encodeNodeId(this.keyRing.getNodeId()),
      },
      undefined,
      async (token) => {
        const halfSignedClaim = token.toSigned();
        const halfSignedClaimEncoded =
          claimsUtils.generateSignedClaim(halfSignedClaim);
        halfSignedClaimResolveP(halfSignedClaimEncoded);
        const readStatus = await input.next();
        if (readStatus.done) {
          throw new claimsErrors.ErrorEmptyStream();
        }
        const receivedClaim = readStatus.value;
        // We need to re-construct the token from the message
        const signedClaim = claimsUtils.parseSignedClaim(
          receivedClaim.signedTokenEncoded,
        );
        const fullySignedToken = Token.fromSigned(signedClaim);
        // Check that the signatures are correct
        const requestingNodePublicKey =
          keysUtils.publicKeyFromNodeId(requestingNodeId);
        if (
          !fullySignedToken.verifyWithPublicKey(
            this.keyRing.keyPair.publicKey,
          ) ||
          !fullySignedToken.verifyWithPublicKey(requestingNodePublicKey)
        ) {
          throw new claimsErrors.ErrorDoublySignedClaimVerificationFailed();
        }
        // Ending the stream
        return fullySignedToken;
      },
    );
    void claimP.catch(() => {});
    yield {
      signedTokenEncoded: await halfSignedClaimP,
    };
    const [, claim] = await claimP;
    // With the claim created we want to add it to the gestalt graph
    const issNodeInfo = {
      nodeId: requestingNodeId,
    };
    const subNodeInfo = {
      nodeId: this.keyRing.getNodeId(),
    };
    await this.gestaltGraph.linkNodeAndNode(issNodeInfo, subNodeInfo, {
      claim: claim as SignedClaim<ClaimLinkNode>,
      meta: {},
    });
  }

  /**
   * Creates a claim on the sigchain granting this node authority over a network to create `ClaimNetworkAccess` claims.
   *
   * @param networkNodeId - The public key NodeId for the root authority for the network
   * @param network - The network URL.
   * @param isPrivate - Indicates if the network is private or not.
   * @param signingHook - A callback used to sign the claim with the network's private key.
   * @param tran
   */
  public async createClaimNetworkAuthority(
    networkNodeId: NodeId,
    network: string,
    isPrivate: boolean,
    signingHook: (token: Token<ClaimNetworkAuthority>) => Promise<Token<Claim>>,
    tran?: DBTransaction,
  ): Promise<[ClaimId, Token<ClaimNetworkAuthority>]> {
    if (tran == null) {
      return await this.db.withTransactionF((tran) =>
        this.createClaimNetworkAuthority(
          networkNodeId,
          network,
          isPrivate,
          signingHook,
          tran,
        ),
      );
    }

    const [claimId, signedClaim] = await this.sigchain.addClaim(
      {
        typ: 'ClaimNetworkAuthority',
        sub: nodesUtils.encodeNodeId(this.keyRing.getNodeId()),
        iss: nodesUtils.encodeNodeId(networkNodeId),
        network,
        isPrivate,
      },
      undefined,
      signingHook,
    );
    const token = Token.fromSigned<ClaimNetworkAuthority>(
      signedClaim as SignedClaim<ClaimNetworkAuthority>,
    );
    this.claimNetworkAuthority = token;
    await this.setClaimNetworkAuthority(token);
    await this.switchNetwork(network);
    return [claimId, token];
  }

  public async createSelfSignedClaimNetworkAccess(
    claimNetworkAuthority: Token<ClaimNetworkAuthority>,
  ): Promise<[ClaimId, Token<ClaimNetworkAccess>]> {
    const thisNodeId = this.keyRing.getNodeId();
    const encodedNetworkAuthority = claimsUtils.generateSignedClaim(
      claimNetworkAuthority.toSigned(),
    );
    if (
      claimNetworkAuthority.payload.sub !== nodesUtils.encodeNodeId(thisNodeId)
    ) {
      throw new claimsErrors.ErrorClaimsVerificationFailed(
        'ClaimNetworkAuthority does not grant authority to this node',
      );
    }
    const network = claimNetworkAuthority.payload.network;
    const isPrivate = claimNetworkAuthority.payload.isPrivate;
    const [claimId, signedClaim] = await this.sigchain.addClaim({
      typ: 'ClaimNetworkAccess',
      iss: nodesUtils.encodeNodeId(thisNodeId),
      sub: nodesUtils.encodeNodeId(thisNodeId),
      network,
      isPrivate,
      signedClaimNetworkAuthorityEncoded: encodedNetworkAuthority,
    });
    const token = Token.fromSigned<ClaimNetworkAccess>(
      signedClaim as SignedClaim<ClaimNetworkAccess>,
    );
    this.claimNetworkAccess = token;
    await this.setClaimNetworkAccess(token);
    await this.switchNetwork(network);
    return [claimId, token];
  }

  /**
   * This takes a `ClaimNetworkAuthority` and tracks it in the database under the network name.
   */
  protected async setClaimNetworkAuthority(
    claimNetworkAuthority: Token<ClaimNetworkAuthority>,
    tran?: DBTransaction,
  ): Promise<void> {
    if (tran == null) {
      return await this.db.withTransactionF((tran) =>
        this.setClaimNetworkAuthority(claimNetworkAuthority, tran),
      );
    }

    const network = claimNetworkAuthority.payload.network;
    await tran.put(
      [...this.nodeManagerClaimNetworkAuthorityPath, network],
      claimNetworkAuthority.payload.jti,
      false,
    );
  }

  /**
   * This returns the `ClaimNetworkAuthority` for the given network.
   */
  protected async getClaimNetworkAuthority(
    network: string,
    tran?: DBTransaction,
  ): Promise<Token<ClaimNetworkAuthority> | undefined> {
    if (tran == null) {
      return await this.db.withTransactionF((tran) =>
        this.getClaimNetworkAuthority(network, tran),
      );
    }

    const jti = await tran.get<ClaimIdEncoded>(
      [...this.nodeManagerClaimNetworkAuthorityPath, network],
      false,
    );
    if (jti == null) return;
    const claim = await this.sigchain.getSignedClaim(
      claimsUtils.decodeClaimId(jti)!,
      tran,
    );
    if (claim == null) return;
    const token = Token.fromSigned(claim);
    claimNetworkAuthorityUtils.assertClaimNetworkAuthority(token.payload);
    return token as Token<ClaimNetworkAuthority>;
  }

  /**
   * This takes a `ClaimNetworkAccess` and tracks it in the database under the network name.
   */
  protected async setClaimNetworkAccess(
    claimNetworkAccess: Token<ClaimNetworkAccess>,
    tran?: DBTransaction,
  ): Promise<void> {
    if (tran == null) {
      return await this.db.withTransactionF((tran) =>
        this.setClaimNetworkAccess(claimNetworkAccess, tran),
      );
    }

    const network = claimNetworkAccess.payload.network;
    await tran.put(
      [...this.nodeManagerClaimNetworkAccessPath, network],
      claimNetworkAccess.payload.jti,
      false,
    );
  }

  /**
   * This returns the `ClaimNetworkAccess` for the given network.
   */
  public async getClaimNetworkAccess(
    network: string,
    tran?: DBTransaction,
  ): Promise<Token<ClaimNetworkAccess> | undefined> {
    if (tran == null) {
      return await this.db.withTransactionF((tran) =>
        this.getClaimNetworkAccess(network, tran),
      );
    }

    const jti = await tran.get<ClaimIdEncoded>(
      [...this.nodeManagerClaimNetworkAccessPath, network],
      false,
    );
    if (jti == null) return;
    const claim = await this.sigchain.getSignedClaim(
      claimsUtils.decodeClaimId(jti)!,
      tran,
    );
    if (claim == null) return;
    const token = Token.fromSigned(claim);
    claimNetworkAccessUtils.assertClaimNetworkAccess(token.payload);
    return token as Token<ClaimNetworkAccess>;
  }

  /**
   * This switches out the active `ClaimNetworkAuthority` and `ClaimNetworkAccess` for the desired network.
   * If no claims exist for the network or no network is provided, then it switches to using no network.
   * In doing so this also updates the `NodeConnectionManager`'s authentication callbacks to use the selected
   * network for authentication.
   * @param network - The Network URL for the desired network to switch to.
   * @param tran
   */
  public async switchNetwork(
    network?: string,
    tran?: DBTransaction,
  ): Promise<void> {
    if (tran == null) {
      return await this.db.withTransactionF((tran) =>
        this.switchNetwork(network, tran),
      );
    }

    if (network == null) {
      this.claimNetworkAuthority = undefined;
      this.claimNetworkAccess = undefined;
      // Use the basic no network behavior
      this.nodeConnectionManager.setAuthenticateNetworkForwardCallback(
        nodesUtils.nodesAuthenticateConnectionForwardDefault,
      );
      this.nodeConnectionManager.setAuthenticateNetworkReverseCallback(
        nodesUtils.nodesAuthenticateConnectionReverseDeny,
      );
      return;
    }

    this.claimNetworkAuthority = await this.getClaimNetworkAuthority(
      network,
      tran,
    );
    this.claimNetworkAccess = await this.getClaimNetworkAccess(network, tran);
    if (this.claimNetworkAccess != null) {
      // Use the claim to verify connections
      this.nodeConnectionManager.setAuthenticateNetworkForwardCallback(
        nodesUtils.nodesAuthenticationConnectionForwardPrivateFactory(
          this.claimNetworkAccess,
        ),
      );
      this.nodeConnectionManager.setAuthenticateNetworkReverseCallback(
        nodesUtils.nodesAuthenticationConnectionReversePrivateFactory(
          this.claimNetworkAccess,
        ),
      );
    } else {
      // Use the basic no network behavior
      this.nodeConnectionManager.setAuthenticateNetworkForwardCallback(
        nodesUtils.nodesAuthenticateConnectionForwardDefault,
      );
      this.nodeConnectionManager.setAuthenticateNetworkReverseCallback(
        nodesUtils.nodesAuthenticateConnectionReverseDeny,
      );
    }
  }

  /**
   * Quick hand utility for checking if the active `ClaimNetworkAuthority` is a private network
   */
  public isClaimNetworkAuthorityPrivate(): boolean | undefined {
    // Return undefined if we're not acting as a network authority
    if (this.claimNetworkAuthority == null) return;
    return this.claimNetworkAuthority.payload.isPrivate;
  }

  /**
   * This creates a cross-signed `ClaimNetworkAccess` on the sigchain. The resulting `ClaimNetworkAccess` is used to
   * authenticate connections between nodes within the network.
   *
   * @param targetNodeId - This is a node with an active `ClaimNetworkAuthority` for the network you wish to join.
   * This usually is a seed node for that network.
   * @param network - The URL of the network you wish to join.
   * @param tran
   * @param ctx
   */
  public async claimNetwork(
    targetNodeId: NodeId,
    network: string,
    tran?: DBTransaction,
    ctx?: ContextTimedInput,
  ): Promise<[ClaimId, Token<ClaimNetworkAccess>]>;
  @startStop.ready(new nodesErrors.ErrorNodeManagerNotRunning())
  public async claimNetwork(
    targetNodeId: NodeId,
    network: string,
    tran: DBTransaction | undefined,
    @decorators.context ctx: ContextTimed,
  ): Promise<[ClaimId, Token<ClaimNetworkAccess>]> {
    if (tran == null) {
      return await this.db.withTransactionF((tran) =>
        this.claimNetwork(targetNodeId, network, tran, ctx),
      );
    }

    // Validating that the ClaimNetworkAuthority is correct.
    const claimNetworkAuthority = await this.remoteClaimNetworkAuthorityGet(
      network,
      targetNodeId,
    );

    // Error out if a network access claim already exists
    if ((await this.getClaimNetworkAccess(network, tran)) != null) {
      throw new Error('TMP network access claim already exists');
    }

    const encodedNetworkAuthority = claimsUtils.generateSignedClaim(
      claimNetworkAuthority.toSigned(),
    );
    const networkNodeId = nodesUtils.decodeNodeId(
      claimNetworkAuthority.payload.iss,
    );
    if (networkNodeId == null) utils.never('failed to decode networkNodeId');

    const subjectNodeId = this.keyRing.getNodeId();
    const isPrivate = claimNetworkAuthority.payload.isPrivate;
    const [claimId, signedClaim] = await this.sigchain.addClaim(
      {
        typ: 'ClaimNetworkAccess',
        iss: nodesUtils.encodeNodeId(targetNodeId),
        sub: nodesUtils.encodeNodeId(subjectNodeId),
        network,
        isPrivate,
        signedClaimNetworkAuthorityEncoded: encodedNetworkAuthority,
      },
      undefined,
      async (token) => {
        // Using the nodeConnection.withConnF so we can use the connection without being authenticated
        return await withF(
          [this.nodeConnectionManager.acquireConnectionInternal(targetNodeId)],
          async ([conn]) => {
            // 2. create the agentClaim message to send
            const halfSignedClaim = token.toSigned();
            const halfSignedClaimEncoded =
              claimsUtils.generateSignedClaim(halfSignedClaim);
            const client = conn.getClient();
            const stream = await client.methods.nodesClaimNetworkSign();
            const writer = stream.writable.getWriter();
            const reader = stream.readable.getReader();
            let fullySignedToken: Token<ClaimNetworkAccess>;
            try {
              await writer.write({
                signedTokenEncoded: halfSignedClaimEncoded,
              });
              // 3. We expect to receive the doubly signed claim
              const readStatus = await reader.read();
              if (readStatus.done) {
                throw new claimsErrors.ErrorEmptyStream(
                  'nodesClaimNetworkSign stream ended too soon, likely due to lack of permission',
                );
              }
              const receivedClaim = readStatus.value;
              // We need to re-construct the token from the message
              const receivedClaimNetworkAccess =
                claimNetworkAccessUtils.parseSignedClaimNetworkAccess(
                  receivedClaim.signedTokenEncoded,
                );
              fullySignedToken = Token.fromSigned(receivedClaimNetworkAccess);
              claimNetworkAccessUtils.verifyClaimNetworkAccess(
                networkNodeId,
                subjectNodeId,
                network,
                fullySignedToken,
              );

              // Next stage is to process the claim for the other node
              const readStatus2 = await reader.read();
              if (readStatus2.done) {
                throw new claimsErrors.ErrorEmptyStream();
              }
              const receivedClaimRemote = readStatus2.value;

              // We need to re-construct the token from the message
              const signedClaimRemote =
                claimNetworkAccessUtils.parseSignedClaimNetworkAccess(
                  receivedClaimRemote.signedTokenEncoded,
                );
              // This is a singly signed claim,
              // we want to verify it before signing and sending back
              const signedTokenRemote = Token.fromSigned(signedClaimRemote);
              signedTokenRemote.signWithPrivateKey(this.keyRing.keyPair);
              // Verify everything is correct
              claimNetworkAccessUtils.verifyClaimNetworkAccess(
                networkNodeId,
                subjectNodeId,
                network,
                signedTokenRemote,
              );
              // 4. X <- responds with double signing the X signed claim <- Y
              const agentClaimedMessageRemote = claimsUtils.generateSignedClaim(
                signedTokenRemote.toSigned(),
              );
              await writer.write({
                signedTokenEncoded: agentClaimedMessageRemote,
              });

              // Check the stream is closed (should be closed by other side)
              const finalResponse = await reader.read();
              if (finalResponse.done != null) {
                await writer.close();
              }
            } catch (e) {
              await writer.abort(e);
              throw e;
            }
            return fullySignedToken;
          },
        );
      },
      tran,
    );
    const token = Token.fromSigned(
      signedClaim as SignedClaim<ClaimNetworkAccess>,
    );
    this.claimNetworkAccess = token;
    await this.setClaimNetworkAccess(token, tran);
    await this.switchNetwork(network, tran);
    return [claimId, token];
  }

  /**
   * This provides the handler side of the ClaimNetwork logic.
   * @param requestingNodeId - The nodeId of the node making the request. This should be taken from the connections
   * certificate to confirm that `ClaimNetworkAccess` is being created for the node requesting it.
   * @param input - The input stream for the RPC handler.
   * @param tran
   */
  public async *handleClaimNetwork(
    requestingNodeId: NodeId,
    input: AsyncIterableIterator<AgentRPCRequestParams<AgentClaimMessage>>,
    tran?: DBTransaction,
  ): AsyncGenerator<AgentRPCResponseResult<AgentClaimMessage>> {
    if (tran == null) {
      return yield* this.db.withTransactionG((tran) =>
        this.handleClaimNetwork(requestingNodeId, input, tran),
      );
    }
    if (this.claimNetworkAuthority == null) {
      throw new nodesErrors.ErrorNodeManagerClaimNetworkAuthorityMissing(
        'Node is not acting as a network authority and can not create a claimNetworkAccess',
      );
    }

    const readStatus = await input.next();
    // If nothing to read, end and destroy
    if (readStatus.done) {
      throw new claimsErrors.ErrorEmptyStream();
    }
    const receivedMessage = readStatus.value;
    const signedClaim = claimsUtils.parseSignedClaim(
      receivedMessage.signedTokenEncoded,
    );
    const token = Token.fromSigned(signedClaim);
    // Verify if token is for our network
    if (
      token.payload.network == null ||
      token.payload.network !== this.claimNetworkAuthority.payload.network
    ) {
      throw new claimsErrors.ErrorClaimsVerificationFailed(
        'Claim does not match expected network',
      );
    }
    // Verify if the token is signed
    if (
      !token.verifyWithPublicKey(
        keysUtils.publicKeyFromNodeId(requestingNodeId),
      )
    ) {
      throw new claimsErrors.ErrorSinglySignedClaimVerificationFailed();
    }
    // If verified, add your own signature to the received claim
    token.signWithPrivateKey(this.keyRing.keyPair);
    // Return the signed claim
    const doublySignedClaim = token.toSigned();
    const halfSignedClaimEncoded =
      claimsUtils.generateSignedClaim(doublySignedClaim);
    yield {
      signedTokenEncoded: halfSignedClaimEncoded,
    };

    // Now we want to send our own claim signed
    const { p: halfSignedClaimP, resolveP: halfSignedClaimResolveP } =
      utils.promise<SignedTokenEncoded>();
    const claimP = this.sigchain.addClaim(
      {
        typ: 'ClaimNetworkAccess',
        iss: nodesUtils.encodeNodeId(this.keyRing.getNodeId()),
        sub: nodesUtils.encodeNodeId(requestingNodeId),
        network: this.claimNetworkAuthority.payload.network,
        isPrivate: this.claimNetworkAuthority.payload.isPrivate,
        signedClaimNetworkAuthorityEncoded: claimsUtils.generateSignedClaim(
          this.claimNetworkAuthority.toSigned(),
        ),
      },
      undefined,
      async (token) => {
        const halfSignedClaim = token.toSigned();
        const halfSignedClaimEncoded =
          claimsUtils.generateSignedClaim(halfSignedClaim);
        halfSignedClaimResolveP(halfSignedClaimEncoded);
        const readStatus = await input.next();
        if (readStatus.done) {
          throw new claimsErrors.ErrorEmptyStream();
        }
        const receivedClaim = readStatus.value;
        // We need to re-construct the token from the message
        const signedClaim =
          claimNetworkAccessUtils.parseSignedClaimNetworkAccess(
            receivedClaim.signedTokenEncoded,
          );
        const fullySignedToken = Token.fromSigned(signedClaim);
        // Check that the signatures are correct
        const networkNodeId = nodesUtils.decodeNodeId(
          this.claimNetworkAuthority!.payload.iss,
        );
        if (networkNodeId == null) {
          utils.never('failed to decode networkNodeId');
        }

        claimNetworkAccessUtils.verifyClaimNetworkAccess(
          networkNodeId,
          requestingNodeId,
          this.claimNetworkAuthority!.payload.network,
          fullySignedToken,
        );
        // Ending the stream
        return fullySignedToken;
      },
    );
    // Prevent async promise handling leak
    void claimP.catch(() => {});
    yield {
      signedTokenEncoded: await halfSignedClaimP,
    };
    const [, claim] = await claimP;
    // With the claim created we want to add it to the gestalt graph
    const issNodeInfo = {
      nodeId: requestingNodeId,
    };
    const subNodeInfo = {
      nodeId: this.keyRing.getNodeId(),
    };
    await this.gestaltGraph.linkNodeAndNode(issNodeInfo, subNodeInfo, {
      claim: claim as SignedClaim<ClaimLinkNode>,
      meta: {},
    });
  }

  /**
   * Gets the `ClaimNetworkAuthority` from the target node. It also verifies its valid and for the expected network.
   * @param network
   * @param targetNodeId
   */
  public async remoteClaimNetworkAuthorityGet(
    network: string,
    targetNodeId: NodeId,
  ): Promise<Token<ClaimNetworkAuthority>> {
    return await withF(
      [this.nodeConnectionManager.acquireConnectionInternal(targetNodeId)],
      async ([conn]) => {
        const client = conn.getClient();
        const receivedClaim =
          await client.methods.nodesClaimNetworkAuthorityGet({});
        const signedClaim =
          claimNetworkAuthorityUtils.parseSignedClaimNetworkAuthority(
            receivedClaim,
          );
        const token = Token.fromSigned(signedClaim);
        const networkNodeId = nodesUtils.decodeNodeId(token.payload.iss);
        if (networkNodeId == null) {
          utils.never('failed to decode networkNodeId');
        }
        claimNetworkAuthorityUtils.verifyClaimNetworkAuthority(
          networkNodeId,
          targetNodeId,
          network,
          token,
        );
        return token;
      },
    );
  }

  /**
   * The handler side logic for `remoteClaimNetworkAuthorityGet`.
   */
  public async handleClaimNetworkAuthorityGet(): Promise<SignedTokenEncoded> {
    if (this.claimNetworkAuthority == null) {
      throw new nodesErrors.ErrorNodeManagerClaimNetworkAuthorityMissing();
    }
    return claimsUtils.generateSignedClaim(
      this.claimNetworkAuthority.toSigned(),
    );
  }

  /**
   * Adds a node to the node graph. This assumes that you have already authenticated the node
   * Updates the node if the node already exists
   * This operation is blocking by default - set `block` to false to make it non-blocking
   * @param nodeId - ID of the node we wish to add
   * @param nodeAddress - Expected address of the node we want to add
   * @param nodeContactAddressData
   * @param block - When true it will wait for any garbage collection to finish before returning.
   * @param force - Flag for if we want to add the node without authenticating or if the bucket is full.
   * This will drop the oldest node in favor of the new.
   * @param connectionConnectTimeoutTime - Timeout for each ping operation during garbage collection.
   * @param ctx
   * @param tran
   */
  public setNode(
    nodeId: NodeId,
    nodeAddress: NodeAddress,
    nodeContactAddressData: NodeContactAddressData,
    block?: boolean,
    force?: boolean,
    connectionConnectTimeoutTime?: number,
    tran?: DBTransaction,
    ctx?: Partial<ContextTimed>,
  ): PromiseCancellable<void>;
  @startStop.ready(new nodesErrors.ErrorNodeManagerNotRunning(), true, [
    'stopping',
  ])
  @decorators.timedCancellable(true)
  public async setNode(
    nodeId: NodeId,
    nodeAddress: NodeAddress,
    nodeContactAddressData: NodeContactAddressData,
    block: boolean = false,
    force: boolean = false,
    connectionConnectTimeoutTime: number = this.connectionConnectTimeoutTime,
    tran: DBTransaction,
    @decorators.context ctx: ContextTimed,
  ): Promise<void> {
    // We don't want to add our own node
    if (nodeId.equals(this.keyRing.getNodeId())) {
      this.logger.debug('Is own NodeId, skipping');
      return;
    }

    if (tran == null) {
      return await this.db.withTransactionF(
        async (tran) =>
          await this.setNode(
            nodeId,
            nodeAddress,
            nodeContactAddressData,
            block,
            force,
            connectionConnectTimeoutTime,
            tran,
            ctx,
          ),
      );
    }

    // Need to await node connection verification, if failed, need to reject connection.

    // When adding a node we need to handle 3 cases
    // 1. The node already exists. We need to update it's last updated field
    // 2. The node doesn't exist and bucket has room. We need to add the node
    //    to the bucket
    // 3. The node doesn't exist and the bucket is full. We need to ping the
    //    oldest node. If the ping succeeds we need to update the lastUpdated of
    //    the oldest node and drop the new one. If the ping fails we delete the
    //    old node and add in the new one.
    const [bucketIndex] = this.nodeGraph.bucketIndex(nodeId);
    // To avoid conflict we want to lock on the bucket index
    await this.nodeGraph.lockBucket(bucketIndex, tran, ctx);

    const nodeContact = await this.nodeGraph.getNodeContact(nodeId, tran);
    // If this is a new entry, check the bucket limit
    const count = await this.nodeGraph.getBucketMetaProp(
      bucketIndex,
      'count',
      tran,
    );

    if (nodeContact != null || count < this.nodeGraph.nodeBucketLimit) {
      // Either already exists or has room in the bucket
      // We want to add or update the node
      await this.nodeGraph.setNodeContactAddressData(
        nodeId,
        nodeAddress,
        nodeContactAddressData,
        tran,
      );
      // Updating the refreshBucket timer
      await this.updateRefreshBucketDelay(
        bucketIndex,
        this.refreshBucketDelayTime,
        true,
        tran,
      );
    } else {
      // We want to add a node but the bucket is full
      if (force) {
        // We just add the new node anyway without checking the old one
        const bucket = await this.nodeGraph.getBucket(
          bucketIndex,
          'connected',
          'asc',
          1,
          tran,
        );
        const oldNodeId = bucket[0]?.[0];
        if (oldNodeId == null) {
          utils.never('bucket should be full in this case');
        }
        this.logger.debug(
          `Force was set, removing ${nodesUtils.encodeNodeId(
            oldNodeId,
          )} and adding ${nodesUtils.encodeNodeId(nodeId)}`,
        );
        await this.nodeGraph.unsetNodeContact(oldNodeId, tran);
        await this.nodeGraph.setNodeContactAddressData(
          nodeId,
          nodeAddress,
          nodeContactAddressData,
          tran,
        );
        // Updating the refreshBucket timer
        await this.updateRefreshBucketDelay(
          bucketIndex,
          this.refreshBucketDelayTime,
          true,
          tran,
        );
        return;
      }
      this.logger.debug(
        `Bucket was full, adding ${nodesUtils.encodeNodeId(
          nodeId,
        )} to pending list`,
      );
      // Add the node to the pending nodes list
      await this.addPendingNode(
        bucketIndex,
        nodeId,
        nodeAddress,
        nodeContactAddressData,
        block,
        connectionConnectTimeoutTime,
        ctx,
        tran,
      );
    }
  }

  /**
   * Removes a node from the NodeGraph
   */
  public async unsetNode(nodeId: NodeId, tran: DBTransaction): Promise<void> {
    return await this.nodeGraph.unsetNodeContact(nodeId, tran);
  }

  /**
   * Gets the specified bucket from the NodeGraph
   */
  public async getBucket(
    bucketIndex: number,
    tran?: DBTransaction,
  ): Promise<NodeBucket | undefined> {
    return await this.nodeGraph.getBucket(
      bucketIndex,
      undefined,
      undefined,
      undefined,
      tran,
    );
  }

  protected garbageCollectBucket(
    bucketIndex: number,
    connectionConnectTimeoutTime?: number,
    ctx?: Partial<ContextTimed>,
    tran?: DBTransaction,
  ): PromiseCancellable<void>;
  @decorators.timedCancellable(true)
  protected async garbageCollectBucket(
    bucketIndex: number,
    connectionConnectTimeoutTime: number = this.connectionConnectTimeoutTime,
    @decorators.context ctx: ContextTimed,
    tran?: DBTransaction,
  ): Promise<void> {
    if (tran == null) {
      return await this.db.withTransactionF(
        async (tran) =>
          await this.garbageCollectBucket(
            bucketIndex,
            connectionConnectTimeoutTime,
            ctx,
            tran,
          ),
      );
    }

    // This needs to:
    //  1. Iterate over every node within the bucket pinging K at a time
    //  2. remove any un-responsive nodes until there is room of all pending
    //    or run out of existing nodes
    //  3. fill in the bucket with pending nodes until full
    //  4. throw out remaining pending nodes

    const pendingNodes = this.pendingNodes.get(bucketIndex);
    // No nodes mean nothing to do
    if (pendingNodes == null || pendingNodes.size === 0) return;
    this.pendingNodes.set(bucketIndex, new Map());
    // Locking on bucket
    await this.nodeGraph.lockBucket(bucketIndex, tran, ctx);
    const semaphore = new Semaphore(this.concurrencyLimit);
    // Iterating over existing nodes
    const bucket = await this.nodeGraph.getBucket(
      bucketIndex,
      'connected',
      'asc',
      this.nodeGraph.nodeBucketLimit,
      tran,
    );
    if (bucket == null) utils.never('bucket should exist');
    let removedNodes = 0;
    const unsetLock = new Lock();
    const pendingPromises: Array<Promise<void>> = [];
    for (const [nodeId, nodeContact] of bucket) {
      if (removedNodes >= pendingNodes.size) break;
      await semaphore.waitForUnlock(ctx);
      if (ctx.signal?.aborted === true) break;
      const [semaphoreReleaser] = await semaphore.lock(ctx)();
      pendingPromises.push(
        (async () => {
          // Ping and remove or update node in bucket
          const pingCtx = {
            signal: ctx.signal,
            timer: connectionConnectTimeoutTime,
          };
          // Getting known addresses for the ping
          const desiredAddresses: Array<NodeAddress> = [];
          for (const [
            nodeContactAddress,
            nodeContactAddressData,
          ] of Object.entries(nodeContact)) {
            if (nodeContactAddressData.mode === 'direct') {
              desiredAddresses.push(
                nodesUtils.parseNodeContactAddress(nodeContactAddress),
              );
            }
          }

          const resolvedAddresses = await networkUtils.resolveHostnames(
            desiredAddresses,
            undefined,
            this.dnsServers,
            ctx,
          );

          const pingResult = await this.pingNodeAddressMultiple(
            nodeId,
            resolvedAddresses,
            pingCtx,
          );

          // If ping fails we remove it, otherwise we don't update
          if (!pingResult) {
            // We don't remove node the ping was aborted
            if (ctx.signal.aborted) return;
            // We need to lock this since it's concurrent
            //  and shares the transaction
            await unsetLock.withF(ctx, async () => {
              await this.unsetNode(nodeId, tran);
              removedNodes += 1;
            });
          }
        })()
          // Clean ensure semaphore is released
          .finally(async () => await semaphoreReleaser()),
      );
    }
    // Wait for pending pings to complete
    await Promise.all(pendingPromises);
    // Fill in bucket with pending nodes
    for (const [
      nodeIdString,
      [address, nodeContactAddressData],
    ] of pendingNodes) {
      if (removedNodes <= 0) break;
      const nodeId = IdInternal.fromString<NodeId>(nodeIdString);
      await this.setNode(
        nodeId,
        address,
        nodeContactAddressData,
        false,
        false,
        undefined,
        tran,
        ctx,
      );
      removedNodes -= 1;
    }
  }

  protected async addPendingNode(
    bucketIndex: number,
    nodeId: NodeId,
    nodeAddress: NodeAddress,
    nodeContactAddressData: NodeContactAddressData,
    block: boolean = false,
    connectionConnectTimeoutTime: number = this.connectionConnectTimeoutTime,
    ctx: ContextTimed,
    tran?: DBTransaction,
  ): Promise<void> {
    if (!this.pendingNodes.has(bucketIndex)) {
      this.pendingNodes.set(bucketIndex, new Map());
    }
    const pendingNodes = this.pendingNodes.get(bucketIndex);
    pendingNodes!.set(nodeId.toString(), [nodeAddress, nodeContactAddressData]);
    // No need to re-set it in the map, Maps are by reference

    // If set to blocking we just run the GC operation here
    //  without setting up a new task
    if (block) {
      await this.garbageCollectBucket(
        bucketIndex,
        connectionConnectTimeoutTime,
        ctx,
        tran,
      );
      return;
    }
    await this.setupGCTask(bucketIndex);
  }

  protected async setupGCTask(bucketIndex: number) {
    // Check and start a `garbageCollect` bucket task
    let scheduled: boolean = false;
    for await (const task of this.taskManager.getTasks('asc', true, [
      this.tasksPath,
      this.gcBucketHandlerId,
      `${bucketIndex}`,
    ])) {
      switch (task.status) {
        case 'queued':
        case 'active':
          // Ignore active tasks
          break;
        case 'scheduled':
          {
            if (scheduled) {
              // Duplicate scheduled are removed
              task.cancel(abortSingletonTaskReason);
              break;
            }
            scheduled = true;
          }
          break;
        default:
          task.cancel(abortSingletonTaskReason);
          break;
      }
    }
    if (!scheduled) {
      // If none were found, schedule a new one
      await this.taskManager.scheduleTask({
        handlerId: this.gcBucketHandlerId,
        parameters: [bucketIndex],
        path: [this.tasksPath, this.gcBucketHandlerId, `${bucketIndex}`],
        lazy: true,
      });
    }
  }

  /**
   * To be called on key renewal. Re-orders all nodes in all buckets with respect
   * to the new node ID.
   */
  public async resetBuckets(): Promise<void> {
    return await this.nodeGraph.resetBuckets();
  }

  /**
   * Kademlia refresh bucket operation.
   * It generates a random node ID within the range of a bucket and does a
   * lookup for that node in the network. This will cause the network to update
   * its node graph information.
   * @param bucketIndex
   * @param connectionConnectTimeoutTime
   * @param ctx
   */
  public refreshBucket(
    bucketIndex: NodeBucketIndex,
    connectionConnectTimeoutTime?: number,
    ctx?: Partial<ContextTimed>,
  ): PromiseCancellable<void>;
  @decorators.timedCancellable(true)
  public async refreshBucket(
    bucketIndex: NodeBucketIndex,
    connectionConnectTimeoutTime: number | undefined = this
      .connectionConnectTimeoutTime,
    @decorators.context ctx: ContextTimed,
  ): Promise<void> {
    // We need to generate a random nodeId for this bucket
    const nodeId = this.keyRing.getNodeId();
    const bucketRandomNodeId = nodesUtils.generateRandomNodeIdForBucket(
      nodeId,
      bucketIndex,
    );
    // We then need to start a findNode procedure
    await this.findNode(
      {
        nodeId: bucketRandomNodeId,
        connectionConnectTimeoutTime: connectionConnectTimeoutTime,
      },
      ctx,
    );
  }

  protected async setupRefreshBucketTasks(tran?: DBTransaction) {
    if (tran == null) {
      return await this.db.withTransactionF(
        async (tran) => await this.setupRefreshBucketTasks(tran),
      );
    }

    this.logger.info('Setting up refreshBucket tasks');
    // 1. Iterate over existing tasks and reset the delay
    const existingTasks: Array<boolean> = new Array(this.nodeGraph.nodeIdBits);
    for await (const task of this.taskManager.getTasks(
      'asc',
      true,
      [this.tasksPath, this.refreshBucketHandlerId],
      tran,
    )) {
      const bucketIndex = parseInt(task.path[0]);
      switch (task.status) {
        case 'scheduled':
          {
            // If it's scheduled then reset delay
            existingTasks[bucketIndex] = true;
            // Total delay is refreshBucketDelay + time since task creation
            const delay =
              performance.now() +
              performance.timeOrigin -
              task.created.getTime() +
              this.refreshBucketDelayTime +
              nodesUtils.refreshBucketsDelayJitter(
                this.refreshBucketDelayTime,
                this.refreshBucketDelayJitter,
              );
            await this.taskManager.updateTask(task.id, { delay }, tran);
          }
          break;
        case 'queued':
        case 'active':
          // If it's running then leave it
          existingTasks[bucketIndex] = true;
          break;
        default:
          // Otherwise, ignore it, should be re-created
          existingTasks[bucketIndex] = false;
      }
    }

    // 2. Recreate any missing tasks for buckets
    for (
      let bucketIndex = 0;
      bucketIndex < existingTasks.length;
      bucketIndex++
    ) {
      const exists = existingTasks[bucketIndex];
      // Can be disabled with 0 delay, only use for testing
      if (!exists && this.refreshBucketDelayTime > 0) {
        // Create a new task
        this.logger.debug(
          `Creating refreshBucket task for bucket ${bucketIndex}`,
        );
        const jitter = nodesUtils.refreshBucketsDelayJitter(
          this.refreshBucketDelayTime,
          this.refreshBucketDelayJitter,
        );
        await this.taskManager.scheduleTask({
          handlerId: this.refreshBucketHandlerId,
          delay: this.refreshBucketDelayTime + jitter,
          lazy: true,
          parameters: [bucketIndex],
          path: [this.tasksPath, this.refreshBucketHandlerId, `${bucketIndex}`],
          priority: 0,
        });
      }
    }
    this.logger.info('Set up refreshBucket tasks');
  }

  /**
   * Cancels all NodeManager tasks.
   * These are ephemeral and will be recreated next time we start.
   */
  protected async stopTasks() {
    const taskPs: Array<Promise<any>> = [];
    for await (const task of this.taskManager.getTasks(undefined, false, [
      this.tasksPath,
    ])) {
      taskPs.push(task.promise());
      task.cancel(abortEphemeralTaskReason);
    }
    await Promise.allSettled(taskPs);
  }

  public async updateRefreshBucketDelay(
    bucketIndex: number,
    delay?: number,
    lazy?: boolean,
    tran?: DBTransaction,
    ctx?: Partial<ContextTimedInput>,
  ): Promise<Task>;
  @startStop.ready(new nodesErrors.ErrorNodeManagerNotRunning(), true, [
    'stopping',
  ])
  @decorators.timedCancellable(true)
  public async updateRefreshBucketDelay(
    bucketIndex: number,
    delay: number = this.refreshBucketDelayTime,
    lazy: boolean = true,
    tran: DBTransaction | undefined,
    @decorators.context ctx: ContextTimed,
  ): Promise<Task> {
    if (tran == null) {
      return await this.db.withTransactionF(
        async (tran) =>
          await this.updateRefreshBucketDelay(
            bucketIndex,
            delay,
            lazy,
            tran,
            ctx,
          ),
      );
    }

    const jitter = nodesUtils.refreshBucketsDelayJitter(
      delay,
      this.refreshBucketDelayJitter,
    );
    let foundTask: Task | undefined;
    let existingTask = false;
    for await (const task of this.taskManager.getTasks(
      'asc',
      true,
      [this.tasksPath, this.refreshBucketHandlerId, `${bucketIndex}`],
      tran,
    )) {
      ctx.signal.throwIfAborted();
      if (!existingTask) {
        foundTask = task;
        // Update the first one
        // total delay is refreshBucketDelay + time since task creation
        // time since task creation = now - creation time;
        const delayNew =
          performance.now() +
          performance.timeOrigin -
          task.created.getTime() +
          delay +
          jitter;
        try {
          await this.taskManager.updateTask(task.id, { delay: delayNew });
          existingTask = true;
        } catch (e) {
          if (e instanceof tasksErrors.ErrorTaskRunning) {
            // Ignore running
            existingTask = true;
          } else if (!(e instanceof tasksErrors.ErrorTaskMissing)) {
            throw e;
          }
        }
        this.logger.debug(
          `Updating refreshBucket task for bucket ${bucketIndex}`,
        );
      } else {
        // These are extra, so we cancel them
        task.cancel(abortSingletonTaskReason);
        this.logger.warn(
          `Duplicate refreshBucket task was found for bucket ${bucketIndex}, cancelling`,
        );
      }
    }
    if (!existingTask) {
      this.logger.debug(
        `No refreshBucket task for bucket ${bucketIndex}, new one was created`,
      );
      foundTask = await this.taskManager.scheduleTask({
        delay: delay + jitter,
        handlerId: this.refreshBucketHandlerId,
        lazy: true,
        parameters: [bucketIndex],
        path: [this.tasksPath, this.refreshBucketHandlerId, `${bucketIndex}`],
        priority: 0,
      });
    }
    if (foundTask == null) utils.never('task should exist');
    return foundTask;
  }

  /**
   * Perform an initial database synchronisation: get k of the closest nodes
   * from each seed node and add them to this database
   * Establish a connection to each node before adding it
   * By default this operation is blocking, set `block` to `false` to make it
   *
   * From the spec:
   * To join the network, a node u must have a contact to an already participating node w. u inserts w into the
   * appropriate k-bucket. u then performs a node lookup for its own node ID. Finally, u refreshes all k-buckets further
   * away than its closest neighbor. During the refreshes, u both populates its own k-buckets and inserts itself into
   * other nodes’ k-buckets as necessary.
   *
   * So this will do 3 steps
   * 1. Connect to the initial nodes
   * 2. do a find-node operation for itself
   * 3. reschedule refresh bucket operations for every bucket above the closest node we found
   *
   */
  public syncNodeGraph(
    network: string | undefined,
    initialNodes: Array<[NodeId, NodeAddress]>,
    connectionConnectTimeoutTime?: number,
    blocking?: boolean,
    ctx?: Partial<ContextTimedInput>,
  ): PromiseCancellable<void>;
  @startStop.ready(new nodesErrors.ErrorNodeManagerNotRunning())
  @decorators.timedCancellable(true)
  public async syncNodeGraph(
    network: string,
    initialNodes: Array<[NodeId, NodeAddress]>,
    connectionConnectTimeoutTime: number = this.connectionConnectTimeoutTime,
    blocking: boolean = false,
    @decorators.context ctx: ContextTimed,
  ): Promise<void> {
    const logger = this.logger.getChild('syncNodeGraph');
    logger.info('Synchronizing NodeGraph');
    if (initialNodes.length === 0) {
      throw new nodesErrors.ErrorNodeManagerSyncNodeGraphFailed(
        'must provide at least 1 initial node',
      );
    }
    const initialNodesParameter = initialNodes.map(([nodeId, address]) => {
      return [nodesUtils.encodeNodeId(nodeId), address] as [
        NodeIdEncoded,
        NodeAddress,
      ];
    });
    if (blocking) {
      await this.syncNodeGraphHandler(
        ctx,
        undefined,
        network,
        initialNodesParameter,
        connectionConnectTimeoutTime,
      );
    } else {
      // Create task
      await this.taskManager.scheduleTask({
        delay: 0,
        handlerId: this.syncNodeGraphHandlerId,
        lazy: true,
        parameters: [
          network,
          initialNodesParameter,
          connectionConnectTimeoutTime,
        ],
        path: [this.tasksPath, this.syncNodeGraphHandlerId],
        priority: 0,
      });
    }
  }
}

export default NodeManager;