support dynamic item columns

3 files changed, 95 insertions, 8 deletions

diff --git a/crunch_global.sql b/crunch_global.sql
index abd8016..8d618ce 100644
--- a/crunch_global.sql
+++ b/crunch_global.sql
@@ -31,9 +31,14 @@ select
     sum(p_s.turret_captures) as turret_captures,
     sum(p_s.gold) as gold,
     sum(p_s.impact_score) as impact_score,
-    sum(0) as surrender
+    sum(p_i.surrender) as surrender,
+    _p_i_items_insert,
+    _p_i_item_grants_insert,
+    _p_i_item_uses_insert,
+    _p_i_item_sells_insert
 from participant p
 join participant_stats p_s on (p_s.participant_api_id = p.api_id)
+join participant_items p_i on (p_i.participant_api_id = p.api_id)
 join filter f on (f.dimension_on = 'global' and (f.name = 'all' or f.id in (select gpf.filter_id from global_point_filters gpf where gpf.match_api_id = p.match_api_id)))
 join series s on (p_s.created_at between s.start and s.end and s.dimension_on = 'global')
 join hero h on (p.hero_id = h.id or h.name = 'all')
@@ -80,4 +85,8 @@ kraken_captures = kraken_captures + values(kraken_captures),
 turret_captures = turret_captures + values(turret_captures),
 gold = gold + values(gold),
 impact_score = impact_score + values(impact_score),
-surrender = surrender + values(surrender)
+surrender = surrender + values(surrender),
+_p_i_items_update,
+_p_i_item_grants_update,
+_p_i_item_uses_update,
+_p_i_item_sells_update
diff --git a/crunch_phases.sql b/crunch_phases.sql
index 92d9283..bc29711 100644
--- a/crunch_phases.sql
+++ b/crunch_phases.sql
@@ -49,10 +49,10 @@ select
     sum(ability_c_level),
     sum(hero_level),
 
-    0 as kda_ratio,
-    0 as kill_participation,
-    0 as cs_per_min,
-    0 as kills_per_min
+    _ph_items_insert,
+    _ph_item_grants_insert,
+    _ph_item_uses_insert,
+    _ph_item_sells_insert
 from participant_phases ph
 join participant p on ph.participant_api_id = p.api_id
 join filter f on (f.dimension_on = 'global' and (f.name = 'all' or f.id in (select gpf.filter_id from global_point_filters gpf where gpf.match_api_id = p.match_api_id)))
@@ -100,4 +100,8 @@ dmg_rcvd_true_others = dmg_rcvd_true_others + values(dmg_rcvd_true_others),
 ability_a_level = ability_a_level + values(ability_a_level),
 ability_b_level = ability_b_level + values(ability_b_level),
 ability_c_level = ability_c_level + values(ability_c_level),
-hero_level = hero_level + values(hero_level)
+hero_level = hero_level + values(hero_level),
+_ph_items_update,
+_ph_item_grants_update,
+_ph_item_uses_update,
+_ph_item_sells_update
diff --git a/worker.js b/worker.js
index 411958f..ba20ee9 100644
--- a/worker.js
+++ b/worker.js
@@ -63,13 +63,87 @@ amqp.connect(RABBITMQ_URI).then(async (rabbit) => {
     await ch.assertQueue(QUEUE, { durable: true });
     await ch.assertQueue(QUEUE + "_failed", { durable: true });
 
+    const model = require("../orm/model")(seq, Seq);
+
     logger.info("configuration", {
         SCRIPT, QUEUE, BATCHSIZE, MAXCONNS, LOAD_TIMEOUT
     });
 
     // load update SQL scripts; scripts use sequelize replacements
     // for the `participant_api_id` array
-    const script = fs.readFileSync(SCRIPT, "utf8");
+    let script = fs.readFileSync(SCRIPT, "utf8");
+
+    // array of all item ids
+    const items = (await model.Item.findAll()).map((i) => i.id);
+    // returns an SQL snippet like this:
+    /* column_create(
+     *      '1', column_get(p_i.item_grants, '1'),
+     *      …
+     *  )
+     */
+    const dynamic_sql = (doCreate, tableName, columnName) => {
+        if (doCreate) {  // insert
+            return `
+column_create(` +
+        (items.map((i) =>
+            // create an array of `columnName, col_get($col, columnName)`
+`   '${i}',
+    coalesce(column_get(${tableName}.${columnName}, '${i}' as int), 0)`
+        ).join(",\n")) + `
+) as ${columnName}`;
+        } else {
+            return `
+${columnName} = column_create(` +
+        (items.map((i) =>
+    // create an array of
+    // `columnName, col_get($col, columnName) + col_get(values($col, columnName))`
+`   '${i}',
+    coalesce(column_get(${columnName}, '${i}' as int), 0)
+    +
+    coalesce(column_get(values(${columnName}), '${i}' as int), 0)`
+        ).join(",\n")) + `
+)`;
+        }
+    };
+    // generate
+    const
+        p_i_items_insert = dynamic_sql(true, 'p_i', 'items'),
+        p_i_item_grants_insert = dynamic_sql(true, 'p_i', 'item_grants'),
+        p_i_item_uses_insert = dynamic_sql(true, 'p_i', 'item_uses'),
+        p_i_item_sells_insert = dynamic_sql(true, 'p_i', 'item_sells'),
+        p_i_items_update = dynamic_sql(false, 'p_i', 'items'),
+        p_i_item_grants_update = dynamic_sql(false, 'p_i', 'item_grants'),
+        p_i_item_uses_update = dynamic_sql(false, 'p_i', 'item_uses'),
+        p_i_item_sells_update = dynamic_sql(false, 'p_i', 'item_sells'),
+        ph_items_insert = dynamic_sql(true, 'ph', 'items'),
+        ph_item_grants_insert = dynamic_sql(true, 'ph', 'item_grants'),
+        ph_item_uses_insert = dynamic_sql(true, 'ph', 'item_uses'),
+        ph_item_sells_insert = dynamic_sql(true, 'ph', 'item_sells'),
+        ph_items_update = dynamic_sql(false, 'ph', 'items'),
+        ph_item_grants_update = dynamic_sql(false, 'ph', 'item_grants'),
+        ph_item_uses_update = dynamic_sql(false, 'ph', 'item_uses'),
+        ph_item_sells_update = dynamic_sql(false, 'ph', 'item_sells')
+    ;
+
+    // replace stubs
+    Object.entries({
+        p_i_items_insert,
+        p_i_item_grants_insert,
+        p_i_item_uses_insert,
+        p_i_item_sells_insert,
+        p_i_items_update,
+        p_i_item_grants_update,
+        p_i_item_uses_update,
+        p_i_item_sells_update,
+        ph_items_insert,
+        ph_item_grants_insert,
+        ph_item_uses_insert,
+        ph_item_sells_insert,
+        ph_items_update,
+        ph_item_grants_update,
+        ph_item_uses_update,
+        ph_item_sells_update,
+    }).forEach(([key, value]) => script = script.replace('_' + key, value));
 
     // fill a buffer and execute an SQL on a bigger (> 1o) batch
     const participants = new Set(),