Let agents discover newly joined hubs

Author: pajama-coder

agent/db.js

@@ -116,6 +116,50 @@ function setHubs(zone, hubs) {
   )
 }
 
+function getHub(id) {
+  return (
+    db.sql('SELECT zone, info FROM hubs WHERE id = ?')
+      .bind(1, id)
+      .exec()
+      .map(r => {
+        try {
+          var hub = JSON.parse(r.info)
+        } catch {
+          var hub = {}
+        }
+        hub.zone = r.zone
+        return hub
+      })[0]
+  )
+}
+
+function setHub(id, hub) {
+  var old = getHub(id)
+  if (old) {
+    var zone = hub.zone || old.zone
+    var info = {
+      ports: hub.ports || old.ports,
+      version: hub.version || old.version,
+    }
+    db.sql('UPDATE hubs SET zone = ?, info = ? WHERE id = ?')
+      .bind(1, zone)
+      .bind(2, JSON.stringify(info))
+      .bind(3, id)
+      .exec()
+  } else {
+    var zone = hub.zone
+    var info = {
+      ports: hub.ports,
+      version: hub.version,
+    }
+    db.sql('INSERT INTO hubs(id, zone, info) VALUES(?, ?, ?)')
+      .bind(1, id)
+      .bind(2, zone)
+      .bind(3, JSON.stringify(info))
+      .exec()
+  }
+}
+
 function recordToMesh(rec) {
   return {
     name: rec.name,
@@ -325,6 +369,8 @@ export default {
   setZones,
   allHubs,
   setHubs,
+  getHub,
+  setHub,
   allMeshes,
   getMesh,
   setMesh,

agent/mesh.js

@@ -682,6 +682,21 @@ export default function (rootDir, listen, config, onConfigUpdate) {
         },
       },
 
+      '/api/hubs/{id}': {
+        'POST': function ({ id }, req) {
+          var info = JSON.decode(req.body)
+          var hub = {
+            zone: info.zone,
+            ports: info.ports,
+            version: info.version,
+          }
+          hubCache.set(id, hub)
+          db.setHub(id, hub)
+          logInfo(`Discovered new hub ${id} in zone '${hub.zone}' with ports: ${info.ports.join(', ')}`)
+          return response(201)
+        },
+      },
+
       '/api/file-data/{hash}': {
         'GET': function ({ hash }) {
           var data = fs.raw(hash)

hub/main.js

@@ -330,6 +330,7 @@ var $hub = null
 var $hubSelected = null
 var $sendEOS = null
 var $sessionID
+var $broadcastID
 var $pingID
 var $pingTime
 
@@ -405,6 +406,7 @@ function start(listen, bootstrap) {
         file: (path, info, ep) => updateFileInfo(path, info, ep, true),
         acl: (username, path, access, since) => updateACL(username, path, access, since),
         eviction: (username, time) => updateEviction(username, time),
+        hub: (id, info) => broadcastHub(id, info),
       },
       log,
     })
@@ -1013,6 +1015,25 @@ var muxToAgent = pipeline($=>$
   )
 )
 
+var broadcastToAgents = pipeline($=>$
+  .onStart(msg => msg)
+  .forkJoin(() => Object.keys(sessions)).to($=>$
+    .onStart(id => { $broadcastID = id })
+    .forkJoin(() => {
+      var hubs = []
+      sessions[$broadcastID].forEach(h => hubs.push(h))
+      return hubs
+    }).to($=>$
+      .onStart(hub => { $hubSelected = hub })
+      .pipe(muxToAgent)
+      .replaceData()
+      .replaceMessage(new StreamEnd)
+    )
+    .replaceMessage(new StreamEnd)
+  )
+  .replaceMessage(new StreamEnd)
+)
+
 var connectEndpoint = pipeline($=>$
   .acceptHTTPTunnel(
     function (req) {
@@ -1433,6 +1454,19 @@ function updateEviction(username, time, expiration) {
   return true
 }
 
+function broadcastHub(id, info) {
+  broadcastToAgents.spawn(
+    new Message({
+      method: 'POST',
+      path: `/api/hubs/${id}`,
+    }, JSON.encode({
+      zone: info.zone,
+      ports: info.ports,
+      version: info.version,
+    }))
+  )
+}
+
 function dumpACL() {
   var all = {}
   Object.entries(acl).forEach(