overleaf/services/track-changes/app/coffee/PackManager.coffee

async = require "async"
_ = require "underscore"
{db, ObjectId, BSON} = require "./mongojs"
logger = require "logger-sharelatex"
LockManager = require "./LockManager"
MongoAWS = require "./MongoAWS"
ProjectIterator = require "./ProjectIterator"

# Sharejs operations are stored in a 'pack' object
#
#  e.g.  a single sharejs update looks like
#
#   {
#     "doc_id" : 549dae9e0a2a615c0c7f0c98,
#     "project_id" : 549dae9c0a2a615c0c7f0c8c,
#     "op" : [ {"p" : 6981,	"d" : "?"	} ],
#     "meta" : {	"user_id" : 52933..., "start_ts" : 1422310693931,	"end_ts" : 1422310693931 },
#     "v" : 17082
#   }
#
#  and a pack looks like this
#
#   {
#     "doc_id" : 549dae9e0a2a615c0c7f0c98,
#     "project_id" : 549dae9c0a2a615c0c7f0c8c,
#     "pack" : [ U1, U2, U3, ...., UN],
#     "meta" : {	"user_id" : 52933..., "start_ts" : 1422310693931,	"end_ts" : 1422310693931 },
#     "v" : 17082
#     "v_end" : ...
#   }
#
#  where U1, U2, U3, .... are single updates stripped of their
#  doc_id and project_id fields (which are the same for all the
#  updates in the pack).
#
#  The pack itself has v and meta fields, this makes it possible to
#  treat packs and single updates in a similar way.
#
#  The v field of the pack itself is from the first entry U1, the
#  v_end field from UN.  The meta.end_ts field of the pack itself is
#  from the last entry UN, the meta.start_ts field from U1.

DAYS = 24 * 3600 * 1000 # one day in milliseconds

module.exports = PackManager =

	MAX_SIZE:  1024*1024 # make these configurable parameters
	MAX_COUNT: 1024

	insertCompressedUpdates: (project_id, doc_id, lastUpdate, newUpdates, temporary, callback = (error) ->) ->
		return callback() if newUpdates.length == 0

		updatesToFlush = []
		updatesRemaining = newUpdates.slice()

		n = lastUpdate?.n || 0
		sz = lastUpdate?.sz || 0

		while updatesRemaining.length and n < PackManager.MAX_COUNT and sz < PackManager.MAX_SIZE
			nextUpdate = updatesRemaining[0]
			nextUpdateSize = BSON.calculateObjectSize(nextUpdate)
			if nextUpdateSize + sz > PackManager.MAX_SIZE and n > 0
				break
			n++
			sz += nextUpdateSize
			updatesToFlush.push updatesRemaining.shift()

		PackManager.flushCompressedUpdates project_id, doc_id, lastUpdate, updatesToFlush, temporary, (error) ->
			return callback(error) if error?
			PackManager.insertCompressedUpdates project_id, doc_id, null, updatesRemaining, temporary, callback

	flushCompressedUpdates:	(project_id, doc_id, lastUpdate, newUpdates, temporary, callback = (error) ->) ->
		return callback() if newUpdates.length == 0
		if lastUpdate? and not (temporary and ((Date.now() - lastUpdate.meta?.start_ts) > 1 * DAYS))
			PackManager.appendUpdatesToExistingPack project_id, doc_id, lastUpdate, newUpdates, temporary, callback
		else
			PackManager.insertUpdatesIntoNewPack project_id, doc_id, newUpdates, temporary, callback

	insertUpdatesIntoNewPack: (project_id, doc_id, newUpdates, temporary, callback = (error) ->) ->
		first = newUpdates[0]
		last = newUpdates[newUpdates.length - 1]
		n = newUpdates.length
		sz = BSON.calculateObjectSize(newUpdates)
		newPack =
			project_id: ObjectId(project_id.toString())
			doc_id: ObjectId(doc_id.toString())
			pack: newUpdates
			n: n
			sz: sz
			meta:
				start_ts: first.meta.start_ts
				end_ts: last.meta.end_ts
			v: first.v
			v_end: last.v
		if temporary
			newPack.expiresAt = new Date(Date.now() + 7 * DAYS)
		logger.log {project_id, doc_id, newUpdates}, "inserting updates into new pack"
		db.docHistory.save newPack, (err, result) ->
			return callback(err) if err?
			if temporary
				return callback()
			else
				PackManager.updateIndex project_id, doc_id, callback

	appendUpdatesToExistingPack: (project_id, doc_id, lastUpdate, newUpdates, temporary, callback = (error) ->) ->
		first = newUpdates[0]
		last = newUpdates[newUpdates.length - 1]
		n = newUpdates.length
		sz = BSON.calculateObjectSize(newUpdates)
		query =
			_id: lastUpdate._id
			project_id: ObjectId(project_id.toString())
			doc_id: ObjectId(doc_id.toString())
			pack: {$exists: true}
		update =
			$push:
				"pack": {$each: newUpdates}
			$inc:
				"n": n
				"sz":  sz
			$set:
				"meta.end_ts": last.meta.end_ts
				"v_end": last.v
		if lastUpdate.expiresAt and temporary
			update.$set.expiresAt = new Date(Date.now() + 7 * DAYS)
		logger.log {project_id, doc_id, lastUpdate, newUpdates}, "appending updates to existing pack"
		db.docHistory.findAndModify {query, update,	new:true, fields:{meta:1,v_end:1}}, callback

	# Retrieve all changes for a document

	getOpsByVersionRange: (project_id, doc_id, fromVersion, toVersion, callback = (error, updates) ->) ->
		PackManager.loadPacksByVersionRange project_id, doc_id, fromVersion, toVersion, (error) ->
			query = {doc_id:ObjectId(doc_id.toString())}
			query.v = {$lte:toVersion} if toVersion?
			query.v_end = {$gte:fromVersion} if fromVersion?
			#console.log "query:", query
			db.docHistory.find(query).sort {v:-1}, (err, result) ->
				return callback(err) if err?
				#console.log "getOpsByVersionRange:", err, result
				updates = []
				opInRange = (op, from, to) ->
					return false if fromVersion? and op.v < fromVersion
					return false if toVersion? and op.v > toVersion
					return true
				for docHistory in result
					#console.log 'adding', docHistory.pack
					for op in docHistory.pack.reverse() when opInRange(op, fromVersion, toVersion)
						op.project_id = docHistory.project_id
						op.doc_id = docHistory.doc_id
						#console.log "added op", op.v, fromVersion, toVersion
						updates.push op
				callback(null, updates)

	loadPacksByVersionRange: (project_id, doc_id, fromVersion, toVersion, callback) ->
		PackManager.getIndex doc_id, (err, indexResult) ->
			return callback(err) if err?
			indexPacks = indexResult?.packs or []
			packInRange = (pack, from, to) ->
				return false if fromVersion? and pack.v_end < fromVersion
				return false if toVersion? and pack.v > toVersion
				return true
			neededIds = (pack._id for pack in indexPacks when packInRange(pack, fromVersion, toVersion))
			PackManager.fetchPacksIfNeeded project_id, doc_id, neededIds, callback

	fetchPacksIfNeeded: (project_id, doc_id, pack_ids, callback) ->
		db.docHistory.find {_id: {$in: (ObjectId(id) for id in pack_ids)}}, {_id:1}, (err, loadedPacks) ->
			return callback(err) if err?
			allPackIds = (id.toString() for id in pack_ids)
			loadedPackIds = (pack._id.toString() for pack in loadedPacks)
			packIdsToFetch = _.difference allPackIds, loadedPackIds
			logger.log {loadedPackIds, allPackIds, packIdsToFetch}, "analysed packs"
			async.eachLimit packIdsToFetch, 4, (pack_id, cb) ->
				MongoAWS.unArchivePack project_id, doc_id, pack_id, cb
			, (err) ->
				return callback(err) if err?
				logger.log "done unarchiving"
				callback()

	# Retrieve all changes across a project

	makeProjectIterator: (project_id, before, callback) ->
		# get all the docHistory Entries
		db.docHistory.find({project_id: ObjectId(project_id)},{pack:false}).sort {"meta.end_ts":-1}, (err, packs) ->
			return callback(err) if err?
			allPacks = []
			seenIds = {}
			for pack in packs
				allPacks.push pack
				seenIds[pack._id] = true
			db.docHistoryIndex.find {project_id: ObjectId(project_id)}, (err, indexes) ->
				return callback(err) if err?
				for index in indexes
					for pack in index.packs when not seenIds[pack._id]
						pack.project_id = index.project_id
						pack.doc_id = index._id
						pack.fromIndex = true
						allPacks.push pack
						seenIds[pack._id] = true
				callback(null, new ProjectIterator(allPacks, before, PackManager.getPackById))

	getPackById: (project_id, doc_id, pack_id, callback) ->
		db.docHistory.findOne {_id: pack_id}, (err, pack) ->
			return callback(err) if err?
			if not pack?
				MongoAWS.unArchivePack project_id, doc_id, pack_id, callback
			else if pack.expiresAt?
				# we only need to touch the TTL on the listing of changes in the project
				# because diffs on individual documents are always done after that
				PackManager.increaseTTL pack, callback
			else
				callback(null, pack)

	increaseTTL: (pack, callback) ->
		if pack.expiresAt < new Date(Date.now() + 6 * DAYS)
			# update cache expiry since we are using this pack
			db.docHistory.findAndModify {
				query: {_id: pack._id}
				update: {$set: {expiresAt: new Date(Date.now() + 7 * DAYS)}}
			}, (err) ->
				return callback(err, pack)
		else
			callback(null, pack)

	# Manage docHistoryIndex collection

	getIndex: (doc_id, callback) ->
		db.docHistoryIndex.findOne {_id:ObjectId(doc_id.toString())}, callback

	getPackFromIndex: (doc_id, pack_id, callback) ->
		db.docHistoryIndex.findOne {_id:ObjectId(doc_id.toString()), "packs._id": pack_id}, {"packs.$":1}, callback

	getLastPackFromIndex: (doc_id, callback) ->
		db.docHistoryIndex.findOne {_id: ObjectId(doc_id.toString())}, {packs:{$slice:-1}}, (err, indexPack) ->
			return callback(err) if err?
			return callback() if not indexPack?
			callback(null,indexPack[0])

	getIndexWithKeys: (doc_id, callback) ->
		PackManager.getIndex doc_id, (err, index) ->
			return callback(err) if err?
			return callback() if not index?
			for pack in index?.packs or []
				index[pack._id] = pack
			callback(null, index)

	initialiseIndex: (project_id, doc_id, callback) ->
		PackManager.findCompletedPacks project_id, doc_id, (err, packs) ->
			#console.log 'err', err, 'packs', packs, packs?.length
			return callback(err) if err?
			return callback() if not packs?
			PackManager.insertPacksIntoIndexWithLock project_id, doc_id, packs, callback

	updateIndex: (project_id, doc_id, callback) ->
		# find all packs prior to current pack
		PackManager.findUnindexedPacks project_id, doc_id, (err, newPacks) ->
			return callback(err) if err?
			return callback() if not newPacks?
			PackManager.insertPacksIntoIndexWithLock project_id, doc_id, newPacks, (err) ->
				return callback(err) if err?
				logger.log {project_id, doc_id, newPacks}, "added new packs to index"
				callback()

	findCompletedPacks: (project_id, doc_id, callback) ->
		query = { doc_id: ObjectId(doc_id.toString()) }
		db.docHistory.find(query, {pack:false}).sort {v:1}, (err, packs) ->
			return callback(err) if err?
			return callback() if not packs?
			return callback() if packs?.length <= 1
			packs.pop()  #  discard the last pack, it's still in progress
			callback(null, packs)

	# findPacks: (project_id, doc_id, queryFilter, callback) ->
	# 	query = { doc_id: ObjectId(doc_id.toString()) }
	# 	query = _.defaults query, queryFilter if queryFilter?
	# 	db.docHistory.find(query, {pack:false}).sort {v:1}, callback

	findUnindexedPacks: (project_id, doc_id, callback) ->
		PackManager.getIndexWithKeys doc_id, (err, indexResult) ->
			return callback(err) if err?
			PackManager.findCompletedPacks project_id, doc_id, (err, historyPacks) ->
				return callback(err) if err?
				return callback() if not historyPacks?
				# select only the new packs not already in the index
				newPacks = (pack for pack in historyPacks when not indexResult?[pack._id]?)
				newPacks = (_.omit(pack, 'doc_id', 'project_id', 'n', 'sz') for pack in newPacks)
				logger.log {project_id, doc_id, n: newPacks.length}, "found new packs"
				callback(null, newPacks)

	insertPacksIntoIndexWithLock: (project_id, doc_id, newPacks, callback) ->
		LockManager.runWithLock(
			"HistoryIndexLock:#{doc_id}",
			(releaseLock) ->
				PackManager._insertPacksIntoIndex project_id, doc_id, newPacks, releaseLock
			callback
		)

	_insertPacksIntoIndex: (project_id, doc_id, newPacks, callback) ->
		db.docHistoryIndex.findAndModify {
			query: {_id:ObjectId(doc_id.toString())}
			update:
				$setOnInsert: project_id: ObjectId(project_id.toString())
				$push:
					packs: {$each: newPacks, $sort: {v: 1}}
			upsert: true
		}, callback

	# Archiving packs to S3

	archivePack: (project_id, doc_id, pack_id, callback) ->
		clearFlagOnError = (err, cb) ->
			if err? # clear the inS3 flag on error
				PackManager.clearPackAsArchiveInProgress project_id, doc_id, pack_id, (err2) ->
					return cb(err2) if err2?
					return cb(err)
			else
				cb()
		async.series [
			(cb) ->
				PackManager.checkArchiveNotInProgress project_id, doc_id, pack_id, cb
			(cb) ->
				PackManager.markPackAsArchiveInProgress project_id, doc_id, pack_id, cb
			(cb) ->
				MongoAWS.archivePack project_id, doc_id, pack_id, (err) ->
					clearFlagOnError(err, cb)
			(cb) ->
				PackManager.checkArchivedPack project_id, doc_id, pack_id, (err) ->
					clearFlagOnError(err, cb)
			(cb) ->
				PackManager.markPackAsArchived project_id, doc_id, pack_id, cb
			(cb) ->
				PackManager.setTTLOnArchivedPack project_id, doc_id, pack_id, callback
		], callback


	checkArchivedPack: (project_id, doc_id, pack_id, callback) ->
		db.docHistory.findOne {_id: pack_id}, (err, pack) ->
			return callback(err) if err?
			return callback new Error("pack not found") if not pack?
			MongoAWS.readArchivedPack project_id, doc_id, pack_id, (err, result) ->
				delete result.last_checked
				delete pack.last_checked
				# need to compare ids as ObjectIds with .equals()
				for key in ['_id', 'project_id', 'doc_id']
					result[key] = pack[key] if result[key].equals(pack[key])
				for op, i in result.pack
					op._id = pack.pack[i]._id if op._id? and op._id.equals(pack.pack[i]._id)
				if _.isEqual pack, result
					callback()
				else
					logger.err {pack, result, jsondiff: JSON.stringify(pack) is JSON.stringify(result)}, "difference when comparing packs"
					callback new Error("pack retrieved from s3 does not match pack in mongo")

	# Processing old packs via worker

	processOldPack: (project_id, doc_id, pack_id, callback) ->
		markAsChecked = (err) ->
			PackManager.markPackAsChecked project_id, doc_id, pack_id, (err2) ->
				return callback(err2) if err2?
				callback(err)
		logger.log {project_id, doc_id}, "processing old packs"
		db.docHistory.findOne {_id:pack_id}, (err, pack) ->
			return markAsChecked(err) if err?
			return markAsChecked() if not pack?
			return callback() if pack.expiresAt? # return directly
			PackManager.updateIndexIfNeeded project_id, doc_id, (err) ->
				return markAsChecked(err) if err?
				PackManager.findUnarchivedPacks project_id, doc_id, (err, unarchivedPacks) ->
					return markAsChecked(err) if err?
					if not unarchivedPacks?.length
						logger.log "no packs need archiving"
						return markAsChecked()
					async.eachSeries unarchivedPacks, (pack, cb) ->
						PackManager.archivePack project_id, doc_id, pack._id, cb
					, (err) ->
						return markAsChecked(err) if err?
						logger.log "done processing"
						markAsChecked()

	updateIndexIfNeeded: (project_id, doc_id, callback) ->
		logger.log {project_id, doc_id}, "archiving old packs"
		PackManager.getIndexWithKeys doc_id, (err, index) ->
			return callback(err) if err?
			if not index?
				PackManager.initialiseIndex project_id, doc_id, callback
			else
				PackManager.updateIndex project_id, doc_id, callback

	markPackAsChecked: (project_id, doc_id, pack_id, callback) ->
		logger.log {project_id, doc_id, pack_id}, "marking pack as checked"
		db.docHistory.findAndModify {
			query: {_id: pack_id}
			update: {$currentDate: {"last_checked":true}}
		}, callback

	findUnarchivedPacks: (project_id, doc_id, callback) ->
		PackManager.getIndex doc_id, (err, indexResult) ->
			return callback(err) if err?
			indexPacks = indexResult?.packs or []
			unArchivedPacks = (pack for pack in indexPacks when not pack.inS3?)
			logger.log {project_id, doc_id, n: unArchivedPacks.length}, "find unarchived packs"
			callback(null, unArchivedPacks)

	# Archive locking flags

	checkArchiveNotInProgress: (project_id, doc_id, pack_id, callback) ->
		logger.log {project_id, doc_id, pack_id}, "checking if archive in progress"
		PackManager.getPackFromIndex doc_id, pack_id, (err, result) ->
			return callback(err) if err?
			return callback new Error("pack not found in index") if not result?
			if result.inS3?
				callback new Error("pack archiving already in progress")
			else
				callback()

	markPackAsArchiveInProgress: (project_id, doc_id, pack_id, callback) ->
		logger.log {project_id, doc_id}, "marking pack as archive in progress status"
		db.docHistoryIndex.findAndModify {
			query: {_id:ObjectId(doc_id.toString()),  packs: {$elemMatch: {"_id": pack_id, inS3: {$exists:false}}}}
			fields:  { "packs.$": 1 }
			update: {$set: {"packs.$.inS3":false}}
		}, (err, result) ->
			return callback(err) if err?
			return callback new Error("archive is already in progress") if not result?
			logger.log {project_id, doc_id, pack_id}, "marked as archive in progress"
			callback()

	clearPackAsArchiveInProgress: (project_id, doc_id, pack_id, callback) ->
		logger.log {project_id, doc_id, pack_id}, "clearing as archive in progress"
		db.docHistoryIndex.findAndModify {
			query: {_id:ObjectId(doc_id.toString()), "packs" : {$elemMatch: {"_id": pack_id, inS3: false}}}
			fields:  { "packs.$": 1 }
			update: {$unset: {"packs.$.inS3":true}}
		}, callback

	markPackAsArchived: (project_id, doc_id, pack_id, callback) ->
		logger.log {project_id, doc_id, pack_id}, "marking pack as archived"
		db.docHistoryIndex.findAndModify {
			query: {_id:ObjectId(doc_id.toString()), "packs" : {$elemMatch: {"_id": pack_id, inS3: false}}}
			fields: { "packs.$": 1 }
			update: {$set: {"packs.$.inS3":true}}
		}, (err, result) ->
			return callback(err) if err?
			return callback new Error("archive is not marked as progress") if not result?
			logger.log {project_id, doc_id, pack_id}, "marked as archived"
			callback()

	setTTLOnArchivedPack: (project_id, doc_id, pack_id, callback) ->
		db.docHistory.findAndModify {
			query: {_id: pack_id}
			update: {$set: {expiresAt: new Date(Date.now() + 1*DAYS)}}
		}, (err) ->
			logger.log {project_id, doc_id, pack_id}, "set expiry on pack"
			callback()