Merge pull request #7 from gregperk/postgres

add postgres as another backing store

Author: gregperk

.gitignore

@@ -7,3 +7,4 @@ dist
 *.swp
 .cscope_db
 npm-debug.log
+.vscode

CHANGELOG.md

@@ -1,5 +1,15 @@
 # Change Log
 
+## [4.0.0]
+### Changed
+- Added Postgres as a supported backing store 
+  (doing so required adding an async `initialize()` to the storage interface 
+  and making the interface's `shutdown()` routine async). 
+- Note: **THIS IS A BREAKING CHANGE** because `storage.initialize()` now
+  needs to be called to ensure your backing store is up and running
+  (well, if your code needs to work properly across all possible backing stores).
+
+
 ## [3.4.0]
 ### Changed
 - AtlasClient.authenticate[...] classmethods now return an instance.
@@ -22,6 +32,7 @@
 
 
 [unreleased]: https://github.com/ForstaLabs/librelay-node/tree/master
+[4.0.0]: https://github.com/ForstaLabs/librelay-node/tree/v4.0.0
 [3.4.0]: https://github.com/ForstaLabs/librelay-node/tree/v3.4.0
 [3.3.0]: https://github.com/ForstaLabs/librelay-node/tree/v3.3.0
 [3.2.0]: https://github.com/ForstaLabs/librelay-node/tree/v3.2.0

README.md

@@ -36,6 +36,10 @@ librelay on a single backing store use
 `librelay.storage.setLabel('<something-unique>')` to shard your storage into
 a unique namespace.
 
+You'll need to ensure your backing store is running properly with a call 
+to (async) `librelay.storage.initialize()`, and if possible you should 
+tear it down before quitting, with (async) `librelay.storage.shutdown()`.
+
 
 Provisioning
 -------

package.json

@@ -1,6 +1,6 @@
 {
   "name": "librelay",
-  "version": "3.4.4",
+  "version": "3.5.0",
   "description": "Signal based end-to-end crypto for Forsta messaging platform",
   "repository": "ForstaLabs/librelay-node",
   "engines": {
@@ -20,6 +20,7 @@
     "libsignal": "1.2.1",
     "long": "3.2.0",
     "node-fetch": "1.7.3",
+    "pg": "^7.4.1",
     "protobufjs": "6.8.0",
     "redis": "2.8.0",
     "utf-8-validate": "3.0.3",

src/storage/backing/fs.js

@@ -140,9 +140,6 @@ class FSBacking extends StorageInterface {
         }
         return regex ? keys.filter(x => x.match(regex)) : keys;
     }
-
-    shutdown() {
-    }
 }
 
 module.exports = FSBacking;

src/storage/backing/index.js

@@ -1,5 +1,6 @@
 module.exports = {
     RedisBacking: require('./redis'),
     FSBacking: require('./fs'),
+    PostgresBacking: require('./postgres'),
     BackingInterface: require('./interface')
 };

src/storage/backing/interface.js

@@ -5,6 +5,9 @@ class StorageInterface {
         this.label = label;
     }
 
+    async initialize() {
+    }
+
     async set(ns, key, value) {
         throw new Error("Not Implemented");
     }

src/storage/backing/postgres.js

@@ -0,0 +1,73 @@
+const StorageInterface = require('./interface');
+const { Client } = require('pg');
+
+class PostgresBacking extends StorageInterface {
+
+    constructor(label) {
+        super(label);
+        this.tableName = 'faux_redis_' + this.label.toLowerCase().replace(/[^a-z0-9_]/g, '_');
+        this.client = new Client({ connectionString: process.env.DATABASE_URL });
+        this.queryCreateTableIfNeeded = `
+            CREATE TABLE IF NOT EXISTS ${this.tableName} (
+                namespace TEXT,
+                key TEXT,
+                value TEXT,
+                PRIMARY KEY (namespace, key)
+            );`;
+
+        this.querySetValue = `
+            INSERT INTO ${this.tableName} (namespace, key, value)
+                VALUES ($1::text, $2::text, $3::text)
+                ON CONFLICT (namespace, key)
+                    DO UPDATE SET value=$3::text`;
+
+        this.queryGetValue = `
+            SELECT value FROM ${this.tableName} WHERE namespace=$1::text AND key=$2::text`;
+
+        this.queryRemoveValue = `
+            DELETE FROM ${this.tableName} WHERE namespace=$1::text AND key=$2::text`;
+
+        this.queryGetKeys = `
+            SELECT key FROM ${this.tableName} WHERE namespace=$1::text`;
+    }
+
+    async initialize() {
+        await this.client.connect();
+        return this.client.query(this.queryCreateTableIfNeeded);
+    }
+
+    async set(ns, key, value) {
+        if (value === undefined) throw new Error("Tried to store undefined");
+        const result = await this.client.query(this.querySetValue, [ns, key, value]);
+        if (result.rowCount !== 1) throw new Error('Failure in postgres set');
+    }
+
+    async get(ns, key) {
+        const result = await this.client.query(this.queryGetValue, [ns, key]);
+        if (result.rowCount !== 1) throw new ReferenceError(key);
+        return result.rows[0].value;
+    }
+
+    async has(ns, key) {
+        const result = await this.client.query(this.queryGetValue, [ns, key]);
+        return result.rowCount === 1;
+    }
+
+    async remove(ns, key) {
+        const result = await this.client.query(this.queryRemoveValue, [ns, key]);
+        return result.rowCount === 1;
+    }
+
+    async keys(ns, regex) {
+        const result = await this.client.query(this.queryGetKeys, [ns]);
+        const keys = result.rows.map(r => r.key);
+        return regex ? keys.filter(x => x.match(regex)) : keys;
+    }
+
+    async shutdown() {
+        await this.client.end();
+        this.client = null;
+    }
+}
+
+module.exports = PostgresBacking;

src/storage/backing/redis.js

@@ -86,7 +86,7 @@ class RedisBacking extends StorageInterface {
         return regex ? keys.filter(x => x.match(regex)) : keys;
     }
 
-    shutdown() {
+    async shutdown() {
         this.client.quit();
         this.client = null;
     }

src/storage/index.js

@@ -65,7 +65,9 @@ exports.get = async (ns, key, defaultValue) => {
     }
     return data && decode(data);
 };
+exports.initialize = () => _backing.initialize();
 exports.set = (ns, key, value) => _backing.set(ns, key, encode(value));
+exports.has = (ns, key, value) => _backing.has(ns, key);
 exports.remove = (ns, key) => _backing.remove(ns, key);
 exports.keys = (ns, re) => _backing.keys(ns, re);
 exports.shutdown = () => _backing.shutdown();
@@ -237,6 +239,7 @@ exports.getDeviceIds = async function(addr) {
 function getBackingClass(name) {
     return {
         redis: exports.backing.RedisBacking,
+        postgres: exports.backing.PostgresBacking,
         fs: exports.backing.FSBacking
     }[name];
 }