<commit_before>/**
 * Worker running tasks on timed intervals
 **/
const config = require('../config');
const constants = require('dotaconstants');
const redis = require('../store/redis');
const queue = require('../store/queue');
const db = require('../store/db');
const queries = require('../store/queries');
const buildSets = require('../store/buildSets');
const utility = require('../util/utility');
// const getMMStats = require('../util/getMMStats');
const vdf = require('simple-vdf');
const async = require('async');
const fs = require('fs');
const moment = require('moment');

const sql = {};
const sqlq = fs.readdirSync('./sql');
sqlq.forEach((f) => {
  sql[f.split('.')[0]] = fs.readFileSync(`./sql/${f}`, 'utf8');
});

function invokeInterval(func, delay) {
  // invokes the function immediately, waits for callback, waits the delay, and then calls it again
  (function invoker() {
    redis.get(`worker:${func.name}`, (err, fresh) => {
      if (err) {
        return setTimeout(invoker, delay);
      }
      if (fresh && config.NODE_ENV !== 'development') {
        console.log('skipping %s', func.name);
        return setTimeout(invoker, delay);
      }
      console.log('running %s', func.name);
      console.time(func.name);
      return func((err) => {
        if (err) {
          // log the error, but wait until next interval to retry
          console.error(err);
        } else {
          // mark success, don't redo until this key expires
          redis.setex(`worker:${func.name}`, (delay / 1000) * 0.9, '1');
        }
        console.timeEnd(func.name);
        setTimeout(invoker, delay);
      });
    });
  }());
}

function doBuildSets(cb) {
  buildSets(db, redis, cb);
}
/*
function doMMStats(cb) {
  getMMStats(redis, cb);
}
*/
function doDistributions(cb) {
  function loadData(key, mapFunc, cb) {
    db.raw(sql[key]).asCallback((err, results) => {
      if (err) {
        return cb(err);
      }
      mapFunc(results);
      return cb(err, results);
    });
  }
  async.parallel({
    country_mmr(cb) {
      const mapFunc = function mapCountryMmr(results) {
        results.rows = results.rows.map((r) => {
          const ref = constants.countries[r.loccountrycode];
          r.common = ref ? ref.name.common : r.loccountrycode;
          return r;
        });
      };
      loadData('country_mmr', mapFunc, cb);
    },
    mmr(cb) {
      const mapFunc = function mapMmr(results) {
        const sum = results.rows.reduce((prev, current) =>
          ({
            count: prev.count + current.count,
          }), {
          count: 0,
        });
        results.rows = results.rows.map((r, i) => {
          r.cumulative_sum = results.rows.slice(0, i + 1).reduce((prev, current) =>
            ({
              count: prev.count + current.count,
            }), {
            count: 0,
          }).count;
          return r;
        });
        results.sum = sum;
      };
      loadData('mmr', mapFunc, cb);
    },
  }, (err, result) => {
    if (err) {
      return cb(err);
    }
    Object.keys(result).forEach((key) => {
      redis.set(`distribution:${key}`, JSON.stringify(result[key]));
    });
    return cb(err);
  });
}

function doQueueCleanup(cb) {
  queue.cleanup(redis, cb);
}

function doProPlayers(cb) {
  const container = utility.generateJob('api_notable', {});
  utility.getData(container.url, (err, body) => {
    if (err) {
      return cb(err);
    }
    return async.each(body.player_infos, (p, cb) => {
      if (p.account_id === 180012313 && p.locked_until < 1502694000) {
        p.locked_until = 1502694000;
      }
      queries.upsert(db, 'notable_players', p, {
        account_id: p.account_id,
      }, cb);
    }, cb);<commit_msg>add Rr~ to ingestion fix<commit_after>/**
 * Worker running tasks on timed intervals
 **/
const config = require('../config');
const constants = require('dotaconstants');
const redis = require('../store/redis');
const queue = require('../store/queue');
const db = require('../store/db');
const queries = require('../store/queries');
const buildSets = require('../store/buildSets');
const utility = require('../util/utility');
// const getMMStats = require('../util/getMMStats');
const vdf = require('simple-vdf');
const async = require('async');
const fs = require('fs');
const moment = require('moment');

const sql = {};
const sqlq = fs.readdirSync('./sql');
sqlq.forEach((f) => {
  sql[f.split('.')[0]] = fs.readFileSync(`./sql/${f}`, 'utf8');
});

function invokeInterval(func, delay) {
  // invokes the function immediately, waits for callback, waits the delay, and then calls it again
  (function invoker() {
    redis.get(`worker:${func.name}`, (err, fresh) => {
      if (err) {
        return setTimeout(invoker, delay);
      }
      if (fresh && config.NODE_ENV !== 'development') {
        console.log('skipping %s', func.name);
        return setTimeout(invoker, delay);
      }
      console.log('running %s', func.name);
      console.time(func.name);
      return func((err) => {
        if (err) {
          // log the error, but wait until next interval to retry
          console.error(err);
        } else {
          // mark success, don't redo until this key expires
          redis.setex(`worker:${func.name}`, (delay / 1000) * 0.9, '1');
        }
        console.timeEnd(func.name);
        setTimeout(invoker, delay);
      });
    });
  }());
}

function doBuildSets(cb) {
  buildSets(db, redis, cb);
}
/*
function doMMStats(cb) {
  getMMStats(redis, cb);
}
*/
function doDistributions(cb) {
  function loadData(key, mapFunc, cb) {
    db.raw(sql[key]).asCallback((err, results) => {
      if (err) {
        return cb(err);
      }
      mapFunc(results);
      return cb(err, results);
    });
  }
  async.parallel({
    country_mmr(cb) {
      const mapFunc = function mapCountryMmr(results) {
        results.rows = results.rows.map((r) => {
          const ref = constants.countries[r.loccountrycode];
          r.common = ref ? ref.name.common : r.loccountrycode;
          return r;
        });
      };
      loadData('country_mmr', mapFunc, cb);
    },
    mmr(cb) {
      const mapFunc = function mapMmr(results) {
        const sum = results.rows.reduce((prev, current) =>
          ({
            count: prev.count + current.count,
          }), {
          count: 0,
        });
        results.rows = results.rows.map((r, i) => {
          r.cumulative_sum = results.rows.slice(0, i + 1).reduce((prev, current) =>
            ({
              count: prev.count + current.count,
            }), {
            count: 0,
          }).count;
          return r;
        });
        results.sum = sum;
      };
      loadData('mmr', mapFunc, cb);
    },
  }, (err, result) => {
    if (err) {
      return cb(err);
    }
    Object.keys(result).forEach((key) => {
      redis.set(`distribution:${key}`, JSON.stringify(result[key]));
    });
    return cb(err);
  });
}

function doQueueCleanup(cb) {
  queue.cleanup(redis, cb);
}

function doProPlayers(cb) {
  const container = utility.generateJob('api_notable', {});
  utility.getData(container.url, (err, body) => {
    if (err) {
      return cb(err);
    }
    return async.each(body.player_infos, (p, cb) => {
      if ((p.account_id === 180012313 || p.account_id === 323792491)
        && p.locked_until < 1502694000) {
        p.locked_until = 1502694000;
      }
      queries.upsert(db, 'notable_players', p, {
        account_id: p.account_id,
      }, cb);
    }, cb);
{'fraction_non_alphanumeric': 0.17417457559149846, 'fraction_numerical': 0.012431493115893597, 'mean_word_length': 2.4258241758241756, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 27, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '7506086', 'n_tokens_mistral': 2473, 'n_tokens_neox': 2392, 'n_words': 513}