supporting indexing projects based on Id (migration task)

Author: Parth Shah

migrations/seedElasticsearchIndex.js

@@ -9,6 +9,20 @@ import RabbitMQService from '../src/services/rabbitmq';
 
 const logger = bunyan.createLogger({ name: 'init-es', level: config.get('logLevel') });
 
+
+/**
+ * Retrieve project ids from cli if provided
+ * @return {Array} list of projectIds
+ */
+function getProjectIds() {
+  let projectIdArg = _.find(process.argv, a => a.indexOf('projectIds') > -1);
+  if (projectIdArg) {
+    projectIdArg = projectIdArg.split('=');
+    return projectIdArg[1].split(',').map(i => parseInt(i, 10));
+  }
+  return [];
+}
+
 Promise.coroutine(function* wrapped() {
   try {
     const rabbit = new RabbitMQService(logger);
@@ -19,23 +33,29 @@ Promise.coroutine(function* wrapped() {
       config.get('pubsubQueueName'),
     );
 
+
+    const projectIds = getProjectIds();
+    const projectWhereClause = (projectIds.length > 0) ? { id: { $in: projectIds } } : { deletedAt: { $eq: null } };
     const projects = yield models.Project.findAll({
-      where: { deletedAt: { $eq: null } },
+      where: projectWhereClause,
       raw: true,
     });
     logger.info(`Retrieved #${projects.length} projects`);
 
+    const memberWhereClause = (projectIds.length > 0)
+      ? { projectId: { $in: projectIds } }
+      : { deletedAt: { $eq: null } };
     let members = yield models.ProjectMember.findAll({
       raw: true,
-      where: { deletedAt: { $eq: null } },
+      where: memberWhereClause,
     });
     logger.info(`Retrieved #${members.length} members`);
     members = _.groupBy(members, 'projectId');
 
     const promises = [];
     _.forEach(projects, (p) => {
       p.members = members[p.id];
-      logger.debug(`Processing #${p.id}`);
+      logger.debug(`Processing Project #${p.id}`);
       promises.push(rabbit.publish('project.initial', p, {}));
     });
     Promise.all(promises)

src/events/projects/index.js

@@ -35,17 +35,17 @@ const projectCreatedHandler = Promise.coroutine(function* (logger, msg, channel)
       return _.merge(single, _.pick(detail, 'handle', 'firstName', 'lastName', 'email'));
     });
     // add the record to the index
-    const result = yield eClient.create({
+    const result = yield eClient.index({
       index: ES_PROJECT_INDEX,
       type: ES_PROJECT_TYPE,
       id: data.id,
       body: data,
     });
-    logger.debug('project indexed successfully', result);
+    logger.debug(`project indexed successfully (projectId: ${data.id})`, result);
     channel.ack(msg);
     return undefined;
   } catch (error) {
-    logger.error('Error proecessing event', error);
+    logger.error(`Error proecessing event (projectId: ${data.id})`, error);
     channel.nack(msg, false, !msg.fields.redelivered);
     return undefined;
   }
@@ -77,11 +77,11 @@ const projectUpdatedHandler = Promise.coroutine(function* (logger, msg, channel)
         doc: merged,
       },
     });
-    logger.debug('project updated successfully in elasticsearh index');
+    logger.debug(`project updated successfully in elasticsearh index, (projectId: ${data.id})`);
     channel.ack(msg);
     return undefined;
   } catch (error) {
-    logger.error('failed to get project document', error);
+    logger.error(`failed to get project document, (projectId: ${data.id})`, error);
     channel.nack(msg, false, !msg.fields.redelivered);
     return undefined;
   }
@@ -102,17 +102,16 @@ const projectDeletedHandler = Promise.coroutine(function* (logger, msg, channel)
       type: ES_PROJECT_TYPE,
       id: data.id,
     });
-    logger.debug('project deleted successfully from elasticsearh index');
+    logger.debug(`project deleted successfully from elasticsearh index (projectId: ${data.id})`);
     channel.ack(msg);
     return undefined;
   } catch (error) {
-    logger.error('failed to delete project document', error);
+    logger.error(`failed to delete project document (projectId: ${data.id})`, error);
     channel.nack(msg, false, !msg.fields.redelivered);
     return undefined;
   }
 });
 
-
 module.exports = {
   projectCreatedHandler,
   projectUpdatedHandler,