@cumulus/deployment/app/message.js

'use strict';

/**
 * Generate a Cumulus Message templates for all the workflows
 * in the stack and upload to s3
 *
 * @param {Object} config - Kes config object
 * @param {Array} outputs - an list of CloudFormation outputs
 * @param {function} uploader - an uploader function
 *
 * @returns {Promise} undefined
 */
let generateTemplates = (() => {
  var _ref = _asyncToGenerator(function* (config, outputs, uploader) {
    // this function only works if there are step functions defined in the deployment
    if (config.stepFunctions) {
      const bucket = config.system_bucket;
      const stack = config.stackName;
      const templates = Object.keys(config.stepFunctions).map(function (name) {
        return generateWorkflowTemplate(name, config.stepFunctions[name], config, outputs);
      });

      // uploads the generated templates to S3
      const workflows = [];
      console.log('Uploading Cumulus Message Templates for each Workflow ...');
      for (let ctr = 0; ctr < templates.length; ctr += 1) {
        const t = templates[ctr];
        const name = t.meta.workflow_name;
        const key = `${stack}/workflows/${name}.json`;
        yield uploader(bucket, key, JSON.stringify(t)); // eslint-disable-line no-await-in-loop
        workflows.push({
          name,
          template: `s3://${bucket}/${key}`,
          definition: config.stepFunctions[name]
        });
      }

      // generate list of workflows and upload it to S3
      // this is used by the /workflows endpoint of the API to return list
      // of existing workflows
      yield uploader(bucket, `${stack}/workflows/list.json`, JSON.stringify(workflows));

      // upload the buckets config
      yield uploader(bucket, `${stack}/workflows/buckets.json`, JSON.stringify(config.buckets));
    }
  });

  return function generateTemplates(_x, _x2, _x3) {
    return _ref.apply(this, arguments);
  };
})();

function _asyncToGenerator(fn) { return function () { var gen = fn.apply(this, arguments); return new Promise(function (resolve, reject) { function step(key, arg) { try { var info = gen[key](arg); var value = info.value; } catch (error) { reject(error); return; } if (info.done) { resolve(value); } else { return Promise.resolve(value).then(function (value) { step("next", value); }, function (err) { step("throw", err); }); } } return step("next"); }); }; }

const get = require('lodash.get');
const isObject = require('lodash.isobject');
const isString = require('lodash.isstring');
const omit = require('lodash.omit');

/**
 * Because both kes and message adapter use Mustache for templating,
 * we add curly brackets to items that are using the [$] and {$} syntax
 * to produce {{$}} and {[$]}
 *
 * @param {Object} cumulusConfig - the CumulusConfig portion of a task definition
 * @returns {Object} updated CumulusConfig
 */
function fixCumulusMessageSyntax(cumulusConfig) {
  if (!cumulusConfig) return {};

  const test = new RegExp('^([\\{]{1}|[\\[]{1})(\\$.*)([\\]]{1}|[\\}]{1})$');

  Object.keys(cumulusConfig).forEach(n => {
    if (isObject(cumulusConfig[n])) {
      // eslint-disable-next-line no-param-reassign
      cumulusConfig[n] = fixCumulusMessageSyntax(cumulusConfig[n]);
    } else if (isString(cumulusConfig[n])) {
      const match = cumulusConfig[n].match(test);
      if (match) {
        // eslint-disable-next-line no-param-reassign
        cumulusConfig[n] = `{${match[0]}}`;
      }
    }
  });

  return cumulusConfig;
}

/**
 * Extracts Cumulus Configuration from each Step Function Workflow
 * and returns it as a separate object
 *
 * @param {Object} config - Kes config object
 * @returns {Object} updated kes config object
 */
function extractCumulusConfigFromSF(config) {
  const workflowConfigs = {};

  // loop through the message adapter config of each step of
  // the step function, add curly brackets to values
  // with dollar sign and remove config key from the
  // definition, otherwise CloudFormation will be mad
  // at us.
  Object.keys(config.stepFunctions).forEach(name => {
    const sf = config.stepFunctions[name];
    workflowConfigs[name] = {};
    Object.keys(sf.States).forEach(n => {
      workflowConfigs[name][n] = fixCumulusMessageSyntax(sf.States[n].CumulusConfig);
      sf.States[n] = omit(sf.States[n], ['CumulusConfig']);
    });
    // eslint-disable-next-line no-param-reassign
    config.stepFunctions[name] = sf;
  });

  // eslint-disable-next-line no-param-reassign
  config.workflowConfigs = workflowConfigs;
  return config;
}

/**
 * Returns the OutputValue of a CloudFormation Output
 *
 * @param {Object} outputs - list of CloudFormation Outputs
 * @param {string} key - the key to return the value of
 *
 * @returns {string} the output value
 */
function findOutputValue(outputs, key) {
  const output = outputs.find(o => o.OutputKey === key);
  if (output) return output.OutputValue;
  return undefined;
}

/**
 * Generates a Cumulus Message template for a Cumulus Workflow
 *
 * @param {string} name - name of the workflow
 * @param {Object} workflow - the Step Function workflow object
 * @param {Object} config - Kes config object
 * @param {Array} outputs - an list of CloudFormation outputs
 *
 * @returns {Object} a Cumulus Message template
 */
function generateWorkflowTemplate(name, workflow, config, outputs) {
  // get cmr password from outputs
  const cmrPassword = findOutputValue(outputs, 'EncryptedCmrPassword');
  const cmr = Object.assign({}, config.cmr, { password: cmrPassword });
  // get launchpad passphrase from outputs
  const launchpadPassphrase = findOutputValue(outputs, 'EncryptedLaunchpadPassphrase');
  const launchpad = Object.assign({}, config.launchpad, { passphrase: launchpadPassphrase });
  const bucket = get(config, 'system_bucket');

  // add the current workflows' state machine arn
  const stateMachineArn = findOutputValue(outputs, `${name}StateMachine`);

  // add queues
  const queues = {};
  const queueExecutionLimits = {};
  if (config.sqs) {
    const queueArns = outputs.filter(o => o.OutputKey.includes('SQSOutput'));

    queueArns.forEach(queue => {
      const queueName = queue.OutputKey.replace('SQSOutput', '');
      const queueUrl = queue.OutputValue;

      queues[queueName] = queueUrl;

      const maxExecutions = get(config.sqs, `${queueName}.maxExecutions`);
      if (maxExecutions) {
        queueExecutionLimits[queueName] = maxExecutions;
      }
    });
  }

  // add the cumulus message config of the current workflow
  const workflowConfig = {};
  const states = get(workflow, 'States', {});
  Object.keys(states).forEach(state => {
    workflowConfig[state] = config.workflowConfigs[name][state];
  });

  // add the s3 uri to all the workflow templates for the current stack
  const templatesUris = {};
  const stepFunctions = get(config, 'stepFunctions', {});
  Object.keys(stepFunctions).forEach(sf => {
    templatesUris[sf] = `s3://${bucket}/${config.stack}/workflows/${sf}.json`;
  });

  const template = {
    cumulus_meta: {
      message_source: 'sfn',
      system_bucket: bucket,
      state_machine: stateMachineArn,
      execution_name: null,
      workflow_start_time: null
    },
    meta: {
      workflow_name: name,
      workflow_tasks: {},
      stack: config.stackName,
      buckets: config.buckets,
      cmr,
      launchpad,
      distribution_endpoint: config.distribution_endpoint,
      collection: {},
      provider: {},
      templates: templatesUris,
      queues,
      queueExecutionLimits
    },
    workflow_config: workflowConfig,
    payload: {},
    exception: null
  };

  return template;
}

module.exports = {
  fixCumulusMessageSyntax,
  extractCumulusConfigFromSF,
  findOutputValue,
  generateWorkflowTemplate,
  generateTemplates
};