mirror of
https://github.com/overleaf/overleaf.git
synced 2024-10-31 21:21:03 -04:00
257 lines
11 KiB
CoffeeScript
257 lines
11 KiB
CoffeeScript
MongoManager = require "./MongoManager"
|
|
PackManager = require "./PackManager"
|
|
RedisManager = require "./RedisManager"
|
|
UpdateCompressor = require "./UpdateCompressor"
|
|
LockManager = require "./LockManager"
|
|
WebApiManager = require "./WebApiManager"
|
|
UpdateTrimmer = require "./UpdateTrimmer"
|
|
logger = require "logger-sharelatex"
|
|
async = require "async"
|
|
_ = require "underscore"
|
|
Settings = require "settings-sharelatex"
|
|
|
|
module.exports = UpdatesManager =
|
|
compressAndSaveRawUpdates: (project_id, doc_id, rawUpdates, temporary, callback = (error) ->) ->
|
|
length = rawUpdates.length
|
|
if length == 0
|
|
return callback()
|
|
|
|
# FIXME: we no longer need the lastCompressedUpdate, so change functions not to need it
|
|
# CORRECTION: we do use it to log the time in case of error
|
|
MongoManager.peekLastCompressedUpdate doc_id, (error, lastCompressedUpdate, lastVersion) ->
|
|
# lastCompressedUpdate is the most recent update in Mongo, and
|
|
# lastVersion is its sharejs version number.
|
|
#
|
|
# The peekLastCompressedUpdate method may pass the update back
|
|
# as 'null' (for example if the previous compressed update has
|
|
# been archived). In this case it can still pass back the
|
|
# lastVersion from the update to allow us to check consistency.
|
|
return callback(error) if error?
|
|
|
|
# Ensure that raw updates start where lastVersion left off
|
|
if lastVersion?
|
|
discardedUpdates = []
|
|
rawUpdates = rawUpdates.slice(0)
|
|
while rawUpdates[0]? and rawUpdates[0].v <= lastVersion
|
|
discardedUpdates.push rawUpdates.shift()
|
|
if discardedUpdates.length
|
|
logger.error project_id: project_id, doc_id: doc_id, discardedUpdates: discardedUpdates, temporary: temporary, lastVersion: lastVersion, "discarded updates already present"
|
|
|
|
if rawUpdates[0]? and rawUpdates[0].v != lastVersion + 1
|
|
ts = lastCompressedUpdate?.meta?.end_ts
|
|
last_timestamp = if ts? then new Date(ts) else 'unknown time'
|
|
error = new Error("Tried to apply raw op at version #{rawUpdates[0].v} to last compressed update with version #{lastVersion} from #{last_timestamp}")
|
|
logger.error err: error, doc_id: doc_id, project_id: project_id, prev_end_ts: ts, temporary: temporary, lastCompressedUpdate: lastCompressedUpdate, "inconsistent doc versions"
|
|
if Settings.trackchanges?.continueOnError and rawUpdates[0].v > lastVersion + 1
|
|
# we have lost some ops - continue to write into the database, we can't recover at this point
|
|
lastCompressedUpdate = null
|
|
else
|
|
return callback error
|
|
|
|
if rawUpdates.length == 0
|
|
return callback()
|
|
|
|
# some old large ops in redis need to be rejected, they predate
|
|
# the size limit that now prevents them going through the system
|
|
REJECT_LARGE_OP_SIZE = 4 * 1024 * 1024
|
|
for rawUpdate in rawUpdates
|
|
opSizes = ((op.i?.length || op.d?.length) for op in rawUpdate?.op or [])
|
|
size = _.max opSizes
|
|
if size > REJECT_LARGE_OP_SIZE
|
|
error = new Error("dropped op exceeding maximum allowed size of #{REJECT_LARGE_OP_SIZE}")
|
|
logger.error err: error, doc_id: doc_id, project_id: project_id, size: size, rawUpdate: rawUpdate, "dropped op - too big"
|
|
rawUpdate.op = []
|
|
|
|
compressedUpdates = UpdateCompressor.compressRawUpdates null, rawUpdates
|
|
PackManager.insertCompressedUpdates project_id, doc_id, lastCompressedUpdate, compressedUpdates, temporary, (error, result) ->
|
|
return callback(error) if error?
|
|
logger.log {project_id, doc_id, orig_v: lastCompressedUpdate?.v, new_v: result.v}, "inserted updates into pack" if result?
|
|
callback()
|
|
|
|
REDIS_READ_BATCH_SIZE: 100
|
|
processUncompressedUpdates: (project_id, doc_id, callback = (error) ->) ->
|
|
UpdateTrimmer.shouldTrimUpdates project_id, (error, temporary) ->
|
|
return callback(error) if error?
|
|
MongoManager.backportProjectId project_id, doc_id, (error) ->
|
|
return callback(error) if error?
|
|
# get the updates as strings from redis (so we can delete them after they are applied)
|
|
RedisManager.getOldestDocUpdates doc_id, UpdatesManager.REDIS_READ_BATCH_SIZE, (error, docUpdates) ->
|
|
return callback(error) if error?
|
|
length = docUpdates.length
|
|
# parse the redis strings into ShareJs updates
|
|
RedisManager.expandDocUpdates docUpdates, (error, rawUpdates) ->
|
|
return callback(error) if error?
|
|
logger.log project_id: project_id, doc_id: doc_id, rawUpdates: rawUpdates, "retrieved raw updates from redis"
|
|
UpdatesManager.compressAndSaveRawUpdates project_id, doc_id, rawUpdates, temporary, (error) ->
|
|
return callback(error) if error?
|
|
logger.log project_id: project_id, doc_id: doc_id, "compressed and saved doc updates"
|
|
# delete the applied updates from redis
|
|
RedisManager.deleteAppliedDocUpdates project_id, doc_id, docUpdates, (error) ->
|
|
return callback(error) if error?
|
|
if length == UpdatesManager.REDIS_READ_BATCH_SIZE
|
|
# There might be more updates
|
|
logger.log project_id: project_id, doc_id: doc_id, "continuing processing updates"
|
|
setTimeout () ->
|
|
UpdatesManager.processUncompressedUpdates project_id, doc_id, callback
|
|
, 0
|
|
else
|
|
logger.log project_id: project_id, doc_id: doc_id, "all raw updates processed"
|
|
callback()
|
|
|
|
processUncompressedUpdatesWithLock: (project_id, doc_id, callback = (error) ->) ->
|
|
LockManager.runWithLock(
|
|
"HistoryLock:#{doc_id}",
|
|
(releaseLock) ->
|
|
UpdatesManager.processUncompressedUpdates project_id, doc_id, releaseLock
|
|
callback
|
|
)
|
|
|
|
processUncompressedUpdatesForProject: (project_id, callback = (error) ->) ->
|
|
RedisManager.getDocIdsWithHistoryOps project_id, (error, doc_ids) ->
|
|
return callback(error) if error?
|
|
jobs = []
|
|
for doc_id in doc_ids
|
|
do (doc_id) ->
|
|
jobs.push (callback) ->
|
|
UpdatesManager.processUncompressedUpdatesWithLock project_id, doc_id, callback
|
|
async.parallelLimit jobs, 5, callback
|
|
|
|
getDocUpdates: (project_id, doc_id, options = {}, callback = (error, updates) ->) ->
|
|
UpdatesManager.processUncompressedUpdatesWithLock project_id, doc_id, (error) ->
|
|
return callback(error) if error?
|
|
#console.log "options", options
|
|
PackManager.getOpsByVersionRange project_id, doc_id, options.from, options.to, (error, updates) ->
|
|
return callback(error) if error?
|
|
callback null, updates
|
|
|
|
getDocUpdatesWithUserInfo: (project_id, doc_id, options = {}, callback = (error, updates) ->) ->
|
|
UpdatesManager.getDocUpdates project_id, doc_id, options, (error, updates) ->
|
|
return callback(error) if error?
|
|
UpdatesManager.fillUserInfo updates, (error, updates) ->
|
|
return callback(error) if error?
|
|
callback null, updates
|
|
|
|
getSummarizedProjectUpdates: (project_id, options = {}, callback = (error, updates) ->) ->
|
|
options.min_count ||= 25
|
|
summarizedUpdates = []
|
|
before = options.before
|
|
nextBeforeTimestamp = null
|
|
UpdatesManager.processUncompressedUpdatesForProject project_id, (error) ->
|
|
return callback(error) if error?
|
|
PackManager.makeProjectIterator project_id, before, (err, iterator) ->
|
|
return callback(err) if err?
|
|
# repeatedly get updates and pass them through the summariser to get an final output with user info
|
|
async.whilst () ->
|
|
#console.log "checking iterator.done", iterator.done()
|
|
return summarizedUpdates.length < options.min_count and not iterator.done()
|
|
, (cb) ->
|
|
iterator.next (err, partialUpdates) ->
|
|
return callback(err) if err?
|
|
#logger.log {partialUpdates}, 'got partialUpdates'
|
|
return cb() if partialUpdates.length is 0 ## FIXME should try to avoid this happening
|
|
nextBeforeTimestamp = partialUpdates[partialUpdates.length - 1].meta.end_ts
|
|
# add the updates to the summary list
|
|
summarizedUpdates = UpdatesManager._summarizeUpdates partialUpdates, summarizedUpdates
|
|
cb()
|
|
, () ->
|
|
# finally done all updates
|
|
#console.log 'summarized Updates', summarizedUpdates
|
|
UpdatesManager.fillSummarizedUserInfo summarizedUpdates, (err, results) ->
|
|
return callback(err) if err?
|
|
callback null, results, if not iterator.done() then nextBeforeTimestamp else undefined
|
|
|
|
fetchUserInfo: (users, callback = (error, fetchedUserInfo) ->) ->
|
|
jobs = []
|
|
fetchedUserInfo = {}
|
|
for user_id of users
|
|
do (user_id) ->
|
|
jobs.push (callback) ->
|
|
WebApiManager.getUserInfo user_id, (error, userInfo) ->
|
|
return callback(error) if error?
|
|
fetchedUserInfo[user_id] = userInfo
|
|
callback()
|
|
|
|
async.series jobs, (err) ->
|
|
return callback(err) if err?
|
|
callback(null, fetchedUserInfo)
|
|
|
|
fillUserInfo: (updates, callback = (error, updates) ->) ->
|
|
users = {}
|
|
for update in updates
|
|
user_id = update.meta.user_id
|
|
if UpdatesManager._validUserId(user_id)
|
|
users[user_id] = true
|
|
|
|
UpdatesManager.fetchUserInfo users, (error, fetchedUserInfo) ->
|
|
return callback(error) if error?
|
|
for update in updates
|
|
user_id = update.meta.user_id
|
|
delete update.meta.user_id
|
|
if UpdatesManager._validUserId(user_id)
|
|
update.meta.user = fetchedUserInfo[user_id]
|
|
callback null, updates
|
|
|
|
fillSummarizedUserInfo: (updates, callback = (error, updates) ->) ->
|
|
users = {}
|
|
for update in updates
|
|
user_ids = update.meta.user_ids or []
|
|
for user_id in user_ids
|
|
if UpdatesManager._validUserId(user_id)
|
|
users[user_id] = true
|
|
|
|
UpdatesManager.fetchUserInfo users, (error, fetchedUserInfo) ->
|
|
return callback(error) if error?
|
|
for update in updates
|
|
user_ids = update.meta.user_ids or []
|
|
update.meta.users = []
|
|
delete update.meta.user_ids
|
|
for user_id in user_ids
|
|
if UpdatesManager._validUserId(user_id)
|
|
update.meta.users.push fetchedUserInfo[user_id]
|
|
else
|
|
update.meta.users.push null
|
|
callback null, updates
|
|
|
|
_validUserId: (user_id) ->
|
|
if !user_id?
|
|
return false
|
|
else
|
|
return !!user_id.match(/^[a-f0-9]{24}$/)
|
|
|
|
TIME_BETWEEN_DISTINCT_UPDATES: fiveMinutes = 5 * 60 * 1000
|
|
_summarizeUpdates: (updates, existingSummarizedUpdates = []) ->
|
|
summarizedUpdates = existingSummarizedUpdates.slice()
|
|
for update in updates
|
|
earliestUpdate = summarizedUpdates[summarizedUpdates.length - 1]
|
|
if earliestUpdate and earliestUpdate.meta.start_ts - update.meta.end_ts < @TIME_BETWEEN_DISTINCT_UPDATES
|
|
# check if the user in this update is already present in the earliest update,
|
|
# if not, add them to the users list of the earliest update
|
|
earliestUpdate.meta.user_ids = _.union earliestUpdate.meta.user_ids, [update.meta.user_id]
|
|
|
|
doc_id = update.doc_id.toString()
|
|
doc = earliestUpdate.docs[doc_id]
|
|
if doc?
|
|
doc.fromV = Math.min(doc.fromV, update.v)
|
|
doc.toV = Math.max(doc.toV, update.v)
|
|
else
|
|
earliestUpdate.docs[doc_id] =
|
|
fromV: update.v
|
|
toV: update.v
|
|
|
|
earliestUpdate.meta.start_ts = Math.min(earliestUpdate.meta.start_ts, update.meta.start_ts)
|
|
earliestUpdate.meta.end_ts = Math.max(earliestUpdate.meta.end_ts, update.meta.end_ts)
|
|
else
|
|
newUpdate =
|
|
meta:
|
|
user_ids: []
|
|
start_ts: update.meta.start_ts
|
|
end_ts: update.meta.end_ts
|
|
docs: {}
|
|
|
|
newUpdate.docs[update.doc_id.toString()] =
|
|
fromV: update.v
|
|
toV: update.v
|
|
newUpdate.meta.user_ids.push update.meta.user_id
|
|
summarizedUpdates.push newUpdate
|
|
|
|
return summarizedUpdates
|