abacus-request/src/index.js

'use strict';

// Simple wrapper around the popular Node request module, adding some logging
// and Express-like route URI templates

const { extend, filter, find, flatten, groupBy, map, omit, pick, sortBy } = require('underscore');

const rrequest = require('request');
const lock = require('abacus-lock').locker('request');
const lru = require('abacus-lrucache');
const moment = require('abacus-moment');
const transform = require('abacus-transform');

const url = require('url');

/* jshint undef: false */
/* jshint unused: false */

// Setup debug log
const debug = require('abacus-debug')('abacus-request');
const edebug = require('abacus-debug')('e-abacus-request');

// Return the list of parameters found in a URI template
const params = (template) => {
  return map(template.match(/:[a-z_][a-z0-9_]*/gi), (k) => k.substr(1));
};

// Generates a URI from an Express-like URI template, use like this:
// request.route('http://.../:x/:y', { x: 1, y: 2 });
// returns http://.../1/2
const route = (template, parms) => {
  return template.replace(/:[a-z_][a-z0-9_]*/gi, (name) => {
    const k = name.substr(1);
    return parms[k] === undefined ? name : parms[k];
  });
};

// Setup request default options
const drequest = rrequest.defaults({
  json: true,
  rejectUnauthorized: !process.env.SKIP_SSL_VALIDATION,
  forever: false,
  pool: {
    maxSockets: parseInt(process.env.MAX_SOCKETS) || 1000
  }
});

// Convert request params to a request target configuration, uri, options and
// cb are all optional, but uri is expected to be given either as the uri
// parameter or as a field in the options parameter
const target = (m, uri, opt, cb) => {
  const callback = typeof opt === 'function' && !cb ? opt : cb;

  // Compute the default options
  const options = (uri, opt) => {
    if (typeof opt === 'object') return extend({}, opt, { route: uri });

    if (typeof uri === 'string') return { route: uri };

    return extend({}, uri, { route: uri.route || uri.uri });
  };

  // Resolve the route URI
  const resolve = (opts) =>
    extend({}, opts, {
      uri: route(opts.route, opts)
    });

  // Determine the target method
  const method = (m, opts) => extend({}, opts, m ? { method: m } : {});

  // Return the request target configuration
  const opts = method(m, resolve(options(uri, opt)));

  return {
    // uri: opts.uri,
    uri: opts.baseUrl ? opts.baseUrl + opts.uri : opts.uri,
    options: opts,
    callback: callback
  };
};

// Return true if the given request target is cacheable
const cacheableTarget = (t) =>
  t.options.cache &&
  t.callback &&
  !(t.options.method !== undefined && t.options.method !== 'GET');

// Return true if the given request target and response are cacheable
const cacheable = (t, res) => {
  if (!cacheableTarget(t))
    return false;
  if (res && res.statusCode !== undefined && res.statusCode !== 200)
    return false;

  return res ? true : false;
};

// Maintain a LRU cache of REST resources
const resources = lru({
  max: 1000,
  maxAge: 1000 * 60 * 20
});

// Cache a response
const cache = (uri, res) => {
  resources.set(uri, pick(res, 'statusCode', 'cookies', 'headers', 'body'));
  debug('Cached %s resource %d %o', uri, res.statusCode || 200, res.body ? res.body : '');
  return res;
};

// Return a resource from the cache
const cached = (uri) => {
  const res = resources.get(uri);
  if (!res)
    debug('Didn\'t find resource %s in cache', uri);
  else
    debug('Found resource %s in cache %d %o', uri, res.statusCode || 200, res.body ? res.body : '');
  return res;
};

const extractMessageFromBody = (body, defaultMessage) => body && body.message || body && body.error || defaultMessage;

// Convert an HTTP result with an error status code to an exception
const httpexc = (res) => {
  const message = extractMessageFromBody(res.body, `HTTP response status code ${res.statusCode}`);
  const exception = extend(new Error(message), res.body);

  // Warning: mutating variable err as that's really the simplest
  // way to set its code property without having to get into the
  // mess of creating a subclass of Error
  exception.code = res.statusCode;
  exception.statusCode = res.statusCode;
  exception.headers = res.headers;

  return exception;
};

// Simple wrapper around the request module function, use like this:
// request('http://localhost/whatever/:x/:y', { x: 10, y: 20 }, (err, res) => {
//   do something with res, res is a Javascript object parsed from the
//   JSON response
// });
const createRequest = () => (uri, opts, cb) => {
  // Convert the input parameters to a request target configuration
  const t = target(undefined, uri, opts, cb);

  // Send the request
  /* eslint complexity: [1, 7] */
  const send = (t, scb) => {
    const requestMsg = `${t.options.method} ${t.uri}`;
    debug(`Sending ${requestMsg}, body %o`, t.options.body || '');

    return drequest(
      t.options,
      scb
        ? (err, res) => {
          // Call back with the response
          if (err) {
            debug(`Error on ${requestMsg}, body: %o, error: %o`, t.options.body || '', err);
            edebug(`Error on ${requestMsg}, error: %o`, err);
            return scb(err, undefined);
          }

          debug('Received %s response %d %o', t.options.method, res.statusCode, res.body || '');

          if (res.statusCode >= 500 && res.statusCode <= 599) return scb(httpexc(res), undefined);

          // Optionally cache REST resources
          if (cacheable(t, res)) cache(t.uri, res);

          return scb(undefined, res);
        }
        : undefined
    );
  };

  // Optionally look for a cached REST resource
  if (cacheableTarget(t))
    return lock(t.uri, (err, unlock) => {
      if (err) {
        t.callback(err);
        unlock();
        return;
      }
      const res = cached(t.uri);
      if (res) {
        t.callback(undefined, res);
        unlock();
        return;
      }

      // Send the request
      send(t, (err, res) => {
        t.callback(err, res);
        unlock();
      });
    });

  // If not cacheable then just sent the request
  return send(t, t.callback);
};

// Return a function that will send a request with the given HTTP method
const singleOp = (method, opt) => {
  return (uri, opts, cb) => {
    const t = target(method, uri, opts, cb);
    t.options = extend(t.options, opt);

    return createRequest()(t.options, t.callback);
  };
};

// Return a function that will send a batch of requests with the given HTTP
// method
const batchOp = (method, opt) => {
  // Batches are sent using an HTTP POST
  const post = singleOp('POST', opt);

  return (reqs, cb) => {
    debug('Sending a batch of %d %s requests', reqs.length, method);

    // Return each request with its index in the list and the corresponding
    // target request options
    const targets = map(reqs, (args, i, reqs) => {
      const cbargs = args.concat(() => undefined);
      return {
        i: i,
        target: target(method, cbargs[0], cbargs[1], cbargs[2])
      };
    });

    // Optionally lookup cached REST resources
    transform.map(
      targets,
      (t, i, targets, ccb) => {
        if (cacheableTarget(t.target)) {
          const res = cached(t.target.uri);
          if (res) return ccb(undefined, extend({}, t, { cached: [undefined, res] }));
        }
        return ccb(undefined, t);
      },
      (err, res) => {
        if (err) cb(err);

        // Collect the cached resources
        const rcached = map(filter(res, (t) => t.cached), (r) => ({ i: r.i, res: r.cached }));

        // Collect the non-cached resources
        const targets = filter(res, (t) => !t.cached);

        // Send the remaining requests, group the calls by target
        // protocol, auth, host and OAuth bearer access token
        const groups = map(
          groupBy(targets, (t) =>
            [
              url.resolve(t.target.uri, '/'),
              t.target.options.headers && t.target.options.headers['x-batch-id']
                ? t.target.options.headers['x-batch-id']
                : '',
              t.target.options.headers && t.target.options.headers.authorization
                ? t.target.options.headers.authorization
                : ''
            ].join('-')
          )
        );

        // Send a single HTTP request per group
        transform.map(
          groups,
          (group, g, groups, gcb) => {
            // Build the request body
            const greq = map(group, (t) =>
              extend({ uri: url.parse(t.target.uri).path }, pick(t.target.options, 'method', 'headers', 'json', 'body'))
            );

            // Use the group's OAuth bearer access token for POST request
            const o =
              greq[0].headers && greq[0].headers.authorization
                ? { headers: pick(greq[0].headers, 'authorization') }
                : {};

            const batchEndpoint = (opt && opt.batch_endpoint) || '/batch';
            debug('Using batch endpoint %s', batchEndpoint);

            // Send the POST request to the target's host /batch path
            post(url.resolve(group[0].target.uri, batchEndpoint), extend(o, { body: greq }), (err, bres) => {
              if (err) {
                gcb(
                  undefined,
                  map(group, (g) => {
                    return { i: g.i, res: [err, undefined] };
                  })
                );
                return;
              }

              // Return the list of results from the response body
              if (bres) {
                // Forward non-200 status codes from /batch to requests
                const httpres = () => {
                  edebug(
                    'Batch response from %s %d %o %o',
                    group[0].target.uri,
                    bres.statusCode,
                    pick(bres.headers, 'www-authenticate') || '',
                    bres.body || ''
                  );

                  gcb(
                    undefined,
                    map(group, (g) => ({
                      i: g.i,
                      res: [undefined, extend({}, bres)]
                    }))
                  );
                };

                // Handle non-200 status codes from /batch
                if (bres.statusCode !== 200) {
                  httpres();
                  return;
                }

                /* eslint complexity: [1, 7] */
                gcb(
                  undefined,
                  map(bres.body, (r, i) => {
                    if (r.statusCode >= 500 && r.statusCode <= 599)
                      return { i: group[i].i, res: [httpexc(r), undefined] };

                    return {
                      i: group[i].i,
                      res: [
                        undefined,
                        {
                          statusCode: r.statusCode || 200,
                          headers: extend(
                            r.headers || omit(bres.headers, ['content-type', 'content-length']) || {},
                            r.header && r.header.Location ? { location: r.header.Location } : {}
                          ),
                          body: r.body
                        }
                      ]
                    };
                  })
                );
              } else
                gcb(
                  undefined,
                  map(group, (g) => {
                    return { i: g.i, res: [undefined, undefined] };
                  })
                );
            });
          },
          (err, gres) => {
            if (err) {
              cb(err);
              return;
            }

            // Optionally cache GET responses
            const fgres = flatten(gres, true);
            transform.map(
              fgres,
              (r, i, fgres, rcb) => {
                const t = find(targets, (t) => t.i === r.i);

                if (cacheable(t.target, r.res[1])) cache(t.target.uri, r.res[1]);

                return rcb();
              },
              () => {
                // Return assembled list of results from the cached list and
                // the list of groups, sorted like the corresponding requests
                const res = map(sortBy(rcached.concat(fgres), (r) => r.i), (r) => r.res);

                cb(undefined, res);
              }
            );
          }
        );
      }
    );
  };
};

// Return a function that pings a URL, useful to wait for the availability of
// an application in test cases for example. Ping can be invoked repeatedly
// Jest waitsFor condition for example without flooding the target URL with
// requests, as it will only ping that URL every 250 msec.

// Warning: pings is a mutable variable, used to record ping times and the
// corresponding responses
let pings = [];

const maxSuccessfulPings = 5;
const counterBarrier = maxSuccessfulPings + 1;

const ping = (uri, opt, reqFn) => {
  const cb = function(err, val) {
    // Warning: mutating variable pings
    if (!err) {
      pings[t.uri].val = val;
      pings[t.uri].count = ((pings[t.uri].count || 0) + 1) % counterBarrier;
      return;
    }

    pings[t.uri].count = 0;
  };

  const t = target(undefined, uri, opt ? opt : cb, opt ? cb : undefined);

  // Warning: mutating variable pings
  pings[t.uri] = pings[t.uri] || { t: moment.now() };

  if (moment.now() - pings[t.uri].t >= 250) {
    // Warning: mutating variable pings
    pings[t.uri].t = moment.now();
    reqFn.options(t.uri, t.options, t.callback);
  }

  return pings[t.uri].count || 0;
};

// Ping the target URI every 250 msec and wait for it to become available
const waitFor = (uri, opt, time, reqFn, cb) => {
  const callback = typeof time === 'function' && !cb ? time : cb;
  const timeout = typeof time === 'function' ? 10000 : time;

  const t = target(undefined, uri, opt, callback);

  debug('Pinging %s', t.uri);
  const i = setInterval(() => {
    const n = reqFn.ping(t.uri, t.options);
    if (n !== 0) debug('%d successful pings', n);
    if (n === maxSuccessfulPings) {
      // Call back after max successful pings
      clearInterval(i);
      clearTimeout(to);
      t.callback(undefined, t.uri);
    }
  }, 250);

  // Time out (by default after 10 seconds)
  const to = setTimeout(() => {
    debug('Timed out after %d ms', timeout);
    clearInterval(i);
    t.callback(new Error('timeout'));
  }, timeout);
};

// Shorthand functions for the various HTTP methods
const extendRequest = (defaultOptions) => {
  const requestFn = extend(createRequest(), {
    get: singleOp('GET', defaultOptions),
    head: singleOp('HEAD', defaultOptions),
    patch: singleOp('PATCH', defaultOptions),
    options: singleOp('OPTIONS', defaultOptions),
    post: singleOp('POST', defaultOptions),
    put: singleOp('PUT', defaultOptions),
    del: singleOp('DELETE', defaultOptions),
    delete: singleOp('DELETE', defaultOptions),

    // Batch versions of the HTTP methods, for use with the batch module
    batch_get: batchOp('GET', defaultOptions),
    batch_head: batchOp('HEAD', defaultOptions),
    batch_patch: batchOp('PATCH', defaultOptions),
    batch_post: batchOp('POST', defaultOptions),
    batch_put: batchOp('PUT', defaultOptions),
    batch_del: batchOp('DELETE', defaultOptions),
    batch_delete: batchOp('DELETE', defaultOptions),

    // Add helpers
    params: params,
    route: route
  });

  // Add ping & wait functions
  return extend(requestFn, {
    ping: (uri, opt) => ping(uri, opt, requestFn),
    waitFor: (uri, opt, time, cb) => waitFor(uri, opt, time, requestFn, cb)
  });
};

const noDefaults = extendRequest();
const defaults = (options) => extendRequest(options);

// Export our public functions
module.exports = noDefaults;
module.exports.defaults = defaults;