Merge pull request #77 from CatalystCode/create-replace-site

Added createSite, removeSite to settings and refactored executeQuery
CatalystCode · Jul 28, 2017 · 87b82fe · 87b82fe
2 parents 8b7ddce + d5e1270
commit 87b82fe
Show file tree

Hide file tree

Showing 9 changed files with 187 additions and 29 deletions.
diff --git a/.gitignore b/.gitignore
@@ -38,3 +38,6 @@ jspm_packages
 
 # IDEA config files
 .idea/
+
+# VS code config
+.vscode
diff --git a/.vscode/settings.json b/.vscode/settings.json
diff --git a/src/clients/cassandra/CassandraConnector.js b/src/clients/cassandra/CassandraConnector.js
@@ -7,6 +7,7 @@ const chunk = require('lodash/chunk');
 const flatten = require('lodash/flatten');
 const trackDependency = require('../appinsights/AppInsightsClient').trackDependency;
 
+const FETCH_SIZE = process.env.FETCH_SIZE || 1000;
 const MAX_OPERATIONS_PER_BATCH = process.env.MAX_OPERATIONS_PER_BATCH || 10;
 const MAX_CONCURRENT_BATCHES = process.env.MAX_CONCURRENT_BATCHES || 50;
 const distance = cassandra.types.distance;
@@ -20,6 +21,11 @@ const options = {
       [distance.local]: CORE_CONNECTIONS_PER_HOST_LOCAL,
       [distance.remote]: CORE_CONNECTIONS_PER_HOST_REMOTE
     } 
+  },
+  queryOptions: {
+    autoPage: true,
+    prepare: true,
+    fetchSize: FETCH_SIZE
   }
 };
 const client = new cassandra.Client(options);
@@ -49,29 +55,38 @@ function executeBatchMutations(mutations) {
       }
     },
     (err) => {
-      if (err) {
-        reject(err);
-      } else {
-        resolve();
-      }
+      if (err) reject(err);
+      else resolve({numBatchMutations: chunkedMutations.length});
     });
   });
 }
 
 /**
  * @param {string} query
  * @param {string[]} params
+ * @param {{fetchSize: int}} [options]
  * @returns {Promise.<object[]>}
  */
-function executeQuery(query, params) { // eslint-disable-line no-unused-vars
-  return new Promise((resolve, reject) => { // eslint-disable-line no-unused-vars
-    client.execute(query, params, { prepare: true }, (err, result) => {
-      if (err) {
-        return reject(err);
-      } else {
-        return resolve(result.rows);
-      }
-    });
+function executeQuery(query, params, options) {
+  return new Promise((resolve, reject) => {
+    try {
+      const rows = [];
+      const stream = client.stream(query, params, options)
+        .on('readable', () => {
+          let row;
+          while((row = stream.read()) != undefined) {
+            rows.push(row);
+          }
+        })
+        .on('error', err => {
+          if (err) reject(err);
+        })
+        .on('end', () => {
+          resolve(rows);
+        });
+    } catch (exception) {
+      reject(exception);
+    }
   });
 }
 

diff --git a/src/clients/storage/BlobStorageClient.js b/src/clients/storage/BlobStorageClient.js
@@ -2,6 +2,7 @@
 
 const Promise = require('promise');
 const request = require('request');
+const trackDependency = require('../appinsights/AppInsightsClient').trackDependency;
 
 /**
  * @param {string} uri
@@ -27,5 +28,5 @@ function fetchJson(uri) {
 }
 
 module.exports = {
-  fetchJson: fetchJson
+  fetchJson: trackDependency(fetchJson, 'BlobStorage', 'fetchJson')
 };
diff --git a/src/resolvers-cassandra/Settings/index.js b/src/resolvers-cassandra/Settings/index.js
@@ -4,7 +4,8 @@ const mutations = require('./mutations');
 const queries = require('./queries');
 
 module.exports = {
-  createOrReplaceSite: mutations.createOrReplaceSite,
+  createSite: mutations.createSite,
+  removeSite: mutations.removeSite,
   modifyFacebookPages: mutations.modifyFacebookPages,
   removeFacebookPages: mutations.removeFacebookPages,
   modifyTrustedTwitterAccounts: mutations.modifyTrustedTwitterAccounts,

diff --git a/src/resolvers-cassandra/Settings/mutations.js b/src/resolvers-cassandra/Settings/mutations.js
@@ -3,21 +3,141 @@
 const Promise = require('promise');
 const uuid = require('uuid/v4');
 const cassandraConnector = require('../../clients/cassandra/CassandraConnector');
+const blobStorageClient = require('../../clients/storage/BlobStorageClient');
 const withRunTime = require('../shared').withRunTime;
 const trackEvent = require('../../clients/appinsights/AppInsightsClient').trackEvent;
-
+const apiUrlBase = process.env.FORTIS_CENTRAL_ASSETS_HOST || 'https://fortiscentral.blob.core.windows.net';
 const STREAM_PIPELINE_TWITTER = 'twitter';
 const STREAM_CONNECTOR_TWITTER = 'Twitter';
 
 const TRUSTED_SOURCES_CONNECTOR_TWITTER = 'Twitter';
 const TRUSTED_SOURCES_CONNECTOR_FACEBOOK = 'FacebookPage';
 const TRUSTED_SOURCES_RANK_DEFAULT = 10;
 
+function createOrReplaceSite(args, res) { // eslint-disable-line no-unused-vars
+  return new Promise((resolve, reject) => { // eslint-disable-line no-unused-vars
+    return reject(
+      'This API call is no longer supported. ' +
+      'Its functionality has been separated into createSite and removeSite.'
+    );
+  });
+}
+
+function _insertTopics(siteType) {
+  return new Promise((resolve, reject) => {
+    if(!siteType || !siteType.length) {
+      return reject('insertTopics: siteType is not defined');
+    }
+    const uri = `${apiUrlBase}/settings/siteTypes/${siteType}/topics/defaultTopics.json`;
+    blobStorageClient.fetchJson(uri)
+      .then(response => {
+        const mutations = response.map(topic => {
+          return {
+            query: `INSERT INTO fortis.watchlist (topicid,topic,lang_code,translations,insertion_time) 
+                    VALUES (?, ?, ?, ?, toTimestamp(now()));`,
+            params: [uuid(), topic.topic, topic.lang_code, topic.translations]
+          };
+        });
+        return mutations;
+      })
+      .then(mutations => {
+        cassandraConnector.executeBatchMutations(mutations)
+        .then(() => {
+          resolve({
+            numTopicsInserted: mutations.length
+          });
+        })
+        .catch(reject);
+      })
+      .catch(reject);
+  });
+}
+
+const insertTopics = trackEvent(_insertTopics, 'Settings.Topics.Insert', (response, err) => ({numTopicsInserted: err ? 0 : response.numTopicsInserted}));
+
 /**
- * @param {{input: {targetBbox: number[], defaultZoomLevel: number, logo: string, title: string, name: string, defaultLocation: number[], storageConnectionString: string, featuresConnectionString: string, mapzenApiKey: string, fbToken: string, supportedLanguages: string[]}}} args
- * @returns {Promise.<{name: string, properties: {targetBBox: number[], defaultZoomLevel: number, logo: string, title: string, defaultLocation: number[], storageConnectionString: string, featuresConnectionString: string, mapzenApiKey: string, fbToken: string, supportedLanguages: string[]}}>}
+ * @param {{input: {siteType: string, targetBbox: number[], defaultZoomLevel: number, logo: string, title: string, name: string, defaultLocation: number[], storageConnectionString: string, featuresConnectionString: string, mapzenApiKey: string, fbToken: string, supportedLanguages: string[]}}} args
+ * @returns {Promise}
  */
-function createOrReplaceSite(args, res) { // eslint-disable-line no-unused-vars
+function createSite(args, res) { // eslint-disable-line no-unused-vars
+  return new Promise((resolve, reject) => {
+    const siteType = args && args.input && args.input.siteType;
+    if(!siteType || !siteType.length) {
+      return reject(`siteType for sitename ${args.input.name} is not defined`);
+    }
+
+    cassandraConnector.executeQuery('SELECT * FROM fortis.sitesettings WHERE sitename = ?;', [args.input.name])
+      .then(rows => {
+        if (!rows || !rows.length) {
+          insertTopics(siteType)
+            .then(() => {
+              return cassandraConnector.executeBatchMutations([{
+                query: `INSERT INTO fortis.sitesettings (
+                  geofence,
+                  defaultzoom,
+                  logo,
+                  title,
+                  sitename,
+                  languages,
+                  insertion_time
+                ) VALUES (?,?,?,?,?,?,toTimestamp(now()))`,
+                params: [
+                  args.input.targetBbox,
+                  args.input.defaultZoomLevel,
+                  args.input.logo,
+                  args.input.title,
+                  args.input.name,
+                  args.input.supportedLanguages
+                ]
+              }]);
+            })
+            .then(() => { 
+              resolve({
+                name: args.input.name,
+                properties: {
+                  targetBbox: args.input.targetBbox,
+                  defaultZoomLevel: args.input.defaultZoomLevel,
+                  logo: args.input.logo,
+                  title: args.input.title,
+                  defaultLocation: args.input.defaultLocation,
+                  supportedLanguages:args.input.supportedLanguages
+                }
+              });
+            })
+            .catch(reject);
+        } 
+        else if (rows.length == 1) return reject(`Site with sitename ${args.input.name} already exists.`);
+        else return reject(`(${rows.length}) number of sites with sitename ${args.input.name} already exist.`);
+      })
+      .catch(reject);
+  });
+}
+
+/**
+ * @param {{input: {siteType: string, targetBbox: number[], defaultZoomLevel: number, logo: string, title: string, name: string, defaultLocation: number[], storageConnectionString: string, featuresConnectionString: string, mapzenApiKey: string, fbToken: string, supportedLanguages: string[]}}} args
+ * @returns {Promise}
+ */
+function removeSite(args, res) { // eslint-disable-line no-unused-vars
+  return new Promise((resolve, reject) => {
+    cassandraConnector.executeBatchMutations([{
+      query: 'DELETE FROM fortis.sitesettings WHERE sitename = ?;',
+      params: [args.input.name]
+    }])
+    .then(() => { 
+      resolve({
+        name: args.input.name,
+        properties: {
+          targetBbox: args.input.targetBbox,
+          defaultZoomLevel: args.input.defaultZoomLevel,
+          logo: args.input.logo,
+          title: args.input.title,
+          defaultLocation: args.input.defaultLocation,
+          supportedLanguages: args.input.supportedLanguages
+        }
+      });
+    })
+    .catch(reject);
+  });
 }
 
 function facebookPagePrimaryKeyValuesToRowKey(values) {
@@ -296,7 +416,9 @@ function removeBlacklist(args, res) { // eslint-disable-line no-unused-vars
 }
 
 module.exports = {
-  createOrReplaceSite: trackEvent(createOrReplaceSite, 'createOrReplaceSite'),
+  createOrReplaceSite: createOrReplaceSite,
+  createSite: trackEvent(createSite, 'createSite'),
+  removeSite: trackEvent(removeSite, 'removeSite'),
   modifyFacebookPages: trackEvent(withRunTime(modifyFacebookPages), 'modifyFacebookPages'),
   removeFacebookPages: trackEvent(withRunTime(removeFacebookPages), 'removeFacebookPages'),
   modifyTrustedTwitterAccounts: trackEvent(withRunTime(modifyTrustedTwitterAccounts), 'modifyTrustedTwitterAccounts'),

diff --git a/src/resolvers-cassandra/Settings/queries.js b/src/resolvers-cassandra/Settings/queries.js
@@ -67,8 +67,8 @@ function twitterAccounts(args, res) { // eslint-disable-line no-unused-vars
   return new Promise((resolve, reject) => {
     const sourcesByConnector = 'SELECT params FROM fortis.streams WHERE connector = ? ALLOW FILTERING';
     cassandraConnector.executeQuery(sourcesByConnector, [CONNECTOR_TWITTER])
-    .then(rows => {
-      const accounts = rows.map(cassandraRowToTwitterAccount);
+    .then(result => {
+      const accounts = result.map(cassandraRowToTwitterAccount);
       resolve({accounts: accounts});
     })
     .catch(reject)

diff --git a/src/resolvers/Settings.js b/src/resolvers/Settings.js
@@ -20,6 +20,22 @@ function sites(args, res) { // eslint-disable-line no-unused-vars
   });
 }
 
+function createSite(args, res) { // eslint-disable-line no-unused-vars
+  return new Promise((resolve, reject) => { // eslint-disable-line no-unused-vars
+    return reject(
+      'This API call is supported in v2. '
+    );
+  });
+}
+
+function removeSite(args, res) { // eslint-disable-line no-unused-vars
+  return new Promise((resolve, reject) => { // eslint-disable-line no-unused-vars
+    return reject(
+      'This API call is supported in v2. '
+    );
+  });
+}
+
 function createOrReplaceSite(args, res) { // eslint-disable-line no-unused-vars
   const siteDefintion = args.input;
   return new Promise((resolve, reject) => {
@@ -268,6 +284,8 @@ function removeBlacklist(args, res) { // eslint-disable-line no-unused-vars
 
 module.exports = {
   sites: sites,
+  createSite: createSite,
+  removeSite: removeSite,
   createOrReplaceSite: createOrReplaceSite,
   modifyFacebookPages: modifyFacebookPages,
   modifyTrustedTwitterAccounts: modifyTrustedTwitterAccounts,

diff --git a/src/schemas/SettingsSchema.js b/src/schemas/SettingsSchema.js
@@ -11,6 +11,8 @@ module.exports = graphql.buildSchema(`
   }
 
   type Mutation {
+    createSite(input: SiteDefinition!): Site
+    removeSite(input: SiteDefinition!): Site
     removeFacebookPages(input: FacebookPageListInput!): FacebookPageCollection
     modifyFacebookPages(input: FacebookPageListInput!): FacebookPageCollection
     createOrReplaceSite(input: SiteDefinition!): Site
@@ -47,7 +49,7 @@ module.exports = graphql.buildSchema(`
 
   type TwitterAccountCollection {
     runTime: String,
-    accounts: [TwitterAccount],
+    accounts: [TwitterAccount]
   }
 
   type TrustedTwitterAccountCollection {
@@ -101,6 +103,7 @@ module.exports = graphql.buildSchema(`
   }
 
   input SiteDefinition {
+    siteType: String,
     targetBbox: [Float],
     defaultZoomLevel: Int,
     logo: String,