Merge branch 'master' into remove_mongo_doc_ops

Conflicts:
	app/coffee/DocOpsManager.coffee
	test/acceptance/coffee/ApplyingUpdatesToADocTests.coffee
	test/acceptance/coffee/FlushingDocsTests.coffee
	test/unit/coffee/DocOpsManager/DocOpsManagerTests.coffee
	test/unit/coffee/RedisManager/prependDocOpsTests.coffee
	test/unit/coffee/RedisManager/pushDocOpTests.coffee

services/document-updater/.gitignore

@@ -43,4 +43,6 @@ app/js/*
 test/unit/js/*
 test/acceptance/js/*
 
+forever/
+
 **.swp

services/document-updater/.travis.yml

@@ -10,8 +10,12 @@ install:
     - npm install
     - grunt install
 
+before_script:
+    - grunt forever:app:start
+
 script:
     - grunt test:unit
+    - grunt test:acceptance
 
 services:
     - redis-server

services/document-updater/Gruntfile.coffee

@@ -5,8 +5,14 @@ module.exports = (grunt) ->
 	grunt.loadNpmTasks 'grunt-available-tasks'
 	grunt.loadNpmTasks 'grunt-execute'
 	grunt.loadNpmTasks 'grunt-bunyan'
+	grunt.loadNpmTasks 'grunt-forever'
 	
 	grunt.initConfig
+		forever:
+			app:
+				options:
+					index: "app.js"
+
 		execute:
 			app:
 				src: "app.js"
@@ -49,12 +55,12 @@ module.exports = (grunt) ->
 
 		mochaTest:
 			unit:
-				src: ['test/unit/js/**/*.js']
+				src: ["test/unit/js/#{grunt.option('feature') or '**'}/*.js"]
 				options:
 					reporter: grunt.option('reporter') or 'spec'
 					grep: grunt.option("grep")
 			acceptance:
-				src: ['test/acceptance/js/**/*.js']
+				src: ["test/acceptance/js/#{grunt.option('feature') or '*'}.js"]
 				options:
 					reporter: grunt.option('reporter') or 'spec'
 					grep: grunt.option("grep")

services/document-updater/app.coffee

@@ -7,7 +7,6 @@ RedisManager = require('./app/js/RedisManager.js')
 UpdateManager = require('./app/js/UpdateManager.js')
 Keys = require('./app/js/RedisKeyBuilder')
 redis = require('redis')
-metrics = require('./app/js/Metrics')
 Errors = require "./app/js/Errors"
 HttpController = require "./app/js/HttpController"
 
@@ -15,33 +14,28 @@ redisConf = Settings.redis.web
 rclient = redis.createClient(redisConf.port, redisConf.host)
 rclient.auth(redisConf.password)
 
+Path = require "path"
+Metrics = require "metrics-sharelatex"
+Metrics.initialize("doc-updater")
+Metrics.mongodb.monitor(Path.resolve(__dirname + "/node_modules/mongojs/node_modules/mongodb"), logger)
+
 app = express()
 app.configure ->
-	app.use(express.logger(':remote-addr - [:date] - :user-agent ":method :url" :status - :response-time ms'));
+	app.use(Metrics.http.monitor(logger));
 	app.use express.bodyParser()
 	app.use app.router
 
-app.configure 'development', ()->
-	console.log "Development Enviroment"
-	app.use express.errorHandler({ dumpExceptions: true, showStack: true })
-
-app.configure 'production', ()->
-	console.log "Production Enviroment"
-	app.use express.logger()
-	app.use express.errorHandler()
-
 rclient.subscribe("pending-updates")
-rclient.on "message", (channel, doc_key)->
+rclient.on "message", (channel, doc_key) ->
 	[project_id, doc_id] = Keys.splitProjectIdAndDocId(doc_key)
-	UpdateManager.processOutstandingUpdatesWithLock project_id, doc_id, (error) ->
-		logger.error err: error, project_id: project_id, doc_id: doc_id, "error processing update" if error?
+	if !Settings.shuttingDown
+		UpdateManager.processOutstandingUpdatesWithLock project_id, doc_id, (error) ->
+			logger.error err: error, project_id: project_id, doc_id: doc_id, "error processing update" if error?
+	else
+		logger.log project_id: project_id, doc_id: doc_id, "ignoring incoming update" 
 
 UpdateManager.resumeProcessing()
 
-app.use (req, res, next)->
-	metrics.inc "http-request"
-	next()
-
 app.get    '/project/:project_id/doc/:doc_id',       HttpController.getDoc
 app.post   '/project/:project_id/doc/:doc_id',       HttpController.setDoc
 app.post   '/project/:project_id/doc/:doc_id/flush', HttpController.flushDocIfLoaded
@@ -50,13 +44,16 @@ app.delete '/project/:project_id',                   HttpController.deleteProjec
 app.post   '/project/:project_id/flush',             HttpController.flushProject
 
 app.get '/total', (req, res)->
-	timer = new metrics.Timer("http.allDocList")	
+	timer = new Metrics.Timer("http.allDocList")	
 	RedisManager.getCountOfDocsInMemory (err, count)->
 		timer.done()
 		res.send {total:count}
 	
 app.get '/status', (req, res)->
-	res.send('document updater is alive')
+	if Settings.shuttingDown
+		res.send 503 # Service unavailable
+	else
+		res.send('document updater is alive')
 
 app.use (error, req, res, next) ->
 	logger.error err: error, "request errored"
@@ -65,6 +62,18 @@ app.use (error, req, res, next) ->
 	else
 		res.send(500, "Oops, something went wrong")
 
+shutdownCleanly = (signal) ->
+	return () ->
+		logger.log signal: signal, "received interrupt, cleaning up"
+		Settings.shuttingDown = true
+		setTimeout () ->
+			logger.log signal: signal, "shutting down"
+			process.exit()
+		, 10000
+
 port = Settings.internal?.documentupdater?.port or Settings.apis?.documentupdater?.port or 3003
 app.listen port, "localhost", ->
 	logger.log("documentupdater-sharelatex server listening on port #{port}")
+
+for signal in ['SIGINT', 'SIGHUP', 'SIGQUIT', 'SIGUSR1', 'SIGUSR2', 'SIGTERM', 'SIGABRT']
+	process.on signal, shutdownCleanly(signal)

services/document-updater/app/coffee/DocOpsManager.coffee

@@ -1,4 +1,5 @@
 RedisManager = require "./RedisManager"
+TrackChangesManager = require "./TrackChangesManager"
 
 module.exports = DocOpsManager =
 	getPreviousDocOps: (project_id, doc_id, start, end, callback = (error, ops) ->) ->
@@ -7,5 +8,9 @@ module.exports = DocOpsManager =
 			callback null, ops
 
 	pushDocOp: (project_id, doc_id, op, callback = (error) ->) ->
-		RedisManager.pushDocOp doc_id, op, callback
+		RedisManager.pushDocOp doc_id, op, (error, version) ->
+			return callback(error) if error?
+			TrackChangesManager.pushUncompressedHistoryOp project_id, doc_id, op, (error) ->
+				return callback(error) if error?
+				callback null, version
 

services/document-updater/app/coffee/DocumentManager.coffee

@@ -40,7 +40,7 @@ module.exports = DocumentManager =
 					return callback(error) if error?
 					callback null, lines, version, ops
 
-	setDoc: (project_id, doc_id, newLines, _callback = (error) ->) ->
+	setDoc: (project_id, doc_id, newLines, source, user_id, _callback = (error) ->) ->
 		timer = new Metrics.Timer("docManager.setDoc")
 		callback = (args...) ->
 			timer.done()
@@ -66,6 +66,8 @@ module.exports = DocumentManager =
 					v: version
 					meta:
 						type: "external"
+						source: source
+						user_id: user_id
 				UpdateManager.applyUpdates project_id, doc_id, [update], (error) ->
 					return callback(error) if error?
 					DocumentManager.flushDocIfLoaded project_id, doc_id, (error) ->
@@ -110,9 +112,9 @@ module.exports = DocumentManager =
 		UpdateManager = require "./UpdateManager"
 		UpdateManager.lockUpdatesAndDo DocumentManager.getDocAndRecentOps, project_id, doc_id, fromVersion, callback
 		
-	setDocWithLock: (project_id, doc_id, lines, callback = (error) ->) ->
+	setDocWithLock: (project_id, doc_id, lines, source, user_id, callback = (error) ->) ->
 		UpdateManager = require "./UpdateManager"
-		UpdateManager.lockUpdatesAndDo DocumentManager.setDoc, project_id, doc_id, lines, callback
+		UpdateManager.lockUpdatesAndDo DocumentManager.setDoc, project_id, doc_id, lines, source, user_id, callback
 		
 	flushDocIfLoadedWithLock: (project_id, doc_id, callback = (error) ->) ->
 		UpdateManager = require "./UpdateManager"

services/document-updater/app/coffee/HttpController.coffee

@@ -32,9 +32,11 @@ module.exports = HttpController =
 		doc_id = req.params.doc_id
 		project_id = req.params.project_id
 		lines = req.body.lines
-		logger.log project_id: project_id, doc_id: doc_id, lines: lines, "setting doc via http"
+		source = req.body.source
+		user_id = req.body.user_id
+		logger.log project_id: project_id, doc_id: doc_id, lines: lines, source: source, user_id: user_id, "setting doc via http"
 		timer = new Metrics.Timer("http.setDoc")
-		DocumentManager.setDocWithLock project_id, doc_id, lines, (error) ->
+		DocumentManager.setDocWithLock project_id, doc_id, lines, source, user_id, (error) ->
 			timer.done()
 			return next(error) if error?
 			logger.log project_id: project_id, doc_id: doc_id, "set doc via http"

services/document-updater/app/coffee/LockManager.coffee

@@ -10,10 +10,10 @@ logger = require "logger-sharelatex"
 module.exports = LockManager =
 	LOCK_TEST_INTERVAL: 50 # 50ms between each test of the lock
 	MAX_LOCK_WAIT_TIME: 10000 # 10s maximum time to spend trying to get the lock
+	REDIS_LOCK_EXPIRY: 30 # seconds. Time until lock auto expires in redis.
 
 	tryLock : (doc_id, callback = (err, isFree)->)->
-		tenSeconds = 10
-		rclient.set keys.blockingKey(doc_id: doc_id), "locked", "EX", 10, "NX", (err, gotLock)->
+		rclient.set keys.blockingKey(doc_id: doc_id), "locked", "EX", LockManager.REDIS_LOCK_EXPIRY, "NX", (err, gotLock)->
 			return callback(err) if err?
 			if gotLock == "OK"
 				metrics.inc "doc-not-blocking"

services/document-updater/app/coffee/Metrics.coffee

@@ -1,23 +1 @@
-StatsD = require('lynx')
-statsd = new StatsD('localhost', 8125, {on_error:->})
-
-buildKey = (key)-> "doc-updater.#{process.env.NODE_ENV}.#{key}"
-
-module.exports =
-	set : (key, value, sampleRate = 1)->
-		statsd.set buildKey(key), value, sampleRate
-
-	inc : (key, sampleRate = 1)->
-		statsd.increment buildKey(key), sampleRate
-
-	Timer : class
-		constructor :(key, sampleRate = 1)->
-			this.start = new Date()
-			this.key = buildKey(key)
-		done:->
-			timeSpan = new Date - this.start
-			statsd.timing(this.key, timeSpan, this.sampleRate)
-
-	gauge : (key, value, sampleRate = 1)->
-		statsd.gauge key, value, sampleRate
-
+module.exports = require "metrics-sharelatex"

services/document-updater/app/coffee/RedisKeyBuilder.coffee

@@ -8,12 +8,15 @@ DOCLINES = "doclines"
 DOCOPS = "DocOps"
 DOCVERSION = "DocVersion"
 DOCIDSWITHPENDINGUPDATES = "DocsWithPendingUpdates"
+DOCSWITHHISTORYOPS = "DocsWithHistoryOps"
+UNCOMPRESSED_HISTORY_OPS = "UncompressedHistoryOps"
 
 module.exports =
 
 	allDocs : ALLDOCSKEY
 	docLines : (op)-> DOCLINES+":"+op.doc_id
 	docOps : (op)-> DOCOPS+":"+op.doc_id
+	uncompressedHistoryOp: (op) -> UNCOMPRESSED_HISTORY_OPS + ":" + op.doc_id
 	docVersion : (op)-> DOCVERSION+":"+op.doc_id
 	projectKey : (op)-> PROJECTKEY+":"+op.doc_id
 	blockingKey : (op)-> BLOCKINGKEY+":"+op.doc_id
@@ -23,6 +26,7 @@ module.exports =
 	docsWithPendingUpdates : DOCIDSWITHPENDINGUPDATES
 	combineProjectIdAndDocId: (project_id, doc_id) -> "#{project_id}:#{doc_id}"
 	splitProjectIdAndDocId: (project_and_doc_id) -> project_and_doc_id.split(":")
+	docsWithHistoryOps: (op) -> DOCSWITHHISTORYOPS + ":" + op.project_id
 	now : (key)->
 		d = new Date()
 		d.getDate()+":"+(d.getMonth()+1)+":"+d.getFullYear()+":"+key

services/document-updater/app/coffee/RedisManager.coffee

@@ -156,12 +156,22 @@ module.exports = RedisManager =
 			version = parseInt(version, 10)
 			callback null, version
 
+	pushUncompressedHistoryOp: (project_id, doc_id, op, callback = (error, length) ->) ->
+		jsonOp = JSON.stringify op
+		multi = rclient.multi()
+		multi.rpush keys.uncompressedHistoryOp(doc_id: doc_id), jsonOp
+		multi.sadd keys.docsWithHistoryOps(project_id: project_id), doc_id
+		multi.exec (error, results) ->
+			return callback(error) if error?
+			[length, _] = results
+			callback(error, length)
+
 	getDocOpsLength: (doc_id, callback = (error, length) ->) ->
 		rclient.llen keys.docOps(doc_id: doc_id), callback
 
 	getDocIdsInProject: (project_id, callback = (error, doc_ids) ->) ->
 		rclient.smembers keys.docsInProject(project_id: project_id), callback
-		
+
 
 getDocumentsProjectId = (doc_id, callback)->
 	rclient.get keys.projectKey({doc_id:doc_id}), (err, project_id)->

services/document-updater/app/coffee/ShareJsDB.coffee

@@ -4,6 +4,7 @@ DocumentManager = require "./DocumentManager"
 RedisManager = require "./RedisManager"
 DocOpsManager = require "./DocOpsManager"
 Errors = require "./Errors"
+logger = require "logger-sharelatex"
 
 module.exports = ShareJsDB =
 	getOps: (doc_key, start, end, callback) ->
@@ -23,15 +24,15 @@ module.exports = ShareJsDB =
 	
 	writeOp: (doc_key, opData, callback) ->
 		[project_id, doc_id] = Keys.splitProjectIdAndDocId(doc_key)
-		DocOpsManager.pushDocOp project_id, doc_id, {op:opData.op, meta:opData.meta}, (error, version) ->
+		DocOpsManager.pushDocOp project_id, doc_id, opData, (error, version) ->
 			return callback error if error?
 
 			if version == opData.v + 1
 				callback()
 			else
-				# The document has been corrupted by the change. For now, throw an exception.
-				# Later, rebuild the snapshot.
-				callback "Version mismatch in db.append. '#{doc_id}' is corrupted."
+				error = new Error("Version mismatch. '#{doc_id}' is corrupted.")
+				logger.error err: error, doc_id: doc_id, project_id: project_id, opVersion: opData.v, expectedVersion: version, "doc is corrupt"
+				callback error
 
 	getSnapshot: (doc_key, callback) ->
 		[project_id, doc_id] = Keys.splitProjectIdAndDocId(doc_key)

services/document-updater/app/coffee/ShareJsUpdateManager.coffee

@@ -44,10 +44,7 @@ module.exports = ShareJsUpdateManager =
 				if error?
 					@_sendError(project_id, doc_id, error)
 					return callback(error)
-				if typeof data.snapshot == "string"
-					docLines = data.snapshot.split("\n")
-				else
-					docLines = data.snapshot.lines
+				docLines = data.snapshot.split(/\r\n|\n|\r/)
 				callback(null, docLines, data.v)
 
 	_listenForOps: (model) ->

services/document-updater/app/coffee/TrackChangesManager.coffee

@@ -0,0 +1,36 @@
+settings = require "settings-sharelatex"
+request  = require "request"
+logger = require "logger-sharelatex"
+RedisManager = require "./RedisManager"
+crypto = require("crypto")
+
+module.exports = TrackChangesManager =
+	flushDocChanges: (project_id, doc_id, callback = (error) ->) ->
+		if !settings.apis?.trackchanges?
+			logger.warn doc_id: doc_id, "track changes API is not configured, so not flushing"
+			return callback()
+
+		url = "#{settings.apis.trackchanges.url}/project/#{project_id}/doc/#{doc_id}/flush"
+		logger.log project_id: project_id, doc_id: doc_id, url: url, "flushing doc in track changes api"
+		request.post url, (error, res, body)->
+			if error?
+				return callback(error)
+			else if res.statusCode >= 200 and res.statusCode < 300
+				return callback(null)
+			else
+				error = new Error("track changes api returned a failure status code: #{res.statusCode}")
+				return callback(error)
+
+	FLUSH_EVERY_N_OPS: 50
+	pushUncompressedHistoryOp: (project_id, doc_id, op, callback = (error) ->) ->
+		RedisManager.pushUncompressedHistoryOp project_id, doc_id, op, (error, length) ->
+			return callback(error) if error?
+			if length > 0 and length % TrackChangesManager.FLUSH_EVERY_N_OPS == 0
+				# Do this in the background since it uses HTTP and so may be too
+				# slow to wait for when processing a doc update.
+				logger.log length: length, doc_id: doc_id, project_id: project_id, "flushing track changes api"
+				TrackChangesManager.flushDocChanges project_id, doc_id,  (error) ->
+					if error?
+						logger.error err: error, doc_id: doc_id, project_id: project_id, "error flushing doc to track changes api"
+			callback()
+

services/document-updater/config/settings.development.coffee

@@ -12,6 +12,8 @@ module.exports =
 			url: "http://localhost:3000"
 			user: "sharelatex"
 			pass: "password"
+		trackchanges:
+			url: "http://localhost:3014"
 
 	redis:
 		web:

services/document-updater/package.json

@@ -14,6 +14,7 @@
     "coffee-script": "1.4.0",
     "settings-sharelatex": "git+https://github.com/sharelatex/settings-sharelatex.git#master",
     "logger-sharelatex": "git+https://github.com/sharelatex/logger-sharelatex.git#master",
+    "metrics-sharelatex": "git+https://github.com/sharelatex/metrics-sharelatex.git#master",
     "sinon": "~1.5.2",
     "mongojs": "0.9.11"
   },
@@ -25,6 +26,7 @@
     "grunt-available-tasks": "~0.4.1",
     "grunt-contrib-coffee": "~0.10.0",
     "bunyan": "~0.22.1",
-    "grunt-bunyan": "~0.5.0"
+    "grunt-bunyan": "~0.5.0",
+    "grunt-forever": "~0.4.2"
   }
 }

services/document-updater/test/acceptance/coffee/ApplyingUpdatesToADocTests.coffee

@@ -2,7 +2,9 @@ sinon = require "sinon"
 chai = require("chai")
 chai.should()
 async = require "async"
+rclient = require("redis").createClient()
 
+MockTrackChangesApi = require "./helpers/MockTrackChangesApi"
 MockWebApi = require "./helpers/MockWebApi"
 DocUpdaterClient = require "./helpers/DocUpdaterClient"
 
@@ -44,6 +46,16 @@ describe "Applying updates to a doc", ->
 				doc.lines.should.deep.equal @result
 				done()
 
+		it "should push the applied updates to the track changes api", (done) ->
+			rclient.lrange "UncompressedHistoryOps:#{@doc_id}", 0, -1, (error, updates) =>
+				throw error if error?
+				JSON.parse(updates[0]).op.should.deep.equal @update.op
+				rclient.sismember "DocsWithHistoryOps:#{@project_id}", @doc_id, (error, result) =>
+					throw error if error?
+					result.should.equal 1
+					done()
+
+
 	describe "when the document is loaded", ->
 		before (done) ->
 			[@project_id, @doc_id] = [DocUpdaterClient.randomId(), DocUpdaterClient.randomId()]
@@ -69,6 +81,13 @@ describe "Applying updates to a doc", ->
 				doc.lines.should.deep.equal @result
 				done()
 
+		it "should push the applied updates to the track changes api", (done) ->
+			rclient.lrange "UncompressedHistoryOps:#{@doc_id}", 0, -1, (error, updates) =>
+				JSON.parse(updates[0]).op.should.deep.equal @update.op
+				rclient.sismember "DocsWithHistoryOps:#{@project_id}", @doc_id, (error, result) =>
+					result.should.equal 1
+					done()
+
 	describe "when the document has been deleted", ->
 		describe "when the ops come in a single linear order", ->
 			before ->
@@ -110,6 +129,16 @@ describe "Applying updates to a doc", ->
 						doc.lines.should.deep.equal @result
 						done()
 
+			it "should push the applied updates to the track changes api", (done) ->
+				rclient.lrange "UncompressedHistoryOps:#{@doc_id}", 0, -1, (error, updates) =>
+					updates = (JSON.parse(u) for u in updates)
+					for appliedUpdate, i in @updates
+						appliedUpdate.op.should.deep.equal updates[i].op
+
+					rclient.sismember "DocsWithHistoryOps:#{@project_id}", @doc_id, (error, result) =>
+						result.should.equal 1
+						done()
+
 		describe "when older ops come in after the delete", ->
 			before ->
 				[@project_id, @doc_id] = [DocUpdaterClient.randomId(), DocUpdaterClient.randomId()]
@@ -160,4 +189,30 @@ describe "Applying updates to a doc", ->
 			DocUpdaterClient.getDoc @project_id, @doc_id, (error, res, doc) =>
 				doc.lines.should.deep.equal @lines
 				done()
-	
+
+	describe "with enough updates to flush to the track changes api", ->
+		beforeEach (done) ->
+			[@project_id, @doc_id] = [DocUpdaterClient.randomId(), DocUpdaterClient.randomId()]
+			MockWebApi.insertDoc @project_id, @doc_id, {
+				lines: @lines
+				version: 0
+			}
+			@updates = []
+			for v in [0..99] # Should flush after 50 ops
+				@updates.push
+					doc_id: @doc_id,
+					op: [i: v.toString(), p: 0]
+					v: v
+
+			sinon.spy MockTrackChangesApi, "flushDoc"
+
+			DocUpdaterClient.sendUpdates @project_id, @doc_id, @updates, (error) =>
+				throw error if error?
+				setTimeout done, 200
+
+		afterEach ->
+			MockTrackChangesApi.flushDoc.restore()
+
+		it "should flush the doc twice", ->
+			MockTrackChangesApi.flushDoc.calledTwice.should.equal true
+

services/document-updater/test/acceptance/coffee/FlushingAProjectTests.coffee

@@ -41,6 +41,7 @@ describe "Flushing a project", ->
 	describe "with documents which have been updated", ->
 		before (done) ->
 			sinon.spy MockWebApi, "setDocumentLines"
+
 			async.series @docs.map((doc) =>
 				(callback) =>
 					DocUpdaterClient.preloadDoc @project_id, doc.id, (error) =>

services/document-updater/test/acceptance/coffee/FlushingDocsTests.coffee

@@ -72,5 +72,4 @@ describe "Flushing a doc to Mongo", ->
 			MockWebApi.setDocumentLines.called.should.equal false
 			MockWebApi.setDocumentVersion.called.should.equal false
 
-			
 

services/document-updater/test/acceptance/coffee/SettingADocumentTests.coffee

@@ -19,6 +19,8 @@ describe "Setting a document", ->
 			v: @version
 		@result = ["one", "one and a half", "two", "three"]
 		@newLines = ["these", "are", "the", "new", "lines"]
+		@source = "dropbox"
+		@user_id = "user-id-123"
 		MockWebApi.insertDoc @project_id, @doc_id, {
 			lines: @lines
 			version: @version
@@ -33,7 +35,7 @@ describe "Setting a document", ->
 				DocUpdaterClient.sendUpdate @project_id, @doc_id, @update, (error) =>
 					throw error if error?
 					setTimeout () =>
-						DocUpdaterClient.setDocLines @project_id, @doc_id, @newLines, (error, res, body) =>
+						DocUpdaterClient.setDocLines @project_id, @doc_id, @newLines, @source, @user_id, (error, res, body) =>
 							@statusCode = res.statusCode
 							done()
 					, 200

services/document-updater/test/acceptance/coffee/helpers/DocUpdaterClient.coffee

@@ -45,11 +45,13 @@ module.exports = DocUpdaterClient =
 		request.post "http://localhost:3003/project/#{project_id}/doc/#{doc_id}/flush", (error, res, body) ->
 			callback error, res, body
 
-	setDocLines: (project_id, doc_id, lines, callback = (error) ->) ->
+	setDocLines: (project_id, doc_id, lines, source, user_id, callback = (error) ->) ->
 		request.post {
 			url: "http://localhost:3003/project/#{project_id}/doc/#{doc_id}"
 			json:
 				lines: lines
+				source: source
+				user_id: user_id
 		}, (error, res, body) ->
 			callback error, res, body
 

services/document-updater/test/acceptance/coffee/helpers/MockTrackChangesApi.coffee

@@ -0,0 +1,20 @@
+express = require("express")
+app = express()
+
+module.exports = MockTrackChangesApi =
+	flushDoc: (doc_id, callback = (error) ->) ->
+		callback()
+
+	run: () ->
+		app.post "/project/:project_id/doc/:doc_id/flush", (req, res, next) =>
+			@flushDoc req.params.doc_id, (error) ->
+				if error?
+					res.send 500
+				else
+					res.send 204
+
+		app.listen 3014, (error) ->
+			throw error if error?
+
+MockTrackChangesApi.run()
+

services/document-updater/test/acceptance/coffee/helpers/MockWebApi.coffee

@@ -40,7 +40,8 @@ module.exports = MockWebApi =
 					else
 						res.send 204
 
-		app.listen(3000)
+		app.listen 3000, (error) ->
+			throw error if error?
 
 MockWebApi.run()
 

services/document-updater/test/unit/coffee/DocOpsManager/DocOpsManagerTests.coffee

@@ -12,6 +12,10 @@ describe "DocOpsManager", ->
 		@DocOpsManager = SandboxedModule.require modulePath, requires:
 			"./RedisManager": @RedisManager = {}
 			"logger-sharelatex": @logger = { log: sinon.stub(), error: sinon.stub() }
+			"./Metrics": @Metrics =
+				Timer: class Timer
+					done: sinon.stub()
+			"./TrackChangesManager": @TrackChangesManager = {}
 
 	describe "getPreviousDocOps", ->
 		beforeEach ->
@@ -28,3 +32,24 @@ describe "DocOpsManager", ->
 
 		it "should call the callback with the ops", ->
 			@callback.calledWith(null, @ops).should.equal true
+
+	describe "pushDocOp", ->
+		beforeEach ->
+			@op = "mock-op"
+			@RedisManager.pushDocOp = sinon.stub().callsArgWith(2, null, @version = 42)
+			@TrackChangesManager.pushUncompressedHistoryOp = sinon.stub().callsArg(3)
+			@DocOpsManager.pushDocOp @project_id, @doc_id, @op, @callback
+
+		it "should push the op in to the docOps list", ->
+			@RedisManager.pushDocOp
+				.calledWith(@doc_id, @op)
+				.should.equal true
+
+		it "should push the op into the pushUncompressedHistoryOp", ->
+			@TrackChangesManager.pushUncompressedHistoryOp
+				.calledWith(@project_id, @doc_id, @op)
+				.should.equal true
+
+		it "should call the callback with the version", ->
+			@callback.calledWith(null, @version).should.equal true
+

services/document-updater/test/unit/coffee/DocumentManager/setDocTests.coffee

@@ -22,6 +22,8 @@ describe "DocumentManager - setDoc", ->
 		@version = 42
 		@ops = ["mock-ops"]
 		@callback = sinon.stub()
+		@source = "dropbox"
+		@user_id = "mock-user-id"
 
 	describe "with plain tex lines", ->
 		beforeEach ->
@@ -34,7 +36,7 @@ describe "DocumentManager - setDoc", ->
 				@DiffCodec.diffAsShareJsOp = sinon.stub().callsArgWith(2, null, @ops)
 				@UpdateManager.applyUpdates = sinon.stub().callsArgWith(3, null)
 				@DocumentManager.flushDocIfLoaded = sinon.stub().callsArg(2)
-				@DocumentManager.setDoc @project_id, @doc_id, @afterLines, @callback
+				@DocumentManager.setDoc @project_id, @doc_id, @afterLines, @source, @user_id, @callback
 
 			it "should get the current doc lines", ->
 				@DocumentManager.getDoc
@@ -48,7 +50,20 @@ describe "DocumentManager - setDoc", ->
 
 			it "should apply the diff as a ShareJS op", ->
 				@UpdateManager.applyUpdates
-					.calledWith(@project_id, @doc_id, [doc: @doc_id, v: @version, op: @ops, meta: { type: "external" }])
+					.calledWith(
+						@project_id,
+						@doc_id,
+						[
+							doc: @doc_id,
+							v: @version,
+							op: @ops,
+							meta: {
+								type: "external"
+								source: @source
+								user_id: @user_id
+							}
+						]
+					)
 					.should.equal true
 
 			it "should flush the doc to Mongo", ->
@@ -62,30 +77,6 @@ describe "DocumentManager - setDoc", ->
 			it "should time the execution", ->
 				@Metrics.Timer::done.called.should.equal true
 
-	describe "with json lines", ->
-		beforeEach ->
-			@beforeLines = [text: "before", text: "lines"]
-			@afterLines = ["after", "lines"]
-			
-		describe "successfully", ->
-			beforeEach ->
-				@DocumentManager.getDoc = sinon.stub().callsArgWith(2, null, @beforeLines, @version)
-				@DiffCodec.diffAsShareJsOp = sinon.stub().callsArgWith(2, null, @ops)
-				@UpdateManager.applyUpdates = sinon.stub().callsArgWith(3, null)
-				@DocumentManager.flushDocIfLoaded = sinon.stub().callsArg(2)
-				@DocumentManager.setDoc @project_id, @doc_id, @afterLines, @callback
-
-			it "should get the current doc lines", ->
-				@DocumentManager.getDoc
-					.calledWith(@project_id, @doc_id)
-					.should.equal true
-
-			it "should return not try to get a diff", ->
-				@DiffCodec.diffAsShareJsOp.called.should.equal false
-
-			it "should call the callback", ->
-				@callback.calledWith(null).should.equal true
-
 	describe "without new lines", ->
 		beforeEach ->
 			@DocumentManager.getDoc = sinon.stub().callsArgWith(2, null, @beforeLines, @version)
@@ -96,7 +87,7 @@ describe "DocumentManager - setDoc", ->
 			
 		it "should not try to get the doc lines", ->
 			@DocumentManager.getDoc.called.should.equal false
-			
+
 		
 
 		

services/document-updater/test/unit/coffee/HttpController/setDocTests.coffee

@@ -19,6 +19,8 @@ describe "HttpController - setDoc", ->
 		@project_id = "project-id-123"
 		@doc_id = "doc-id-123"
 		@lines = ["one", "two", "three"]
+		@source = "dropbox"
+		@user_id = "user-id-123"
 		@res =
 			send: sinon.stub()
 		@req =
@@ -27,16 +29,18 @@ describe "HttpController - setDoc", ->
 				doc_id: @doc_id
 			body:
 				lines: @lines
+				source: @source
+				user_id: @user_id
 		@next = sinon.stub()
 
 	describe "successfully", ->
 		beforeEach ->
-			@DocumentManager.setDocWithLock = sinon.stub().callsArgWith(3)
+			@DocumentManager.setDocWithLock = sinon.stub().callsArgWith(5)
 			@HttpController.setDoc(@req, @res, @next)
 
 		it "should set the doc", ->
 			@DocumentManager.setDocWithLock
-				.calledWith(@project_id, @doc_id)
+				.calledWith(@project_id, @doc_id, @lines, @source, @user_id)
 				.should.equal true
 
 		it "should return a successful No Content response", ->
@@ -46,7 +50,7 @@ describe "HttpController - setDoc", ->
 
 		it "should log the request", ->
 			@logger.log
-				.calledWith(doc_id: @doc_id, project_id: @project_id, lines: @lines, "setting doc via http")
+				.calledWith(doc_id: @doc_id, project_id: @project_id, lines: @lines, source: @source, user_id: @user_id, "setting doc via http")
 				.should.equal true
 
 		it "should time the request", ->
@@ -54,7 +58,7 @@ describe "HttpController - setDoc", ->
 
 	describe "when an errors occurs", ->
 		beforeEach ->
-			@DocumentManager.setDocWithLock = sinon.stub().callsArgWith(3, new Error("oops"))
+			@DocumentManager.setDocWithLock = sinon.stub().callsArgWith(5, new Error("oops"))
 			@HttpController.setDoc(@req, @res, @next)
 
 		it "should call next with the error", ->

services/document-updater/test/unit/coffee/LockManager/tryLockTests.coffee

@@ -21,7 +21,7 @@ describe 'LockManager - trying the lock', ->
 			@LockManager.tryLock @doc_id, @callback
 
 		it "should set the lock key with an expiry if it is not set", ->
-			@set.calledWith("Blocking:#{@doc_id}", "locked", "EX", 10, "NX")
+			@set.calledWith("Blocking:#{@doc_id}", "locked", "EX", 30, "NX")
 				.should.equal true
 
 		it "should return the callback with true", ->

services/document-updater/test/unit/coffee/RedisManager/clearDocFromPendingUpdatesSetTests.coffee

@@ -12,6 +12,7 @@ describe "RedisManager.clearDocFromPendingUpdatesSet", ->
 		@RedisManager = SandboxedModule.require modulePath, requires:
 			"redis" : createClient: () =>
 				@rclient = auth:->
+			"logger-sharelatex": {}
 
 		@rclient.srem = sinon.stub().callsArg(2)
 		@RedisManager.clearDocFromPendingUpdatesSet(@project_id, @doc_id, @callback)

services/document-updater/test/unit/coffee/RedisManager/getDocsWithPendingUpdatesTests.coffee

@@ -10,6 +10,7 @@ describe "RedisManager.getDocsWithPendingUpdates", ->
 		@RedisManager = SandboxedModule.require modulePath, requires:
 			"redis" : createClient: () =>
 				@rclient = auth:->
+			"logger-sharelatex": {}
 
 		@docs = [{
 			doc_id: "doc-id-1"

services/document-updater/test/unit/coffee/RedisManager/pushUncompressedHistoryOpTests.coffee

@@ -0,0 +1,41 @@
+sinon = require('sinon')
+chai = require('chai')
+should = chai.should()
+modulePath = "../../../../app/js/RedisManager.js"
+SandboxedModule = require('sandboxed-module')
+
+describe "RedisManager.pushUncompressedHistoryOp", ->
+	beforeEach ->
+		@RedisManager = SandboxedModule.require modulePath, requires:
+			"redis": createClient: () =>
+				@rclient =
+					auth: () ->
+					multi: () => @rclient
+			"logger-sharelatex": @logger = {log: sinon.stub()}
+		@doc_id = "doc-id-123"
+		@project_id = "project-id-123"
+		@callback = sinon.stub()
+
+	describe "successfully", ->
+		beforeEach ->
+			@op = { op: [{ i: "foo", p: 4 }] }
+			@rclient.rpush = sinon.stub()
+			@rclient.sadd = sinon.stub()
+			@rclient.exec = sinon.stub().callsArgWith(0, null, [@length = 42, "1"])
+			@RedisManager.pushUncompressedHistoryOp @project_id, @doc_id, @op, @callback
+		
+		it "should push the doc op into the doc ops list", ->
+			@rclient.rpush
+				.calledWith("UncompressedHistoryOps:#{@doc_id}", JSON.stringify(@op))
+				.should.equal true
+
+		it "should add the doc_id to the set of which records the project docs", ->
+			@rclient.sadd
+				.calledWith("DocsWithHistoryOps:#{@project_id}", @doc_id)
+				.should.equal true
+
+		it "should call the callback with the length", ->
+			@callback.calledWith(null, @length).should.equal true
+
+
+

services/document-updater/test/unit/coffee/ShareJsDB/WriteOpsTests.coffee

@@ -26,11 +26,8 @@ describe "ShareJsDB.writeOps", ->
 			@ShareJsDB.writeOp @doc_key, @opData, @callback
 
 		it "should write the op to redis", ->
-			op =
-				op: @opData.op
-				meta: @opData.meta
 			@DocOpsManager.pushDocOp
-				.calledWith(@project_id, @doc_id, op)
+				.calledWith(@project_id, @doc_id, @opData)
 				.should.equal true
 
 		it "should call the callback without an error", ->
@@ -46,7 +43,7 @@ describe "ShareJsDB.writeOps", ->
 			@ShareJsDB.writeOp @doc_key, @opData, @callback
 
 		it "should call the callback with an error", ->
-			@callback.calledWith(sinon.match.string).should.equal true
+			@callback.calledWith(new Error()).should.equal true
 
 
 	

services/document-updater/test/unit/coffee/ShareJsUpdateManagerTests.coffee

@@ -29,113 +29,74 @@ describe "ShareJsUpdateManager", ->
 			@ShareJsUpdateManager.getNewShareJsModel = sinon.stub().returns(@model)
 			@ShareJsUpdateManager._listenForOps = sinon.stub()
 			@ShareJsUpdateManager.removeDocFromCache = sinon.stub().callsArg(1)
+			@updates = [
+				{p: 4, t: "foo"}
+				{p: 6, t: "bar"}
+			]
+			@updatedDocLines = ["one", "two"]
 
-		describe "with a text document", ->
-			beforeEach ->
-				@updates = [
-					{p: 4, t: "foo"}
-					{p: 6, t: "bar"}
-				]
-				@updatedDocLines = ["one", "two"]
+		describe "successfully", ->
+			beforeEach (done) ->
+				@model.getSnapshot.callsArgWith(1, null, {snapshot: @updatedDocLines.join("\n"), v: @version})
+				@ShareJsUpdateManager.applyUpdates @project_id, @doc_id, @updates, (err, docLines, version) =>
+					@callback(err, docLines, version)
+					done()
 
-			describe "successfully", ->
-				beforeEach (done) ->
-					@model.getSnapshot.callsArgWith(1, null, {snapshot: @updatedDocLines.join("\n"), v: @version})
-					@ShareJsUpdateManager.applyUpdates @project_id, @doc_id, @updates, (err, docLines, version) =>
-						@callback(err, docLines, version)
-						done()
+			it "should create a new ShareJs model", ->
+				@ShareJsUpdateManager.getNewShareJsModel
+					.called.should.equal true
 
-				it "should create a new ShareJs model", ->
-					@ShareJsUpdateManager.getNewShareJsModel
-						.called.should.equal true
+			it "should listen for ops on the model", ->
+				@ShareJsUpdateManager._listenForOps
+					.calledWith(@model)
+					.should.equal true
 
-				it "should listen for ops on the model", ->
-					@ShareJsUpdateManager._listenForOps
-						.calledWith(@model)
-						.should.equal true
+			it "should send each update to ShareJs", ->
+				for update in @updates
+					@model.applyOp
+						.calledWith("#{@project_id}:#{@doc_id}", update).should.equal true
 
-				it "should send each update to ShareJs", ->
-					for update in @updates
-						@model.applyOp
-							.calledWith("#{@project_id}:#{@doc_id}", update).should.equal true
+			it "should get the updated doc lines", ->
+				@model.getSnapshot
+					.calledWith("#{@project_id}:#{@doc_id}")
+					.should.equal true
 
-				it "should get the updated doc lines", ->
-					@model.getSnapshot
-						.calledWith("#{@project_id}:#{@doc_id}")
-						.should.equal true
+			it "should return the updated doc lines", ->
+				@callback.calledWith(null, @updatedDocLines, @version).should.equal true
 
-				it "should return the updated doc lines", ->
-					@callback.calledWith(null, @updatedDocLines, @version).should.equal true
+		describe "when applyOp fails", ->
+			beforeEach (done) ->
+				@error = new Error("Something went wrong")
+				@ShareJsUpdateManager._sendError = sinon.stub()
+				@model.applyOp = sinon.stub().callsArgWith(2, @error)
+				@ShareJsUpdateManager.applyUpdates @project_id, @doc_id, @updates, (err, docLines, version) =>
+					@callback(err, docLines, version)
+					done()
 
-			describe "when applyOp fails", ->
-				beforeEach (done) ->
-					@error = new Error("Something went wrong")
-					@ShareJsUpdateManager._sendError = sinon.stub()
-					@model.applyOp = sinon.stub().callsArgWith(2, @error)
-					@ShareJsUpdateManager.applyUpdates @project_id, @doc_id, @updates, (err, docLines, version) =>
-						@callback(err, docLines, version)
-						done()
+			it "should call sendError with the error", ->
+				@ShareJsUpdateManager._sendError
+					.calledWith(@project_id, @doc_id, @error)
+					.should.equal true
 
-				it "should call sendError with the error", ->
-					@ShareJsUpdateManager._sendError
-						.calledWith(@project_id, @doc_id, @error)
-						.should.equal true
+			it "should call the callback with the error", ->
+				@callback.calledWith(@error).should.equal true
 
-				it "should call the callback with the error", ->
-					@callback.calledWith(@error).should.equal true
+		describe "when getSnapshot fails", ->
+			beforeEach (done) ->
+				@error = new Error("Something went wrong")
+				@ShareJsUpdateManager._sendError = sinon.stub()
+				@model.getSnapshot.callsArgWith(1, @error)
+				@ShareJsUpdateManager.applyUpdates @project_id, @doc_id, @updates, (err, docLines, version) =>
+					@callback(err, docLines, version)
+					done()
 
-			describe "when getSnapshot fails", ->
-				beforeEach (done) ->
-					@error = new Error("Something went wrong")
-					@ShareJsUpdateManager._sendError = sinon.stub()
-					@model.getSnapshot.callsArgWith(1, @error)
-					@ShareJsUpdateManager.applyUpdates @project_id, @doc_id, @updates, (err, docLines, version) =>
-						@callback(err, docLines, version)
-						done()
+			it "should call sendError with the error", ->
+				@ShareJsUpdateManager._sendError
+					.calledWith(@project_id, @doc_id, @error)
+					.should.equal true
 
-				it "should call sendError with the error", ->
-					@ShareJsUpdateManager._sendError
-						.calledWith(@project_id, @doc_id, @error)
-						.should.equal true
-
-				it "should call the callback with the error", ->
-					@callback.calledWith(@error).should.equal true
-
-		describe "with a JSON document", ->
-			beforeEach ->
-				@updates = [
-					{p: ["lines", 0], dl: { foo: "bar "}}
-				]
-				@docLines = [text: "one", text: "two"]
-
-			describe "successfully", ->
-				beforeEach (done) ->
-					@model.getSnapshot.callsArgWith(1, null, {snapshot: {lines: @docLines}, v: @version})
-					@ShareJsUpdateManager.applyUpdates @project_id, @doc_id, @updates, (err, docLines, version) =>
-						@callback(err, docLines, version)
-						done()
-
-				it "should create a new ShareJs model", ->
-					@ShareJsUpdateManager.getNewShareJsModel
-						.called.should.equal true
-
-				it "should listen for ops on the model", ->
-					@ShareJsUpdateManager._listenForOps
-						.calledWith(@model)
-						.should.equal true
-
-				it "should send each update to ShareJs", ->
-					for update in @updates
-						@model.applyOp
-							.calledWith("#{@project_id}:#{@doc_id}", update).should.equal true
-
-				it "should get the updated doc lines", ->
-					@model.getSnapshot
-						.calledWith("#{@project_id}:#{@doc_id}")
-						.should.equal true
-
-				it "should return the updated doc lines", ->
-					@callback.calledWith(null, @docLines, @version).should.equal true
+			it "should call the callback with the error", ->
+				@callback.calledWith(@error).should.equal true
 
 	describe "_listenForOps", ->
 		beforeEach ->

services/document-updater/test/unit/coffee/TrackChangesManager/TrackChangesManagerTests.coffee

@@ -0,0 +1,92 @@
+SandboxedModule = require('sandboxed-module')
+sinon = require('sinon')
+require('chai').should()
+modulePath = require('path').join __dirname, '../../../../app/js/TrackChangesManager'
+
+describe "TrackChangesManager", ->
+	beforeEach ->
+		@TrackChangesManager = SandboxedModule.require modulePath, requires:
+			"request": @request = {}
+			"settings-sharelatex": @Settings = {}
+			"logger-sharelatex": @logger = { log: sinon.stub(), error: sinon.stub() }
+			"./RedisManager": @RedisManager = {}
+		@project_id = "mock-project-id"
+		@doc_id = "mock-doc-id"
+		@callback = sinon.stub()
+
+	describe "flushDocChanges", ->
+		beforeEach ->
+			@Settings.apis =
+				trackchanges: url: "http://trackchanges.example.com"
+
+		describe "successfully", ->
+			beforeEach ->
+				@request.post = sinon.stub().callsArgWith(1, null, statusCode: 204)
+				@TrackChangesManager.flushDocChanges @project_id, @doc_id, @callback
+
+			it "should send a request to the track changes api", ->
+				@request.post
+					.calledWith("#{@Settings.apis.trackchanges.url}/project/#{@project_id}/doc/#{@doc_id}/flush")
+					.should.equal true
+
+			it "should return the callback", ->
+				@callback.calledWith(null).should.equal true
+
+		describe "when the track changes api returns an error", ->
+			beforeEach ->
+				@request.post = sinon.stub().callsArgWith(1, null, statusCode: 500)
+				@TrackChangesManager.flushDocChanges @project_id, @doc_id, @callback
+
+			it "should return the callback with an error", ->
+				@callback.calledWith(new Error("track changes api return non-success code: 500")).should.equal true
+
+	describe "pushUncompressedHistoryOp", ->
+		beforeEach ->
+			@op = "mock-op"
+			@TrackChangesManager.flushDocChanges = sinon.stub().callsArg(2)
+
+		describe "pushing the op", ->
+			beforeEach ->
+				@RedisManager.pushUncompressedHistoryOp = sinon.stub().callsArgWith(3, null, 1)
+				@TrackChangesManager.pushUncompressedHistoryOp @project_id, @doc_id, @op, @callback
+
+			it "should push the op into redis", ->
+				@RedisManager.pushUncompressedHistoryOp
+					.calledWith(@project_id, @doc_id, @op)
+					.should.equal true
+
+			it "should call the callback", ->
+				@callback.called.should.equal true
+
+			it "should not try to flush the op", ->
+				@TrackChangesManager.flushDocChanges.called.should.equal false
+
+		describe "when there are a multiple of FLUSH_EVERY_N_OPS ops", ->
+			beforeEach ->
+				@RedisManager.pushUncompressedHistoryOp =
+					sinon.stub().callsArgWith(3, null, 2 * @TrackChangesManager.FLUSH_EVERY_N_OPS)
+				@TrackChangesManager.pushUncompressedHistoryOp @project_id, @doc_id, @op, @callback
+
+			it "should tell the track changes api to flush", ->
+				@TrackChangesManager.flushDocChanges
+					.calledWith(@project_id, @doc_id)
+					.should.equal true
+
+		describe "when TrackChangesManager errors", ->
+			beforeEach ->
+				@RedisManager.pushUncompressedHistoryOp =
+					sinon.stub().callsArgWith(3, null, 2 * @TrackChangesManager.FLUSH_EVERY_N_OPS)
+				@TrackChangesManager.flushDocChanges = sinon.stub().callsArgWith(2, @error = new Error("oops"))
+				@TrackChangesManager.pushUncompressedHistoryOp @project_id, @doc_id, @op, @callback
+
+			it "should log out the error", ->
+				@logger.error
+					.calledWith(
+						err: @error
+						doc_id: @doc_id
+						project_id: @project_id
+						"error flushing doc to track changes api"
+					)
+					.should.equal true
+
+