move files to backend subdir, push data in submodule

9 files changed, 544 insertions, 0 deletions

diff --git a/backend/api.js b/backend/api.js
new file mode 100644
index 0000000..30b4937
--- /dev/null
+++ b/backend/api.js
@@ -0,0 +1,42 @@
+#!/usr/bin/node
+
+const requestP = require('request-promise'),
+  config = require('./config'),
+  Future = require('fluture'),
+  FutureRetry = require('fluture-retry');
+
+const API_KEY = process.env.API_KEY || require('./secret').apikey;
+
+const requestF = Future.encaseP(requestP);
+
+function apiRequest(path, query) {
+  return FutureRetry.retryLinearly(requestF({
+    uri: config.api.baseUrl + path,
+    headers: {
+      'X-Title-Id': 'semc-vainglory',
+      'Authorization': API_KEY
+    },
+    qs: query,
+    json: true,
+    gzip: true,
+    forever: true,
+    timeout: config.api.timeout * 1000,
+    strictSSL: true
+  }));
+}
+
+function awsRequest(url) {
+  return FutureRetry.retryLinearly(requestF({
+    uri: url,
+    json: true,
+    gzip: true,
+    forever: true,
+    timeout: config.api.timeout * 1000,
+    strictSSL: true
+  }));
+}
+
+module.exports = {
+  apiRequest,
+  awsRequest,
+};
diff --git a/backend/config.js b/backend/config.js
new file mode 100644
index 0000000..fe51ca3
--- /dev/null
+++ b/backend/config.js
@@ -0,0 +1,59 @@
+#!/usr/bin/node
+
+const crypto = require('crypto');
+
+const sha256 = (x) => crypto.createHash('sha256').update(x, 'utf8').digest('hex');
+
+const hashSettings = (config) => sha256(
+  config.api.regions +
+  config.api.modes +
+  config.api.requestsPerInterval +
+  config.api.interval +
+  config.api.intervalUnit
+).substring(0, 8);
+
+module.exports = {
+  api: {
+    baseUrl: 'https://api.dc01.gamelockerapp.com',
+    timeout: 3,
+    /* available: eu, na, sg, cn, sa, ea */
+    regions: ['sg', 'eu', 'na'],
+    /* available: casual, 5v5_pvp_casual, ranked, 5v5_pvp_ranked, blitz_pvp_ranked, casual_aral */
+    modes: ['blitz_pvp_ranked', 'casual_aral'],
+    /* available: 2, 3, 4, 5 */
+    matchesPerRequest: 5,
+    /* How many equidistant requests to create
+     * matchesPerRequest * requestsPerInterval = sample size for that interval
+     * currently requestsPerInterval <= API key rate limit but should work if it isn't significantly higher
+     * rough estimate of upper limits (na/eu/sg, depends on region and time of day):
+     * blitz: 240
+     * battle royale: 30
+     * 3v3 casual: 30
+     * 3v3 ranked: 100
+     * 5v5 casual: 30
+     * 5v5 ranked: 40
+     * if filtering for multiple game modes, use the sum of the upper limits */
+    requestsPerInterval: 50,
+    /* the size of a batch */
+    interval: 1,
+    intervalUnit: 'hours',
+    /* How many batches to request per minute
+     * requestsPerInterval * intervalsPer Minute <= API key rate limit */
+    intervalsPerMinute: 1,
+  },
+  etl: {
+    /* attributes to be extracted and renamed from main API data */
+    participant: [ ['Winner', ['attributes', 'stats', 'winner']] ],
+    roster: [ ],
+    match: [ ['Mode', ['attributes', 'gameMode']] ],
+  },
+  reduce: {
+    /* criteria or 'filters' */
+    group: ['Actor', 'Talent', 'Mode'],
+    /* stats */
+    sum: ['Level', 'Winner'],
+  },
+  file: {
+    pattern: (moment) => `../data/${hashSettings(module.exports)}/${moment.toISOString()}.json`,
+  },
+};
diff --git a/backend/etl.js b/backend/etl.js
new file mode 100644
index 0000000..7334b7d
--- /dev/null
+++ b/backend/etl.js
@@ -0,0 +1,176 @@
+#!/usr/bin/node
+
+const R = require('ramda'),
+  Future = require('fluture'),
+  utils = require('./utils'),
+  api = require('./api');
+
+// API data cleanup
+const mapSideToTeam = (side) => side == 'left/blue' ? 'Left' : 'Right';
+
+function loadFPayloads(config) {
+  const objOfPathsConfig = {
+    match: R.map(R.apply(utils.objOfPath), config.match),
+    roster: R.map(R.apply(utils.objOfPath), config.roster),
+    participant: R.map(R.apply(utils.objOfPath), config.participant),
+  }; // TODO Ramda-ify
+
+  /*
+   * 1     Pull API JSON
+   * 2     Extract Participant, Roster, Match, Assets
+   * 3.1   Transform Participant, Roster, Match to flat structure
+   * 3.2   Pull and filter Assets for Events
+   * 3.2.1 Transform Events: Add Match meta data
+   * 4     Join Events to flat structure
+   */
+
+  const propIncludedFilterType = R.curry((type) =>
+    R.compose(R.filter(R.propEq('type', type)), R.prop('included'))
+  );
+
+  const extractJsonToAssets = propIncludedFilterType('asset');
+  const extractJsonToParticipants = propIncludedFilterType('participant');
+  const extractJsonToRosters = propIncludedFilterType('roster');
+  const extractJsonToMatches = R.prop('data');
+
+  const filterTypeEqTalentEquipped = R.pipe(
+    R.filter(R.propEq('type', 'TalentEquipped')),
+    R.map(R.prop('payload'))
+  );
+
+  const transformParticipantToPayload = R.converge(utils.mergeAllArgs,
+    R.concat(objOfPathsConfig.participant, [
+      utils.objOfPath('Actor', ['attributes', 'actor']),
+      utils.objOfPath('_participant_id', ['id']),
+    ])
+  );
+
+  const transformRosterToPayload = R.converge(utils.mergeAllArgs,
+    R.concat(objOfPathsConfig.roster, [
+      R.pipe(
+        R.path(['attributes', 'stats', 'side']),
+        mapSideToTeam,
+        R.objOf('Team'),
+      ),
+      utils.objOfPath('_roster_id', ['id']),
+      R.pipe(
+        R.path(['relationships', 'participants', 'data']),
+        R.map(R.prop('id')),
+        R.objOf('_participant_ids')
+      )
+    ])
+  );
+
+  const transformMatchToPayload = R.converge(utils.mergeAllArgs,
+    R.concat(objOfPathsConfig.match, [
+      utils.objOfPath('_match_id', ['id']),
+      utils.objOfPath('_asset_id', ['relationships', 'assets', 'data', 0, 'id']),
+      R.pipe(
+        R.path(['relationships', 'rosters', 'data']),
+        R.map(R.prop('id')),
+        R.objOf('_roster_ids')
+      )
+    ])
+  );
+
+  /* SQL style join using relationship info stored in _ attrs */
+  // TODO refactor
+  const joinParticipantPayloadToRosterPayload = R.pipe(
+    R.xprod,
+    R.filter(
+      R.converge(R.contains, [
+        R.path([0, '_participant_id']),
+        R.path([1, '_participant_ids'])
+      ])
+    ),
+    R.map(R.mergeAll)
+  );
+
+  const joinParticipantRosterPayloadToMatchPayload = R.pipe(
+    R.xprod,
+    R.filter(
+      R.converge(R.contains, [
+        R.path([0, '_roster_id']),
+        R.path([1, '_roster_ids'])
+      ])
+    ),
+    R.map(R.mergeAll)
+  );
+
+  // join participant -> roster -> match
+  const joinedParticipantRosterPayloads = R.converge(joinParticipantPayloadToRosterPayload, [
+    R.compose(R.map(transformParticipantToPayload), extractJsonToParticipants),
+    R.compose(R.map(transformRosterToPayload), extractJsonToRosters),
+  ]);
+  const joinedParticipantRosterMatchPayloads = R.converge(joinParticipantRosterPayloadToMatchPayload, [
+    joinedParticipantRosterPayloads,
+    R.compose(R.map(transformMatchToPayload), extractJsonToMatches),
+  ]);
+
+  // chain json -> constructed payload,
+  // wrap in future for compatibility with `loadFTelemetryPayloads`
+  const loadFParticipantRosterMatchPayloads = R.pipe(
+    joinedParticipantRosterMatchPayloads,
+    Future.of,
+  );
+
+  // join [telemetry, asset id] -> { telemetry, asset id }
+  const joinTelemetryPayloadsToId = R.pipe(
+    R.apply(R.xprod),
+    R.map(R.mergeAll),
+  );
+
+  // chain json -> api payload,
+  // add additional meta information to events
+  const loadFTelemetryPayloads = R.pipe(
+    extractJsonToAssets,
+    R.map(R.converge(R.pipe(
+      Future.both,
+      R.map(joinTelemetryPayloadsToId),
+    ), [
+      R.pipe(
+        R.path(['attributes', 'URL']),
+        api.awsRequest,
+        R.map(filterTypeEqTalentEquipped),
+      ),
+      R.pipe(
+        utils.objOfPath('_asset_id', ['id']),
+        R.of,
+        Future.of,
+      ),
+    ])),
+    Future.parallel(Infinity),
+    R.map(R.unnest),
+  );
+
+  // given an array of telemetry payloads and an array of constructed payloads,
+  // return merged telemetry + constructed payloads for identical players & matches
+  // problem: some do not have 'NoTalent' (5v5, 3v3 standard)! -> need right join not inner join
+  const joinParticipantRosterMatchPayloadsToTelemetryPayloads = R.pipe(
+    R.apply(utils.naturalJoinRight),
+    R.map(R.mergeAll),
+  );
+
+  const loadFParticipantRosterMatchTelemetryPayloads = R.pipe(
+    api.apiRequest,
+    R.chain(R.converge(R.pipe(
+      Future.both,
+      R.map(R.pipe(
+        joinParticipantRosterMatchPayloadsToTelemetryPayloads,
+        R.map(R.pipe(
+          R.omit(['_asset_id', '_participant_id', '_participant_ids', '_roster_id', '_roster_ids', '_match_id', 'Team']),
+          R.set(R.lensProp('Count'), 1),
+        )),
+      )),
+    ), [
+      loadFParticipantRosterMatchPayloads,
+      loadFTelemetryPayloads,
+    ]))
+  );
+
+  return loadFParticipantRosterMatchTelemetryPayloads;
+}
+
+module.exports = {
+  loadFPayloads,
+};
diff --git a/backend/file.js b/backend/file.js
new file mode 100644
index 0000000..b750921
--- /dev/null
+++ b/backend/file.js
@@ -0,0 +1,39 @@
+#!/usr/bin/node
+
+const R = require('ramda'),
+  Future = require('fluture'),
+  fs = require('fs'),
+  fsPath = require('fs-path');
+
+function loadFPayloads(config) {
+  return R.curry((file) =>
+    Future.node((cb) => fs.readFile(file, 'utf8', cb))
+          .chain(Future.encase(JSON.parse))
+  );
+}
+
+function saveFPayloads(config) {
+  return R.curry((file, data) =>
+    Future.node((cb) => fsPath.writeFile(file, JSON.stringify(data), cb))
+    .map(() => data)
+  );
+}
+
+function saveFPayloadsTimestamped(config) {
+  return R.curry((timestamp, data) =>
+    saveFPayloads(config)(config.pattern(timestamp), data)
+  );
+}
+
+function loadFPayloadsTimestamped(config) {
+  return R.curry((timestamp) =>
+    loadFPayloads(config)(config.pattern(timestamp))
+  );
+}
+
+module.exports = {
+  loadFPayloads,
+  loadFPayloadsTimestamped,
+  saveFPayloads,
+  saveFPayloadsTimestamped,
+};
diff --git a/backend/index.js b/backend/index.js
new file mode 100644
index 0000000..5da0f82
--- /dev/null
+++ b/backend/index.js
@@ -0,0 +1,27 @@
+#!/usr/bin/node
+
+const R  = require('ramda'),
+  Future = require('fluture'),
+  moment = require('moment'),
+  reduce = require('./reduce'),
+  source = require('./source'),
+  config = require('./config');
+
+const loadFTimestamped = source.loadFTimestamped(config);
+const deriveStatistics = reduce.deriveStatistics(config.reduce);
+const aggregatePayloads = reduce.aggregatePayloads(config.reduce);
+
+function main() {
+  const firstOfMay = moment('2018-05-01');
+
+  const later = R.curry((base, hs) => base.clone().add(hs, 'hours'));
+  const hours = R.range(0, 24 * 5);
+  const laterMoments = R.map(later(firstOfMay), hours);
+
+  const futures = R.map(loadFTimestamped, laterMoments);
+  Future.parallel(1, futures)
+    .map(R.compose(deriveStatistics, aggregatePayloads, R.unnest))
+    .fork(console.error, (d) => console.log(JSON.stringify(d, null, 2)));
+}
+
+main();
diff --git a/backend/params.js b/backend/params.js
new file mode 100644
index 0000000..5edf658
--- /dev/null
+++ b/backend/params.js
@@ -0,0 +1,39 @@
+#!/usr/bin/node
+
+const R = require('ramda'),
+  moment = require('moment');
+
+function paramsForHourSample(config) {
+  const requestArgsForRegionStartEnd = R.curry((region, start, end) => [
+    `/shards/${region}/matches`, {
+      'sort': '-createdAt',
+      'filter[gameMode]': R.join(',', config.modes),
+      'filter[createdAt-start]': start.toISOString(),
+      'filter[createdAt-end]': end.toISOString(),
+      'page[limit]': config.matchesPerRequest,
+      'page[offset]': '0',
+    }
+  ]);
+
+  const intervalMinutes = moment.duration(config.interval, config.intervalUnit).as('minutes');
+  const requestsPerMinutePerRegion = Math.floor(config.requestsPerInterval / config.regions.length);
+  const splitDurationMinutes = Math.floor(intervalMinutes / requestsPerMinutePerRegion);
+  const splitMoments = (hour) => R.map(
+    (offsetIndex) => [
+      hour.clone().minutes(offsetIndex * splitDurationMinutes),
+      hour.clone().minutes((offsetIndex + 1) * splitDurationMinutes),
+    ],
+    R.range(0, requestsPerMinutePerRegion),
+  );
+
+  return R.pipe(
+    splitMoments,
+    R.xprod(config.regions),
+    R.map(R.unnest),
+    R.map(R.apply(requestArgsForRegionStartEnd)),
+  );
+}
+
+module.exports = {
+  paramsForHourSample,
+};
diff --git a/backend/reduce.js b/backend/reduce.js
new file mode 100644
index 0000000..3ac9da2
--- /dev/null
+++ b/backend/reduce.js
@@ -0,0 +1,51 @@
+#!/usr/bin/node
+
+const R = require('ramda'),
+  utils = require('./utils');
+
+function aggregatePayloads(config) {
+  // Ramda group() only groups neighbors, so sort beforehand
+  const sorter  = R.props(config.group);
+  const grouper = utils.eqPropses(config.group);
+
+  // sum & count
+  const aggregator = R.pipe(
+    R.sortBy(sorter),
+    R.groupWith(grouper),
+    R.map(
+      R.converge(utils.mergeAllArgs, [
+        R.pipe(
+          R.project(config.group),
+          R.prop(0),
+        ),
+        R.pipe(
+          R.project(R.concat(config.sum, R.of('Count'))),
+          utils.mergeAllWith(R.add),
+          utils.unlist,
+        ),
+      ]),
+    ),
+  );
+
+  return aggregator;
+}
+
+function deriveStatistics(config) {
+  const sorter = R.compose(R.defaultTo(0), R.prop('Winner'));
+
+  // divide sums by counts = averages
+  const transformer = R.map(
+    R.converge(utils.mergeAllArgs, [
+      utils.projectProps(config.group),
+      utils.divPropsByProp('Count', config.sum),
+      utils.projectProps(['Count']),
+    ])
+  );
+
+  return R.compose(transformer, R.sortBy(sorter));
+}
+
+module.exports = {
+  aggregatePayloads,
+  deriveStatistics,
+};
diff --git a/backend/source.js b/backend/source.js
new file mode 100644
index 0000000..0f4fd28
--- /dev/null
+++ b/backend/source.js
@@ -0,0 +1,39 @@
+#!/usr/bin/node
+
+const R  = require('ramda'),
+  Future = require('fluture'),
+  etl    = require('./etl'),
+  params = require('./params'),
+  reduce = require('./reduce'),
+  file   = require('./file');
+
+function doFetchProcessStoreHour(config) {
+  const intervalSleep = 60 / config.api.intervalsPerMinute * 1000;
+
+  const paramsForHourSample = params.paramsForHourSample(config.api);
+  const loadFApi            = etl.loadFPayloads(config.etl);
+  const saveFPayloads       = file.saveFPayloadsTimestamped(config.file);
+  const aggregatePayloads   = reduce.aggregatePayloads(config.reduce);
+
+  const requestsForHour = (timestamp) => R.map(R.apply(loadFApi), paramsForHourSample(timestamp));
+
+  return (timestamp) => Future.parallel(1, requestsForHour(timestamp))
+    .map(R.compose(aggregatePayloads, R.unnest))
+    .chainRej((err) => { console.error(err); return Future.of({}); }) // fail silently
+    .chain(saveFPayloads(timestamp));
+    //.chain(Future.after(intervalSleep));
+  // TODO should sleep after request; throttling rpm should happen
+  // at a higher level though. CPU / IO / AWS requests stall
+  // as well
+}
+
+function loadFTimestamped(config) {
+  // load from file or fall back to API and store
+  return (timestamp) =>
+    file.loadFPayloads(config)(config.file.pattern(timestamp))
+    .or(doFetchProcessStoreHour(config)(timestamp));
+};
+
+module.exports = {
+  loadFTimestamped,
+};
diff --git a/backend/utils.js b/backend/utils.js
new file mode 100644
index 0000000..6cff356
--- /dev/null
+++ b/backend/utils.js
@@ -0,0 +1,72 @@
+#!/usr/bin/node
+
+const R = require('ramda');
+
+const lengthEq = (n) => R.compose(R.equals(n), R.length);
+const list = R.unapply(R.identity);
+const unlist = R.apply(R.identity);
+const mergeAllArgs = R.compose(R.mergeAll, list);
+const objOfPath = R.curry((objOf, path) => R.compose(R.objOf(objOf), R.path(path)));
+const andAll = R.all(R.equals(R.T()));
+const andAllArgs = R.compose(andAll, list);
+
+const projectProps = R.curry(
+  (proj, d) => R.compose(R.prop(0), R.project(proj), R.of)(d)
+);
+
+const joinRight = R.curry((mapper1, mapper2, t1, t2) => {
+  let indexed = R.indexBy(mapper1, t1);
+  return t2.map((t2row) => R.merge(t2row, indexed[mapper2(t2row)]));
+});
+
+const commonKeys = R.useWith(R.intersection, [
+  R.keys,
+  R.keys,
+]);
+
+const naturalJoinRight = R.curry((t1, t2) => joinRight(
+  R.props(commonKeys(t1[0], t2[0])),
+  R.props(commonKeys(t1[0], t2[0])),
+  t2,
+  t1,
+));
+
+const eqPropses = R.curry((props) => // TODO should create function with arity 3 but doesn't because it's nested
+  R.converge(andAllArgs, R.map(R.eqProps, props))
+);
+
+const mergeAllWith = (merger) => R.until(lengthEq(1), R.pipe( // TODO same issue
+  R.splitEvery(2),
+  R.map(
+    R.ifElse(lengthEq(2), // TODO refactor
+             R.apply(R.mergeWith(merger)),
+             R.apply(R.identity))
+  )
+));
+
+const divPropsByProp = R.curry((prop, props, data) => { // TODO refactor
+  return R.fromPairs(
+    R.map((aProp) =>
+          R.contains(aProp, props) ?
+            [aProp, R.prop(aProp, data) / R.prop(prop, data)]
+          : [aProp, R.prop(aProp, data)],
+          R.keys(data),
+    )
+  );
+});
+
+module.exports = {
+  list,
+  unlist,
+  mergeAllArgs,
+  objOfPath,
+  andAll,
+  andAllArgs,
+  joinRight,
+  naturalJoinRight,
+  eqPropses,
+  lengthEq,
+  mergeAllWith,
+  projectProps,
+  divPropsByProp,
+};