Updates

state-crdts/hyper-p2p-crdt-version-vector/index.js

@@ -1,31 +1,22 @@
 require('bare-process/global')
 const EventEmitter = require('bare-events')
-const { assertNonEmpty } = require('../../_shared/lib/errors.js')
+const {
+  crdtStats,
+  compareVectors,
+  happensBefore,
+  concurrentVectors,
+  assertPeer
+} = require('../../_shared/crdt-base.js')
 const { initModuleSwarm, gossipSend } = require('../../_shared/p2p-bare.js')
 
 const PROTOCOL = 'crdt-version-vector/v1'
 
-function compareVectors (a, b) {
-  const keys = new Set([...Object.keys(a || {}), ...Object.keys(b || {})])
-  let aDom = false
-  let bDom = false
-  for (const k of keys) {
-    const av = a[k] || 0
-    const bv = b[k] || 0
-    if (av > bv) aDom = true
-    if (bv > av) bDom = true
-  }
-  if (!aDom && !bDom) return 'equal'
-  if (aDom && !bDom) return 'after'
-  if (bDom && !aDom) return 'before'
-  return 'concurrent'
-}
-
 class HyperP2PCrdtVersionVector extends EventEmitter {
   constructor (opts = {}) {
     super()
     this.topic = opts.topic || null
     this.keyPair = opts.keyPair || require('hypercore-crypto').keyPair()
+    this.localPeer = opts.peerId || 'local'
     this._clock = new Map()
     this._stats = { ops: 0, gossipIn: 0, gossipOut: 0 }
     this.swarm = null
@@ -33,15 +24,19 @@ class HyperP2PCrdtVersionVector extends EventEmitter {
   }
 
   increment (peer) {
-    assertNonEmpty(peer, 'peer')
-    const v = (this._clock.get(peer) || 0) + 1
-    this._clock.set(peer, v)
+    const p = assertPeer(peer)
+    const v = (this._clock.get(p) || 0) + 1
+    this._clock.set(p, v)
     this._stats.ops++
-    gossipSend(this, { type: 'crdt-version-vector-sync', peer, value: v })
+    gossipSend(this, { type: 'crdt-version-vector-sync', peer: p, value: v })
     this._stats.gossipOut++
     return v
   }
 
+  tick (peer = null) {
+    return this.increment(peer || this.localPeer)
+  }
+
   merge (other) {
     if (!other || typeof other !== 'object') return 0
     let n = 0
@@ -52,6 +47,7 @@ class HyperP2PCrdtVersionVector extends EventEmitter {
         n++
       }
     }
+    if (n) this.emit('merge', { updated: n })
     return n
   }
 
@@ -59,10 +55,22 @@ class HyperP2PCrdtVersionVector extends EventEmitter {
     return compareVectors(a, b)
   }
 
-  toJSON () {
+  happensBefore (a, b) {
+    return happensBefore(a, b)
+  }
+
+  isConcurrent (a, b) {
+    return concurrentVectors(a, b)
+  }
+
+  snapshot () {
     return Object.fromEntries(this._clock)
   }
 
+  toJSON () {
+    return this.snapshot()
+  }
+
   _onGossip (d) {
     if (!d || d.type !== 'crdt-version-vector-sync' || !d.peer) return
     this._stats.gossipIn++
@@ -70,7 +78,7 @@ class HyperP2PCrdtVersionVector extends EventEmitter {
   }
 
   getStats () {
-    return { ...this._stats, peers: this._clock.size, protocol: PROTOCOL }
+    return crdtStats(this._stats, PROTOCOL, { peers: this._clock.size })
   }
 
   async ready () {
@@ -88,7 +96,12 @@ class HyperP2PCrdtVersionVector extends EventEmitter {
     if (this.swarm) await this.swarm.destroy().catch(() => {})
     this.swarm = null
     this._peerMsgs = null
+    this.emit('closed')
   }
 }
 
-module.exports = { HyperP2PCrdtVersionVector, PROTOCOL, compareVectors }
+module.exports = {
+  HyperP2PCrdtVersionVector,
+  PROTOCOL,
+  compareVectors
+}

state-crdts/hyper-p2p-crdt-version-vector/package.json

@@ -1,6 +1,6 @@
 {
   "name": "hyper-p2p-crdt-version-vector",
-  "version": "0.3.1",
+  "version": "0.3.2",
   "description": "Version vector CRDT.",
   "main": "index.js",
   "type": "commonjs",

state-crdts/hyper-p2p-crdt-version-vector/test/test.js

@@ -16,6 +16,16 @@ test('increment and merge', async (t) => {
   await m.close()
 })
 
+test('tick and ordering helpers', async (t) => {
+  const m = new HyperP2PCrdtVersionVector({ peerId: 'local' })
+  m.tick()
+  const snap = m.snapshot()
+  t.is(snap.local, 1)
+  t.ok(m.happensBefore({ a: 1 }, { a: 2 }))
+  t.ok(m.isConcurrent({ a: 1, b: 2 }, { a: 2, b: 1 }))
+  await m.close()
+})
+
 test('compare vectors', (t) => {
   t.is(compareVectors({ a: 2 }, { a: 1 }), 'after')
   t.is(compareVectors({ a: 1 }, { a: 2 }), 'before')
@@ -32,5 +42,6 @@ test('getStats', async (t) => {
   const m = new HyperP2PCrdtVersionVector()
   m.increment('x')
   t.is(m.getStats().protocol, PROTOCOL)
+  t.is(m.getStats().mode, 'crdt')
   await m.close()
 })