Merge 60bb983 into d3b0208

api/helpers/economy/deduct-from-player.js

@@ -71,11 +71,7 @@ module.exports = {
         economyAction: 'deduct'
       })
 
-      let actionsCompleted = await sails.helpers.redis.get(`server:${playerToDeductFrom.server}:economyActionsCompleted`);
-      if (!actionsCompleted) {
-        actionsCompleted = 1
-      }
-      await sails.helpers.redis.set(`server:${playerToDeductFrom.server}:economyActionsCompleted`, parseInt(actionsCompleted) + 1);
+      await sails.helpers.redis.incr(`server:${playerToDeductFrom.server}:economyActionsCompleted`);
       await sails.helpers.economy.deleteOldData(playerToDeductFrom.server);
 
       return exits.success();

api/helpers/economy/give-to-player.js

@@ -66,11 +66,7 @@ module.exports = {
         economyAction: 'give'
       })
 
-      let actionsCompleted = await sails.helpers.redis.get(`server:${playerToGiveTo.server}:economyActionsCompleted`);
-      if (!actionsCompleted) {
-        actionsCompleted = 1
-      }
-      await sails.helpers.redis.set(`server:${playerToGiveTo.server}:economyActionsCompleted`, parseInt(actionsCompleted) + 1);
+      await sails.helpers.redis.incr(`server:${playerToGiveTo.server}:economyActionsCompleted`);
       await sails.helpers.economy.deleteOldData(playerToGiveTo.server);
 
       return exits.success();

api/helpers/get-queue-object.js

@@ -18,8 +18,9 @@ module.exports = {
   fn: function (inputs, exits) {
     if (process.env.IS_TEST) {
       return exits.success({
-        on: function() {},
-        process: function() {},
+        on: function () { },
+        process: function () { },
+        empty: function () { },
         add: function() {}
       });
     }

api/helpers/redis/ensure-cache-exists.js

@@ -0,0 +1,28 @@
+module.exports = {
+
+
+    friendlyName: 'Ensure cache exists',
+
+
+    description: 'Makes sure the in memory cache exists',
+
+
+    inputs: {},
+
+
+    exits: {
+
+    },
+
+
+    fn: async function (inputs, exits) {
+
+        if (!sails.cache) {
+            sails.cache = {};
+        }
+
+        return exits.success();
+    }
+
+};
+

api/helpers/redis/set.js

@@ -36,47 +36,10 @@ module.exports = {
 
 
   fn: async function (inputs, exits) {
-    const datastore = sails.getDatastore('cache');
-    if (datastore.config.adapter === 'sails-redis') {
-      if (inputs.ex) {
-        if (_.isUndefined(inputs.ttl)) {
-          return exits.error(`When setting ex true you must provide a TTL.`)
-        }
-
-        sails.getDatastore('cache').leaseConnection(function during(redisConnection, proceed) {
-          redisConnection.set(inputs.keyString, inputs.value, 'EX', inputs.ttl, (err, reply) => {
-            if (err) return proceed(err);
-
-            return proceed(undefined, reply)
-          })
-        }).exec((err, result) => {
-          if (err) return exits.error(err);
-
-          return exits.success(result);
-        })
-
-      } else {
-
-        sails.getDatastore('cache').leaseConnection(function during(redisConnection, proceed) {
-          redisConnection.set(inputs.keyString, inputs.value, (err, reply) => {
-            if (err) return proceed(err);
-
-            return proceed(undefined, reply)
-          })
-        }).exec((err, result) => {
-          if (err) return exits.error(err);
-
-          return exits.success(result);
-        })
-      }
-    } else {
-      if (!sails.cache) {
-        sails.cache = {};
-      }
-      sails.cache[inputs.keyString] = inputs.value;
-      return exits.success();
-    }
+    await sails.helpers.redis.ensureCacheExists();
 
+    sails.cache[inputs.keyString] = inputs.value;
+    return exits.success();
   }
 
 };

api/hooks/playerTracking/index.js

@@ -105,7 +105,7 @@ module.exports = function definePlayerTrackingHook(sails) {
           //await deleteInventoryData(server);
           await sails.helpers.redis.set(`server:${serverId}:trackingCyclesCompleted`, 0);
         } else {
-          await sails.helpers.redis.set(`server:${serverId}:trackingCyclesCompleted`, currentCycles + 1);
+          await sails.helpers.redis.incr(`server:${serverId}:trackingCyclesCompleted`);
         }
 
         let dateEnded = new Date();

api/hooks/sdtdLogs/LoggingObject.js

@@ -39,16 +39,17 @@ class LoggingObject extends EventEmitter {
   }
 
   async addFetchJob() {
+    sails.log.debug(`Adding new fetch job for server ${this.serverId} - last log line: ${this.lastLogLine}`);
     this.queue.add(
       {
         serverId: this.serverId,
-        lastLogLine: this.lastLogLine // FIXME - currently ignored
+        lastLogLine: this.lastLogLine
       },
       {
         timeout: 5000,
         removeOnComplete: 100,
         removeOnFail: 100,
-        attempts: 1,
+        attempts: 0,
         delay: this.intervalTime
       }
     );
@@ -70,18 +71,25 @@ class LoggingObject extends EventEmitter {
       // Fail silently
     }
 
-    await this.addFetchJob()
+    await this.addFetchJob("init")
   }
 
   async handleError(error) {
     sails.log.error(inspect(error));
   }
 
-  async handleFailedJob(job, err) {
+  async handleFailedJob(jobId, err) {
+    const job = await this.queue.getJob(jobId);
+
+
+    if (job.data.serverId != this.serverId) {
+      // not one of ours
+      return;
+    }
+
     // A job failed with reason `err`!
     sails.log.error(`Queue error: ${inspect(err)}`);
     await this._failedHandler();
-    await this.addFetchJob();
     return;
   }
 
@@ -90,7 +98,8 @@ class LoggingObject extends EventEmitter {
       result = JSON.parse(result);
     }
 
-    if (result.serverId.toString() !== this.serverId) {
+
+    if (result.serverId != this.serverId) {
       // not one of ours
       return;
     }
@@ -118,16 +127,21 @@ class LoggingObject extends EventEmitter {
       this.emit(newLog.type, enrichedLog.data);
     }
 
+    if (this.failed) {
+      await this.setFailedToZero();
+      this.failed = false;
+    }
+
     // If the server is in slowmode and we receive data again, this shows the server is back online
     if (this.slowmode) {
       this.slowmode = false;
       await this.stop();
       await this.init();
+      return
     }
 
-    await this.setFailedToZero();
-    await this.setLastLogLine(result.lastLogLine);
-    await this.addFetchJob();
+    await this.setLastLogLine();
+    await this.addFetchJob("completedHandler");
   }
 
   async destroy() {
@@ -178,7 +192,7 @@ class LoggingObject extends EventEmitter {
       `sdtdserver:${this.serverId}:sdtdLogs:lastSuccess`
     );
     lastSuccess = parseInt(lastSuccess);
-    if (counter > 25) {
+    if (counter > 100) {
       let prettyLastSuccess = new Date(lastSuccess);
 
       if (!this.slowmode) {
@@ -190,6 +204,7 @@ class LoggingObject extends EventEmitter {
         this.slowmode = true;
         await this.stop();
         await this.init(300000);
+        return
       }
 
       if (lastSuccess + threeDaysInMs < Date.now()) {
@@ -199,6 +214,7 @@ class LoggingObject extends EventEmitter {
         await sails.helpers.meta.setServerInactive(this.serverId);
       }
     }
+    await this.addFetchJob("failedHandler");
   }
 }
 

api/hooks/sdtdLogs/index.js

@@ -28,6 +28,9 @@ module.exports = function sdtdLogs(sails) {
         sails.log.info('Initializing custom hook (`sdtdLogs`)');
 
         try {
+          // Make sure there are no lingering jobs
+          // TODO: Once we scale this to multiple processes, this should happen differently
+          await sails.helpers.getQueueObject('logs').empty();
           let enabledServers = await SdtdConfig.find({
             loggingEnabled: true,
             inactive: false,

api/hooks/sdtdLogs/logProcessor.js

@@ -1,7 +1,6 @@
-const {promisify} = require("util");
 const handleLogLine = require("./handleLogLine");
 
-module.exports = async function(job) {
+module.exports = async function (job) {
   const resultLogs = [];
   let lastLogLine = job.data.lastLogLine;
   // If latest log line is not found, get it from the server