insert assets, squish concurrency bugs

3 files changed, 66 insertions, 26 deletions

diff --git a/model.js b/model.js
index 0ff813e..867643b 100644
--- a/model.js
+++ b/model.js
@@ -5,18 +5,24 @@ var JsonField = require("sequelize-json");
 
 module.exports = (seq, Seq) => {
     let Match = seq.import("./models/match.js"),
+        Asset = seq.import("./models/asset.js"),
         Roster = seq.import("./models/roster.js"),
+        //Team = seq.import("./models/team.js"),
         Participant = seq.import("./models/participant.js"),
-        Player = seq.import("./models/player.js");
+        ParticipantItemUse = seq.import("./models/participant_item_use.js"),
+        Player = seq.import("./models/player.js"),
+        Item = seq.import("./models/item.js");
 
-    //Match.hasMany(Asset, {as: "Assets"});
-    //Asset.belongsTo(Match);
+    Match.hasMany(Asset, {as: "Assets"});
+    Asset.belongsTo(Match);
     Roster.belongsTo(Match, { foreignKey: "match_api_id", targetKey: "api_id" });
     //Roster.hasOne(Team, {as: "Team"});
     //Team.belongsTo(Roster);
     Participant.belongsTo(Roster, { foreignKey: "roster_api_id", targetKey: "api_id" });
     Participant.hasOne(Player, { foreignKey: "player_api_id", targetKey: "api_id" });
+    ParticipantItemUse.belongsTo(Participant, { foreignKey: "participant_api_id", targetKey: "api_id" });
+    ParticipantItemUse.hasOne(Item);
 
     //return {Match, Roster, Team, Participant, Player, Asset};
-    return {Match, Roster, Participant, Player};
+    return {Match, Roster, Participant, ParticipantItemUse, Player, Asset, Item};
 };
diff --git a/models/asset.js b/models/asset.js
index 54a9a82..af8d040 100644
--- a/models/asset.js
+++ b/models/asset.js
@@ -10,9 +10,34 @@ module.exports = function(sequelize, DataTypes) {
     },
     api_id: {
       type: DataTypes.STRING(191),
+      allowNull: false,
+      unique: true
+    },
+    match_api_id: {
+      type: DataTypes.STRING(191),
+      allowNull: false
+    },
+    url: {
+      type: DataTypes.STRING(191),
+      allowNull: false
+    },
+    content_type: {
+      type: DataTypes.STRING(191),
+      allowNull: false
+    },
+    created_at: {
+      type: DataTypes.TIME,
+      allowNull: false
+    },
+    description: {
+      type: DataTypes.STRING(191),
+      allowNull: false
+    },
+    filename: {
+      type: DataTypes.STRING(191),
       allowNull: false
     },
-    type: {
+    name: {
       type: DataTypes.STRING(191),
       allowNull: false
     }
diff --git a/worker.js b/worker.js
index 5176189..65d6e8d 100644
--- a/worker.js
+++ b/worker.js
@@ -7,7 +7,7 @@ var amqp = require("amqplib"),
 
 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 + 2 + 3*2 + 3*2),
+    BATCHSIZE = process.env.PROCESSOR_BATCH || 50,  // matches
     IDLE_TIMEOUT = process.env.PROCESSOR_IDLETIMEOUT || 500;  // ms
 
 (async () => {
@@ -37,11 +37,11 @@ var RABBITMQ_URI = process.env.RABBITMQ_URI || "amqp://localhost",
         if (timer === undefined)
             timer = setTimeout(process, IDLE_TIMEOUT)
         if (queue.length == BATCHSIZE)
-            process();
+            await process();
     }, { noAck: false });
 
     async function process() {
-        console.log("processing batch");
+        console.log("processing batch", queue.length);
 
         // clean up to allow processor to accept while we wait for db
         let matchmsgs = queue.slice();
@@ -71,7 +71,9 @@ var RABBITMQ_URI = process.env.RABBITMQ_URI || "amqp://localhost",
         // UPSERT
         try {
             let matches = matchmsgs.map((msg) => JSON.parse(msg.content));
-            await matches.forEach(async (match) => {
+            await Promise.all(matches.map(async (match) => {
+                console.log("processing match", match.id);
+
                 // flatten jsonapi nested response into our db structure-like shape
                 match.rosters = match.rosters.map((roster) => {
                     roster.participants = roster.participants.map((participant) => {
@@ -80,6 +82,7 @@ var RABBITMQ_URI = process.env.RABBITMQ_URI || "amqp://localhost",
                     });
                     return flatten(roster);
                 });
+                match.assets = match.assets.map((asset) => flatten(asset));
                 match = flatten(match);
 
                 // upsert match
@@ -89,15 +92,15 @@ var RABBITMQ_URI = process.env.RABBITMQ_URI || "amqp://localhost",
 
                 // upsert children
                 // before, add foreign keys and other missing information (shardId)
-                await match.rosters.forEach(async (roster) => {
+                await Promise.all(match.rosters.map(async (roster) => {
                     roster.match_api_id = match.api_id;
                     roster.shard_id = match.shard_id;
 
                     await model.Roster.upsert(roster, {
-                        include: [ model.Participant, model.Team ]
+                        include: [ model.Participant/*, model.Team */]
                     });
 
-                    await roster.participants.forEach(async (participant) => {
+                    await Promise.all(roster.participants.map(async (participant) => {
                         participant.player.shard_id = participant.shard_id;
                         await model.Player.upsert(participant.player);
 
@@ -107,32 +110,38 @@ var RABBITMQ_URI = process.env.RABBITMQ_URI || "amqp://localhost",
                         await model.Participant.upsert(participant, {
                             include: [ model.Player ]
                         });
-                    });
+                    }));
 
                     if (roster.team != null) {
                         roster.team.shard_id = roster.shard_id;
                         roster.team.roster_api_id = roster.api_id;
-                        await model.Team.upsert(roster.team);
+                        /*await model.Team.upsert(roster.team);*/
                     }
-                });
+                }));
 
-                await match.assets.forEach(async (asset) => {
+                await Promise.all(match.assets.map(async (asset) => {
+                    asset.match_api_id = match.api_id;
+                    asset.shard_id = match.shard_id;
+                    asset.url = asset.uRL;  // camelcasization failed ;)
                     await model.Asset.upsert(asset);
-                });
-            });
+                }));
+            }));
 
             // COMMIT
             await transaction.commit();
-            await ch.ack(matchmsgs.pop(), true);  // ack all messages until the last
+            console.log("acking batch");
+            await Promise.all(matchmsgs.map(async (msg) => {
+                await ch.ack(msg);
+            }));
             // request child jobs, notify player
-            await matches.forEach(async (m) => {
-                await m.rosters.forEach(async (r) => {
-                    await r.participants.forEach(async (p) => {
+            await Promise.all(matches.map(async (m) => {
+                await Promise.all(m.rosters.map(async (r) => {
+                    await Promise.all(r.participants.map(async (p) => {
                         await ch.publish("amq.topic", p.player.name, new Buffer("process_commit"));
-                    });
-                });
-            });
-        } catch (err) {  // TODO catch only SQL error, also catch errors in the forEach
+                    }));
+                }));
+            }));
+        } catch (err) {  // TODO catch only SQL error, also catch errors in the promises
             console.error(err);
             await ch.nack(matchmsgs.pop(), true, true);  // nack all messages until the last and requeue
             // TODO don't requeue broken records