CHANGELOG.md

@@ -1,5 +1,3 @@
-# [0.1.0-develop.3](https://git.lumeweb.com/LumeWeb/libs5/compare/v0.1.0-develop.2...v0.1.0-develop.3) (2023-08-31)
-
 # [0.1.0-develop.2](https://git.lumeweb.com/LumeWeb/libs5/compare/v0.1.0-develop.1...v0.1.0-develop.2) (2023-08-31)
 
 

npm-shrinkwrap.json

@@ -1,12 +1,12 @@
 {
   "name": "@lumeweb/libs5",
-  "version": "0.1.0-develop.3",
+  "version": "0.1.0-develop.2",
   "lockfileVersion": 3,
   "requires": true,
   "packages": {
     "": {
       "name": "@lumeweb/libs5",
-      "version": "0.1.0-develop.3",
+      "version": "0.1.0-develop.2",
       "dependencies": {
         "@noble/curves": "^1.1.0",
         "@noble/hashes": "^1.3.1",

package.json

@@ -1,6 +1,6 @@
 {
   "name": "@lumeweb/libs5",
-  "version": "0.1.0-develop.3",
+  "version": "0.1.0-develop.2",
   "type": "module",
   "main": "lib/index.js",
   "repository": {

src/constants.ts

@@ -133,4 +133,3 @@ export const storageLocationTypeArchive = 0;
 export const storageLocationTypeFile = 3;
 export const storageLocationTypeFull = 5;
 export const storageLocationTypeBridge = 7;
-export const supportedFeatures = 3;

src/index.ts

@@ -1,14 +1,8 @@
 import { S5Node } from "#node.js";
 import type { S5NodeConfig } from "#node.js";
-import type { SignedRegistryEntry } from "#service/registry.js";
 
 export * from "./types.js";
-export {
-  createTransportSocket,
-  isTransport,
-  createTransportPeer,
-} from "./transports/index.js";
-export type { S5NodeConfig, SignedRegistryEntry };
+export type { S5NodeConfig };
 
 export function createNode(config: S5NodeConfig) {
   return new S5Node(config);

src/messages/handshakeOpen.ts

@@ -1,33 +0,0 @@
-import Packer from "#serialization/pack.js";
-import { protocolMethodHandshakeDone, supportedFeatures } from "#constants.js";
-import { S5Node } from "#node.js";
-import { Peer } from "#types.js";
-import Unpacker from "#serialization/unpack.js";
-
-export default async function (
-  node: S5Node,
-  peer: Peer,
-  data: Unpacker,
-  rawData: Uint8Array,
-) {
-  const p = new Packer();
-  p.packInt(protocolMethodHandshakeDone);
-  p.packBinary(data.unpackBinary());
-  let peerNetworkId: string | null = null;
-  try {
-    peerNetworkId = data.unpackString();
-  } catch {}
-
-  if (this.networkId && peerNetworkId !== this.networkId) {
-    throw `Peer is in different network: ${peerNetworkId}`;
-  }
-
-  p.packInt(supportedFeatures);
-  p.packInt(node.services.p2p.selfConnectionUris.length);
-  for (const uri of this.selfConnectionUris) {
-    p.packString(uri.toString());
-  }
-  // TODO Protocol version
-  // p.packInt(protocolVersion);
-  peer.sendMessage(await this.signMessageSimple(p.takeBytes()));
-}

src/messages/index.ts

@@ -1,12 +0,0 @@
-import { P2PMessageHandler } from "#types.js";
-import handshakeOpen from "#messages/handshakeOpen.js";
-
-const messages = new Map<number, P2PMessageHandler>(
-  Object.entries({
-    protocolMethodHandshakeOpen: handshakeOpen,
-  }).map(([key, value]) => [Number(key), value]),
-);
-
-Object.freeze(messages);
-
-export default messages;

src/messages/registryEntry.ts

@@ -1,13 +0,0 @@
-import { S5Node } from "#node.js";
-import { Peer } from "#types.js";
-import Unpacker from "#serialization/unpack.js";
-
-export default async function (
-  node: S5Node,
-  peer: Peer,
-  data: Unpacker,
-  rawData: Uint8Array,
-) {
-  const sre = node.services.registry.deserializeRegistryEntry(rawData);
-  await node.services.registry.set(sre, false, peer);
-}

src/messages/registryQuery.ts

@@ -1,16 +0,0 @@
-import { S5Node } from "#node.js";
-import { Peer } from "#types.js";
-import Unpacker from "#serialization/unpack.js";
-
-export default async function (
-  node: S5Node,
-  peer: Peer,
-  data: Unpacker,
-  rawData: Uint8Array,
-) {
-  const pk = data.unpackBinary();
-  const sre = await node.services.registry.getFromDB(pk);
-  if (sre !== null) {
-    peer.sendMessage(node.services.registry.serializeRegistryEntry(sre));
-  }
-}

src/messages/signedMessage.ts

@@ -1,20 +0,0 @@
-import { S5Node } from "#node.js";
-import { Peer } from "#types.js";
-import Unpacker from "#serialization/unpack.js";
-import messages from "#messages/signedMessages/index.js";
-
-export default async function (
-  node: S5Node,
-  peer: Peer,
-  data: Unpacker,
-  rawData: Uint8Array,
-  verifyId = true,
-) {
-  const sm = await node.services.p2p.unpackAndVerifySignature(data);
-  const u = Unpacker.fromPacked(sm.message);
-  const method = data.unpackInt();
-
-  if (method !== null && messages.has(method)) {
-    await messages.get(method)?.(node, peer, u, sm, verifyId);
-  }
-}

src/messages/signedMessages/announcePeers.ts

@@ -1,39 +0,0 @@
-import { S5Node } from "#node.js";
-import { Peer, SignedMessage } from "#types.js";
-import Unpacker from "#serialization/unpack.js";
-import { equalBytes } from "@noble/curves/abstract/utils";
-import { URL } from "url";
-import NodeId from "#nodeId.js";
-
-export default async function (
-  node: S5Node,
-  peer: Peer,
-  data: Unpacker,
-  message: SignedMessage,
-  verifyId: boolean,
-) {
-  const length = data.unpackInt() as number;
-  for (let i = 0; i < length; i++) {
-    const peerIdBinary = data.unpackBinary();
-    const id = new NodeId(peerIdBinary);
-
-    const isConnected = data.unpackBool() as boolean;
-
-    const connectionUrisCount = data.unpackInt() as number;
-
-    const connectionUris: URL[] = [];
-
-    for (let i = 0; i < connectionUrisCount; i++) {
-      connectionUris.push(new URL(data.unpackString() as string));
-    }
-
-    if (connectionUris.length > 0) {
-      // TODO Fully support multiple connection uris
-      const uri = new URL(connectionUris[0].toString());
-      uri.username = id.toBase58();
-      if (!this.reconnectDelay.has(NodeId.decode(uri.username).toString())) {
-        node.services.p2p.connectToNode([uri]);
-      }
-    }
-  }
-}

src/messages/signedMessages/handshakeDone.ts

@@ -1,65 +0,0 @@
-import { S5Node } from "#node.js";
-import { Peer, SignedMessage } from "#types.js";
-import Unpacker from "#serialization/unpack.js";
-import { equalBytes } from "@noble/curves/abstract/utils";
-import { URL } from "url";
-
-export default async function (
-  node: S5Node,
-  peer: Peer,
-  data: Unpacker,
-  message: SignedMessage,
-  verifyId: boolean,
-) {
-  const challenge = data.unpackBinary();
-
-  if (!equalBytes(peer.challenge, challenge)) {
-    throw "Invalid challenge";
-  }
-
-  const pId = message.nodeId;
-
-  if (!verifyId) {
-    peer.id = pId;
-  } else {
-    if (!peer.id.equals(pId)) {
-      throw "Invalid transports id on initial list";
-    }
-  }
-
-  peer.isConnected = true;
-
-  const supportedFeatures = data.unpackInt();
-
-  if (supportedFeatures !== 3) {
-    throw "Remote node does not support required features";
-  }
-
-  node.services.p2p.peers.set(peer.id.toString(), peer);
-  node.services.p2p.reconnectDelay.set(peer.id.toString(), 1);
-
-  const connectionUrisCount = data.unpackInt() as number;
-
-  peer.connectionUris = [];
-  for (let i = 0; i < connectionUrisCount; i++) {
-    peer.connectionUris.push(new URL(data.unpackString() as string));
-  }
-
-  this.logger.info(
-    `[+] ${peer.id.toString()} (${peer.renderLocationUri().toString()})`,
-  );
-
-  node.services.p2p.sendPublicPeersToPeer(
-    peer,
-    Array.from(node.services.p2p.peers.values()),
-  );
-  for (const p of this._peers.values()) {
-    if (p.id.equals(peer.id)) {
-      continue;
-    }
-
-    if (p.isConnected) {
-      this.sendPublicPeersToPeer(p, [peer]);
-    }
-  }
-}

src/messages/signedMessages/index.ts

@@ -1,12 +0,0 @@
-import { P2PSignedMessageHandler } from "#types.js";
-import handshakeDone from "#messages/signedMessages/handshakeDone.js";
-
-const messages = new Map<number, P2PSignedMessageHandler>(
-  Object.entries({
-    protocolMethodHandshakeOpen: handshakeDone,
-  }).map(([key, value]) => [Number(key), value]),
-);
-
-Object.freeze(messages);
-
-export default messages;

src/messages/storageLocation.ts

@@ -1,77 +0,0 @@
-import { S5Node } from "#node.js";
-import { Peer } from "#types.js";
-import Unpacker from "#serialization/unpack.js";
-import { Multihash } from "#multihash.js";
-import { decodeEndian } from "#util.js";
-import { mkeyEd25519 } from "#constants.js";
-import { ed25519 } from "@noble/curves/ed25519";
-import NodeId from "#nodeId.js";
-import StorageLocation from "#storage.js";
-
-export default async function (
-  node: S5Node,
-  peer: Peer,
-  data: Unpacker,
-  rawData: Uint8Array,
-) {
-  const hash = new Multihash(rawData.subarray(1, 34));
-  const type = rawData[34];
-  const expiry = decodeEndian(rawData.subarray(35, 39));
-  const partCount = rawData[39];
-  const parts: string[] = [];
-  let cursor = 40;
-  for (let i = 0; i < partCount; i++) {
-    const length = decodeEndian(rawData.subarray(cursor, cursor + 2));
-    cursor += 2;
-    parts.push(
-      new TextDecoder().decode(rawData.subarray(cursor, cursor + length)),
-    );
-    cursor += length;
-  }
-  cursor++;
-
-  const publicKey = rawData.subarray(cursor, cursor + 33);
-  const signature = rawData.subarray(cursor + 33);
-
-  if (publicKey[0] !== mkeyEd25519) {
-    throw `Unsupported public key type ${mkeyEd25519}`;
-  }
-
-  if (
-    !ed25519.verify(
-      signature,
-      rawData.subarray(0, cursor),
-      publicKey.subarray(1),
-    )
-  ) {
-    return;
-  }
-
-  const nodeId = new NodeId(publicKey);
-  await node.addStorageLocation({
-    hash,
-    nodeId,
-    location: new StorageLocation(type, parts, expiry),
-    message: rawData,
-    config: this.node.config,
-  });
-
-  const list = this.hashQueryRoutingTable.get(hash) || new Set<NodeId>();
-  for (const peerId of list) {
-    if (peerId.equals(nodeId)) {
-      continue;
-    }
-    if (peerId.equals(peer.id)) {
-      continue;
-    }
-
-    if (this._peers.has(peerId.toString())) {
-      try {
-        this._peers.get(peerId.toString())?.sendMessage(event);
-      } catch (e) {
-        this.logger.catched(e);
-      }
-    }
-  }
-  this.hashQueryRoutingTable.delete(hash);
-}

src/node.ts

@@ -40,29 +40,11 @@ export interface S5NodeConfig {
 
 export class S5Node {
   private _nodeConfig: S5NodeConfig;
-
+  private _config?: S5Config;
   constructor(config: S5NodeConfig) {
     this._nodeConfig = config;
   }
 
-  private _config?: S5Config;
-
-  get config() {
-    return this._config as S5Config;
-  }
-
-  get services() {
-    return this._config?.services as S5Services;
-  }
-
-  get db() {
-    return this._config?.db as AbstractLevel<Uint8Array, string, Uint8Array>;
-  }
-
-  get logger() {
-    return this._config?.logger as Logger;
-  }
-
   public async start() {
     this._config = {
       keyPair: this._nodeConfig.keyPair,
@@ -80,63 +62,77 @@ export class S5Node {
     await p2p.start();
   }
 
-  async readStorageLocationsFromDB(
-    hash: Multihash,
-  ): Promise<Map<number, Map<string, Map<number, any>>>> {
-    const map = new Map<number, Map<string, Map<number, any>>>();
-    const bytes = await this.db.get(stringifyHash(hash));
-    if (bytes === null) {
-      return map;
-    }
-    const unpacker = Unpacker.fromPacked(bytes);
-    const mapLength = unpacker.unpackMapLength();
-    for (let i = 0; i < mapLength; i++) {
-      const type = unpacker.unpackInt() as number;
-      const innerMap = new Map<string, Map<number, any>>();
-      map.set(type, innerMap);
-      const innerMapLength = unpacker.unpackMapLength();
-      for (let j = 0; j < innerMapLength; j++) {
-        const nodeId = new NodeId(unpacker.unpackBinary());
-        innerMap.set(
-          nodeId.toString(),
-          new Map(unpacker.unpackMap() as [number, any][]),
-        );
-      }
-    }
+  get services() {
+    return this._config?.services as S5Services;
+  }
+  get config() {
+    return this._config as S5Config;
+  }
+  get db() {
+    return this._config?.db as AbstractLevel<Uint8Array, string, Uint8Array>;
+  }
+  get logger() {
+    return this._config?.logger as Logger;
+  }
+}
+
+export async function readStorageLocationsFromDB({
+  hash,
+  config,
+}: {
+  hash: Multihash;
+  config: S5Config;
+}): Promise<Map<number, Map<NodeId, Map<number, any>>>> {
+  const map = new Map<number, Map<NodeId, Map<number, any>>>();
+  const bytes = await config.db.get(stringifyHash(hash));
+  if (bytes === null) {
     return map;
   }
-
-  async addStorageLocation({
-    hash,
-    nodeId,
-    location,
-    message,
-    config,
-  }: {
-    hash: Multihash;
-    nodeId: NodeId;
-    location: StorageLocation;
-    message?: Uint8Array;
-    config: S5Config;
-  }) {
-    const map = await this.readStorageLocationsFromDB(hash);
-    const innerMap =
-      map.get(location.type) || new Map<string, Map<number, any>>();
-    map.set(location.type, innerMap);
-
-    const locationMap = new Map<number, any>([
-      [1, location.parts],
-      // [2, location.binaryParts],
-      [3, location.expiry],
-      [4, message],
-    ]);
-
-    innerMap.set(nodeId.toString(), locationMap);
-    await config.cacheDb.put(
-      stringifyHash(hash),
-      new Packer().pack(map).takeBytes(),
-    );
+  const unpacker = Unpacker.fromPacked(bytes);
+  const mapLength = unpacker.unpackMapLength();
+  for (let i = 0; i < mapLength; i++) {
+    const type = unpacker.unpackInt() as number;
+    const innerMap = new Map<NodeId, Map<number, any>>();
+    map.set(type, innerMap);
+    const innerMapLength = unpacker.unpackMapLength();
+    for (let j = 0; j < innerMapLength; j++) {
+      const nodeId = new NodeId(unpacker.unpackBinary());
+      innerMap.set(nodeId, new Map(unpacker.unpackMap() as [number, any][]));
+    }
   }
+  return map;
+}
+
+export async function addStorageLocation({
+  hash,
+  nodeId,
+  location,
+  message,
+  config,
+}: {
+  hash: Multihash;
+  nodeId: NodeId;
+  location: StorageLocation;
+  message?: Uint8Array;
+  config: S5Config;
+}) {
+  const map = this.readStorageLocationsFromDB(hash);
+  const innerMap =
+    map.get(location.type) || new Map<NodeId, Map<number, any>>();
+  map.set(location.type, innerMap);
+
+  const locationMap = new Map<number, any>([
+    [1, location.parts],
+    // [2, location.binaryParts],
+    [3, location.expiry],
+    [4, message],
+  ]);
+
+  innerMap.set(nodeId, locationMap);
+  await config.cacheDb.put(
+    stringifyHash(hash),
+    new Packer().pack(map).takeBytes(),
+  );
 }
 
 export function stringifyBytes(data: Uint8Array) {

src/peer/tcp.ts

@@ -3,7 +3,6 @@ import NodeId from "../nodeId.js";
 import * as net from "net";
 import { URL } from "url";
 import { decodeEndian } from "../util.js";
-import * as console from "console";
 
 export class TcpPeer implements Peer {
   connectionUris: Array<URL>;
@@ -80,17 +79,15 @@ export class TcpPeer implements Peer {
       this._socket.on("error", onError);
     }
   }
-
-  public static async connect(uri: URL): Promise<net.Socket> {
-    const host = uri.hostname;
-    const port = parseInt(uri.port);
-    return new Promise((resolve, reject) => {
-      const socket = net.connect(port, host, () => {
-        resolve(socket);
-      });
-      socket.on("error", (err) => {
-        reject(err);
-      });
-    });
-  }
+}
+
+export async function connect(port: number, host: string): Promise<net.Socket> {
+  return new Promise((resolve, reject) => {
+    const socket = net.connect(port, host, () => {
+      resolve(socket);
+    });
+    socket.on("error", (err) => {
+      reject(err);
+    });
+  });
 }

src/peer/webSocket.ts

@@ -56,16 +56,15 @@ export class WebSocketPeer implements Peer {
       this._socket.addEventListener("error", onError);
     }
   }
-
-  public static async connect(uri: URL): Promise<WebSocket> {
-    return new Promise((resolve, reject) => {
-      const socket = new WebSocket(uri);
-      socket.addEventListener("open", () => {
-        resolve(socket);
-      });
-      socket.addEventListener("error", (err) => {
-        reject(err);
-      });
-    });
-  }
+}
+export async function connect(uri: string): Promise<WebSocket> {
+  return new Promise((resolve, reject) => {
+    const socket = new WebSocket(uri);
+    socket.addEventListener("open", () => {
+      resolve(socket);
+    });
+    socket.addEventListener("error", (err) => {
+      reject(err);
+    });
+  });
 }

src/service/p2p.ts

@@ -1,37 +1,41 @@
 import { Multihash } from "../multihash.js";
 import NodeId from "../nodeId.js";
-import { Logger, Peer, SignedMessage } from "../types.js";
+import { equalBytes } from "@noble/curves/abstract/utils";
+import { Logger, Peer, S5Config, SignedMessage } from "../types.js";
 import KeyPairEd25519 from "../ed25519.js";
 import * as crypto from "crypto";
 import {
+  mkeyEd25519,
   protocolMethodAnnouncePeers,
+  protocolMethodHandshakeDone,
   protocolMethodHandshakeOpen,
   protocolMethodHashQuery,
   protocolMethodSignedMessage,
+  recordTypeRegistryEntry,
   recordTypeStorageLocation,
   storageLocationTypeFull,
 } from "../constants.js";
 import defer from "p-defer";
-import { calculateScore, encodeEndian } from "#util.js";
+import { calculateScore, decodeEndian, encodeEndian } from "#util.js";
 import Packer from "#serialization/pack.js";
 import Unpacker from "#serialization/unpack.js";
 import { ed25519 } from "@noble/curves/ed25519";
 import { AbstractLevel, AbstractSublevel } from "abstract-level";
 import StorageLocation from "#storage.js";
-import { S5Node, stringifyNode } from "#node.js";
+import { addStorageLocation, S5Node, stringifyNode } from "#node.js";
 import { URL } from "url";
 import { Buffer } from "buffer";
-import {
-  createTransportPeer,
-  createTransportSocket,
-} from "#transports/index.js";
-import messages from "#messages/index.js";
+import { connect as tcpConnect, TcpPeer } from "../peer/tcp.js";
+import { connect as wsConnect, WebSocketPeer } from "../peer/webSocket.js";
 
 export class P2PService {
+  private node: S5Node;
   private logger: Logger;
   private nodeKeyPair: KeyPairEd25519;
   private localNodeId?: NodeId;
   private networkId?: string;
+  private reconnectDelay: Map<string, number> = new Map();
+  private selfConnectionUris: Array<URL> = [];
   private nodesDb?: AbstractSublevel<
     AbstractLevel<Uint8Array, string, Uint8Array>,
     Uint8Array,
@@ -41,7 +45,7 @@ export class P2PService {
   private hashQueryRoutingTable: Map<Multihash, Set<NodeId>> = new Map();
 
   constructor(node: S5Node) {
-    this._node = node;
+    this.node = node;
     this.networkId = node.config.p2p?.network;
     this.nodeKeyPair = node.config.keyPair;
     this.logger = node.logger;
@@ -49,24 +53,6 @@ export class P2PService {
     node.config.services.p2p = this;
   }
 
-  private _node: S5Node;
-
-  get node(): S5Node {
-    return this._node;
-  }
-
-  private _reconnectDelay: Map<string, number> = new Map();
-
-  get reconnectDelay(): Map<string, number> {
-    return this._reconnectDelay;
-  }
-
-  private _selfConnectionUris: Array<URL> = [];
-
-  get selfConnectionUris(): Array<URL> {
-    return this._selfConnectionUris;
-  }
-
   private _peers: Map<string, Peer> = new Map();
 
   get peers(): Map<string, Peer> {
@@ -75,11 +61,11 @@ export class P2PService {
 
   async init(): Promise<void> {
     this.localNodeId = new NodeId(this.nodeKeyPair.publicKey); // Define the NodeId constructor
-    this.nodesDb = this._node.db.sublevel<string, Uint8Array>("s5-nodes", {});
+    this.nodesDb = this.node.db.sublevel<string, Uint8Array>("s5-nodes", {});
   }
 
   async start(): Promise<void> {
-    const initialPeers = this._node.config?.p2p?.peers?.initial || [];
+    const initialPeers = this.node.config?.p2p?.peers?.initial || [];
 
     for (const p of initialPeers) {
       this.connectToNode([new URL(p)]);
@@ -98,14 +84,196 @@ export class P2PService {
 
     const completer = defer<void>();
 
+    const supportedFeatures = 3; // 0b00000011
+
     peer.listenForMessages(
       async (event: Uint8Array) => {
         let u = Unpacker.fromPacked(event);
         const method = u.unpackInt();
+        if (method === protocolMethodHandshakeOpen) {
+          const p = new Packer();
+          p.packInt(protocolMethodHandshakeDone);
+          p.packBinary(u.unpackBinary());
+          let peerNetworkId: string | null = null;
+          try {
+            peerNetworkId = u.unpackString();
+          } catch {}
 
-        if (method !== null && messages.has(method)) {
-          await messages.get(method)?.(this.node, peer, u, event, verifyId);
+          if (this.networkId && peerNetworkId !== this.networkId) {
+            throw `Peer is in different network: ${peerNetworkId}`;
+          }
+
+          p.packInt(supportedFeatures);
+          p.packInt(this.selfConnectionUris.length);
+          for (const uri of this.selfConnectionUris) {
+            p.packString(uri.toString());
+          }
+          // TODO Protocol version
+          // p.packInt(protocolVersion);
+          peer.sendMessage(await this.signMessageSimple(p.takeBytes()));
+          return;
+        } else if (method === recordTypeRegistryEntry) {
+          const sre =
+            this.node.services.registry.deserializeRegistryEntry(event);
+          await this.node.services.registry.set(sre, false, peer);
+          return;
+        } else if (method === recordTypeStorageLocation) {
+          const hash = new Multihash(event.subarray(1, 34));
+          const type = event[34];
+          const expiry = decodeEndian(event.subarray(35, 39));
+          const partCount = event[39];
+          const parts: string[] = [];
+          let cursor = 40;
+          for (let i = 0; i < partCount; i++) {
+            const length = decodeEndian(event.subarray(cursor, cursor + 2));
+            cursor += 2;
+            parts.push(
+              new TextDecoder().decode(event.subarray(cursor, cursor + length)),
+            );
+            cursor += length;
+          }
+          cursor++;
+
+          const publicKey = event.subarray(cursor, cursor + 33);
+          const signature = event.subarray(cursor + 33);
+
+          if (publicKey[0] !== mkeyEd25519) {
+            throw `Unsupported public key type ${mkeyEd25519}`;
+          }
+
+          if (
+            !ed25519.verify(
+              signature,
+              event.subarray(0, cursor),
+              publicKey.subarray(1),
+            )
+          ) {
+            return;
+          }
+
+          const nodeId = new NodeId(publicKey);
+          await addStorageLocation({
+            hash,
+            nodeId,
+            location: new StorageLocation(type, parts, expiry),
+            message: event,
+            config: this.node.config,
+          });
+
+          const list =
+            this.hashQueryRoutingTable.get(hash) || new Set<NodeId>();
+          for (const peerId of list) {
+            if (peerId.equals(nodeId)) {
+              continue;
+            }
+            if (peerId.equals(peer.id)) {
+              continue;
+            }
+
+            if (this._peers.has(peerId.toString())) {
+              try {
+                this._peers.get(peerId.toString())?.sendMessage(event);
+              } catch (e) {
+                this.logger.catched(e);
+              }
+            }
+          }
+          this.hashQueryRoutingTable.delete(hash);
         }
+
+        if (method === protocolMethodSignedMessage) {
+          const sm = await this.unpackAndVerifySignature(u);
+          u = Unpacker.fromPacked(sm.message);
+          const method2 = u.unpackInt();
+
+          if (method2 === protocolMethodHandshakeDone) {
+            const challenge = u.unpackBinary();
+
+            if (!equalBytes(peer.challenge, challenge)) {
+              throw "Invalid challenge";
+            }
+
+            const pId = sm.nodeId;
+
+            if (!verifyId) {
+              peer.id = pId;
+            } else {
+              if (!peer.id.equals(pId)) {
+                throw "Invalid peer id on initial list";
+              }
+            }
+
+            peer.isConnected = true;
+
+            const supportedFeatures = u.unpackInt();
+
+            if (supportedFeatures !== 3) {
+              throw "Remote node does not support required features";
+            }
+
+            this._peers.set(peer.id.toString(), peer);
+            this.reconnectDelay.set(peer.id.toString(), 1);
+
+            const connectionUrisCount = u.unpackInt() as number;
+
+            peer.connectionUris = [];
+            for (let i = 0; i < connectionUrisCount; i++) {
+              peer.connectionUris.push(new URL(u.unpackString() as string));
+            }
+
+            this.logger.info(
+              `[+] ${peer.id.toString()} (${peer
+                .renderLocationUri()
+                .toString()})`,
+            );
+
+            this.sendPublicPeersToPeer(peer, Array.from(this._peers.values()));
+            for (const p of this._peers.values()) {
+              if (p.id.equals(peer.id)) continue;
+
+              if (p.isConnected) {
+                this.sendPublicPeersToPeer(p, [peer]);
+              }
+            }
+
+            return;
+          } else if (method2 === protocolMethodAnnouncePeers) {
+            const length = u.unpackInt() as number;
+            for (let i = 0; i < length; i++) {
+              const peerIdBinary = u.unpackBinary();
+              const id = new NodeId(peerIdBinary);
+
+              const isConnected = u.unpackBool() as boolean;
+
+              const connectionUrisCount = u.unpackInt() as number;
+
+              const connectionUris: URL[] = [];
+
+              for (let i = 0; i < connectionUrisCount; i++) {
+                connectionUris.push(new URL(u.unpackString() as string));
+              }
+
+              if (connectionUris.length > 0) {
+                // TODO Fully support multiple connection uris
+                const uri = new URL(connectionUris[0].toString());
+                uri.username = id.toBase58();
+                if (
+                  !this.reconnectDelay.has(
+                    NodeId.decode(uri.username).toString(),
+                  )
+                ) {
+                  this.connectToNode([uri]);
+                }
+              }
+            }
+          }
+        } /* else if (method === protocolMethodRegistryQuery) {
+          const pk = u.unpackBinary();
+          const sre = node.registry.getFromDB(pk);
+          if (sre !== null) {
+            peer.sendMessage(node.registry.serializeRegistryEntry(sre));
+          }
+        }*/
       },
       {
         onDone: async () => {
@@ -288,7 +456,7 @@ export class P2PService {
       );
     }
 
-    const protocol = connectionUri.protocol.replace(":", "");
+    const protocol = connectionUri.protocol;
 
     if (!connectionUri.username) {
       throw new Error("Connection URI does not contain node id");
@@ -296,9 +464,9 @@ export class P2PService {
 
     const id = NodeId.decode(connectionUri.username);
 
-    this._reconnectDelay.set(
+    this.reconnectDelay.set(
       id.toString(),
-      this._reconnectDelay.get(id.toString()) || 1,
+      this.reconnectDelay.get(id.toString()) || 1,
     );
 
     if (id.equals(this.localNodeId)) {
@@ -307,12 +475,21 @@ export class P2PService {
 
     try {
       this.logger.verbose(`[connect] ${connectionUri}`);
+      if (protocol === "tcp:") {
+        const ip = connectionUri.hostname;
+        const port = parseInt(connectionUri.port);
+        const socket = await tcpConnect(port, ip);
 
-      const socket = await createTransportSocket(protocol, connectionUri);
-      await this.onNewPeer(
-        createTransportPeer(protocol, socket, [connectionUri]),
-        true,
-      );
+        const peer = new TcpPeer(socket, [connectionUri]);
+        peer.id = id;
+
+        await this.onNewPeer(peer, true);
+      } else {
+        const channel = await wsConnect(connectionUri.toString());
+        const peer = new WebSocketPeer(channel, [connectionUri]);
+        peer.id = id;
+        await this.onNewPeer(peer, true);
+      }
     } catch (e) {
       if (retried) {
         return;
@@ -321,8 +498,8 @@ export class P2PService {
 
       this.logger.catched(e);
 
-      const delay = this._reconnectDelay.get(id.toString())!;
-      this._reconnectDelay.set(id.toString(), delay * 2);
+      const delay = this.reconnectDelay.get(id.toString())!;
+      this.reconnectDelay.set(id.toString(), delay * 2);
       await new Promise((resolve) => setTimeout(resolve, delay * 1000));
 
       await this.connectToNode(connectionUris, retried);

src/service/registry.ts

@@ -16,7 +16,7 @@ import { EventEmitter } from "events";
 import KeyPairEd25519 from "#ed25519.js";
 import { S5Node, stringifyBytes } from "#node.js";
 
-export interface SignedRegistryEntry {
+interface SignedRegistryEntry {
   pk: Uint8Array; // public key with multicodec prefix
   revision: number; // revision number of this entry, maximum is (256^8)-1
   data: Uint8Array; // data stored in this entry, can have a maximum length of 48 bytes
@@ -68,7 +68,7 @@ export class RegistryService {
         throw new Error("Data too long");
       }
 
-      const isValid = this.verifyRegistryEntry(sre);
+      const isValid = verifyRegistryEntry(sre);
       if (!isValid) {
         throw new Error("Invalid signature found");
       }
@@ -81,7 +81,7 @@ export class RegistryService {
         if (existingEntry.revision === sre.revision) {
           return;
         } else if (existingEntry.revision > sre.revision) {
-          const updateMessage = this.serializeRegistryEntry(existingEntry);
+          const updateMessage = serializeRegistryEntry(existingEntry);
           receivedFrom.sendMessage(updateMessage);
           return;
         }
@@ -95,7 +95,7 @@ export class RegistryService {
     const key = new Multihash(sre.pk);
     this.streams.get(key.toString())?.emit("event", sre);
 
-    this.db?.put(stringifyBytes(sre.pk), this.serializeRegistryEntry(sre));
+    this.db?.put(stringifyBytes(sre.pk), serializeRegistryEntry(sre));
 
     this.broadcastEntry(sre, receivedFrom);
   }
@@ -104,7 +104,7 @@ export class RegistryService {
   // TODO: If there are more than X peers, only broadcast to subscribed nodes (routing table) and shard-nodes (256)
   broadcastEntry(sre: SignedRegistryEntry, receivedFrom?: Peer): void {
     this.logger.verbose("[registry] broadcastEntry");
-    const updateMessage = this.serializeRegistryEntry(sre);
+    const updateMessage = serializeRegistryEntry(sre);
 
     for (const p of Object.values(this.node.services.p2p.peers)) {
       if (receivedFrom == null || p.id !== receivedFrom.id) {
@@ -216,25 +216,25 @@ export class RegistryService {
       signature: event.slice(43 + dataLength),
     };
   }
-
-  public verifyRegistryEntry(sre: SignedRegistryEntry): boolean {
-    const list: Uint8Array = Uint8Array.from([
-      recordTypeRegistryEntry,
-      ...encodeEndian(sre.revision, 8),
-      sre.data.length, // 1 byte
-      ...sre.data,
-    ]);
-
-    return ed25519.verify(list, sre.signature, sre.pk.slice(1));
-  }
-  public serializeRegistryEntry(sre: SignedRegistryEntry): Uint8Array {
-    return Uint8Array.from([
-      recordTypeRegistryEntry,
-      ...sre.pk,
-      ...encodeEndian(sre.revision, 8),
-      sre.data.length,
-      ...sre.data,
-      ...sre.signature,
-    ]);
-  }
+}
+
+function verifyRegistryEntry(sre: SignedRegistryEntry): boolean {
+  const list: Uint8Array = Uint8Array.from([
+    recordTypeRegistryEntry,
+    ...encodeEndian(sre.revision, 8),
+    sre.data.length, // 1 byte
+    ...sre.data,
+  ]);
+
+  return ed25519.verify(list, sre.signature, sre.pk.slice(1));
+}
+function serializeRegistryEntry(sre: SignedRegistryEntry): Uint8Array {
+  return Uint8Array.from([
+    recordTypeRegistryEntry,
+    ...sre.pk,
+    ...encodeEndian(sre.revision, 8),
+    sre.data.length,
+    ...sre.data,
+    ...sre.signature,
+  ]);
 }

src/transports/index.ts

@@ -1,42 +0,0 @@
-import { URL } from "url";
-import { TcpPeer } from "#transports/tcp.js";
-import { WebSocketPeer } from "#transports/webSocket.js";
-import { PeerStatic } from "#types.js";
-
-const transports = new Map<string, PeerStatic>();
-
-export function registerTransport(type: string, transport: PeerStatic) {
-  transports.set(type, transport);
-}
-
-export function isTransport(type: string) {
-  return transports.has(type);
-}
-
-export function createTransportSocket(type: string, uri: URL) {
-  if (!isTransport(type)) {
-    throw new Error(`transport ${type} does not exist`);
-  }
-
-  const transport = transports.get(type) as PeerStatic;
-
-  return transport.connect(uri);
-}
-
-export function createTransportPeer(
-  type: string,
-  socket: any,
-  connectionUris: URL[] = [],
-) {
-  if (!isTransport(type)) {
-    throw new Error(`transport ${type} does not exist`);
-  }
-
-  const transport = transports.get(type) as PeerStatic;
-
-  return new transport(socket, connectionUris);
-}
-
-registerTransport("tcp", TcpPeer);
-registerTransport("ws", WebSocketPeer);
-registerTransport("wss", WebSocketPeer);

src/types.ts

@@ -3,14 +3,13 @@ import KeyPairEd25519 from "#ed25519.js";
 import { AbstractLevel } from "abstract-level";
 import { P2PService } from "./service/p2p.js";
 import { RegistryService } from "./service/registry.js";
-import { S5Node } from "#node.js";
-import Unpacker from "#serialization/unpack.js";
 
 export interface Peer {
   id: NodeId;
   connectionUris: Array<URL>;
   isConnected: boolean;
   challenge: Uint8Array;
+
   sendMessage(message: Uint8Array): void;
 
   listenForMessages(
@@ -29,12 +28,6 @@ export interface Peer {
   renderLocationUri(): string;
 }
 
-// Define the static side of the class
-export interface PeerStatic {
-  new (_socket: any, uri: URL[]): Peer;
-  connect(uri: URL): Promise<any>;
-}
-
 export interface Logger {
   info(s: string): void;
   verbose(s: string): void;
@@ -65,19 +58,3 @@ export interface SignedMessage {
   nodeId: NodeId;
   message: Uint8Array;
 }
-
-export type P2PMessageHandler = (
-  node: S5Node,
-  peer: Peer,
-  data: Unpacker,
-  rawData: Uint8Array,
-  verifyId: boolean,
-) => Promise<void>;
-
-export type P2PSignedMessageHandler = (
-  node: S5Node,
-  peer: Peer,
-  data: Unpacker,
-  message: SignedMessage,
-  verifyId: boolean,
-) => Promise<void>;