rewrite in NodeJS

8 files changed, 224 insertions, 208 deletions

diff --git a/.gitignore b/.gitignore
index 9a05e2d..00cbbdf 100644
--- a/.gitignore
+++ b/.gitignore
@@ -1,91 +1,59 @@
-# Byte-compiled / optimized / DLL files
-__pycache__/
-*.py[cod]
-*$py.class
+# Logs
+logs
+*.log
+npm-debug.log*
+yarn-debug.log*
+yarn-error.log*
 
-# C extensions
-*.so
+# Runtime data
+pids
+*.pid
+*.seed
+*.pid.lock
 
-# Distribution / packaging
-.Python
-env/
-build/
-develop-eggs/
-dist/
-downloads/
-eggs/
-.eggs/
-lib/
-lib64/
-parts/
-sdist/
-var/
-wheels/
-*.egg-info/
-.installed.cfg
-*.egg
+# Directory for instrumented libs generated by jscoverage/JSCover
+lib-cov
 
-# PyInstaller
-#  Usually these files are written by a python script from a template
-#  before PyInstaller builds the exe, so as to inject date/other infos into it.
-*.manifest
-*.spec
+# Coverage directory used by tools like istanbul
+coverage
 
-# Installer logs
-pip-log.txt
-pip-delete-this-directory.txt
+# nyc test coverage
+.nyc_output
 
-# Unit test / coverage reports
-htmlcov/
-.tox/
-.coverage
-.coverage.*
-.cache
-nosetests.xml
-coverage.xml
-*,cover
-.hypothesis/
+# Grunt intermediate storage (http://gruntjs.com/creating-plugins#storing-task-files)
+.grunt
 
-# Translations
-*.mo
-*.pot
+# Bower dependency directory (https://bower.io/)
+bower_components
 
-# Django stuff:
-*.log
-local_settings.py
+# node-waf configuration
+.lock-wscript
 
-# Flask stuff:
-instance/
-.webassets-cache
+# Compiled binary addons (http://nodejs.org/api/addons.html)
+build/Release
 
-# Scrapy stuff:
-.scrapy
+# Dependency directories
+node_modules/
+jspm_packages/
 
-# Sphinx documentation
-docs/_build/
+# Typescript v1 declaration files
+typings/
 
-# PyBuilder
-target/
+# Optional npm cache directory
+.npm
 
-# Jupyter Notebook
-.ipynb_checkpoints
+# Optional eslint cache
+.eslintcache
 
-# pyenv
-.python-version
+# Optional REPL history
+.node_repl_history
 
-# celery beat schedule file
-celerybeat-schedule
+# Output of 'npm pack'
+*.tgz
 
-# dotenv
-.env
+# Yarn Integrity file
+.yarn-integrity
 
-# virtualenv
-.venv/
-venv/
-ENV/
-
-# Spyder project settings
-.spyderproject
+# dotenv environment variables file
+.env
 
-# Rope project settings
-.ropeproject
diff --git a/.gitmodules b/.gitmodules
index 47695ea..e69de29 100644
--- a/.gitmodules
+++ b/.gitmodules
@@ -1,3 +0,0 @@
-[submodule "joblib"]
-	path = joblib
-	url = https://gitlab.com/vainglorygame/joblib.git
diff --git a/Dockerfile b/Dockerfile
new file mode 100644
index 0000000..9c3debd
--- /dev/null
+++ b/Dockerfile
@@ -0,0 +1,9 @@
+FROM node:7.7-alpine
+
+RUN mkdir -p /usr/src/app
+WORKDIR /usr/src/app
+
+COPY . /usr/src/app
+RUN npm install && npm cache clean
+
+CMD ["node", "worker.js"]
diff --git a/api.py b/api.py
deleted file mode 100644
index d8daf46..0000000
--- a/api.py
+++ /dev/null
@@ -1,126 +0,0 @@
-#!/usr/bin/python3
-
-import asyncio
-import os
-import glob
-import logging
-import json
-import asyncpg
-
-import joblib.worker
-
-queue_db = {
-    "host": os.environ.get("POSTGRESQL_SOURCE_HOST") or "localhost",
-    "port": os.environ.get("POSTGRESQL_SOURCE_PORT") or 5433,
-    "user": os.environ.get("POSTGRESQL_SOURCE_USER") or "vainraw",
-    "password": os.environ.get("POSTGRESQL_SOURCE_PASSWORD") or "vainraw",
-    "database": os.environ.get("POSTGRESQL_SOURCE_DB") or "vainsocial-raw"
-}
-
-db_config = {
-    "host": os.environ.get("POSTGRESQL_DEST_HOST") or "localhost",
-    "port": os.environ.get("POSTGRESQL_DEST_PORT") or 5432,
-    "user": os.environ.get("POSTGRESQL_DEST_USER") or "vainweb",
-    "password": os.environ.get("POSTGRESQL_DEST_PASSWORD") or "vainweb",
-    "database": os.environ.get("POSTGRESQL_DEST_DB") or "vainsocial-web"
-}
-
-
-class Cruncher(joblib.worker.Worker):
-    def __init__(self):
-        self._con = None
-        super().__init__(jobtype="crunch")
-
-    async def connect(self, dbconf, queuedb):
-        """Connect to database."""
-        logging.warning("connecting to database")
-        await super().connect(**queuedb)
-        self._con = await asyncpg.connect(**dbconf)
-
-    async def setup(self):
-        pass
-
-    async def _windup(self):
-        self._tr = self._con.transaction()
-        await self._tr.start()
-
-    async def _teardown(self, failed):
-        if failed:
-            await self._tr.rollback()
-        else:
-            await self._tr.commit()
-
-    async def _execute_job(self, jobid, payload, priority):
-        """Finish a job."""
-        dimension_id = payload["dimension"]
-
-        logging.debug("%s: crunching dimension %s",
-                      jobid, dimension_id)
-
-        dimension = await self._con.fetchrow("""
-        SELECT * FROM stats_dimensions WHERE id=$1
-        """, dimension_id)
-        if dimension is None:
-            logging.warning("%s: invalid dimension '%s', exiting",
-                            jobid, dimension_id)
-            raise joblib.worker.JobFailed("invalid dimension '" + dimension_id
-                                          + "'",
-                                          False)
-
-        table = dimension["dimension_on"]
-        field = dimension["name"]
-        value = dimension["value"]
-
-        if table == "hero":
-            filter_query = ""
-            if field == "patch":
-                filter_query = "WHERE match.patch_version='" + value + "'"
-            if field == "recent_matches":
-                filter_query = "WHERE match.api_id IN (SELECT api_id FROM match ORDER BY created_at DESC LIMIT " + value + ")"
-            if field == "skill_tier":
-                filter_query = "WHERE participant.skill_tier=" + value
-
-            stats = await self._con.fetch("""
-            SELECT
-                heros.id,
-                SUM(participant.winner::INT)/COUNT(participant.winner)::FLOAT AS win_rate,
-                COUNT(participant.hero)/(SELECT COUNT(hero) FROM participant)::FLOAT AS pick_rate,
-                SUM(60*participant.minion_kills/match.duration::FLOAT)/COUNT(participant.farm)::FLOAT AS cs_per_min,
-                0 AS gold_per_min
-            FROM participant
-
-            JOIN participant_stats ON participant.api_id=participant_stats.participant_api_id
-            JOIN roster ON roster.api_id=participant.roster_api_id
-            JOIN match ON match.api_id=roster.match_api_id
-
-            JOIN heros ON heros.api_name=hero
-            """
-            + filter_query +
-            """
-            GROUP BY heros.id
-            """)
-
-            for stat in stats:
-                stat_id = await self._con.fetchrow("""
-                    INSERT INTO hero_stats(win_rate, pick_rate, cs_per_min, gold_per_min)
-                    VALUES($1, $2, $3, $4)
-                    RETURNING hero_stats.id
-                    """, stat["win_rate"], stat["pick_rate"], stat["cs_per_min"],
-                    stat["gold_per_min"])
-                await self._con.fetch("""
-                    INSERT INTO hero_dimension(hero_id, dimension_id, stats_id, computed_on)
-                    VALUES($1, $2, $3, NOW())
-                    """, stat["id"], dimension["id"],
-                    stat_id["id"])
-
-async def startup():
-    worker = Cruncher()
-    await worker.connect(db_config, queue_db)
-    await worker.setup()
-    await worker.run(batchlimit=1)
-
-
-logging.basicConfig(level=logging.DEBUG)
-
-loop = asyncio.get_event_loop()
-loop.run_until_complete(startup())
diff --git a/joblib b/joblib
deleted file mode 160000
-Subproject 7ced841aa44a7dd7bd2d3a68ceb989ef488ce44
diff --git a/package.json b/package.json
new file mode 100644
index 0000000..b3db938
--- /dev/null
+++ b/package.json
@@ -0,0 +1,19 @@
+{
+  "name": "cruncher",
+  "version": "2.0.0",
+  "description": "",
+  "main": "worker.js",
+  "dependencies": {
+    "amqplib": "^0.5.1",
+    "mysql": "^2.13.0",
+    "object-hash": "^1.1.7",
+    "sequelize": "^3.30.4",
+    "sleep-promise": "^2.0.0"
+  },
+  "devDependencies": {},
+  "scripts": {
+    "test": "echo \"Error: no test specified\" && exit 1"
+  },
+  "author": "schneefux",
+  "license": "UNLICENSED"
+}
diff --git a/requirements.txt b/requirements.txt
deleted file mode 100644
index 2e1ea39..0000000
--- a/requirements.txt
+++ /dev/null
@@ -1,5 +0,0 @@
-appdirs==1.4.2
-asyncpg==0.8.4
-packaging==16.8
-pyparsing==2.1.10
-six==1.10.0
diff --git a/worker.js b/worker.js
new file mode 100644
index 0000000..1f1263e
--- /dev/null
+++ b/worker.js
@@ -0,0 +1,154 @@
+#!/usr/bin/node
+/* jshint esnext:true */
+"use strict";
+
+var amqp = require("amqplib"),
+    Seq = require("sequelize"),
+    sleep = require("sleep-promise"),
+    hash = require("object-hash");
+
+var RABBITMQ_URI = process.env.RABBITMQ_URI,
+    DATABASE_URI = process.env.DATABASE_URI;
+
+(async () => {
+    let seq, model, rabbit, ch;
+
+    while (true) {
+        try {
+            seq = new Seq(DATABASE_URI),
+            rabbit = await amqp.connect(RABBITMQ_URI),
+            ch = await rabbit.createChannel();
+            await ch.assertQueue("crunch", {durable: true});
+            break;
+        } catch (err) {
+            console.error(err);
+            await sleep(5000);
+        }
+    }
+    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);
+
+    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; };
+        }
+
+        try {
+            await action;
+            console.log("acking");
+            await ch.ack(msg);
+        } catch (err) {
+            console.error(err);
+            await ch.nack(msg, false, false);  // nack and do not requeue
+        }
+    }, { noAck: false });
+
+    async function calculateHeroStats(dimension_id, filter) {
+        let win_rate, pick_rate, gold_per_min;
+
+        // 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,
+            include: [ {
+                model: model.Roster,
+                attributes: [],
+                include: [ {
+                    model: model.Match,
+                    attributes: []
+                } ]
+            }, {
+                model: model.Heros,
+                attributes: []
+            } ]
+        });
+
+        // TODO (workaround) filter unmapped actor<->hero
+        filter["$hero.id$"] = { $ne: null };
+        let data = await model.Participant.findAll({
+            where: filter,
+            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" ]
+            ],
+            group: q("hero.name"),
+            include: [ {
+                model: model.Roster,
+                attributes: [],
+                include: [ {
+                    model: model.Match,
+                    attributes: []
+                } ]
+            }, {
+                model: model.Heros,
+                attributes: []
+            } ]
+        });
+        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 });
+            }))
+        );
+    }
+})();