Implemented improved plugin for synchronous collaboration, with conflict detection and server-side persistent state.

lib/versioning/History.js

@@ -3,18 +3,26 @@
 const { v4: uuidv4 } = require("uuid");
 
 class Operation {
-  constructor(id, value, parents) {
+  constructor(id, detail) {
     this.id = id;
-    this.value = value;
-    this.parents = parents; // object mapping every key in value to a parent Operation
+    this.detail = detail;
   }
   // Basically replaces JS references by IDs.
   // Result can be JSON'd with constant time+space complexity. Useful for sharing an edit over the network.
   serialize(op) {
+    const self = this; // workaround
     return {
       id: this.id,
-      value: this.value,
-      parentIds: Object.fromEntries(Object.entries(this.parents).map(([key,parent])=>[key,parent.id])),
+      detail: Object.fromEntries(
+        (function*() {
+          for (const [key, {value, parent, depth}] of self.detail.entries()) {
+            yield [key, {
+              value,
+              parentId: parent.id,
+              depth,
+            }];
+          }
+        })()),
     }
   }
 }
@@ -25,7 +33,7 @@ class Context {
     this.fetchCallback = fetchCallback;
 
     // "Global" stuff. Operations have GUIDs but can also be shared between Histories. For instance, the 'initial' operation is the common root of all model histories. We could have put these things in a global variable, but that would make it more difficult to mock 'remoteness' (separate contexts) in tests.
-    this.initialOp = new Operation("0", {}, {}); // The parent of all parentless Operations. Root of all histories.
+    this.initialOp = new Operation("0", new Map()); // The parent of all parentless Operations. Root of all histories.
     this.ops = new Map(); // contains all pending or resolved operation-requests; mapping from operation-id to Promise resolving to Operation.
     this.ops.set(this.initialOp.id, Promise.resolve(this.initialOp));
   }
@@ -44,7 +52,6 @@ class Context {
   receiveOperation(serialized) {
     let promise = this.ops.get(serialized.id);
     if (promise === undefined) {
-      // console.log("Hopefully ASSERTION DOESN'T FAIL...")
       promise = this._awaitParents(serialized);
       this.ops.set(serialized.id, promise);
     }
@@ -52,12 +59,15 @@ class Context {
   }
 
   // Internal function. Do not use directly.
-  async _awaitParents({id, value, parentIds}) {
-    // if (!this.ops.has(id)) throw new Error("ASSERTION FAILED: must have been added to ops already");
-    const dependencies = Object.entries(parentIds).map(async ([key, parentId]) => [key, await this.requestOperation(parentId)]);
-    const pairs = await Promise.all(dependencies);
-    const parents = Object.fromEntries(pairs);
-    return new Operation(id, value, parents);
+  async _awaitParents({id, detail}) {
+    const dependencies = Object.entries(detail).map(async ([key, {value, parentId, depth}]) => {
+      return [key, {
+        value,
+        parent: await this.requestOperation(parentId),
+        depth,
+      }];
+    });
+    return new Operation(id, new Map(await Promise.all(dependencies)));
   }
 }
 
@@ -80,16 +90,24 @@ class History {
   _getHead(key) {
     const op = this.heads.get(key);
     if (op !== undefined) {
-      return op;
-    }
-    return this.context.initialOp;
+      return {
+        op,
+        depth: op.detail.get(key).depth,
+      };
+    };
+    return {
+      op: this.context.initialOp,
+      depth: 0,
+    };
   }
 
-  _exec(op) {
+  _exec(op, setState) {
     // update mapping to point to op
-    for (const [key, val] of Object.entries(op.value)) {
+    for (const [key, {value}] of op.detail.entries()) {
       this.heads.set(key, op); // update HEAD ptr
-      this.setState(key, val);
+      if (setState) {
+        this.setState(key, value);
+      }
     }
   }
 
@@ -110,19 +128,24 @@ class History {
 
   // To be called when a new user operation has happened locally.
   // The new operation advances HEADs.
-  new(value) {
-    const parents = {};
-    const newOp = new Operation(uuidv4(), value, parents);
-    for (const [key,val] of Object.entries(value)) {
-      const parent = this._getHead(key);
-      parents[key] = parent;
+  new(v, setState=true) {
+    const newId = uuidv4();
+    const detail = new Map(Object.entries(v).map(([key,value]) => {
+      const {op: parent, depth} = this._getHead(key);
+      return [key, {
+        value,
+        parent,
+        depth: depth + 1,
+      }];
+    }));
+    const newOp = new Operation(newId, detail);
+    for (const [key, {parent}] of detail.entries()) {
       this._setChild(parent, key, newOp);
     }
+    this._exec(newOp, setState);
 
-    this._exec(newOp);
-
-    this.context.ops.set(newOp.id, Promise.resolve(newOp));
-    this.ops.set(newOp.id, newOp);
+    this.context.ops.set(newId, Promise.resolve(newOp));
+    this.ops.set(newId, newOp);
 
     return newOp;
   }
@@ -136,7 +159,7 @@ class History {
     }
 
     let exec = true;
-    for (const [key, parent] of Object.entries(op.parents)) {
+    for (const [key, {parent}] of op.detail.entries()) {
       if (!this.ops.has(parent.id)) {
         // Update this History with operation's dependencies first
         this.autoMerge(parent);
@@ -146,12 +169,12 @@ class History {
       const sibling = this._getChild(parent, key);
       if (sibling) {
         // Conflict
-        if (this.resolve(op.value, sibling.value)) {
+        if (this.resolve(op, sibling)) {
           // console.log("conflict: op wins")
           const visited = new Set();
           const rollback = op => {
             visited.add(op); // Children form a DAG, with possible 'diamond' shapes -> prevent same operation from being visited more than once.
-            for (const [key, parent] of Object.entries(op.parents)) {
+            for (const [key, {parent}] of op.detail.entries()) {
               // recurse, child-first
               const child = this._getChild(op, key);
               if (child && !visited.has(child)) {
@@ -159,8 +182,14 @@ class History {
                 rollback(child);
               }
               // rollback
-              this.heads.set(key, parent);
-              this.setState(key, parent.value[key]);
+              if (parent === this.context.initialOp) {
+                // Invariant: HEADs never contains initialOp
+                this.heads.delete(key);
+                this.setState(key, undefined);
+              } else {
+                this.heads.set(key, parent);
+                this.setState(key, parent.detail.get(key).value);
+              }
             }
           };
           // Received operation wins conflict - state must be rolled back before executing it
@@ -176,14 +205,14 @@ class History {
       }
       // won (or no conflict):
       this._setChild(parent, key, op);
-      if (parent !== this._getHead(key)) {
+      if (parent !== this._getHead(key).op) {
         // only execute received operation if it advances HEAD
         exec = false;
       }
     }
 
     if (exec) {
-      this._exec(op);
+      this._exec(op, true);
     }
 
     this.ops.set(op.id, op);
@@ -195,6 +224,27 @@ class History {
     this.autoMerge(op);
     return op;
   }
+
+  // Get operations in history in a sequence, such that any operation's dependencies precede it in the list. To reproduce the state of this History, operations can be executed in the returned order (front to back), and are guaranteed to not give conflicts.
+  getOpsSequence() {
+    const added = new Set([this.context.initialOp]);
+    const visiting = new Set();
+    const seq = [];
+    const visit = op => {
+      if (!added.has(op)) {
+        visiting.add(op);
+        for (const [key, {parent}] of op.detail.entries()) {
+          visit(parent);
+        }
+        seq.push(op);
+        added.add(op);
+      }
+    }
+    for (const op of this.heads.values()) {
+      visit(op);
+    }
+    return seq;
+  }
 }
 
-module.exports = { Context, History };
+module.exports = { Context, History, uuidv4 };

lib/versioning/README.txt

@@ -0,0 +1,34 @@
+Synchronous collaboration for drawio.
+
+
+Steps to run server
+-------------------
+
+ - Install NodeJS and NPM
+
+ - In 'lib' dir, run:
+      npm i serve-handler ws uuid
+
+ - Setup server state directory.
+      mkdir /desired/path
+      mkdir /desired/path/ops
+      mkdir /desired/path/branches
+
+ - In root dir, run:
+      DRAWIOSTATEDIR=/desired/path node lib/versioning/run_server.js
+
+
+Steps to build + run client
+---------------------------
+
+ - Build SCCD model:
+      cd lib/versioning
+      python -m sccd.compiler.sccdc -l javascript -p eventloop client.xml -o client.js
+
+ - Browserify plugin:
+      cd src/main/webapp/plugins/cdf
+      browserify versioning.js > versioning.browser.js
+
+ - Make sure server is running and navigate to
+      http://localhost:8700/src/main/webapp?dev=1&p=versioning
+

lib/versioning/build_client.sh

@@ -0,0 +1,2 @@
+#!/bin/sh
+python -m sccd.compiler.sccdc -l javascript -p eventloop -o client.js client.xml && browserify ../../src/main/webapp/plugins/cdf/versioning.js > ../../src/main/webapp/plugins/cdf/versioning.browser.js

lib/versioning/client.xml

@@ -0,0 +1,372 @@
+<?xml version="1.0" ?>
+<diagram author="Joeri Exelmans" name="client">
+  <description>Example of a browser-based WebSocket client with heartbeats</description>
+  <top></top>
+  <inport name="in"/>
+  <outport name="out"/>
+  <class name="Main" default="true">
+    <relationships>
+      <association name="socket" class="Socket" min="1" max="1"/>
+    </relationships>
+    <inport name="ack"/>
+    <inport name="socket"/>
+    <constructor>
+      <parameter name="uiState"/>
+      <body>
+        console.log("Main constructor");
+        this.uiState = uiState;
+        this.connected = false;
+      </body>
+    </constructor>
+    <scxml
+      big_step_maximality="take_many"
+      internal_event_lifeline="next_combo_step">
+      <parallel id="p">
+
+        <state id="response_handler">
+          <onentry>
+            <script>
+              this.reqHandlers = new Map();
+              this.reqCounter = 0;
+              this.sendReq = (obj, handler) => {
+                const reqId = this.reqCounter++;
+                const req = {
+                  reqId,
+                  ...obj,
+                };
+                console.log("Sending req", req);
+                this.reqHandlers.set(reqId, handler);
+                this.addEvent(new Event("send", null, [req]));
+
+                // allow to cancel the request handler
+                return () => {
+                  this.reqHandlers.delete(reqId);
+                };
+              }
+            </script>
+          </onentry>
+          <state id="default">
+            <transition event="received" target=".">
+              <parameter name="parsed"/>
+              <script>
+                if (parsed.type === 'ack') {
+                  const handler = this.reqHandlers.get(parsed.reqId);
+                  if (handler) {
+                    handler(parsed);
+                  }
+                }
+              </script>
+            </transition>
+            <transition event="disconnected" target=".">
+              <script>
+                // no way we will get a response for our pending requests
+                this.reqHandlers.clear();
+                this.reqCounter = 0;
+              </script>
+            </transition>
+          </state>
+        </state>
+
+        <state id="mode" initial="uninitialized">
+          <state id="uninitialized">
+            <transition event="init_offline" port="in" target="../async_disconnected"/>
+            <transition event="init_join" port="in" target="../can_leave/session_set/rejoin_when_online">
+              <parameter name="sessionId"/>
+              <script>
+                this.sessionId = sessionId;
+              </script>
+            </transition>
+          </state>
+
+          <state id="async_disconnected">
+            <onentry><script>
+              this.uiState.setOfflineDisconnected();
+            </script></onentry>
+            <transition event="connected" target="../async_connected"/>
+          </state><!-- async_disconnected -->
+
+          <state id="async_connected">
+            <onentry><script>
+              this.uiState.setOfflineConnected();
+            </script></onentry>
+
+            <transition event="disconnected" target="../async_disconnected"/>
+
+            <transition event="join" port="in" target="../can_leave/session_set/waiting_for_join_ack">
+              <parameter name="sessionId"/>
+              <script> this.sessionId = sessionId; </script>
+            </transition>
+
+            <transition event="new_share" port="in" target="../can_leave/waiting_for_new_share_ack">
+              <parameter name="ops"/>
+              <script>
+                // console.log("new share event, ops=", ops);
+                this.ops = ops;
+              </script>
+            </transition>
+          </state><!-- async_connected -->
+
+          <state id="can_leave" initial="session_set">
+
+            <transition event="leave" cond='this.connected' port="in" target="../async_connected"/>
+            <transition event="leave" cond='this.disconnected' port="in" target="../async_disconnected"/>
+
+            <onexit><script>
+              this.sendReq({type:"leave"});
+            </script></onexit>
+
+            <state id="waiting_for_new_share_ack">
+              <onentry><script>
+                this.uiState.setCreatingShare();
+                this.deleteNewShareHandler = this.sendReq(
+                  {type: "new_share", ops: this.ops},
+                  res => {
+                    // called during transition (as action code), so no need to 'wake up' the controller like we do with 'addInput':
+                    this.addEvent(new Event("ack_new_share", "ack", [res.sessionId]));
+                  });
+              </script></onentry>
+              <onexit><script>
+                this.deleteNewShareHandler();
+              </script></onexit>
+              <transition event="disconnected" target="../reshare_when_online"/>
+              <transition event="ack_new_share" port="ack" target="../session_set/joined">
+                <parameter name="sessionId"/>
+                <script>
+                  this.sessionId = sessionId;
+                </script>
+                <raise event="ack_new_share" port="out" scope="output">
+                  <parameter expr="sessionId"/>
+                </raise>
+              </transition>
+            </state><!-- waiting_for_new_share_ack -->
+
+            <state id="reshare_when_online">
+              <onentry><script>
+                this.uiState.setReconnecting();
+              </script></onentry>
+              <transition event="connected" target="../waiting_for_new_share_ack"/>
+            </state>
+
+            <state id="session_set" initial="waiting_for_join_ack">
+              <onentry><script>
+                if (!this.sessionId) {
+                  throw new Error("DEBUG: no session id");
+                }
+                this.uiState.setSession(this.sessionId);
+              </script></onentry>
+              <onexit> <script>
+                this.uiState.unsetSession();
+              </script> </onexit>
+
+               <state id="waiting_for_join_ack">
+                  <onentry><script>
+                    this.uiState.setJoining();
+
+                    this.deleteJoinHandler = this.sendReq(
+                      {type: "join", sessionId: this.sessionId},
+                      res => {
+                        // called during transition (as action code), so no need to 'wake up' the controller like we do with 'addInput':
+                        this.addEvent(new Event("ack_join", "ack", [res.ops]));
+                      });
+                  </script></onentry>
+                  <onexit><script>
+                    this.deleteJoinHandler();
+                  </script></onexit>
+                  <transition event="ack_join" port="ack" target="../joined">
+                    <parameter name="ops"/>
+                    <raise event="ack_join" port="out" scope="output">
+                      <parameter expr="ops"/>
+                    </raise>
+                  </transition>
+                  <transition event="disconnected" target="../rejoin_when_online"/>
+                </state><!-- waiting_for_join_ack -->
+
+                <state id="joined">
+                  <onentry>
+                    <script>
+                      this.uiState.setOnline();
+                    </script>
+                    <raise event="joined" port="out" scope="output"/>
+                  </onentry>
+                  <onexit>
+                    <raise event="left" port="out" scope="output"/>
+                  </onexit>
+                  <transition event="received" cond="parsed.type === 'pub_edit'" target=".">
+                    <parameter name="parsed"/>
+                    <script>
+                      if (parsed.sessionId !== this.sessionId) {
+                        throw new Error("Unexpected: received edit for another session:" +  parsed.sessionId);
+                      }
+                    </script>
+                    <raise event="received_op" port="out" scope="output">
+                      <parameter expr="parsed.op"/>
+                    </raise>
+                  </transition>
+                  <transition event="new_edit" port="in" target=".">
+                    <parameter name="serialized"/>
+                    <script>
+                      // console.log("sending sessionId", this.sessionId);
+                      this.sendReq(
+                        {type:"new_edit", sessionId: this.sessionId, op: serialized},
+                        res => {
+                          // this.unacknowledged.delete(serialized.id);
+                        });
+                      // this.unacknowledged.set(serialized.id, serialized);
+                    </script>
+                  </transition>
+
+                  <transition event="disconnected" target="../rejoin_when_online"/>
+                </state><!-- joined -->
+                <state id="rejoin_when_online">
+                  <onentry><script>
+                    this.uiState.setReconnecting();
+                  </script></onentry>
+                  <!-- Re-join when connection is restored -->
+                  <transition event="connected" target="../waiting_for_join_ack"/>
+                </state><!-- rejoin when online -->
+              </state><!-- session_set -->
+          </state><!-- can_leave -->
+
+        </state><!-- mode -->
+
+        <state id="socket_region" initial="disconnected">
+          <state id="disconnected">
+            <transition event="connect" port="in" target="../connecting_or_connected">
+              <parameter name="addr"/>
+              <script>
+                console.log("received connect", addr)
+                this.addr = addr;
+              </script>
+            </transition>
+          </state>
+
+          <state id="connecting_or_connected" initial="connecting">
+            <!-- Within this state, we do our best to establish a connection, and also attempt to re-establish a connection if an error occurs or if the server closes its end. -->
+
+            <transition event="disconnect" port="in" target="../disconnected">
+              <script>
+                this.socket.close();
+              </script>
+            </transition>
+            <state id="connecting">
+              <onentry>
+                <script>
+                  this.socket = new WebSocket(this.addr);
+
+                  // Translate socket events to statechart events:
+
+                  this.socket.onopen = event => {
+                    this.controller.addInput("open", this.inports["socket"], []);
+                  };
+
+                  this.socket.onmessage = event => {
+                    let parsed;
+                    try {
+                      parsed = JSON.parse(event.data);
+                    } catch (e) {
+                      console.log("received unparsable message", e);
+                      return;
+                    }
+
+                    this.controller.addInput("message", this.inports["socket"], [parsed]);
+                  };
+
+                  this.socket.onerror = event => {
+                    // From what I see, this event only happens when the socket could not connect.
+                    // An 'error' happening after the connection is established, will be indicated by a close event.
+                    this.controller.addInput("error", this.inports["socket"], []);
+                  };
+
+                  this.socket.onclose = event => {
+                    this.controller.addInput("close", this.inports["socket"], []);
+                  };
+                </script>
+              </onentry>
+              <transition event="open" port="socket" target="../connected"/>
+              <transition event="error" port="socket" target="../wait_reconnect">
+                <!-- Error event - emitted when connection could not be established  -->
+                <raise event="error"/>
+              </transition>
+            </state>
+
+            <parallel id="connected">
+              <onentry>
+                <raise event="connected"/>
+                <script> this.connected = true; </script>
+              </onentry>
+              <onexit>
+                <raise event="disconnected"/>
+                <script> this.connected = false; </script>
+              </onexit>
+
+              <!-- Server closed its end -->
+              <transition event="close" port="socket" target="../wait_reconnect"/>
+
+              <state id="send_receive_region">
+                <state id="ready">
+                  <transition event="send" target=".">
+                    <parameter name="json"/>
+                    <script>
+                      this.socket.send(JSON.stringify(json));
+                    </script>
+                  </transition>
+                  <transition event="message" port="socket" target="." cond="parsed.type !== 'pong'">
+                    <parameter name="parsed"/>
+                    <raise event="received">
+                      <parameter expr="parsed"/>
+                    </raise>
+                  </transition>
+                </state>
+              </state>
+
+              <state id="connection_monitor_region" initial="all_good">
+                <parallel id="all_good">
+                  <state id="send_pings">
+                    <state id="waiting">
+                      <!-- reset ping timer each time we send anything -->
+                      <transition event="send" port="in" target="."/>
+                      <!-- when not having sent anything for 1s, send a ping to let server know we're still alive -->
+                      <transition after="500" target=".">
+                        <script>
+                          this.socket.send(JSON.stringify({type:"ping"}));
+                        </script>
+                      </transition>
+                    </state>
+                  </state>
+                  <state id="receive_pongs">
+                    <state id="waiting">
+                      <!-- reset pong timer each time we receive anything -->
+                      <transition event="message" port="socket" target="."/>
+                      <!-- when not having received anything for 3s, decide the server and/or connection are (temporarily) dead -->
+                      <transition after="7000" target=".">
+                        <!-- WORKAROUND: for some reason, target cannot be '../../../timeout' (limitation of 'main' SCCD, fixed in 'joeri' branch), so we generate an internal event to make the transition from higher up -->
+                        <raise event="timeout"/>
+                      </transition>
+                    </state>
+                  </state>
+                  <!-- part of WORKAROUND, see above -->
+                  <transition event="timeout" target="../timeout">
+                    <raise event="timeout"/>
+                  </transition>
+                </parallel>
+                <state id="timeout">
+                  <!-- don't send pings - just wait for any observable server activity -->
+                  <transition event="message" port="socket" target="../all_good">
+                    <raise event="timeout_recovered"/>
+                  </transition>
+                </state>
+              </state>
+            </parallel>
+
+            <state id="wait_reconnect">
+              <!-- This is an intermediate state where we wait for a little while, before trying to reconnect -->
+              <transition after="1" target="../connecting"/>
+            </state>
+          </state>
+        </state><!-- socket region -->
+
+      </parallel>
+
+    </scxml>
+  </class>
+</diagram>

lib/versioning/run_server.js

@@ -0,0 +1,217 @@
+// NodeJS script
+
+const fs = require('fs/promises');
+const fsConstants = require('fs').constants;
+const path = require('path');
+const {EOL} = require('os');
+
+const http = require('http');
+const handler = require('serve-handler');
+
+const IDLENGTH = 36; // UUID v4
+
+const port = process.env.DRAWIOPORT || 8700;
+const stateDir = process.env.DRAWIOSTATEDIR || ".";
+
+async function startServer() {
+  const opsDir = path.join(stateDir, 'ops');
+  const sessionDir = path.join(stateDir, 'sessions');
+  try {
+    process.stdout.write("Can read/write in directory '" + opsDir + "' ? ...");
+    await fs.access(opsDir, fsConstants.R_OK | fsConstants.W_OK);
+    process.stdout.write("OK" + EOL);
+    process.stdout.write("Can read/write in directory '" + sessionDir + "' ? ...");
+    await fs.access(sessionDir, fsConstants.R_OK | fsConstants.W_OK);
+    process.stdout.write("OK" + EOL);
+  } catch (e) {
+    process.stdout.write(EOL + "Please make sure the following directories exist and are writable:" + EOL);
+    process.stdout.write("  " + opsDir + EOL);
+    process.stdout.write("  " + sessionDir + EOL);
+    process.exit(1);
+  }
+
+  const { WebSocketServer } = require('ws');
+  const { v4: uuidv4 } = require("uuid");
+
+  function asyncSleep(ms) {
+    return new Promise(resolve => setTimeout(resolve, ms));
+  }
+
+
+  const httpServer = http.createServer((request, response) => {
+    // You pass two more arguments for config and middleware
+    // More details here: https://github.com/vercel/serve-handler#options
+    console.log(request.method, request.url)
+    return handler(request, response);
+  });
+
+  const wsServer = new WebSocketServer({
+    server: httpServer,
+    path: "/websocket",
+  });
+
+  // Provides async reads and writes to files, while guaranteeing that for every file, a read always reads the latest write.
+  class QueuedIO {
+    constructor(path) {
+      this.path = path;
+      this.queues = new Map();
+    }
+    _queuedIO(f, filename, ...args) {
+      const lastIO = this.queues.get(filename) || Promise.resolve();
+      const nextIO = lastIO.then(() => f(path.join(this.path, filename), ...args));
+      this.queues.set(filename, nextIO.catch(e => {/*console.log("silently caught", e)*/}));
+      return nextIO;
+    }
+    write(filename, data) {
+      // console.log("writing file", filename);
+      return this._queuedIO((path, data) => fs.writeFile(path, data, {encoding: 'utf8'}), filename, data);
+    }
+    append(filename, data) {
+      // console.log("appending to file", filename);
+      const result = this._queuedIO(fs.appendFile, filename, data);
+      // result.then(() => console.log("success")).catch(e => console.log("COULD NOT APPEND", e));
+      return result;
+    }
+    stat(filename) {
+      // console.log("statting file", filename);
+      return this._queuedIO(fs.stat, filename);
+    }
+    read(filename) {
+      // console.log("reading file", filename);
+      return this._queuedIO(path => fs.readFile(path, {encoding:'utf8'}), filename);
+    }
+    writeJSON(filename, json) {
+      // console.log("writeJSONing file", filename);
+      return this.write(filename, JSON.stringify(json));
+    }
+    readJSON(filename) {
+      // console.log("readJSONing file", filename);
+      return this.read(filename).then(JSON.parse);
+    }
+  }
+
+  const opsDB = new QueuedIO(opsDir);
+  const sessionDB = new QueuedIO(sessionDir);
+
+  // mapping of modelId to set of sockets
+  const subscriptions = new Map();
+
+  wsServer.on('connection', function(socket) {
+    console.log("Client connected.")
+    const mySubscriptions = new Set();
+
+    function join(sessionId) {
+      const subbedSockets = subscriptions.get(sessionId);
+      if (subbedSockets) {
+        subbedSockets.add(socket);
+      } else {
+        subscriptions.set(sessionId, new Set([socket]));
+      }
+      mySubscriptions.add(sessionId);
+    }
+    function leave(sessionId) {
+      const subbedSockets = subscriptions.get(sessionId);
+      if (subbedSockets) {
+        subbedSockets.delete(socket);
+      }
+      mySubscriptions.delete(sessionId);      
+    }
+    function leaveAll() {
+      mySubscriptions.forEach(leave);
+    }
+
+    async function handle(req) {
+      if (req.type === "ping") {
+        return {type: "pong"};
+      }
+      else {
+        console.log("-> ", req);
+
+        if (req.type === "join") {
+          const { reqId, sessionId } = req;
+          leaveAll(); // a client can only join 1 session at a time
+          join(sessionId);
+          // Reply with all operations in session
+          const session = await sessionDB.read(sessionId).catch(e => {
+            if (e.code === 'ENOENT') {
+              return ""; // Act as if session exists and is empty
+            } else {
+              throw e;
+            }
+          });
+          const opIds = new Array(session.length / IDLENGTH);
+          for (let i=0; i<opIds.length; i+=1) {
+            const offset = i * IDLENGTH;
+            opIds[i] = session.slice(offset, offset + IDLENGTH);
+          }
+          const ops = await Promise.all(opIds.map(async id => ({id, detail: await opsDB.readJSON(id)})));
+          return {type: "ack", reqId, ops};
+        }
+
+        else if (req.type === "leave") {
+          const { reqId } = req;
+          leaveAll();
+          return {type: "ack", reqId};
+        }
+
+        else if (req.type === "new_share") {
+          const { reqId, ops } = req;
+          const sessionId = uuidv4();
+          await Promise.all(ops.map(({id, detail}) => opsDB.writeJSON(id, detail)));
+          await sessionDB.write(sessionId, ops.map(op=>op.id).join(''));
+          return {type: "ack", reqId, sessionId};
+        }
+
+        else if (req.type === "new_edit") {
+          const { reqId, sessionId, op: {id, detail} } = req;
+          await opsDB.writeJSON(id, detail);
+          await sessionDB.append(sessionId, id); // Creates file if it doesn't exist yet
+          // Best effort broadcast to all subscribers
+          (async () => {
+            const subbedSockets = subscriptions.get(sessionId) || new Set();
+            const broadcastMsg = {type:"pub_edit", sessionId, op: {id, detail}};
+            console.log("BROADCAST to", subbedSockets.size-1,"subscribers...", broadcastMsg);
+            const stringified = JSON.stringify(broadcastMsg);
+            // await asyncSleep(3000);
+            subbedSockets.forEach(subbed => {
+              if (subbed !== socket) { // don't echo
+                try {
+                  subbed.send(stringified); // forward incoming request to all
+                } catch (e) {
+                  console.log("Error:", e, "Closing socket...")
+                  subbed.close(); // in case of an error, it is the client's responsibility to re-subscribe and 'catch up'.
+                }
+              }
+            });
+          })();
+          return {type: "ack", reqId};
+        }
+      }
+    }
+
+    socket.on('message', function(data) {
+      const req = JSON.parse(data);
+      handle(req).then(json => {
+        if (json.type !== 'pong')
+          console.log("<- ", json);
+        socket.send(JSON.stringify(json));
+      }).catch(e => {
+        console.log("Error handling", req, e);
+      })
+    });
+
+    socket.on('close', function() {
+      leaveAll();
+      console.log("Client disconnected.");
+    })
+  });
+
+  httpServer.listen(port);
+  console.log("Listening on", port);
+}
+
+
+startServer().catch(e => {
+  console.log("Error in startServer:", e);
+  process.exit(1);
+});

lib/versioning/test_History.js

@@ -71,12 +71,12 @@ async function runTest(verbose) {
     if (verbose) console.log(...arguments);
   }
 
-  function resolve(value1, value2) {
+  function resolve(op1, op2) {
     // info("resolve...", props1, props2)
-    if (value1.geometry !== value2.geometry) {
-      return value1.geometry > value2.geometry;
+    if (op1.detail.get('geometry').value !== op2.detail.get('geometry').value) {
+      return op1.detail.get('geometry').value > op2.detail.get('geometry').value;
     }
-    return value1.style > value2.style;
+    return op1.detail.get('style').value > op2.detail.get('style').value;
   }
 
   function createAppState(label) {
@@ -94,9 +94,7 @@ async function runTest(verbose) {
     const {setState, state} = createAppState(label);
     // const context = new Context(requestCallback); // simulate 'remoteness' by creating a new context for every History.
 
-    const history = new History(context, setState, resolve,
-      true, // check_assertions
-    );
+    const history = new History(context, setState, resolve);
     return {history, state};
   }
 
@@ -133,7 +131,6 @@ async function runTest(verbose) {
   {
     info("\nTest case: Multi-user without conflict\n")
 
-
     // Local and remote are just names for our histories.
     const localContext = new Context(noFetch);
     const remoteContext = new Context(noFetch);
@@ -144,6 +141,8 @@ async function runTest(verbose) {
     const localOp1 = localHistory.new({geometry: 1});
     await remoteHistory.receiveAndMerge(localOp1.serialize());
 
+    console.log("11")
+
     const remoteOp2 = remoteHistory.new({geometry: 2}); // happens after (hence, overwrites) op1
     await localHistory.receiveAndMerge(remoteOp2.serialize());
 
@@ -233,6 +232,32 @@ async function runTest(verbose) {
     await remoteContext.receiveOperation(localOps[4].serialize());
     assert(deepEqual(fetched, [localOps[1].id, localOps[0].id, localOps[3].id]));
   }
+
+  {
+    info("\nTest case: Get as sequence\n")
+
+    const {history} = createHistory("local", new Context(noFetch));
+
+    const ops = [
+      history.new({x:1, y:1}), // 0
+      history.new({x:2}),      // 1 depends on 0
+      history.new({y:2}),      // 2 depends on 0
+      history.new({x:3, z:3}), // 3 depends on 1
+      history.new({a:4}),      // 4
+      history.new({a:5}),      // 5 depends on 4
+      history.new({a:6, z:6}), // 6 depends on 5, 3
+    ];
+
+    const seq = history.getOpsSequence();
+    console.log(seq.map(op => op.serialize()));
+
+    assert(seq.indexOf(ops[1]) > seq.indexOf(0));
+    assert(seq.indexOf(ops[2]) > seq.indexOf(0));
+    assert(seq.indexOf(ops[3]) > seq.indexOf(1));
+    assert(seq.indexOf(ops[5]) > seq.indexOf(4));
+    assert(seq.indexOf(ops[6]) > seq.indexOf(5));
+    assert(seq.indexOf(ops[6]) > seq.indexOf(3));
+  }
 }
 
 runTest(/* verbose: */ true).then(() => {

sccd

@@ -1 +1 @@
-Subproject commit 15fa3e579bfc6a229ec7fe885c542a31ea391c48
+Subproject commit 49f0b975629052b13bc681a06f4c0858ca93e958

shell.nix

@@ -0,0 +1,28 @@
+{ pkgs ? import <nixpkgs> {} }:
+  pkgs.mkShell {
+    buildInputs = [
+      # to run tests and to run demo server
+      pkgs.nodejs
+      pkgs.nodePackages.npm
+
+      # To build plugin
+      pkgs.nodePackages.browserify
+
+      # to run standalone app
+      pkgs.electron
+
+      # to build (produce minified JS)
+      pkgs.ant
+      pkgs.jre8_headless
+
+      # for SCCD
+      pkgs.python39
+      pkgs.python39Packages.websockets
+    ];
+    # environment variable for standalone app. ignored by browser.
+    shellHook = ''
+      export DRAWIO_ENV=dev
+      #export PYTHONPATH=$PYTHONPATH:~/cdf/repos/drawio/sccd/sccd
+    '';
+    PYTHONPATH = ./sccd;
+}

src/main/webapp/js/diagramly/App.js

@@ -334,6 +334,8 @@ App.pluginRegistry = {
 	'logevents': 'plugins/cdf/logevents.js',
 	'ftgpm': 'plugins/cdf/ftgpm.js',
 	'ftgpm-edit': 'plugins/cdf/ftgpm-edit.js',
+	// 'versioning': 'plugins/cdf/versioning.js',
+	'versioning': 'plugins/cdf/versioning.browser.js',
 	'sendshapes': 'plugins/cdf/sendshapes.js',
 	'screenshare': 'plugins/cdf/screenshare.js',
 	'svg-viewport': 'plugins/cdf/svg-viewport.js',

src/main/webapp/plugins/cdf/versioning.js

@@ -0,0 +1,499 @@
+// Build this plugin with 'browserify':
+//   browserify versioning.js > versioning.browser.js
+
+Draw.loadPlugin(async function(ui) {
+  window.ui = ui;
+  const graph = ui.editor.graph;
+  const model = graph.model;
+
+  await loadScript("../../../../../sccd/docs/runtimes/javascript/statecharts_core.js");
+  await loadScript("../../../../../lib/versioning/client.js");
+
+  class EnabledState {
+    constructor() {
+      this.enabled = false;
+    }
+    enable() {
+      this.enabled = true;
+    }
+    disable() {
+      this.enabled = false;
+    }
+  }
+
+  class UIState {
+    constructor() {
+      this.statusTextNode = document.createTextNode("")
+      this.shareEnabled = new EnabledState();
+      this.joinEnabled = new EnabledState();
+      this.leaveEnabled = new EnabledState();
+    }
+
+    install(ui) {
+      ui.toolbar.addSeparator();
+      ui.menus.put('collab', new Menu((menu, parent) => {
+        const buttonShare = menu.addItem("Share this diagram",
+          null, // image
+          () => {
+            const ops = history.getOpsSequence();
+            const serialized = ops.map(op => op.serialize());
+            controller.addInput("new_share", "in", [serialized]);
+          }, // callback
+          menu,
+          null, // ?
+          this.shareEnabled.enabled); // enabled
+        const buttonJoin = menu.addItem("Join session",
+          null, // image
+          () => {
+            const sessionId = window.prompt("Enter session ID:");
+            if (sessionId) {
+              controller.addInput("join", "in", [sessionId]);
+            }
+          }, // callback
+          menu,
+          null, // ?
+          this.joinEnabled.enabled); // enabled
+        const buttonLeave = menu.addItem("Leave session",
+          null, // image
+          () => {
+            controller.addInput("leave", "in", []);
+          }, // callback
+          menu,
+          null, // ?
+          this.leaveEnabled.enabled); // enabled
+      }));
+
+      const screenshareMenu = ui.toolbar.addMenu('', "Collaboration", true, 'collab');
+      screenshareMenu.style.width = '100px';
+      screenshareMenu.showDisabled = true;
+      screenshareMenu.style.whiteSpace = 'nowrap';
+      screenshareMenu.style.position = 'relative';
+      screenshareMenu.style.overflow = 'hidden';
+      screenshareMenu.innerHTML = "Collaboration" + ui.toolbar.dropdownImageHtml;
+
+      ui.toolbar.addSeparator();
+
+      const statusEl = document.createElement('div');
+      statusEl.classList.add("geLabel");
+      statusEl.style = "white-space: nowrap; position: relative;";
+      statusEl.appendChild(this.statusTextNode);
+      ui.toolbar.container.appendChild(statusEl);
+    }
+
+    setSession(sessionId) {
+      const searchParams = new URLSearchParams(window.location.search);
+      searchParams.set('sessionId', sessionId);
+      const newLocation = window.location.origin + window.location.pathname + '?' + searchParams.toString() + window.location.hash;
+      window.history.replaceState({}, "", newLocation);
+    }
+
+    unsetSession(sessionId) {
+      const searchParams = new URLSearchParams(window.location.search);
+      searchParams.delete('sessionId');
+      const newLocation = window.location.origin + window.location.pathname + '?' + searchParams.toString() + window.location.hash;
+      window.history.replaceState({}, "", newLocation);
+    }
+
+    setOfflineDisconnected() {
+      graph.setEnabled(true);
+      this.statusTextNode.textContent = "Offline (no server connection)";
+
+      this.shareEnabled.disable();
+      this.joinEnabled.disable();
+      this.leaveEnabled.disable();
+    }
+
+    setOfflineConnected() {
+      graph.setEnabled(true);
+      this.statusTextNode.textContent = "Offline";
+
+      this.shareEnabled.enable();
+      this.joinEnabled.enable();
+      this.leaveEnabled.disable();
+    }
+
+    setJoining(sessionId) {
+      graph.setEnabled(false);
+      this.statusTextNode.textContent = "Joining session...";
+
+      this.shareEnabled.disable();
+      this.joinEnabled.disable();
+      this.leaveEnabled.enable();
+    }
+
+    setCreatingShare() {
+      graph.setEnabled(false);
+      this.statusTextNode.textContent = "Creating share...";
+
+      this.shareEnabled.disable();
+      this.joinEnabled.disable();
+      this.leaveEnabled.enable();
+    }
+
+    setOnline(sessionId) {
+      graph.setEnabled(true);
+      this.statusTextNode.textContent = "Online";
+
+      this.shareEnabled.disable();
+      this.joinEnabled.disable();
+      this.leaveEnabled.enable();
+    }
+
+    setReconnecting() {
+      graph.setEnabled(false);
+      this.statusTextNode.textContent = "(re)Connecting...";
+
+      this.shareEnabled.disable();
+      this.joinEnabled.disable();
+      this.leaveEnabled.enable();
+    }
+  }
+
+  const uiState = new UIState();
+
+  const {Context, History, uuidv4} = require("../../../../../lib/versioning/History.js");
+
+  const context = new Context(async id => {
+    throw new Error("Fetch is forbidden. ", id)
+  });
+
+  const codec = new mxCodec();
+  const xmlSerializer = new XMLSerializer();
+  const xmlParser = new DOMParser();
+
+  const setState = (key, value) => {
+    // console.log("setState", key, value);
+
+    function getCell(cellId) {
+      let cell = model.cells[cellId];
+      if (cell === undefined) {
+        // throw new Error("NO SUCH CELL:", cellId);
+      }
+      return cell;
+    }
+
+    function createCell(cellId, isVertex, isEdge) {
+      cell = new mxCell("");
+      cell.setId(cellId);
+      cell.setVertex(isVertex);
+      cell.setEdge(isEdge);
+      model.cells[cellId] = cell; // HACK!
+      return cell;      
+    }
+
+    try {
+      // temporarily disable listener:
+      listenForEdits = false;
+
+      if (key === "root") {
+        // the only global 'key'
+
+        const rootId = value;
+        const root = getCell(rootId) || createCell(rootId, false, false);
+        model.setRoot(root);
+      }
+      else {
+        // key is concatenation: <cellId> + '_' + <attribute>
+        const splitPoint = key.lastIndexOf('_');
+        const attribute = key.substring(splitPoint+1);
+        const cellId = key.substring(0, splitPoint-2);
+        const isVertex = key[splitPoint-1] === 'V';
+        const isEdge = key[splitPoint-1] === 'E';
+
+        const cell = getCell(cellId) || createCell(cellId, isVertex, isEdge);
+
+        const attrHandlers = {
+          geometry: () => {
+            const serializedGeometry = value;
+            const parsed = xmlParser.parseFromString(serializedGeometry, 'text/xml');
+            const geometry = codec.decode(parsed.documentElement);
+            model.setGeometry(cell, geometry);
+          },
+          style: () => {
+            const style = value;
+            model.setStyle(cell, style);
+          },
+          parent: () => {
+            const cellId = value;
+            const parent = model.cells[cellId];
+            // The following appears to create a mxChildChange object, indicating that it is this the correct way to set the parent...
+            model.add(parent, cell, null);
+          },
+          value: () => {
+            let v;
+            if (typeof value === 'object') {
+              const {xmlEncoded} = value;
+              v = xmlParser.parseFromString(xmlEncoded, 'text/xml');
+            } else {
+              v = value;
+            }
+            model.setValue(cell, v);
+          },
+          source: () => {
+            const sourceId = value;
+            if (sourceId === null) {
+              model.setTerminal(cell, null, true);
+            } else {
+              const source = getCell(sourceId);
+              if (source === undefined) {
+                throw new Error("NO SUCH CELL:", cellId);
+              }
+              model.setTerminal(cell, source, true);
+            }
+          },
+          target: () => {
+            const targetId = value;
+            if (targetId === null) {
+              model.setTerminal(cell, null, false);
+            } else {
+              const target = getCell(targetId);
+              if (target === undefined) {
+                throw new Error("NO SUCH CELL:", cellId);
+              }
+              model.setTerminal(cell, target, false);
+            }
+          },
+          collapsed: () => {
+            const collapsed = value;
+            model.setCollapsed(cell, collapsed);
+          },
+          visible: () => {
+            const visible = value;
+            model.setVisible(cell, visible);
+          },
+          vertex: () => {
+            const vertex = value;
+            cell.setVertex(vertex);
+          },
+          edge: () => {
+            const edge = value;
+            cell.setEdge(edge);
+          },
+        };
+        attrHandlers[attribute]();
+      }
+    } finally {
+      // re-enable:
+      listenForEdits = true;
+    }
+  };
+
+  const resolve = (a, b) => {
+    console.log("CONFLICT between", a.id, "and", b.id);
+    // Cheap & fun way to define a total ordering on all elements:
+    return a.id > b.id;
+  };
+
+  let history, mergePromise;
+  function resetHistory() {
+    history = new History(context, setState, resolve);
+    mergePromise = Promise.resolve();
+  }
+  resetHistory();
+
+  function queuedMerge(serializedOp) {
+    mergePromise = mergePromise.then(() => {
+      return history.context.receiveOperation(serializedOp).then(op => {
+        history.autoMerge(op);
+        console.log("Merged ", op.id);
+      });
+    });
+  }
+  function leaveOnError() {
+    mergePromise.catch(err => {
+      console.log("Unexpected error merging", err);
+      controller.addInput("leave", "in", []);
+    });
+  }
+
+  const controller = new client.Controller(uiState, new JsEventLoop());
+
+  controller.addMyOwnOutputListener({
+    'add': event => {
+      console.log("output event:", event);
+      if (event.name === "ack_join") {
+        const [ops] = event.parameters;
+        console.log("Outevent ack_join", ops.length, "ops..");
+        try {
+          listenForEdits = false;
+          model.clear();
+          resetHistory();
+          for (const op of ops) {
+            queuedMerge(op);
+          }
+          leaveOnError();
+        } finally {
+          listenForEdits = true;
+        }
+      }
+      else if (event.name === "ack_new_share") {
+        // nothing needs to happen
+        const [sessionId] = event.parameters;
+      }
+      else if (event.name === "received_op") {
+        const [op] = event.parameters;
+        queuedMerge(op);
+        leaveOnError();
+      }
+      else if (event.name === "left") {
+
+      }
+    }
+  });
+
+  
+  {
+    // Upon loading the page, if a 'sessionId' URL parameter is given, automatically join the given session once the connection with the server is established.
+    const searchParams = new URLSearchParams(window.location.search);
+    const sessionId = searchParams.get('sessionId');
+    if (sessionId) {
+      controller.addInput("init_join", "in", [sessionId]);
+    } else {
+      controller.addInput("init_offline", "in", []);
+    }
+  }
+  controller.addInput("connect", "in", ["ws://localhost:8700/websocket"]);
+
+
+  controller.start();
+
+  function keyPrefix(cell) {
+    return cell.id + '_' + (cell.isVertex() ? 'V' : cell.isEdge() ? 'E' : 'X') + '_';
+  }
+
+  function encodeCompleteCellState(cell, deltaObj) {
+    const prefix = keyPrefix(cell);
+
+    if (cell.geometry) {
+      deltaObj[prefix + 'geometry'] = xmlSerializer.serializeToString(
+        codec.encode(
+          cell.geometry));
+    }
+    if (cell.style) {
+      deltaObj[prefix + 'style'] = cell.style;
+    }
+    if (cell.parent) {
+      deltaObj[prefix + 'parent'] = cell.parent.id;
+    }
+    if (cell.value) {
+      // For some reason, 'value' can be an HTML string (when only a label is defined, or it can be a DOM object
+      if (cell.value.constructor === String) {
+        deltaObj[prefix + 'value'] = cell.value;
+      } else {
+        deltaObj[prefix + 'value'] = {
+          xmlEncoded: xmlSerializer.serializeToString(cell.value),
+        };
+      }
+    }
+    if (cell.source) {
+      deltaObj[prefix + 'source'] = cell.source.id;
+    }
+    if (cell.target) {
+      deltaObj[prefix + 'target'] = cell.target.id;
+    }
+    if (cell.collapsed) {
+      deltaObj[prefix + 'collapsed'] = cell.collapsed;
+    }
+    if (cell.visible !== undefined) {
+      deltaObj[prefix + 'visible'] = cell.visible;
+    }
+  }
+
+  const changeToDelta = (change, deltaObj) => {
+    if (change.constructor === mxRootChange) {
+
+      if (change.root === change.previous) {
+        // This doesn't seem to occur:
+        console.log("no root change")
+        return; // no change
+      }
+      deltaObj['root'] = change.root.id;
+
+      for (const cell of Object.values(model.cells)) {
+        encodeCompleteCellState(cell, deltaObj);
+      }
+    }
+    else {
+      if (change.constructor === mxGeometryChange) {
+        deltaObj[keyPrefix(change.cell) + 'geometry'] = change.geometry ?
+          xmlSerializer.serializeToString(
+            codec.encode(change.geometry)) : null;
+      }
+      else if (change.constructor === mxStyleChange) {
+        deltaObj[keyPrefix(change.cell) + 'style'] = change.style;
+      }
+      else if (change.constructor === mxChildChange) {
+        if (change.previous) {
+          // cell has a previous parent
+          console.log("previous parent");
+          deltaObj[keyPrefix(change.child) + 'parent'] = change.parent ? change.parent.id: null;
+        } else {
+          // no previous parent -> cell was created:
+          console.log("no previous parent");
+          encodeCompleteCellState(change.child, deltaObj);
+        }
+      }
+      else if (change.constructor === mxValueChange) {
+        if (change.value.constructor === String) {
+          deltaObj[keyPrefix(change.cell) + 'value'] = change.value;
+        } else {
+          deltaObj[keyPrefix(change.cell) + 'value'] = {
+            xmlEncoded: xmlSerializer.serializeToString(change.value),
+          };
+        }
+      }
+      else if (change.constructor === mxTerminalChange) {
+        deltaObj[keyPrefix(change.cell) + (change.source ? 'source' : 'target')] =
+          change.terminal ? change.terminal.id : null;
+      }
+      else if (change.constructor === mxCollapseChange) {
+        deltaObj[keyPrefix(change.cell) + 'collapsed'] = change.collapsed;
+      }
+      else if (change.constructor === mxVisibleChange) {
+        deltaObj[keyPrefix(change.cell) + 'visible'] = change.visible;
+      }
+      else if (change.constructor === mxCellAttributeChange) {
+        // Mentioned in mxGraph documentation, but what is this, and does it actually occur?
+        throw new Error("DEBUG: Don't know what to do with mxCellAttributeChange");
+      }
+    }
+  }
+
+  // Fired when a local change happens
+  let listenForEdits = true;
+  model.addListener(mxEvent.NOTIFY, function(sender, event) {
+    if (listenForEdits) {
+      console.log("NOTIFY:", event.properties.edit.changes);
+
+      const delta = {};
+      for (const change of event.properties.edit.changes) {
+        changeToDelta(change, delta);
+      }
+      if (Object.keys(delta).length > 0) {
+        const op = history.new(delta, 
+          false // do NOT update mxGraphModel with the change (the change was already executed)
+        );
+        const serializedOp = op.serialize();
+        console.log("OP:", serializedOp);
+        controller.addInput("new_edit", "in", [serializedOp]);
+      }
+    }
+  });
+
+  // UI stuff
+  uiState.install(ui);
+
+  document.addEventListener('keydown', e => {
+    console.log(e);
+    if (e.code === 'KeyC') {
+      console.log("KeyC pressed: Clear model and reset history");
+      resetHistory();
+      model.clear();
+    }
+
+    if (e.code === 'KeyH') {
+      const seq = history.getOpsSequence();
+      console.log(seq.map(op => op.serialize()));
+    }
+  })
+});