rewrite for new schema

1 files changed, 125 insertions, 87 deletions

diff --git a/worker.js b/worker.js
index 1f1263e..fbced5f 100644
--- a/worker.js
+++ b/worker.js
@@ -8,14 +8,16 @@ var amqp = require("amqplib"),
     hash = require("object-hash");
 
 var RABBITMQ_URI = process.env.RABBITMQ_URI,
-    DATABASE_URI = process.env.DATABASE_URI;
+    DATABASE_URI = process.env.DATABASE_URI,
+    BATCHSIZE = parseInt(process.env.BATCHSIZE) || 5,  // players + globals
+    IDLE_TIMEOUT = parseFloat(process.env.IDLE_TIMEOUT) || 1000;  // ms
 
 (async () => {
     let seq, model, rabbit, ch;
 
     while (true) {
         try {
-            seq = new Seq(DATABASE_URI),
+            seq = new Seq(DATABASE_URI, { logging: false }),
             rabbit = await amqp.connect(RABBITMQ_URI),
             ch = await rabbit.createChannel();
             await ch.assertQueue("crunch", {durable: true});
@@ -27,50 +29,16 @@ var RABBITMQ_URI = process.env.RABBITMQ_URI,
     }
     model = require("../orm/model")(seq, Seq);
 
-    let queue = [],
-        timer = undefined;
-
     await seq.sync();
 
-    // no batching for cruncher, not worth it
-    // we'll read a million tables and commit just one record
-    // once per hour
-    await ch.prefetch(1);
+    await ch.prefetch(1);  // TODO batching
 
     ch.consume("crunch", async (msg) => {
-        let dim = JSON.parse(msg.content),
-            filter = dim.filter, dimension, filter_hash,
-            action;
-        console.log("crunching dimension", dim);
-
-        // we don't want to store duplicate filters
-        // but we can't reliably compare JSON in SQL either
-        // so we hash the object and lookup/store a sha1
-        filter_hash = hash(dim.filter, { unorderedArrays: true });
-        try {
-            dimension = (await model.StatsDimensions.findOrCreate({
-                where: {
-                    dimension_on: dim.dimension_on,
-                    filter_hash: filter_hash
-                }
-            }))[0];
-        } catch (err) {
-            console.error(err);
-            await ch.nack(msg, false, false);
-            return;
-        }
-
-        // not implemented -> error
-        switch (dimension.dimension_on) {
-            case "hero":
-                action = calculateHeroStats(dimension.id, filter);
-                break;
-            default:
-                action = async () => { throw "dimension_on not implemented " + dimension.dimension_on; };
-        }
+        if (msg.properties.type == "global") {}
 
         try {
-            await action;
+            if (msg.properties.type == "player")
+                await calculate_player_point(msg.content.toString());
             console.log("acking");
             await ch.ack(msg);
         } catch (err) {
@@ -79,23 +47,103 @@ var RABBITMQ_URI = process.env.RABBITMQ_URI,
         }
     }, { noAck: false });
 
-    async function calculateHeroStats(dimension_id, filter) {
+    async function calculate_player_point(player_api_id) {
         let win_rate, pick_rate, gold_per_min;
 
+        let player = await model.Player.findOne(
+            { where: { api_id: player_api_id } });
+        // TODO cache dimensions
+        // Series and Filter are special
+        let dimensions = [model.Series, model.Filter, model.Hero, model.GameMode],
+            player_dimensions = ["series", "filter", "hero", "game_mode"],
+            dimension_cache = [];
+        // create a 3D array
+        // [[ ["hero", "Vox"], ["hero", "Taka"], …], [ ["game_mode", "ranked"], … ], …]
+        // (will not use "Vox" but the index instead)
+        await Promise.all(
+            dimensions.map(async (d, idx) =>
+                dimension_cache[idx] = 
+                    (await d.findAll()).map((o) => [d, o]))
+        );
+
+        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 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 …
+        let combos = cartesian(dimension_cache);
+        await Promise.all(combos.map(async (combo) => {
+            let where_aggr = {},
+                where_links = {};
+            // create skeleton: where hero_id=$hero
+            // for aggregation, use series as range
+            // for links, use the id
+            combo.map((tuple) => {
+                // [table name, table element]
+                if (player_dimensions.indexOf(tuple[0].tableName) > -1) {
+                    if (tuple[1].get("name") != "all") {
+                        // exclude series & filter, added below
+                        if (tuple[0].tableName == "filter") {
+                            // merge custom filters
+                            if (tuple[1].get("filter_on") == "player")
+                                Object.assign(where_aggr,
+                                    tuple[1].get("filter"));
+                        } else if (tuple[0].tableName == "series") {
+                            // series is special,
+                            // use start < date < end comparison
+                            where_aggr.created_at = { $between: [
+                                tuple[1].get("start"),
+                                tuple[1].get("end")
+                            ] }
+                        } else where_aggr["$participant." + tuple[0].tableName + ".id$"] =
+                            tuple[1].id
+                    }
+                    where_links[tuple[0].tableName + "_id"] = tuple[1].id
+                }
+            });
+            // make it player specific
+            where_aggr["$participant.player_api_id$"] = player_api_id;
+            where_links["player_id"] = player.id;
+            // aggregate participant_stats with our condition
+            let stats = await aggregate_stats(where_aggr);
+            if (stats != undefined) {
+                console.log("inserting stats for player", player.name);
+                Object.assign(stats, where_links);
+                await model.PlayerPoint.upsert(stats, {
+                    where: where_links
+                });
+            }
+        }));
+    }
+
+    // return aggregated stats based on $where as WHERE clauses
+    async function aggregate_stats(where) {
         // in literals: q -> column name, e -> function or string
         let q = (qry) => seq.dialect.QueryGenerator.quote(qry),
             e = (qry) => seq.dialect.QueryGenerator.escape(qry);
 
-        // TODO (workaround) ignore old API data where participant.gold was null
-        filter["$participant.gold$"] = { $ne: null };
-        console.log(filter);
-
         // 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" ]
 
-        // calculate stats
-        let total_participants = await model.Participant.count({
-            where: filter,
+        let associations = [ {
+            model: model.Participant,
+            as: "participant",
+            attributes: [],
             include: [ {
                 model: model.Roster,
                 attributes: [],
@@ -103,52 +151,42 @@ var RABBITMQ_URI = process.env.RABBITMQ_URI,
                     model: model.Match,
                     attributes: []
                 } ]
-            }, {
-                model: model.Heros,
-                attributes: []
+                }, {
+                    model: model.Hero,
+                    as: "hero",
+                    attributes: []
+                }, {
+                    model: model.Series,
+                    as: "series",
+                    attributes: []
+                }, {
+                    model: model.GameMode,
+                    as: "game_mode",
+                    attributes: []
+                }, {
+                    model: model.Role,
+                    as: "role",
+                    attributes: []
             } ]
+        } ];
+
+        // TODO cache this?
+        let total_participants = await model.ParticipantStats.count({
+            where: where,
+            include: associations
         });
+        if (total_participants == 0) return undefined;  // not enough data
 
-        // TODO (workaround) filter unmapped actor<->hero
-        filter["$hero.id$"] = { $ne: null };
-        let data = await model.Participant.findAll({
-            where: filter,
+        let data = await model.ParticipantStats.findOne({
+            where: where,
             attributes: [
-                // meta
-                [ seq.col("hero.name"), "hero_name" ],
-                [ seq.col("hero.id"), "hero_id" ],
-
-                // stats, see `hero_stats` table
                 [ seq.literal(`${q(seq.fn("count", seq.col("participant.id") ))} / ${total_participants}`), "pick_rate" ],
                 [ seq.fn("avg", seq.cast(seq.col("participant.winner"), "int") ), "win_rate" ],
-                [ seq.fn("sum", seq.literal(`${q("participant.gold")} / ${q("roster.match.duration")}`)), "gold_per_min" ],
-                [ seq.fn("avg", seq.literal(`60.0 * ${q("participant.minion_kills")} / ${q("roster.match.duration")}`)), "cs_per_min" ]
+                [ seq.fn("sum", seq.literal(`${q("participant_stats.gold")} / ${q("participant.roster.match.duration")}`)), "gold_per_min" ],
+                [ seq.fn("avg", seq.literal(`60.0 * ${q("participant_stats.minion_kills")} / ${q("participant.roster.match.duration")}`)), "cs_per_min" ]
             ],
-            group: q("hero.name"),
-            include: [ {
-                model: model.Roster,
-                attributes: [],
-                include: [ {
-                    model: model.Match,
-                    attributes: []
-                } ]
-            }, {
-                model: model.Heros,
-                attributes: []
-            } ]
+            include: associations
         });
-        console.log("hero stats", filter, data.map((d) => d.dataValues));
-        // insert hero x dimension x stats
-        await seq.transaction({ autocommit: false }, (transaction) =>
-            Promise.all(data.map(async (record) => {
-                let hero_stat_db = await model.HeroStats.create(record.dataValues);
-                await model.HeroDimension.create({
-                    hero_id: record.get("hero_id"),
-                    dimension_id: dimension_id,
-                    stats_id: hero_stat_db.get("id"),
-                    computed_on: new Date()
-                }, { transaction: transaction });
-            }))
-        );
+        return data.dataValues;
     }
 })();