Overleaf/server-ce/migrations/3_pack_docHistory_collection.coffee

Settings = require "settings-sharelatex"
fs = require("fs")
mongojs = require("mongojs")
ObjectId = mongojs.ObjectId
db = mongojs(Settings.mongo.url, ['docs','docHistory', 'docHistoryStats'])
_ = require("underscore")
async = require("async")
exec = require("child_process").exec
BSON = db.bson.BSON

logger = {
	log: ->
	err: ->
}

needToExit = false
handleExit = () ->
	needToExit = true
	console.log('Got signal.  Shutting down.')

process.on 'SIGINT', handleExit
process.on 'SIGHUP', handleExit

finished_docs_path = "/tmp/finished-docs-3"
all_docs_path = "/tmp/all-docs-3"
unmigrated_docs_path = "/tmp/unmigrated-docs-3"

finished_docs = {}
if fs.existsSync(finished_docs_path)
	for id in fs.readFileSync(finished_docs_path,'utf-8').split("\n")
		finished_docs[id] = true

getAndWriteDocids = (callback)->
	console.log "finding all doc id's - #{new Date().toString()}"
	db.docs.find {}, {_id:1}, (err, ids)->
		console.log "total found docs in mongo #{ids.length} - #{new Date().toString()}"
		ids = _.pluck ids, '_id'
		ids = _.filter ids, (id)-> id?
		fileData = ids.join("\n")
		fs.writeFileSync all_docs_path + ".tmp", fileData
		fs.renameSync all_docs_path + ".tmp", all_docs_path
		callback(err, ids)

loadDocIds = (callback)->
	console.log "loading doc ids from #{all_docs_path}"
	data = fs.readFileSync all_docs_path, "utf-8"
	ids = data.split("\n")
	console.log "loaded #{ids.length} doc ids from #{all_docs_path}"
	callback null, ids

getDocIds = (callback)->
	exists = fs.existsSync all_docs_path
	if exists
		loadDocIds callback
	else
		getAndWriteDocids callback

markDocAsProcessed = (doc_id, callback)->
	finished_docs[doc_id] = true
	fs.appendFile finished_docs_path, "#{doc_id}\n", callback

markDocAsUnmigrated = (doc_id, callback)->
	console.log "#{doc_id} unmigrated"
	markDocAsProcessed doc_id, (err)->
		fs.appendFile unmigrated_docs_path, "#{doc_id}\n", callback

checkIfDocHasBeenProccessed = (doc_id, callback)->
	callback(null, finished_docs[doc_id])

processNext = (doc_id, callback)->
	if !doc_id? or doc_id.length == 0
		return callback()
	if needToExit
		return callback(new Error("graceful shutdown"))
	checkIfDocHasBeenProccessed doc_id, (err, hasBeenProcessed)->
		if hasBeenProcessed
			console.log "#{doc_id} already processed, skipping"
			return callback()
		PackManager._packDocHistory doc_id, {}, (err) ->
			if err?
				console.log "error processing #{doc_id}"
				markDocAsUnmigrated doc_id, callback
			else
				markDocAsProcessed doc_id, callback

updateIndexes = (callback) ->
	async.series [
		(cb) ->
			console.log "create index"
			db.docHistory.ensureIndex { project_id: 1, "meta.end_ts": 1, "meta.start_ts": -1 }, { background: true }, cb
		(cb) ->
			console.log "drop index"
			db.docHistory.dropIndex { project_id: 1, "meta.end_ts": 1 }, cb
		(cb) ->
			console.log "drop index"
			db.docHistory.dropIndex { project_id: 1, "pack.0.meta.end_ts": 1, "meta.end_ts": 1}, cb
	], (err, results) ->
		console.log "all done"
		callback(err)

exports.migrate = (client, done = ->)->
	getDocIds (err, ids)->
		totalDocCount = ids.length
		alreadyFinishedCount = Object.keys(finished_docs).length
		t0 = Date.now()
		printProgress = () ->
			count = Object.keys(finished_docs).length
			processedFraction = (count-alreadyFinishedCount)/totalDocCount
			remainingFraction = (totalDocCount-count)/totalDocCount
			t = Date.now()
			dt = (t-t0)*remainingFraction/processedFraction
			estFinishTime = new Date(t + dt)
			console.log "completed #{count}/#{totalDocCount} processed=#{processedFraction.toFixed(2)} remaining=#{remainingFraction.toFixed(2)} elapsed=#{(t-t0)/1000} est Finish=#{estFinishTime}"
		interval = setInterval printProgress, 3*1000

		nextId = null

		testFn = () ->
			return false if needToExit
			id = ids.shift()
			while id? and finished_docs[id]	# skip finished
				id = ids.shift()
			nextId = id
			return nextId?

		executeFn = (cb) ->
			processNext nextId, cb

		async.whilst testFn, executeFn, (err)->
			if err?
				console.error err, "at end of jobs"
			else
				console.log "finished at #{new Date}"
			clearInterval interval
			done(err)

exports.rollback = (client, done)->
	done()

# process.nextTick () ->
# 	exports.migrate () ->
# 		console.log "done"

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

# copied from track-changes/app/coffee/PackManager.coffee

PackManager =
	MAX_SIZE:  1024*1024 # make these configurable parameters
	MAX_COUNT: 512

	convertDocsToPacks: (docs, callback) ->
		packs = []
		top = null
		docs.forEach (d,i) ->
			# skip existing packs
			if d.pack?
				top = null
				return
			sz = BSON.calculateObjectSize(d)
			# decide if this doc can be added to the current pack
			validLength = top? && (top.pack.length < PackManager.MAX_COUNT)
			validSize = top? && (top.sz + sz < PackManager.MAX_SIZE)
			bothPermanent = top? && (top.expiresAt? is false) && (d.expiresAt? is false)
			bothTemporary = top? && (top.expiresAt? is true) && (d.expiresAt? is true)
			within1Day = bothTemporary && (d.meta.start_ts - top.meta.start_ts < 24 * 3600 * 1000)
			if top? && validLength && validSize && (bothPermanent || (bothTemporary && within1Day))
				top.pack = top.pack.concat {v: d.v, meta: d.meta,  op: d.op, _id: d._id}
				top.sz += sz
				top.n += 1
				top.v_end = d.v
				top.meta.end_ts = d.meta.end_ts
				top.expiresAt = d.expiresAt if top.expiresAt?
				return
			else
				# create a new pack
				top = _.clone(d)
				top.pack = [ {v: d.v, meta: d.meta,  op: d.op, _id: d._id} ]
				top.meta = { start_ts: d.meta.start_ts, end_ts: d.meta.end_ts }
				top.sz = sz
				top.n = 1
				top.v_end = d.v
				delete top.op
				delete top._id
				packs.push top

		callback(null, packs)

	checkHistory: (docs, callback) ->
		errors = []
		prev = null
		error = (args...) ->
			errors.push args
		docs.forEach (d,i) ->
			if d.pack?
				n = d.pack.length
				last = d.pack[n-1]
				error('bad pack v_end', d) if d.v_end != last.v
				error('bad pack start_ts', d) if d.meta.start_ts != d.pack[0].meta.start_ts
				error('bad pack end_ts', d) if d.meta.end_ts != last.meta.end_ts
				d.pack.forEach (p, i) ->
					prev = v
					v = p.v
					error('bad version', v, 'in', p) if v <= prev
					#error('expired op', p, 'in pack') if p.expiresAt?
			else
				prev = v
				v = d.v
				error('bad version', v, 'in', d) if v <= prev
		if errors.length
			callback(errors)
		else
			callback()

	insertPack: (packObj, callback) ->
		bulk = db.docHistory.initializeOrderedBulkOp()
		doc_id = packObj.doc_id
		expect_nInserted = 1
		expect_nRemoved = packObj.pack.length
		logger.log {doc_id: doc_id}, "adding pack, removing #{expect_nRemoved} ops"
		bulk.insert packObj
		ids = (op._id for op in packObj.pack)
		bulk.find({_id:{$in:ids}}).remove()
		bulk.execute (err, result) ->
			if err?
				logger.error {doc_id: doc_id}, "error adding pack"
				callback(err, result)
			else if result.nInserted != expect_nInserted or result.nRemoved != expect_nRemoved
				logger.error {doc_id: doc_id, result}, "unexpected result adding pack"
				callback(new Error(
					msg: 'unexpected result'
					expected: {expect_nInserted, expect_nRemoved}
				), result)
			else
				db.docHistoryStats.update {doc_id:doc_id}, {
					$inc:{update_count:-expect_nRemoved},
					$currentDate:{last_packed:true}
				}, {upsert:true}, () ->
					callback(err, result)

	# retrieve document ops/packs and check them
	getDocHistory: (doc_id, callback) ->
		db.docHistory.find({doc_id:ObjectId(doc_id)}).sort {v:1}, (err, docs) ->
			return callback(err) if err?
			# for safety, do a consistency check of the history
			logger.log {doc_id}, "checking history for document"
			PackManager.checkHistory docs, (err) ->
				return callback(err) if err?
				callback(err, docs)
				#PackManager.deleteExpiredPackOps docs, (err) ->
				#	return callback(err) if err?
				#	callback err, docs

	packDocHistory: (doc_id, options, callback) ->
		if typeof callback == "undefined" and typeof options == 'function'
			callback = options
			options = {}
		LockManager.runWithLock(
			"HistoryLock:#{doc_id}",
			(releaseLock) ->
				PackManager._packDocHistory(doc_id, options, releaseLock)
			,	callback
		)

	_packDocHistory: (doc_id, options, callback) ->
		logger.log {doc_id},"starting pack operation for document history"

		PackManager.getDocHistory doc_id, (err, docs) ->
			return callback(err) if err?
			origDocs = 0
			origPacks = 0
			for d in docs
				if d.pack? then	origPacks++	else origDocs++
			PackManager.convertDocsToPacks docs, (err, packs) ->
				return callback(err) if err?
				total = 0
				for p in packs
					total = total + p.pack.length
				logger.log {doc_id, origDocs, origPacks, newPacks: packs.length, totalOps: total}, "document stats"
				if packs.length
					if options['dry-run']
						logger.log {doc_id}, 'dry-run, skipping write packs'
						return callback()
					PackManager.savePacks packs, (err) ->
						return callback(err) if err?
						# check the history again
						PackManager.getDocHistory doc_id, callback
				else
					logger.log {doc_id}, "no packs to write"
					# keep a record that we checked this one to avoid rechecking it
					db.docHistoryStats.update {doc_id:doc_id}, {
						$currentDate:{last_checked:true}
					}, {upsert:true}, () ->
						callback null, null

	DB_WRITE_DELAY: 100

	savePacks: (packs, callback) ->
		async.eachSeries packs, PackManager.safeInsert, (err, result) ->
			if err?
				logger.log {err, result}, "error writing packs"
				callback err, result
			else
				callback()

	safeInsert: (packObj, callback) ->
		PackManager.insertPack packObj, (err, result) ->
			setTimeout () ->
				callback(err,result)
			, PackManager.DB_WRITE_DELAY

	deleteExpiredPackOps: (docs, callback) ->
		now = Date.now()
		toRemove = []
		toUpdate = []
		docs.forEach (d,i) ->
			if d.pack?
				newPack = d.pack.filter (op) ->
					if op.expiresAt? then op.expiresAt > now else true
				if newPack.length == 0
					toRemove.push d
				else if newPack.length < d.pack.length
					# adjust the pack properties
					d.pack = newPack
					first = d.pack[0]
					last = d.pack[d.pack.length - 1]
					d.v_end = last.v
					d.meta.start_ts = first.meta.start_ts
					d.meta.end_ts = last.meta.end_ts
					toUpdate.push d
		if toRemove.length or toUpdate.length
			bulk = db.docHistory.initializeOrderedBulkOp()
			toRemove.forEach (pack) ->
				console.log "would remove", pack
				#bulk.find({_id:pack._id}).removeOne()
			toUpdate.forEach (pack) ->
				console.log "would update", pack
				#bulk.find({_id:pack._id}).updateOne(pack);
			bulk.execute callback
		else
			callback()