our own json patch lib and everything working

client/index.ts

@@ -1,6 +1,7 @@
 export * from './lib/source/poll-source'
 export * from './lib/source/ws-source'
 export * from './lib/source/source'
+export * from './lib/json-patch-lib'
 export * from './lib/patch-db'
 export * from './lib/store'
 export * from './lib/types'

client/lib/json-patch-lib.ts

@@ -0,0 +1,86 @@
+export interface Validator<T> {
+  (operation: Operation, index: number, document: T, existingPathFragment: string): void
+}
+
+export interface BaseOperation {
+  path: string
+}
+
+export interface AddOperation<T> extends BaseOperation {
+  op: 'add'
+  value: T
+}
+
+export interface RemoveOperation extends BaseOperation {
+  op: 'remove'
+}
+
+export interface ReplaceOperation<T> extends BaseOperation {
+  op: 'replace'
+  value: T
+}
+
+export type Doc = { [key: string]: any }
+
+export type Operation = AddOperation<any> | RemoveOperation | ReplaceOperation<any>
+
+export function getValueByPointer (document: any, pointer: string): any {
+  const pathArr = pointer.split('/')
+  pathArr.shift()
+  try {
+    return pathArr.reduce((acc, next) => acc[next], document)
+  } catch (e) {
+    return undefined
+  }
+}
+
+export function applyOperation (document: Doc, op: Operation): Operation | null {
+  let undo: Operation | null = null
+  const pathArr = op.path.split('/')
+  pathArr.shift()
+  pathArr.reduce((node, key, i) => {
+    if (!isObject) {
+      throw Error('patch cannot be applied.  Path contains non object')
+    }
+
+    if (i < pathArr.length - 1) {
+      // iterate node
+      return node[key]
+    }
+
+    // if last key
+    const curVal = node[key]
+    if (op.op === 'add' || op.op === 'replace') {
+      node[key] = op.value
+      if (curVal) {
+        undo = {
+          op: 'replace',
+          path: op.path,
+          value: curVal,
+        }
+      } else {
+        undo = {
+          op: 'remove',
+          path: op.path,
+        }
+      }
+    } else {
+      delete node[key]
+      if (curVal) {
+        undo = {
+          op: 'add',
+          path: op.path,
+          value: curVal,
+        }
+      }
+    }
+  }, document)
+
+  return undo
+}
+
+function isObject (val: any): val is Doc {
+  return typeof val === 'object' && !Array.isArray(val) && val !== null
+}
+
+

client/lib/patch-db.ts

@@ -4,8 +4,6 @@ import { Source } from './source/source'
 import { Store } from './store'
 import { DBCache, Http } from './types'
 
-export { Operation } from 'fast-json-patch'
-
 export class PatchDB<T> {
   store: Store<T>
 

client/lib/store.ts

@@ -1,14 +1,19 @@
 import { DBCache, Dump, Http, Revision, Update } from './types'
-import { applyPatch, getValueByPointer } from 'fast-json-patch'
 import { BehaviorSubject, Observable } from 'rxjs'
 import { finalize } from 'rxjs/operators'
+import { applyOperation, getValueByPointer, Operation } from './json-patch-lib'
 import BTree from 'sorted-btree'
 
+export interface StashEntry {
+  revision: Revision
+  undo: Operation[]
+}
+
 export class Store<T> {
   cache: DBCache<T>
   sequence$: BehaviorSubject<number>
-  private nodes: { [path: string]: BehaviorSubject<any> } = { }
-  private stashed = new BTree<number, Revision>()
+  private watchedNodes: { [path: string]: BehaviorSubject<any> } = { }
+  private stash = new BTree<number, StashEntry>()
 
   constructor (
     private readonly http: Http<T>,
@@ -27,18 +32,18 @@ export class Store<T> {
   watch$<P1 extends keyof T, P2 extends keyof T[P1], P3 extends keyof T[P1][P2], P4 extends keyof T[P1][P2][P3], P5 extends keyof T[P1][P2][P3][P4], P6 extends keyof T[P1][P2][P3][P4][P5]> (p1: P1, p2: P2, p3: P3, p4: P4, p5: P5, p6: P6): Observable<T[P1][P2][P3][P4][P5][P6]>
   watch$ (...args: (string | number)[]): Observable<any> {
     const path = `/${args.join('/')}`
-    if (!this.nodes[path]) {
-      this.nodes[path] = new BehaviorSubject(getValueByPointer(this.cache.data, path))
-      this.nodes[path].pipe(
-        finalize(() => delete this.nodes[path]),
+    if (!this.watchedNodes[path]) {
+      this.watchedNodes[path] = new BehaviorSubject(getValueByPointer(this.cache.data, path))
+      this.watchedNodes[path].pipe(
+        finalize(() => delete this.watchedNodes[path]),
       )
     }
-    return this.nodes[path].asObservable()
+    return this.watchedNodes[path].asObservable()
   }
 
   update (update: Update<T>): void {
-    // if stale, return
-    if (update.id <= this.cache.sequence) return
+    // if old or known, return
+    if (update.id <= this.cache.sequence || this.stash.get(update.id)) return
 
     if (this.isRevision(update)) {
       this.handleRevision(update)
@@ -48,54 +53,94 @@ export class Store<T> {
   }
 
   reset (): void {
-    Object.values(this.nodes).forEach(node => node.complete())
-    this.stashed.clear()
+    Object.values(this.watchedNodes).forEach(node => node.complete())
+    this.stash.clear()
   }
 
   private handleRevision (revision: Revision): void {
     // stash the revision
-    this.stashed.set(revision.id, revision)
-    // if revision is futuristic, fetch missing revisions and return
+    this.stash.set(revision.id, { revision, undo: [] })
+
+    // if revision is futuristic, fetch missing revisions
     if (revision.id > this.cache.sequence + 1) {
       this.http.getRevisions(this.cache.sequence)
-      return
-    // if revision is next in line, apply contiguous stashed
-    } else {
-      this.processStashed(revision.id)
     }
+
+    this.processStashed(revision.id)
   }
 
   private handleDump (dump: Dump<T>): void {
     this.cache.data = dump.value
-    this.stashed.deleteRange(this.cache.sequence, dump.id, false)
-    this.updateNodesByPath('')
+    this.stash.deleteRange(this.cache.sequence, dump.id, false)
+    this.updateWatchedNodes('')
     this.updateSequence(dump.id)
     this.processStashed(dump.id + 1)
   }
 
   private processStashed (id: number): void {
-    while (true) {
-      const revision = this.stashed.get(id)
-      if (!revision) break
-      applyPatch(this.cache.data, revision.patch, true, true)
-      revision.patch.map(op => {
-        this.updateNodesByPath(op.path)
-      })
-      this.updateSequence(id)
-      id++
-    }
-    this.stashed.deleteRange(0, id, false)
+    this.undoRevisions(id)
+    this.applyRevisions(id)
   }
 
-  private updateNodesByPath (revisionPath: string) {
-    Object.keys(this.nodes).forEach(nodePath => {
-      if (!this.nodes[nodePath]) return
-      if (nodePath.includes(revisionPath) || revisionPath.includes(nodePath)) {
-        try {
-          this.nodes[nodePath].next(getValueByPointer(this.cache.data, nodePath))
-        } catch (e) {
-          this.nodes[nodePath].complete()
-          delete this.nodes[nodePath]
+  private undoRevisions (id: number): void {
+    let stashEntry = this.stash.get(this.stash.maxKey() as number)
+
+    while (stashEntry && stashEntry.revision.id > id) {
+      stashEntry.undo.forEach(u => {
+        applyOperation(document, u)
+      })
+      stashEntry = this.stash.nextLowerPair(stashEntry.revision.id)?.[1]
+    }
+  }
+
+  private applyRevisions (id: number): void {
+    let revision = this.stash.get(id)?.revision
+
+    while (revision) {
+      let undo: Operation[] = []
+
+      let success = false
+      try {
+        revision.patch.forEach(op => {
+          const u = applyOperation(this.cache.data, op)
+          if (u) undo.push(u)
+        })
+        success = true
+      } catch (e) {
+        undo.forEach(u => {
+          applyOperation(document, u)
+        })
+        undo = []
+      }
+
+      if (success) {
+        revision.patch.forEach(op => {
+          this.updateWatchedNodes(op.path)
+        })
+      }
+
+      if (revision.id === this.cache.sequence + 1) {
+        this.updateSequence(revision.id)
+      } else {
+        this.stash.set(revision.id, { revision, undo })
+      }
+
+      // increment revision for next loop
+      revision = this.stash.nextHigherPair(revision.id)?.[1].revision
+    }
+
+    // delete all old stashed revisions
+    this.stash.deleteRange(0, this.cache.sequence, false)
+  }
+
+  private updateWatchedNodes (revisionPath: string) {
+    Object.keys(this.watchedNodes).forEach(path => {
+      if (path.includes(revisionPath) || revisionPath.includes(path)) {
+        const val = getValueByPointer(this.cache.data, path)
+        if (val !== undefined) {
+          this.watchedNodes[path].next(val)
+        } else {
+          this.watchedNodes[path].complete()
         }
       }
     })

client/lib/types.ts

@@ -1,4 +1,4 @@
-import { Operation } from 'fast-json-patch'
+import { Operation } from './json-patch-lib'
 
 // revise a collection of nodes.
 export type Revision = { id: number, patch: Operation[], expireId: string | null }

client/package.json

@@ -11,23 +11,13 @@
     "test": "mocha -r ts-node/register tests/**/*.test.ts"
   },
   "dependencies": {
-    "fast-json-patch": "^3.0.0-1",
-    "jsonpointer": "^4.1.0",
-    "mobx": "^6.1.4",
-    "mobx-utils": "^6.0.3",
     "rxjs": "^6.6.3",
     "sorted-btree": "^1.5.0",
     "uuid": "^8.3.2"
   },
   "devDependencies": {
-    "@types/chai": "^4.2.14",
-    "@types/chai-string": "^1.4.2",
-    "@types/mocha": "^8.2.0",
     "@types/node": "^15.0.0",
     "@types/uuid": "^8.3.0",
-    "chai": "^4.2.0",
-    "chai-string": "^1.5.0",
-    "mocha": "^8.2.1",
     "ts-node": "^9.1.1",
     "tslint": "^6.1.0",
     "typescript": "4.1.5"