Feature/rethink migration

.gitignore

@@ -2,4 +2,5 @@
 node_modules
 env.sh
 stash-passport-demo
-data
+data
+out

.vscode/launch.json

@@ -0,0 +1,18 @@
+{
+    // Use IntelliSense to learn about possible attributes.
+    // Hover to view descriptions of existing attributes.
+    // For more information, visit: https://go.microsoft.com/fwlink/?linkid=830387
+    "version": "0.2.0",
+    "configurations": [
+        {
+            "type": "node",
+            "request": "attach",
+            "name": "Attach to Remote",
+            "address": localhost,
+            "port": 9229,
+            "protocol": "inspector",
+            //"localRoot": "${workspaceFolder}",
+            //"remoteRoot": "Absolute path to the remote directory containing the program"
+        }
+    ]
+}

bin/install

@@ -0,0 +1,6 @@
+#! /usr/bin/env bash
+
+. /home/vagrant/.nvm/nvm.sh
+
+DIR="$( cd "$( dirname "${BASH_SOURCE[0]}" )" && pwd )"
+eval "nvm exec 8.9.0 npm install --production --prefix ${DIR}/../ $@"

bin/loom → bin/loom.ts

@@ -1,12 +1,13 @@
 #! /usr/bin/env node
 'use strict';
-var loom = require('..');
 
-var path = require('path');
-var util = require('util');
-var Loader = require('yaml-config-loader');
-var yargs = require('yargs');
-var loader = new Loader({stopOnError: true});
+import * as loom from '../index';
+import * as path from 'path';
+import * as util from 'util';
+import * as yargs from 'yargs';
+import * as Loader from 'yaml-config-loader';
+
+const loader = new Loader({stopOnError: true});
 
 loader.on('error', function(error) {
   if (error.name === 'YAMLException') {
@@ -15,15 +16,15 @@ loader.on('error', function(error) {
   throw error;
 });
 
-var argv = yargs
+yargs
   .describe('port', 'The port to listen on.')
   .alias('port', 'p')
   .describe('config', 'A YAML config file or directory of yaml files to load, can be invoked multiple times and later files will override earlier.')
   .alias('config', 'c')
   .describe('help', 'Display this help message.')
   .alias('help', 'h');
 
-argv = yargs.argv;
+const argv = yargs.argv;
 
 if (argv.help) {
   yargs.showHelp();
@@ -44,7 +45,7 @@ loader.addMapping({
   storageCompress: 'storage.compress',
 });
 
-var configKeys = [
+const configKeys = [
   'server',
   'db',
   'storage',
@@ -53,25 +54,26 @@ var configKeys = [
 
 // When you use remapping, later mappings tend to replace the entire
 // structure rather than overwriting the component key.
-var loaderAddOptions = {
-  deepMerge: configKeys,
+const loaderAddOptions = {
+  deepMerge: configKeys
 };
 
-loader.add(path.resolve(path.join(__dirname, '..', 'defaults.yaml')), loaderAddOptions);
+loader.add(path.resolve(path.join(process.env.PWD, 'defaults.yaml')), loaderAddOptions);
 loader.addAndNormalizeObject(process.env, loaderAddOptions);
 
 if (argv.config) {
   if (typeof argv.config === 'string') {
     argv.config = [argv.config];
   }
+
   for (let filePath of argv.config) {
     loader.add(path.resolve(filePath), loaderAddOptions);
   }
 }
 
 loader.addAndNormalizeObject(argv, loaderAddOptions);
 
-var cachedConfig = null;
+let cachedConfig = null;
 module.exports = {
   load: function(cb) {
     if (cachedConfig) {

bin/mergeMigrations

@@ -0,0 +1,61 @@
+#! /usr/bin/env node
+'use strict';
+
+const util = require('util');
+const fs = require('fs');
+const path = require('path');
+const Loader = require('yaml-config-loader');
+const yargs = require('yargs');
+const loader = new Loader();
+
+const argv = yargs
+  .describe('config', 'A YAML config file or directory of yaml files to load, can be invoked multiple times and later files will override earlier.')
+  .alias('config', 'c')
+  .alias('knexfile', 'k')
+  .argv;
+
+loader.on('error', function(error) {
+  if (error.name === 'YAMLException') {
+    console.error(util.print('Error parsing YAML file `', error.filePath, '`:', error.reason));
+    console.error(error);
+  }
+});
+
+loader.add(path.resolve(path.join(__dirname, '..', 'defaults.yaml')));
+
+if (argv.config) {
+  loader.add(path.resolve(argv.config));
+}
+
+loader.addAndNormalizeObject({});
+
+loader.load(function(error, config) {
+  let baseMigrationDir = path.resolve(path.join(__dirname, '..', 'baseMigrations'));
+  let migrationDir = path.resolve(path.join(__dirname, '..', 'migrations'));
+  let pluginMigrationDir = config.pluginDirectory !== null ? path.resolve(path.join(config.pluginDirectory, 'migrations')) : null;
+
+  fs.readdir(baseMigrationDir, function(error, files) {
+    if (error) {
+      console.error(error);
+      process.exit(1);
+    }
+    files.forEach(function(file) {
+      fs.createReadStream(path.join(baseMigrationDir, file))
+        .pipe(fs.createWriteStream(path.join(migrationDir, file)));
+
+    });
+  });
+
+  if (pluginMigrationDir) {
+    fs.readdir(pluginMigrationDir, function(error, files) {
+      if (error) {
+        console.error(error);
+        process.exit(1);
+      }
+      files.forEach(function(file) {
+        fs.createReadStream(path.join(pluginMigrationDir, file))
+          .pipe(fs.createWriteStream(path.join(migrationDir, file)));    });
+    });
+  }
+});
+

bin/migrate

@@ -0,0 +1,68 @@
+#! /bin/bash
+
+MY_PATH="`dirname \"$0\"`"              # relative
+MY_PATH="`( cd \"$MY_PATH\" && pwd )`"  # absolutized and normalized
+
+while test $# -gt 0; do
+  case "$1" in
+    -h|--help)
+      echo "Handles merging migrations from plugins and running them."
+      echo " "
+      echo "options:"
+      echo "-h, --help      show help"
+      echo "-c,             specify the full path to config.yaml to use."
+      echo "-k,             specify the full path to the knexfile to use."
+      echo "-e, --env       specify the enviroment from the knexfile to use."
+      exit 0
+      ;;
+    -c)
+      shift
+      if test $# -gt 0; then
+        config=$1
+      else
+	echo "no config specified"
+	exit 1
+      fi
+      shift
+      ;;
+    -k)
+      shift
+      if test $# -gt 0; then
+        knexfile=$1
+      else
+	echo "no knexfile specified"
+	exit 1
+      fi
+      shift
+      ;;
+    -e|--env)
+      shift
+      if test $# -gt 0; then
+        env=$1
+      else
+  echo "no enviroment specified"
+  exit 1
+      fi
+      shift
+      ;;
+    *)
+      break
+      ;;
+  esac
+done
+
+if [[ -z $env ]]; then
+  enviroment="production"
+else
+  echo "Set default enviroment as : $env"
+  enviroment=$env
+fi
+
+if [[ -z $knexfile ]]; then
+  node_modules/.bin/knex-migrate up --env $enviroment
+else
+  cp $knexfile $MY_PATH/../alt-knex.js
+  node_modules/.bin/knex-migrate up --env $enviroment --knexfile $MY_PATH/../alt-knex.js
+  rm $MY_PATH/../alt-knex.js
+fi
+

bin/start

@@ -0,0 +1,6 @@
+#! /usr/bin/env bash
+
+. /home/vagrant/.nvm/nvm.sh
+
+DIR="$( cd "$( dirname "${BASH_SOURCE[0]}" )" && pwd )"
+eval "nvm exec 8.9.0 npm run start --prefix ${DIR}/../ $@"

bin/stream.ts

@@ -0,0 +1,21 @@
+#!/usr/bin/env node
+'use strict';
+
+const url = process.argv[2];
+const streamid = process.argv[3];
+
+if (!url || !streamid) {
+  console.error(`Reads input from stdin and streams it to the server
+
+usage: ${process.argv[1]} url streamid [--force]`);
+  process.exit(1);
+}
+
+let client = require('../lib/client')({url: url});
+
+let stream = client.createWriteStream({}, {
+  id: streamid,
+  force: process.argv.indexOf('--force') > 1
+});
+
+process.stdin.pipe(stream);

migrations/0001-rethink-to-file-streams.js → dataMigration/0001-rethink-to-file-streams.js

@@ -1,7 +1,7 @@
 'use strict';
 
 // Run as:
-// node migrations/0001-rethink-to-file-streams -c loom.yaml | bunyan
+// node dataMigration/0001-rethink-to-file-streams -c loom.yaml | bunyan
 
 var loader = require('../bin/loom');
 var log = require('../lib/logger').getLogger('migration 0001');

dataMigration/0002-rethink-to-postgre.ts

@@ -0,0 +1,89 @@
+'use strict';
+
+// Run as:
+// node dataMigration/0001-rethink-to-file-streams -c loom.yaml | bunyan
+
+var loader = require("../bin/loom");
+import * as logger from "../lib/logger";
+let log = logger.getLogger('migration 0002');
+import { rethink } from "../lib/rethink";
+import { Database } from "../lib/knex";
+import * as co from 'co';
+import * as fs from 'fs';
+
+import { RethinkStorage } from "../lib/models/rethink_storage";
+import { PostgreStorage } from "../lib/models/postgre_storage";
+
+var listRethinkStreams = function(cb) {
+  var Meta = rethink.models.Meta;
+  Meta.run(function(err, streams) {
+    if (err) { return cb(err); }
+
+    cb(null, streams.map((s)=>s));
+  });
+};
+
+// returns a promise that resolves to true or false
+var metaRowExists = function(id, postgreStorage) {
+  return new Promise(function(accept, reject) {
+    postgreStorage.loadStream(id, (err) =>{
+      accept(err);
+    });
+  });
+};
+
+// returns a promise that resoslves when src if fully written to dest
+var writeStream = function(src, dest) {
+  return new Promise(function(accept, reject) {
+    src
+      .pipe(dest)
+      .on('finish', accept)
+      .on('error', reject);
+  });
+};
+
+
+loader.load(function(err, config) {
+  log.info(config, 'using config:');
+
+  try {
+    rethink.connect(config.db);
+    var rethinkStorage = new RethinkStorage(config.storage);
+    var postgreStorage = new PostgreStorage(config.storage);
+
+    listRethinkStreams(function(err, streamIds) {
+      if (err) {
+        return log.error(err);
+      }
+
+      co(function*() {
+        for (let meta of streamIds) {
+          var exists = yield metaRowExists(meta.id, postgreStorage);
+
+          if (exists) {
+           log.info(`exists, skipping:  ${meta.id}`); 
+           log.info(`${exists}`); 
+          }
+          else {
+            log.info(`writing stream to: ${meta.id}`);
+
+            let metaData = meta.meta.metaData;
+
+            postgreStorage.saveStream(meta.id,meta.meta);
+
+          }
+        }
+
+        setTimeout(()=> rethink.thinky.r.getPool().drain(), 1000);
+      }).catch(function(err) {
+        console.error(err.stack);
+      });
+    });
+  }
+  catch (e) {
+    console.error(e.stack);
+  }
+});
+
+
+