path: root/worker.js

#!/usr/bin/node
/* jshint esnext:true */
"use strict";

const amqp = require("amqplib"),
    Promise = require("bluebird"),
    winston = require("winston"),
    loggly = require("winston-loggly-bulk"),
    Seq = require("sequelize"),
    elasticsearch = require("elasticsearch");

const RABBITMQ_URI = process.env.RABBITMQ_URI || "amqp://localhost",
    DATABASE_URI = process.env.DATABASE_URI,
    ELASTICSEARCH_URI = process.env.ELASTICSEARCH_URI || "localhost:9200",
    QUEUE = process.env.QUEUE || "reap",
    INDEX = process.env.INDEX || "phase",
    JOIN_PLAYER = process.env.JOIN_PLAYER == "true",  // include player records
    SPLIT_INDEX = process.env.SPLIT_INDEX != "false",  // split index into index_seriesName
    LOGGLY_TOKEN = process.env.LOGGLY_TOKEN,
    BATCHSIZE = parseInt(process.env.BATCHSIZE) || 10,
    IDLE_TIMEOUT = parseInt(process.env.IDLE_TIMEOUT) || 1000,  // ms
    MAXCONNS = parseInt(process.env.MAXCONNS) || 20;

const logger = new (winston.Logger)({
    transports: [
        new (winston.transports.Console)({
            timestamp: true,
            colorize: true
        })
    ]
});

// loggly integration
if (LOGGLY_TOKEN)
    logger.add(winston.transports.Loggly, {
        inputToken: LOGGLY_TOKEN,
        subdomain: "kvahuja",
        tags: ["backend", "reaper", QUEUE],
        json: true
    });


amqp.connect(RABBITMQ_URI).then(async (rabbit) => {
    process.on("SIGINT", () => {
        rabbit.close();
        process.exit();
    });

    // connect to rabbit & db
    const seq = new Seq(DATABASE_URI, {
            logging: false,
            pool: {
                max: MAXCONNS
            }
        }),
        elastic = new elasticsearch.Client({ host: ELASTICSEARCH_URI, log: "info" });

    const ch = await rabbit.createChannel();
    await ch.assertQueue(QUEUE, { durable: true });
    await ch.assertQueue(QUEUE + "_failed", { durable: true });
    await ch.prefetch(BATCHSIZE);

    logger.info("configuration", {
        QUEUE, BATCHSIZE, MAXCONNS, IDLE_TIMEOUT
    });

    const model = require("../orm/model")(seq, Seq);

    let phase_data = new Set();
    let msg_buffer = new Set();
    let idle_timer = undefined;

    ch.consume(QUEUE, async (msg) => {
        const payload = JSON.parse(msg.content);
        phase_data.add(payload);
        msg_buffer.add(msg);

        // timeout after last job
        if (idle_timer != undefined)
            clearTimeout(idle_timer);
        idle_timer = setTimeout(tryProcess, IDLE_TIMEOUT);
        if (phase_data.size == BATCHSIZE)
            await tryProcess();
    }, { noAck: false });

    // wrap process() in message handler
    async function tryProcess() {
        const msgs = new Set(msg_buffer);
        msg_buffer.clear();

        logger.info("processing batch");

        // clean up to allow reaper to accept while we wait for db
        clearTimeout(idle_timer);
        idle_timer = undefined;

        const phase_objects = new Set(phase_data);
        phase_data.clear();

        try {
            await reap(phase_objects);

            logger.info("acking batch", { size: msgs.size });
            await Promise.map(msgs, async (m) => await ch.ack(m));
        } catch (err) {
            // log, move to error queue and NACK
            logger.error(err);
            await Promise.map(msgs, async (m) => {
                await ch.sendToQueue(QUEUE + "_failed",
                    m.content, { persistent: true });
                await ch.nack(m, false, false);
            });
        }
    }

    async function reap(phase_objects) {
        const db_profiler = logger.startTimer(),
            data = [].concat(... await Promise.map(phase_objects, async (po) => {
                // TODO would be better to get all in bulk using the id
                let phases = await model.ParticipantPhases.findAll({
                    attributes: [
                        "id", "created_at", "updated_at", "start", "end",
                        "participant_api_id",
                        "kills", "deaths", "assists", /* "farm" ,*/
                        "minion_kills", "jungle_kills",
                        "non_jungle_minion_kills",
                        "crystal_mine_captures", "gold_mine_captures",
                        "kraken_captures", "turret_captures",
                        /*"gold",*/
                        "dmg_true_hero", "dmg_true_kraken",
                        "dmg_true_turret", "dmg_true_vain_turret",
                        "dmg_true_others",
                        "dmg_dealt_hero", "dmg_dealt_kraken",
                        "dmg_dealt_turret", "dmg_dealt_vain_turret",
                        "dmg_dealt_others",
                        "dmg_rcvd_dealt_hero", "dmg_rcvd_true_hero",
                        "dmg_rcvd_dealt_others", "dmg_rcvd_true_others",
                        "ability_a_level", "ability_b_level", "ability_c_level",
                        "hero_level",
                        /* scores, */
                        "draft_position", "ban", "pick",
                        "heal_heal_hero", "heal_healed_hero", "heal_heal_ally", "heal_healed_ally",
                        "heal_heal_other", "heal_healed_other", "heal_rcvd_heal_hero", "heal_rcvd_healed_hero",
                        "heal_rcvd_heal_ally", "heal_rcvd_healed_ally",
                        "heal_rcvd_healed_vamp",
                        "heal_rcvd_heal_other", "heal_rcvd_healed_other",
                        /*seq.fn("COLUMN_JSON", "items"),*/
                        [ seq.cast(seq.fn("COLUMN_JSON", seq.col("participant_phases.item_grants")), "char"), "item_grants" ],
                        "item_grants_inorder",
                        [ seq.cast(seq.fn("COLUMN_JSON", seq.col("participant_phases.item_sells")), "char"), "item_sells" ],
                        "ability_a_use", "ability_b_use", "ability_c_use",
                        "ability_a_damage_true", "ability_a_damage_dealt",
                        "ability_b_damage_true", "ability_b_damage_dealt",
                        "ability_c_damage_true", "ability_c_damage_dealt",
                        "ability_perk_damage_true", "ability_perk_damage_dealt",
                        "ability_aa_damage_true", "ability_aa_damage_dealt",
                        "ability_aacrit_damage_true", "ability_aacrit_damage_dealt",
                        [ seq.cast(seq.fn("COLUMN_JSON", seq.col("participant_phases.item_uses")), "char"), "item_uses" ]/*,
                        seq.fn("COLUMN_JSON", "player_damage")*/
                    ],
                    where: {
                        "$participant.match_api_id$": po.match_api_id,
                        start: po.start,
                        end: po.end
                    },
                    include: [ {
                        model: model.Participant,
                        include: [
                            model.Region,
                            model.Hero,
                            model.Series,
                            model.GameMode,
                            model.Role,

                            model.ParticipantStats,
                            {
                                model: model.ParticipantItems,
                                attributes: [
                                    "id", "shard_id", "participant_api_id",
                                    [ seq.cast(seq.fn("COLUMN_JSON", seq.col("participant->participant_items.items")), "char"), "items" ],
                                    [ seq.cast(seq.fn("COLUMN_JSON", seq.col("participant->participant_items.item_grants")), "char"), "item_grants" ],
                                    [ seq.cast(seq.fn("COLUMN_JSON", seq.col("participant->participant_items.item_uses")), "char"), "item_uses" ],
                                    [ seq.cast(seq.fn("COLUMN_JSON", seq.col("participant->participant_items.item_sells")), "char"), "item_sells" ],
                                    "surrender",
                                    "trueskill_ranked_mu", "trueskill_ranked_sigma"
                                ],
                            },
                            model.Roster,
                            model.Match
                        ].concat(JOIN_PLAYER? [ model.Player ] : [])
                    }, {
                        model: model.Hero,
                        as: "hero_ban"
                    }, {
                        model: model.Hero,
                        as: "hero_pick"
                    }
                    ],
                    raw: true
                });
                // parse dynamic columns output
                phases.forEach((data) => {
                    //data.items = JSON.parse(data.items);
                    if (data.item_grants != "")
                        data.item_grants = JSON.parse(data.item_grants);
                    else data.item_grants = {};
                    if (data.item_sells != "")
                        data.item_sells = JSON.parse(data.item_sells);
                    else data.item_sells = {};
                    if (data.item_uses != "")
                        data.item_uses = JSON.parse(data.item_uses);
                    else data.item_uses = {};

                    if (data["participant.participant_items.items"] != "")
                        data["participant.participant_items.items"] = JSON.parse(data["participant.participant_items.items"]);
                    else data["participant.participant_items.items"] = {};
                    if (data["participant.participant_items.item_grants"] != "")
                        data["participant.participant_items.item_grants"] = JSON.parse(data["participant.participant_items.item_grants"]);
                    else data["participant.participant_items.item_grants"] = {};
                    if (data["participant.participant_items.item_sells"] != "")
                        data["participant.participant_items.item_sells"] = JSON.parse(data["participant.participant_items.item_sells"]);
                    else data["participant.participant_items.item_sells"] = {};
                    if (data["participant.participant_items.item_uses"] != "")
                        data["participant.participant_items.item_uses"] = JSON.parse(data["participant.participant_items.item_uses"]);
                    else data["participant.participant_items.item_uses"] = {};
                });
                return phases;
            } ) );
        db_profiler.done("database transaction");

        if (data.length == 0) {
            logger.error("no data!");
            return;
        }

        const es_profiler = logger.startTimer();
        await elastic.bulk({
            body: [].concat(... data.map((d) => [
                { index: {
                    _index:
                        SPLIT_INDEX?
                            `${INDEX}_${d["participant.series.name"].toLowerCase().replace(/ |\./g, "_")}`
                        :
                            INDEX,
                    _type: INDEX,
                    _id: `${d.participant_api_id}@${d.start}+${d.end}`
                } },
                d
            ]) )
        });
        es_profiler.done("elastic bulk request");
    }
});

process.on("unhandledRejection", (err) => {
    logger.error(err);
    process.exit(1);  // fail hard and die
});