diff options
Diffstat (limited to 'worker.js')
| -rw-r--r-- | worker.js | 351 |
1 files changed, 276 insertions, 75 deletions
@@ -4,22 +4,24 @@ var amqp = require("amqplib"), Seq = require("sequelize"), + snakeCaseKeys = require("snakecase-keys"), + item_name_map = require("../orm/items"), sleep = require("sleep-promise"); -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 - IDLE_TIMEOUT = process.env.PROCESSOR_IDLETIMEOUT || 500; // ms +var RABBITMQ_URI = process.env.RABBITMQ_URI, + DATABASE_URI = process.env.DATABASE_URI, + BATCHSIZE = parseInt(process.env.PROCESSOR_BATCH) || 50 * (1 + 5), // matches + players + IDLE_TIMEOUT = parseFloat(process.env.PROCESSOR_IDLETIMEOUT) || 500; // ms (async () => { let seq, model, rabbit, ch; while (true) { try { - seq = new Seq(DATABASE_URI, { logging: () => {} }), - rabbit = await amqp.connect(RABBITMQ_URI), + seq = new Seq(DATABASE_URI, { logging: () => {} }); + rabbit = await amqp.connect(RABBITMQ_URI); ch = await rabbit.createChannel(); - await ch.assertQueue("compile", {durable: true}); + await ch.assertQueue("process", {durable: true}); await ch.assertQueue("analyze", {durable: true}); break; } catch (err) { @@ -27,47 +29,255 @@ var RABBITMQ_URI = process.env.RABBITMQ_URI || "amqp://localhost", await sleep(5000); } } + model = require("../orm/model")(seq, Seq); let queue = [], timer = undefined; + let item_db_map = {}, // "Halcyon Potion" to id + hero_db_map = {}, // "*SAW*" to id + series_db_map = {}, // date to series id + game_mode_db_map = {}, // "ranked" to id + role_db_map = {}; // "captain" to id + + /* recreate for debugging + await seq.query("SET FOREIGN_KEY_CHECKS=0"); + await seq.sync({force: true}); + */ await seq.sync(); + // TODO instead of object, use Map + await Promise.all([ + model.Item.findAll() + .map((item) => item_db_map[item.name] = item.id), + model.Hero.findAll() + .map((hero) => hero_db_map[hero.name] = hero.id), + model.Series.findAll() + .map((series) => series_db_map[series.name] = series.id), + model.GameMode.findAll() + .map((mode) => game_mode_db_map[mode.name] = mode.id), + model.Role.findAll() + .map((role) => role_db_map[role.name] = role.id) + ]); // 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) => { + ch.consume("process", async (msg) => { queue.push(msg); // fill queue until batchsize or idle if (timer === undefined) - timer = setTimeout(process, IDLE_TIMEOUT) + timer = setTimeout(process, IDLE_TIMEOUT); if (queue.length == BATCHSIZE) await process(); }, { noAck: false }); async function process() { - console.log("compiling batch", queue.length); + console.log("processing batch", queue.length); // clean up to allow processor to accept while we wait for db - let msgs = queue.slice(); - queue = []; clearTimeout(timer); timer = undefined; + let msgs = queue; + queue = []; + + // helper to convert API response into flat JSON + // db structure is (almost) 1:1 the API structure + // so we can insert the flat API response as-is + function flatten(obj) { + let attrs = obj.attributes || {}, + stats = attrs.stats || {}, + o = Object.assign({}, obj, attrs, stats); + o.api_id = o.id; // rename + delete o.id; + delete o.type; + delete o.attributes; + delete o.stats; + delete o.relationships; + return o; + } + + // helper: true if object is not in record arr + // also, sort out invalid objects (undefined or null) + let is_in = (arr, obj) => obj == undefined || + arr.map((o) => o.api_id).indexOf(obj.api_id) > -1; + - // aggregate & bulk insert - let participant_stats_records = [], - player_updates = []; // [[what, where]] + // we aggregate record objects to do a bulk insert + let match_records = [], + roster_records = [], + participant_records = [], + participant_stats_records = [], + player_records = [], + asset_records = [], + participant_item_records = []; + + // populate `_records` + // data from `/player` + msgs.filter((m) => m.properties.type == "player").map((msg) => { + let players = JSON.parse(msg.content); + players.map((p) => { + let player = flatten(p); + console.log("processing player", player.name); + if (!is_in(player_records, player)) + player_records.push(player); + }); + }); - // 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)).filter((p) => p != undefined), - participants = msgs.filter((m) => m.properties.type == "participant").map( - (m) => JSON.parse(m.content)).filter((p) => p != undefined); + // data from `/matches` + msgs.filter((m) => m.properties.type == "match").map((msg) => { + let match = JSON.parse(msg.content); + console.log("processing match", match.id); + // flatten jsonapi nested response into our db structure-like shape + // also, push missing fields and snakecasify + match.rosters = match.rosters.map((roster) => { + roster.matchApiId = match.id; + roster.attributes.shardId = match.attributes.shardId; + roster.createdAt = match.createdAt; + + roster.participants = roster.participants.map((participant) => { + participant.attributes.shardId = roster.attributes.shardId; + participant.rosterApiId = roster.id; + participant.createdAt = roster.createdAt; + participant.playerApiId = participant.player.id; + + // API bug fixes + // items on AFK is `null` not `{}` + participant.attributes.stats.itemGrants = participant.attributes.stats.itemGrants || {}; + participant.attributes.stats.itemSells = participant.attributes.stats.itemSells || {}; + participant.attributes.stats.itemUses = participant.attributes.stats.itemUses || {}; + // jungle_kills is `null` in BR + participant.attributes.stats.jungleKills = participant.attributes.stats.jungleKills || 0; + + // map items: names/id -> name -> db + let itms = [], + item_use = (arr, action) => + arr.map((item) => { return { + participant_api_id: participant.id, + item_id: item_db_map[item_name_map[item]], + action: action + } }), + item_arr_from_obj = (obj) => + [].concat(... // 3 flatten + Object.entries(obj).map( // 1 map over (key, value) + (tuple) => Array(tuple[1]).fill(tuple[0]))) // 2 create Array [key] * value + + itms = itms.concat(item_use(participant.attributes.stats.items, "final")); + itms = itms.concat(item_use(item_arr_from_obj(participant.attributes.stats.itemGrants), "grant")); + itms = itms.concat(item_use(item_arr_from_obj(participant.attributes.stats.itemUses), "use")); + itms = itms.concat(item_use(item_arr_from_obj(participant.attributes.stats.itemSells), "sell")); + + // for debugging: + let items_missing_name = [].concat(... + Object.keys(participant.attributes.stats.itemGrants), + participant.attributes.stats.items) + .filter((i) => Object.keys(item_name_map).indexOf(i) == -1); + if (items_missing_name.length > 0) console.error("missing API name -> name mapping for", items_missing_name); + + let items_missing_db = [].concat(... + Object.keys(participant.attributes.stats.itemGrants), + participant.attributes.stats.items) + .filter((i) => Object.keys(item_db_map).indexOf(item_name_map[i]) == -1); + if (items_missing_db.length > 0) console.error("missing name -> DB ID mapping for", items_missing_db); + + // redefine participant.items for our custom map + participant.attributes.stats.participant_items = itms; + + participant.player.attributes.shardId = participant.attributes.shardId; + participant.player = flatten(participant.player); + return flatten(participant); + }); + return flatten(roster); + }); + match.assets = match.assets.map((asset) => { + asset.matchApiId = match.id; + asset.attributes.shardId = match.attributes.shardId; + return flatten(asset); + }); + match = snakeCaseKeys(flatten(match)); + + // after conversion, create the array of records + // there is a low chance of a match being duplicated in a batch, + // skip it and its children + if (!is_in(match_records, match)) { + match_records.push(match); + match.rosters.map((r) => { + roster_records.push(r); + r.participants.map((p) => { + let p_pstats = calculate_participant_stats(match, r, p); + // participant gets split into participant and p_stats + participant_records.push(p_pstats[0]); + participant_stats_records.push(p_pstats[1]); + // deduplicate player + // in a batch, it is very likely that players are duplicated + // so this improves performance a bit + if (!is_in(player_records, p.player)) player_records.push(p.player); + p.participant_items.map((i) => { + participant_item_records.push(i); + }); + }); + }); + match.assets.map((a) => { + asset_records.push(a); + }); + } + }); + + // now access db + try { + console.log("inserting batch into db"); + // upsert whole batch in parallel + await seq.transaction({ autocommit: false }, async (transaction) => { + await Promise.all([ + model.Match.bulkCreate(match_records, { + include: [ model.Roster, model.Asset ], + updateOnDuplicate: [], // all + transaction: transaction + }), + model.Roster.bulkCreate(roster_records, { + include: [ model.Roster ], + updateOnDuplicate: [], + transaction: transaction + }), + model.Participant.bulkCreate(participant_records, { + include: [ model.Player ], + updateOnDuplicate: [], + transaction: transaction + }), + model.ParticipantStats.bulkCreate(participant_stats_records, { + include: [ model.Participant ], + updateOnDuplicate: [], + transaction: transaction + }), + model.Player.bulkCreate(player_records, { + updateOnDuplicate: [], + transaction: transaction + }), + model.ItemParticipant.bulkCreate(participant_item_records, { + include: [ model.Participant ], + updateOnDuplicate: [], + transaction: transaction + }), + model.Asset.bulkCreate(asset_records, { + updateOnDuplicate: [], + transaction: transaction + }) + ]); + }); + await Promise.all(msgs.map((m) => ch.ack(m)) ); + } catch (err) { + // this should only happen for Deadlocks in prod + // it *must not* fail due to broken schema or missing dependency + // TODO: eliminate such cases earlier in the chain + // and immediately NACK those broken matches, requeueing only the rest + console.error(err); + await Promise.all(msgs.map((m) => ch.nack(m, true)) ); // requeue + return; // give up + } + /* // collect information and populate _record arrays await Promise.all([ // collect player information @@ -99,25 +309,6 @@ var RABBITMQ_URI = process.env.RABBITMQ_URI || "amqp://localhost", { where: { api_id: player_api_id } } ]); })), - // calculate participant fields - Promise.all(participants.map(async (api_participant) => { - participant_stats_records.push(calculate_stats( - await model.Participant.findOne({ - where: { api_id: api_participant.api_id }, - include: [ { - model: model.Roster, - include: [ - model.Match - ] - }, { - model: model.ItemParticipant, - as: "items", - include: [ model.Item ] - } - ] - }) - )) - })) ]); // load records into db @@ -125,22 +316,17 @@ var RABBITMQ_URI = process.env.RABBITMQ_URI || "amqp://localhost", console.log("inserting batch into db"); await seq.transaction({ autocommit: false }, async (transaction) => { await Promise.all([ - model.ParticipantStats.bulkCreate(participant_stats_records, { - updateOnDuplicate: [], // all - transaction: transaction - }), player_updates.map(async (pu) => await model.Player.update(pu[0], pu[1])) ]); }); - } catch (err) { - // this should only happen for deadlocks or non-data related issues - console.error(err); - await Promise.all(msgs.map((m) => ch.nack(m, true)) ); // requeue - return; // give up - } - console.log("acking batch"); - await Promise.all(msgs.map((m) => ch.ack(m)) ); + */ + + // notify web + await Promise.all(player_records.map(async (p) => + await ch.publish("amq.topic", "player." + p.name, new Buffer("matches_update")) )); + if (match_records.length > 0) + await ch.publish("amq.topic", "global", new Buffer("matches_update")); // notify analyzer Promise.all(participant_stats_records.map(async (p) => @@ -149,36 +335,51 @@ var RABBITMQ_URI = process.env.RABBITMQ_URI || "amqp://localhost", type: "participant" }) )); - - // notify web - await Promise.all([ - Promise.all(players.map(async (p) => await ch.publish("amq.topic", "player." + p.name, - new Buffer("stats_update")) )), - Promise.all(participant_stats_records.map(async (p) => await ch.publish("amq.topic", - "participant." + p.participant_api_id, new Buffer("stats_update")) )) - ]); } - // based on the participant db record from the end of the match, - // calculate a participant_stats record and return it - function calculate_stats(participant) { - if (participant == null) { console.error("got nonexisting participant!"); return; } - let participant_stats = {}; - participant_stats.participant_api_id = participant.get("api_id"); + // Split participant API data into participant and participant_stats + // Should not need to query db here. + function calculate_participant_stats(match, roster, participant) { + let p_s = {}, // participant_stats_record + p = {}; // participant_record - participant_stats.kills = participant.get("kills"); + // meta + p_s.participant_api_id = participant.api_id; + p_s.final = true; // these are the stats at the end of the match + p_s.updated_at = new Date(); + p_s.created_at = new Date(); // TODO set to match.created_at+match.duration - if (participant.roster.get("hero_kills") == 0) participant_stats.kill_participation = 0; - else participant_stats.kill_participation = participant.get("kills") / participant.roster.get("hero_kills"); + // attributes to copy from API to participant + // these don't change over the duration of the match + // (or aren't in Telemetry) + ["api_id", "shard_id", "player_api_id", "roster_api_id", + "winner", "went_afk", "first_afk_time", + "skin_key", "skill_tier", "level", + "karma_level", "actor", + "farm"].map((attr) => + p[attr] = participant[attr]); - participant_stats.sustain_score = participant.items.reduce((score, item) => { - if (item.get("action") == "final") { - if (["Eve of Harvest", "Serpent Mask"].indexOf(item.item.get("name")) != -1) - return score + 20; - } + // attributes to copy from API to participant_stats + // with Telemetry, these will be calculated in intervals + ["kills", "deaths", "assists", "minion_kills", + "jungle_kills", "non_jungle_minion_kills", + "crystal_mine_captures", "gold_mine_captures", + "kraken_captures", "turret_captures", + "gold"].map((attr) => + p_s[attr] = participant[attr]); + + // traits calculations + if (roster.hero_kills == 0) p_s.kill_participation = 0; + else p_s.kill_participation = p_s.kills / roster.hero_kills; + + p_s.sustain_score = participant.items.reduce((score, item) => { + // items[], itemGrants{}, itemUse{}, itemSells{} are the API objects + // old item names to clean names need to be mapped via `item_name_map[oldname]` + if (["Eve of Harvest", "Serpent Mask"].indexOf(item_name_map[item]) != -1) + return score + 20; return score; }, 0); - return participant_stats; + return [p, p_s]; } })(); |