implement 0.7.1 migration

Author: kebetsi

components/storage/src/migrations/0.7.1.js

@@ -1,3 +1,72 @@
+var async = require('async'),
+  toString = require('components/utils').toString;
+
 /**
- * Created by iliakebets on 23.01.17.
+ * v0.7.1:
+ *
+ * - Fixes streams with parentId='' to parentId=null
  */
+module.exports = function (context, callback) {
+  context.database.getCollection({name: 'users'}, function (err, usersCol) {
+    if (err) { return callback(err); }
+
+    usersCol.find({}).toArray(function (err, users) {
+      if (err) { return callback(err); }
+
+      async.forEachSeries(users, migrateUser, function (err) {
+        if (err) { return callback(err); }
+
+        context.logInfo('Data version is now 0.7.1');
+        callback();
+      });
+    });
+  });
+
+  function migrateUser(user, callback) {
+    context.logInfo('Migrating user ' + toString.user(user) + '...');
+    async.series([
+      function updateStreamsStructure(stepDone) {
+        context.database.getCollection({name: user._id + '.streams'}, function (err, streamsCol) {
+          if (err) {
+            context.logError(err, 'retrieving streams collection');
+            return stepDone(err);
+          }
+
+          var streamsCursor = streamsCol.find(),
+            completed = false;
+          async.until(function () { return completed; }, migrateStreams,
+            context.stepCallbackFn('migrating events structure', stepDone));
+
+          function migrateStreams(streamDone) {
+            streamsCursor.nextObject(function (err, stream) {
+              if (err) { return setImmediate(streamDone.bind(null, err)); }
+              if (! stream) {
+                completed = true;
+                return setImmediate(streamDone);
+              }
+
+              if (stream.parentId !== '') {
+                return setImmediate(streamDone);
+              }
+
+              var update = {
+                $set: {
+                  parentId: null
+                }
+              };
+
+              streamsCol.update({_id: stream._id}, update, streamDone);
+            });
+          }
+        });
+      }
+    ], function (err) {
+      if (err) {
+        context.logError(err, 'migrating user');
+        return callback(err);
+      }
+      context.logInfo('Successfully migrated user ' + toString.user(user) + '.');
+      callback();
+    });
+  }
+};

components/storage/src/migrations/index.js

@@ -9,5 +9,6 @@ module.exports = {
   '0.3.0': require('./0.3.0.js'),
   '0.4.0': require('./0.4.0.js'),
   '0.5.0': require('./0.5.0.js'),
-  '0.7.0': require('./0.7.0.js')
+  '0.7.0': require('./0.7.0.js'),
+  '0.7.1': require('./0.7.1.js')
 };