remove libp2p, use only kubo

README.md

@@ -42,5 +42,4 @@ npx tsx src/indexer/start.ts
 
 ## TODO
 
-When using Kubo node, libp2p is not needed at all.
 Insert pubkey instead of ss58 address if we want data to be compatible across networks

package.json

@@ -13,19 +13,11 @@
     "format": "prettier --write src/"
   },
   "dependencies": {
-    "@chainsafe/libp2p-gossipsub": "^13.0.0",
-    "@chainsafe/libp2p-noise": "^15.0.0",
-    "@libp2p/bootstrap": "^10.0.16",
-    "@libp2p/mplex": "^10.0.18",
-    "@libp2p/ping": "^1.0.14",
-    "@libp2p/tcp": "^9.0.18",
-    "@libp2p/webtransport": "^4.0.21",
     "@multiformats/multiaddr": "^12.2.1",
     "@polkadot/util": "^12.6.2",
     "@polkadot/util-crypto": "^12.6.2",
     "buffer": "^6.0.3",
     "kubo-rpc-client": "^3.0.4",
-    "libp2p": "^1.3.0",
     "multiformats": "^13.1.0",
     "pg": "^8.11.5",
     "pretty-bytes": "^6.1.1",
@@ -33,8 +25,6 @@
     "vue-router": "^4.3.0"
   },
   "devDependencies": {
-    "@libp2p/interface": "^1.1.4",
-    "@libp2p/interface-pubsub": "^4.0.1",
     "@rushstack/eslint-patch": "^1.3.3",
     "@tsconfig/node20": "^20.1.2",
     "@types/node": "^20.11.28",

src/collector.ts

 import { decodeAddress, signatureVerify } from '@polkadot/util-crypto'
 import { u8aToHex } from '@polkadot/util'
-import { type Message } from '@libp2p/interface-pubsub'
 import { CID } from 'multiformats'
 import { kubo } from './kubo'
 import { bigintToUint8Array } from './utils'
 import type { IndexRequest } from './types'
-import { addToIndex } from './processor'
+
+// === payload verification
 
 // build payload to sign
 export function buildBytesPayload(ir: IndexRequest) {
@@ -23,46 +23,48 @@ export function isValidSignature(signedMessage: Uint8Array, signature: string, a
   return signatureVerify(signedMessage, signature, hexPublicKey).isValid
 }
 
-// ===
+// === message handling
 
-// message handling function
-// TODO allow more control on what to do on messages (filtering, trust list...)
-// NOTE: currently unused, more like a template
-function handleMessage(message: CustomEvent<Message>) {
-  const msg = new TextDecoder().decode(message.detail.data)
-  //   const log = `[${message.detail.topic}]: ` + msg
-  //   console.log(log)
+// returns a pubsub message handler
+// TODO allow more control on what to do from invalid message (instead of simple logging)
+//   so that it could be used by the app message handler
+// TODO allow more control on what to do before processing valid messages (filtering, trust list...)
+//   this allows to give more details on what is considered valid
+export function getPubSubHandler(validMessageHandler: (cid: CID, dag: IndexRequest) => void) {
+  return (message: any) => {
+    // decode message as text
+    const msg = new TextDecoder().decode(message.data).trim()
 
-  try {
-    const cid = CID.parse(msg)
-    console.log(msg)
+    try {
+      const cid = CID.parse(msg)
 
-    kubo.dag.get(cid).then(function (d) {
-      const dag = d.value as IndexRequest
-      // TODO some validation on dag to ensure it is exact format
-      // TODO some validation on timestamp to prevent too much control on the key
-      //   example: 0 < Date.now() - timestamp < 1 minute
-      const bytesPayload = buildBytesPayload(dag)
-      const isValid = isValidSignature(bytesPayload, dag.sig, dag.pubk)
-      if (isValid) {
-        // at this point we can apply different treatment based on the key
-        // we could for example have a trust list published in ipfs
-        // this is mostly for spam prevention
-        // but for the moment we will just assume that we trust any key and add the index request to the pool
+      kubo.dag.get(cid).then(function (d) {
+        const dag = d.value as IndexRequest
+        // TODO some validation on dag to ensure it is exact format
+        // TODO some validation on timestamp to prevent too much control on the key
+        //   example: 0 < Date.now() - timestamp < 1 minute
+        const bytesPayload = buildBytesPayload(dag)
+        const isValid = isValidSignature(bytesPayload, dag.sig, dag.pubk)
+        if (isValid) {
+          // at this point we can apply different treatment based on the key
+          // we could for example have a trust list published in ipfs
+          // this is mostly for spam prevention
+          // but for the moment we will just assume that we trust any key and add the index request to the pool
 
-        // we can also apply different treatment based on the key:
-        // high trust => common timestamp AMT
-        // low trust => sandboxed HAMT
+          // we can also apply different treatment based on the key:
+          // high trust => common timestamp AMT
+          // low trust => sandboxed HAMT
 
-        addToIndex(cid, dag)
+          validMessageHandler(cid, dag)
 
-        // this is all that this indexer does
-        // the rest (autopinning, postgres indexing... will be managed by an other part of the indexer)
-      } else {
-        console.log('[invalid sig] ' + msg)
-      }
-    })
-  } catch {
-    console.log('[invalid] ' + msg)
+          // this is all that this indexer does
+          // the rest (autopinning, postgres indexing... will be managed by an other part of the indexer)
+        } else {
+          console.log('[invalid sig] ' + msg)
+        }
+      })
+    } catch {
+      console.log('[invalid] ' + msg)
+    }
   }
 }

src/consts.ts

 import { CID } from 'multiformats'
 
+// topic used for diffusion of index requests
+export const TOPIC = 'ddd'
+
 // base used for index key
 export const BASE = 16
 // current root index

src/indexer/start.ts

-import { TOPIC } from '../p2p'
+import { TOPIC } from '../consts'
 import { timestampToKey, mergeInodesSync, arrayToVinode, publishHistory } from '../processor'
-import { buildBytesPayload, isValidSignature } from '../collector'
+import { getPubSubHandler } from '../collector'
 import { kubo } from '../kubo'
 import type { IndexRequest } from '../types'
 import { CID } from 'multiformats'
@@ -11,45 +11,10 @@ import { getRootCIDfromArgs } from './utils'
 // === HANDLERS ===
 
 // pubsub message handler
-function handleMessage(message: any) {
-  // console.log('received message')
-  // console.log(message)
-  const msg = new TextDecoder().decode(message.data).trim()
-
-  try {
-    const cid = CID.parse(msg)
-    // console.log(msg)
-
-    kubo.dag.get(cid).then(function (d) {
-      const dag = d.value as IndexRequest
-      // TODO some validation on dag to ensure it is exact format
-      // TODO some validation on timestamp to prevent too much control on the key
-      //   example: 0 < Date.now() - timestamp < 1 minute
-      const bytesPayload = buildBytesPayload(dag)
-      const isValid = isValidSignature(bytesPayload, dag.sig, dag.pubk)
-      if (isValid) {
-        // at this point we can apply different treatment based on the key
-        // we could for example have a trust list published in ipfs
-        // this is mostly for spam prevention
-        // but for the moment we will just assume that we trust any key and add the index request to the pool
-
-        // we can also apply different treatment based on the key:
-        // high trust => common timestamp AMT
-        // low trust => sandboxed HAMT
-
-        processQueue.push([cid, dag])
-        events.emit(evtype.trigger)
-
-        // this is all that this indexer does
-        // the rest (autopinning, postgres indexing... will be managed by an other part of the indexer)
-      } else {
-        console.log('[invalid sig] ' + msg)
-      }
-    })
-  } catch {
-    console.log('[invalid] ' + msg)
-  }
-}
+const handleMessage = getPubSubHandler((cid, dag) => {
+  processQueue.push([cid, dag])
+  events.emit(evtype.trigger)
+})
 
 // queue event handler
 function handleBatch() {

src/main.ts

@@ -6,28 +6,21 @@ import router from './router'
 import { feed } from './global'
 
 // collector dependencies
-import { local_peer, TOPIC, libp2pForBrowser } from './p2p'
-import type { PubSub } from '@libp2p/interface'
-import { multiaddr } from '@multiformats/multiaddr'
+import { TOPIC } from './consts'
 import { CID } from 'multiformats'
 import { kubo } from './kubo'
 import type { IndexRequest } from './types'
 import { buildBytesPayload, isValidSignature } from './collector'
-import { addToIndex } from './processor'
 
-// start libp2p
-// there is no need to do that if we have a kubo node other than demo purpose
-const libp2p = await libp2pForBrowser()
-await libp2p.start()
-await libp2p.dial(local_peer.map(multiaddr))
-const pubsub = libp2p.services.pubsub as PubSub
+// activate this for debug purpose if you want to do indexing directly from the browser app
+// instead of the nodejs proces
+// import { addToIndex } from './processor'
 
 // === define what to do of pubsub messages and start listening ===
 
 // message handler used for demo purpose when in app mode, to display results in the feed
-pubsub.addEventListener('message', (message) => {
-  const msg = new TextDecoder().decode(message.detail.data)
-  // const log = `[${message.detail.topic}]: ` + msg
+function handleMessage(message: any) {
+  const msg = new TextDecoder().decode(message.data).trim()
   try {
     const cid = CID.parse(msg)
     feed.value.push(msg)
@@ -45,9 +38,9 @@ pubsub.addEventListener('message', (message) => {
   } catch {
     feed.value.push('[invalid] ' + msg)
   }
-})
+}
 // subscribe
-pubsub.subscribe(TOPIC)
+kubo.pubsub.subscribe(TOPIC, handleMessage)
 
 // === APP UI ===
 const app = createApp(App)

src/p2p.ts

-import { createLibp2p } from 'libp2p'
-import { bootstrap } from '@libp2p/bootstrap'
-import { gossipsub } from '@chainsafe/libp2p-gossipsub'
-import { webTransport } from '@libp2p/webtransport'
-import type { Libp2p } from '@libp2p/interface'
-
-export const local_peer = [
-  '/ip4/127.0.0.1/tcp/4001/p2p/12D3KooWF44SaSomGuUSNycgpkRwQcxcsMYeNbtn6XCHPR2ojodv',
-  '/ip4/127.0.0.1/udp/4001/quic-v1/p2p/12D3KooWF44SaSomGuUSNycgpkRwQcxcsMYeNbtn6XCHPR2ojodv',
-  '/ip4/127.0.0.1/udp/4001/quic-v1/webtransport/certhash/uEiCG3LMdW9dtRC_kXft2TlE66kZ4RX1MzQksLwxRgbJMVw/certhash/uEiC6q4csXlqjvPg_z28SK2fu59NKwOLBDud7j-6Cdn958Q/p2p/12D3KooWF44SaSomGuUSNycgpkRwQcxcsMYeNbtn6XCHPR2ojodv'
-]
-
-export const TOPIC = 'ddd'
-
-// return libp2p object compatible with browser transport
-export function libp2pForBrowser(): Promise<Libp2p> {
-  return createLibp2p({
-    peerDiscovery: [
-      bootstrap({
-        list: local_peer
-      })
-    ],
-    transports: [webTransport()],
-    services: {
-      pubsub: gossipsub()
-    },
-    connectionGater: {
-      denyDialMultiaddr: async () => false
-    }
-  })
-}

src/views/FeedView.vue

 <script setup lang="ts">
 import { feed } from '@/global'
 import { ref, type Ref, onMounted } from 'vue'
-import { TOPIC, libp2pForBrowser } from '@/p2p'
-import type { PeerId } from '@libp2p/interface'
 import { CID } from 'multiformats'
 import FeedRow from '../components/FeedRow.vue'
 import { kubo } from '../kubo'
+import { TOPIC } from '../consts'
 import type { IndexRequest } from '../types'
-import type { PubSub } from '@libp2p/interface'
 
-const subscribers: Ref<PeerId[]> = ref([])
-let pubsub: PubSub | undefined = undefined
+const subscribers: Ref<string[]> = ref([])
 
-function subscribe() {
-  if (pubsub) pubsub.subscribe(TOPIC)
-}
-function getSubscribers() {
-  if (pubsub) subscribers.value = pubsub.getSubscribers(TOPIC)
+async function getSubscribers() {
+  subscribers.value = (await kubo.pubsub.peers(TOPIC)).map((e) => e.toString())
 }
 function get_dag(cid: CID): Promise<IndexRequest> {
   return kubo.dag.get(cid).then((d) => d.value)
 }
-
-function start() {
-  libp2pForBrowser().then((libp2p) => (pubsub = libp2p.services.pubsub as PubSub))
-}
-
-onMounted(start)
 </script>
 
 <template>
   <div>
     <h1>Feed</h1>
     <p>Pubsub feed</p>
-    <p>topic: {{ TOPIC }} <button @click="subscribe">subscribe</button></p>
-    <p>subscriber list <button @click="getSubscribers">get list</button></p>
+    <p>topic "{{ TOPIC }}" subscriber list <button @click="getSubscribers">get</button></p>
     <ul>
       <li v-for="s in subscribers" :key="s.toString()">{{ s }}</li>
     </ul>