Comments. Bug fixes. Test fixes.

Author: mrtcode

lib/controller.js

@@ -12,7 +12,6 @@ var trace = require('debug')('arkivo:trace');
 var B = require('bluebird');
 
 var config = require('./config').controller;
-var zotero = require('./zotero');
 var common = require('./common');
 
 var extend = common.extend;
@@ -29,6 +28,78 @@ var Listener = require('./listener');
 
 /** @module arkivo */
 
+/**
+ * # How it works?
+ * ## What happens on startup?
+ * 1. Arkivo is started.
+ * 2. Continues processing previous Kue jobs [subscribe, unusbscribe, sync].
+ * 3. Creates listener, connects to stream-server, subscribes to all
+ *    topics from all subscriptions.
+ * 4. Stream-server responds with subscriptionsCreated, and the listener
+ *    emits 'update' event for each subscription that is listening
+ *    for the specific topic and key pair.
+ * 5. Sync jobs are scheduled for each subscription. And this basically means
+ *    that on each stream-server (re)connect we (re)synchronize all subscriptions.
+ *
+ * ## What happens when creating a subscription?
+ * 1. The URL is validated to make sure it's an items url and we have access to it.
+ * 2. If the URL is ok, the new subscription is created and saved.
+ * 3. Two concurrent operations are issued:
+ *    subscribe to stream-server, and schedule the first sync job.
+ * 4. Listener receives 'subscriptionsCreated' response, and emits
+ *    'update' events only for subscriptions which are newly subscribed
+ *    to stream-server.
+ * 5. Each 'update' event should schedule a 'sync' job for the updated
+ *    subscription, but in practise the job will be skipped because,
+ *    it should already exists when created in point 4.
+ *
+ * ## What happens when a topic is updated?
+ * 1. Listener receives 'topicUpdated' from stream-server.
+ * 2. Each subscription that is listening for the updated topic
+ *    will get the 'update' event.
+ * 3. Each update event should schedule a 'sync' job for that subscription,
+ *    unless there already exists a 'sync' job for the subscription.
+ *
+ * ## What happens when removing subscription?
+ * 1. Destroys the actual subscription.
+ * 2. Removes the subscription from the listener.
+ * 3. Listener checks if there is no more subscriptions listening
+ *    for the same topic(+key) combination, and usubscribes from
+ *    stream-server if so.
+ *
+ * ---
+ *
+ * # Optional stream-server
+ * Stream-sever can be disabled by setting listen:false in config.
+ * ## What this would change?
+ * * On startup we only continue to do the previously scheduled jobs.
+ * * But we don't initialize the listener.
+ * * Therefore we don't add subscriptions to it.
+ * * Therefore the listener doesn't subscribe to stream-server.
+ * * Therefore we don't get subscriptionsCreated response from server,
+ *   which emits 'update' event and schedules 'sync' jobs.
+ * * But, 'sync' job is scheduled on subscription creation.
+ * * And also it can be triggered by `POST /api/sync` API request.
+ *
+ * ---
+ *
+ * # The Kue based 'sync' queue
+ * The Kue based sync queue is probably unnecessary and can be replaced with
+ * a simple array marking subscriptions that need to be updated:
+ *  var subscriptionsToUpdate = [
+ *    "vrrui780gb",
+ *    "olxbwld2t4",
+ *  ]
+ *
+ *  Especially because Kue doesn't have a simple way to deduplicate jobs.
+ *
+ *  The parameter 'skip' (to skip fetching already existing items)
+ *  can be saved together with the subscription,
+ *  therefore we would no longer need to persist the sync queue.
+ *  On each restart all subscriptions are resynchronized anyway.
+ *  Unless there is something else why we need the persisting sync queue?
+ *
+ */
 
 /**
  * @class Controller
@@ -103,13 +174,14 @@ Controller.prototype.subscribe = function (data, job) {
   // Theoretically the validated url can become invalid
   // when stream-server tries to create subscription.
   // But in practice this shouldn't be a problem.
-  return validateUrl(data.url, data.key)
+  return Subscription.validateItemsUrl(data.url, data.key)
     .then(function () {
       return subscription.save();
     })
     .then(function () {
-      if (this.options.listen)
+      if (this.options.listen) {
         this.listener.add(subscription);
+      }
       // If sync job would finish faster than stream-server
       // would add the new subscription, this could result in missed updates,
       // but each new stream-server subscription triggers another sync,
@@ -118,6 +190,7 @@ Controller.prototype.subscribe = function (data, job) {
         id: subscription.id,
         skip: data.skip
       }, 0, this.options.attempts);
+
       return subscription;
     }.bind(this));
 };
@@ -151,7 +224,7 @@ Controller.prototype.unsubscribe = function (data, job) {
     .call('destroy')
 
     .tap(function (s) {
-      if (this.options.listen) this.listener.remove(s);
+      if (this.options.listen) this.listener.remove(s.id);
     }.bind(this));
 };
 
@@ -206,7 +279,7 @@ Controller.prototype.synchronize = function (data, job) {
 
     subscriptions = Subscription
       .load(data.id)
-      .then(listify);
+      .then(function (subscription) { return [subscription]; });
   }
 
   return subscriptions
@@ -463,29 +536,5 @@ function progressor(job, total) {
 //   };
 // }
 
-function listify(i) { return [i]; }
-
-function validateUrl(url, key) {
-  return new B(function (resolve, reject) {
-    if (!(/^(\/?(users|groups)\/\d+\/?(publications\/items|items))/).test(url)) {
-      return reject(new Error('Invalid URL'));
-    }
-
-    var headers = {};
-    if (key) {
-      headers['Zotero-API-Key'] = key;
-    }
-    zotero.request({
-      method: 'get',
-      path: url,
-      headers: headers
-    }, null, function (error, resp) {
-      // This directly returns all dataserver errors
-      if (error) return reject(error);
-      resolve(resp);
-    });
-  });
-}
-
 // --- Exports ---
 module.exports = Controller;

lib/http/index.js

@@ -38,6 +38,8 @@ Server.prototype.initialize = function () {
 
   this.app = express();
 
+  this.app.set('json spaces', 2);
+
   this.app.use(this.options.api, q.app);
   this.app.use(this.options.api, routes.api);