refactored globals into classes

README.md

@@ -35,7 +35,8 @@ Then if their clocks drift relative to ours, we can seek consensus among a broad
 But at that point just run ntpd. Can still do consensus to verify
 but probably no need to implement custom time synchronization protocol.
 
-Wait NTP is centralized isn't it, not peer to peer...
+Apparently PTP, Precision Time Protocol, is a thing.
+PTP affords for a layer of user defined priority for best clock selection.
 
 ## Peering
 
@@ -118,7 +119,7 @@ To demonstrate the example application, you can open multiple terminals, and in
 export DEBUG="*,-express"
 export RHIZOME_REQUEST_BIND_PORT=4000
 export RHIZOME_PUBLISH_BIND_PORT=4001
-export RHIZOME_SEED_PEERS='127.0.0.1:4002, 127.0.0.1:4004'
+export RHIZOME_SEED_PEERS='localhost:4002, localhost:4004'
 export RHIZOME_HTTP_API_PORT=3000
 export RHIZOME_PEER_ID=peer1
 npm run example-app
@@ -128,7 +129,7 @@ npm run example-app
 export DEBUG="*,-express"
 export RHIZOME_REQUEST_BIND_PORT=4002
 export RHIZOME_PUBLISH_BIND_PORT=4003
-export RHIZOME_SEED_PEERS='127.0.0.1:4000, 127.0.0.1:4004'
+export RHIZOME_SEED_PEERS='localhost:4000, localhost:4004'
 export RHIZOME_HTTP_API_PORT=3001
 export RHIZOME_PEER_ID=peer2
 npm run example-app
@@ -138,7 +139,7 @@ npm run example-app
 export DEBUG="*,-express"
 export RHIZOME_REQUEST_BIND_PORT=4004
 export RHIZOME_PUBLISH_BIND_PORT=4005
-export RHIZOME_SEED_PEERS='127.0.0.1:4000, 127.0.0.1:4002'
+export RHIZOME_SEED_PEERS='localhost:4000, localhost:4002'
 export RHIZOME_HTTP_API_PORT=3002
 export RHIZOME_PEER_ID=peer3
 npm run example-app

__tests__/run.ts

@@ -0,0 +1,60 @@
+import Debug from 'debug';
+import {RhizomeNode, RhizomeNodeConfig} from "../src/node";
+import {TypedCollection} from '../src/typed-collection';
+const debug = Debug('test:run');
+
+type User = {
+  id?: string;
+  name: string;
+  nameLong?: string;
+  email?: string;
+  age: number;
+};
+
+class App extends RhizomeNode {
+  constructor(config?: Partial<RhizomeNodeConfig>) {
+    super(config);
+    const users = new TypedCollection<User>("users");
+    users.rhizomeConnect(this);
+  }
+}
+
+describe('Run', () => {
+  let app: App;
+
+  beforeAll(async () => {
+    app = new App({
+      // TODO expose more conveniently as test config options
+      httpPort: 5000,
+      httpEnable: true,
+      requestBindPort: 5001,
+      publishBindPort: 5002,
+    });
+    await app.start();
+  });
+
+  afterAll(async () => {
+    debug('attempting to stop app');
+    await app.stop();
+  });
+
+  it('can put a new user', async () => {
+    const res = await fetch('http://localhost:5000/users', {
+      method: 'PUT',
+      headers: {'Content-Type': 'application/json'},
+      body: JSON.stringify({
+        name: "Peon",
+        id: "peon-1",
+        age: 263
+      })
+    });
+    const data = await res.json();
+    expect(data).toMatchObject({
+      properties: {
+        name: "Peon",
+        id: "peon-1",
+        age: 263
+      }
+    });
+  });
+});

src/collection.ts

@@ -3,12 +3,11 @@
 // It should enable operations like removing a property removes the value from the entities in the collection
 // It could then be further extended with e.g. table semantics like filter, sort, join
 
+import {randomUUID} from "node:crypto";
 import EventEmitter from "node:events";
-import { deltasAccepted, publishDelta, subscribeDeltas } from "./deltas";
+import {RhizomeNode} from "./node";
-import { Entity, EntityProperties, EntityPropertiesDeltaBuilder } from "./object-layer";
+import {Entity, EntityProperties, EntityPropertiesDeltaBuilder} from "./object-layer";
-import { Delta } from "./types";
+import {Delta} from "./types";
-import { randomUUID } from "node:crypto";
-import {myRequestAddr} from "./peers";
 
 // type Property = {
 //   name: string,
@@ -24,13 +23,24 @@ import {myRequestAddr} from "./peers";
 // }
 
 export class Collection {
+  rhizomeNode?: RhizomeNode;
+  name: string;
   entities = new Map<string, Entity>();
   eventStream = new EventEmitter();
-  constructor() {
+
-    subscribeDeltas((delta: Delta) => {
+  constructor(name: string) {
+    this.name = name;
+  }
+
+  rhizomeConnect(rhizomeNode: RhizomeNode) {
+    this.rhizomeNode = rhizomeNode;
+
+    rhizomeNode.deltaStream.subscribeDeltas((delta: Delta) => {
       // TODO: Make sure this is the kind of delta we're looking for
       this.applyDelta(delta);
     });
+
+    rhizomeNode.httpApi.serveCollection(this);
   }
 
   // Applies the javascript rules for updating object values,
@@ -45,7 +55,7 @@ export class Collection {
       entity.id = entityId;
       eventType = 'create';
     }
-    const deltaBulider = new EntityPropertiesDeltaBuilder(entityId);
+    const deltaBulider = new EntityPropertiesDeltaBuilder(this.rhizomeNode!, entityId);
 
     if (!properties) {
       // Let's interpret this as entity deletion
@@ -135,9 +145,9 @@ export class Collection {
     const deltas: Delta[] = [];
     const entity = this.updateEntity(entityId, properties, true, deltas);
     deltas.forEach(async (delta: Delta) => {
-      delta.receivedFrom = myRequestAddr;
+      delta.receivedFrom = this.rhizomeNode!.myRequestAddr;
-      deltasAccepted.push(delta);
+      this.rhizomeNode!.deltaStream.deltasAccepted.push(delta);
-      await publishDelta(delta);
+      await this.rhizomeNode!.deltaStream.publishDelta(delta);
     });
     return entity;
   }
@@ -146,8 +156,8 @@ export class Collection {
     const deltas: Delta[] = [];
     this.updateEntity(entityId, undefined, true, deltas);
     deltas.forEach(async (delta: Delta) => {
-      deltasAccepted.push(delta);
+      this.rhizomeNode!.deltaStream.deltasAccepted.push(delta);
-      await publishDelta(delta);
+      await this.rhizomeNode!.deltaStream.publishDelta(delta);
     });
   }
 

src/config.ts

@@ -6,7 +6,7 @@ import {PeerAddress} from "./types";
 
 export const LEVEL_DB_DIR = process.env.RHIZOME_LEVEL_DB_DIR ?? './data';
 export const CREATOR = process.env.USER!;
-export const HOST_ID = process.env.RHIZOME_PEER_ID || randomUUID();
+export const PEER_ID = process.env.RHIZOME_PEER_ID || randomUUID();
 export const ADDRESS = process.env.RHIZOME_ADDRESS ?? 'localhost';
 export const REQUEST_BIND_ADDR = process.env.RHIZOME_REQUEST_BIND_ADDR || ADDRESS;
 export const REQUEST_BIND_PORT = parseInt(process.env.RHIZOME_REQUEST_BIND_PORT || '4000');
@@ -14,7 +14,7 @@ export const REQUEST_BIND_HOST = process.env.RHIZOME_REQUEST_BIND_HOST || REQUES
 export const PUBLISH_BIND_ADDR = process.env.RHIZOME_PUBLISH_BIND_ADDR || ADDRESS;
 export const PUBLISH_BIND_PORT = parseInt(process.env.RHIZOME_PUBLISH_BIND_PORT || '4001');
 export const PUBLISH_BIND_HOST = process.env.RHIZOME_PUBLISH_BIND_HOST || PUBLISH_BIND_ADDR;
-export const HTTP_API_ADDR = process.env.RHIZOME_HTTP_API_ADDR || ADDRESS || '127.0.0.1';
+export const HTTP_API_ADDR = process.env.RHIZOME_HTTP_API_ADDR || ADDRESS || 'localhost';
 export const HTTP_API_PORT = parseInt(process.env.RHIZOME_HTTP_API_PORT || '3000');
 export const HTTP_API_ENABLE = process.env.RHIZOME_HTTP_API_ENABLE === 'true';
 export const SEED_PEERS: PeerAddress[]  = (process.env.RHIZOME_SEED_PEERS || '').split(',')

src/deltas.ts

@@ -1,103 +1,94 @@
+import Debug from 'debug';
 import EventEmitter from 'node:events';
 import objectHash from 'object-hash';
-import {myRequestAddr} from './peers';
+import {RhizomeNode} from './node';
-import {publishSock, subscribeSock} from './pub-sub';
+import {Decision, Delta} from './types';
-import {Decision, Delta, PeerAddress} from './types';
-import Debug from 'debug';
 const debug = Debug('deltas');
 
-export const deltaStream = new EventEmitter();
+export class DeltaStream {
+  rhizomeNode: RhizomeNode;
+  deltaStream = new EventEmitter();
+  deltasProposed: Delta[] = [];
+  deltasAccepted: Delta[] = [];
+  deltasRejected: Delta[] = [];
+  deltasDeferred: Delta[] = [];
+  hashesReceived = new Set<string>();
 
-export const deltasProposed: Delta[] = [];
+  constructor(rhizomeNode: RhizomeNode) {
-export const deltasAccepted: Delta[] = [];
+    this.rhizomeNode = rhizomeNode;
-export const deltasRejected: Delta[] = [];
-export const deltasDeferred: Delta[] = [];
-
-export const hashesReceived = new Set<string>();
-
-export function applyPolicy(delta: Delta): Decision {
-  return !!delta && Decision.Accept;
-}
-
-export function receiveDelta(delta: Delta) {
-  // Deduplication: if we already received this delta, disregard it
-  const hash = objectHash(delta);
-  if (!hashesReceived.has(hash)) {
-    hashesReceived.add(hash);
-    deltasProposed.push(delta);
   }
-}
 
-export function ingestDelta(delta: Delta) {
+  applyPolicy(delta: Delta): Decision {
-  const decision = applyPolicy(delta);
+    return !!delta && Decision.Accept;
-  switch (decision) {
-    case Decision.Accept:
-      deltasAccepted.push(delta);
-      deltaStream.emit('delta', {delta});
-      break;
-    case Decision.Reject:
-      deltasRejected.push(delta);
-      break;
-    case Decision.Defer:
-      deltasDeferred.push(delta);
-      break;
   }
-}
 
-export function ingestNext(): boolean {
+  receiveDelta(delta: Delta) {
-  const delta = deltasProposed.shift();
+    // Deduplication: if we already received this delta, disregard it
-  if (!delta) {
+    const hash = objectHash(delta);
-    return false;
+    if (!this.hashesReceived.has(hash)) {
-  }
+      this.hashesReceived.add(hash);
-  ingestDelta(delta);
+      this.deltasProposed.push(delta);
-  return true;
-}
-
-export function ingestAll() {
-  while (ingestNext());
-}
-
-export function ingestNextDeferred(): boolean {
-  const delta = deltasDeferred.shift();
-  if (!delta) {
-    return false;
-  }
-  ingestDelta(delta);
-  return true;
-}
-
-export function ingestAllDeferred() {
-  while (ingestNextDeferred());
-}
-
-export function subscribeDeltas(fn: (delta: Delta) => void) {
-  deltaStream.on('delta', ({delta}) => {
-    fn(delta);
-  });
-}
-
-export async function publishDelta(delta: Delta) {
-  debug(`Publishing delta: ${JSON.stringify(delta)}`);
-  await publishSock.send(["deltas", myRequestAddr.toAddrString(), serializeDelta(delta)]);
-}
-
-function serializeDelta(delta: Delta) {
-  return JSON.stringify(delta);
-}
-
-function deserializeDelta(input: string) {
-  return JSON.parse(input);
-}
-
-export async function runDeltas() {
-  for await (const [topic, sender, msg] of subscribeSock) {
-    if (topic.toString() !== "deltas") {
-      continue;
     }
-    const delta = deserializeDelta(msg.toString());
+  }
-    delta.receivedFrom = PeerAddress.fromString(sender.toString());
+
-    debug(`Received delta: ${JSON.stringify(delta)}`);
+  ingestDelta(delta: Delta) {
-    ingestDelta(delta);
+    const decision = this.applyPolicy(delta);
+    switch (decision) {
+      case Decision.Accept:
+        this.deltasAccepted.push(delta);
+        this.deltaStream.emit('delta', {delta});
+        break;
+      case Decision.Reject:
+        this.deltasRejected.push(delta);
+        break;
+      case Decision.Defer:
+        this.deltasDeferred.push(delta);
+        break;
+    }
+  }
+
+  ingestNext(): boolean {
+    const delta = this.deltasProposed.shift();
+    if (!delta) {
+      return false;
+    }
+    this.ingestDelta(delta);
+    return true;
+  }
+
+  ingestAll() {
+    while (this.ingestNext());
+  }
+
+  ingestNextDeferred(): boolean {
+    const delta = this.deltasDeferred.shift();
+    if (!delta) {
+      return false;
+    }
+    this.ingestDelta(delta);
+    return true;
+  }
+
+  ingestAllDeferred() {
+    while (this.ingestNextDeferred());
+  }
+
+  subscribeDeltas(fn: (delta: Delta) => void) {
+    this.deltaStream.on('delta', ({delta}) => {
+      fn(delta);
+    });
+  }
+
+  async publishDelta(delta: Delta) {
+    debug(`Publishing delta: ${JSON.stringify(delta)}`);
+    await this.rhizomeNode.pubSub.publish("deltas", this.serializeDelta(delta));
+  }
+
+  serializeDelta(delta: Delta) {
+    return JSON.stringify(delta);
+  }
+
+  deserializeDelta(input: string) {
+    return JSON.parse(input);
   }
 }
-

src/example-app.ts

@@ -1,12 +1,7 @@
-import {HTTP_API_ENABLE} from "./config";
-import {runDeltas} from "./deltas";
-import {runHttpApi} from "./http-api";
-import {Entity} from "./object-layer";
-import {askAllPeersForDeltas, subscribeToSeeds} from "./peers";
-import {bindPublish, } from "./pub-sub";
-import {bindReply, runRequestHandlers} from "./request-reply";
-import {TypedCollection} from "./typed-collection";
 import Debug from 'debug';
+import {RhizomeNode} from "./node";
+import {Entity} from "./object-layer";
+import {TypedCollection} from "./typed-collection";
 const debug = Debug('example-app');
 
 // As an app we want to be able to write and read data.
@@ -23,21 +18,9 @@ type User = {
 };
 
 (async () => {
-  const users = new TypedCollection<User>();
+  const rhizomeNode = new RhizomeNode();
-
+  const users = new TypedCollection<User>("users");
-  await bindPublish();
+  users.rhizomeConnect(rhizomeNode);
-  await bindReply();
-  if (HTTP_API_ENABLE) {
-    runHttpApi({users});
-  }
-
-  runDeltas();
-  runRequestHandlers();
-  await new Promise((resolve) => setTimeout(resolve, 500));
-  subscribeToSeeds();
-  await new Promise((resolve) => setTimeout(resolve, 500));
-  askAllPeersForDeltas();
-  await new Promise((resolve) => setTimeout(resolve, 1000));
 
   users.onUpdate((u: Entity) => {
     debug('User updated:', u);
@@ -47,6 +30,9 @@ type User = {
     debug('New user!:', u);
   });
 
+
+  await rhizomeNode.start()
+
   const taliesin = users.put(undefined, {
     // id: 'taliesin-1',
     name: 'Taliesin',

src/http-api.ts

@@ -1,25 +1,22 @@
 import Debug from "debug";
 import express from "express";
+import {FSWatcher} from "fs";
 import {readdirSync, readFileSync, watch} from "fs";
+import {Server} from "http";
 import path, {join} from "path";
 import {Converter} from "showdown";
 import {Collection} from "./collection";
-import {HTTP_API_ADDR, HTTP_API_PORT} from "./config";
+import {RhizomeNode} from "./node";
-import {deltasAccepted} from "./deltas";
-import {peers} from "./peers";
 import {Delta} from "./types";
 const debug = Debug('http-api');
 
-type CollectionsToServe = {
-  [key: string]: Collection;
-};
-
 const docConverter = new Converter({
   completeHTMLDocument: true,
   // simpleLineBreaks: true,
   tables: true,
   tasklists: true
 });
+
 const htmlDocFromMarkdown = (md: string): string => docConverter.makeHtml(md);
 
 type mdFileInfo = {
@@ -31,6 +28,8 @@ type mdFileInfo = {
 class MDFiles {
   files = new Map<string, mdFileInfo>();
   readme?: mdFileInfo;
+  dirWatcher?: FSWatcher;
+  readmeWatcher?: FSWatcher;
 
   readFile(name: string) {
     const md = readFileSync(join('./markdown', `${name}.md`)).toString();
@@ -66,7 +65,7 @@ class MDFiles {
   }
 
   watchDir() {
-    watch('./markdown', null, (eventType, filename) => {
+    this.dirWatcher = watch('./markdown', null, (eventType, filename) => {
       if (!filename) return;
       if (!filename.endsWith(".md")) return;
 
@@ -91,7 +90,7 @@ class MDFiles {
   }
 
   watchReadme() {
-    watch('./README.md', null, (eventType, filename) => {
+   this.readmeWatcher = watch('./README.md', null, (eventType, filename) => {
       if (!filename) return;
 
       switch (eventType) {
@@ -104,127 +103,143 @@ class MDFiles {
       }
     });
   }
+
+  close() {
+    this.dirWatcher?.close();
+    this.readmeWatcher?.close();
+  }
 }
 
-export function runHttpApi(collections?: CollectionsToServe) {
+export class HttpApi {
-  const app = express();
+  rhizomeNode: RhizomeNode;
-  app.use(express.json());
+  app = express();
+  mdFiles = new MDFiles();
+  server?: Server;
 
-  // Get list of markdown files
+  constructor(rhizomeNode: RhizomeNode) {
-  const mdFiles = new MDFiles();
+    this.rhizomeNode = rhizomeNode;
-  mdFiles.readDir();
+    this.app.use(express.json());
-  mdFiles.readReadme();
+  }
-  mdFiles.watchDir();
-  mdFiles.watchReadme();
 
-  // Serve README
+  start() {
-  app.get('/html/README', (_req: express.Request, res: express.Response) => {
+    // Scan and watch for markdown files
-    const html = mdFiles.getReadmeHTML();
+    this.mdFiles.readDir();
-    res.setHeader('content-type', 'text/html').send(html);
+    this.mdFiles.readReadme();
-  });
+    this.mdFiles.watchDir();
+    this.mdFiles.watchReadme();
 
-  // Serve markdown files as html
+    // Serve README
-  app.get('/html/:name', (req: express.Request, res: express.Response) => {
+    this.app.get('/html/README', (_req: express.Request, res: express.Response) => {
-    let html = mdFiles.getHtml(req.params.name);
+      const html = this.mdFiles.getReadmeHTML();
-    if (!html) {
-      res.status(404);
-      html = htmlDocFromMarkdown('# 404\n\n## [Index](/html)');
-    }
-    res.setHeader('content-type', 'text/html');
-    res.send(html);
-  });
-
-  // Serve index
-  {
-    let md = `# Files\n\n`;
-    md += `[README](/html/README)\n\n`;
-    for (const name of mdFiles.list()) {
-      md += `- [${name}](./${name})\n`;
-    }
-    const html = htmlDocFromMarkdown(md);
-
-    app.get('/html', (_req: express.Request, res: express.Response) => {
       res.setHeader('content-type', 'text/html').send(html);
     });
+
+    // Serve markdown files as html
+    this.app.get('/html/:name', (req: express.Request, res: express.Response) => {
+      let html = this.mdFiles.getHtml(req.params.name);
+      if (!html) {
+        res.status(404);
+        html = htmlDocFromMarkdown('# 404\n\n## [Index](/html)');
+      }
+      res.setHeader('content-type', 'text/html');
+      res.send(html);
+    });
+
+    // Serve index
+    {
+      let md = `# Files\n\n`;
+      md += `[README](/html/README)\n\n`;
+      for (const name of this.mdFiles.list()) {
+        md += `- [${name}](./${name})\n`;
+      }
+      const html = htmlDocFromMarkdown(md);
+
+      this.app.get('/html', (_req: express.Request, res: express.Response) => {
+        res.setHeader('content-type', 'text/html').send(html);
+      });
+    }
+
+    // Serve list of all deltas accepted
+    // TODO: This won't scale well
+    this.app.get("/deltas", (_req: express.Request, res: express.Response) => {
+      res.json(this.rhizomeNode.deltaStream.deltasAccepted);
+    });
+
+    // Get the number of deltas ingested by this node
+    this.app.get("/deltas/count", (_req: express.Request, res: express.Response) => {
+      res.json(this.rhizomeNode.deltaStream.deltasAccepted.length);
+    });
+
+    // Get the list of peers seen by this node (including itself)
+    this.app.get("/peers", (_req: express.Request, res: express.Response) => {
+      res.json(this.rhizomeNode.peers.peers.map(({reqAddr, publishAddr, isSelf, isSeedPeer}) => {
+        const deltasAcceptedCount = this.rhizomeNode.deltaStream.deltasAccepted
+          .filter((delta: Delta) => {
+            return delta.receivedFrom?.addr == reqAddr.addr &&
+              delta.receivedFrom?.port == reqAddr.port;
+          })
+          .length;
+        const peerInfo = {
+          reqAddr: reqAddr.toAddrString(),
+          publishAddr: publishAddr?.toAddrString(),
+          isSelf,
+          isSeedPeer,
+          deltaCount: {
+            accepted: deltasAcceptedCount
+          }
+        };
+        return peerInfo;
+      }));
+    });
+
+    // Get the number of peers seen by this node (including itself)
+    this.app.get("/peers/count", (_req: express.Request, res: express.Response) => {
+      res.json(this.rhizomeNode.peers.peers.length);
+    });
+
+    const {httpAddr, httpPort} = this.rhizomeNode.config;
+    this.server = this.app.listen(httpPort, httpAddr, () => {
+      debug(`HTTP API bound to ${httpAddr}:${httpPort}`);
+    });
   }
 
-  // Set up API routes
+  serveCollection(collection: Collection) {
+    const {name} = collection;
 
-  if (collections) {
+    // Get the ID of all domain entities
-    for (const [name, collection] of Object.entries(collections)) {
+    this.app.get(`/${name}/ids`, (_req: express.Request, res: express.Response) => {
-      debug(`collection: ${name}`);
+      res.json({ids: collection.getIds()});
+    });
 
-      // Get the ID of all domain entities
+    // Get a single domain entity by ID
-      app.get(`/${name}/ids`, (_req: express.Request, res: express.Response) => {
+    this.app.get(`/${name}/:id`, (req: express.Request, res: express.Response) => {
-        res.json({ids: collection.getIds()});
+      const {params: {id}} = req;
-      });
+      const ent = collection.get(id);
+      res.json(ent);
+    });
 
-      // Get a single domain entity by ID
+    // Add a new domain entity
-      app.get(`/${name}/:id`, (req: express.Request, res: express.Response) => {
+    // TODO: schema validation
-        const {params: {id}} = req;
+    this.app.put(`/${name}`, (req: express.Request, res: express.Response) => {
-        const ent = collection.get(id);
+      const {body: properties} = req;
-        res.json(ent);
+      const ent = collection.put(properties.id, properties);
-      });
+      res.json(ent);
+    });
 
-      // Add a new domain entity
+    // Update a domain entity
-      // TODO: schema validation
+    this.app.put(`/${name}/:id`, (req: express.Request, res: express.Response) => {
-      app.put(`/${name}`, (req: express.Request, res: express.Response) => {
+      const {body: properties, params: {id}} = req;
-        const {body: properties} = req;
+      if (properties.id && properties.id !== id) {
-        const ent = collection.put(properties.id, properties);
+        res.status(400).json({error: "ID Mismatch", param: id, property: properties.id});
-        res.json(ent);
+        return;
-      });
+      }
-
+      const ent = collection.put(id, properties);
-      // Update a domain entity
+      res.json(ent);
-      app.put(`/${name}/:id`, (req: express.Request, res: express.Response) => {
+    });
-        const {body: properties, params: {id}} = req;
-        if (properties.id && properties.id !== id) {
-          res.status(400).json({error: "ID Mismatch", param: id, property: properties.id});
-          return;
-        }
-        const ent = collection.put(id, properties);
-        res.json(ent);
-      });
-    }
   }
 
-  app.get("/deltas", (_req: express.Request, res: express.Response) => {
+  async stop() {
-    // TODO: streaming
+    this.server?.close();
-    res.json(deltasAccepted);
+    this.mdFiles.close();
-  });
+  }
-
-  // Get the number of deltas ingested by this node
-  app.get("/deltas/count", (_req: express.Request, res: express.Response) => {
-    res.json(deltasAccepted.length);
-  });
-
-  // Get the list of peers seen by this node (including itself)
-  app.get("/peers", (_req: express.Request, res: express.Response) => {
-    res.json(peers.map(({reqAddr, publishAddr, isSelf, isSeedPeer}) => {
-      const deltasAcceptedCount = deltasAccepted
-        .filter((delta: Delta) => {
-          return delta.receivedFrom?.addr == reqAddr.addr &&
-            delta.receivedFrom?.port == reqAddr.port;
-        })
-        .length;
-      const peerInfo = {
-        reqAddr: reqAddr.toAddrString(),
-        publishAddr: publishAddr?.toAddrString(),
-        isSelf,
-        isSeedPeer,
-        deltaCount: {
-          accepted: deltasAcceptedCount
-        }
-      };
-      return peerInfo;
-    }));
-  });
-
-  // Get the number of peers seen by this node (including itself)
-  app.get("/peers/count", (_req: express.Request, res: express.Response) => {
-    res.json(peers.length);
-  });
-
-  app.listen(HTTP_API_PORT, HTTP_API_ADDR, () => {
-    debug(`HTTP API bound to http://${HTTP_API_ADDR}:${HTTP_API_PORT}`);
-  });
 }

src/node.ts

@@ -0,0 +1,82 @@
+import Debug from 'debug';
+import {CREATOR, HTTP_API_ADDR, HTTP_API_ENABLE, HTTP_API_PORT, PEER_ID, PUBLISH_BIND_ADDR, PUBLISH_BIND_HOST, PUBLISH_BIND_PORT, REQUEST_BIND_ADDR, REQUEST_BIND_HOST, REQUEST_BIND_PORT, SEED_PEERS} from './config';
+import {DeltaStream} from './deltas';
+import {HttpApi} from './http-api';
+import {Peers} from './peers';
+import {PubSub} from './pub-sub';
+import {RequestReply} from './request-reply';
+import {PeerAddress} from './types';
+import {Collection} from './collection';
+const debug = Debug('rhizome-node');
+
+export type RhizomeNodeConfig = {
+  requestBindAddr: string;
+  requestBindHost: string;
+  requestBindPort: number;
+  publishBindAddr: string;
+  publishBindHost: string;
+  publishBindPort: number;
+  httpAddr: string;
+  httpPort: number;
+  httpEnable: boolean;
+  seedPeers: PeerAddress[];
+  peerId: string;
+  creator: string; // TODO each host should be able to support multiple users
+};
+
+// So that we can run more than one instance in the same process (for testing)
+export class RhizomeNode {
+  config: RhizomeNodeConfig;
+  pubSub: PubSub;
+  requestReply: RequestReply;
+  httpApi: HttpApi;
+  deltaStream: DeltaStream;
+  peers: Peers;
+  myRequestAddr: PeerAddress;
+  myPublishAddr: PeerAddress;
+
+  constructor(config?: Partial<RhizomeNodeConfig>) {
+    this.config = {
+      requestBindAddr: REQUEST_BIND_ADDR,
+      requestBindHost: REQUEST_BIND_HOST,
+      requestBindPort: REQUEST_BIND_PORT,
+      publishBindAddr: PUBLISH_BIND_ADDR,
+      publishBindHost: PUBLISH_BIND_HOST,
+      publishBindPort: PUBLISH_BIND_PORT,
+      httpAddr: HTTP_API_ADDR,
+      httpPort: HTTP_API_PORT,
+      httpEnable: HTTP_API_ENABLE,
+      seedPeers: SEED_PEERS,
+      peerId: PEER_ID,
+      creator: CREATOR,
+      ...config
+    };
+    debug('config', this.config);
+    this.myRequestAddr = new PeerAddress(this.config.requestBindHost, this.config.requestBindPort);
+    this.myPublishAddr = new PeerAddress(this.config.publishBindHost, this.config.publishBindPort);
+    this.pubSub = new PubSub(this);
+    this.requestReply = new RequestReply(this);
+    this.httpApi = new HttpApi(this);
+    this.deltaStream = new DeltaStream(this);
+    this.peers = new Peers(this);
+  }
+
+  async start() {
+    this.pubSub.start();
+    this.requestReply.start();
+    if (this.config.httpEnable) {
+      this.httpApi.start();
+    }
+    await new Promise((resolve) => setTimeout(resolve, 500));
+    this.peers.subscribeToSeeds();
+    await new Promise((resolve) => setTimeout(resolve, 500));
+    this.peers.askAllPeersForDeltas();
+    await new Promise((resolve) => setTimeout(resolve, 1000));
+  }
+
+  async stop() {
+    await this.pubSub.stop();
+    await this.requestReply.stop();
+    await this.httpApi.stop();
+  }
+}

src/object-layer.ts

@@ -7,8 +7,8 @@
 // - As typescript interfaces?
 // - As typescript classes?
 
-import { CREATOR, HOST_ID } from "./config";
+import {RhizomeNode} from "./node";
-import { Delta, PropertyTypes } from "./types";
+import {Delta, PropertyTypes} from "./types";
 
 export type EntityProperties = {
   [key: string]: PropertyTypes;
@@ -29,10 +29,10 @@ export class Entity {
 export class EntityPropertiesDeltaBuilder {
   delta: Delta;
 
-  constructor(entityId: string) {
+  constructor(rhizomeNode: RhizomeNode, entityId: string) {
     this.delta = {
-      creator: CREATOR,
+      creator: rhizomeNode.config.creator,
-      host: HOST_ID,
+      host: rhizomeNode.config.peerId,
       pointers: [{
         localContext: 'id',
         target: entityId,

src/peers.ts

@@ -1,9 +1,10 @@
-import {PUBLISH_BIND_HOST, PUBLISH_BIND_PORT, REQUEST_BIND_HOST, REQUEST_BIND_PORT, SEED_PEERS} from "./config";
-import {deltasAccepted, ingestAll, receiveDelta} from "./deltas";
-import {connectSubscribe} from "./pub-sub";
-import {PeerRequest, registerRequestHandler, RequestSocket, ResponseSocket} from "./request-reply";
-import {Delta, PeerAddress} from "./types";
 import Debug from 'debug';
+import {Message} from 'zeromq';
+import {SEED_PEERS} from "./config";
+import {RhizomeNode} from "./node";
+import {Subscription} from './pub-sub';
+import {PeerRequest, RequestSocket, ResponseSocket} from "./request-reply";
+import {Delta, PeerAddress} from "./types";
 const debug = Debug('peers');
 
 export enum PeerMethods {
@@ -11,48 +12,50 @@ export enum PeerMethods {
   AskForDeltas
 }
 
-export const myRequestAddr = new PeerAddress(REQUEST_BIND_HOST, REQUEST_BIND_PORT);
-export const myPublishAddr = new PeerAddress(PUBLISH_BIND_HOST, PUBLISH_BIND_PORT);
-
-registerRequestHandler(async (req: PeerRequest, res: ResponseSocket) => {
-  debug('inspecting peer request');
-  switch (req.method) {
-    case PeerMethods.GetPublishAddress: {
-      debug('it\'s a request for our publish address');
-      await res.send(myPublishAddr.toAddrString());
-      break;
-    }
-    case PeerMethods.AskForDeltas: {
-      debug('it\'s a request for deltas');
-      // TODO: stream these rather than
-      // trying to write them all in one message
-      await res.send(JSON.stringify(deltasAccepted));
-      break;
-    }
-  }
-});
-
 class Peer {
+  rhizomeNode: RhizomeNode;
   reqAddr: PeerAddress;
-  reqSock: RequestSocket;
+  reqSock?: RequestSocket;
   publishAddr: PeerAddress | undefined;
   isSelf: boolean;
   isSeedPeer: boolean;
-  constructor(addr: string, port: number) {
+  subscription?: Subscription;
-    this.reqAddr = new PeerAddress(addr, port);
+
-    this.reqSock = new RequestSocket(addr, port);
+  constructor(rhizomeNode: RhizomeNode, reqAddr: PeerAddress) {
-    this.isSelf = addr === myRequestAddr.addr && port === myRequestAddr.port;
+    this.rhizomeNode = rhizomeNode;
-    this.isSeedPeer = !!SEED_PEERS.find((seedPeer) =>
+    this.reqAddr = reqAddr;
-      addr === seedPeer.addr && port === seedPeer.port);
+    this.isSelf = reqAddr.isEqual(this.rhizomeNode.myRequestAddr);
+    this.isSeedPeer = !!SEED_PEERS.find((seedPeer) => reqAddr.isEqual(seedPeer));
   }
-  async subscribe() {
+
-    if (!this.publishAddr) {
+  async request(method: PeerMethods): Promise<Message> {
-      const res = await this.reqSock.request(PeerMethods.GetPublishAddress);
+    if (!this.reqSock) {
-      // TODO: input validation
+      this.reqSock = new RequestSocket(this.reqAddr);
-      this.publishAddr = PeerAddress.fromString(res.toString());
-      connectSubscribe(this.publishAddr!);
     }
+    return this.reqSock.request(method);
   }
+
+  async subscribeDeltas() {
+    if (!this.publishAddr) {
+      debug(`requesting publish addr from peer ${this.reqAddr.toAddrString()}`);
+      const res = await this.request(PeerMethods.GetPublishAddress);
+      this.publishAddr = PeerAddress.fromString(res.toString());
+      debug(`received publish addr ${this.publishAddr.toAddrString()} from peer ${this.reqAddr.toAddrString()}`);
+    }
+
+    this.subscription = this.rhizomeNode.pubSub.subscribe(
+      this.publishAddr,
+      "deltas",
+      (sender, msg) => {
+        const delta = this.rhizomeNode.deltaStream.deserializeDelta(msg.toString());
+        delta.receivedFrom = sender;
+        debug(`Received delta: ${JSON.stringify(delta)}`);
+        this.rhizomeNode.deltaStream.ingestDelta(delta);
+      });
+
+    this.subscription.start();
+  }
+
   async askForDeltas(): Promise<Delta[]> {
     // TODO as a first approximation we are trying to cram the entire history
     // of accepted deltas, into one (potentially gargantuan) json message.
@@ -60,42 +63,69 @@ class Peer {
     // Third pass should find a way to reduce the number of deltas transmitted.
 
     // TODO: requestTimeout
-    const res = await this.reqSock.request(PeerMethods.AskForDeltas);
+    const res = await this.request(PeerMethods.AskForDeltas);
     const deltas = JSON.parse(res.toString());
     return deltas;
   }
 }
 
-export const peers: Peer[] = [];
+export class Peers {
+  rhizomeNode: RhizomeNode;
+  peers: Peer[] = [];
 
-peers.push(new Peer(myRequestAddr.addr, myRequestAddr.port));
+  constructor(rhizomeNode: RhizomeNode) {
+    this.rhizomeNode = rhizomeNode;
 
-function newPeer(addr: string, port: number) {
+    // Add self to the list of peers, but don't connect
-  const peer = new Peer(addr, port);
+    this.addPeer(this.rhizomeNode.myRequestAddr);
-  peers.push(peer);
-  return peer;
-}
 
-export async function subscribeToSeeds() {
+    this.rhizomeNode.requestReply.registerRequestHandler(async (req: PeerRequest, res: ResponseSocket) => {
-  SEED_PEERS.forEach(async ({addr, port}, idx) => {
+      debug('inspecting peer request');
-    debug(`SEED PEERS[${idx}]=${addr}:${port}`);
+      switch (req.method) {
-    const peer = newPeer(addr, port);
+        case PeerMethods.GetPublishAddress: {
-    await peer.subscribe();
+          debug('it\'s a request for our publish address');
-  });
+          await res.send(this.rhizomeNode.myPublishAddr.toAddrString());
-}
+          break;
-
+        }
-//! TODO Expect abysmal scaling properties with this function
+        case PeerMethods.AskForDeltas: {
-export async function askAllPeersForDeltas() {
+          debug('it\'s a request for deltas');
-  peers
+          // TODO: stream these rather than
-    .filter(({isSelf}) => !isSelf)
+          // trying to write them all in one message
-    .forEach(async (peer, idx) => {
+          await res.send(JSON.stringify(this.rhizomeNode.deltaStream.deltasAccepted));
-      debug(`Asking peer ${idx} for deltas`);
+          break;
-      const deltas = await peer.askForDeltas();
+        }
-      debug(`received ${deltas.length} deltas from ${peer.reqAddr.toAddrString()}`);
-      for (const delta of deltas) {
-        delta.receivedFrom = peer.reqAddr;
-        receiveDelta(delta);
       }
-      ingestAll();
     });
+
+  }
+
+  addPeer(addr: PeerAddress): Peer {
+    const peer = new Peer(this.rhizomeNode, addr);
+    this.peers.push(peer);
+    debug('added peer', addr);
+    return peer;
+  }
+
+  async subscribeToSeeds() {
+    SEED_PEERS.forEach(async (addr, idx) => {
+      debug(`SEED PEERS[${idx}]=${addr.toAddrString()}`);
+      const peer = this.addPeer(addr);
+      await peer.subscribeDeltas();
+    });
+  }
+
+  //! TODO Expect abysmal scaling properties with this function
+  async askAllPeersForDeltas() {
+    this.peers.filter(({isSelf}) => !isSelf)
+      .forEach(async (peer, idx) => {
+        debug(`Asking peer ${idx} for deltas`);
+        const deltas = await peer.askForDeltas();
+        debug(`received ${deltas.length} deltas from ${peer.reqAddr.toAddrString()}`);
+        for (const delta of deltas) {
+          delta.receivedFrom = peer.reqAddr;
+          this.rhizomeNode.deltaStream.receiveDelta(delta);
+        }
+        this.rhizomeNode.deltaStream.ingestAll();
+      });
+  }
 }

src/pub-sub.ts

@@ -1,23 +1,79 @@
-import {Publisher, Subscriber} from 'zeromq';
-import {PUBLISH_BIND_ADDR, PUBLISH_BIND_PORT} from './config';
-import {PeerAddress} from './types';
 import Debug from 'debug';
+import {Message, Publisher, Subscriber} from 'zeromq';
+import {RhizomeNode} from './node';
+import {PeerAddress} from './types';
 const debug = Debug('pub-sub');
 
-export const publishSock = new Publisher();
+export type SubscribedMessageHandler = (sender: PeerAddress, msg: Message) => void;
-export const subscribeSock = new Subscriber();
 
-export async function bindPublish() {
+// TODO: Allow subscribing to multiple topics on one socket
-  const addrStr = `tcp://${PUBLISH_BIND_ADDR}:${PUBLISH_BIND_PORT}`;
+export class Subscription {
-  await publishSock.bind(addrStr);
+  sock = new Subscriber();
-  debug(`Publishing socket bound to ${addrStr}`);
+  topic: string;
+  publishAddr: PeerAddress;
+  publishAddrStr: string;
+  cb: SubscribedMessageHandler;
+
+  constructor(publishAddr: PeerAddress, topic: string, cb: SubscribedMessageHandler) {
+    this.cb = cb;
+    this.topic = topic;
+    this.publishAddr = publishAddr;
+    this.publishAddrStr = `tcp://${this.publishAddr.toAddrString()}`;
+  }
+
+  async start() {
+    this.sock.connect(this.publishAddrStr);
+    this.sock.subscribe(this.topic);
+    debug(`Subscribing to ${this.topic} topic on ${this.publishAddrStr}`);
+
+    for await (const [, sender, msg] of this.sock) {
+      const senderAddr = PeerAddress.fromString(sender.toString());
+      this.cb(senderAddr, msg);
+    }
+  }
 }
 
-export function connectSubscribe(publishAddr: PeerAddress) {
+export class PubSub {
-  // TODO: peer discovery
+  rhizomeNode: RhizomeNode;
-  const addrStr = `tcp://${publishAddr.toAddrString()}`;
+  publishSock: Publisher;
-  debug('connectSubscribe', {addrStr});
+  publishAddrStr: string;
-  subscribeSock.connect(addrStr);
+  subscriptions: Subscription[] = [];
-  subscribeSock.subscribe("deltas");
+
-  debug(`Subscribing to ${addrStr}`);
+  constructor(rhizomeNode: RhizomeNode) {
+    this.rhizomeNode = rhizomeNode;
+    this.publishSock = new Publisher();
+
+    const {publishBindAddr, publishBindPort} = this.rhizomeNode.config;
+    this.publishAddrStr = `tcp://${publishBindAddr}:${publishBindPort}`;
+  }
+
+  async start() {
+    await this.publishSock.bind(this.publishAddrStr);
+    debug(`Publishing socket bound to ${this.publishAddrStr}`);
+  }
+
+  async publish(topic: string, msg: string) {
+    await this.publishSock.send([
+      topic,
+      this.rhizomeNode.myRequestAddr.toAddrString(),
+      msg
+    ]);
+  }
+
+  subscribe(publishAddr: PeerAddress, topic: string, cb: SubscribedMessageHandler): Subscription {
+    const subscription = new Subscription(publishAddr, topic, cb);
+    this.subscriptions.push(subscription);
+    return subscription;
+  }
+
+  async stop() {
+    await this.publishSock.unbind(this.publishAddrStr);
+    this.publishSock.close();
+    this.publishSock = new Publisher();
+
+    for (const subscription of this.subscriptions) {
+      subscription.sock.close();
+      debug('subscription socket is closed?', subscription.sock.closed);
+    }
+  }
 }

src/query.ts

@@ -1,15 +0,0 @@
-import { Query, QueryResult, } from './types';
-import { deltasAccepted } from './deltas';
-import { applyFilter } from './filter';
-
-// export const queryResultMemo = new Map<Query, QueryResult>();
-
-export function issueQuery(query: Query): QueryResult {
-  const deltas = applyFilter(deltasAccepted, query.filterExpr);
-  return {
-    deltas
-    // TODO: Materialized view / state collapse snapshot
-  };
-}
-
-

src/request-reply.ts

@@ -1,8 +1,9 @@
-import { Request, Reply, Message } from 'zeromq';
+import {Request, Reply, Message} from 'zeromq';
-import { REQUEST_BIND_PORT, REQUEST_BIND_ADDR} from './config';
+import {EventEmitter} from 'node:events';
-import { EventEmitter } from 'node:events';
+import {PeerMethods} from './peers';
-import { PeerMethods } from './peers';
 import Debug from 'debug';
+import {RhizomeNode} from './node';
+import {PeerAddress} from './types';
 const debug = Debug('request-reply');
 
 export type PeerRequest = {
@@ -11,32 +12,28 @@ export type PeerRequest = {
 
 export type RequestHandler = (req: PeerRequest, res: ResponseSocket) => void;
 
-export const replySock = new Reply();
+// TODO: Retain handle to request socket for each peer, so we only need to open once
-const requestStream = new EventEmitter();
+export class RequestSocket {
+  sock = new Request();
 
-export async function bindReply() {
+  constructor(addr: PeerAddress) {
-  const addrStr = `tcp://${REQUEST_BIND_ADDR}:${REQUEST_BIND_PORT}`;
+    const addrStr = `tcp://${addr.addr}:${addr.port}`;
-  await replySock.bind(addrStr);
+    this.sock.connect(addrStr);
-  debug(`Reply socket bound to ${addrStr}`);
+    debug(`Request socket connecting to ${addrStr}`);
-}
-
-export async function runRequestHandlers() {
-  for await (const [msg] of replySock) {
-    debug(`Received message`, {msg: msg.toString()});
-    const req = peerRequestFromMsg(msg);
-    requestStream.emit('request', req);
   }
-}
 
-function peerRequestFromMsg(msg: Message): PeerRequest | null {
+  async request(method: PeerMethods): Promise<Message> {
-  let req: PeerRequest | null = null;
+    const req: PeerRequest = {
-  try {
+      method
-    const obj = JSON.parse(msg.toString());
+    };
-    req = {...obj};
+    await this.sock.send(JSON.stringify(req));
-  } catch(e) {
+    // Wait for a response.
-    debug('error receiving command', e);
+    // TODO: Timeout
+    // TODO: Retry
+    // this.sock.receiveTimeout = ...
+    const [res] = await this.sock.receive();
+    return res;
   }
-  return req;
 }
 
 export class ResponseSocket {
@@ -53,30 +50,54 @@ export class ResponseSocket {
   }
 }
 
-export function registerRequestHandler(handler: RequestHandler) {
+function peerRequestFromMsg(msg: Message): PeerRequest | null {
-  requestStream.on('request', (req) => {
+  let req: PeerRequest | null = null;
-    const res = new ResponseSocket(replySock);
+  try {
-    handler(req, res);
+    const obj = JSON.parse(msg.toString());
-  });
+    req = {...obj};
+  } catch (e) {
+    debug('error receiving command', e);
+  }
+  return req;
 }
 
-export class RequestSocket {
+export class RequestReply {
-  sock = new Request();
+  rhizomeNode: RhizomeNode;
-  constructor(host: string, port: number) {
+  replySock = new Reply();
-    const addrStr = `tcp://${host}:${port}`;
+  requestStream = new EventEmitter();
-    this.sock.connect(addrStr);
+  requestBindAddrStr: string;
-    debug(`Request socket connecting to ${addrStr}`);
+
+  constructor(rhizomeNode: RhizomeNode) {
+    this.rhizomeNode = rhizomeNode;
+    const {requestBindAddr, requestBindPort} = this.rhizomeNode.config;
+    this.requestBindAddrStr = `tcp://${requestBindAddr}:${requestBindPort}`;
   }
-  async request(method: PeerMethods): Promise<Message> {
+
-    const req: PeerRequest = {
+  // Listen for incoming requests
-      method
+  async start() {
-    };
+
-    await this.sock.send(JSON.stringify(req));
+    await this.replySock.bind(this.requestBindAddrStr);
-    // Wait for a response.
+    debug(`Reply socket bound to ${this.requestBindAddrStr}`);
-    // TODO: Timeout
+
-    // TODO: Retry
+    for await (const [msg] of this.replySock) {
-    // this.sock.receiveTimeout = ...
+      debug(`Received message`, {msg: msg.toString()});
-    const [res] = await this.sock.receive();
+      const req = peerRequestFromMsg(msg);
-    return res;
+      this.requestStream.emit('request', req);
+    }
+  }
+
+  // Add a top level handler for incoming requests.
+  // Each handler will get a copy of every message.
+  registerRequestHandler(handler: RequestHandler) {
+    this.requestStream.on('request', (req) => {
+      const res = new ResponseSocket(this.replySock);
+      handler(req, res);
+    });
+  }
+
+  async stop() {
+    await this.replySock.unbind(this.requestBindAddrStr);
+    this.replySock.close();
+    this.replySock = new Reply();
   }
 }

src/typed-collection.ts

@@ -5,6 +5,7 @@ export class TypedCollection<T extends EntityProperties> extends Collection {
   put(id: string | undefined, properties: T): Entity {
     return super.put(id, properties);
   }
+
   get(id: string): Entity | undefined {
     return super.get(id);
   }

src/types.ts

@@ -46,19 +46,27 @@ export type Properties = {[key: string]: PropertyTypes};
 export class PeerAddress {
   addr: string;
   port: number;
+
   constructor(addr: string, port: number) {
     this.addr = addr;
     this.port = port;
   }
+
   static fromString(addrString: string): PeerAddress {
     const [addr, port] = addrString.trim().split(':');
     return new PeerAddress(addr, parseInt(port));
   }
+
   toAddrString() {
     return `${this.addr}:${this.port}`;
   }
+
   toJSON() {
     return this.toAddrString();
   }
+
+  isEqual(other: PeerAddress) {
+    return this.addr === other.addr && this.port === other.port;
+  }
 };