rewrite in node

9 files changed, 231 insertions, 221 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/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 fbe2cef..0000000
--- a/api.py
+++ /dev/null
@@ -1,92 +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 "vaindock_postgres_raw",
-    "port": os.environ.get("POSTGRESQL_SOURCE_PORT") or 5532,
-    "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 "vaindock_postgres_web",
-    "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 Compiler(joblib.worker.Worker):
-    def __init__(self):
-        self._con = None
-        self._queries = {}
-        super().__init__(jobtype="compile")
-
-    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):
-        """Initialize the database."""
-        scriptroot = os.path.realpath(
-            os.path.join(os.getcwd(), os.path.dirname(__file__)))
-        for path in glob.glob(scriptroot + "/queries/*/*.sql"):
-            # utf-8-sig is used by pgadmin, doesn't hurt to specify
-            # directory names: web target table
-            table = os.path.basename(os.path.dirname(path))
-            with open(path, "r", encoding="utf-8-sig") as file:
-                try:
-                    self._queries[table].append(file.read())
-                except KeyError:
-                    self._queries[table] = [file.read()]
-                logging.info("loaded query '%s'", table)
-
-
-    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."""
-        object_id = payload["id"]
-        table = payload["type"]
-        if table not in self._queries:
-            return
-        logging.debug("%s: compiling '%s' from '%s'",
-                      jobid, object_id, table)
-        tasks = []
-        for query in self._queries[table]:
-            tasks.append(asyncio.ensure_future(
-                self._con.execute(query, object_id)))
-        await asyncio.gather(*tasks)
-
-
-async def startup():
-    worker = Compiler()
-    await worker.connect(db_config, queue_db)
-    await worker.setup()
-    await worker.run(batchlimit=5000)
-
-
-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..0f84dca
--- /dev/null
+++ b/package.json
@@ -0,0 +1,17 @@
+{
+  "name": "compiler",
+  "version": "2.0.0",
+  "description": "",
+  "main": "model.js",
+  "dependencies": {
+    "amqplib": "^0.5.1",
+    "mysql": "^2.13.0",
+    "sequelize": "^3.30.4"
+  },
+  "devDependencies": {},
+  "scripts": {
+    "test": "echo \"Error: no test specified\" && exit 1"
+  },
+  "author": "schneefux",
+  "license": "UNLICENSED"
+}
diff --git a/queries/participant/stats.sql b/queries/participant/stats.sql
deleted file mode 100644
index fe4025f..0000000
--- a/queries/participant/stats.sql
+++ /dev/null
@@ -1,21 +0,0 @@
-WITH stats AS (
-    SELECT
-        participant.api_id,
-        match.patch_version,
-        CASE WHEN roster.hero_kills=0
-        THEN 0
-        ELSE (kills+assists)/roster.hero_kills::float
-        END AS kills_participation,
-        CASE WHEN participant.deaths=0
-        THEN 0
-        ELSE (kills+assists)/deaths::float
-        END AS kda
-    FROM participant
-    JOIN roster ON roster.api_id=participant.roster_api_id
-    JOIN match ON match.api_id=roster.match_api_id
-    WHERE participant.api_id = $1
-)
-INSERT INTO participant_stats(participant_api_id, patch_version, kills_participation, kda)
-SELECT * FROM stats
-ON CONFLICT(participant_api_id) DO
-UPDATE SET(participant_api_id, patch_version, kills_participation, kda) = (SELECT * FROM stats)
diff --git a/queries/player/count_player_matches.sql b/queries/player/count_player_matches.sql
deleted file mode 100644
index 226cacc..0000000
--- a/queries/player/count_player_matches.sql
+++ /dev/null
@@ -1,29 +0,0 @@
-with stats as (
-with plr as (
-select 
-    match.patch_version as patch_version,
-    player.api_id as player_api_id,
-    match.game_mode as game_mode,
-    participant.winner as winner
-from
-    match, roster, participant, player
-where
-    match.api_id = roster.match_api_id and
-    roster.api_id = participant.roster_api_id and
-    participant.player_api_id = player.api_id and
-    player.api_id = $1)
-select
-    (select distinct(player_api_id) from plr),
-    (select distinct(patch_version) from plr),
-    (select count(*) as played from plr), 
-    (select count(*) as wins from plr where winner = true), 
-    (select count(*) as casual_played from plr where game_mode like '%casual%'),
-    (select count(*) as casual_wins from plr where game_mode like '%casual%' and winner = true),
-    (select count(*) as ranked_played from plr where game_mode like '%ranked%'),
-    (select count(*) as ranked_wins from plr where game_mode like '%ranked%' and winner = true),
-    (select 0 as streak)
-)
-INSERT INTO player_stats(player_api_id, patch_version, played, wins, casual_played, casual_wins, ranked_played, ranked_wins, streak)
-    SELECT * FROM stats
-ON CONFLICT(player_api_id) DO
-    UPDATE SET (player_api_id, patch_version, played, wins, casual_played, casual_wins, ranked_played, ranked_wins, streak) = (SELECT * FROM stats)
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..6d05e24
--- /dev/null
+++ b/worker.js
@@ -0,0 +1,163 @@
+#!/usr/bin/node
+/* jshint esnext:true */
+'use strict';
+
+var amqp = require("amqplib"),
+    Seq = require("sequelize");
+
+var RABBITMQ_URI = process.env.RABBITMQ_URI || "amqp://localhost",
+    DATABASE_URI = process.env.DATABASE_URI || "sqlite:///db.sqlite",
+    BATCHSIZE = process.env.PROCESSOR_BATCH || 50 * (1 + 5),  // matches + players + teams
+    IDLE_TIMEOUT = process.env.PROCESSOR_IDLETIMEOUT || 500;  // ms
+
+(async () => {
+    let seq = new Seq(DATABASE_URI),
+        model = require("../orm/model")(seq, Seq),
+        rabbit = await amqp.connect(RABBITMQ_URI),
+        ch = await rabbit.createChannel();
+
+    let queue = [],
+        timer = undefined;
+
+    await seq.sync();
+
+    await ch.assertQueue("compile", {durable: true});
+    // as long as the queue is filled, msg are not ACKed
+    // server sends as long as there are less than `prefetch` unACKed
+    await ch.prefetch(BATCHSIZE);
+
+    ch.consume("compile", async (msg) => {
+        queue.push(msg);
+
+        // fill queue until batchsize or idle
+        if (timer === undefined)
+            timer = setTimeout(process, IDLE_TIMEOUT)
+        if (queue.length == BATCHSIZE)
+            await process();
+    }, { noAck: false });
+
+    async function process() {
+        console.log("compiling batch", queue.length);
+
+        // clean up to allow processor to accept while we wait for db
+        let msgs = queue.slice();
+        queue = [];
+        clearTimeout(timer);
+        timer = undefined;
+
+        // BEGIN
+        let transaction = await seq.transaction({ autocommit: false });
+
+        // UPSERT
+        try {
+            // processor sends to queue with a custom "type" so compiler can filter
+            // m.content: player.api_id
+            let players      = msgs.filter((m) => m.properties.type == "player").map((m) => JSON.parse(m.content)),
+                participants = msgs.filter((m) => m.properties.type == "participant").map((m) => JSON.parse(m.content));
+
+            await Promise.all(players.map(async (player) => {
+                let player_api_id = player.api_id,
+                    player_ext = {};
+
+                player_ext.player_api_id = player_api_id;
+                //player_ext.series = ""
+
+                // TODO parallelize
+
+                player_ext.played = await model.Participant.count({
+                    where: {
+                        player_api_id: player_api_id
+                    }
+                });
+                player_ext.wins = await model.Participant.count({
+                    where: {
+                        player_api_id: player_api_id,
+                        winner: true
+                    }
+                });
+
+                // TODO maybe this can be done in fewer/combined/subqueries
+                let count_matches_where = async (where) => {
+                    return (await model.Participant.findOne({
+                        where: where,
+                        attributes: [[seq.fn("COUNT", "$roster.match$"), "count"]],
+                        include: [ {
+                            model: model.Roster,
+                            attributes: [],
+                            include: [ {
+                                model: model.Match,
+                                attributes: []
+                            } ]
+                        } ]
+                    })).get("count");
+                };
+                player_ext.played_casual = await count_matches_where({
+                    player_api_id: player_api_id,
+                    "$roster.match.game_mode$": "casual"
+                });
+                player_ext.played_ranked = await count_matches_where({
+                    player_api_id: player_api_id,
+                    "$roster.match.game_mode$": "ranked"
+                });
+                player_ext.wins_casual = await count_matches_where({
+                    player_api_id: player_api_id,
+                    winner: true,
+                    "$roster.match.game_mode$": "casual"
+                });
+                player_ext.wins_ranked = await count_matches_where({
+                    player_api_id: player_api_id,
+                    winner: true,
+                    "$roster.match.game_mode$": "ranked"
+                });
+
+                await model.PlayerExt.upsert(player_ext, {
+                    include: [ model.Participant ],
+                    transaction: transaction
+                });
+            }));
+            await Promise.all(participants.map(async (api_participant) => {
+                let participant = await model.Participant.findOne({
+                    where: {
+                        api_id: api_participant.api_id
+                    },
+                    attributes: ["api_id", "kills", "assists", "deaths", seq.col("roster.hero_kills")],
+                    include: [
+                        model.Roster
+                    ]
+                }),
+                    participant_ext = {};
+
+                participant_ext.participant_api_id = participant.api_id;
+                participant_ext.series = ""  // TODO rm
+
+                if (participant.roster.hero_kills == 0)
+                    participant_ext.kills_participation = 0;
+                else
+                    participant_ext.kills_participation = (participant.kills + participant.assists) / participant.roster.hero_kills;
+
+                if (participant.deaths == 0)
+                    participant_ext.kda = 0;
+                else
+                    participant_ext.kda = (participant.kills + participant.assists) / participant.deaths;
+
+                await model.ParticipantExt.upsert(participant_ext, {
+                    include: [ model.Participant ],
+                    transaction: transaction
+                });
+            }));
+
+            // COMMIT
+            await transaction.commit();
+            console.log("acking batch");
+            await ch.ack(msgs.pop(), true);  // ack all messages until the last
+
+            // notify web
+            await Promise.all(players.map(async (p) => await ch.publish("amq.topic", p.name, new Buffer("compile_commit")) ));
+        } catch (err) {  // TODO catch only SQL error, also catch errors in the promises
+            console.error(err);
+            await transaction.rollback();
+            await ch.nack(msgs.pop(), true, true);  // nack all messages until the last and requeue
+            // TODO don't requeue broken records
+        }
+    }
+})();