Merge fork

.gitignore

@@ -0,0 +1 @@
+node_modules

CHANGELOG.md

@@ -1,3 +1,20 @@
+## 4.0.1
+
+ - upgrade sharedb
+
+
+## 4.0.0
+
+ - upgrade `pg` from 7.4.1 to 8.5.1 to prevent silent failure for node v14 or later, reported in [this issue](https://github.com/brianc/node-postgres/issues/2317). 
+ - fix submit ops failed due to version mismatched - https://github.com/share/sharedb-postgres/issues/8
+
+## Note about re-versioning
+
+Original `sharedb-postgres` seems to have been not maintained for a long time since 2018. Thus we made a fork and maintain it as `@plotdb/sharedb-postgre`.
+
+
+# Change log in original repo
+
 ## 3.0.0
 
 Thanks to @billwashere, we upgraded to a more modern version of `pg` (4.5.1 ->

README.md

@@ -1,22 +1,68 @@
-# sharedb-postgres
+# @plotdb/sharedb-postgres
 
-PostgreSQL database adapter for [sharedb](https://github.com/share/sharedb). This
-driver can be used both as a snapshot store and oplog.
+PostgreSQL database adapter for [sharedb](https://github.com/share/sharedb). This driver can be used both as a snapshot store and oplog.
 
 Doesn't support queries (yet?).
 
 Moderately experimental. (This drives [Synaptograph](https://www.synaptograph.com)'s backend, and [@nornagon](https://github.com/nornagon) hasn't noticed any issues so far.)
 
 
+## Note about versioning
+
+This is a fork from the [original `sharedb-postgres`](https://github.com/share/sharedb-postgres) and its relative forks (see [billwashere](https://github.com/billwashere/sharedb-postgres-jsonb), [zbryikt](https://github.com/zbryikt/sharedb-postgres-jsonb). It seems to have been not maintained for a long time since 2018, Thus we decide to fork it and maintain it as `@plotdb/sharedb-postgre`.
+
+
+## Installation 
+
+```cmd
+npm i @plotdb/sharedb-postgres
+```
+
+
+## Requirements
+
+Due to the fix to resolve [high concurency issues](https://github.com/share/sharedb-postgres/issues/1) Postgres 9.5+ is now required.
+
+## Migrating older versions
+
+Older versions of this adaptor used the data type json. You will need to alter the data type prior to using if you are upgrading. 
+
+```PLpgSQL
+ALTER TABLE ops
+  ALTER COLUMN operation
+  SET DATA TYPE jsonb
+  USING operation::jsonb;
+
+ALTER TABLE snapshots
+  ALTER COLUMN data
+  SET DATA TYPE jsonb
+  USING data::jsonb;
+```
+
 ## Usage
 
-`sharedb-postgres` wraps native [node-postgres](https://github.com/brianc/node-postgres), and it supports the same configuration options.
+`sharedb-postgres-jsonb` wraps native [node-postgres](https://github.com/brianc/node-postgres), and it supports the same configuration options.
 
 To instantiate a sharedb-postgres wrapper, invoke the module and pass in your
-PostgreSQL configuration as an argument. For example:
+PostgreSQL configuration as an argument or use environmental arguments. 
+
+For example using environmental arugments:
+
+```js
+var db = require('@plotdb/sharedb-postgres')();
+var backend = require('sharedb')({db: db})
+```
+
+Then executing via the command line 
+
+```
+PGUSER=dbuser  PGPASSWORD=secretpassword PGHOST=database.server.com PGDATABASE=mydb PGPORT=5433 npm start
+```
+
+Example using an object
 
 ```js
-var db = require('sharedb-postgres')({host: 'localhost', database: 'mydb'});
+var db = require('@plotdb/sharedb-postgres')({host: 'localhost', database: 'mydb'});
 var backend = require('sharedb')({db: db})
 ```
 

index.js

@@ -9,6 +9,7 @@ function PostgresDB(options) {
 
   this.closed = false;
 
+  this.pg_config = options;
   this.pool = new pg.Pool(options);
 };
 module.exports = PostgresDB;
@@ -22,11 +23,6 @@ PostgresDB.prototype.close = function(callback) {
   if (callback) callback();
 };
 
-function rollback(client, done) {
-  client.query('ROLLBACK', function(err) {
-    return done(err);
-  })
-}
 
 // Persists an op and snapshot if it is for the next version. Calls back with
 // callback(err, succeeded)
@@ -41,84 +37,78 @@ PostgresDB.prototype.commit = function(collection, id, op, snapshot, options, ca
    * }
    * snapshot: PostgresSnapshot
    */
-  this.pool.connect(function(err, client, done) {
-    if (err) {
-      done(client);
-      callback(err);
-      return;
-    }
-    function commit() {
-      client.query('COMMIT', function(err) {
-        done(err);
-        if (err) {
-          callback(err);
-        } else {
-          callback(null, true);
-        }
-      })
+    this.pool.connect((err, client, done) => {
+      if (err) {
+        done(client);
+        callback(err);
+        return;
+      }
+    /*
+    * This query uses common table expression to upsert the snapshot table 
+    * (iff the new version is exactly 1 more than the latest table or if
+    * the document id does not exists)
+    *
+    * It will then insert into the ops table if it is exactly 1 more than the 
+    * latest table or it the first operation and iff the previous insert into
+    * the snapshot table is successful.
+    *
+    * This result of this query the version of the newly inserted operation
+    * If either the ops or the snapshot insert fails then 0 rows are returned
+    *
+    * If 0 zeros are return then the callback must return false   
+    *
+    * Casting is required as postgres thinks that collection and doc_id are
+    * not varchar  
+    */  
+    const query = {
+      name: 'sdb-commit-op-and-snap',
+      text: `WITH snapshot_id AS (
+  INSERT INTO snapshots (collection, doc_id, doc_type, version, data)
+  SELECT $1::varchar collection, $2::varchar doc_id, $4 doc_type, $3 v, $5 d
+  WHERE $3 = (
+    SELECT version+1 v
+    FROM snapshots
+    WHERE collection = $1 AND doc_id = $2
+    FOR UPDATE
+  ) OR NOT EXISTS (
+    SELECT 1
+    FROM snapshots
+    WHERE collection = $1 AND doc_id = $2
+    FOR UPDATE
+  )
+  ON CONFLICT (collection, doc_id) DO UPDATE SET version = $3, data = $5, doc_type = $4
+  RETURNING version
+)
+INSERT INTO ops (collection, doc_id, version, operation)
+SELECT $1::varchar collection, $2::varchar doc_id, $3 v, $6 operation
+WHERE (
+  $3 = (
+    SELECT max(version)+1
+    FROM ops
+    WHERE collection = $1 AND doc_id = $2
+  ) OR NOT EXISTS (
+    SELECT 1
+    FROM ops
+    WHERE collection = $1 AND doc_id = $2
+  )
+) AND EXISTS (SELECT 1 FROM snapshot_id)
+RETURNING version`,
+      values: [collection,id,snapshot.v, snapshot.type, snapshot.data,op]
     }
-    client.query(
-      'SELECT max(version) AS max_version FROM ops WHERE collection = $1 AND doc_id = $2',
-      [collection, id],
-      function(err, res) {
-        var max_version = res.rows[0].max_version;
-        if (max_version == null)
-          max_version = 0;
-        if (snapshot.v !== max_version + 1) {
-          return callback(null, false);
-        }
-        client.query('BEGIN', function(err) {
-          client.query(
-            'INSERT INTO ops (collection, doc_id, version, operation) VALUES ($1, $2, $3, $4)',
-            [collection, id, snapshot.v, op],
-            function(err, res) {
-              if (err) {
-                // TODO: if err is "constraint violation", callback(null, false) instead
-                rollback(client, done);
-                callback(err);
-                return;
-              }
-              if (snapshot.v === 1) {
-                client.query(
-                  'INSERT INTO snapshots (collection, doc_id, doc_type, version, data) VALUES ($1, $2, $3, $4, $5)',
-                  [collection, id, snapshot.type, snapshot.v, snapshot.data],
-                  function(err, res) {
-                    // TODO:
-                    // if the insert was successful and did insert, callback(null, true)
-                    // if the insert was successful and did not insert, callback(null, false)
-                    // if there was an error, rollback and callback(error)
-                    if (err) {
-                      rollback(client, done);
-                      callback(err);
-                      return;
-                    }
-                    commit();
-                  }
-                )
-              } else {
-                client.query(
-                  'UPDATE snapshots SET doc_type = $3, version = $4, data = $5 WHERE collection = $1 AND doc_id = $2 AND version = ($4 - 1)',
-                  [collection, id, snapshot.type, snapshot.v, snapshot.data],
-                  function(err, res) {
-                    // TODO:
-                    // if any rows were updated, success
-                    // if 0 rows were updated, rollback and not success
-                    // if error, rollback and not success
-                    if (err) {
-                      rollback(client, done);
-                      callback(err);
-                      return;
-                    }
-                    commit();
-                  }
-                )
-              }
-            }
-          )
-        })
+    client.query(query, (err, res) => {
+      if (err) {
+        callback(err)
+      } else if(res.rows.length === 0) {
+        done(client);
+        callback(null,false)
+      } 
+      else {
+        done(client);
+        callback(null,true)
       }
-    )
-  })
+    })
+
+    })
 };
 
 // Get the named document from the database. The callback is called with (err,
@@ -181,9 +171,12 @@ PostgresDB.prototype.getOps = function(collection, id, from, to, options, callba
       callback(err);
       return;
     }
-    client.query(
-      'SELECT version, operation FROM ops WHERE collection = $1 AND doc_id = $2 AND version >= $3 AND version < $4',
-      [collection, id, from, to],
+
+    var cmd = 'SELECT version, operation FROM ops WHERE collection = $1 AND doc_id = $2 AND version > $3 ';
+    var params = [collection, id, from];
+    if(to || to == 0) { cmd += ' AND version <= $4'; params.push(to)}
+    cmd += ' order by version';
+    client.query( cmd, params,
       function(err, res) {
         done();
         if (err) {