RPC/function delegation in hypercore with networking

hypercore-2way-RPC-demo.js

const Hypercore = require('hypercore')
const Hyperswarm = require('hyperswarm')
const crypto = require('hypercore-crypto')

// Available functions for delegation
const AVAILABLE_FUNCTIONS = {
  add: (a, b) => a + b,
  multiply: (a, b) => a * b,
  greet: (name) => `Hello, ${name}!`,
  getCurrentTime: () => new Date().toISOString(),
  processData: (data) => ({
    processed: true,
    length: data.length,
    uppercase: data.toUpperCase()
  }),
  fibonacci: (n) => {
    if (n <= 1) return n
    return AVAILABLE_FUNCTIONS.fibonacci(n - 1) + AVAILABLE_FUNCTIONS.fibonacci(n - 2)
  }
}

class DemoRunner {
  constructor() {
    this.cleanup = []
  }

  log(message, type = 'info') {
    const timestamp = new Date().toISOString().substr(11, 8)
    const prefix = {
      info: '📝',
      success: '✅',
      error: '❌',
      delegator: '📤',
      executor: '⚡',
      peer1: '🔵',
      peer2: '🟣'
    }[type] || '📝'
    
    console.log(`[${timestamp}] ${prefix} ${message}`)
  }

  async sleep(ms) {
    return new Promise(resolve => setTimeout(resolve, ms))
  }

  async runOneWayDemo() {
    this.log('='.repeat(60), 'info')
    this.log('DEMO 1: ONE-WAY FUNCTION DELEGATION', 'info')
    this.log('='.repeat(60), 'info')
    
    // Create delegator (sends function calls)
    const delegatorFeed = new Hypercore(`./demo-oneway-delegator-${Date.now()}`, {
      valueEncoding: 'json'
    })
    await delegatorFeed.ready()
    
    const delegatorSwarm = new Hyperswarm()
    this.cleanup.push(() => delegatorSwarm.destroy())
    this.cleanup.push(() => delegatorFeed.close())
    
    // Create executor (receives and runs function calls)
    const executorFeed = new Hypercore(`./demo-oneway-executor-${Date.now()}`, delegatorFeed.key, {
      valueEncoding: 'json'
    })
    await executorFeed.ready()
    
    const executorSwarm = new Hyperswarm()
    this.cleanup.push(() => executorSwarm.destroy())
    this.cleanup.push(() => executorFeed.close())
    
    this.log(`Feed key: ${delegatorFeed.key.toString('hex')}`, 'info')
    
    // Set up executor to listen for function calls
    let executedCount = 0
    const executeFunction = async (entry) => {
      const { id, function: funcName, args, timestamp } = entry
      
      this.log(`Received: ${funcName}(${args.join(', ')}) [ID: ${id.substr(0, 8)}...]`, 'executor')
      
      try {
        if (AVAILABLE_FUNCTIONS[funcName]) {
          const result = AVAILABLE_FUNCTIONS[funcName](...args)
          this.log(`Executed -> ${JSON.stringify(result)}`, 'success')
          executedCount++
        } else {
          this.log(`Unknown function: ${funcName}`, 'error')
        }
      } catch (error) {
        this.log(`Execution error: ${error.message}`, 'error')
      }
    }
    
    // Process existing entries
    for (let i = 0; i < executorFeed.length; i++) {
      const entry = await executorFeed.get(i)
      await executeFunction(entry)
    }
    
    // Listen for new entries
    executorFeed.on('append', async () => {
      const entry = await executorFeed.get(executorFeed.length - 1)
      await executeFunction(entry)
    })
    
    // Set up networking
    delegatorSwarm.join(delegatorFeed.discoveryKey)
    executorSwarm.join(executorFeed.discoveryKey)
    
    let connected = false
    delegatorSwarm.on('connection', (connection) => {
      if (!connected) {
        this.log('Delegator connected to executor', 'success')
        connected = true
      }
      delegatorFeed.replicate(connection)
    })
    
    executorSwarm.on('connection', (connection) => {
      executorFeed.replicate(connection)
    })
    
    // Wait for connection
    await this.sleep(2000)
    
    // Send function calls from delegator
    const calls = [
      { func: 'add', args: ['5', '3'] },
      { func: 'multiply', args: ['4', '7'] },
      { func: 'greet', args: ['Alice'] },
      { func: 'getCurrentTime', args: [] },
      { func: 'processData', args: ['hello-world'] },
      { func: 'fibonacci', args: ['7'] }
    ]
    
    this.log('Starting function delegation...', 'delegator')
    
    for (const call of calls) {
      const functionCall = {
        id: crypto.randomBytes(16).toString('hex'),
        function: call.func,
        args: call.args,
        timestamp: Date.now()
      }
      
      this.log(`Delegating: ${call.func}(${call.args.join(', ')})`, 'delegator')
      await delegatorFeed.append(functionCall)
      await this.sleep(1000)
    }
    
    // Wait for all executions to complete
    await this.sleep(2000)
    
    this.log(`One-way demo completed! Executed ${executedCount} functions`, 'success')
    
    // Cleanup
    await delegatorSwarm.destroy()
    await executorSwarm.destroy()
    await delegatorFeed.close()
    await executorFeed.close()
  }

  async runBidirectionalDemo() {
    this.log('='.repeat(60), 'info')
    this.log('DEMO 2: BIDIRECTIONAL FUNCTION DELEGATION', 'info')
    this.log('='.repeat(60), 'info')
    
    // Create feeds for requests and responses
    const timestamp = Date.now()
    const requestsFeed = new Hypercore(`./demo-requests-${timestamp}`, {
      valueEncoding: 'json'
    })
    const responsesFeed = new Hypercore(`./demo-responses-${timestamp}`, {
      valueEncoding: 'json'
    })
    
    await requestsFeed.ready()
    await responsesFeed.ready()
    
    this.log(`Requests feed: ${requestsFeed.key.toString('hex')}`, 'info')
    this.log(`Responses feed: ${responsesFeed.key.toString('hex')}`, 'info')
    
    // Create two peers
    const peer1 = await this.createBidirectionalPeer('Peer1', requestsFeed, responsesFeed)
    const peer2 = await this.createBidirectionalPeer('Peer2', requestsFeed, responsesFeed)
    
    // Wait for connections
    await this.sleep(2000)
    
    // Demo sequence: Peer1 calls functions on Peer2, then vice versa
    this.log('Starting bidirectional function calls...', 'info')
    
    const demo1Calls = [
      { peer: peer1, func: 'add', args: ['10', '5'] },
      { peer: peer1, func: 'greet', args: ['Bob'] },
      { peer: peer1, func: 'fibonacci', args: ['6'] }
    ]
    
    for (const call of demo1Calls) {
      try {
        this.log(`${call.peer.name} calling: ${call.func}(${call.args.join(', ')})`, 'peer1')
        const result = await call.peer.callRemoteFunction(call.func, ...call.args)
        this.log(`${call.peer.name} received result: ${JSON.stringify(result)}`, 'success')
      } catch (error) {
        this.log(`${call.peer.name} call failed: ${error.message}`, 'error')
      }
      await this.sleep(1500)
    }
    
    await this.sleep(1000)
    
    const demo2Calls = [
      { peer: peer2, func: 'multiply', args: ['3', '8'] },
      { peer: peer2, func: 'processData', args: ['test-data'] },
      { peer: peer2, func: 'getCurrentTime', args: [] }
    ]
    
    for (const call of demo2Calls) {
      try {
        this.log(`${call.peer.name} calling: ${call.func}(${call.args.join(', ')})`, 'peer2')
        const result = await call.peer.callRemoteFunction(call.func, ...call.args)
        this.log(`${call.peer.name} received result: ${JSON.stringify(result)}`, 'success')
      } catch (error) {
        this.log(`${call.peer.name} call failed: ${error.message}`, 'error')
      }
      await this.sleep(1500)
    }
    
    this.log('Bidirectional demo completed!', 'success')
    
    // Cleanup
    await peer1.cleanup()
    await peer2.cleanup()
    await requestsFeed.close()
    await responsesFeed.close()
  }

  async createBidirectionalPeer(name, requestsFeed, responsesFeed) {
    const swarm = new Hyperswarm()
    const pendingRequests = new Map()
    const processedRequests = new Set()
    
    // Join swarms
    swarm.join(requestsFeed.discoveryKey)
    swarm.join(responsesFeed.discoveryKey)
    
    swarm.on('connection', (connection) => {
      requestsFeed.replicate(connection)
      responsesFeed.replicate(connection)
    })
    
    // Process requests
    const processRequest = async (request) => {
      const { id, function: funcName, args, from } = request
      
      if (from === name || processedRequests.has(id)) {
        return
      }
      
      processedRequests.add(id)
      
      this.log(`${name} executing: ${funcName}(${args.join(', ')}) for ${from}`, name.toLowerCase())
      
      try {
        if (AVAILABLE_FUNCTIONS[funcName]) {
          const result = AVAILABLE_FUNCTIONS[funcName](...args)
          
          const response = {
            id: crypto.randomBytes(16).toString('hex'),
            requestId: id,
            result,
            from: name,
            to: from,
            timestamp: Date.now(),
            success: true
          }
          
          await responsesFeed.append(response)
        }
      } catch (error) {
        const response = {
          id: crypto.randomBytes(16).toString('hex'),
          requestId: id,
          error: error.message,
          from: name,
          to: from,
          timestamp: Date.now(),
          success: false
        }
        
        await responsesFeed.append(response)
      }
    }
    
    // Process responses
    const processResponse = (response) => {
      const { requestId, result, error, to, success } = response
      
      if (to !== name) return
      
      if (pendingRequests.has(requestId)) {
        const { resolve, reject } = pendingRequests.get(requestId)
        
        if (success) {
          resolve(result)
        } else {
          reject(new Error(error))
        }
        
        pendingRequests.delete(requestId)
      }
    }
    
    // Set up listeners
    requestsFeed.on('append', async () => {
      const request = await requestsFeed.get(requestsFeed.length - 1)
      await processRequest(request)
    })
    
    responsesFeed.on('append', async () => {
      const response = await responsesFeed.get(responsesFeed.length - 1)
      processResponse(response)
    })
    
    // Process existing entries
    for (let i = 0; i < requestsFeed.length; i++) {
      const request = await requestsFeed.get(i)
      await processRequest(request)
    }
    
    for (let i = 0; i < responsesFeed.length; i++) {
      const response = await responsesFeed.get(i)
      processResponse(response)
    }
    
    return {
      name,
      swarm,
      async callRemoteFunction(functionName, ...args) {
        return new Promise(async (resolve, reject) => {
          const requestId = crypto.randomBytes(16).toString('hex')
          
          const request = {
            id: requestId,
            function: functionName,
            args,
            from: name,
            timestamp: Date.now()
          }
          
          pendingRequests.set(requestId, { resolve, reject })
          
          setTimeout(() => {
            if (pendingRequests.has(requestId)) {
              pendingRequests.delete(requestId)
              reject(new Error('Request timeout'))
            }
          }, 5000)
          
          await requestsFeed.append(request)
        })
      },
      async cleanup() {
        await swarm.destroy()
      }
    }
  }

  async run() {
    console.log('🚀 Function Delegation Demo')
    console.log('===========================')
    console.log('This demo shows how to delegate function calls between peers')
    console.log('using hypercore feeds for communication.\n')
    
    try {
      await this.runOneWayDemo()
      await this.sleep(2000)
      await this.runBidirectionalDemo()
      
      this.log('🎉 All demos completed successfully!', 'success')
      
    } catch (error) {
      this.log(`Demo failed: ${error.message}`, 'error')
      console.error(error)
    } finally {
      // Cleanup any remaining resources
      for (const cleanup of this.cleanup) {
        try {
          await cleanup()
        } catch (e) {
          // Ignore cleanup errors
        }
      }
      
      // Clean up demo files
      try {
        const fs = require('fs')
        const path = require('path')
        const files = fs.readdirSync('.')
        for (const file of files) {
          if (file.startsWith('demo-')) {
            fs.rmSync(file, { recursive: true, force: true })
          }
        }
      } catch (e) {
        // Ignore cleanup errors
      }
      
      process.exit(0)
    }
  }
}

// Run the demo
const demo = new DemoRunner()
demo.run().catch(console.error)

output.log

➜  function-delegation node hypercore-2way-RPC-demo.js
🚀 Function Delegation Demo
===========================
This demo shows how to delegate function calls between peers
using hypercore feeds for communication.

[14:57:27] 📝 ============================================================
[14:57:27] 📝 DEMO 1: ONE-WAY FUNCTION DELEGATION
[14:57:27] 📝 ============================================================
[14:57:27] 📝 Feed key: 6dda8663ffa91679e4ade0add05d218c012ffbb6ff2abf60b42ab11c752e0bfa
[14:57:29] 📤 Starting function delegation...
[14:57:29] 📤 Delegating: add(5, 3)
[14:57:30] 📤 Delegating: multiply(4, 7)
[14:57:31] 📤 Delegating: greet(Alice)
[14:57:32] 📤 Delegating: getCurrentTime()
[14:57:33] 📤 Delegating: processData(hello-world)
[14:57:34] ✅ Delegator connected to executor
[14:57:34] ⚡ Received: processData(hello-world) [ID: e819a8f0...]
[14:57:34] ✅ Executed -> {"processed":true,"length":11,"uppercase":"HELLO-WORLD"}
[14:57:34] 📤 Delegating: fibonacci(7)
[14:57:34] ⚡ Received: fibonacci(7) [ID: 975ee27a...]
[14:57:34] ✅ Executed -> 13
[14:57:37] ✅ One-way demo completed! Executed 2 functions
[14:57:42] 📝 ============================================================
[14:57:42] 📝 DEMO 2: BIDIRECTIONAL FUNCTION DELEGATION
[14:57:42] 📝 ============================================================
[14:57:42] 📝 Requests feed: 2c0640e4cf400cefce6b536b8823f3f8721a8d02f185f7fc572e56ff4555dbbe
[14:57:42] 📝 Responses feed: 450afb2921c24d336fc4f3551774de20bdcc90a7ae742243afc6b46e3e9021e8
[14:57:44] 📝 Starting bidirectional function calls...
[14:57:44] 🔵 Peer1 calling: add(10, 5)
[14:57:44] 🟣 Peer2 executing: add(10, 5) for Peer1
[14:57:44] ✅ Peer1 received result: "105"
[14:57:46] 🔵 Peer1 calling: greet(Bob)
[14:57:46] 🟣 Peer2 executing: greet(Bob) for Peer1
[14:57:46] ✅ Peer1 received result: "Hello, Bob!"
[14:57:47] 🔵 Peer1 calling: fibonacci(6)
[14:57:47] 🟣 Peer2 executing: fibonacci(6) for Peer1
[14:57:47] ✅ Peer1 received result: 8
[14:57:50] 🟣 Peer2 calling: multiply(3, 8)
[14:57:50] 🔵 Peer1 executing: multiply(3, 8) for Peer2
[14:57:50] ✅ Peer2 received result: 24
[14:57:51] 🟣 Peer2 calling: processData(test-data)
[14:57:51] 🔵 Peer1 executing: processData(test-data) for Peer2
[14:57:51] ✅ Peer2 received result: {"processed":true,"length":9,"uppercase":"TEST-DATA"}
[14:57:53] 🟣 Peer2 calling: getCurrentTime()
[14:57:53] 🔵 Peer1 executing: getCurrentTime() for Peer2
[14:57:53] ✅ Peer2 received result: "2025-06-03T14:57:53.442Z"
[14:57:54] ✅ Bidirectional demo completed!
[14:57:58] ✅ 🎉 All demos completed successfully!
[14:57:58] 📝 Check the examples in this directory to see the full implementations.