agendaJobs.js

const oauth = Npm.require('oauth');
const Agenda = Npm.require('agenda');

Meteor.startup(function(){
   const mongoURL = process.env.MONGO_URL;
   var agenda = new Agenda({db: {address: mongoURL}});

   agenda.define('sync_inventory', Meteor.bindEnvironment(function(job, done) {
     try {
       syncInventory();
     } catch (err) {
       handleSyncError(err);
     }
     if(done) done();
   }));

   agenda.define('process_orders', Meteor.bindEnvironment(function(job, done) {
     const cursor = ReactionCore.Collections.Orders.find({
       "workflow.status" : "coreOrderWorkflow/processing"
     });
     for (order of cursor.fetch()) {
       const val = process.env.WINE_LOGISTIX_PROCESS_ORDERS || Meteor.settings.easywine.WINE_LOGISTIX_PROCESS_ORDERS;
       const isDryRun = !val || val === "0" || typeof(val) === "string" && val.toLowerCase() !== "true";
       const dryRunInfo = isDryRun ? "DRYRUN: " : "";
       const msg = `${dryRunInfo}About to create order at Winelogistix.`;
       ReactionCore.Log.info(order._id, msg);
       if (!isDryRun) {
         EasyWine.winelogistixOrderCreate(order, function(err, success) {
           if (success) {
              Meteor.call("workflow/pushOrderWorkflow", "coreOrderWorkflow", "completed", order);
           }
         });
       }
     }
     if(done) done();
   }));

   agenda.on('ready', function() {
     ReactionCore.Log.trace("+++ agenda startup +++");
     agenda.now('sync_inventory');
     agenda.now('process_orders');
     agenda.every('*/30 * * * *', 'sync_inventory');
     agenda.every('*/5 * * * *', 'process_orders');
     agenda.start();
   });
});

Meteor.methods({
  // Implemented as server method to have this.userId available, through
  // which the user is impersonated
  "easywine/updateStockImpersonated": function(sku, quantity) {
    check(sku, String);
    check(quantity, Number);
    if (!Meteor.isServer) return;
    const email = process.env.REACTION_EMAIL || Meteor.settings.reaction.REACTION_EMAIL;
    const accountId = Meteor.users.findOne({
      "emails.address": email
    })._id;

    // impersonate as Admin user
    // this.setUserId(accountId); //does throw error "Can't call setUserId on a server initiated method call"
    this.userId = accountId;

    // Update needs to happen directly after imporsonating. Otherwise the user context will be lost.
    const res = ReactionCore.Collections.Products.update({
      sku: sku
 //     $or: [
 //       {"inventoryQuantity": {$gt: quantity}},
 //       {"inventoryQuantity": -1}]
    }, {
      $set: {
        "inventoryQuantity": quantity
      }
    }, {
      selector: { type: "variant" },
      validate: false,
      multi: true
    });
    if (res == 1) {
       // record changed
       ReactionCore.Log.trace("Updated inventoryQuantity of article " + sku + ": " + quantity);
    } else {
      // Record unchanged.
       const msg = "No update for inventoryQuantity of article " + sku + ".";
       ReactionCore.Log.trace(msg);
    }
    return res;
  }
});

function syncInventory() {
   ReactionCore.Log.info("Sync inventory with wine-logistix");
   // Iterate over all product variants
   let wines = ReactionCore.Collections.Products.find({type: "variant"}).fetch();
   let articles = [];
   let i =0;
   for (let wine of wines) {
     articles.push(wine.sku);
   }

   getLiveStock(articles, function(err, data){
     if (err) {
       handleSyncError(err);
     }
     try {
       processStockList(data);
     } catch (err) {
       handleSyncError(err);
     }
   });
}

function getLiveStock(articles, cb) {
  let requestToken = null;
  let accessToken = null;
  let oauthToken = null;
  let oauthTokenSecret = null;
  let key = process.env.WINE_LOGISTIX_CONSUMER_KEY || Meteor.settings.easywine.WINE_LOGISTIX_CONSUMER_KEY;
  let secret = process.env.WINE_LOGISTIX_CONSUMER_SECRET || Meteor.settings.easywine.WINE_LOGISTIX_CONSUMER_SECRET;
  // Create the client we use to request everything.
  // The first two parameters (requestToken and accessToken) aren't needed since we use it for 2-legged
  // and not 3-legged
  let request = new oauth.OAuth(requestToken, accessToken, key, secret, '1.0', null, 'HMAC-SHA1');
  const endpoint = "http://wine-logistix.de/API/v3/livestock/?artno=" + articles.join(',');
  request.get(endpoint, null, null, Meteor.bindEnvironment(function(err, data, res) {
    if (err) {
      cb(err, null);
    } else {
      // ReactionCore.Log.debug("REST call to " + endpoint + " was successful. Response: " + data);
      ReactionCore.Log.debug("getLiveStock: REST call to " + endpoint + " was successful");
      cb(null, JSON.parse(data));
    }
  }));
}

function processStockList(data) {
  if (data && data.success === true && data.message === "OK") {
    for (let item of data.livestock) {
       const wine = item.article;
       if (wine.message !== "OK") {
         // e.g article not found
         if (wine.message.indexOf("article not found") >= 0) {
           // Ignore.
           ReactionCore.Log.warn(`Error updating article ${wine.art_no}: ${wine.message}`);
           wine.stock_amount = 0;
         } else {
           throw Error("SKU: " + wine.art_no + " | Error: " + wine.message);
         }
       }
       Meteor.call("easywine/updateStockImpersonated", wine.art_no, wine.stock_amount);
    }
  } else {
    throw Error(data && data.message);
  }
}

function handleSyncError(msg) {
  // var recipient = process.env.REACTION_EMAIL || Meteor.settings.easywine.REACTION_EMAIL;
  const recipient = process.env.WEBMASTER_EMAIL || Meteor.settings.easywine.WEBMASTER_EMAIL;
  //var recipient = "michael.jenny@projekt-und-partner.de";
  ReactionCore.Log.error(msg, "Can't sync inventory.");

  // configure email
  ReactionCore.configureMailUrl();
  // don't send emails unless email server configured
  if (!process.env.MAIL_URL) {
    ReactionCore.Log.warn(`Mail not configured`);
  }
  try {
    return Email.send({
      to: recipient,
      from: `easywine`,
      subject: `Livestock syncing failed`,
      html: "Es trat ein Fehler beim Synchronisieren auf.\r\n " +  msg
    });
  } catch (e) {
    ReactionCore.Log.warn(e, "Unable to send email, check configuration and port.");
  }
}

function reportStock() {
   let wines = ReactionCore.Collections.Products.find({type: "variant"}).fetch();
   wines.sort(function(a,b){
       return a.inventoryQuantity > b.inventoryQuantity;
   });
}