rework cruncher

3 files changed, 177 insertions, 253 deletions

diff --git a/crunch_global.sql b/crunch_global.sql
index b0fae3a..7bb6263 100644
--- a/crunch_global.sql
+++ b/crunch_global.sql
@@ -1,8 +1,3 @@
-INSERT IGNORE INTO `keys` (`id`, `key`, `value`) VALUES(3, 'last_participant_crunched', 0), (4, 'participant_crunch_steps', 1000);
-
-SET @lpcid = (SELECT `value` FROM `keys` WHERE `key`='last_participant_crunched');
-SET @lpc_step = (SELECT `value` FROM `keys` WHERE `key`='participant_crunch_steps');
-
 INSERT LOW_PRIORITY INTO global_point
 SELECT
     NULL,
@@ -64,7 +59,7 @@ JOIN `build` ON `build`.`name` = 'all'
 JOIN `role` ON `role`.`id` = `participant`.`role_id` OR `role`.`name` = 'all'
 JOIN `region` ON `region`.`name` = `participant`.`shard_id` OR `region`.`name` = 'all'
 
-WHERE `participant`.`id` > (SELECT @lpcid) and `participant`.`id` < (SELECT @lpcid + @lpc_step)
+WHERE `participant`.`api_id` IN (:participant_api_ids)
 
 GROUP BY `series`.`id`, `filter`.`id`, `hero`.`id`, `game_mode`.`id`, `skill_tier`.`id`, `build`.`id`, `role`.`id`, `region`.`id`
 ORDER BY `participant`.`id`
@@ -105,6 +100,4 @@ ON DUPLICATE KEY UPDATE
 `utility_score` = `utility_score` + VALUES(`utility_score`),
 `synergy_score` = `synergy_score` + VALUES(`synergy_score`),
 `build_score` = `build_score` + VALUES(`build_score`),
-`offmeta_score` = `offmeta_score` + VALUES(`offmeta_score`);
-
-UPDATE `keys` SET `value` = (SELECT @lpcid + @lpc_step) WHERE `key` = 'last_participant_crunched';
+`offmeta_score` = `offmeta_score` + VALUES(`offmeta_score`)
diff --git a/crunch_player.sql b/crunch_player.sql
new file mode 100644
index 0000000..d0e6616
--- /dev/null
+++ b/crunch_player.sql
@@ -0,0 +1,104 @@
+INSERT LOW_PRIORITY INTO player_point
+SELECT
+    NULL,
+    `participant`.`player_api_id`,
+
+    COUNT(`participant`.`id`) AS `played`,
+    SUM(CAST(`participant`.`winner` AS INT)) AS `wins`,
+    SUM(`duration`) AS `time_spent`,
+    
+    `series`.`id`,
+    `role`.`id`,
+    `filter`.`id`,
+    `hero`.`id`,
+    `game_mode`.`id`,
+    NOW(),
+    
+    SUM(`participant_stats`.`kills`) AS `kills`,
+    SUM(`participant_stats`.`deaths`) AS `deaths`,
+    SUM(`participant_stats`.`assists`) AS `assists`,
+    SUM(`participant_stats`.`minion_kills`) AS `minion_kills`,
+    SUM(`participant_stats`.`jungle_kills`) AS `jungle_kills`,
+    SUM(`participant_stats`.`non_jungle_minion_kills`) AS `non_jungle_minion_kills`,
+    SUM(`participant_stats`.`crystal_mine_captures`) AS `crystal_mine_captures`,
+    SUM(`participant_stats`.`gold_mine_captures`) AS `gold_mine_captures`,
+    SUM(`participant_stats`.`kraken_captures`) AS `kraken_captures`,
+    SUM(`participant_stats`.`turret_captures`) AS `turret_captures`,
+    SUM(`participant_stats`.`gold`) AS `gold`,
+    SUM(`participant_stats`.`hero_level`) AS `hero_level`,
+    SUM(`participant_stats`.`kda_ratio`) AS `kda_ratio`,
+    SUM(`participant_stats`.`kill_participation`) AS `kill_participation`,
+    SUM(`participant_stats`.`cs_per_min`) AS `cs_per_min`,
+    SUM(`participant_stats`.`kills_per_min`) AS `kills_per_min`,
+    SUM(`participant_stats`.`impact_score`) AS `impact_score`,
+    SUM(`participant_stats`.`objective_score`) AS `objective_score`,
+    SUM(`participant_stats`.`damage_cp_score`) AS `damage_cp_score`,
+    SUM(`participant_stats`.`damage_wp_score`) AS `damage_wp_score`,
+    SUM(`participant_stats`.`sustain_score`) AS `sustain_score`,
+    SUM(`participant_stats`.`farm_lane_score`) AS `farm_lane_score`,
+    SUM(`participant_stats`.`kill_score`) AS `kill_score`,
+    SUM(`participant_stats`.`objective_lane_score`) AS `objective_lane_score`,
+    SUM(`participant_stats`.`farm_jungle_score`) AS `farm_jungle_score`,
+    SUM(`participant_stats`.`peel_score`) AS `peel_score`,
+    SUM(`participant_stats`.`kill_assist_score`) AS `kill_assist_score`,
+    SUM(`participant_stats`.`objective_jungle_score`) AS `objective_jungle_score`,
+    SUM(`participant_stats`.`vision_score`) AS `vision_score`,
+    SUM(`participant_stats`.`heal_score`) AS `heal_score`,
+    SUM(`participant_stats`.`assist_score`) AS `assist_score`,
+    SUM(`participant_stats`.`utility_score`) AS `utility_score`,
+    SUM(`participant_stats`.`synergy_score`) AS `synergy_score`,
+    SUM(`participant_stats`.`build_score`) AS `build_score`,
+    SUM(`participant_stats`.`offmeta_score`) AS `offmeta_score`
+FROM `participant`
+JOIN `participant_stats` ON `participant_stats`.`participant_api_id` = `participant`.`api_id`
+JOIN `series` ON `participant_stats`.`created_at` BETWEEN `series`.`start` AND `series`.`end` AND `series`.`dimension_on` = 'global'
+JOIN `filter` ON `filter`.`name` = 'all' AND `filter`.`dimension_on` = 'global'
+JOIN `hero` ON `hero`.`id` = `participant`.`hero_id` OR `hero`.`name` = 'all'
+JOIN `game_mode` ON `game_mode`.`id` = `participant`.`game_mode_id` OR `game_mode`.`name` = 'all'
+JOIN `skill_tier` ON `participant`.`skill_tier` BETWEEN `skill_tier`.`start` AND `skill_tier`.`end` OR `skill_tier`.`name` = 'all'
+JOIN `build` ON `build`.`name` = 'all'
+JOIN `role` ON `role`.`id` = `participant`.`role_id` OR `role`.`name` = 'all'
+JOIN `region` ON `region`.`name` = `participant`.`shard_id` OR `region`.`name` = 'all'
+
+WHERE `participant`.`api_id` IN (:participant_api_ids)
+
+GROUP BY `series`.`id`, `filter`.`id`, `hero`.`id`, `game_mode`.`id`, `skill_tier`.`id`, `build`.`id`, `role`.`id`, `region`.`id`
+ORDER BY `participant`.`id`
+
+ON DUPLICATE KEY UPDATE
+`played` = `played` + VALUES(`played`),
+`wins` = `wins` + VALUES(`wins`),
+`kills` = `kills` + VALUES(`kills`),
+`deaths` = `deaths` + VALUES(`deaths`),
+`assists` = `assists` + VALUES(`assists`),
+`minion_kills` = `minion_kills` + VALUES(`minion_kills`),
+`jungle_kills` = `jungle_kills` + VALUES(`jungle_kills`),
+`non_jungle_minion_kills` = `non_jungle_minion_kills` + VALUES(`non_jungle_minion_kills`),
+`crystal_mine_captures` = `crystal_mine_captures` + VALUES(`crystal_mine_captures`),
+`gold_mine_captures` = `gold_mine_captures` + VALUES(`gold_mine_captures`),
+`kraken_captures` = `kraken_captures` + VALUES(`kraken_captures`),
+`turret_captures` = `turret_captures` + VALUES(`turret_captures`),
+`gold` = `gold` + VALUES(`gold`),
+`kda_ratio` = `kda_ratio` + VALUES(`kda_ratio`),
+`kill_participation` = `kill_participation` + VALUES(`kill_participation`),
+`cs_per_min` = `cs_per_min` + VALUES(`cs_per_min`),
+`kills_per_min` = `kills_per_min` + VALUES(`kills_per_min`),
+`impact_score` = `impact_score` + VALUES(`impact_score`),
+`objective_score` = `objective_score` + VALUES(`objective_score`),
+`damage_cp_score` = `damage_cp_score` + VALUES(`damage_cp_score`),
+`damage_wp_score` = `damage_wp_score` + VALUES(`damage_wp_score`),
+`sustain_score` = `sustain_score` + VALUES(`sustain_score`),
+`farm_lane_score` = `farm_lane_score` + VALUES(`farm_lane_score`),
+`kill_score` = `kill_score` + VALUES(`kill_score`),
+`objective_lane_score` = `objective_lane_score` + VALUES(`objective_lane_score`),
+`farm_jungle_score` = `farm_jungle_score` + VALUES(`farm_jungle_score`),
+`peel_score` = `peel_score` + VALUES(`peel_score`),
+`kill_assist_score` = `kill_assist_score` + VALUES(`kill_assist_score`),
+`objective_jungle_score` = `objective_jungle_score` + VALUES(`objective_jungle_score`),
+`vision_score` = `vision_score` + VALUES(`vision_score`),
+`heal_score` = `heal_score` + VALUES(`heal_score`),
+`assist_score` = `assist_score` + VALUES(`assist_score`),
+`utility_score` = `utility_score` + VALUES(`utility_score`),
+`synergy_score` = `synergy_score` + VALUES(`synergy_score`),
+`build_score` = `build_score` + VALUES(`build_score`),
+`offmeta_score` = `offmeta_score` + VALUES(`offmeta_score`)
diff --git a/worker.js b/worker.js
index 7b09a9d..c367111 100644
--- a/worker.js
+++ b/worker.js
@@ -2,21 +2,30 @@
 /* jshint esnext:true */
 "use strict";
 
+/*
+ * cruncher calculates global_point and player_point for stats.
+ * It listens to the queue `crunch` and expects a JSON with
+ * a string that is a `participant.api_id`.
+ * cruncher will update the sums of points
+ * and send a notification to web.
+ */
+
 const amqp = require("amqplib"),
     Promise = require("bluebird"),
+    fs = Promise.promisifyAll(require("fs")),
     winston = require("winston"),
     loggly = require("winston-loggly-bulk"),
     Seq = require("sequelize"),
-    sleep = require("sleep-promise"),
-    hash = require("object-hash");
+    sleep = require("sleep-promise");
 
 const RABBITMQ_URI = process.env.RABBITMQ_URI,
     DATABASE_URI = process.env.DATABASE_URI,
     LOGGLY_TOKEN = process.env.LOGGLY_TOKEN,
-    // number of inserts in one statement
-    CHUNKSIZE = parseInt(process.env.CHUNKSIZE) || 300,
-    MAXCONNS = parseInt(process.env.MAXCONNS) || 10,  // how many concurrent actions
-    CRUNCHERS = parseInt(process.env.CRUNCHERS) || 4;  // how many players to crunch concurrently
+    // size of connection pool
+    MAXCONNS = parseInt(process.env.MAXCONNS) || 3,
+    // number of participants to calculate at once
+    BATCHSIZE = parseInt(process.env.BATCHSIZE) || 1000,
+    LOAD_TIMEOUT = parseInt(process.env.LOAD_TIMEOUT) || 5;  // s
 
 const logger = new (winston.Logger)({
         transports: [
@@ -36,16 +45,11 @@ if (LOGGLY_TOKEN)
         json: true
     });
 
-// helpers
-// split an array into arrays of max chunksize
-function* chunks(arr) {
-    for (let c=0, len=arr.length; c<len; c+=CHUNKSIZE)
-        yield arr.slice(c, c+CHUNKSIZE);
-}
 
 (async () => {
-    let seq, model, rabbit, ch;
+    let seq, rabbit, ch;
 
+    // connect to rabbit & db
     while (true) {
         try {
             seq = new Seq(DATABASE_URI, {
@@ -61,246 +65,69 @@ function* chunks(arr) {
             await sleep(5000);
         }
     }
-    model = require("../orm/model")(seq, Seq);
 
-    function cartesian(arr) {
-        return Array.prototype.reduce.call(arr, function(a, b) {
-            let ret = [];
-            a.forEach(function(a) {
-                b.forEach(function(b) {
-                    ret.push(a.concat([b]));
-                });
-            });
-            return ret;
-        }, [[]]);
-    }
-
-    // create a 3D array
-    // [[ ["hero", "Vox"], ["hero", "Taka"], …], [ ["game_mode", "ranked"], … ], …]
-    // (will not use "Vox" but the index instead)
-    async function dimensions_for(dimensions, on) {
-        let cache = [];
-        await Promise.map(dimensions, async (d, idx) =>
-            cache[idx] = (await d.findAll()).map(
-                (o) => [d, o]).filter((t) =>
-                    t[1].get("dimension_on") == null
-                        || t[1].get("dimension_on") == on)
-        );
-        return cache;
-    }
-    const player_dimensions = await dimensions_for(
-        [model.Series, model.Filter, model.Hero, model.Role,
-            model.GameMode], "player"),
-        global_dimensions = await dimensions_for(
-        [model.Series, model.Filter, model.Hero, model.Role,
-            model.GameMode, model.Skilltier, model.Build,
-            model.Region], "global");
+    // load update SQL scripts; scripts use sequelize replacements
+    // for the `participant_api_id` array
+    const player_script = fs.readFileSync("crunch_player.sql", "utf8"),
+        global_script = fs.readFileSync("crunch_global.sql", "utf8");
 
-    // return every possible [query, insert] combination
-    function calculate_point(points) {
-        return points.map((point) => {
-            // Series and Filter are special
-            let where_aggr = {},
-                where_links = {};
-            // create skeleton: where hero_id=$hero
-            // for aggregation, use series as range
-            // for links, use the id
-            point.forEach((tuple) => {
-                // [table name, table element]
-                if (tuple[1].get("name") != "all") {
-                    // exclude series & filter, added below
-                    if (tuple[0].tableName == "filter") {
-                        // merge custom filters
-                        Object.assign(where_aggr,
-                            tuple[1].get("filter"));
-                    // series and skill_tier are ranged filters
-                    } else if (tuple[0].tableName == "series") {
-                        // use start < date < end comparison
-                        where_aggr.created_at = { $between: [
-                            tuple[1].get("start"),
-                            tuple[1].get("end")
-                        ] }
-                    } else if (tuple[0].tableName == "skill_tier") {
-                        where_aggr["$participant.skill_tier$"] = { $between: [
-                            tuple[1].get("start"),
-                            tuple[1].get("end")
-                        ] }
-                    // build is a special ranged filter
-                    } else if (tuple[0].tableName == "build") {
-                        // TODO!
-                    } else if (tuple[0].tableName == "region") {
-                        // not joined via id, joined via name
-                        where_aggr["$participant.shard_id$"] = tuple[1].get("name");
-                    // most filters are directly as $filter_id on participant
-                    } else where_aggr["$participant." + tuple[0].tableName + "_id$"] =
-                        tuple[1].id
-                }
-                where_links[tuple[0].tableName + "_id"] = tuple[1].get("id");
-            });
-            return [where_aggr, where_links];
-        });
-    }
+    // fill a buffer and execute an SQL on a bigger (> 1o) batch
+    const participants_player = new Set(),
+        participants = new Set(),
+        // store the msgs that should be ACKed
+        buffer = new Set();
+    let timeout = undefined;
 
-    // create an array with every possible combination
-    // hero x game mode x …
-    // Vox  x casual    x …
-    // SAW  x casual    x …
-    // …
-    // Vox  x ranked    x …
-    // SAW  x ranked    x …
-    // …
-    // Vox  x ANY       x …
-    const player_points = calculate_point(cartesian(player_dimensions)),
-        global_points = calculate_point(cartesian(global_dimensions));
+    // set maximum allowed number of unacked msgs
+    await ch.prefetch(BATCHSIZE);
+    ch.consume("crunch", (msg) => {
+        const api_id = msg.content.toString();
+        if (msg.type == "global")
+            participants.add(api_id);
+        // else exclusively add data to player, used for player refresh
+        participants_player.add(api_id);
+        buffer.add(msg);
+        if (timeout == undefined) timeout = setTimeout(crunch, LOAD_TIMEOUT*1000);
+        if (buffer.size >= BATCHSIZE) crunch();
+    }, { noAck: false });
 
-    await ch.prefetch(CRUNCHERS);
+    // execute the scripts
+    async function crunch() {
+        const profiler = logger.startTimer();
+        logger.info("crunching");
 
-    ch.consume("crunch", async (msg) => {
-        const player_id = msg.content.toString();
-        logger.info("working",
-            { type: msg.properties.type, id: player_id });
+        // prevent async issues
+        const api_ids_player = [...participants_player],
+            api_ids = [...participants],
+            msgs = new Set(buffer);
+        participants.clear();
+        participants_player.clear();
+        buffer.clear();
+        clearTimeout(timeout);
+        timeout = undefined;
 
-        let profiler = logger.startTimer();
-        // service wide stats
-        if (msg.properties.type == "global") {
-            try {
-                await calculate_global_point();
-                logger.info("acking");
-                await ch.ack(msg);
-                // tell web
-                await ch.publish("amq.topic", "global", new Buffer("points_update"));
-            } catch (err) {
-                logger.error("SQL error", err);
-                await ch.nack(msg, false, true);  // requeue
-            }
-        }
-        // player stats
-        if (msg.properties.type == "player") {
-            try {
-                await calculate_player_point(player_id);
-                logger.info("acking");
-                await ch.ack(msg);
-            } catch (err) {
-                logger.error("SQL error", err);
-                await ch.nack(msg, false, true);  // requeue
-            }
-            // tell web
-            const player = await model.Player.findOne({
-                where: { api_id: player_id },
-                attributes: ["name"]
+        if (api_ids.length > 0)
+            await seq.query(global_script, {
+                replacements: { participant_api_ids: api_ids },
+                type: seq.QueryTypes.UPSERT
+            });
+        if (api_ids_player.length > 0)
+            await seq.query(player_script, {
+                replacements: { participant_api_ids: api_ids_player },
+                type: seq.QueryTypes.UPSERT
             });
-            if (player != null) {
-                logger.info("updated player", { name: player.get("name") });
-                await ch.publish("amq.topic", "player." + player.get("name"),
-                    new Buffer("points_update"));
-            }
-        }
-        profiler.done("calculations for " +
-            msg.properties.type + " " + player_id);
-    }, { noAck: false });
-
-    async function calculate_global_point() {
-        logger.info("crunching global stats, this could take a while");
-        await Promise.map(global_points, async (tuple, idx, len) => {
-            const progress = Math.floor(100*100 * (1-idx/len)) / 100,
-                where_aggr = tuple[0], where_links = tuple[1];
-            // aggregate participant_stats with our condition
-            let stats = await aggregate_stats(where_aggr);
-            stats.updated_at = seq.fn("NOW");
-            logger.info("inserting global stat", { progress: progress });
-            Object.assign(stats, where_links);
-            await model.GlobalPoint.upsert(stats);
-        }, { concurrency: MAXCONNS });
-        logger.info("committing");
-    }
-
-    async function calculate_player_point(player_api_id) {
-        let point_records = [];
-        logger.info("crunching player", { id: player_api_id });
 
-        await Promise.map(player_points, async (tuple) => {
-            const where_aggr = tuple[0],
-                where_links = tuple[1];
-            // make it player specific
-            where_aggr["$participant.player_api_id$"] = player_api_id;
-            where_aggr["final"] = true;  // only end of match stats
-            where_links["player_api_id"] = player_api_id;
-            // aggregate participant_stats with our condition
-            let stats = await aggregate_stats(where_aggr);
-            if (stats != undefined) {
-                stats.updated_at = seq.fn("NOW");
-                Object.assign(stats, where_links);
-                point_records.push(stats);
-            }
-        }, { concurrency: MAXCONNS });
+        // ack
+        await Promise.map(msgs, async (m) => await ch.ack(m));
+        // notify web
+        // TODO notify for player too
+        await ch.publish("amq.topic", "global", new Buffer("points_update"));
 
-        logger.info("inserting into db");
-        await seq.transaction({ autocommit: false }, async (transaction) => {
-            await Promise.map(chunks(point_records), async (p_r) =>
-                model.PlayerPoint.bulkCreate(p_r, {
-                    updateOnDuplicate: [],  // all
-                    transaction: transaction
-                }), { concurrency: MAXCONNS }
-            )
-        });
+        profiler.done("crunched");
     }
+})();
 
-    // return aggregated stats based on $where as WHERE clauses
-    async function aggregate_stats(where) {
-        // in literals: q -> column name, e -> function or string
-        const q = (qry) => seq.dialect.QueryGenerator.quote(qry),
-            e = (qry) => seq.dialect.QueryGenerator.escape(qry);
-
-        // alternative for win rate
-        //[ seq.literal(`${e(seq.fn("sum", seq.cast(seq.col("participant.winner"), "int") ))} / ${e(seq.fn("count", seq.col("participant.id")))}`), "win_rate" ]
-
-        // short to sum a participant row as player stat with the same name
-        const sum = (name) => [ seq.fn("sum", seq.col("participant_stats." + name)), name ];
+process.on("unhandledRejection", function(reason, promise) {
+    logger.error(reason);
+});
 
-        const data = await model.ParticipantStats.findOne({
-            where: where,
-            attributes: [
-                [ seq.fn("count", seq.col("participant.id")), "played" ],
-                [ seq.fn("sum", seq.col("duration")), "time_spent" ],
-                [ seq.fn("sum", seq.cast(seq.col("participant.winner"), "int") ), "wins" ],
-                sum("kills"),
-                sum("deaths"),
-                sum("assists"),
-                sum("minion_kills"),
-                sum("jungle_kills"),
-                sum("non_jungle_minion_kills"),
-                sum("crystal_mine_captures"),
-                sum("turret_captures"),
-                sum("kda_ratio"),
-                sum("kill_participation"),
-                sum("impact_score"),
-                sum("objective_score"),
-                sum("damage_cp_score"),
-                sum("damage_wp_score"),
-                sum("sustain_score"),
-                sum("farm_lane_score"),
-                sum("kill_score"),
-                sum("objective_lane_score"),
-                sum("farm_jungle_score"),
-                sum("peel_score"),
-                sum("kill_assist_score"),
-                sum("objective_jungle_score"),
-                sum("vision_score"),
-                sum("heal_score"),
-                sum("assist_score"),
-                sum("utility_score"),
-                sum("synergy_score"),
-                sum("build_score"),
-                sum("offmeta_score"),
-                sum("kraken_captures"),
-                sum("gold")
-            ],
-            include: [ {
-                model: model.Participant,
-                as: "participant",
-                attributes: []
-            } ]
-        });
-        return data.dataValues;
-    }
-})();