keeping track of peers from whom deltas are received

__tests__/peer-address.ts

@@ -0,0 +1,13 @@
+import {PeerAddress} from '../src/types';
+
+describe('PeerAddress', () => {
+  it('toString()', () => {
+    const addr = new PeerAddress('localhost', 1000);
+    expect(addr.toString()).toBe("localhost:1000");
+  });
+  it('fromString()', () => {
+    const addr = PeerAddress.fromString("localhost:1000");
+    expect(addr.addr).toBe("localhost");
+    expect(addr.port).toBe(1000);
+  });
+});

eslint.config.js

@@ -4,4 +4,9 @@ import tseslint from 'typescript-eslint';
 export default tseslint.config(
   eslint.configs.recommended,
   tseslint.configs.recommended,
+  {
+    ignores: [
+      "dist/",
+    ],
+  }
 );

package.json

@@ -5,7 +5,11 @@
   "scripts": {
     "start": "node --experimental-strip-types --experimental-transform-types src/main.ts",
     "lint": "eslint",
-    "test": "echo \"Error: no test specified\" && exit 1"
+    "test": "jest"
+  },
+  "jest": {
+    "testEnvironment": "node",
+    "preset": "ts-jest"
   },
   "author": "",
   "license": "Unlicense",
@@ -20,9 +24,12 @@
   "devDependencies": {
     "@eslint/js": "^9.17.0",
     "@types/express": "^5.0.0",
+    "@types/jest": "^29.5.14",
     "@types/node": "^22.10.2",
     "eslint": "^9.17.0",
     "eslint-config-airbnb-base-typescript": "^1.1.0",
+    "jest": "^29.7.0",
+    "ts-jest": "^29.2.5",
     "typescript": "^5.7.2",
     "typescript-eslint": "^8.18.0"
   }

src/collection-layer.ts

@@ -2,9 +2,8 @@
 // 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 EventEmitter from "node:events";
-import { publishDelta, subscribeDeltas } from "./deltas";
+import { deltasAccepted, publishDelta, subscribeDeltas } from "./deltas";
 import { Entity, EntityProperties, EntityPropertiesDeltaBuilder } from "./object-layer";
 import { Delta } from "./types";
 import { randomUUID } from "node:crypto";
@@ -45,10 +44,8 @@ export class Collection {
   entities = new Map<string, Entity>();
   eventStream = new EventEmitter();
   constructor() {
-    console.log('COLLECTION SUBSCRIBING TO DELTA STREAM');
     subscribeDeltas((delta: Delta) => {
       // TODO: Make sure this is the kind of delta we're looking for
-      console.log('COLLECTION RECEIVED DELTA');
       this.applyDelta(delta);
     });
     this.eventStream.on('create', (entity: Entity) => {
@@ -69,6 +66,7 @@ export class Collection {
       eventType = 'create';
     }
     const deltaBulider = new EntityPropertiesDeltaBuilder(entityId);
+    console.log('deltaBulider -->', deltaBulider.delta);
 
     if (!properties) {
       // Let's interpret this as entity deletion
@@ -157,6 +155,7 @@ export class Collection {
     const deltas: Delta[] = [];
     const entity = this.updateEntity(entityId, properties, true, deltas);
     deltas.forEach(async (delta: Delta) => {
+      deltasAccepted.push(delta);
       await publishDelta(delta);
     });
     return entity;
@@ -165,6 +164,7 @@ export class Collection {
     const deltas: Delta[] = [];
     this.updateEntity(entityId, undefined, true, deltas);
     deltas.forEach(async (delta: Delta) => {
+      deltasAccepted.push(delta);
       await publishDelta(delta);
     });
   }

src/config.ts

@@ -1,17 +1,19 @@
+import {randomUUID} from "crypto";
+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 = process.env.HOST!;
+export const HOST_ID = process.env.RHIZOME_PEER_ID || randomUUID();
-export const ADDRESS = process.env.ADDRESS ?? '127.0.0.1';
+export const ADDRESS = process.env.RHIZOME_ADDRESS ?? '127.0.0.1';
-export const REQUEST_BIND_PORT = parseInt(process.env.REQUEST_BIND_PORT || '4000');
+export const REQUEST_BIND_PORT = parseInt(process.env.RHIZOME_REQUEST_BIND_PORT || '4000');
-export const PUBLISH_BIND_PORT = parseInt(process.env.PUBLISH_BIND_PORT || '4001');
+export const REQUEST_BIND_ADDR = process.env.RHIZOME_REQUEST_BIND_ADDR || ADDRESS || '127.0.0.1';
-export const REQUEST_BIND_ADDR = process.env.ADDRESS || '127.0.0.1';
+export const REQUEST_BIND_HOST = process.env.RHIZOME_REQUEST_BIND_HOST || REQUEST_BIND_ADDR || '127.0.0.1';
-export const PUBLISH_BIND_ADDR = process.env.ADDRESS || '127.0.0.1';
+export const PUBLISH_BIND_PORT = parseInt(process.env.RHIZOME_PUBLISH_BIND_PORT || '4001');
-export const HTTP_API_PORT = parseInt(process.env.HTTP_API_PORT || '3000');
+export const PUBLISH_BIND_ADDR = process.env.RHIZOME_PUBLISH_BIND_ADDR || ADDRESS || '127.0.0.1';
-export const HTTP_API_ADDR = process.env.ADDRESS || '127.0.0.1';
+export const PUBLISH_BIND_HOST = process.env.RHIZOME_PUBLISH_BIND_HOST || PUBLISH_BIND_ADDR || '127.0.0.1';
-export const ENABLE_HTTP_API = process.env.ENABLE_HTTP_API === 'true';
+export const HTTP_API_PORT = parseInt(process.env.RHIZOME_HTTP_API_PORT || '3000');
-export const SEED_PEERS = (process.env.SEED_PEERS || '').split(',')
+export const HTTP_API_ADDR = process.env.RHIZOME_HTTP_API_ADDR || ADDRESS || '127.0.0.1';
+export const HTTP_API_ENABLE = process.env.RHIZOME_HTTP_API_ENABLE === 'true';
+export const SEED_PEERS: PeerAddress[]  = (process.env.RHIZOME_SEED_PEERS || '').split(',')
   .filter(x => !!x)
-  .map((peer: string) => {
+  .map((peer: string) => PeerAddress.fromString(peer));
-    const [addr, port] = peer.trim().split(':');
-    return {addr, port: parseInt(port)};
-  });

src/deltas.ts

@@ -1,6 +1,8 @@
 import EventEmitter from 'node:events';
-import { Delta, Decision } from './types';
+import {REQUEST_BIND_HOST, REQUEST_BIND_PORT} from './config';
-import { publishSock, subscribeSock } from './pub-sub';
+import {publishSock, subscribeSock} from './pub-sub';
+import {Decision, Delta, PeerAddress} from './types';
+import {myRequestAddr} from './peers';
 
 export const deltaStream = new EventEmitter();
 
@@ -22,7 +24,7 @@ export function ingestDelta(delta: Delta) {
   switch (decision) {
     case Decision.Accept:
       deltasAccepted.push(delta);
-      deltaStream.emit('delta', { delta });
+      deltaStream.emit('delta', {delta});
       break;
     case Decision.Reject:
       deltasRejected.push(delta);
@@ -67,7 +69,7 @@ export function subscribeDeltas(fn: (delta: Delta) => void) {
 
 export async function publishDelta(delta: Delta) {
   console.log(`Publishing delta: ${JSON.stringify(delta)}`);
-  await publishSock.send(["deltas", serializeDelta(delta)])
+  await publishSock.send(["deltas", myRequestAddr.toAddrString(), serializeDelta(delta)]);
 }
 
 function serializeDelta(delta: Delta) {
@@ -79,11 +81,12 @@ function deserializeDelta(input: string) {
 }
 
 export async function runDeltas() {
-  for await (const [topic, msg] of subscribeSock) {
+  for await (const [topic, sender, msg] of subscribeSock) {
     if (topic.toString() !== "deltas") {
       continue;
     }
     const delta = deserializeDelta(msg.toString());
+    delta.receivedFrom = PeerAddress.fromString(sender.toString());
     console.log(`Received delta: ${JSON.stringify(delta)}`);
     ingestDelta(delta);
   }

src/example-app.ts

@@ -1,14 +1,14 @@
 // We can start to use deltas to express relational data in a given context
 
 import express from "express";
-import { bindPublish, } from "./pub-sub";
+import {Collection} from "./collection-layer";
-import { deltasAccepted, deltasProposed, runDeltas } from "./deltas";
+import {HTTP_API_ENABLE, HTTP_API_ADDR, HTTP_API_PORT, SEED_PEERS} from "./config";
-import { Entity } from "./object-layer";
+import {deltasAccepted, deltasProposed, runDeltas} from "./deltas";
-import { Collection } from "./collection-layer";
+import {Entity} from "./object-layer";
-import { bindReply, runRequestHandlers } from "./request-reply";
+import {askAllPeersForDeltas, peers, subscribeToSeeds} from "./peers";
-import { askAllPeersForDeltas, subscribeToSeeds } from "./peers";
+import {bindPublish, } from "./pub-sub";
-import { ENABLE_HTTP_API, HTTP_API_ADDR, HTTP_API_PORT } from "./config";
+import {bindReply, runRequestHandlers} from "./request-reply";
-
+import {Delta, PeerAddress} from "./types";
 
 // As an app we want to be able to write and read data.
 // The data is whatever shape we define it to be in a given context.
@@ -16,7 +16,6 @@ import { ENABLE_HTTP_API, HTTP_API_ADDR, HTTP_API_PORT } from "./config";
 // e.g. entities and their properties.
 
 // This implies at least one layer on top of the underlying primitive deltas.
-
 type UserProperties = {
   id?: string;
   name: string;
@@ -48,11 +47,13 @@ class Users {
 }
 
 (async () => {
+  console.log('1');
   const users = new Users();
+  console.log('2');
 
   const app = express()
   app.get("/ids", (req: express.Request, res: express.Response) => {
-      res.json({ ids: users.getIds()});
+    res.json({ids: users.getIds()});
   });
 
   app.get("/deltas", (req: express.Request, res: express.Response) => {
@@ -65,14 +66,39 @@ class Users {
     res.json(deltasAccepted.length);
   });
 
-  if (ENABLE_HTTP_API) {
+  app.get("/peers", (req: express.Request, res: express.Response) => {
+    res.json(peers.map(({reqAddr, publishAddr}) => {
+      const isSeedPeer = !!SEED_PEERS.find(({addr, port}) =>
+        addr === reqAddr.addr && port === reqAddr.port);
+      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(),
+        isSeedPeer,
+        deltaCount: {
+          accepted: deltasAcceptedCount
+        }
+      };
+      return peerInfo;
+    }));
+  });
+
+  if (HTTP_API_ENABLE) {
     app.listen(HTTP_API_PORT, HTTP_API_ADDR, () => {
-        console.log(`HTTP API bound to http://${HTTP_API_ADDR}:${HTTP_API_PORT}`);
+      console.log(`HTTP API bound to http://${HTTP_API_ADDR}:${HTTP_API_PORT}`);
     });
   }
 
+  console.log('3');
   await bindPublish();
+  console.log('3a');
   await bindReply();
+  console.log('3b');
   runDeltas();
   runRequestHandlers();
   await new Promise((resolve) => setTimeout(resolve, 500));
@@ -81,11 +107,7 @@ class Users {
   askAllPeersForDeltas();
   await new Promise((resolve) => setTimeout(resolve, 1000));
 
-  setInterval(() => {
+  console.log('4');
-    console.log('deltasProposed count', deltasProposed.length,
-      'deltasAccepted count', deltasAccepted.length);
-  }, 5000)
-
   const taliesin = users.upsert({
     // id: 'taliesin-1',
     name: 'Taliesin',

src/main.ts

@@ -1,6 +1,6 @@
 import express from "express";
 import { runDeltas } from "./deltas";
-import {ENABLE_HTTP_API, HTTP_API_ADDR, HTTP_API_PORT} from "./config";
+import {HTTP_API_ENABLE, HTTP_API_ADDR, HTTP_API_PORT} from "./config";
 
 const app = express()
 
@@ -8,7 +8,7 @@ app.get("/", (req: express.Request, res: express.Response) => {
     res.json({ message: "Welcome to the Express + TypeScript Server!" });
 });
 
-if (ENABLE_HTTP_API) {
+if (HTTP_API_ENABLE) {
   app.listen(HTTP_API_PORT, HTTP_API_ADDR, () => {
       console.log(`HTTP API bound to http://${HTTP_API_ADDR}:${HTTP_API_PORT}`);
   });

src/object-layer.ts

@@ -7,7 +7,7 @@
 // - As typescript interfaces?
 // - As typescript classes?
 
-import { CREATOR, HOST } from "./config";
+import { CREATOR, HOST_ID } from "./config";
 import { Delta, PropertyTypes } from "./types";
 
 export type EntityProperties = {
@@ -31,7 +31,7 @@ export class EntityPropertiesDeltaBuilder {
   constructor(entityId: string) {
     this.delta = {
       creator: CREATOR,
-      host: HOST,
+      host: HOST_ID,
       pointers: [{
         localContext: 'id',
         target: entityId,

src/peers.ts

@@ -1,22 +1,23 @@
-import { PUBLISH_BIND_ADDR, PUBLISH_BIND_PORT } from "./config";
+import {PUBLISH_BIND_HOST, PUBLISH_BIND_PORT, REQUEST_BIND_HOST, REQUEST_BIND_PORT, SEED_PEERS} from "./config";
-import { registerRequestHandler, PeerRequest, ResponseSocket } from "./request-reply";
-import { RequestSocket, } from "./request-reply";
-import { SEED_PEERS } from "./config";
-import {connectSubscribe} from "./pub-sub";
 import {deltasAccepted, deltasProposed, ingestAll, receiveDelta} from "./deltas";
-import {Delta} from "./types";
+import {connectSubscribe} from "./pub-sub";
+import {PeerRequest, registerRequestHandler, RequestSocket, ResponseSocket} from "./request-reply";
+import {Delta, PeerAddress} from "./types";
 
 export enum PeerMethods {
   GetPublishAddress,
   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) => {
   console.log('inspecting peer request');
   switch (req.method) {
     case PeerMethods.GetPublishAddress: {
       console.log('it\'s a request for our publish address');
-      await res.send(publishAddr);
+      await res.send(myPublishAddr.toAddrString());
       break;
     }
     case PeerMethods.AskForDeltas: {
@@ -29,29 +30,20 @@ registerRequestHandler(async (req: PeerRequest, res: ResponseSocket) => {
   }
 });
 
-export type PeerAddress = {
-  addr: string,
-  port: number
-};
-
-const publishAddr: PeerAddress = {
-  addr: PUBLISH_BIND_ADDR,
-  port: PUBLISH_BIND_PORT
-};
-
 class Peer {
+  reqAddr: PeerAddress;
   reqSock: RequestSocket;
   publishAddr: PeerAddress | undefined;
   constructor(addr: string, port: number) {
+    this.reqAddr = new PeerAddress(addr, port);
     this.reqSock = new RequestSocket(addr, port);
   }
   async subscribe() {
     if (!this.publishAddr) {
       const res = await this.reqSock.request(PeerMethods.GetPublishAddress);
       // TODO: input validation
-      const {addr, port} = JSON.parse(res.toString());
+      this.publishAddr = PeerAddress.fromString(res.toString());
-      this.publishAddr = {addr, port};
+      connectSubscribe(this.publishAddr!);
-      connectSubscribe(addr, port);
     }
   }
   async askForDeltas(): Promise<Delta[]> {
@@ -67,12 +59,12 @@ class Peer {
   }
 }
 
-const peers: Peer[] = [];
+export const peers: Peer[] = [];
 
 function newPeer(addr: string, port: number) {
-    const peer = new Peer(addr, port);
+  const peer = new Peer(addr, port);
-    peers.push(peer);
+  peers.push(peer);
-    return peer;
+  return peer;
 }
 
 export async function subscribeToSeeds() {
@@ -90,6 +82,7 @@ export async function askAllPeersForDeltas() {
     const deltas = await peer.askForDeltas();
     console.log('received deltas:', deltas);
     for (const delta of deltas) {
+      delta.receivedFrom = peer.reqAddr;
       receiveDelta(delta);
     }
     console.log('deltasProposed count', deltasProposed.length);

src/pub-sub.ts

@@ -1,18 +1,21 @@
-import { Publisher, Subscriber } from 'zeromq';
+import {Publisher, Subscriber} from 'zeromq';
-import { PUBLISH_BIND_PORT, PUBLISH_BIND_ADDR} from './config';
+import {PUBLISH_BIND_ADDR, PUBLISH_BIND_PORT} from './config';
+import {PeerAddress} from './types';
 
 export const publishSock = new Publisher();
 export const subscribeSock = new Subscriber();
 
 export async function bindPublish() {
   const addrStr = `tcp://${PUBLISH_BIND_ADDR}:${PUBLISH_BIND_PORT}`;
+    console.log('addrStr:', addrStr);
   await publishSock.bind(addrStr);
   console.log(`Publishing socket bound to ${addrStr}`);
 }
 
-export function connectSubscribe(host: string, port: number) {
+export function connectSubscribe(publishAddr: PeerAddress) {
   // TODO: peer discovery
-  const addrStr = `tcp://${host}:${port}`;
+  const addrStr = `tcp://${publishAddr.toAddrString()}`;
+  console.log('connectSubscribe', {addrStr});
   subscribeSock.connect(addrStr);
   subscribeSock.subscribe("deltas");
   console.log(`Subscribing to ${addrStr}`);

src/types.ts

@@ -4,10 +4,11 @@ export type Pointer = {
   targetContext?: string
 };
 
-export type Delta =   {
+export type Delta = {
   creator: string,
   host: string,
   pointers: Pointer[],
+  receivedFrom?: PeerAddress,
 }
 
 export type DeltaContext = Delta & {
@@ -15,17 +16,17 @@ export type DeltaContext = Delta & {
 };
 
 export type Query = {
-    filterExpr: JSON
+  filterExpr: JSON
 };
 
 export type QueryResult = {
-    deltas: Delta[]
+  deltas: Delta[]
 };
 
 export enum Decision {
-    Accept,
+  Accept,
-    Reject,
+  Reject,
-    Defer
+  Defer
 };
 
 
@@ -37,3 +38,24 @@ export type FilterGenerator = () => FilterExpr;
 
 export type PropertyTypes = string | number | undefined;
 
+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() {
+    console.log('toAddrStr...', {addr: this.addr, port: this.port});
+    return `${this.addr}:${this.port}`;
+  }
+  toJSON() {
+    console.log('toAddrStr...', {addr: this.addr, port: this.port});
+    return `${this.addr}:${this.port}`;
+  }
+};
+