diff --git a/client/app/app.styl b/client/app/app.styl index 42fd458..e5a1b0a 100644 --- a/client/app/app.styl +++ b/client/app/app.styl @@ -1,62 +1,63 @@ @import "font-awesome/css/font-awesome.css" @import "bootstrap/dist/css/bootstrap.css" // // Bootstrap Fonts // @font-face font-family: 'Glyphicons Halflings' src: url('../bower_components/bootstrap/fonts/glyphicons-halflings-regular.eot') src: url('../bower_components/bootstrap/fonts/glyphicons-halflings-regular.eot?#iefix') format('embedded-opentype'), url('../bower_components/bootstrap/fonts/glyphicons-halflings-regular.woff') format('woff'), url('../bower_components/bootstrap/fonts/glyphicons-halflings-regular.ttf') format('truetype'), url('../bower_components/bootstrap/fonts/glyphicons-halflings-regular.svg#glyphicons_halflingsregular') format('svg'); // // Font Awesome Fonts // @font-face font-family: 'FontAwesome' src: url('../bower_components/font-awesome/fonts/fontawesome-webfont.eot?v=4.1.0') src: url('../bower_components/font-awesome/fonts/fontawesome-webfont.eot?#iefix&v=4.1.0') format('embedded-opentype'), url('../bower_components/font-awesome/fonts/fontawesome-webfont.woff?v=4.1.0') format('woff'), url('../bower_components/font-awesome/fonts/fontawesome-webfont.ttf?v=4.1.0') format('truetype'), url('../bower_components/font-awesome/fonts/fontawesome-webfont.svg?v=4.1.0#fontawesomeregular') format('svg'); font-weight: normal font-style: normal // // App-wide Styles // .browsehappy background #ccc color #000 margin 0.2em 0 padding 0.2em 0 body background-color whitesmoke .text-strong font-weight bold // Component styles are injected through grunt // injector @import 'account/login/login.styl'; @import 'editor/editor.styl'; @import 'main/main.styl'; @import 'templates/templates.styl'; @import 'users/users.styl'; @import 'createMenu/createMenu.styl'; @import 'documentList/documentList.styl'; @import 'exportButton/exportButton.styl'; @import 'import/import.styl'; @import 'labelEditor/labelEditor.styl'; @import 'modal/modal.styl'; @import 'navbar/navbar.styl'; +@import 'saveButton/saveButton.styl'; @import 'titleEditor/titleEditor.styl'; @import 'wodo/editor.styl'; // endinjector diff --git a/client/app/editor/editor.jade b/client/app/editor/editor.jade index e56d102..7392f51 100644 --- a/client/app/editor/editor.jade +++ b/client/app/editor/editor.jade @@ -1,7 +1,9 @@ div.toolbar div.toolbar-item.title title-editor div.toolbar-item export-button + div.toolbar-item + save-button wodo-editor.wodo diff --git a/client/components/saveButton/saveButton.controller.js b/client/components/saveButton/saveButton.controller.js new file mode 100644 index 0000000..d0f6f13 --- /dev/null +++ b/client/components/saveButton/saveButton.controller.js @@ -0,0 +1,28 @@ +'use strict'; + +angular.module('manticoreApp') +.controller('SaveButtonCtrl', function ($scope, $timeout) { + $scope.label = 'Save'; + $scope.isSaving = false; + + $scope.save = function () { + $scope.label = 'Saving'; + $scope.isSaving = true; + + var socket = $scope.editor.clientAdaptor.getSocket(); + socket.emit('save', function (err) { + $timeout(function () { + if (err) { + $scope.label = 'Error while saving'; + } else { + $scope.label = 'Saved just now'; + } + }); + + $timeout(function () { + $scope.label = 'Save'; + $scope.isSaving = false; + }, 5000); + }); + }; +}); diff --git a/client/components/saveButton/saveButton.directive.js b/client/components/saveButton/saveButton.directive.js new file mode 100644 index 0000000..3aa2c27 --- /dev/null +++ b/client/components/saveButton/saveButton.directive.js @@ -0,0 +1,10 @@ +'use strict'; + +angular.module('manticoreApp') + .directive('saveButton', function () { + return { + templateUrl: 'components/saveButton/saveButton.html', + restrict: 'E', + controller: 'SaveButtonCtrl' + }; + }); diff --git a/client/components/saveButton/saveButton.jade b/client/components/saveButton/saveButton.jade new file mode 100644 index 0000000..642d4b0 --- /dev/null +++ b/client/components/saveButton/saveButton.jade @@ -0,0 +1,2 @@ +div.save-button + button.btn.btn-primary(type='button' ng-click='save()' ng-disabled='isSaving') {{label}} diff --git a/client/components/saveButton/saveButton.styl b/client/components/saveButton/saveButton.styl new file mode 100644 index 0000000..a52ebc5 --- /dev/null +++ b/client/components/saveButton/saveButton.styl @@ -0,0 +1,3 @@ +.save-button + padding: 10px 10px; + height: 50px; diff --git a/client/components/wodo/adaptor.service.js b/client/components/wodo/adaptor.service.js index 1ac97d6..63c5bcb 100644 --- a/client/components/wodo/adaptor.service.js +++ b/client/components/wodo/adaptor.service.js @@ -1,285 +1,298 @@ /*jslint unparam: true*/ /*global runtime, core, ops, io*/ 'use strict'; angular.module('manticoreApp') .factory('Adaptor', function () { var OperationRouter = function (socket, odfContainer, errorCb) { var EVENT_BEFORESAVETOFILE = 'beforeSaveToFile', EVENT_SAVEDTOFILE = 'savedToFile', EVENT_HASLOCALUNSYNCEDOPERATIONSCHANGED = 'hasLocalUnsyncedOperationsChanged', EVENT_HASSESSIONHOSTCONNECTIONCHANGED = 'hasSessionHostConnectionChanged', EVENT_MEMBERADDED = 'memberAdded', EVENT_MEMBERCHANGED = 'memberChanged', EVENT_MEMBERREMOVED = 'memberRemoved', eventNotifier = new core.EventNotifier([ EVENT_BEFORESAVETOFILE, EVENT_SAVEDTOFILE, EVENT_HASLOCALUNSYNCEDOPERATIONSCHANGED, EVENT_HASSESSIONHOSTCONNECTIONCHANGED, EVENT_MEMBERADDED, EVENT_MEMBERCHANGED, EVENT_MEMBERREMOVED, ops.OperationRouter.signalProcessingBatchStart, ops.OperationRouter.signalProcessingBatchEnd ]), operationFactory, playbackFunction, lastServerSyncHeadId = 0, sendClientOpspecsLock = false, sendClientOpspecsTask, hasSessionHostConnection = true, unplayedServerOpSpecQueue = [], unsyncedClientOpSpecQueue = [], operationTransformer = new ops.OperationTransformer(), /**@const*/sendClientOpspecsDelay = 300; function playbackOpspecs(opspecs) { var op, i; if (!opspecs.length) { return; } eventNotifier.emit(ops.OperationRouter.signalProcessingBatchStart, {}); for (i = 0; i < opspecs.length; i += 1) { op = operationFactory.create(opspecs[i]); if (op !== null) { if (!playbackFunction(op)) { eventNotifier.emit(ops.OperationRouter.signalProcessingBatchEnd, {}); errorCb('opExecutionFailure'); return; } } else { eventNotifier.emit(ops.OperationRouter.signalProcessingBatchEnd, {}); errorCb('Unknown opspec: ' + runtime.toJson(opspecs[i])); return; } } eventNotifier.emit(ops.OperationRouter.signalProcessingBatchEnd, {}); } function handleNewServerOpsWithUnsyncedClientOps(serverOps) { var transformResult = operationTransformer.transform(unsyncedClientOpSpecQueue, serverOps); if (!transformResult) { errorCb('Has unresolvable conflict.'); return false; } unsyncedClientOpSpecQueue = transformResult.opSpecsA; unplayedServerOpSpecQueue = unplayedServerOpSpecQueue.concat(transformResult.opSpecsB); return true; } function handleNewClientOpsWithUnplayedServerOps(clientOps) { var transformResult = operationTransformer.transform(clientOps, unplayedServerOpSpecQueue); if (!transformResult) { errorCb('Has unresolvable conflict.'); return false; } unsyncedClientOpSpecQueue = unsyncedClientOpSpecQueue.concat(transformResult.opSpecsA); unplayedServerOpSpecQueue = transformResult.opSpecsB; return true; } function receiveServerOpspecs(headId, serverOpspecs) { if (unsyncedClientOpSpecQueue.length > 0) { handleNewServerOpsWithUnsyncedClientOps(serverOpspecs); // could happen that ops from server make client ops obsolete if (unsyncedClientOpSpecQueue.length === 0) { eventNotifier.emit(EVENT_HASLOCALUNSYNCEDOPERATIONSCHANGED, false); } } else { // apply directly playbackOpspecs(serverOpspecs); } lastServerSyncHeadId = headId; } function sendClientOpspecs() { var originalUnsyncedLength = unsyncedClientOpSpecQueue.length; if (originalUnsyncedLength) { sendClientOpspecsLock = true; socket.emit('commit_ops', { head: lastServerSyncHeadId, ops: unsyncedClientOpSpecQueue }, function (response) { if (response.conflict === true) { sendClientOpspecs(); } else { lastServerSyncHeadId = response.head; // on success no other server ops should have sneaked in meanwhile, so no need to check // got no other client ops meanwhile? if (unsyncedClientOpSpecQueue.length === originalUnsyncedLength) { unsyncedClientOpSpecQueue.length = 0; // finally apply all server ops collected while waiting for sync playbackOpspecs(unplayedServerOpSpecQueue); unplayedServerOpSpecQueue.length = 0; eventNotifier.emit(EVENT_HASLOCALUNSYNCEDOPERATIONSCHANGED, false); sendClientOpspecsLock = false; } else { // send off the new client ops directly unsyncedClientOpSpecQueue.splice(0, originalUnsyncedLength); sendClientOpspecs(); } } }); } } this.setOperationFactory = function (f) { operationFactory = f; }; this.setPlaybackFunction = function (f) { playbackFunction = f; }; this.push = function (operations) { var clientOpspecs = [], now = Date.now(), hasLocalUnsyncedOpsBefore = (unsyncedClientOpSpecQueue.length !== 0), hasLocalUnsyncedOpsNow; operations.forEach(function(op) { var opspec = op.spec(); opspec.timestamp = now; clientOpspecs.push(opspec); }); playbackOpspecs(clientOpspecs); if (unplayedServerOpSpecQueue.length > 0) { handleNewClientOpsWithUnplayedServerOps(clientOpspecs); } else { unsyncedClientOpSpecQueue = unsyncedClientOpSpecQueue.concat(clientOpspecs); } hasLocalUnsyncedOpsNow = (unsyncedClientOpSpecQueue.length !== 0); if (hasLocalUnsyncedOpsNow !== hasLocalUnsyncedOpsBefore) { eventNotifier.emit(EVENT_HASLOCALUNSYNCEDOPERATIONSCHANGED, hasLocalUnsyncedOpsNow); } sendClientOpspecsTask.trigger(); }; this.requestReplay = function (cb) { var cbOnce = function () { eventNotifier.unsubscribe(ops.OperationRouter.signalProcessingBatchEnd, cbOnce); cb(); }; // hack: relies on at least addmember op being added for ourselves and being executed eventNotifier.subscribe(ops.OperationRouter.signalProcessingBatchEnd, cbOnce); socket.emit('replay', {}); }; this.close = function (cb) { cb(); }; this.subscribe = function (eventId, cb) { eventNotifier.subscribe(eventId, cb); }; this.unsubscribe = function (eventId, cb) { eventNotifier.unsubscribe(eventId, cb); }; this.hasLocalUnsyncedOps = function () { return unsyncedClientOpSpecQueue.length !== 0; }; this.hasSessionHostConnection = function () { return hasSessionHostConnection; }; function init() { + var replayed = false, + followupHead, + followupOps = []; + sendClientOpspecsTask = core.Task.createTimeoutTask(function () { if (!sendClientOpspecsLock) { sendClientOpspecs(); } }, sendClientOpspecsDelay); socket.on('replay', function (data) { receiveServerOpspecs(data.head, data.ops); + replayed = true; + if (followupHead && followupHead > data.head) { + receiveServerOpspecs(followupHead, followupOps); + } + }); - socket.on('new_ops', function (data) { + socket.on('new_ops', function (data) { + if (replayed) { receiveServerOpspecs(data.head, data.ops); - }); + } else { + followupHead = data.head; + followupOps = followupOps.concat(data.ops); + } }); } init(); }; var ClientAdaptor = function (documentId, authToken, connectedCb, kickedCb, disconnectedCb) { var self = this, memberId, genesisUrl, socket; this.getMemberId = function () { return memberId; }; this.getGenesisUrl = function () { return genesisUrl; }; this.createOperationRouter = function (odfContainer, errorCb) { runtime.assert(Boolean(memberId), 'You must be connected to a session before creating an operation router'); return new OperationRouter(socket, odfContainer, errorCb); }; this.joinSession = function (cb) { socket.on('join_success', function handleJoinSuccess(data) { socket.removeListener('join_success', handleJoinSuccess); memberId = data.memberId; genesisUrl = data.genesisUrl; cb(memberId); }); socket.emit('join', { documentId: documentId }); }; this.leaveSession = function (cb) { socket.emit('leave', {}, cb); socket.removeAllListeners(); }; this.getSocket = function () { return socket; }; this.destroy = function () { socket.disconnect(); }; function init() { socket = io({ query: 'token=' + authToken, forceNew: true }); socket.on('connect', connectedCb); socket.on('kick', kickedCb); socket.on('disconnect', disconnectedCb); } init(); }; return ClientAdaptor; }); diff --git a/client/components/wodo/editor.controller.js b/client/components/wodo/editor.controller.js index 08ccede..9b90b1a 100644 --- a/client/components/wodo/editor.controller.js +++ b/client/components/wodo/editor.controller.js @@ -1,91 +1,92 @@ 'use strict'; /*global Wodo*/ angular.module('manticoreApp') .controller('WodoCtrl', function ($scope, Auth, Adaptor) { var editorInstance, clientAdaptor, editorOptions = { collabEditingEnabled: true, allFeaturesEnabled: true }, onConnectCalled = false; function closeEditing() { editorInstance.leaveSession(function () { $scope.$apply(function () { $scope.joined = false; }); clientAdaptor.leaveSession(function () { console.log('Closed editing, left session.'); }); }); } function handleEditingError(error) { alert('Something went wrong!\n' + error); console.log(error); closeEditing(); } function openEditor() { Wodo.createCollabTextEditor('wodoContainer', editorOptions, function (err, editor) { editorInstance = editor; $scope.editor = editor; + editorInstance.clientAdaptor = clientAdaptor; editorInstance.addEventListener(Wodo.EVENT_UNKNOWNERROR, handleEditingError); editorInstance.joinSession(clientAdaptor, function () { $scope.$apply(function () { $scope.joined = true; }); }); }); } function boot() { clientAdaptor = new Adaptor( $scope.document._id, Auth.getToken(), function onConnect() { console.log('onConnect'); if (onConnectCalled) { console.log('Reconnecting not yet supported'); return; } onConnectCalled = true; clientAdaptor.joinSession(function (memberId) { if (!memberId) { console.log('Could not join; memberId not received'); } else { console.log('Joined with memberId ' + memberId); openEditor(); } }); }, function onKick() { console.log('onKick'); closeEditing(); }, function onDisconnect() { console.log('onDisconnect'); } ); } function destroy (cb) { if (editorInstance) { closeEditing(); editorInstance.destroy(cb); } else { if (clientAdaptor) { clientAdaptor.leaveSession(); clientAdaptor.destroy(); cb(); } } } this.boot = boot; this.destroy = destroy; }); diff --git a/client/index.html b/client/index.html index ef0e933..f31aeb3 100644 --- a/client/index.html +++ b/client/index.html @@ -1,92 +1,94 @@
+ + diff --git a/package.json b/package.json index 180d542..77b1204 100644 --- a/package.json +++ b/package.json @@ -1,97 +1,98 @@ { "name": "manticore", "version": "0.0.0", "main": "server/app.js", "dependencies": { "express": "~4.0.0", "morgan": "~1.0.0", "body-parser": "~1.5.0", "method-override": "~1.0.0", "serve-favicon": "~2.0.1", "cookie-parser": "~1.0.1", "express-session": "~1.0.2", "errorhandler": "~1.0.0", "compression": "~1.0.1", "lodash": "~2.4.1", "jade": "~1.2.0", "mongoose": "~3.8.8", "gridfs-stream": "1.1.1", "jsonwebtoken": "^0.3.0", "express-jwt": "^0.1.3", "passport": "~0.2.0", "passport-local": "~0.1.6", "composable-middleware": "^0.3.0", "connect-mongo": "^0.4.1", "multer": "0.1.8", "socket.io": "1.3.5", "socketio-jwt": "4.2.0", "async": "~1.3.0", "phantom": "~0.7.2", - "dav": "~1.7.6" + "dav": "~1.7.6", + "request": "~2.60.0" }, "devDependencies": { "grunt": "~0.4.4", "grunt-autoprefixer": "~0.7.2", "grunt-wiredep": "~1.8.0", "grunt-concurrent": "~0.5.0", "grunt-contrib-clean": "~0.5.0", "grunt-contrib-concat": "~0.4.0", "grunt-contrib-copy": "~0.5.0", "grunt-contrib-cssmin": "~0.9.0", "grunt-contrib-htmlmin": "~0.2.0", "grunt-contrib-imagemin": "~0.7.1", "grunt-contrib-jshint": "~0.10.0", "grunt-contrib-uglify": "~0.4.0", "grunt-contrib-watch": "~0.6.1", "grunt-contrib-jade": "^0.11.0", "grunt-google-cdn": "~0.4.0", "grunt-newer": "~0.7.0", "grunt-ng-annotate": "^0.2.3", "grunt-filerev": "~2.2.0", "grunt-svgmin": "~0.4.0", "grunt-usemin": "~3.0.0", "grunt-env": "~0.4.1", "grunt-node-inspector": "~0.1.5", "grunt-nodemon": "~0.2.0", "grunt-angular-templates": "^0.5.4", "grunt-dom-munger": "^3.4.0", "grunt-protractor-runner": "^1.1.0", "grunt-asset-injector": "^0.1.0", "grunt-karma": "~0.8.2", "grunt-build-control": "DaftMonk/grunt-build-control", "grunt-mocha-test": "~0.10.2", "grunt-contrib-stylus": "latest", "jit-grunt": "^0.5.0", "time-grunt": "~0.3.1", "grunt-express-server": "~0.4.17", "grunt-open": "~0.2.3", "open": "~0.0.4", "jshint-stylish": "~0.1.5", "connect-livereload": "~0.4.0", "karma-ng-scenario": "~0.1.0", "karma-firefox-launcher": "~0.1.3", "karma-script-launcher": "~0.1.0", "karma-html2js-preprocessor": "~0.1.0", "karma-ng-jade2js-preprocessor": "^0.1.2", "karma-jasmine": "~0.1.5", "karma-chrome-launcher": "~0.1.3", "requirejs": "~2.1.11", "karma-requirejs": "~0.2.1", "karma-coffee-preprocessor": "~0.2.1", "karma-jade-preprocessor": "0.0.11", "karma-phantomjs-launcher": "~0.1.4", "karma": "~0.12.9", "karma-ng-html2js-preprocessor": "~0.1.0", "supertest": "~0.11.0", "should": "~3.3.1" }, "engines": { "node": ">=0.10.0" }, "scripts": { "start": "node server/app.js", "test": "grunt test", "update-webdriver": "node node_modules/grunt-protractor-runner/node_modules/protractor/bin/webdriver-manager update" }, "private": true } diff --git a/server/api/document/document.controller.js b/server/api/document/document.controller.js index 817e54f..45b003b 100644 --- a/server/api/document/document.controller.js +++ b/server/api/document/document.controller.js @@ -1,171 +1,89 @@ /** * Using Rails-like standard naming convention for endpoints. * GET /documents -> index * POST /documents -> create * GET /documents/:id -> show * PUT /documents/:id -> update * DELETE /documents/:id -> destroy */ 'use strict'; var _ = require('lodash'); var mongoose = require('mongoose'); var Grid = require('gridfs-stream'); var multer = require('multer'); +var storage = require('./storage'); + var DocumentChunk = require('./document.model').DocumentChunk; var Document = require('./document.model').Document; var Template = require('../template/template.model'); var gfs = Grid(mongoose.connection.db, mongoose.mongo); // Get list of documents -exports.index = function(req, res) { - Document.find().populate('creator').exec(function (err, documents) { - if(err) { return handleError(res, err); } - return res.json(200, documents); - }); -}; +exports.index = storage.index; // Get a single document -exports.show = function(req, res) { - Document.findById(req.params.id, function (err, document) { - if(err) { return handleError(res, err); } - if(!document) { return res.send(404); } - return res.json(document); - }); -}; +exports.show = storage.show; -exports.upload = function (req, res, next) { - multer({ - upload: null, - limits: { - fileSize: 1024 * 1024 * 20, // 20 Megabytes - files: 5 - }, - onFileUploadStart: function (file) { - var chunkId = new mongoose.Types.ObjectId(), - fileId = new mongoose.Types.ObjectId(); - - var firstChunk = new DocumentChunk({ - _id: chunkId, - fileId: fileId - }); - var newDocument = new Document({ - title: file.originalname, - creator: req.user._id, - chunks: [chunkId] - }); - this.upload = gfs.createWriteStream({ - _id: fileId, - filename: file.originalname, - mode: 'w', - chunkSize: 1024 * 4, - content_type: file.mimetype, - root: 'fs' - }); - this.upload.on('finish', function () { - firstChunk.save(function (err) { - if (!err) { - newDocument.save(); - } - }); - }); - }, - onFileUploadData: function (file, data) { - this.upload.write(data); - }, - onFileUploadComplete: function (file) { - this.upload.end(); - } - })(req, res, next); -}; +// Middleware for handling file uploads +exports.upload = storage.upload; exports.acknowledgeUpload = function (req, res) { return res.send(200); }; exports.showSnapshot = function(req, res) { var snapshotId = req.params.id; gfs.findOne({_id: snapshotId}, function (err, file) { if (err) { return handleError(res, err); } if (!file) { return res.send(404); } var download = gfs.createReadStream({ _id: snapshotId }); download.on('error', function (err) { return handleError(res, err); }); res.set('Content-Type', file.contentType); res.attachment(file.filename) download.pipe(res); }); }; -// Creates a new document in the DB. -exports.create = function(req, res) { - Document.create(req.body, function(err, document) { - if(err) { return handleError(res, err); } - return res.json(201, document); - }); -}; +exports.createFromTemplate = storage.createFromTemplate; -exports.createFromTemplate = function (req, res) { - Template.findById(req.params.id, function (err, template) { - if (err) { return handleError(res, err); } - if (!template) { return res.send(404); } - - var chunkId = new mongoose.Types.ObjectId(); - - var firstChunk = new DocumentChunk({ - _id: chunkId, - fileId: template.fileId - }); - var newDocument = new Document({ - title: template.title, - creator: req.user._id, - chunks: [chunkId] - }); - - firstChunk.save(function (err) { - if (!err) { - newDocument.save(function (err) { - return res.json(201, newDocument); - }); - } - }) - }); -}; +exports.createChunkFromSnapshot = storage.createChunkFromSnapshot; // Updates an existing document in the DB. exports.update = function(req, res) { if(req.body._id) { delete req.body._id; } Document.findById(req.params.id, function (err, document) { if (err) { return handleError(res, err); } if(!document) { return res.send(404); } var updated = _.merge(document, req.body); updated.save(function (err) { if (err) { return handleError(res, err); } return res.json(200, document); }); }); }; // Deletes a document from the DB. exports.destroy = function(req, res) { Document.findById(req.params.id, function (err, document) { if(err) { return handleError(res, err); } if(!document) { return res.send(404); } document.remove(function(err) { if(err) { return handleError(res, err); } return res.send(204); }); }); }; function handleError(res, err) { return res.send(500, err); } diff --git a/server/api/document/document.model.js b/server/api/document/document.model.js index 7c21ee8..197ca07 100644 --- a/server/api/document/document.model.js +++ b/server/api/document/document.model.js @@ -1,27 +1,39 @@ 'use strict'; var mongoose = require('mongoose'), Schema = mongoose.Schema; +var storageTypes = ['webdav']; + /* - * Each Document Chunk has an associated ODF snapshot file within - * GridFS of the same ID. + * Each DocumentChunk has an associated ODF snapshot file within + * GridFS, a list of operations required to bring the snapshot into + * a workable initial state for the chunk, and a list of operations + * that signifies the edit history after the aforementioned document + * state */ var DocumentChunk = new Schema({ - operations: { type: Array, default: [] }, - fileId: { type: Schema.Types.ObjectId } + sequence: { type: Number, default: 0 }, + snapshot: { + fileId: { type: Schema.Types.ObjectId, required: true }, + operations: { type: Array, default: [] } + }, + operations: { type: Array, default: [] } }); var DocumentSchema = new Schema({ title: String, created: { type: Date, default: Date.now, required: true }, date: { type: Date, default: Date.now }, creator: { type: Schema.Types.ObjectId, ref: 'User' }, editors: { type: [{type: Schema.Types.ObjectId, ref: 'User'}], default: [] }, - chunks: { type: [{type: Schema.Types.ObjectId, ref: 'DocumentChunk'}], required: true } + chunks: { type: [{type: Schema.Types.ObjectId, ref: 'DocumentChunk'}], default: [] }, + live: { type: Boolean, default: false }, + provider: String, + webdav: {} }); module.exports = { DocumentChunk: mongoose.model('DocumentChunk', DocumentChunk), Document: mongoose.model('Document', DocumentSchema) }; diff --git a/server/api/document/storage/index.js b/server/api/document/storage/index.js new file mode 100644 index 0000000..81b3033 --- /dev/null +++ b/server/api/document/storage/index.js @@ -0,0 +1,4 @@ +'use strict'; + +var config = require('../../../config/environment'); +module.exports = require('./' + config.storage.type); diff --git a/server/api/document/storage/local/index.js b/server/api/document/storage/local/index.js new file mode 100644 index 0000000..3839a7f --- /dev/null +++ b/server/api/document/storage/local/index.js @@ -0,0 +1,147 @@ +'use strict'; + +var multer = require('multer'); +var mongoose = require('mongoose'); +var Grid = require('gridfs-stream'); + +var Document = require('../../document.model').Document; +var DocumentChunk = require('../../document.model').DocumentChunk; +var Template = require('../../../template/template.model'); + +var gfs = Grid(mongoose.connection.db, mongoose.mongo); + +exports.index = function (req, res) { + var userId = req.user._id; + + Document.find({ + '$or': [ + { 'creator': userId }, + { 'editors': { '$in': [userId] } } + ] + }).populate('creator', 'name email').exec(function (err, documents) { + if(err) { return handleError(res, err); } + return res.json(200, documents); + }); +}; + +exports.show = function(req, res) { + Document.findById(req.params.id, function (err, document) { + if(err) { return handleError(res, err); } + if(!document) { return res.send(404); } + return res.json(document); + }); +}; + +exports.upload = function (req, res, next) { + multer({ + upload: null, + limits: { + fileSize: 1024 * 1024 * 20, // 20 Megabytes + files: 5 + }, + onFileUploadStart: function (file) { + var chunkId = new mongoose.Types.ObjectId(), + fileId = new mongoose.Types.ObjectId(); + + var firstChunk = new DocumentChunk({ + _id: chunkId, + snapshot: { + fileId: fileId + } + }); + var newDocument = new Document({ + title: file.originalname, + creator: req.user._id, + chunks: [chunkId] + }); + this.upload = gfs.createWriteStream({ + _id: fileId, + filename: file.originalname, + mode: 'w', + chunkSize: 1024 * 4, + content_type: file.mimetype, + root: 'fs' + }); + this.upload.on('finish', function () { + firstChunk.save(function (err) { + if (!err) { + newDocument.save(); + } + }); + }); + }, + onFileUploadData: function (file, data) { + this.upload.write(data); + }, + onFileUploadComplete: function (file) { + this.upload.end(); + } + })(req, res, next); +}; + + +exports.createFromTemplate = function (req, res) { + Template.findById(req.params.id, function (err, template) { + if (err) { return handleError(res, err); } + if (!template) { return res.send(404); } + + var chunkId = new mongoose.Types.ObjectId(); + + var firstChunk = new DocumentChunk({ + _id: chunkId, + snapshot: { + fileId: template.fileId + } + }); + var newDocument = new Document({ + title: template.title, + creator: req.user._id, + chunks: [chunkId] + }); + + firstChunk.save(function (err) { + if (!err) { + newDocument.save(function (err) { + return res.json(201, newDocument); + }); + } + }) + }); +}; + +exports.createChunkFromSnapshot = function (document, snapshot, cb) { + var chunkId = new mongoose.Types.ObjectId(), + fileId = new mongoose.Types.ObjectId(); + + var writeStream = gfs.createWriteStream({ + _id: fileId, + filename: document.title + '_' + Date.now(), + mode: 'w', + chunkSize: 1024 * 4, + content_type: 'application/vnd.oasis.opendocument.text', + root: 'fs' + }); + + writeStream.write(new Buffer(snapshot.data), function () { + writeStream.end(function () { + var chunk = new DocumentChunk({ + _id: chunkId, + sequence: snapshot.sequence, + snapshot: { + fileId: fileId, + operations: snapshot.operations + } + }); + chunk.save(function (err) { + if (err) { return cb(err); } + document.chunks.push(chunkId); + document.markModified('chunks'); + cb(null, chunk); + }); + }); + }); +}; + +function handleError(res, err) { + return res.send(500, err); +} diff --git a/server/api/document/storage/webdav/index.js b/server/api/document/storage/webdav/index.js new file mode 100644 index 0000000..8952686 --- /dev/null +++ b/server/api/document/storage/webdav/index.js @@ -0,0 +1,450 @@ +'use strict'; + +var _ = require('lodash'); +var dav = require('dav'); +var async = require('async'); +var mongoose = require('mongoose'); +var https = require('https'); +var fs = require('fs'); +var url = require('url'); +var path = require('path'); +var Grid = require('gridfs-stream'); +var multer = require('multer'); +var request = require('request'); +var querystring = require('querystring'); + +var config = require('../../../../config/environment'); + +var User = require('../../../user/user.model'); +var Document = require('../../document.model').Document; +var DocumentChunk = require('../../document.model').DocumentChunk; +var Template = require('../../../template/template.model'); + +var gfs = Grid(mongoose.connection.db, mongoose.mongo); + +function makeDavClient (user) { + return new dav.Client( + new dav.transport.Basic(new dav.Credentials({ + username: user.webdav.username, + password: user.webdav.password + })), + { + baseUrl: config.storage.server + } + ); +} + +function saveToGridFS(user, href, fileId, cb) { + var file = gfs.createWriteStream({ + _id: fileId, + filename: href.split('/').pop(), + mode: 'w', + chunkSize: 1024 * 4, + content_type: 'application/vnd.oasis.opendocument.text', + root: 'fs' + }); + + request.get({ + url: config.storage.server + href, + auth: { + user: user.webdav.username, + pass: user.webdav.password, + } + }) + .on('error', function (err) { + cb(err); + }) + .pipe(file); + + file.on('finish', cb); +} + +function makeContentId(webdavDoc) { + return webdavDoc.props.getetag + webdavDoc.props.getlastmodified; +} + +// The WebDAV server is always authoritative. +function synchronizeUserFilesToDB(user, webdavDocuments, objectCache, cb) { + Document.find({ + '$or': [ + { 'creator': user._id }, + { 'editors': { '$in': [user._id] } } + ] + }) + .populate('creator', 'name email').exec(function (err, dbDocuments) { + // Since we're doing things that could possibly invalidate DB documents, + // we have to work with the live cache + var persistenceQueue = []; + var finalDocuments = []; + + dbDocuments.forEach(function (dbDoc) { + var trackedDoc; + if (dbDoc.creator._id.equals(user._id)) { + trackedDoc = objectCache.getTrackedObject(dbDoc); + finalDocuments.push(trackedDoc); + } else { + finalDocuments.push(dbDoc); + return; + } + + var sameHrefDoc = _.find(webdavDocuments, function (doc) { + return doc.href === trackedDoc.webdav.href; + }); + var sameContentDoc = _.find(webdavDocuments, function (doc) { + return makeContentId(doc) === trackedDoc.webdav.contentId; + }); + + if (!sameContentDoc) { + if (!sameHrefDoc) { + // Document has been effectively deleted, take it offline and remove + trackedDoc.live = false; + persistenceQueue.push(function (cb) { trackedDoc.remove(cb); }); + finalDocuments.pop(); + } else { + // Document has been modified, take it offline, throw away local changes, and update content ID + trackedDoc.live = false; + trackedDoc.chunks.length = 0; + trackedDoc.markModified('chunks'); + trackedDoc.webdav.contentId = makeContentId(sameHrefDoc); + trackedDoc.date = Date.parse(sameHrefDoc.props.getlastmodified); + trackedDoc.markModified('webdav'); + persistenceQueue.push(function (cb) { trackedDoc.save(cb); }); + } + } else { + if (!sameHrefDoc) { + // Document has been moved, just update href and save without interrupting anything + trackedDoc.webdav.href = sameContentDoc.href; + trackedDoc.markModified('webdav'); + persistenceQueue.push(function (cb) { trackedDoc.save(cb); }); + } else { + if (makeContentId(sameHrefDoc) === makeContentId(sameContentDoc)) { + // nothing changed + } else { + // Document moved to sameContentDoc.props.href and there is a new doc at sameHrefDoc.href + trackedDoc.webdav.href = sameContentDoc.href; + trackedDoc.markModified('webdav'); + persistenceQueue.push(function (cb) { trackedDoc.save(cb); }); + } + } + } + }); + + // Once modified documents are persisted, purge non-live documents from the cache + async.parallel(persistenceQueue, function (err) { + if (err) return cb(err); + + dbDocuments.forEach(function (dbDoc) { + if (objectCache.isTracked(dbDoc) && !dbDoc.live) { + objectCache.forgetTrackedObject(dbDoc); + } + }); + + return cb(null, finalDocuments); + }); + }); +} + +exports.index = function (req, res) { + var davClient = makeDavClient(req.user); + + // Retrieve a list of files and filter by mimetype + davClient.send(dav.request.propfind({ + depth: 'infinity', + props: [ + { name: 'getcontenttype', namespace: dav.ns.DAV }, + { name: 'getetag', namespace: dav.ns.DAV }, + { name: 'getlastmodified', namespace: dav.ns.DAV } + ] + }), config.storage.path) + .then( + function success(response) { + var webdavDocuments = _(response) + .filter(function (item) { + return item.props.getcontenttype === 'application/vnd.oasis.opendocument.text'; + }) + .map(function (item) { + item.href = querystring.unescape(item.href); + return item; + }) + .value(); + + synchronizeUserFilesToDB(req.user, webdavDocuments, req.app.get('objectCache'), function (err, updatedDocuments) { + if (err) return handleError(res, err); + // Transform the webdavDocuments array into something more usable + // The fake ID of this document is represented as a combination of it's href and content id + webdavDocuments = webdavDocuments.map(function (doc) { + return { + _id: new Buffer(doc.href + '__' + makeContentId(doc)).toString('base64'), + title: doc.href.split('/').pop(), + creator: req.user.profile, + date: Date.parse(doc.props.getlastmodified), + webdav: { + href: doc.href, + contentId: makeContentId(doc) + } + }; + }); + // Merge the two document types, such that hrefs are unique in the resulting list. DB docs have priority here + var mergedDocuments = _.uniq(updatedDocuments.concat(webdavDocuments), function (doc) { + return (doc.webdav && doc.webdav.contentId) || makeContentId(doc); + }); + return res.json(200, mergedDocuments); + }); + }, + function failure(error) { + handleError(res, error); + }); +}; + +function createFirstChunk(user, href, cb) { + var chunkId = new mongoose.Types.ObjectId(), + fileId = new mongoose.Types.ObjectId(); + + var firstChunk = new DocumentChunk({ + _id: chunkId, + snapshot: { + fileId: fileId + } + }); + + saveToGridFS(user, href, fileId, function(err) { + if (err) { return cb(err); } + firstChunk.save(function (err) { + cb(err, firstChunk); + }); + }); +} + +exports.show = function(req, res) { + if (mongoose.Types.ObjectId.isValid(req.params.id)) { + Document.findById(req.params.id, function (err, document) { + if(err) { return handleError(res, err); } + if(!document) { return res.send(404); } + if (document.chunks.length) { return res.json(200, document); } + // If the document had been invalidated, it has no chunks, so generate one + createFirstChunk(req.user, document.webdav.href, function (err, firstChunk) { + if (err) { return handleError(res, err); } + document.chunks.push(firstChunk); + document.save(function () { + res.json(200, document); + }) + }); + }); + } else { + var identifier = new Buffer(req.params.id, 'base64').toString('ascii').split('__'), + href = identifier[0], + contentId = identifier[1]; + Document.findOne({ + 'webdav.contentId': contentId, + }, function (err, document) { + if (err) { return handleError(res, err); } + if (document) { + return res.json(200, document); + } else { + var davClient = makeDavClient(req.user); + davClient.send(dav.request.propfind({ props: [] }), href) + .then( + function success(response) { + var webdavDoc = response[0]; + createFirstChunk(req.user, href, function(err, firstChunk) { + if (err) { return handleError(res, err); } + Document.create({ + title: href.split('/').pop(), + creator: req.user._id, + created: Date.parse(webdavDoc.props.getlastmodified), + date: Date.parse(webdavDoc.props.getlastmodified), + chunks: [firstChunk._id], + webdav: { + href: href, + contentId: makeContentId(webdavDoc) + } + }, function (err, document) { + if (err) { return handleError(res, err); } + res.json(201, document); + }); + }) + }, + function failure(error) { + res.send(404, error); + }); + } + }); + } +}; + +function uploadToServer(user, readStream, href, replace, cb) { + var nonConflictingPath; + + function upload () { + readStream.pipe(request.put({ + url: nonConflictingPath, + auth: { + user: user.webdav.username, + pass: user.webdav.password, + }, + headers: { + 'Content-Type': 'application/vnd.oasis.opendocument.text' + } + }, cb)); + } + + if (replace) { + nonConflictingPath = config.storage.server + href; + upload(); + } else { + makeDavClient(user).send(dav.request.propfind({ + depth: 1, + props: [ + { name: 'getcontenttype', namespace: dav.ns.DAV }, + { name: 'getetag', namespace: dav.ns.DAV }, + { name: 'getlastmodified', namespace: dav.ns.DAV } + ] + }), path.dirname(href)) + .then( + function success(response) { + function makePath(dir, basename, ext) { + return dir + '/' + basename + ext; + } + + var files = _(response) + .filter(function (item) { + return item.props.getcontenttype === 'application/vnd.oasis.opendocument.text'; + }).map(function (item) { + item.href = querystring.unescape(item.href); + return item; + }).value(); + + var iteration = 0, + extension = path.extname(href), + basename = path.basename(href, extension), + basename_i = basename, + dirname = path.dirname(href); + + for (var i = 0; i < files.length; i++) { + if (files[i].href === makePath(dirname, basename_i, extension)) { + iteration++; + basename_i = basename + ' (' + iteration + ')'; + i = -1; + } + } + + nonConflictingPath = config.storage.server + makePath(dirname, basename_i, extension); + upload(); + }, + function failure(err) { + cb(err); + }); + } +} + +exports.upload = function (req, res, next) { + multer({ + upload: null, + limits: { + fileSize: 1024 * 1024 * 20, // 20 Megabytes + files: 5 + }, + onFileUploadComplete: function (file) { + uploadToServer(req.user, + fs.createReadStream(file.path), + config.storage.path + '/' + file.originalname, + false, + function (err) { + if (err) { console.log (err); } + } + ); + } + })(req, res, next); +}; + +exports.createFromTemplate = function (req, res) { + Template.findById(req.params.id, function (err, template) { + if (err) { return handleError(res, err); } + if (!template) { return res.send(404); } + + var chunkId = new mongoose.Types.ObjectId(); + + var firstChunk = new DocumentChunk({ + _id: chunkId, + snapshot: { + fileId: template.fileId + } + }); + var newDocument = new Document({ + title: template.title, + creator: req.user._id, + chunks: [chunkId] + }); + + uploadToServer( + req.user, + gfs.createReadStream({ _id: template.fileId }), + config.storage.path + '/' + template.title + '.odt', + false, + function (err, response) { + if (err) { return handleError(res.err); } + newDocument.webdav = { + href: response.request.uri.path, + contentId: response.headers.etag + response.headers.date + }; + + firstChunk.save(function (err) { + if (!err) { + newDocument.save(function (err) { + return res.json(201, newDocument); + }); + } + }); + } + ); + + }); +}; + +exports.createChunkFromSnapshot = function (document, snapshot, cb) { + var chunkId = new mongoose.Types.ObjectId(), + fileId = new mongoose.Types.ObjectId(); + + var writeStream = gfs.createWriteStream({ + _id: fileId, + filename: document.title + '_' + Date.now(), + mode: 'w', + chunkSize: 1024 * 4, + content_type: 'application/vnd.oasis.opendocument.text', + root: 'fs' + }); + + writeStream.end(new Buffer(snapshot.data), function () { + User.findById(document.creator._id, function (err, user) { + if (err) { return cb(err); } + uploadToServer( + user, + gfs.createReadStream({ _id: fileId }), + document.webdav.href, + true, + function (err, response) { + if (err) { return cb(err); } + var chunk = new DocumentChunk({ + _id: chunkId, + sequence: snapshot.sequence, + snapshot: { + fileId: fileId, + operations: snapshot.operations + } + }); + chunk.save(function (err) { + if (err) { return cb(err); } + document.webdav.contentId = response.headers.etag + response.headers.date; + document.chunks.push(chunkId); + document.markModified('chunks'); + cb(null, chunk); + }); + }); + }); + }); +}; + +function handleError(res, err) { + console.log(err); + return res.send(500, err); +} diff --git a/server/api/user/user.model.js b/server/api/user/user.model.js index 2a96013..48a06a0 100644 --- a/server/api/user/user.model.js +++ b/server/api/user/user.model.js @@ -1,145 +1,151 @@ 'use strict'; var mongoose = require('mongoose'); var Schema = mongoose.Schema; var crypto = require('crypto'); var authTypes = ['webdav']; +var WebDAVSchema = new Schema({ + username: String, + password: String +}); + var UserSchema = new Schema({ name: String, email: { type: String, lowercase: true }, role: { type: String, default: 'user' }, hashedPassword: String, provider: String, + webdav: {}, salt: String }); /** * Virtuals */ UserSchema .virtual('password') .set(function(password) { this._password = password; this.salt = this.makeSalt(); this.hashedPassword = this.encryptPassword(password); }) .get(function() { return this._password; }); // Public profile information UserSchema .virtual('profile') .get(function() { return { 'name': this.name, 'role': this.role }; }); // Non-sensitive info we'll be putting in the token UserSchema .virtual('token') .get(function() { return { '_id': this._id, 'role': this.role }; }); /** * Validations */ // Validate empty email UserSchema .path('email') .validate(function(email) { return email.length; }, 'Email cannot be blank'); // Validate empty password UserSchema .path('hashedPassword') .validate(function(hashedPassword) { if (authTypes.indexOf(this.provider) !== -1) return true; return hashedPassword.length; }, 'Password cannot be blank'); // Validate email is not taken UserSchema .path('email') .validate(function(value, respond) { var self = this; this.constructor.findOne({email: value}, function(err, user) { if(err) throw err; if(user) { if(self.id === user.id) return respond(true); return respond(false); } respond(true); }); }, 'The specified email address is already in use.'); var validatePresenceOf = function(value) { return value && value.length; }; /** * Pre-save hook */ UserSchema .pre('save', function(next) { if (!this.isNew) return next(); if (!validatePresenceOf(this.hashedPassword) && authTypes.indexOf(this.provider) === -1) next(new Error('Invalid password')); else next(); }); /** * Methods */ UserSchema.methods = { /** * Authenticate - check if the passwords are the same * * @param {String} plainText * @return {Boolean} * @api public */ authenticate: function(plainText) { return this.encryptPassword(plainText) === this.hashedPassword; }, /** * Make salt * * @return {String} * @api public */ makeSalt: function() { return crypto.randomBytes(16).toString('base64'); }, /** * Encrypt password * * @param {String} password * @return {String} * @api public */ encryptPassword: function(password) { if (!password || !this.salt) return ''; var salt = new Buffer(this.salt, 'base64'); return crypto.pbkdf2Sync(password, salt, 10000, 64).toString('base64'); } }; module.exports = mongoose.model('User', UserSchema); diff --git a/server/app.js b/server/app.js index 3900244..62b2223 100644 --- a/server/app.js +++ b/server/app.js @@ -1,74 +1,74 @@ /** * Main application file */ 'use strict'; // Set default node environment to development process.env.NODE_ENV = process.env.NODE_ENV || 'development'; var express = require('express'); var mongoose = require('mongoose'); var config = require('./config/environment'); -var Adaptor = require('./components/adaptor'); -var ObjectCache = require('./components/objectcache'); // Connect to database mongoose.connect(config.mongo.uri, config.mongo.options); // Populate DB with sample data if(config.seedDB) { require('./config/seed'); } // Setup server +var Adaptor = require('./components/adaptor'); +var ObjectCache = require('./components/objectcache'); var app = express(); +app.set('objectCache', new ObjectCache()); + var server = require('http').createServer(app); var socketio = require('socket.io')(server, { path: '/socket.io' }); var adaptor; -var objectCache; var sockets = []; require('./config/socketio')(socketio); require('./config/express')(app); require('./routes')(app); server.on('connection', function (socket) { sockets.push(socket); socket.on('close', function () { sockets.splice(sockets.indexOf(socket), 1); }); }); // Start server server.listen(config.port, config.ip, function () { console.log('Express server listening on %d, in %s mode', config.port, app.get('env')); - objectCache = new ObjectCache(); - adaptor = new Adaptor(app, socketio, objectCache); + adaptor = new Adaptor(app, socketio, app.get('objectCache')); }); function destroy() { adaptor.destroy(function () { console.log('All realtime clients disconnected.'); - objectCache.destroy(function () { + app.get('objectCache').destroy(function () { console.log('All objects persisted to DB.'); sockets.forEach(function (socket) { socket.destroy(); }); server.close(function () { console.log('HTTP server shut down.'); mongoose.disconnect(function () { console.log('DB connection closed.'); console.log('Everything successfully shut down. Bye!'); process.exit(); }); }); }); }); } process.on("SIGTERM", destroy); process.on("SIGINT", destroy); // Expose app exports = module.exports = app; diff --git a/server/auth/webdav/passport.js b/server/auth/webdav/passport.js index 8a0ce50..77e9482 100644 --- a/server/auth/webdav/passport.js +++ b/server/auth/webdav/passport.js @@ -1,54 +1,58 @@ var passport = require('passport'); var LocalStrategy = require('passport-local').Strategy; var dav = require('dav'); -dav.debug.enabled = true; + exports.setup = function (User, config) { passport.use(new LocalStrategy({ usernameField: 'email', passwordField: 'password' // this is the virtual field on the model }, function(email, password, done) { var client = new dav.Client( new dav.transport.Basic(new dav.Credentials({ username: email, password: password })), { baseUrl: config.storage.server } ); client.send( dav.request.basic({ method: 'OPTIONS', data: ''}), config.storage.path ).then( function success() { User.findOne({ email: email.toLowerCase() }, function (err, user) { if (err) { return done(err); } if (!user) { var newUser = new User({ name: email, email: email, provider: 'webdav', - role: 'user' + role: 'user', + webdav: { + username: email, + password: password + } }); newUser.save(function (err, user) { if (err) { return done(err); } if (!err) { return done(null, user); } }); } else { return done(null, user); } }); }, function failure(err) { return done(err); } ); } )); }; diff --git a/server/components/adaptor/index.js b/server/components/adaptor/index.js index ed38ffd..dfb4f43 100644 --- a/server/components/adaptor/index.js +++ b/server/components/adaptor/index.js @@ -1,84 +1,100 @@ /* * Copyright (C) 2015 KO GmbH * * @licstart * This file is part of Kotype. * * Kotype is free software: you can redistribute it and/or modify it * under the terms of the GNU Affero General Public License (GNU AGPL) * as published by the Free Software Foundation, either version 3 of * the License, or (at your option) any later version. * * Kotype is distributed in the hope that it will be useful, but * WITHOUT ANY WARRANTY; without even the implied warranty of * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the * GNU Affero General Public License for more details. * * You should have received a copy of the GNU Affero General Public License * along with Kotype. If not, see . * @licend * * @source: https://github.com/kogmbh/Kotype/ */ "use strict"; /*jslint nomen: true, unparam: true */ /*global require, console, setInterval, module */ var async = require("async"), Room = require("./room"), Document = require("../../api/document/document.model").Document, User = require("../../api/user/user.model"); // Maintains an in-memory cache of users, documents, and sessions. // And writes/reads them from the DB on demand. var ServerAdaptor = function (app, socketServer, objectCache) { var rooms = {}; function addToRoom(documentId, socket) { var room = rooms[documentId]; if (!room) { - Document.findById(documentId, function (err, doc) { + Document.findById(documentId).populate('creator', 'name email') + .exec(function (err, doc) { if (err) { return console.log(err); } if (!doc) { return console.log("documentId unknown:"+documentId); } - room = new Room(app, doc, objectCache, function () { + var document = objectCache.getTrackedObject(doc); + document.live = true; + rooms[documentId] = room = new Room(app, document, objectCache, function () { rooms[documentId] = room; room.attachSocket(socket); }); + + var interval = setInterval(function () { + if (!document.live) { + clearInterval(interval); + objectCache.forgetTrackedObject(document); + room.destroy(function () { + delete rooms[documentId]; + }); + } + }, 500); }); - } else { + } else if (room.isAvailable()) { room.attachSocket(socket); + } else { + console.log("Room currently unavailable, disconnecting client.") + socket.disconnect(); } } this.destroy = function (callback) { async.each(Object.keys(rooms), function (documentId, cb) { rooms[documentId].destroy(cb); }, function () { rooms = {}; callback() }); }; function init() { socketServer.on("connection", function (socket) { User.findById(socket.decoded_token._id, function (err, user) { if (err) { return console.log(err); } socket.user = user; socket.on("join", function (data) { var documentId = data.documentId; if (documentId) { console.log("Authorized user " + user.name + " for document " + documentId); addToRoom(documentId, socket); } else { console.log("Error: Client did not specify a document ID"); } }); }); }); } init(); }; module.exports = ServerAdaptor; diff --git a/server/components/adaptor/recorder.js b/server/components/adaptor/recorder.js index b174b48..f454981 100644 --- a/server/components/adaptor/recorder.js +++ b/server/components/adaptor/recorder.js @@ -1,136 +1,144 @@ var phantom = require('phantom'); var fs = require('fs'); var EventEmitter = require('events').EventEmitter; var config = require('../../config/environment'); var Recorder = function (lastChunk, cb) { var self = this, - baseSnapshotUrl = 'http://' + (config.ip || 'localhost') + ':' + config.port + '/api/documents/snapshot/' + lastChunk.fileId, + baseSnapshotUrl = 'http://' + (config.ip || 'localhost') + ':' + config.port + '/api/documents/snapshot/' + lastChunk.snapshot.fileId, emitter = new EventEmitter(), snapshotReadyCb = function () {}, ph, page; function phantomCallback(data) { function documentLoaded(data) { cb(); } switch(data.event) { case 'log': console.log('PhantomJS Log :', data.message); break; case 'documentLoaded': documentLoaded(data); break; default: emitter.emit(data.event, data); } } /* jshint ignore:start */ - function loadDocument(url, operations) { - console.log = function (message) { + function loadDocument(url, sequence, operations) { + window.console.log = function (message) { window.callPhantom({event: 'log', message: message}); }; var odfElement = document.getElementById('odf'); document.odfCanvas = new odf.OdfCanvas(odfElement); document.odfCanvas.addListener('statereadychange', function () { + // The "sequence" is the number of ops executed so far in the canonical document history + window.sequence = sequence; document.session = new ops.Session(document.odfCanvas); document.odtDocument = document.session.getOdtDocument(); var operationRouter = new ops.OperationRouter(); operationRouter.push = function (opspecs) { var op, i; if (!opspecs.length) { return; } for (i = 0; i < opspecs.length; i++) { op = document.session.getOperationFactory().create(opspecs[i]); if (op && op.execute(document.session.getOdtDocument())) { + window.sequence++; // console.log('Just executed op ' + opspecs[i].optype + 'by ' + opspecs[i].memberid); } else { break; } } }; document.session.setOperationRouter(operationRouter); document.session.enqueue(operations); window.callPhantom({event: 'documentLoaded'}); }); document.odfCanvas.load(url); } /* jshint ignore:end */ this.push = function (operations, cb) { page.evaluate(function (opspecs) { document.session.enqueue(opspecs); }, function () { cb(); }, operations); }; this.getSnapshot = function (cb) { var eventId = Date.now(); emitter.once('snapshotReady' + eventId, function (data) { cb(data); }); page.evaluate(function (eventId) { var doc = document.odtDocument, ops = []; doc.getMemberIds().forEach(function (memberId) { ops.push({ optype: 'AddMember', timestamp: Date.now(), memberid: memberId, setProperties: doc.getMember(memberId).getProperties() }); var cursor = doc.getCursor(memberId); if (cursor) { var selection = doc.getCursorSelection(memberId); ops.push({ optype: 'AddCursor', timestamp: Date.now(), memberid: memberId }); ops.push({ optype: 'MoveCursor', timestamp: Date.now(), memberid: memberId, position: selection.position, length: selection.length, selectionType: cursor.getSelectionType() }); } }); document.odfCanvas.odfContainer().createByteArray(function (data) { window.callPhantom({ event: 'snapshotReady' + eventId, operations: ops, - data: data + data: data, + sequence: window.sequence }); }); }, function () {}, eventId); }; this.destroy = function (cb) { ph.exit(); cb(); }; function init() { - phantom.create('--web-security=no', function (ph) { + phantom.create('--web-security=no', function (instance) { + ph = instance; ph.createPage(function (p) { p.open('file://' + __dirname + '/odf.html', function (status) { if (status === 'success') { page = p; page.set('onCallback', phantomCallback); - page.evaluate(loadDocument, function (){}, baseSnapshotUrl, lastChunk.operations); + page.evaluate(loadDocument, function (){}, + baseSnapshotUrl, + lastChunk.sequence - lastChunk.snapshot.operations.length, + lastChunk.snapshot.operations.concat(lastChunk.operations)); } else { return cb(new Error('Could not initialize recorder module.')); } }); }); }); } init(); }; module.exports = Recorder; diff --git a/server/components/adaptor/room.js b/server/components/adaptor/room.js index 719339a..d565b08 100644 --- a/server/components/adaptor/room.js +++ b/server/components/adaptor/room.js @@ -1,401 +1,478 @@ /* * Copyright (C) 2015 KO GmbH * * @licstart * This file is part of Kotype. * * Kotype is free software: you can redistribute it and/or modify it * under the terms of the GNU Affero General Public License (GNU AGPL) * as published by the Free Software Foundation, either version 3 of * the License, or (at your option) any later version. * * Kotype is distributed in the hope that it will be useful, but * WITHOUT ANY WARRANTY; without even the implied warranty of * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the * GNU Affero General Public License for more details. * * You should have received a copy of the GNU Affero General Public License * along with Kotype. If not, see . * @licend * * @source: https://github.com/kogmbh/Kotype/ */ var async = require("async"); var _ = require("lodash"); var RColor = require('../colors'); var DocumentChunk = require("../../api/document/document.model").DocumentChunk; +var DocumentController = require("../../api/document/document.controller"); + var Recorder = require('./recorder'); -var Room = function (app, doc, objectCache, cb) { - var document, - chunk, +var Room = function (app, document, objectCache, cb) { + var ChunkManager = function (seedChunk) { + var serverSeq, + chunks = []; + + function getOperationsAfter(seq) { + var ops = []; + + for (var i = chunks.length - 1; i >= 0; i--) { + if (chunks[i].sequence >= seq) { + ops = chunks[i].operations.concat(ops); + } else { + var basedOn = seq - chunks[i].sequence; + ops = chunks[i].operations.slice(basedOn).concat(ops); + break; + } + } + + return ops; + } + this.getOperationsAfter = getOperationsAfter; + + function appendOperations(ops) { + var lastChunk = getLastChunk(); + lastChunk.operations = lastChunk.operations.concat(ops); + serverSeq += ops.length; + } + this.appendOperations = appendOperations; + + function appendChunk(chunk) { + var trackedChunk = objectCache.getTrackedObject(chunk); + chunks.push(trackedChunk); + serverSeq = trackedChunk.sequence + trackedChunk.operations.length; + } + this.appendChunk = appendChunk; + + function getLastChunk() { + return _.last(chunks); + } + this.getLastChunk = getLastChunk; + + this.getServerSequence = function () { + return serverSeq; + }; + + appendChunk(seedChunk); + }; + + var chunkManager, recorder, hasCursor = {}, sockets = [], userColorMap = {}, randomColor = new RColor(), - serverSeq = 0; + saveInProgress = false, + isAvailable = false; function trackTitle(ops) { var newTitle, i; for (i = 0; i < ops.length; i += 1) { if (ops[i].optype === "UpdateMetadata" && ops[i].setProperties["dc:title"] !== undefined) { newTitle = ops[i].setProperties["dc:title"]; } } if (newTitle !== undefined) { if (newTitle.length === 0) { newTitle = "Untitled Document"; } } if (newTitle) { document.title = newTitle; } } function trackEditors() { // TODO: rather track by ops, to decouple from socket implementation sockets.forEach(function (socket) { var _id = socket.user._id; if (document.editors.indexOf(_id) === -1) { document.editors.push(_id); } }); } function trackCursors(ops) { var i; for (i = 0; i < ops.length; i += 1) { if (ops[i].optype === "AddCursor") { hasCursor[ops[i].memberid] = true; } if (ops[i].optype === "RemoveCursor") { hasCursor[ops[i].memberid] = false; } } } + // Removes all cursors and members in the correct order within the last chunk function sanitizeDocument() { - var ops = chunk.operations, + var chunk = chunkManager.getLastChunk(), + ops = chunk.snapshot.operations.concat(chunk.operations), unbalancedCursors = {}, unbalancedMembers = {}, lastAccessDate = document.date, newOps = [], i; for (i = 0; i < ops.length; i += 1) { if (ops[i].optype === "AddCursor") { unbalancedCursors[ops[i].memberid] = true; } else if (ops[i].optype === "RemoveCursor") { unbalancedCursors[ops[i].memberid] = false; } else if (ops[i].optype === "AddMember") { unbalancedMembers[ops[i].memberid] = true; } else if (ops[i].optype === "RemoveMember") { unbalancedMembers[ops[i].memberid] = false; } } Object.keys(unbalancedCursors).forEach(function (memberId) { if (unbalancedCursors[memberId]) { newOps.push({ optype: "RemoveCursor", memberid: memberId, timestamp: lastAccessDate }); } }); Object.keys(unbalancedMembers).forEach(function (memberId) { if (unbalancedMembers[memberId]) { newOps.push({ optype: "RemoveMember", memberid: memberId, timestamp: lastAccessDate }); } }); if (newOps.length) { // Update op stack - chunk.operations = chunk.operations.concat(newOps); - serverSeq = chunk.operations.length; + chunkManager.appendOperations(newOps); } } function broadcastMessage(message, data) { sockets.forEach(function (peerSocket) { peerSocket.emit(message, data) }); } function sendOpsToMember(socket, ops) { socket.emit("new_ops", { - head: serverSeq, + head: chunkManager.getServerSequence(), ops: ops }); } - function setupMemberSnapshot(socket, operations, genesisSeq) { + function setupMemberSnapshot(socket, snapshot) { socket.emit("replay", { - head: serverSeq, - ops: operations.concat(getOpsAfter(genesisSeq)) + head: chunkManager.getServerSequence(), + ops: snapshot.operations.concat(chunkManager.getOperationsAfter(snapshot.sequence)) }); } function broadcastOpsByMember(socket, ops) { if (!ops.length) { return; } sockets.forEach(function (peerSocket) { if (peerSocket.memberId !== socket.memberId) { sendOpsToMember(peerSocket, ops); } }); } function writeOpsToDocument(ops, cb) { - if (!ops.length) { + if (!ops.length || !document.live) { cb(); } recorder.push(ops, function () { trackTitle(ops); trackEditors(); // Update op stack - chunk.operations = chunk.operations.concat(ops); - serverSeq = chunk.operations.length; + chunkManager.appendOperations(ops); // Update modified date document.date = new Date(); cb(); }); } function addMember(user, cb) { var memberId, op, timestamp = Date.now(), color = userColorMap[user._id]; memberId = user.name + "_" + timestamp.toString(); // Let user colors persist in a Room even after they've // left and joined. if (!color) { userColorMap[user._id] = color = randomColor.get(true, 0.7); } op = { optype: "AddMember", memberid: memberId, timestamp: timestamp, setProperties: { fullName: user.name, color: color } }; writeOpsToDocument([op], function () { cb(memberId, [op]); }); } function removeMember(memberId, cb) { var ops = [], timestamp = Date.now(); if (hasCursor[memberId]) { ops.push({ optype: "RemoveCursor", memberid: memberId, timestamp: timestamp }); } ops.push({ optype: "RemoveMember", memberid: memberId, timestamp: timestamp }); writeOpsToDocument(ops, function () { cb(ops); }); } - function getOpsAfter(basedOn) { - return chunk.operations.slice(basedOn, serverSeq); - } - this.socketCount = function () { return sockets.length; }; this.attachSocket = function (socket) { // Add the socket to the room and give the // client it's unique memberId addMember(socket.user, function (memberId, ops) { socket.memberId = memberId; sockets.push(socket); broadcastOpsByMember(socket, ops); // Generate genesis URL with the latest document version's snapshot // We use a two-time URL because WebODF makes two identical GET requests (!) var genesisUrl = '/genesis/' + Date.now().toString(), - genesisSeq = serverSeq, usages = 0; recorder.getSnapshot(function (snapshot) { + var buffer = new Buffer(snapshot.data); app.get(genesisUrl, function (req, res) { usages++; res.set('Content-Type', 'application/vnd.oasis.opendocument.text'); res.attachment(document.title); - res.send(new Buffer(snapshot.data)); + res.send(buffer); var routes = app._router.stack; if (usages === 2) { + buffer = null; for (var i = 0; i < routes.length; i++) { if (routes[i].path === genesisUrl) { routes.splice(i, 1); break; } } } }); socket.emit("join_success", { memberId: memberId, genesisUrl: genesisUrl }); // Service replay requests socket.on("replay", function () { - setupMemberSnapshot(socket, snapshot.operations, genesisSeq); + setupMemberSnapshot(socket, snapshot); }); // Store, analyze, and broadcast incoming commits socket.on("commit_ops", function (data, cb) { var clientSeq = data.head, ops = data.ops; - if (clientSeq === serverSeq) { + if (clientSeq === chunkManager.getServerSequence()) { writeOpsToDocument(ops, function () { cb({ conflict: false, - head: serverSeq + head: chunkManager.getServerSequence() }); trackCursors(ops); broadcastOpsByMember(socket, data.ops); }); } else { cb({ conflict: true }); } }); + // Service save requests. A save is a commit + + socket.on("save", function (cb) { + // Saves are blocking inside the phantomjs process, and they affect everyone, + // therefore use a lock. + if (saveInProgress) { + var checkIfSaved = setInterval(function () { + if (!saveInProgress) { + clearInterval(checkIfSaved); + cb(); + } + }, 1000); + } else { + saveInProgress = true; + recorder.getSnapshot(function (snapshot) { + DocumentController.createChunkFromSnapshot(document, snapshot, + function (err, chunk) { + saveInProgress = false; + if (err) { return cb(err); } + chunkManager.appendChunk(chunk); + cb(); + }); + }); + } + }); + // Service various requests socket.on("access_get", function (data, cb) { cb({ access: document.isPublic ? "public" : "normal" }); }); if (socket.user.identity !== "guest") { socket.on("access_change", function (data) { document.isPublic = data.access === "public"; broadcastMessage("access_changed", { access: data.access === "public" ? "public" : "normal" }); if (data.access !== "public") { sockets.forEach(function (peerSocket) { if (peerSocket.user.identity === "guest") { console.log(peerSocket.user.name); removeSocket(peerSocket); } }); } }); } }); // Handle dropout events socket.on("leave", function () { removeSocket(socket); }); socket.on("disconnect", function () { removeSocket(socket); }); }); }; function detachSocket(socket, callback) { removeMember(socket.memberId, function (ops) { broadcastOpsByMember(socket, ops); socket.removeAllListeners(); function lastCB() { socket.removeAllListeners(); if (callback) { callback(); } } // If a socket that is already connected is being // removed, this means that this is a deliberate // kicking-out, and not a natural event that could // result in a reconnection later. Therefore, clean // up. if (socket.connected) { console.log(socket.user.name + " is connected, removing"); socket.on('disconnect', lastCB); socket.emit("kick"); socket.emit("disconnect"); } else { console.log(socket.user.name + " is not connected, removing"); lastCB(); } }); } function removeSocket(socket) { var index = sockets.indexOf(socket); detachSocket(socket); if (index !== -1) { sockets.splice(index, 1); } } this.getDocument = function () { return document; }; + this.isAvailable = function () { + return isAvailable; + }; + this.destroy = function (callback) { async.each(sockets, function (socket, cb) { detachSocket(socket, cb); }, function () { + //objectCache.forgetTrackedObject(chunk); + document.live = false; sockets.length = 0; recorder.destroy(callback); }); }; function init() { // Setup caching - document = objectCache.getTrackedObject(doc); DocumentChunk.findById(_.last(document.chunks), function (err, lastChunk) { - chunk = objectCache.getTrackedObject(lastChunk); + chunkManager = new ChunkManager(lastChunk); // Sanitize leftovers from previous session, if any sanitizeDocument(); - recorder = new Recorder(chunk, function () { + recorder = new Recorder(chunkManager.getLastChunk(), function () { + isAvailable = true; cb(); }); }); } init(); }; module.exports = Room; diff --git a/server/components/objectcache/index.js b/server/components/objectcache/index.js index a9a484a..c5b184b 100644 --- a/server/components/objectcache/index.js +++ b/server/components/objectcache/index.js @@ -1,89 +1,89 @@ /* * Copyright (C) 2015 KO GmbH * * @licstart * This file is part of Kotype. * * Kotype is free software: you can redistribute it and/or modify it * under the terms of the GNU Affero General Public License (GNU AGPL) * as published by the Free Software Foundation, either version 3 of * the License, or (at your option) any later version. * * Kotype is distributed in the hope that it will be useful, but * WITHOUT ANY WARRANTY; without even the implied warranty of * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the * GNU Affero General Public License for more details. * * You should have received a copy of the GNU Affero General Public License * along with Kotype. If not, see . * @licend * * @source: https://github.com/kogmbh/Kotype/ */ /*jslint nomen: true, unparam: true */ /*global require, console, setInterval, module */ var mongoose = require("mongoose"), async = require("async"), ObjectCache; -// Maintains an in-memory cache of objects from mongoose collections, +// Maintains an in-memory cache of documents from mongoose collections, // and writes them to the DB periodically. var ObjectCache = function () { "use strict"; var objects = {}, timer, writeInterval = 1000 * 5; function isTracked(object) { return objects.hasOwnProperty(object._id); } this.isTracked = isTracked; function getTrackedObject(object) { var id = object._id; if (!objects[id]) { objects[id] = object; } return objects[id]; } this.getTrackedObject = getTrackedObject; function forgetTrackedObject(object) { var id = object._id; if (objects.hasOwnProperty(id)) { delete objects[id]; } } this.forgetTrackedObject = forgetTrackedObject; function saveObjects(callback) { async.each(Object.keys(objects), function (id, cb) { - if (objects[id].isModified()) { + if (objects[id].isModified() && objects[id].live !== false) { objects[id].save(cb); } else { cb(); } }, callback); } this.destroy = function (callback) { clearInterval(timer); saveObjects(callback); }; function init() { timer = setInterval(function () { saveObjects(); }, writeInterval); } init(); }; module.exports = ObjectCache; diff --git a/server/config/local.env.sample.js b/server/config/local.env.sample.js index 8604415..2976d7a 100644 --- a/server/config/local.env.sample.js +++ b/server/config/local.env.sample.js @@ -1,25 +1,25 @@ 'use strict'; // Use local.env.js for environment variables that grunt will set when the server starts locally. // Use for your api keys, secrets, etc. This file should not be tracked by git. // // You will need to set these on the server you deploy to. module.exports = { DOMAIN: 'http://localhost:9000', SESSION_SECRET: 'manticore-secret', // Control debug level for modules using visionmedia/debug DEBUG: '', /* * Supported authentication strategies. * 1. 'local' for storing everything in Mongo/GridFS, auth using username/password * 2. 'webdav' for linking with a WebDAV server, auth using WebDAV credentials */ STORAGE: 'webdav', // More configuration for the chosen auth type. None required for 'local' WEBDAV_SERVER: 'https://apps.kolabnow.com', - WEBDAV_PATH: '/' + WEBDAV_PATH: '/files/Files' };