Merge branch 'master' into docs-781

.travis.yml

@@ -3,16 +3,16 @@ before_script:
   - sudo mv /etc/apt/sources.list.d/pgdg-source.list* /tmp
   - sudo apt-get -qq purge postgis* postgresql*
   - sudo rm -Rf /var/lib/postgresql /etc/postgresql
-  - sudo apt-add-repository --yes ppa:cartodb/postgresql-9.3
+  - sudo apt-add-repository --yes ppa:cartodb/postgresql-9.5
   - sudo apt-add-repository --yes ppa:cartodb/gis
   - sudo apt-get update
-  - sudo apt-get install -q postgresql-9.3-postgis-2.1
+  - sudo apt-get install -q postgresql-9.5-postgis-2.2
-  - sudo apt-get install -q postgresql-contrib-9.3
+  - sudo apt-get install -q postgresql-contrib-9.5
-  - sudo apt-get install -q postgresql-plpython-9.3
+  - sudo apt-get install -q postgresql-plpython-9.5
   - sudo apt-get install -q postgis
   - sudo apt-get install -q gdal-bin
   - sudo apt-get install -q ogr2ogr2-static-bin
-  - echo -e "local\tall\tall\ttrust\nhost\tall\tall\t127.0.0.1/32\ttrust\nhost\tall\tall\t::1/128\ttrust" |sudo tee /etc/postgresql/9.3/main/pg_hba.conf
+  - echo -e "local\tall\tall\ttrust\nhost\tall\tall\t127.0.0.1/32\ttrust\nhost\tall\tall\t::1/128\ttrust" |sudo tee /etc/postgresql/9.5/main/pg_hba.conf
   - sudo service postgresql restart
   - psql -c 'create database template_postgis;' -U postgres
   - psql -c 'CREATE EXTENSION postgis;' -U postgres -d template_postgis

NEWS.md

@@ -1,6 +1,56 @@
-1.29.2 - 2016-mm-dd
+1.33.1 - 2016-mm-dd
 -------------------
 
+New features:
+ * Allow to setup more than one domain to validate oauth against.
+
+
+1.33.0 - 2016-07-01
+-------------------
+
+New features:
+ * Add `<%= job_id %>` template support for onerror and onsuccess fallback queries.
+
+
+1.32.0 - 2016-06-30
+-------------------
+
+New features:
+ * Broadcast after enqueueing jobs to improve query distribution load.
+
+
+1.31.0 - 2016-06-29
+-------------------
+
+New features:
+ * Adds start and end time for batch queries with fallback.
+ * Add `<%= error_message %>` template support for onerror fallback queries.
+
+
+1.30.1 - 2016-06-23
+-------------------
+
+Bug fixes:
+ * Fixed issue with profiling in Batch API #318
+
+
+1.30.0 - 2016-06-14
+-------------------
+
+Announcements:
+ * Now Batch API sends stats metrics to statsd server #312
+ * Now Batch API sets "skipped" instead of "pending" to queries that won't be performed #311
+
+ Bug fixes:
+  * Fixed issue with error handling in Batch API #316
+
+
+1.29.2 - 2016-05-25
+-------------------
+
+Bug fixes:
+ * Fixed issue with status transition in fallback jobs #308
+
 
 1.29.1 - 2016-05-24
 -------------------

app/app.js

@@ -180,22 +180,21 @@ function App() {
 
     var userDatabaseService = new UserDatabaseService(metadataBackend);
 
-    var jobQueue = new JobQueue(metadataBackend);
     var jobPublisher = new JobPublisher(redis);
+    var jobQueue = new JobQueue(metadataBackend, jobPublisher);
     var userIndexer = new UserIndexer(metadataBackend);
-    var jobBackend = new JobBackend(metadataBackend, jobQueue, jobPublisher, userIndexer);
+    var jobBackend = new JobBackend(metadataBackend, jobQueue, userIndexer);
     var userDatabaseMetadataService = new UserDatabaseMetadataService(metadataBackend);
     var jobCanceller = new JobCanceller(userDatabaseMetadataService);
     var jobService = new JobService(jobBackend, jobCanceller);
 
-
     var genericController = new GenericController();
     genericController.route(app);
 
     var queryController = new QueryController(userDatabaseService, tableCache, statsd_client);
     queryController.route(app);
 
-    var jobController = new JobController(userDatabaseService, jobService, jobCanceller);
+    var jobController = new JobController(userDatabaseService, jobService, statsd_client);
     jobController.route(app);
 
     var cacheStatusController = new CacheStatusController(tableCache);
@@ -210,7 +209,7 @@ function App() {
     var isBatchProcess = process.argv.indexOf('--no-batch') === -1;
 
     if (global.settings.environment !== 'test' && isBatchProcess) {
-        app.batch = batchFactory(metadataBackend);
+        app.batch = batchFactory(metadataBackend, statsd_client);
         app.batch.start();
     }
 

app/auth/oauth.js

@@ -3,6 +3,8 @@ var _ = require('underscore');
 var OAuthUtil = require('oauth-client');
 var step = require('step');
 var assert = require('assert');
+var CdbRequest = require('../models/cartodb_request');
+var cdbReq = new CdbRequest();
 
 var oAuth = (function(){
   var me = {
@@ -60,79 +62,87 @@ var oAuth = (function(){
     return removed;
   };
 
+  me.getAllowedHosts= function() {
+    var oauthConfig = global.settings.oauth || {};
+    return oauthConfig.allowedHosts || ['carto.com', 'cartodb.com'];
+  };
 
   // do new fancy get User ID
   me.verifyRequest = function(req, metadataBackend, callback) {
     var that = this;
     //TODO: review this
     var httpProto = req.protocol;
-    var passed_tokens;
+    if(!httpProto || (httpProto !== 'http' && httpProto !== 'https')) {
-    var ohash;
+      var msg = "Unknown HTTP protocol " + httpProto + ".";
+      var unknownProtocolErr = new Error(msg);
+      unknownProtocolErr.http_status = 500;
+      return callback(unknownProtocolErr);
+    }
+
+    var username = cdbReq.userByReq(req);
+    var requestTokens;
     var signature;
 
     step(
       function getTokensFromURL(){
         return oAuth.parseTokens(req);
       },
-      function getOAuthHash(err, data){
+      function getOAuthHash(err, _requestTokens) {
         assert.ifError(err);
 
         // this is oauth request only if oauth headers are present
-        this.is_oauth_request = !_.isEmpty(data);
+        this.is_oauth_request = !_.isEmpty(_requestTokens);
 
         if (this.is_oauth_request) {
-          passed_tokens = data;
+          requestTokens = _requestTokens;
-          that.getOAuthHash(metadataBackend, passed_tokens.oauth_token, this);
+          that.getOAuthHash(metadataBackend, requestTokens.oauth_token, this);
         } else {
           return null;
         }
       },
-      function regenerateSignature(err, data){
+      function regenerateSignature(err, oAuthHash){
         assert.ifError(err);
         if (!this.is_oauth_request) {
             return null;
         }
 
-        ohash = data;
+        var consumer     = OAuthUtil.createConsumer(oAuthHash.consumer_key, oAuthHash.consumer_secret);
-        var consumer     = OAuthUtil.createConsumer(ohash.consumer_key, ohash.consumer_secret);
+        var access_token = OAuthUtil.createToken(oAuthHash.access_token_token, oAuthHash.access_token_secret);
-        var access_token = OAuthUtil.createToken(ohash.access_token_token, ohash.access_token_secret);
         var signer       = OAuthUtil.createHmac(consumer, access_token);
 
         var method = req.method;
-        var host   = req.headers.host;
+        var hostsToValidate = {};
+        var requestHost = req.headers.host;
+        hostsToValidate[requestHost] = true;
+        that.getAllowedHosts().forEach(function(allowedHost) {
+            hostsToValidate[username + '.' + allowedHost] = true;
+        });
 
-        if(!httpProto || (httpProto !== 'http' && httpProto !== 'https')) {
-          var msg = "Unknown HTTP protocol " + httpProto + ".";
-          err = new Error(msg);
-          err.http_status = 500;
-          callback(err);
-          return;
-        }
-        
-        var path   = httpProto + '://' + host + req.path;
         that.splitParams(req.query);
-
-        // remove signature from passed_tokens
-        signature = passed_tokens.oauth_signature;
-        delete passed_tokens.oauth_signature;
-
-        var joined = {};
-
         // remove oauth_signature from body
         if(req.body) {
             delete req.body.oauth_signature;
         }
-        _.extend(joined, req.body ? req.body : null);
+        signature = requestTokens.oauth_signature;
-        _.extend(joined, passed_tokens);
+        // remove signature from requestTokens
-        _.extend(joined, req.query);
+        delete requestTokens.oauth_signature;
+        var requestParams = _.extend({}, req.body, requestTokens, req.query);
 
-        return signer.sign(method, path, joined);
+        var hosts = Object.keys(hostsToValidate);
+        var requestSignatures = hosts.map(function(host) {
+            var url = httpProto + '://' + host + req.path;
+            return signer.sign(method, url, requestParams);
+        });
+
+        return requestSignatures.reduce(function(validSignature, requestSignature) {
+            if (signature === requestSignature && !_.isUndefined(requestSignature)) {
+                validSignature = true;
+            }
+            return validSignature;
+        }, false);
       },
-      function checkSignature(err, data){
+      function finishValidation(err, hasValidSignature) {
-        assert.ifError(err);
+        return callback(err, hasValidSignature || null);
-
-        //console.log(data + " should equal the provided signature: " + signature);
-        callback(err, (signature === data && !_.isUndefined(data)) ? true : null);
       }
     );
   };

app/controllers/job_controller.js

@@ -26,9 +26,10 @@ function getMaxSizeErrorMessage(sql) {
     );
 }
 
-function JobController(userDatabaseService, jobService) {
+function JobController(userDatabaseService, jobService, statsdClient) {
     this.userDatabaseService = userDatabaseService;
     this.jobService = jobService;
+    this.statsdClient = statsdClient || { increment: function () {} };
 }
 
 function bodyPayloadSizeMiddleware(req, res, next) {
@@ -101,17 +102,26 @@ JobController.prototype.cancelJob = function (req, res) {
                 return handleException(err, res);
             }
 
-            if ( req.profiler ) {
-                req.profiler.done('cancelJob');
-                res.header('X-SQLAPI-Profiler', req.profiler.toJSONString());
-            }
-
             if (global.settings.api_hostname) {
-              res.header('X-Served-By-Host', global.settings.api_hostname);
+                res.header('X-Served-By-Host', global.settings.api_hostname);
             }
 
             if (result.host) {
-              res.header('X-Served-By-DB-Host', result.host);
+                res.header('X-Served-By-DB-Host', result.host);
+            }
+
+            if ( req.profiler ) {
+                req.profiler.done('cancelJob');
+                req.profiler.end();
+                req.profiler.sendStats();
+
+                res.header('X-SQLAPI-Profiler', req.profiler.toJSONString());
+            }
+
+            if ( err ) {
+                self.statsdClient.increment('sqlapi.job.error');
+            } else {
+                self.statsdClient.increment('sqlapi.job.success');
             }
 
             res.send(result.job);
@@ -168,17 +178,26 @@ JobController.prototype.listJob = function (req, res) {
                 return handleException(err, res);
             }
 
-            if ( req.profiler ) {
-                req.profiler.done('listJob');
-                res.header('X-SQLAPI-Profiler', req.profiler.toJSONString());
-            }
-
             if (global.settings.api_hostname) {
-              res.header('X-Served-By-Host', global.settings.api_hostname);
+                res.header('X-Served-By-Host', global.settings.api_hostname);
             }
 
             if (result.host) {
-              res.header('X-Served-By-DB-Host', result.host);
+                res.header('X-Served-By-DB-Host', result.host);
+            }
+
+            if ( req.profiler ) {
+                req.profiler.done('listJob');
+                req.profiler.end();
+                req.profiler.sendStats();
+
+                res.header('X-SQLAPI-Profiler', req.profiler.toJSONString());
+            }
+
+            if ( err ) {
+                self.statsdClient.increment('sqlapi.job.error');
+            } else {
+                self.statsdClient.increment('sqlapi.job.success');
             }
 
             res.send(result.jobs);
@@ -234,17 +253,26 @@ JobController.prototype.getJob = function (req, res) {
                 return handleException(err, res);
             }
 
-            if ( req.profiler ) {
-                req.profiler.done('getJob');
-                res.header('X-SQLAPI-Profiler', req.profiler.toJSONString());
-            }
-
             if (global.settings.api_hostname) {
-              res.header('X-Served-By-Host', global.settings.api_hostname);
+                res.header('X-Served-By-Host', global.settings.api_hostname);
             }
 
             if (result.host) {
-              res.header('X-Served-By-DB-Host', result.host);
+                res.header('X-Served-By-DB-Host', result.host);
+            }
+
+            if ( req.profiler ) {
+                req.profiler.done('getJob');
+                req.profiler.end();
+                req.profiler.sendStats();
+
+                res.header('X-SQLAPI-Profiler', req.profiler.toJSONString());
+            }
+
+            if ( err ) {
+                self.statsdClient.increment('sqlapi.job.error');
+            } else {
+                self.statsdClient.increment('sqlapi.job.success');
             }
 
             res.send(result.job);
@@ -306,11 +334,6 @@ JobController.prototype.createJob = function (req, res) {
                 return handleException(err, res);
             }
 
-            if ( req.profiler ) {
-                req.profiler.done('persistJob');
-                res.header('X-SQLAPI-Profiler', req.profiler.toJSONString());
-            }
-
             if (global.settings.api_hostname) {
                 res.header('X-Served-By-Host', global.settings.api_hostname);
             }
@@ -319,6 +342,20 @@ JobController.prototype.createJob = function (req, res) {
                 res.header('X-Served-By-DB-Host', result.host);
             }
 
+            if ( req.profiler ) {
+                req.profiler.done('persistJob');
+                req.profiler.end();
+                req.profiler.sendStats();
+
+                res.header('X-SQLAPI-Profiler', req.profiler.toJSONString());
+            }
+
+            if ( err ) {
+                self.statsdClient.increment('sqlapi.job.error');
+            } else {
+                self.statsdClient.increment('sqlapi.job.success');
+            }
+
             res.status(201).send(result.job);
         }
     );
@@ -378,17 +415,26 @@ JobController.prototype.updateJob = function (req, res) {
                 return handleException(err, res);
             }
 
-            if ( req.profiler ) {
-                req.profiler.done('updateJob');
-                res.header('X-SQLAPI-Profiler', req.profiler.toJSONString());
-            }
-
             if (global.settings.api_hostname) {
-              res.header('X-Served-By-Host', global.settings.api_hostname);
+                res.header('X-Served-By-Host', global.settings.api_hostname);
             }
 
             if (result.host) {
-              res.header('X-Served-By-DB-Host', result.host);
+                res.header('X-Served-By-DB-Host', result.host);
+            }
+
+            if ( req.profiler ) {
+                req.profiler.done('updateJob');
+                req.profiler.end();
+                req.profiler.sendStats();
+
+                res.header('X-SQLAPI-Profiler', req.profiler.toJSONString());
+            }
+
+            if ( err ) {
+                self.statsdClient.increment('sqlapi.job.error');
+            } else {
+                self.statsdClient.increment('sqlapi.job.success');
             }
 
             res.send(result.job);

app/controllers/query_controller.js

@@ -243,7 +243,7 @@ QueryController.prototype.handleQuery = function (req, res) {
                 }
 
                 if ( req.profiler ) {
-                  req.profiler.sendStats(); // TODO: do on nextTick ?
+                  req.profiler.sendStats();
                 }
                 if (self.statsd_client) {
                   if ( err ) {

batch/index.js

@@ -15,20 +15,19 @@ var JobBackend = require('./job_backend');
 var JobService = require('./job_service');
 var Batch = require('./batch');
 
-module.exports = function batchFactory (metadataBackend) {
+module.exports = function batchFactory (metadataBackend, statsdClient) {
     var queueSeeker = new QueueSeeker(metadataBackend);
     var jobSubscriber = new JobSubscriber(redis, queueSeeker);
-    var jobQueuePool = new JobQueuePool(metadataBackend);
     var jobPublisher = new JobPublisher(redis);
-    var jobQueue =  new JobQueue(metadataBackend);
+    var jobQueuePool = new JobQueuePool(metadataBackend, jobPublisher);
+    var jobQueue =  new JobQueue(metadataBackend, jobPublisher);
     var userIndexer = new UserIndexer(metadataBackend);
-    var jobBackend = new JobBackend(metadataBackend, jobQueue, jobPublisher, userIndexer);
+    var jobBackend = new JobBackend(metadataBackend, jobQueue, userIndexer);
     var userDatabaseMetadataService = new UserDatabaseMetadataService(metadataBackend);
-    // TODO: down userDatabaseMetadataService
+    var queryRunner = new QueryRunner(userDatabaseMetadataService);
-    var queryRunner = new QueryRunner();
     var jobCanceller = new JobCanceller(userDatabaseMetadataService);
     var jobService = new JobService(jobBackend, jobCanceller);
-    var jobRunner = new JobRunner(jobService, jobQueue, queryRunner, userDatabaseMetadataService);
+    var jobRunner = new JobRunner(jobService, jobQueue, queryRunner, statsdClient);
 
     return new Batch(jobSubscriber, jobQueuePool, jobRunner, jobService);
 };

batch/job_backend.js

@@ -14,10 +14,9 @@ var finalStatus = [
     jobStatus.UNKNOWN
 ];
 
-function JobBackend(metadataBackend, jobQueueProducer, jobPublisher, userIndexer) {
+function JobBackend(metadataBackend, jobQueueProducer, userIndexer) {
     this.metadataBackend = metadataBackend;
     this.jobQueueProducer = jobQueueProducer;
-    this.jobPublisher = jobPublisher;
     this.userIndexer = userIndexer;
 }
 
@@ -29,8 +28,7 @@ function toRedisParams(job) {
     for (var property in obj) {
         if (obj.hasOwnProperty(property)) {
             redisParams.push(property);
-            // TODO: this should be moved to job model ??
+            if (property === 'query' && typeof obj[property] !== 'string') {
-            if ((property === 'query' || property === 'status') && typeof obj[property] !== 'string') {
                 redisParams.push(JSON.stringify(obj[property]));
             } else {
                 redisParams.push(obj[property]);
@@ -65,7 +63,6 @@ function toObject(job_id, redisParams, redisValues) {
     return obj;
 }
 
-// TODO: is it really necessary??
 function isJobFound(redisValues) {
     return redisValues[0] && redisValues[1] && redisValues[2] && redisValues[3] && redisValues[4];
 }
@@ -119,9 +116,6 @@ JobBackend.prototype.create = function (job, callback) {
                     return callback(err);
                 }
 
-                // broadcast to consumers
-                self.jobPublisher.publish(job.host);
-
                 self.userIndexer.add(job.user, job.job_id, function (err) {
                   if (err) {
                       return callback(err);

batch/job_queue.js

@@ -1,13 +1,23 @@
 'use strict';
 
-function JobQueue(metadataBackend) {
+function JobQueue(metadataBackend, jobPublisher) {
     this.metadataBackend = metadataBackend;
+    this.jobPublisher = jobPublisher;
     this.db = 5;
     this.redisPrefix = 'batch:queues:';
 }
 
 JobQueue.prototype.enqueue = function (job_id, host, callback) {
-    this.metadataBackend.redisCmd(this.db, 'LPUSH', [ this.redisPrefix + host, job_id ], callback);
+    var self = this;
+
+    this.metadataBackend.redisCmd(this.db, 'LPUSH', [ this.redisPrefix + host, job_id ], function (err) {
+        if (err) {
+            return callback(err);
+        }
+
+        self.jobPublisher.publish(host);
+        callback();
+    });
 };
 
 JobQueue.prototype.dequeue = function (host, callback) {

batch/job_queue_pool.js

@@ -2,8 +2,9 @@
 
 var JobQueue = require('./job_queue');
 
-function JobQueuePool(metadataBackend) {
+function JobQueuePool(metadataBackend, jobPublisher) {
     this.metadataBackend = metadataBackend;
+    this.jobPublisher = jobPublisher;
     this.queues = {};
 }
 
@@ -29,7 +30,7 @@ JobQueuePool.prototype.list = function () {
 
 JobQueuePool.prototype.createQueue = function (host) {
     this.queues[host] = {
-        queue: new JobQueue(this.metadataBackend),
+        queue: new JobQueue(this.metadataBackend, this.jobPublisher),
         currentJobId: null
     };
 

batch/job_runner.js

@@ -2,17 +2,21 @@
 
 var errorCodes = require('../app/postgresql/error_codes').codeToCondition;
 var jobStatus = require('./job_status');
+var Profiler = require('step-profiler');
 
-function JobRunner(jobService, jobQueue, queryRunner, userDatabaseMetadataService) {
+function JobRunner(jobService, jobQueue, queryRunner, statsdClient) {
     this.jobService = jobService;
     this.jobQueue = jobQueue;
     this.queryRunner = queryRunner;
-    this.userDatabaseMetadataService = userDatabaseMetadataService; // TODO: move to queryRunner
+    this.statsdClient = statsdClient;
 }
 
 JobRunner.prototype.run = function (job_id, callback) {
     var self = this;
 
+    var profiler = new Profiler({ statsd_client: self.statsdClient });
+    profiler.start('sqlapi.batch.job');
+
     self.jobService.get(job_id, function (err, job) {
         if (err) {
             return callback(err);
@@ -31,55 +35,59 @@ JobRunner.prototype.run = function (job_id, callback) {
                 return callback(err);
             }
 
-            self._run(job, query, callback);
+            profiler.done('running');
+
+            self._run(job, query, profiler, callback);
         });
     });
 };
 
-JobRunner.prototype._run = function (job, query, callback) {
+JobRunner.prototype._run = function (job, query, profiler, callback) {
     var self = this;
 
-    // TODO: move to query
+    self.queryRunner.run(job.data.job_id, query, job.data.user, function (err /*, result */) {
-    self.userDatabaseMetadataService.getUserMetadata(job.data.user, function (err, userDatabaseMetadata) {
         if (err) {
+            if (!err.code) {
+                return callback(err);
+            }
+            // if query has been cancelled then it's going to get the current
+            // job status saved by query_canceller
+            if (errorCodes[err.code.toString()] === 'query_canceled') {
+                return self.jobService.get(job.data.job_id, callback);
+            }
+        }
+
+        try {
+            if (err) {
+                profiler.done('failed');
+                job.setStatus(jobStatus.FAILED, err.message);
+            } else {
+                profiler.done('success');
+                job.setStatus(jobStatus.DONE);
+            }
+        } catch (err) {
             return callback(err);
         }
 
-        self.queryRunner.run(job.data.job_id, query, userDatabaseMetadata, function (err /*, result */) {
+        self.jobService.save(job, function (err, job) {
             if (err) {
-                // if query has been cancelled then it's going to get the current
-                // job status saved by query_canceller
-                if (errorCodes[err.code.toString()] === 'query_canceled') {
-                    return self.jobService.get(job.data.job_id, callback);
-                }
-            }
-
-            try {
-                if (err) {
-                    job.setStatus(jobStatus.FAILED, err.message);
-                } else {
-                    job.setStatus(jobStatus.DONE);
-                }
-            } catch (err) {
                 return callback(err);
             }
 
-            self.jobService.save(job, function (err, job) {
+            profiler.done('done');
+            profiler.end();
+            profiler.sendStats();
+
+            if (!job.hasNextQuery()) {
+                return callback(null, job);
+            }
+
+            self.jobQueue.enqueue(job.data.job_id, job.data.host, function (err) {
                 if (err) {
                     return callback(err);
                 }
 
-                if (!job.hasNextQuery()) {
+                callback(null, job);
-                    return callback(null, job);
-                }
-
-                self.jobQueue.enqueue(job.data.job_id, userDatabaseMetadata.host, function (err) {
-                    if (err) {
-                        return callback(err);
-                    }
-
-                    callback(null, job);
-                });
             });
         });
     });

batch/job_status.js

@@ -6,6 +6,7 @@ var JOB_STATUS_ENUM = {
     DONE: 'done',
     CANCELLED: 'cancelled',
     FAILED: 'failed',
+    SKIPPED: 'skipped',
     UNKNOWN: 'unknown'
 };
 

batch/models/job_base.js

@@ -1,28 +1,22 @@
 'use strict';
 
-var assert = require('assert');
+var util = require('util');
 var uuid = require('node-uuid');
+var JobStateMachine = require('./job_state_machine');
 var jobStatus = require('../job_status');
-var validStatusTransitions = [
-    [jobStatus.PENDING, jobStatus.RUNNING],
-    [jobStatus.PENDING, jobStatus.CANCELLED],
-    [jobStatus.PENDING, jobStatus.UNKNOWN],
-    [jobStatus.RUNNING, jobStatus.DONE],
-    [jobStatus.RUNNING, jobStatus.FAILED],
-    [jobStatus.RUNNING, jobStatus.CANCELLED],
-    [jobStatus.RUNNING, jobStatus.PENDING],
-    [jobStatus.RUNNING, jobStatus.UNKNOWN]
-];
 var mandatoryProperties = [
     'job_id',
     'status',
     'query',
     'created_at',
     'updated_at',
+    'host',
     'user'
 ];
 
 function JobBase(data) {
+    JobStateMachine.call(this);
+
     var now = new Date().toISOString();
 
     this.data = data;
@@ -39,24 +33,10 @@ function JobBase(data) {
         this.data.updated_at = now;
     }
 }
+util.inherits(JobBase, JobStateMachine);
 
 module.exports = JobBase;
 
-JobBase.prototype.isValidStatusTransition = function (initialStatus, finalStatus) {
-    var transition = [ initialStatus, finalStatus ];
-
-    for (var i = 0; i < validStatusTransitions.length; i++) {
-        try {
-            assert.deepEqual(transition, validStatusTransitions[i]);
-            return true;
-        } catch (e) {
-            continue;
-        }
-    }
-
-    return false;
-};
-
 // should be implemented by childs
 JobBase.prototype.getNextQuery = function () {
     throw new Error('Unimplemented method');
@@ -104,7 +84,7 @@ JobBase.prototype.setQuery = function (query) {
 JobBase.prototype.setStatus = function (finalStatus, errorMesssage) {
     var now = new Date().toISOString();
     var initialStatus = this.data.status;
-    var isValid = this.isValidStatusTransition(initialStatus, finalStatus);
+    var isValid = this.isValidTransition(initialStatus, finalStatus);
 
     if (!isValid) {
         throw new Error('Cannot set status from ' + initialStatus + ' to ' + finalStatus);

batch/models/job_fallback.js

@@ -3,34 +3,29 @@
 var util = require('util');
 var JobBase = require('./job_base');
 var jobStatus = require('../job_status');
-var breakStatus = [
+var QueryFallback = require('./query/query_fallback');
-    jobStatus.CANCELLED,
+var MainFallback = require('./query/main_fallback');
-    jobStatus.FAILED,
+var QueryFactory = require('./query/query_factory');
-    jobStatus.UNKNOWN
-];
-function isBreakStatus(status) {
-    return breakStatus.indexOf(status) !== -1;
-}
-var finalStatus = [
-    jobStatus.CANCELLED,
-    jobStatus.DONE,
-    jobStatus.FAILED,
-    jobStatus.UNKNOWN
-];
-function isFinalStatus(status) {
-    return finalStatus.indexOf(status) !== -1;
-}
 
 function JobFallback(jobDefinition) {
     JobBase.call(this, jobDefinition);
 
     this.init();
+
+    this.queries = [];
+    for (var i = 0; i < this.data.query.query.length; i++) {
+        this.queries[i] = QueryFactory.create(this.data, i);
+    }
+
+    if (MainFallback.is(this.data)) {
+        this.fallback = new MainFallback();
+    }
 }
 util.inherits(JobFallback, JobBase);
 
 module.exports = JobFallback;
 
-// from user: {
+// 1. from user: {
 //     query: {
 //         query: [{
 //             query: 'select ...',
@@ -39,7 +34,8 @@ module.exports = JobFallback;
 //         onerror: 'select ...'
 //     }
 // }
-// from redis: {
+//
+// 2. from redis: {
 //     status: 'pending',
 //     fallback_status: 'pending'
 //     query: {
@@ -63,11 +59,7 @@ JobFallback.is = function (query) {
     }
 
     for (var i = 0; i < query.query.length; i++) {
-        if (!query.query[i].query) {
+        if (!QueryFallback.is(query.query[i])) {
-            return false;
-        }
-
-        if (typeof query.query[i].query !== 'string') {
             return false;
         }
     }
@@ -76,98 +68,65 @@ JobFallback.is = function (query) {
 };
 
 JobFallback.prototype.init = function () {
-    // jshint maxcomplexity: 8
     for (var i = 0; i < this.data.query.query.length; i++) {
-        if ((this.data.query.query[i].onsuccess || this.data.query.query[i].onerror) &&
+        if (shouldInitStatus(this.data.query.query[i])){
-            !this.data.query.query[i].status) {
             this.data.query.query[i].status = jobStatus.PENDING;
+        }
+        if (shouldInitQueryFallbackStatus(this.data.query.query[i])) {
             this.data.query.query[i].fallback_status = jobStatus.PENDING;
-        } else if (!this.data.query.query[i].status){
-            this.data.query.query[i].status = jobStatus.PENDING;
         }
     }
 
-    if ((this.data.query.onsuccess || this.data.query.onerror) && !this.data.status) {
+    if (shouldInitStatus(this.data)) {
         this.data.status = jobStatus.PENDING;
-        this.data.fallback_status = jobStatus.PENDING;
+    }
 
-    } else if (!this.data.status) {
+    if (shouldInitFallbackStatus(this.data)) {
-        this.data.status = jobStatus.PENDING;
+        this.data.fallback_status = jobStatus.PENDING;
+    }
+};
+
+function shouldInitStatus(jobOrQuery) {
+    return !jobOrQuery.status;
+}
+
+function shouldInitQueryFallbackStatus(query) {
+    return (query.onsuccess || query.onerror) && !query.fallback_status;
+}
+
+function shouldInitFallbackStatus(job) {
+    return (job.query.onsuccess || job.query.onerror) && !job.fallback_status;
+}
+
+JobFallback.prototype.getNextQueryFromQueries = function () {
+    for (var i = 0; i < this.queries.length; i++) {
+        if (this.queries[i].hasNextQuery(this.data)) {
+            return this.queries[i].getNextQuery(this.data);
+        }
+    }
+};
+
+JobFallback.prototype.hasNextQueryFromQueries = function () {
+    return !!this.getNextQueryFromQueries();
+};
+
+JobFallback.prototype.getNextQueryFromFallback = function () {
+    if (this.fallback && this.fallback.hasNextQuery(this.data)) {
+
+        return this.fallback.getNextQuery(this.data);
     }
 };
 
 JobFallback.prototype.getNextQuery = function () {
-    var query = this._getNextQueryFromQuery();
+    var query = this.getNextQueryFromQueries();
 
     if (!query) {
-        query = this._getNextQueryFromJobFallback();
+        query = this.getNextQueryFromFallback();
     }
 
     return query;
 };
 
-JobFallback.prototype._hasNextQueryFromQuery = function () {
-    return !!this._getNextQueryFromQuery();
-};
-
-JobFallback.prototype._getNextQueryFromQuery = function () {
-    // jshint maxcomplexity: 8
-    for (var i = 0; i < this.data.query.query.length; i++) {
-
-        if (this.data.query.query[i].fallback_status) {
-            if (this._isNextQuery(i)) {
-                return this.data.query.query[i].query;
-            } else if (this._isNextQueryOnSuccess(i)) {
-                return this.data.query.query[i].onsuccess;
-            } else if (this._isNextQueryOnError(i)) {
-                return this.data.query.query[i].onerror;
-            } else if (isBreakStatus(this.data.query.query[i].status)) {
-                return;
-            }
-        } else if (this.data.query.query[i].status === jobStatus.PENDING) {
-            return this.data.query.query[i].query;
-        }
-    }
-};
-
-JobFallback.prototype._getNextQueryFromJobFallback = function () {
-    if (this.data.fallback_status) {
-        if (this._isNextQueryOnSuccessJob()) {
-            return this.data.query.onsuccess;
-        } else if (this._isNextQueryOnErrorJob()) {
-            return this.data.query.onerror;
-        }
-    }
-};
-
-JobFallback.prototype._isNextQuery = function (index) {
-    return this.data.query.query[index].status === jobStatus.PENDING;
-};
-
-JobFallback.prototype._isNextQueryOnSuccess = function (index) {
-    return this.data.query.query[index].status === jobStatus.DONE &&
-        this.data.query.query[index].onsuccess &&
-        this.data.query.query[index].fallback_status === jobStatus.PENDING;
-};
-
-JobFallback.prototype._isNextQueryOnError = function (index) {
-    return this.data.query.query[index].status === jobStatus.FAILED &&
-        this.data.query.query[index].onerror &&
-        this.data.query.query[index].fallback_status === jobStatus.PENDING;
-};
-
-JobFallback.prototype._isNextQueryOnSuccessJob = function () {
-    return this.data.status === jobStatus.DONE &&
-        this.data.query.onsuccess &&
-        this.data.fallback_status === jobStatus.PENDING;
-};
-
-JobFallback.prototype._isNextQueryOnErrorJob = function () {
-    return this.data.status === jobStatus.FAILED &&
-        this.data.query.onerror &&
-        this.data.fallback_status === jobStatus.PENDING;
-};
-
 JobFallback.prototype.setQuery = function (query) {
     if (!JobFallback.is(query)) {
         throw new Error('You must indicate a valid SQL');
@@ -178,121 +137,72 @@ JobFallback.prototype.setQuery = function (query) {
 
 JobFallback.prototype.setStatus = function (status, errorMesssage) {
     var now = new Date().toISOString();
-    var resultFromQuery = this._setQueryStatus(status, errorMesssage);
-    var resultFromJob = this._setJobStatus(status, resultFromQuery.isChangeAppliedToQueryFallback, errorMesssage);
 
-    if (!resultFromJob.isValid && !resultFromQuery.isValid) {
+    var hasChanged = this.setQueryStatus(status, this.data, errorMesssage);
-        throw new Error('Cannot set status from ' + this.data.status + ' to ' + status);
+    hasChanged = this.setJobStatus(status, this.data, hasChanged, errorMesssage);
+    hasChanged = this.setFallbackStatus(status, this.data, hasChanged);
+
+    if (!hasChanged.isValid) {
+        throw new Error('Cannot set status to ' + status);
     }
 
     this.data.updated_at = now;
 };
 
-JobFallback.prototype._getLastStatusFromFinishedQuery = function () {
+JobFallback.prototype.setQueryStatus = function (status, job, errorMesssage) {
-    var lastStatus =  jobStatus.DONE;
+    return this.queries.reduce(function (hasChanged, query) {
-
+        var result = query.setStatus(status, this.data, hasChanged, errorMesssage);
-    for (var i = 0; i < this.data.query.query.length; i++) {
+        return result.isValid ? result : hasChanged;
-        if (this.data.query.query[i].fallback_status) {
+    }.bind(this), { isValid: false, appliedToFallback: false });
-            if (isFinalStatus(this.data.query.query[i].status)) {
-                lastStatus = this.data.query.query[i].status;
-            } else {
-                break;
-            }
-        } else {
-            if (isFinalStatus(this.data.query.query[i].status)) {
-                lastStatus = this.data.query.query[i].status;
-            } else {
-                break;
-            }
-        }
-    }
-
-    return lastStatus;
 };
 
-JobFallback.prototype._setJobStatus = function (status, isChangeAppliedToQueryFallback, errorMesssage) {
+JobFallback.prototype.setJobStatus = function (status, job, hasChanged, errorMesssage) {
-    var isValid = false;
+    var result = {
-
+        isValid: false,
-    status = this._shiftJobStatus(status, isChangeAppliedToQueryFallback);
+        appliedToFallback: false
-
-    isValid = this.isValidStatusTransition(this.data.status, status);
-
-    if (isValid) {
-        this.data.status = status;
-    } else if (this.data.fallback_status) {
-
-        isValid = this.isValidStatusTransition(this.data.fallback_status, status);
-
-        if (isValid) {
-            this.data.fallback_status = status;
-        }
-    }
-
-    if (status === jobStatus.FAILED && errorMesssage && !isChangeAppliedToQueryFallback) {
-        this.data.failed_reason = errorMesssage;
-    }
-
-    return {
-        isValid: isValid
     };
+
+    status = this.shiftStatus(status, hasChanged);
+
+    result.isValid = this.isValidTransition(job.status, status);
+    if (result.isValid) {
+        job.status = status;
+        if (status === jobStatus.FAILED && errorMesssage && !hasChanged.appliedToFallback) {
+            job.failed_reason = errorMesssage;
+        }
+    }
+
+    return result.isValid ? result : hasChanged;
 };
 
-JobFallback.prototype._shiftJobStatus = function (status, isChangeAppliedToQueryFallback) {
+JobFallback.prototype.setFallbackStatus = function (status, job, hasChanged) {
+    var result = hasChanged;
+
+    if (this.fallback && !this.hasNextQueryFromQueries()) {
+        result = this.fallback.setStatus(status, job, hasChanged);
+    }
+
+    return result.isValid ? result : hasChanged;
+};
+
+JobFallback.prototype.shiftStatus = function (status, hasChanged) {
     // jshint maxcomplexity: 7
-
+    if (hasChanged.appliedToFallback) {
-    // In some scenarios we have to change the normal flow in order to keep consistency
+        if (!this.hasNextQueryFromQueries() && (status === jobStatus.DONE || status === jobStatus.FAILED)) {
-    // between query's status and job's status.
+            status = this.getLastFinishedStatus();
-
-    if (isChangeAppliedToQueryFallback) {
-        if (!this._hasNextQueryFromQuery() && (status === jobStatus.DONE || status === jobStatus.FAILED)) {
-            status = this._getLastStatusFromFinishedQuery();
         } else if (status === jobStatus.DONE || status === jobStatus.FAILED){
             status = jobStatus.PENDING;
         }
-    } else if (this._hasNextQueryFromQuery() && status !== jobStatus.RUNNING) {
+    } else if (this.hasNextQueryFromQueries() && status !== jobStatus.RUNNING) {
         status = jobStatus.PENDING;
     }
 
     return status;
 };
 
-
+JobFallback.prototype.getLastFinishedStatus = function () {
-JobFallback.prototype._setQueryStatus = function (status, errorMesssage) {
+    return this.queries.reduce(function (lastFinished, query) {
-    // jshint maxcomplexity: 7
+        var status = query.getStatus(this.data);
-    var isValid = false;
+        return this.isFinalStatus(status) ? status : lastFinished;
-    var isChangeAppliedToQueryFallback = false;
+    }.bind(this), jobStatus.DONE);
-
-    for (var i = 0; i < this.data.query.query.length; i++) {
-        isValid = this.isValidStatusTransition(this.data.query.query[i].status, status);
-
-        if (isValid) {
-            this.data.query.query[i].status = status;
-
-            if (status === jobStatus.FAILED && errorMesssage) {
-                this.data.query.query[i].failed_reason = errorMesssage;
-            }
-
-            break;
-        }
-
-        if (this.data.query.query[i].fallback_status) {
-            isValid = this.isValidStatusTransition(this.data.query.query[i].fallback_status, status);
-
-            if (isValid) {
-                this.data.query.query[i].fallback_status = status;
-
-                if (status === jobStatus.FAILED && errorMesssage) {
-                    this.data.query.query[i].failed_reason = errorMesssage;
-                }
-
-                isChangeAppliedToQueryFallback = true;
-                break;
-            }
-        }
-    }
-
-    return {
-        isValid: isValid,
-        isChangeAppliedToQueryFallback: isChangeAppliedToQueryFallback
-    };
 };

batch/models/job_multiple.js

@@ -76,7 +76,7 @@ JobMultiple.prototype.setStatus = function (finalStatus, errorMesssage) {
     }
 
     for (var i = 0; i < this.data.query.length; i++) {
-        var isValid = JobMultiple.super_.prototype.isValidStatusTransition(this.data.query[i].status, finalStatus);
+        var isValid = JobMultiple.super_.prototype.isValidTransition(this.data.query[i].status, finalStatus);
 
         if (isValid) {
             this.data.query[i].status = finalStatus;

batch/models/job_state_machine.js

@@ -0,0 +1,46 @@
+'use strict';
+
+var assert = require('assert');
+var jobStatus = require('../job_status');
+var finalStatus = [
+    jobStatus.CANCELLED,
+    jobStatus.DONE,
+    jobStatus.FAILED,
+    jobStatus.UNKNOWN
+];
+
+var validStatusTransitions = [
+    [jobStatus.PENDING, jobStatus.RUNNING],
+    [jobStatus.PENDING, jobStatus.CANCELLED],
+    [jobStatus.PENDING, jobStatus.UNKNOWN],
+    [jobStatus.PENDING, jobStatus.SKIPPED],
+    [jobStatus.RUNNING, jobStatus.DONE],
+    [jobStatus.RUNNING, jobStatus.FAILED],
+    [jobStatus.RUNNING, jobStatus.CANCELLED],
+    [jobStatus.RUNNING, jobStatus.PENDING],
+    [jobStatus.RUNNING, jobStatus.UNKNOWN]
+];
+
+function JobStateMachine () {
+}
+
+module.exports = JobStateMachine;
+
+JobStateMachine.prototype.isValidTransition = function (initialStatus, finalStatus) {
+    var transition = [ initialStatus, finalStatus ];
+
+    for (var i = 0; i < validStatusTransitions.length; i++) {
+        try {
+            assert.deepEqual(transition, validStatusTransitions[i]);
+            return true;
+        } catch (e) {
+            continue;
+        }
+    }
+
+    return false;
+};
+
+JobStateMachine.prototype.isFinalStatus = function (status) {
+    return finalStatus.indexOf(status) !== -1;
+};

batch/models/query/fallback.js

@@ -0,0 +1,78 @@
+'use strict';
+
+var util = require('util');
+var QueryBase = require('./query_base');
+var jobStatus = require('../../job_status');
+
+function Fallback(index) {
+    QueryBase.call(this, index);
+}
+util.inherits(Fallback, QueryBase);
+
+module.exports = Fallback;
+
+Fallback.is = function (query) {
+    if (query.onsuccess || query.onerror) {
+        return true;
+    }
+    return false;
+};
+
+Fallback.prototype.getNextQuery = function (job) {
+    if (this.hasOnSuccess(job)) {
+        return this.getOnSuccess(job);
+    }
+    if (this.hasOnError(job)) {
+        return this.getOnError(job);
+    }
+};
+
+Fallback.prototype.getOnSuccess = function (job) {
+    if (job.query.query[this.index].status === jobStatus.DONE &&
+        job.query.query[this.index].fallback_status === jobStatus.PENDING) {
+        var onsuccessQuery = job.query.query[this.index].onsuccess;
+        if (onsuccessQuery) {
+            onsuccessQuery = onsuccessQuery.replace(/<%=\s*job_id\s*%>/g, job.job_id);
+        }
+        return onsuccessQuery;
+    }
+};
+
+Fallback.prototype.hasOnSuccess = function (job) {
+    return !!this.getOnSuccess(job);
+};
+
+Fallback.prototype.getOnError = function (job) {
+    if (job.query.query[this.index].status === jobStatus.FAILED &&
+        job.query.query[this.index].fallback_status === jobStatus.PENDING) {
+        var onerrorQuery = job.query.query[this.index].onerror;
+        if (onerrorQuery) {
+            onerrorQuery = onerrorQuery.replace(/<%=\s*job_id\s*%>/g, job.job_id);
+            onerrorQuery = onerrorQuery.replace(/<%=\s*error_message\s*%>/g, job.query.query[this.index].failed_reason);
+        }
+        return onerrorQuery;
+    }
+};
+
+Fallback.prototype.hasOnError = function (job) {
+    return !!this.getOnError(job);
+};
+
+Fallback.prototype.setStatus = function (status, job, errorMessage) {
+    var isValid = false;
+
+    isValid = this.isValidTransition(job.query.query[this.index].fallback_status, status);
+
+    if (isValid) {
+        job.query.query[this.index].fallback_status = status;
+        if (status === jobStatus.FAILED && errorMessage) {
+            job.query.query[this.index].failed_reason = errorMessage;
+        }
+    }
+
+    return isValid;
+};
+
+Fallback.prototype.getStatus = function (job) {
+    return job.query.query[this.index].fallback_status;
+};

batch/models/query/main_fallback.js

@@ -0,0 +1,74 @@
+'use strict';
+
+var util = require('util');
+var QueryBase = require('./query_base');
+var jobStatus = require('../../job_status');
+
+function MainFallback() {
+    QueryBase.call(this);
+}
+util.inherits(MainFallback, QueryBase);
+
+module.exports = MainFallback;
+
+MainFallback.is = function (job) {
+    if (job.query.onsuccess || job.query.onerror) {
+        return true;
+    }
+    return false;
+};
+
+MainFallback.prototype.getNextQuery = function (job) {
+    if (this.hasOnSuccess(job)) {
+        return this.getOnSuccess(job);
+    }
+
+    if (this.hasOnError(job)) {
+        return this.getOnError(job);
+    }
+};
+
+MainFallback.prototype.getOnSuccess = function (job) {
+    if (job.status === jobStatus.DONE && job.fallback_status === jobStatus.PENDING) {
+        return job.query.onsuccess;
+    }
+};
+
+MainFallback.prototype.hasOnSuccess = function (job) {
+    return !!this.getOnSuccess(job);
+};
+
+MainFallback.prototype.getOnError = function (job) {
+    if (job.status === jobStatus.FAILED && job.fallback_status === jobStatus.PENDING) {
+        return job.query.onerror;
+    }
+};
+
+MainFallback.prototype.hasOnError = function (job) {
+    return !!this.getOnError(job);
+};
+
+MainFallback.prototype.setStatus = function (status, job, previous) {
+    var isValid = false;
+    var appliedToFallback = false;
+
+    if (previous.isValid && !previous.appliedToFallback) {
+        if (this.isFinalStatus(status) && !this.hasNextQuery(job)) {
+            isValid = this.isValidTransition(job.fallback_status, jobStatus.SKIPPED);
+
+            if (isValid) {
+                job.fallback_status = jobStatus.SKIPPED;
+                appliedToFallback = true;
+            }
+        }
+    } else if (!previous.isValid) {
+        isValid = this.isValidTransition(job.fallback_status, status);
+
+        if (isValid) {
+            job.fallback_status = status;
+            appliedToFallback = true;
+        }
+    }
+
+    return { isValid: isValid, appliedToFallback: appliedToFallback };
+};

batch/models/query/query.js

@@ -0,0 +1,51 @@
+'use strict';
+
+var util = require('util');
+var QueryBase = require('./query_base');
+var jobStatus = require('../../job_status');
+
+function Query(index) {
+    QueryBase.call(this, index);
+}
+util.inherits(Query, QueryBase);
+
+module.exports = Query;
+
+Query.is = function (query) {
+    if (query.query && typeof query.query === 'string') {
+        return true;
+    }
+
+    return false;
+};
+
+Query.prototype.getNextQuery = function (job) {
+    if (job.query.query[this.index].status === jobStatus.PENDING) {
+        return job.query.query[this.index].query;
+    }
+};
+
+Query.prototype.setStatus = function (status, job, errorMesssage) {
+    var isValid = false;
+
+    isValid = this.isValidTransition(job.query.query[this.index].status, status);
+
+    if (isValid) {
+        job.query.query[this.index].status = status;
+        if (status === jobStatus.RUNNING) {
+            job.query.query[this.index].started_at = new Date().toISOString();
+        }
+        if (this.isFinalStatus(status)) {
+            job.query.query[this.index].ended_at = new Date().toISOString();
+        }
+        if (status === jobStatus.FAILED && errorMesssage) {
+            job.query.query[this.index].failed_reason = errorMesssage;
+        }
+    }
+
+    return isValid;
+};
+
+Query.prototype.getStatus = function (job) {
+    return job.query.query[this.index].status;
+};

batch/models/query/query_base.js

@@ -0,0 +1,31 @@
+'use strict';
+
+var util = require('util');
+var JobStateMachine = require('../job_state_machine');
+
+function QueryBase(index) {
+    JobStateMachine.call(this);
+
+    this.index = index;
+}
+util.inherits(QueryBase, JobStateMachine);
+
+module.exports = QueryBase;
+
+// should be implemented
+QueryBase.prototype.setStatus = function () {
+    throw new Error('Unimplemented method');
+};
+
+// should be implemented
+QueryBase.prototype.getNextQuery = function () {
+    throw new Error('Unimplemented method');
+};
+
+QueryBase.prototype.hasNextQuery = function (job) {
+    return !!this.getNextQuery(job);
+};
+
+QueryBase.prototype.getStatus = function () {
+    throw new Error('Unimplemented method');
+};

batch/models/query/query_factory.js

@@ -0,0 +1,16 @@
+'use strict';
+
+var QueryFallback = require('./query_fallback');
+
+function QueryFactory() {
+}
+
+module.exports = QueryFactory;
+
+QueryFactory.create = function (job, index) {
+    if (QueryFallback.is(job.query.query[index])) {
+        return new QueryFallback(job, index);
+    }
+
+    throw new Error('there is no query class for the provided query');
+};

batch/models/query/query_fallback.js

@@ -0,0 +1,75 @@
+'use strict';
+
+var util = require('util');
+var QueryBase = require('./query_base');
+var Query = require('./query');
+var Fallback = require('./fallback');
+var jobStatus = require('../../job_status');
+
+function QueryFallback(job, index) {
+    QueryBase.call(this, index);
+
+    this.init(job, index);
+}
+
+util.inherits(QueryFallback, QueryBase);
+
+QueryFallback.is = function (query) {
+    if (Query.is(query)) {
+        return true;
+    }
+    return false;
+};
+
+QueryFallback.prototype.init = function (job, index) {
+    this.query = new Query(index);
+
+    if (Fallback.is(job.query.query[index])) {
+        this.fallback = new Fallback(index);
+    }
+};
+
+QueryFallback.prototype.getNextQuery = function (job) {
+    if (this.query.hasNextQuery(job)) {
+        return this.query.getNextQuery(job);
+    }
+
+    if (this.fallback && this.fallback.hasNextQuery(job)) {
+        return this.fallback.getNextQuery(job);
+    }
+};
+
+QueryFallback.prototype.setStatus = function (status, job, previous, errorMesssage) {
+    // jshint maxcomplexity: 9
+    var isValid = false;
+    var appliedToFallback = false;
+
+    if (previous.isValid && !previous.appliedToFallback) {
+        if (status === jobStatus.FAILED || status === jobStatus.CANCELLED) {
+            this.query.setStatus(jobStatus.SKIPPED, job, errorMesssage);
+
+            if (this.fallback) {
+                this.fallback.setStatus(jobStatus.SKIPPED, job);
+            }
+        }
+    } else if (!previous.isValid) {
+        isValid = this.query.setStatus(status, job, errorMesssage);
+
+        if (this.fallback) {
+            if (!isValid) {
+                isValid = this.fallback.setStatus(status, job, errorMesssage);
+                appliedToFallback = true;
+            } else if (isValid && this.isFinalStatus(status) && !this.fallback.hasNextQuery(job)) {
+                this.fallback.setStatus(jobStatus.SKIPPED, job);
+            }
+        }
+    }
+
+    return { isValid: isValid, appliedToFallback: appliedToFallback };
+};
+
+QueryFallback.prototype.getStatus = function (job) {
+    return this.query.getStatus(job);
+};
+
+module.exports = QueryFallback;

batch/query_runner.js

@@ -2,37 +2,43 @@
 
 var PSQL = require('cartodb-psql');
 
-function QueryRunner() {
+function QueryRunner(userDatabaseMetadataService) {
+    this.userDatabaseMetadataService = userDatabaseMetadataService;
 }
 
 module.exports = QueryRunner;
 
-QueryRunner.prototype.run = function (job_id, sql, userDatabaseMetadata, callback) {
+QueryRunner.prototype.run = function (job_id, sql, user, callback) {
-    var pg = new PSQL(userDatabaseMetadata, {}, { destroyOnError: true });
+    this.userDatabaseMetadataService.getUserMetadata(user, function (err, userDatabaseMetadata) {
-
+        if (err) {
-    pg.query('SET statement_timeout=0', function (err) {
-        if(err) {
             return callback(err);
         }
 
-        // mark query to allow to users cancel their queries
+        var pg = new PSQL(userDatabaseMetadata, {}, { destroyOnError: true });
-        sql = '/* ' + job_id + ' */ ' + sql;
 
-        pg.eventedQuery(sql, function (err, query) {
+        pg.query('SET statement_timeout=0', function (err) {
-            if (err) {
+            if(err) {
                 return callback(err);
             }
 
-            query.on('error', callback);
+            // mark query to allow to users cancel their queries
+            sql = '/* ' + job_id + ' */ ' + sql;
 
-            query.on('end', function (result) {
+            pg.eventedQuery(sql, function (err, query) {
-                // only if result is present then query is done sucessfully otherwise an error has happened
+                if (err) {
-                // and it was handled by error listener
+                    return callback(err);
-                if (result) {
-                    callback(null, result);
                 }
+
+                query.on('error', callback);
+
+                query.on('end', function (result) {
+                    // only if result is present then query is done sucessfully otherwise an error has happened
+                    // and it was handled by error listener
+                    if (result) {
+                        callback(null, result);
+                    }
+                });
             });
         });
     });
-
 };

config/environments/production.js.example

@@ -85,4 +85,7 @@ module.exports.health = {
     username: 'development',
     query: 'select 1'
 };
+module.exports.oauth = {
+    allowedHosts: ['carto.com', 'cartodb.com']
+};
 module.exports.disabled_file = 'pids/disabled';

config/environments/test.js.example

@@ -74,4 +74,7 @@ module.exports.health = {
     username: 'vizzuality',
     query: 'select 1'
 };
+module.exports.oauth = {
+    allowedHosts: ['localhost.lan:8080', 'localhostdb.lan:8080']
+};
 module.exports.disabled_file = 'pids/disabled';

npm-shrinkwrap.json

@@ -1,6 +1,6 @@
 {
   "name": "cartodb_sql_api",
-  "version": "1.29.2",
+  "version": "1.33.1",
   "dependencies": {
     "cartodb-psql": {
       "version": "0.6.1",

package.json

@@ -5,7 +5,7 @@
     "keywords": [
         "cartodb"
     ],
-    "version": "1.29.2",
+    "version": "1.33.1",
     "repository": {
         "type": "git",
         "url": "git://github.com/CartoDB/CartoDB-SQL-API.git"

test/acceptance/batch.test.js

@@ -22,10 +22,10 @@ var metadataBackend = require('cartodb-redis')({
 describe('batch module', function() {
     var dbInstance = 'localhost';
     var username = 'vizzuality';
-    var jobQueue =  new JobQueue(metadataBackend);
     var jobPublisher = new JobPublisher(redis);
+    var jobQueue =  new JobQueue(metadataBackend, jobPublisher);
     var userIndexer = new UserIndexer(metadataBackend);
-    var jobBackend = new JobBackend(metadataBackend, jobQueue, jobPublisher, userIndexer);
+    var jobBackend = new JobBackend(metadataBackend, jobQueue, userIndexer);
     var userDatabaseMetadataService = new UserDatabaseMetadataService(metadataBackend);
     var jobCanceller = new JobCanceller(userDatabaseMetadataService);
     var jobService = new JobService(jobBackend, jobCanceller);

test/acceptance/job.callback-template.test.js

@@ -0,0 +1,225 @@
+require('../helper');
+
+var assert = require('../support/assert');
+var app = require(global.settings.app_root + '/app/app')();
+var querystring = require('qs');
+var metadataBackend = require('cartodb-redis')({
+    host: global.settings.redis_host,
+    port: global.settings.redis_port,
+    max: global.settings.redisPool,
+    idleTimeoutMillis: global.settings.redisIdleTimeoutMillis,
+    reapIntervalMillis: global.settings.redisReapIntervalMillis
+});
+var batchFactory = require('../../batch');
+var jobStatus = require('../../batch/job_status');
+
+describe('Batch API callback templates', function () {
+
+    function createJob(jobDefinition, callback) {
+        assert.response(app, {
+            url: '/api/v2/sql/job?api_key=1234',
+            headers: {
+                'Content-Type': 'application/x-www-form-urlencoded',
+                host: 'vizzuality.cartodb.com'
+            },
+            method: 'POST',
+            data: querystring.stringify(jobDefinition)
+        }, {
+            status: 201
+        }, function (res, err) {
+            if (err) {
+                return callback(err);
+            }
+            return callback(null, JSON.parse(res.body));
+        });
+    }
+
+    function getJobStatus(jobId, callback) {
+        assert.response(app, {
+            url: '/api/v2/sql/job/' + jobId + '?api_key=1234&',
+            headers: {
+                host: 'vizzuality.cartodb.com'
+            },
+            method: 'GET'
+        }, {
+            status: 200
+        }, function (res, err) {
+            if (err) {
+                return callback(err);
+            }
+            return callback(null, JSON.parse(res.body));
+        });
+    }
+
+    function getQueryResult(query, callback) {
+        assert.response(app, {
+            url: '/api/v2/sql?' + querystring.stringify({q: query, api_key: 1234}),
+            headers: {
+                host: 'vizzuality.cartodb.com'
+            },
+            method: 'GET'
+        }, {
+            status: 200
+        }, function (res, err) {
+            if (err) {
+                return callback(err);
+            }
+            return callback(null, JSON.parse(res.body));
+        });
+    }
+
+    function validateExpectedResponse(actual, expected) {
+        actual.query.forEach(function(actualQuery, index) {
+            var expectedQuery = expected.query[index];
+            assert.ok(expectedQuery);
+            Object.keys(expectedQuery).forEach(function(expectedKey) {
+                assert.equal(
+                    actualQuery[expectedKey],
+                    expectedQuery[expectedKey],
+                    'Expected value for key "' + expectedKey + '" does not match: ' + actualQuery[expectedKey] + ' ==' +
+                    expectedQuery[expectedKey] + ' at query index=' + index + '. Full response: ' +
+                    JSON.stringify(actual, null, 4)
+                );
+            });
+            var propsToCheckDate = ['started_at', 'ended_at'];
+            propsToCheckDate.forEach(function(propToCheckDate) {
+                if (actualQuery.hasOwnProperty(propToCheckDate)) {
+                    assert.ok(new Date(actualQuery[propToCheckDate]));
+                }
+            });
+        });
+
+        assert.equal(actual.onsuccess, expected.onsuccess);
+        assert.equal(actual.onerror, expected.onerror);
+    }
+
+    var batch = batchFactory(metadataBackend);
+
+    before(function () {
+        batch.start();
+    });
+
+    after(function (done) {
+        batch.stop();
+        batch.drain(function () {
+            metadataBackend.redisCmd(5, 'DEL', [ 'batch:queues:localhost' ], done);
+        });
+    });
+
+    describe.skip('should use templates for error_message and job_id onerror callback', function () {
+        var jobResponse;
+        before(function(done) {
+            getQueryResult('create table test_batch_errors (job_id text, error_message text)', function(err) {
+                if (err) {
+                    return done(err);
+                }
+                createJob({
+                    "query": {
+                        "query": [
+                            {
+                                "query": "SELECT * FROM invalid_table",
+                                "onerror": "INSERT INTO test_batch_errors " +
+                                    "values ('<%= job_id %>', '<%= error_message %>')"
+                            }
+                        ]
+                    }
+                }, function(err, job) {
+                    jobResponse = job;
+                    return done(err);
+                });
+            });
+        });
+
+        it('should keep the original templated query but use the error message', function (done) {
+            var expectedQuery = {
+                query: [
+                    {
+                        "query": "SELECT * FROM invalid_table",
+                        "onerror": "INSERT INTO test_batch_errors values ('<%= job_id %>', '<%= error_message %>')",
+                        status: 'failed',
+                        fallback_status: 'done'
+                    }
+                ]
+            };
+
+            var interval = setInterval(function () {
+                getJobStatus(jobResponse.job_id, function(err, job) {
+                    if (job.status === jobStatus.FAILED) {
+                        clearInterval(interval);
+                        validateExpectedResponse(job.query, expectedQuery);
+                        getQueryResult('select * from test_batch_errors', function(err, result) {
+                            if (err) {
+                                return done(err);
+                            }
+                            assert.equal(result.rows[0].job_id, jobResponse.job_id);
+                            assert.equal(result.rows[0].error_message, 'relation "invalid_table" does not exist');
+                            getQueryResult('drop table test_batch_errors', done);
+                        });
+                    } else if (job.status === jobStatus.DONE || job.status === jobStatus.CANCELLED) {
+                        clearInterval(interval);
+                        done(new Error('Job ' + job.job_id + ' is ' + job.status + ', expected to be "failed"'));
+                    }
+                });
+            }, 50);
+        });
+    });
+
+    describe('should use template for job_id onsuccess callback', function () {
+        var jobResponse;
+        before(function(done) {
+            createJob({
+                "query": {
+                    "query": [
+                        {
+                            query: "create table batch_jobs (job_id text)"
+                        },
+                        {
+                            "query": "SELECT 1",
+                            "onsuccess": "INSERT INTO batch_jobs values ('<%= job_id %>')"
+                        }
+                    ]
+                }
+            }, function(err, job) {
+                jobResponse = job;
+                return done(err);
+            });
+        });
+
+        it('should keep the original templated query but use the job_id', function (done) {
+            var expectedQuery = {
+                query: [
+                    {
+                        query: "create table batch_jobs (job_id text)",
+                        status: 'done'
+                    },
+                    {
+                        query: "SELECT 1",
+                        onsuccess: "INSERT INTO batch_jobs values ('<%= job_id %>')",
+                        status: 'done',
+                        fallback_status: 'done'
+                    }
+                ]
+            };
+
+            var interval = setInterval(function () {
+                getJobStatus(jobResponse.job_id, function(err, job) {
+                    if (job.status === jobStatus.DONE) {
+                        clearInterval(interval);
+                        validateExpectedResponse(job.query, expectedQuery);
+                        getQueryResult('select * from batch_jobs', function(err, result) {
+                            if (err) {
+                                return done(err);
+                            }
+                            assert.equal(result.rows[0].job_id, jobResponse.job_id);
+                            getQueryResult('drop table batch_jobs', done);
+                        });
+                    } else if (job.status === jobStatus.FAILED || job.status === jobStatus.CANCELLED) {
+                        clearInterval(interval);
+                        done(new Error('Job ' + job.job_id + ' is ' + job.status + ', expected to be "done"'));
+                    }
+                });
+            }, 50);
+        });
+    });
+
+});

test/acceptance/job.fallback.test.js

@@ -15,6 +15,25 @@ var jobStatus = require('../../batch/job_status');
 
 describe('Batch API fallback job', function () {
 
+    function validateExpectedResponse(actual, expected) {
+        actual.query.forEach(function(actualQuery, index) {
+            var expectedQuery = expected.query[index];
+            assert.ok(expectedQuery);
+            Object.keys(expectedQuery).forEach(function(expectedKey) {
+                assert.equal(actualQuery[expectedKey], expectedQuery[expectedKey]);
+            });
+            var propsToCheckDate = ['started_at', 'ended_at'];
+            propsToCheckDate.forEach(function(propToCheckDate) {
+                if (actualQuery.hasOwnProperty(propToCheckDate)) {
+                    assert.ok(new Date(actualQuery[propToCheckDate]));
+                }
+            });
+        });
+
+        assert.equal(actual.onsuccess, expected.onsuccess);
+        assert.equal(actual.onerror, expected.onerror);
+    }
+
     var batch = batchFactory(metadataBackend);
 
     before(function () {
@@ -31,7 +50,6 @@ describe('Batch API fallback job', function () {
     describe('"onsuccess" on first query should be triggered', function () {
         var fallbackJob = {};
 
-
         it('should create a job', function (done) {
             assert.response(app, {
                 url: '/api/v2/sql/job?api_key=1234',
@@ -86,7 +104,7 @@ describe('Batch API fallback job', function () {
                     var job = JSON.parse(res.body);
                     if (job.status === jobStatus.DONE) {
                         clearInterval(interval);
-                        assert.deepEqual(job.query, expectedQuery);
+                        validateExpectedResponse(job.query, expectedQuery);
                         done();
                     } else if (job.status === jobStatus.FAILED || job.status === jobStatus.CANCELLED) {
                         clearInterval(interval);
@@ -133,7 +151,7 @@ describe('Batch API fallback job', function () {
                     "query": "SELECT * FROM untitle_table_4",
                     "onerror": "SELECT * FROM untitle_table_4 limit 1",
                     "status": "done",
-                    "fallback_status": "pending"
+                    "fallback_status": "skipped"
                 }]
             };
             var interval = setInterval(function () {
@@ -153,7 +171,7 @@ describe('Batch API fallback job', function () {
                     var job = JSON.parse(res.body);
                     if (job.status === jobStatus.DONE) {
                         clearInterval(interval);
-                        assert.deepEqual(job.query, expectedQuery);
+                        validateExpectedResponse(job.query, expectedQuery);
                         done();
                     } else if (job.status === jobStatus.FAILED || job.status === jobStatus.CANCELLED) {
                         clearInterval(interval);
@@ -221,7 +239,7 @@ describe('Batch API fallback job', function () {
                     var job = JSON.parse(res.body);
                     if (job.status === jobStatus.FAILED) {
                         clearInterval(interval);
-                        assert.deepEqual(job.query, expectedQuery);
+                        validateExpectedResponse(job.query, expectedQuery);
                         done();
                     } else if (job.status === jobStatus.DONE || job.status === jobStatus.CANCELLED) {
                         clearInterval(interval);
@@ -268,7 +286,7 @@ describe('Batch API fallback job', function () {
                     query: 'SELECT * FROM nonexistent_table /* query should fail */',
                     onsuccess: 'SELECT * FROM untitle_table_4 limit 1',
                     status: 'failed',
-                    fallback_status: 'pending',
+                    fallback_status: 'skipped',
                     failed_reason: 'relation "nonexistent_table" does not exist'
                 }]
             };
@@ -290,7 +308,7 @@ describe('Batch API fallback job', function () {
                     var job = JSON.parse(res.body);
                     if (job.status === jobStatus.FAILED) {
                         clearInterval(interval);
-                        assert.deepEqual(job.query, expectedQuery);
+                        validateExpectedResponse(job.query, expectedQuery);
                         done();
                     } else if (job.status === jobStatus.DONE || job.status === jobStatus.CANCELLED) {
                         clearInterval(interval);
@@ -358,7 +376,7 @@ describe('Batch API fallback job', function () {
                     var job = JSON.parse(res.body);
                     if (job.status === jobStatus.DONE && job.fallback_status === jobStatus.DONE) {
                         clearInterval(interval);
-                        assert.deepEqual(job.query, expectedQuery);
+                        validateExpectedResponse(job.query, expectedQuery);
                         done();
                     } else if (job.status === jobStatus.FAILED || job.status === jobStatus.CANCELLED) {
                         clearInterval(interval);
@@ -424,9 +442,9 @@ describe('Batch API fallback job', function () {
                         return done(err);
                     }
                     var job = JSON.parse(res.body);
-                    if (job.status === jobStatus.FAILED && job.fallback_status === jobStatus.PENDING) {
+                    if (job.status === jobStatus.FAILED && job.fallback_status === jobStatus.SKIPPED) {
                         clearInterval(interval);
-                        assert.deepEqual(job.query, expectedQuery);
+                        validateExpectedResponse(job.query, expectedQuery);
                         done();
                     } else if (job.status === jobStatus.FAILED || job.status === jobStatus.CANCELLED) {
                         clearInterval(interval);
@@ -495,7 +513,7 @@ describe('Batch API fallback job', function () {
                     var job = JSON.parse(res.body);
                     if (job.status === jobStatus.FAILED && job.fallback_status === jobStatus.DONE) {
                         clearInterval(interval);
-                        assert.deepEqual(job.query, expectedQuery);
+                        validateExpectedResponse(job.query, expectedQuery);
                         done();
                     } else if (job.status === jobStatus.DONE || job.status === jobStatus.CANCELLED) {
                         clearInterval(interval);
@@ -560,9 +578,9 @@ describe('Batch API fallback job', function () {
                         return done(err);
                     }
                     var job = JSON.parse(res.body);
-                    if (job.status === jobStatus.DONE && job.fallback_status === jobStatus.PENDING) {
+                    if (job.status === jobStatus.DONE && job.fallback_status === jobStatus.SKIPPED) {
                         clearInterval(interval);
-                        assert.deepEqual(job.query, expectedQuery);
+                        validateExpectedResponse(job.query, expectedQuery);
                         done();
                     } else if (job.status === jobStatus.FAILED || job.status === jobStatus.CANCELLED) {
                         clearInterval(interval);
@@ -633,7 +651,7 @@ describe('Batch API fallback job', function () {
                     var job = JSON.parse(res.body);
                     if (job.status === jobStatus.DONE && job.fallback_status === jobStatus.DONE) {
                         clearInterval(interval);
-                        assert.deepEqual(job.query, expectedQuery);
+                        validateExpectedResponse(job.query, expectedQuery);
                         done();
                     } else if (job.status === jobStatus.FAILED || job.status === jobStatus.CANCELLED) {
                         clearInterval(interval);
@@ -709,7 +727,7 @@ describe('Batch API fallback job', function () {
                     var job = JSON.parse(res.body);
                     if (job.status === jobStatus.DONE) {
                         clearInterval(interval);
-                        assert.deepEqual(job.query, expectedQuery);
+                        validateExpectedResponse(job.query, expectedQuery);
                         done();
                     } else if (job.status === jobStatus.FAILED || job.status === jobStatus.CANCELLED) {
                         clearInterval(interval);
@@ -759,13 +777,13 @@ describe('Batch API fallback job', function () {
                     "query": "SELECT * FROM nonexistent_table /* should fail */",
                     "onsuccess": "SELECT * FROM untitle_table_4 limit 1",
                     "status": "failed",
-                    "fallback_status": "pending",
+                    "fallback_status": "skipped",
                     "failed_reason": 'relation "nonexistent_table" does not exist'
                 }, {
                     "query": "SELECT * FROM untitle_table_4 limit 2",
                     "onsuccess": "SELECT * FROM untitle_table_4 limit 3",
-                    "status": "pending",
+                    "status": "skipped",
-                    "fallback_status": "pending"
+                    "fallback_status": "skipped"
                 }]
             };
 
@@ -786,7 +804,7 @@ describe('Batch API fallback job', function () {
                     var job = JSON.parse(res.body);
                     if (job.status === jobStatus.FAILED) {
                         clearInterval(interval);
-                        assert.deepEqual(job.query, expectedQuery);
+                        validateExpectedResponse(job.query, expectedQuery);
                         done();
                     } else if (job.status === jobStatus.DONE || job.status === jobStatus.CANCELLED) {
                         clearInterval(interval);
@@ -842,7 +860,7 @@ describe('Batch API fallback job', function () {
                     "query": "SELECT * FROM nonexistent_table /* should fail */",
                     "onsuccess": "SELECT * FROM untitle_table_4 limit 3",
                     "status": "failed",
-                    "fallback_status": "pending",
+                    "fallback_status": "skipped",
                     "failed_reason": 'relation "nonexistent_table" does not exist'
                 }]
             };
@@ -864,7 +882,7 @@ describe('Batch API fallback job', function () {
                     var job = JSON.parse(res.body);
                     if (job.status === jobStatus.FAILED) {
                         clearInterval(interval);
-                        assert.deepEqual(job.query, expectedQuery);
+                        validateExpectedResponse(job.query, expectedQuery);
                         done();
                     } else if (job.status === jobStatus.DONE || job.status === jobStatus.CANCELLED) {
                         clearInterval(interval);
@@ -875,6 +893,219 @@ describe('Batch API fallback job', function () {
         });
     });
 
+    describe('"onerror" should not be triggered for any query and "skipped"', function () {
+        var fallbackJob = {};
+
+        it('should create a job', function (done) {
+            assert.response(app, {
+                url: '/api/v2/sql/job?api_key=1234',
+                headers: {
+                    'Content-Type': 'application/x-www-form-urlencoded',
+                    'host': 'vizzuality.cartodb.com'
+                },
+                method: 'POST',
+                data: querystring.stringify({
+                    query: {
+                        query: [{
+                            query: "SELECT * FROM untitle_table_4 limit 1",
+                            onerror: "SELECT * FROM untitle_table_4 limit 2"
+                        }, {
+                            query: "SELECT * FROM untitle_table_4 limit 3",
+                            onerror: "SELECT * FROM untitle_table_4 limit 4"
+                        }]
+                    }
+                })
+            }, {
+                status: 201
+            }, function (res, err) {
+                if (err) {
+                    return done(err);
+                }
+                fallbackJob = JSON.parse(res.body);
+                done();
+            });
+        });
+
+        it('job should be failed', function (done) {
+            var expectedQuery = {
+                query: [{
+                    query: 'SELECT * FROM untitle_table_4 limit 1',
+                    onerror: 'SELECT * FROM untitle_table_4 limit 2',
+                    status: 'done',
+                    fallback_status: 'skipped'
+                }, {
+                    query: 'SELECT * FROM untitle_table_4 limit 3',
+                    onerror: 'SELECT * FROM untitle_table_4 limit 4',
+                    status: 'done',
+                    fallback_status: 'skipped'
+                }]
+            };
+
+            var interval = setInterval(function () {
+                assert.response(app, {
+                    url: '/api/v2/sql/job/' + fallbackJob.job_id + '?api_key=1234&',
+                    headers: {
+                        'Content-Type': 'application/x-www-form-urlencoded',
+                        'host': 'vizzuality.cartodb.com'
+                    },
+                    method: 'GET'
+                }, {
+                    status: 200
+                }, function (res, err) {
+                    if (err) {
+                        return done(err);
+                    }
+                    var job = JSON.parse(res.body);
+                    if (job.status === jobStatus.DONE) {
+                        clearInterval(interval);
+                        validateExpectedResponse(job.query, expectedQuery);
+                        done();
+                    } else if (job.status === jobStatus.FAILED || job.status === jobStatus.CANCELLED) {
+                        clearInterval(interval);
+                        done(new Error('Job ' + job.job_id + ' is ' + job.status + ', expected to be done'));
+                    }
+                });
+            }, 50);
+        });
+    });
+
+    describe('"onsuccess" should be "skipped"', function () {
+        var fallbackJob = {};
+
+        it('should create a job', function (done) {
+            assert.response(app, {
+                url: '/api/v2/sql/job?api_key=1234',
+                headers: {
+                    'Content-Type': 'application/x-www-form-urlencoded',
+                    'host': 'vizzuality.cartodb.com'
+                },
+                method: 'POST',
+                data: querystring.stringify({
+                    query: {
+                        query: [{
+                            query: "SELECT * FROM untitle_table_4 limit 1, /* should fail */",
+                            onsuccess: "SELECT * FROM untitle_table_4 limit 2"
+                        }]
+                    }
+                })
+            }, {
+                status: 201
+            }, function (res, err) {
+                if (err) {
+                    return done(err);
+                }
+                fallbackJob = JSON.parse(res.body);
+                done();
+            });
+        });
+
+        it('job should be failed', function (done) {
+            var expectedQuery = {
+                query: [{
+                    query: 'SELECT * FROM untitle_table_4 limit 1, /* should fail */',
+                    onsuccess: 'SELECT * FROM untitle_table_4 limit 2',
+                    status: 'failed',
+                    fallback_status: 'skipped',
+                    failed_reason: 'syntax error at end of input'
+                }]
+            };
+
+            var interval = setInterval(function () {
+                assert.response(app, {
+                    url: '/api/v2/sql/job/' + fallbackJob.job_id + '?api_key=1234&',
+                    headers: {
+                        'Content-Type': 'application/x-www-form-urlencoded',
+                        'host': 'vizzuality.cartodb.com'
+                    },
+                    method: 'GET'
+                }, {
+                    status: 200
+                }, function (res, err) {
+                    if (err) {
+                        return done(err);
+                    }
+                    var job = JSON.parse(res.body);
+                    if (job.status === jobStatus.FAILED) {
+                        clearInterval(interval);
+                        validateExpectedResponse(job.query, expectedQuery);
+                        done();
+                    } else if (job.status === jobStatus.DONE || job.status === jobStatus.CANCELLED) {
+                        clearInterval(interval);
+                        done(new Error('Job ' + job.job_id + ' is ' + job.status + ', expected to be failed'));
+                    }
+                });
+            }, 50);
+        });
+    });
+
+
+    describe('"onsuccess" should not be triggered and "skipped"', function () {
+        var fallbackJob = {};
+
+        it('should create a job', function (done) {
+            assert.response(app, {
+                url: '/api/v2/sql/job?api_key=1234',
+                headers: {
+                    'Content-Type': 'application/x-www-form-urlencoded',
+                    'host': 'vizzuality.cartodb.com'
+                },
+                method: 'POST',
+                data: querystring.stringify({
+                    query: {
+                        query: [{
+                            query: "SELECT * FROM untitle_table_4 limit 1, /* should fail */",
+                        }],
+                        onsuccess: "SELECT * FROM untitle_table_4 limit 2"
+                    }
+                })
+            }, {
+                status: 201
+            }, function (res, err) {
+                if (err) {
+                    return done(err);
+                }
+                fallbackJob = JSON.parse(res.body);
+                done();
+            });
+        });
+
+        it('job should be failed', function (done) {
+            var expectedQuery = {
+                query: [{
+                    query: 'SELECT * FROM untitle_table_4 limit 1, /* should fail */',
+                    status: 'failed',
+                    failed_reason: 'syntax error at end of input'
+                }],
+                onsuccess: 'SELECT * FROM untitle_table_4 limit 2'
+            };
+
+            var interval = setInterval(function () {
+                assert.response(app, {
+                    url: '/api/v2/sql/job/' + fallbackJob.job_id + '?api_key=1234&',
+                    headers: {
+                        'Content-Type': 'application/x-www-form-urlencoded',
+                        'host': 'vizzuality.cartodb.com'
+                    },
+                    method: 'GET'
+                }, {
+                    status: 200
+                }, function (res, err) {
+                    if (err) {
+                        return done(err);
+                    }
+                    var job = JSON.parse(res.body);
+                    if (job.status === jobStatus.FAILED && job.fallback_status === jobStatus.SKIPPED) {
+                        clearInterval(interval);
+                        validateExpectedResponse(job.query, expectedQuery);
+                        done();
+                    } else if (job.status === jobStatus.DONE || job.status === jobStatus.CANCELLED) {
+                        clearInterval(interval);
+                        done(new Error('Job ' + job.job_id + ' is ' + job.status + ', expected to be failed'));
+                    }
+                });
+            }, 50);
+        });
+    });
 
     describe('"onsuccess" for first query should fail', function () {
         var fallbackJob = {};
@@ -942,7 +1173,7 @@ describe('Batch API fallback job', function () {
                     var job = JSON.parse(res.body);
                     if (job.status === jobStatus.DONE) {
                         clearInterval(interval);
-                        assert.deepEqual(job.query, expectedQuery);
+                        validateExpectedResponse(job.query, expectedQuery);
                         done();
                     } else if (job.status === jobStatus.FAILED || job.status === jobStatus.CANCELLED) {
                         clearInterval(interval);
@@ -1019,7 +1250,7 @@ describe('Batch API fallback job', function () {
                     var job = JSON.parse(res.body);
                     if (job.status === jobStatus.DONE) {
                         clearInterval(interval);
-                        assert.deepEqual(job.query, expectedQuery);
+                        validateExpectedResponse(job.query, expectedQuery);
                         done();
                     } else if (job.status === jobStatus.FAILED || job.status === jobStatus.CANCELLED) {
                         clearInterval(interval);
@@ -1097,7 +1328,7 @@ describe('Batch API fallback job', function () {
                     var job = JSON.parse(res.body);
                     if (job.status === jobStatus.DONE && job.fallback_status === jobStatus.DONE) {
                         clearInterval(interval);
-                        assert.deepEqual(job.query, expectedQuery);
+                        validateExpectedResponse(job.query, expectedQuery);
                         done();
                     } else if (job.status === jobStatus.FAILED || job.status === jobStatus.CANCELLED) {
                         clearInterval(interval);
@@ -1177,7 +1408,7 @@ describe('Batch API fallback job', function () {
                     var job = JSON.parse(res.body);
                     if (job.status === jobStatus.DONE && job.fallback_status === jobStatus.DONE) {
                         clearInterval(interval);
-                        assert.deepEqual(job.query, expectedQuery);
+                        validateExpectedResponse(job.query, expectedQuery);
                         done();
                     } else if (job.status === jobStatus.FAILED || job.status === jobStatus.CANCELLED) {
                         clearInterval(interval);
@@ -1248,13 +1479,13 @@ describe('Batch API fallback job', function () {
                     var job = JSON.parse(res.body);
                     if (job.status === jobStatus.RUNNING && job.fallback_status === jobStatus.PENDING) {
                         clearInterval(interval);
-                        assert.deepEqual(job.query, expectedQuery);
+                        validateExpectedResponse(job.query, expectedQuery);
                         done();
                     } else if (job.status === jobStatus.DONE ||
                             job.status === jobStatus.FAILED ||
                             job.status === jobStatus.CANCELLED) {
                         clearInterval(interval);
-                        done(new Error('Job ' + job.job_id + ' is ' + job.status + ', expected to be done'));
+                        done(new Error('Job ' + job.job_id + ' is ' + job.status + ', expected to be running'));
                     }
                 });
             }, 50);
@@ -1266,7 +1497,7 @@ describe('Batch API fallback job', function () {
                     "query": "SELECT pg_sleep(3)",
                     "onsuccess": "SELECT pg_sleep(0)",
                     "status": "cancelled",
-                    "fallback_status": "pending"
+                    "fallback_status": "skipped"
                 }],
                 "onsuccess": "SELECT pg_sleep(0)"
             };
@@ -1285,8 +1516,8 @@ describe('Batch API fallback job', function () {
                     return done(err);
                 }
                 var job = JSON.parse(res.body);
-                if (job.status === jobStatus.CANCELLED && job.fallback_status === jobStatus.PENDING) {
+                if (job.status === jobStatus.CANCELLED && job.fallback_status === jobStatus.SKIPPED) {
-                    assert.deepEqual(job.query, expectedQuery);
+                    validateExpectedResponse(job.query, expectedQuery);
                     done();
                 } else if (job.status === jobStatus.DONE || job.status === jobStatus.FAILED) {
                     done(new Error('Job ' + job.job_id + ' is ' + job.status + ', expected to be cancelled'));
@@ -1355,7 +1586,7 @@ describe('Batch API fallback job', function () {
                     if (job.query.query[0].status === jobStatus.DONE &&
                         job.query.query[0].fallback_status === jobStatus.RUNNING) {
                         clearInterval(interval);
-                        assert.deepEqual(job.query, expectedQuery);
+                        validateExpectedResponse(job.query, expectedQuery);
                         done();
                     } else if (job.query.query[0].status === jobStatus.DONE ||
                             job.query.query[0].status === jobStatus.FAILED ||
@@ -1394,8 +1625,8 @@ describe('Batch API fallback job', function () {
                     return done(err);
                 }
                 var job = JSON.parse(res.body);
-                if (job.status === jobStatus.CANCELLED && job.fallback_status === jobStatus.PENDING) {
+                if (job.status === jobStatus.CANCELLED && job.fallback_status === jobStatus.SKIPPED) {
-                    assert.deepEqual(job.query, expectedQuery);
+                    validateExpectedResponse(job.query, expectedQuery);
                     done();
                 } else if (job.status === jobStatus.DONE || job.status === jobStatus.FAILED) {
                     done(new Error('Job ' + job.job_id + ' is ' + job.status + ', expected to be cancelled'));
@@ -1403,4 +1634,165 @@ describe('Batch API fallback job', function () {
             });
         });
     });
+
+    describe('should fail first "onerror" and job "onerror" and skip the other ones', function () {
+        var fallbackJob = {};
+
+        it('should create a job', function (done) {
+            assert.response(app, {
+                url: '/api/v2/sql/job?api_key=1234',
+                headers: {
+                    'Content-Type': 'application/x-www-form-urlencoded',
+                    'host': 'vizzuality.cartodb.com'
+                },
+                method: 'POST',
+                data: querystring.stringify({
+                    "query": {
+                        "query": [{
+                            "query": "SELECT * FROM atm_madrid limit 1, should fail",
+                            "onerror": "SELECT * FROM atm_madrid limit 2"
+                        }, {
+                            "query": "SELECT * FROM atm_madrid limit 3",
+                            "onerror": "SELECT * FROM atm_madrid limit 4"
+                        }],
+                        "onerror": "SELECT * FROM atm_madrid limit 5"
+                    }
+                })
+            }, {
+                status: 201
+            }, function (res, err) {
+                if (err) {
+                    return done(err);
+                }
+                fallbackJob = JSON.parse(res.body);
+                done();
+            });
+        });
+
+        it('job should fail', function (done) {
+            var expectedQuery = {
+                query: [{
+                    query: 'SELECT * FROM atm_madrid limit 1, should fail',
+                    onerror: 'SELECT * FROM atm_madrid limit 2',
+                    status: 'failed',
+                    fallback_status: 'failed',
+                    failed_reason: 'relation "atm_madrid" does not exist'
+                }, {
+                    query: 'SELECT * FROM atm_madrid limit 3',
+                    onerror: 'SELECT * FROM atm_madrid limit 4',
+                    status: 'skipped',
+                    fallback_status: 'skipped'
+                }],
+                onerror: 'SELECT * FROM atm_madrid limit 5'
+            };
+
+            var interval = setInterval(function () {
+                assert.response(app, {
+                    url: '/api/v2/sql/job/' + fallbackJob.job_id + '?api_key=1234&',
+                    headers: {
+                        'Content-Type': 'application/x-www-form-urlencoded',
+                        'host': 'vizzuality.cartodb.com'
+                    },
+                    method: 'GET'
+                }, {
+                    status: 200
+                }, function (res, err) {
+                    if (err) {
+                        return done(err);
+                    }
+                    var job = JSON.parse(res.body);
+                    if (job.status === jobStatus.FAILED && job.fallback_status === jobStatus.FAILED) {
+                        clearInterval(interval);
+                        validateExpectedResponse(job.query, expectedQuery);
+                        done();
+                    } else if (job.status === jobStatus.DONE || job.status === jobStatus.CANCELLED) {
+                        clearInterval(interval);
+                        done(new Error('Job ' + job.job_id + ' is ' + job.status + ', expected to be failed'));
+                    }
+                });
+            }, 50);
+        });
+    });
+
+    describe('should run first "onerror" and job "onerror" and skip the other ones', function () {
+        var fallbackJob = {};
+
+        it('should create a job', function (done) {
+            assert.response(app, {
+                url: '/api/v2/sql/job?api_key=1234',
+                headers: {
+                    'Content-Type': 'application/x-www-form-urlencoded',
+                    'host': 'vizzuality.cartodb.com'
+                },
+                method: 'POST',
+                data: querystring.stringify({
+                    "query": {
+                        "query": [{
+                            "query": "SELECT * FROM untitle_table_4 limit 1, should fail",
+                            "onerror": "SELECT * FROM untitle_table_4 limit 2"
+                        }, {
+                            "query": "SELECT * FROM untitle_table_4 limit 3",
+                            "onerror": "SELECT * FROM untitle_table_4 limit 4"
+                        }],
+                        "onerror": "SELECT * FROM untitle_table_4 limit 5"
+                    }
+                })
+            }, {
+                status: 201
+            }, function (res, err) {
+                if (err) {
+                    return done(err);
+                }
+                fallbackJob = JSON.parse(res.body);
+                done();
+            });
+        });
+
+        it('job should fail', function (done) {
+            var expectedQuery = {
+                "query": [
+                  {
+                    "query": "SELECT * FROM untitle_table_4 limit 1, should fail",
+                    "onerror": "SELECT * FROM untitle_table_4 limit 2",
+                    "status": "failed",
+                    "fallback_status": "done",
+                    "failed_reason": "LIMIT #,# syntax is not supported"
+                  },
+                  {
+                    "query": "SELECT * FROM untitle_table_4 limit 3",
+                    "onerror": "SELECT * FROM untitle_table_4 limit 4",
+                    "status": "skipped",
+                    "fallback_status": "skipped"
+                  }
+                ],
+                "onerror": "SELECT * FROM untitle_table_4 limit 5"
+            };
+
+            var interval = setInterval(function () {
+                assert.response(app, {
+                    url: '/api/v2/sql/job/' + fallbackJob.job_id + '?api_key=1234&',
+                    headers: {
+                        'Content-Type': 'application/x-www-form-urlencoded',
+                        'host': 'vizzuality.cartodb.com'
+                    },
+                    method: 'GET'
+                }, {
+                    status: 200
+                }, function (res, err) {
+                    if (err) {
+                        return done(err);
+                    }
+                    var job = JSON.parse(res.body);
+                    if (job.status === jobStatus.FAILED && job.fallback_status === jobStatus.DONE) {
+                        clearInterval(interval);
+                        validateExpectedResponse(job.query, expectedQuery);
+                        done();
+                    } else if (job.status === jobStatus.DONE || job.status === jobStatus.CANCELLED) {
+                        clearInterval(interval);
+                        done(new Error('Job ' + job.job_id + ' is ' + job.status + ', expected to be failed'));
+                    }
+                });
+            }, 50);
+        });
+    });
 });

test/acceptance/job.timing.test.js

@@ -0,0 +1,201 @@
+require('../helper');
+
+var assert = require('../support/assert');
+var app = require(global.settings.app_root + '/app/app')();
+var querystring = require('qs');
+var metadataBackend = require('cartodb-redis')({
+    host: global.settings.redis_host,
+    port: global.settings.redis_port,
+    max: global.settings.redisPool,
+    idleTimeoutMillis: global.settings.redisIdleTimeoutMillis,
+    reapIntervalMillis: global.settings.redisReapIntervalMillis
+});
+var batchFactory = require('../../batch');
+var jobStatus = require('../../batch/job_status');
+
+describe('Batch API query timing', function () {
+
+    function createJob(jobDefinition, callback) {
+        assert.response(app, {
+            url: '/api/v2/sql/job?api_key=1234',
+            headers: {
+                'Content-Type': 'application/x-www-form-urlencoded',
+                host: 'vizzuality.cartodb.com'
+            },
+            method: 'POST',
+            data: querystring.stringify(jobDefinition)
+        }, {
+            status: 201
+        }, function (res, err) {
+            if (err) {
+                return callback(err);
+            }
+            return callback(null, JSON.parse(res.body));
+        });
+    }
+
+    function getJobStatus(jobId, callback) {
+        assert.response(app, {
+            url: '/api/v2/sql/job/' + jobId + '?api_key=1234&',
+            headers: {
+                host: 'vizzuality.cartodb.com'
+            },
+            method: 'GET'
+        }, {
+            status: 200
+        }, function (res, err) {
+            if (err) {
+                return callback(err);
+            }
+            return callback(null, JSON.parse(res.body));
+        });
+    }
+
+    function validateExpectedResponse(actual, expected) {
+        actual.query.forEach(function(actualQuery, index) {
+            var expectedQuery = expected.query[index];
+            assert.ok(expectedQuery);
+            Object.keys(expectedQuery).forEach(function(expectedKey) {
+                assert.equal(actualQuery[expectedKey], expectedQuery[expectedKey]);
+            });
+            var propsToCheckDate = ['started_at', 'ended_at'];
+            propsToCheckDate.forEach(function(propToCheckDate) {
+                if (actualQuery.hasOwnProperty(propToCheckDate)) {
+                    assert.ok(new Date(actualQuery[propToCheckDate]));
+                }
+            });
+        });
+
+        assert.equal(actual.onsuccess, expected.onsuccess);
+        assert.equal(actual.onerror, expected.onerror);
+    }
+
+    var batch = batchFactory(metadataBackend);
+
+    before(function () {
+        batch.start();
+    });
+
+    after(function (done) {
+        batch.stop();
+        batch.drain(function () {
+            metadataBackend.redisCmd(5, 'DEL', [ 'batch:queues:localhost' ], done);
+        });
+    });
+
+    describe('should report start and end time for each query with fallback queries', function () {
+        var jobResponse;
+        before(function(done) {
+            createJob({
+                "query": {
+                    "query": [
+                        {
+                            "query": "SELECT * FROM untitle_table_4 limit 1",
+                            "onerror": "SELECT * FROM untitle_table_4 limit 2"
+                        },
+                        {
+                            "query": "SELECT * FROM untitle_table_4 limit 3",
+                            "onerror": "SELECT * FROM untitle_table_4 limit 4"
+                        }
+                    ],
+                    "onerror": "SELECT * FROM untitle_table_4 limit 5"
+                }
+            }, function(err, job) {
+                jobResponse = job;
+                return done(err);
+            });
+        });
+
+        it('should expose started_at and ended_at for all queries with fallback mechanism', function (done) {
+            var expectedQuery = {
+                query: [{
+                    query: 'SELECT * FROM untitle_table_4 limit 1',
+                    onerror: 'SELECT * FROM untitle_table_4 limit 2',
+                    status: 'done',
+                    fallback_status: 'skipped'
+                }, {
+                    query: 'SELECT * FROM untitle_table_4 limit 3',
+                    onerror: 'SELECT * FROM untitle_table_4 limit 4',
+                    status: 'done',
+                    fallback_status: 'skipped'
+                }],
+                onerror: 'SELECT * FROM untitle_table_4 limit 5'
+            };
+
+            var interval = setInterval(function () {
+                getJobStatus(jobResponse.job_id, function(err, job) {
+                    if (job.status === jobStatus.DONE) {
+                        clearInterval(interval);
+                        validateExpectedResponse(job.query, expectedQuery);
+                        job.query.query.forEach(function(actualQuery) {
+                            assert.ok(actualQuery.started_at);
+                            assert.ok(actualQuery.ended_at);
+                        });
+                        done();
+                    } else if (job.status === jobStatus.FAILED || job.status === jobStatus.CANCELLED) {
+                        clearInterval(interval);
+                        done(new Error('Job ' + job.job_id + ' is ' + job.status + ', expected to be "done"'));
+                    }
+                });
+            }, 50);
+        });
+    });
+
+    describe('should report start and end time for each query also for failing queries', function () {
+        var jobResponse;
+        before(function(done) {
+            createJob({
+                "query": {
+                    "query": [
+                        {
+                            "query": "SELECT * FROM untitle_table_4 limit 1",
+                            "onerror": "SELECT * FROM untitle_table_4 limit 2"
+                        },
+                        {
+                            "query": "SELECT * FROM untitle_table_4 limit 3 failed",
+                            "onerror": "SELECT * FROM untitle_table_4 limit 4"
+                        }
+                    ],
+                    "onerror": "SELECT * FROM untitle_table_4 limit 5"
+                }
+            }, function(err, job) {
+                jobResponse = job;
+                return done(err);
+            });
+        });
+
+        it('should expose started_at and ended_at for all queries with fallback mechanism (failed)', function (done) {
+            var expectedQuery = {
+                query: [{
+                    query: 'SELECT * FROM untitle_table_4 limit 1',
+                    onerror: 'SELECT * FROM untitle_table_4 limit 2',
+                    status: 'done',
+                    fallback_status: 'skipped'
+                }, {
+                    query: 'SELECT * FROM untitle_table_4 limit 3 failed',
+                    onerror: 'SELECT * FROM untitle_table_4 limit 4',
+                    status: 'failed',
+                    fallback_status: 'done'
+                }],
+                onerror: 'SELECT * FROM untitle_table_4 limit 5'
+            };
+
+            var interval = setInterval(function () {
+                getJobStatus(jobResponse.job_id, function(err, job) {
+                    if (job.status === jobStatus.FAILED) {
+                        clearInterval(interval);
+                        validateExpectedResponse(job.query, expectedQuery);
+                        job.query.query.forEach(function(actualQuery) {
+                            assert.ok(actualQuery.started_at);
+                            assert.ok(actualQuery.ended_at);
+                        });
+                        done();
+                    } else if (job.status === jobStatus.DONE || job.status === jobStatus.CANCELLED) {
+                        clearInterval(interval);
+                        done(new Error('Job ' + job.job_id + ' is ' + job.status + ', expected to be "failed"'));
+                    }
+                });
+            }, 50);
+        });
+    });
+});

test/prepare_db.sh

@@ -65,31 +65,34 @@ export PGHOST PGPORT
 if test x"$PREPARE_PGSQL" = xyes; then
 
   echo "preparing postgres..."
+  echo "PostgreSQL server version: `psql -A -t -c 'select version()'`"
   dropdb ${TEST_DB} # 2> /dev/null # error expected if doesn't exist, but not otherwise
   createdb -Ttemplate_postgis -EUTF8 ${TEST_DB} || die "Could not create test database"
-  psql -c 'CREATE EXTENSION "uuid-ossp";' ${TEST_DB}
+  psql -c 'CREATE EXTENSION IF NOT EXISTS "uuid-ossp";' ${TEST_DB}
-  cat test.sql |
+  psql -c "CREATE EXTENSION IF NOT EXISTS plpythonu;" ${TEST_DB}
-    sed "s/:PUBLICUSER/${PUBLICUSER}/" |
-    sed "s/:PUBLICPASS/${PUBLICPASS}/" |
-    sed "s/:TESTUSER/${TESTUSER}/" |
-    sed "s/:TESTPASS/${TESTPASS}/" |
-    psql -v ON_ERROR_STOP=1 ${TEST_DB} || exit 1
 
-  echo "Populating windshaft_test database with reduced populated places data"
+  LOCAL_SQL_SCRIPTS='test populated_places_simple_reduced'
-  cat ./fixtures/populated_places_simple_reduced.sql |
+  REMOTE_SQL_SCRIPTS='CDB_QueryStatements CDB_QueryTables CDB_CartodbfyTable CDB_TableMetadata CDB_ForeignTable CDB_UserTables CDB_ColumnNames CDB_ZoomFromScale CDB_OverviewsSupport CDB_Overviews'
-    sed "s/:PUBLICUSER/${PUBLICUSER}/" |
-    sed "s/:PUBLICPASS/${PUBLICPASS}/" |
-    sed "s/:TESTUSER/${TESTUSER}/" |
-    sed "s/:TESTPASS/${TESTPASS}/" |
-    psql -v ON_ERROR_STOP=1 ${TEST_DB} || exit 1
 
-  # TODO: send in a single run, togheter with test.sql
+  CURL_ARGS=""
-  psql -c "CREATE EXTENSION plpythonu;" ${TEST_DB}
+  for i in ${REMOTE_SQL_SCRIPTS}
-  for i in CDB_QueryStatements CDB_QueryTables CDB_CartodbfyTable CDB_TableMetadata CDB_ForeignTable CDB_UserTables CDB_ColumnNames CDB_ZoomFromScale CDB_Overviews
   do
-    curl -L -s https://github.com/CartoDB/cartodb-postgresql/raw/master/scripts-available/$i.sql -o support/$i.sql
+    CURL_ARGS="${CURL_ARGS}\"https://github.com/CartoDB/cartodb-postgresql/raw/master/scripts-available/$i.sql\" -o support/sql/$i.sql "
-    cat support/$i.sql | sed -e 's/cartodb\./public./g' -e "s/''cartodb''/''public''/g" \
+  done
-        | psql -v ON_ERROR_STOP=1 ${TEST_DB} || exit 1
+  echo "Downloading and updating: ${REMOTE_SQL_SCRIPTS}"
+  echo ${CURL_ARGS} | xargs curl -L -s
+
+  psql -c "CREATE EXTENSION IF NOT EXISTS plpythonu;" ${TEST_DB}
+  ALL_SQL_SCRIPTS="${REMOTE_SQL_SCRIPTS} ${LOCAL_SQL_SCRIPTS}"
+  for i in ${ALL_SQL_SCRIPTS}
+  do
+    cat support/sql/${i}.sql |
+      sed -e 's/cartodb\./public./g' -e "s/''cartodb''/''public''/g" |
+      sed "s/:PUBLICUSER/${PUBLICUSER}/" |
+      sed "s/:PUBLICPASS/${PUBLICPASS}/" |
+      sed "s/:TESTUSER/${TESTUSER}/" |
+      sed "s/:TESTPASS/${TESTPASS}/" |
+      PGOPTIONS='--client-min-messages=WARNING' psql -q -v ON_ERROR_STOP=1 ${TEST_DB} > /dev/null || exit 1
   done
 
 fi

test/unit/batch/job_queue.js

@@ -11,29 +11,29 @@ describe('batch API job queue', function () {
                 });
             }
         };
-        this.jobQueue = new JobQueue(this.metadataBackend);
+        this.jobPublisher = {
+            publish: function () {}
+        };
+        this.jobQueue = new JobQueue(this.metadataBackend, this.jobPublisher);
     });
 
     it('.enqueue() should enqueue the provided job', function (done) {
-        this.jobQueue.enqueue('irrelevantJob', 'irrelevantHost', function (err, username) {
+        this.jobQueue.enqueue('irrelevantJob', 'irrelevantHost', function (err) {
             assert.ok(!err);
-            assert.equal(username, 'irrelevantJob');
             done();
         });
     });
 
     it('.dequeue() should dequeue the next job', function (done) {
-        this.jobQueue.dequeue('irrelevantHost', function (err, username) {
+        this.jobQueue.dequeue('irrelevantHost', function (err) {
             assert.ok(!err);
-            assert.equal(username, 'irrelevantJob');
             done();
         });
     });
 
     it('.enqueueFirst() should dequeue the next job', function (done) {
-        this.jobQueue.enqueueFirst('irrelevantJob', 'irrelevantHost', function (err, username) {
+        this.jobQueue.enqueueFirst('irrelevantJob', 'irrelevantHost', function (err) {
             assert.ok(!err);
-            assert.equal(username, 'irrelevantJob');
             done();
         });
     });

test/unit/oauth.test.js

@@ -78,7 +78,7 @@ it('test can access oauth hash for a user based on access token (oauth_token)',
 });
 
 it('test non existant oauth hash for a user based on oauth_token returns empty hash', function(done){
-    var req = {query:{}, headers:{authorization:full_oauth_header}};
+    var req = {query:{}, params: { user: 'vizzuality' }, headers:{authorization:full_oauth_header}};
     var tokens = oAuth.parseTokens(req);
 
     oAuth.getOAuthHash(metadataBackend, tokens.oauth_token, function(err, data){
@@ -91,6 +91,7 @@ it('test non existant oauth hash for a user based on oauth_token returns empty h
 it('can return user for verified signature', function(done){
     var req = {query:{},
         headers:{authorization:real_oauth_header, host: 'vizzuality.testhost.lan' },
+        params: { user: 'vizzuality' },
         protocol: 'http',
         method: 'GET',
         path: '/api/v1/tables'
@@ -103,9 +104,31 @@ it('can return user for verified signature', function(done){
     });
 });
 
+it('can return user for verified signature (for other allowed domains)', function(done){
+    var oAuthGetAllowedHostsFn = oAuth.getAllowedHosts;
+    oAuth.getAllowedHosts = function() {
+        return ['testhost.lan', 'testhostdb.lan'];
+    };
+    var req = {query:{},
+        headers:{authorization:real_oauth_header, host: 'vizzuality.testhostdb.lan' },
+        params: { user: 'vizzuality' },
+        protocol: 'http',
+        method: 'GET',
+        path: '/api/v1/tables'
+    };
+
+    oAuth.verifyRequest(req, metadataBackend, function(err, data){
+        oAuth.getAllowedHosts = oAuthGetAllowedHostsFn;
+        assert.ok(!err, err);
+        assert.equal(data, 1);
+        done();
+    });
+});
+
 it('returns null user for unverified signatures', function(done){
     var req = {query:{},
         headers:{authorization:real_oauth_header, host: 'vizzuality.testyhost.lan' },
+        params: { user: 'vizzuality' },
         protocol: 'http',
         method: 'GET',
         path: '/api/v1/tables'
@@ -121,6 +144,7 @@ it('returns null user for no oauth', function(done){
     var req = {
         query:{},
         headers:{},
+        params: { user: 'vizzuality' },
         protocol: 'http',
         method: 'GET',
         path: '/api/v1/tables'