much more efficient

client/lib/patch-db.ts

@@ -111,38 +111,24 @@ export class PatchDB<T extends { [key: string]: any }> {
   watch$(...args: (string | number)[]): Observable<any> {
     const path = `/${args.join('/')}`
 
-    return new Observable(subscriber => {
+    if (!this.watchedNodes[path]) {
       const data = this.cache$.value.data
       const value = getValueByPointer(data, path)
-      const source = this.watchedNodes[path] || new BehaviorSubject(value)
-      const subscription = source.subscribe(subscriber)
+      this.watchedNodes[path] = new BehaviorSubject(value)
+    }
 
-      this.watchedNodes[path] = source
-      this.updateWatchedNode(path, data)
-
-      return () => {
-        subscription.unsubscribe()
-
-        if (!source.observed) {
-          source.complete()
-          delete this.watchedNodes[path]
-        }
-      }
-    })
+    return this.watchedNodes[path]
   }
 
   proccessUpdates(updates: Update<T>[], cache: DBCache<T>) {
     updates.forEach(update => {
-      if (update.id <= cache.sequence) return
-
       if (this.isRevision(update)) {
-        if (update.id > cache.sequence + 1) {
-          console.error(
-            `Received futuristic revision. Expected ${
-              cache.sequence + 1
-            }, got ${update.id}`,
+        const expected = cache.sequence + 1
+        if (update.id < expected) return
+        if (update.id > expected) {
+          return console.error(
+            `Received futuristic revision. Expected ${expected}, got ${update.id}`,
           )
-          return
         }
         this.handleRevision(update, cache)
       } else {
@@ -154,22 +140,25 @@ export class PatchDB<T extends { [key: string]: any }> {
   }
 
   private handleRevision(revision: Revision, cache: DBCache<T>): void {
+    // apply opperations
     revision.patch.forEach(op => {
       applyOperation(cache, op)
-      this.updateWatchedNodes(op.path, cache.data)
+    })
+    // update watched nodes
+    revision.patch.forEach(op => {
+      Object.keys(this.watchedNodes).forEach(watchedPath => {
+        const revisionPath = op.path
+        if (revisionPath.includes(watchedPath)) {
+          this.updateWatchedNode(watchedPath, cache.data)
+        }
+      })
     })
   }
 
   private handleDump(dump: Dump<T>, cache: DBCache<T>): void {
     cache.data = { ...dump.value }
-    this.updateWatchedNodes('', cache.data)
-  }
-
-  private updateWatchedNodes(revisionPath: string, data: T) {
     Object.keys(this.watchedNodes).forEach(path => {
-      if (path.includes(revisionPath) || revisionPath.includes(path)) {
-        this.updateWatchedNode(path, data)
-      }
+      this.updateWatchedNode(path, cache.data)
     })
   }