implement league fetcher, use async for match insert flow

sql/create_tables.sql

@@ -221,3 +221,20 @@ CREATE INDEX ON match_logs(match_id, attackername_slot) WHERE attackername_slot
 CREATE INDEX ON match_logs(match_id, targetname_slot) WHERE targetname_slot IS NOT NULL;
 CREATE INDEX ON match_logs(match_id, sourcename_slot) WHERE sourcename_slot IS NOT NULL;
 CREATE INDEX ON match_logs(match_id, targetsourcename_slot) WHERE targetsourcename_slot IS NOT NULL;
+
+CREATE TABLE picks_bans(
+  match_id bigint REFERENCES matches(match_id) ON DELETE CASCADE,
+  is_pick boolean,
+  hero_id int,
+  team smallint,
+  order smallint,
+  PRIMARY KEY (match_id, order)
+);
+
+CREATE TABLE leagues(
+  leagueid bigint PRIMARY KEY,
+  ticket varchar(255),
+  banner varchar(255),
+  tier varchar(255),
+  name varchar(255)
+)

store/queries.js

@@ -191,6 +191,7 @@ function insertMatch(db, redis, match, options, cb)
     //we want to insert into matches, then insert into player_matches for each entry in players
     async.series(
     {
+        "ipm": isProMatch,
         "u": upsertMatch,
         "uc": upsertMatchCassandra,
         "uml": upsertMatchLogs,
@@ -206,23 +207,38 @@ function insertMatch(db, redis, match, options, cb)
         return cb(err, results.dp);
     });
 
+    function isProMatch(cb)
+    {
+        //TODO check redis/postgres for professional/premium league
+        match.isProMatch = match.leagueid >= 0;
+        cb();
+    }
+
     function upsertMatch(cb)
     {
-        if (!config.ENABLE_POSTGRES_MATCH_STORE_WRITE && !match.leagueid)
+        if (!config.ENABLE_POSTGRES_MATCH_STORE_WRITE && !match.isProMatch)
         {
             return cb();
         }
         db.transaction(function(trx)
         {
-            upsert(trx, 'matches', match,
+            async.series(
             {
-                match_id: match.match_id
-            }, function(err)
+                "im": insertMatch,
+                "ipm": insertPlayerMatches,
+                "ipb": insertPicksBans
+            }, exit);
+
+            function insertMatch(cb)
             {
-                if (err)
+                upsert(trx, 'matches', match,
                 {
-                    return exit(err);
-                }
+                    match_id: match.match_id
+                }, cb);
+            }
+
+            function insertPlayerMatches(cb)
+            {
                 async.each(players || [], function(pm, cb)
                 {
                     pm.match_id = match.match_id;
@@ -231,30 +247,37 @@ function insertMatch(db, redis, match, options, cb)
                         match_id: pm.match_id,
                         player_slot: pm.player_slot
                     }, cb);
-                }, exit);
-                //TODO create leagues table
-                //TODO create worker task to get leagues
-                //TODO create table for picks/bans
-                //TODO insert picks/bans
-                //TODO remove dota_unknown to save space
-                //TODO figure out whehter to store expanded or raw data
-/*
-"picks_bans":[{"is_pick":false,"hero_id":41,"team":1,"order":0},{"is_pick":false,"hero_id":38,"team":0,"order":1},{"is_pick":false,"hero_id":6,"team":1,"order":2},{"is_pick":false,"hero_id":62,"team":0,"order":3},{"is_pick":true,"hero_id":29,"team":1,"order":4},{"is_pick":true,"hero_id":20,"team":0,"order":5},{"is_pick":true,"hero_id":65,"team":0,"order":6},{"is_pick":true,"hero_id":87,"team":1,"order":7},{"is_pick":false,"hero_id":93,"team":0,"order":8},{"is_pick":false,"hero_id":74,"team":1,"order":9},{"is_pick":false,"hero_id":110,"team":0,"order":10},{"is_pick":false,"hero_id":54,"team":1,"order":11},{"is_pick":true,"hero_id":18,"team":0,"order":12},{"is_pick":true,"hero_id":107,"team":1,"order":13},{"is_pick":true,"hero_id":103,"team":0,"order":14},{"is_pick":true,"hero_id":49,"team":1,"order":15},{"is_pick":false,"hero_id":63,"team":0,"order":16},{"is_pick":false,"hero_id":98,"team":1,"order":17},{"is_pick":true,"hero_id":67,"team":1,"order":18},{"is_pick":true,"hero_id":8,"team":0,"order":19}]
-*/
-                function exit(err)
+                }, cb);
+            }
+
+            function insertPicksBans(cb)
+            {
+                async.each(match.picks_bans || [], function(p, cb)
                 {
-                    if (err)
+                    //order is a reserved keyword
+                    p.ord = p.order;
+                    p.match_id = match.match_id;
+                    upsert(trx, 'picks_bans', p,
                     {
-                        console.error(err);
-                        trx.rollback(err);
-                    }
-                    else
-                    {
-                        trx.commit();
-                    }
-                    cb(err);
+                        match_id: p.match_id,
+                        ord: p.ord
+                    }, cb);
+                }, cb);
+            }
+
+            function exit(err)
+            {
+                if (err)
+                {
+                    console.error(err);
+                    trx.rollback(err);
                 }
-            });
+                else
+                {
+                    trx.commit();
+                }
+                cb(err);
+            }
         });
     }
 
@@ -462,7 +485,7 @@ function insertMatch(db, redis, match, options, cb)
                 duration: match.duration,
                 replay_blob_key: match.replay_blob_key,
                 pgroup: match.pgroup,
-                leagueid: match.leagueid,
+                isProMatch: match.isProMatch,
             },
             {
                 lifo: options.lifo,

svc/parser.js

@@ -267,7 +267,7 @@ function runParse(match, job, cb)
                 var message = "time spent on post-processing match ";
                 console.time(message);
                 var meta = processMetadata(entries);
-                var logs = match.leagueid ? processReduce(entries, match, meta) : undefined;
+                var logs = match.isProMatch ? processReduce(entries, match, meta) : undefined;
                 var res = processExpand(entries, meta);
                 var parsed_data = processParsedData(res.parsed_data);
                 var teamfights = processTeamfights(res.tf_data, meta);

svc/scanner.js

@@ -112,12 +112,7 @@ function start()
 
             function processMatch(match, cb)
             {
-                if (config.ENABLE_PRO_PARSING && match.leagueid)
-                {
-                    //parse tournament games
-                    match.parse_status = 0;
-                }
-                else if (match.players.some(function(p)
+                if (match.players.some(function(p)
                     {
                         return (p.account_id in trackedPlayers);
                     }))

svc/worker.js

@@ -124,6 +124,36 @@ invokeInterval(function notablePlayers(cb)
         }, cb);
     });
 }, 10 * 60 * 1000);
+invokeInterval(function leagues(cb)
+{
+    utility.getData('https://raw.githubusercontent.com/dotabuff/d2vpkr/master/dota/scripts/items/leagues.json', function(err, leagues)
+    {
+        if (err)
+        {
+            return cb(err);
+        }
+        utility.getData('https://raw.githubusercontent.com/dotabuff/d2vpkr/master/dota/scripts/items/items_game.json', function(err, items)
+        {
+            if (err)
+            {
+                return cb(err);
+            }
+            var arr = [];
+            for (var key in leagues)
+            {
+                arr.push(
+                {
+                    leagueid: key,
+                    ticket: leagues[key].ticket,
+                    banner: leagues[key].banner,
+                    tier: leagues[key].tier,
+                    name: items.items_game.items[key].name
+                });
+            }
+            db('leagues').insert(arr).asCallback(cb);
+        });
+    });
+}, 10 * 60 * 1000);
 
 function invokeInterval(func, delay)
 {