Merge pull request #2 from ncthbrt/fixes/v4-compatibility

ncthbrt · web-flow · commit 5f8b8ba0d75f · 2017-12-04T13:46:05.000+02:00
Fixes/v4 compatibility
diff --git a/lib/index.js b/lib/index.js
@@ -1,17 +1,17 @@
-const { AbstractPersistenceEngine, PersistedEvent } = require('nact/lib/persistence');
-const PgRx = require('pg-reactive');
-const { Promise } = require('bluebird');
+const { AbstractPersistenceEngine, PersistedEvent, PersistedSnapshot } = require('nact/lib/persistence');
 require('rxjs');
+const pgp = require('pg-promise')();
+const Rx = require('rxjs');
 const { create } = require('./schema');
 const assert = require('assert');
 
 class PostgresPersistenceEngine extends AbstractPersistenceEngine {
   constructor (connectionString, { createIfNotExists = true, tablePrefix = '', ...settings } = {}) {
     super();
-    this.db = new PgRx(connectionString);
+    this.db = pgp(connectionString);
     this.tablePrefix = tablePrefix;
     if (createIfNotExists) {
-      this.db.query(create(settings.tablePrefix)).catch(console.error).subscribe();
+      this.db.none(create(settings.tablePrefix)).catch(console.error);
     }
   }
 
@@ -25,6 +25,15 @@ class PostgresPersistenceEngine extends AbstractPersistenceEngine {
     );
   }
 
+  static mapDbModelToSnapshotDomainModel (dbSnapshot) {
+    return new PersistedSnapshot(
+      dbSnapshot.data,
+      Number.parseInt(dbSnapshot.sequence_nr),
+      dbSnapshot.persistence_key,
+      Number.parseInt(dbSnapshot.created_at)
+    );
+  }
+
   events (persistenceKey, offset = 0, limit = null, tags) {
     assert(typeof (persistenceKey) === 'string');
     assert(Number.isInteger(offset));
@@ -39,34 +48,73 @@ class PostgresPersistenceEngine extends AbstractPersistenceEngine {
                     OFFSET $2
                     LIMIT $3
                   `;
-    const args = [ persistenceKey, offset, limit, tags ].filter(x => x !== undefined);
 
-    return this.db.query(query, args)
-                  .retry(5)
-                  .map(PostgresPersistenceEngine.mapDbModelToDomainModel);
+    const args = [persistenceKey, offset, limit, tags].filter(x => x !== undefined);
+
+    return Rx.Observable
+      .of([1])
+      // Perform query
+      .mergeMap((_) => this.db.any(query, args))
+      // Retry the query if it fails
+      .retry(5)
+      // Flatten array so that it is returned as a stream of events
+      .mergeMap(x => x)
+      .map(PostgresPersistenceEngine.mapDbModelToDomainModel);
   }
 
   persist (persistedEvent) {
-    return new Promise((resolve, reject) => {
-      const query = ` INSERT INTO ${this.tablePrefix}event_journal (          
+    const query = ` 
+      INSERT INTO ${this.tablePrefix}event_journal (          
+        persistence_key,
+        sequence_nr,    
+        created_at,   
+        data,          
+        tags          
+      ) VALUES ($1, $2, $3, $4, $5)
+      RETURNING ordering;
+    `;
+    return this.db.one(
+      query, [
+        persistedEvent.key,
+        persistedEvent.sequenceNumber,
+        persistedEvent.createdAt,
+        persistedEvent.data,
+        persistedEvent.tags
+      ]
+    );
+  }
+
+  latestSnapshot (persistenceKey) {
+    assert(typeof (persistenceKey) === 'string');
+
+    const query = ` SELECT * from ${this.tablePrefix}snapshot_store 
+    WHERE persistence_key = $1
+    AND is_deleted = false   
+    ORDER BY sequence_nr DESC    
+    LIMIT 1
+  `;
+
+    return this.db.oneOrNone(query, [persistenceKey]).then(PostgresPersistenceEngine.mapDbModelToSnapshotDomainModel);
+  }
+
+  takeSnapshot (persistedSnapshot) {
+    const query = ` INSERT INTO ${this.tablePrefix}snapshot_store (          
           persistence_key,
           sequence_nr,    
           created_at,   
-          data,          
-          tags          
+          data
         )
-        VALUES ($1, $2, $3, $4, $5)
+        VALUES ($1, $2, $3, $4)
         RETURNING ordering;
       `;
-      this.db.query(
-          query,
-        [ persistedEvent.key,
-          persistedEvent.sequenceNumber,
-          persistedEvent.createdAt,
-          persistedEvent.data,
-          persistedEvent.tags
-        ]).catch(e => { reject(e); return e; }).subscribe(resolve);
-    });
+    return this.db.one(
+        query, [
+          persistedSnapshot.key,
+          persistedSnapshot.sequenceNumber,
+          persistedSnapshot.createdAt,
+          persistedSnapshot.data
+        ]
+    );
   }
 }
 
diff --git a/lib/schema/index.js b/lib/schema/index.js
@@ -9,6 +9,15 @@ module.exports.create = (tablePrefix = '') => `
     tags TEXT ARRAY DEFAULT ARRAY[]::TEXT[],
     CONSTRAINT event_journal_uq UNIQUE (persistence_key, sequence_nr)
   );  
+  CREATE TABLE IF NOT EXISTS ${tablePrefix}snapshot_store (
+    ordering BIGSERIAL NOT NULL PRIMARY KEY,
+    persistence_key VARCHAR(255) NOT NULL,
+    sequence_nr BIGINT NOT NULL,    
+    created_at BIGINT NOT NULL,   
+    data JSONB NOT NULL,
+    is_deleted BOOLEAN NOT NULL DEFAULT FALSE    
+  );  
   `;
 
-module.exports.destroy = (tablePrefix = '') => `DROP TABLE IF EXISTS ${tablePrefix}event_journal CASCADE;`;
+module.exports.destroy = (tablePrefix = '') => `DROP TABLE IF EXISTS ${tablePrefix}event_journal CASCADE; 
+DROP TABLE IF EXISTS ${tablePrefix}snapshot_store CASCADE;`;
diff --git a/package.json b/package.json
@@ -1,6 +1,6 @@
 {
   "name": "nact-persistence-postgres",
-  "version": "3.1.5",
+  "version": "4.0.0",
   "description": "nact-persistence-postgres ⇒ nact + postresql = your services have never been so persistent",
   "main": "lib/index.js",
   "scripts": {
@@ -16,24 +16,24 @@
   },
   "keywords": [
     "PostgreSQL",
-    "persistence",    
-    "event sourcing", 
+    "persistence",
+    "event sourcing",
     "actors",
-    "actor-model",    
+    "actor-model",
     "akka",
     "akka.net",
     "non-blocking",
     "actor system",
-    "erlang"       
+    "erlang"
   ],
   "author": "Nick Cuthbert (https://github.com/ncthbrt)",
   "license": "Apache-2.0",
   "bugs": {
     "url": "http://github.com/ncthbrt/nact-persistence-postgres/issues"
   },
   "dependencies": {
-    "nact": "^3.1.4",
-    "pg-reactive": "^0.3.2",
+    "nact": "^4.0.0",
+    "pg-promise": "^7.3.2",
     "rxjs": "^5.4.3"
   },
   "devDependencies": {
@@ -44,7 +44,6 @@
     "husky": "^0.14.3",
     "mocha": "^4.0.0",
     "nyc": "^11.2.1",
-    "pg-promise": "^7.0.0",
     "semistandard": "^11.0.0"
   }
 }
diff --git a/package.json.orig b/package.json.orig
@@ -0,0 +1,53 @@
+{
+  "name": "nact-persistence-postgres",
+<<<<<<< HEAD
+  "version": "4.0.0",
+=======
+  "version": "3.1.5",
+>>>>>>> origin/master
+  "description": "nact-persistence-postgres ⇒ nact + postresql = your services have never been so persistent",
+  "main": "lib/index.js",
+  "scripts": {
+    "test": "yarn run lint && nyc mocha ./test/*.js",
+    "interactive-cover": "nyc --reporter=html mocha ./test/*.js && open ./coverage/index.html",
+    "lint": "semistandard",
+    "coverage": "nyc report --reporter=text-lcov | coveralls",
+    "prepush": "yarn run test"
+  },
+  "repository": {
+    "type": "git",
+    "url": "git://github.com/ncthbrt/nact-persistence-postgres"
+  },
+  "keywords": [
+    "PostgreSQL",
+    "persistence",
+    "event sourcing",
+    "actors",
+    "actor-model",
+    "akka",
+    "akka.net",
+    "non-blocking",
+    "actor system",
+    "erlang"
+  ],
+  "author": "Nick Cuthbert (https://github.com/ncthbrt)",
+  "license": "Apache-2.0",
+  "bugs": {
+    "url": "http://github.com/ncthbrt/nact-persistence-postgres/issues"
+  },
+  "dependencies": {
+    "nact": "^4.0.0",
+    "pg-promise": "^7.3.2",
+    "rxjs": "^5.4.3"
+  },
+  "devDependencies": {
+    "chai": "^4.1.1",
+    "chai-as-promised": "^7.1.1",
+    "coveralls": "^3.0.0",
+    "eslint": "^4.8.0",
+    "husky": "^0.14.3",
+    "mocha": "^4.0.0",
+    "nyc": "^11.2.1",
+    "semistandard": "^11.0.0"
+  }
+}
diff --git a/scripts/teardown-db.sh b/scripts/teardown-db.sh
diff --git a/test/postgres-persistence.js b/test/postgres-persistence.js
@@ -7,7 +7,7 @@ chai.should();
 const { Promise } = require('bluebird');
 const delay = Promise.delay;
 const { PostgresPersistenceEngine } = require('../lib');
-const { PersistedEvent } = require('nact/lib/persistence');
+const { PersistedEvent, PersistedSnapshot } = require('nact/lib/persistence');
 const pgp = require('pg-promise')();
 const { destroy } = require('../lib/schema');
 
@@ -29,7 +29,10 @@ const connectionString = 'postgres://postgres:testpassword@localhost:5431/testdb
 describe('PostgresPersistenceEngine', function () {
   const db = pgp(connectionString);
 
-  afterEach(() => { db.query(destroy()); });
+  afterEach(() => {
+    db.query(destroy());
+  });
+
   it('should not create database if createIfNotExists is set to false', async function () {
     new PostgresPersistenceEngine(connectionString, { createIfNotExists: false });
     await delay(300);
@@ -41,7 +44,9 @@ describe('PostgresPersistenceEngine', function () {
   });
 
   describe('#persist', function () {
-    afterEach(() => { db.query(destroy()); });
+    afterEach(() => {
+      db.query(destroy());
+    });
 
     it('should store values in database', async function () {
       const engine = new PostgresPersistenceEngine(connectionString);
@@ -64,6 +69,56 @@ describe('PostgresPersistenceEngine', function () {
     });
   });
 
+  describe('#takeSnapshot', function () {
+    afterEach(() => {
+      db.query(destroy());
+    });
+
+    it('should store values in database', async function () {
+      const engine = new PostgresPersistenceEngine(connectionString);
+      await retry(async () => {
+        const snapshot1 = new PersistedSnapshot({ message: 'hello' }, 1, 'test');
+        const snapshot2 = new PersistedSnapshot({ message: 'goodbye' }, 2, 'test');
+        const snapshot3 = new PersistedSnapshot({ message: 'hello' }, 1, 'test2');
+        await engine.takeSnapshot(snapshot1);
+        await engine.takeSnapshot(snapshot2);
+        await engine.takeSnapshot(snapshot3);
+
+        const result =
+          (await db.many('SELECT * FROM snapshot_store WHERE persistence_key = \'test\' ORDER BY sequence_nr'))
+            .map(PostgresPersistenceEngine.mapDbModelToSnapshotDomainModel);
+
+        result.should.be.lengthOf(2).and.deep.equal([snapshot1, snapshot2]);
+        const result2 = await db.one('SELECT * FROM snapshot_store WHERE persistence_key = \'test2\'');
+        PostgresPersistenceEngine.mapDbModelToSnapshotDomainModel(result2).should.deep.equal(snapshot3);
+      }, 7, 50);
+    });
+  });
+
+  describe('#latestSnapshot', function () {
+    const snapshot1 = new PersistedSnapshot({ message: 'hello' }, 1, 'test3');
+    const snapshot2 = new PersistedSnapshot({ message: 'goodbye' }, 2, 'test3');
+    const snapshot3 = new PersistedSnapshot({ message: 'hello again' }, 3, 'test3');
+    let engine;
+
+    beforeEach(async () => {
+      engine = new PostgresPersistenceEngine(connectionString);
+      await retry(async () => {
+        await engine.takeSnapshot(snapshot1);
+        await engine.takeSnapshot(snapshot2);
+        await engine.takeSnapshot(snapshot3);
+      }, 7, 50);
+    });
+    afterEach(() => {
+      db.query(destroy());
+    });
+
+    it('should be able to retrieve latest snapshot', async function () {
+      const result = await engine.latestSnapshot('test3');
+      result.should.deep.equal(snapshot3);
+    });
+  });
+
   describe('#events', async function () {
     const event1 = new PersistedEvent({ message: 'hello' }, 1, 'test3', ['a', 'b', 'c']);
     const event2 = new PersistedEvent({ message: 'goodbye' }, 2, 'test3', ['a']);
@@ -78,7 +133,9 @@ describe('PostgresPersistenceEngine', function () {
         await engine.persist(event3);
       }, 7, 50);
     });
-    afterEach(() => { db.query(destroy()); });
+    afterEach(() => {
+      db.query(destroy());
+    });
 
     it('should be able to retrieve previously persisted events', async function () {
       const result = await new Promise((resolve, reject) => {
diff --git a/yarn.lock b/yarn.lock
