UpdatesManager.coffee 4.0 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107
  1. MongoManager = require "./MongoManager"
  2. RedisManager = require "./RedisManager"
  3. UpdateCompressor = require "./UpdateCompressor"
  4. LockManager = require "./LockManager"
  5. WebApiManager = require "./WebApiManager"
  6. logger = require "logger-sharelatex"
  7. async = require "async"
  8. module.exports = UpdatesManager =
  9. compressAndSaveRawUpdates: (doc_id, rawUpdates, callback = (error) ->) ->
  10. length = rawUpdates.length
  11. if length == 0
  12. return callback()
  13. MongoManager.popLastCompressedUpdate doc_id, (error, lastCompressedUpdate) ->
  14. return callback(error) if error?
  15. # Ensure that raw updates start where lastCompressedUpdate left off
  16. if lastCompressedUpdate?
  17. rawUpdates = rawUpdates.slice(0)
  18. while rawUpdates[0]? and rawUpdates[0].v <= lastCompressedUpdate.v
  19. rawUpdates.shift()
  20. if rawUpdates[0]? and rawUpdates[0].v != lastCompressedUpdate.v + 1
  21. error = new Error("Tried to apply raw op at version #{rawUpdates[0].v} to last compressed update with version #{lastCompressedUpdate.v}")
  22. logger.error err: error, doc_id: doc_id, "inconsistent doc versions"
  23. # Push the update back into Mongo - catching errors at this
  24. # point is useless, we're already bailing
  25. MongoManager.insertCompressedUpdates doc_id, [lastCompressedUpdate], () ->
  26. return callback error
  27. return
  28. compressedUpdates = UpdateCompressor.compressRawUpdates lastCompressedUpdate, rawUpdates
  29. MongoManager.insertCompressedUpdates doc_id, compressedUpdates, (error) ->
  30. return callback(error) if error?
  31. logger.log doc_id: doc_id, rawUpdatesLength: length, compressedUpdatesLength: compressedUpdates.length, "compressed doc updates"
  32. callback()
  33. REDIS_READ_BATCH_SIZE: 100
  34. processUncompressedUpdates: (doc_id, callback = (error) ->) ->
  35. RedisManager.getOldestRawUpdates doc_id, UpdatesManager.REDIS_READ_BATCH_SIZE, (error, rawUpdates) ->
  36. return callback(error) if error?
  37. length = rawUpdates.length
  38. UpdatesManager.compressAndSaveRawUpdates doc_id, rawUpdates, (error) ->
  39. return callback(error) if error?
  40. logger.log doc_id: doc_id, "compressed and saved doc updates"
  41. RedisManager.deleteOldestRawUpdates doc_id, length, (error) ->
  42. return callback(error) if error?
  43. if length == UpdatesManager.REDIS_READ_BATCH_SIZE
  44. # There might be more updates
  45. logger.log doc_id: doc_id, "continuing processing updates"
  46. setTimeout () ->
  47. UpdatesManager.processUncompressedUpdates doc_id, callback
  48. , 0
  49. else
  50. logger.log doc_id: doc_id, "all raw updates processed"
  51. callback()
  52. processUncompressedUpdatesWithLock: (doc_id, callback = (error) ->) ->
  53. LockManager.runWithLock(
  54. "HistoryLock:#{doc_id}",
  55. (releaseLock) ->
  56. UpdatesManager.processUncompressedUpdates doc_id, releaseLock
  57. callback
  58. )
  59. getUpdates: (doc_id, options = {}, callback = (error, updates) ->) ->
  60. UpdatesManager.processUncompressedUpdatesWithLock doc_id, (error) ->
  61. return callback(error) if error?
  62. MongoManager.getUpdates doc_id, options, callback
  63. getUpdatesWithUserInfo: (doc_id, options = {}, callback = (error, updates) ->) ->
  64. UpdatesManager.getUpdates doc_id, options, (error, updates) ->
  65. return callback(error) if error?
  66. UpdatesManager.fillUserInfo updates, (error, updates) ->
  67. return callback(error) if error?
  68. callback null, updates
  69. fillUserInfo: (updates, callback = (error, updates) ->) ->
  70. users = {}
  71. for update in updates
  72. if UpdatesManager._validUserId(update.meta.user_id)
  73. users[update.meta.user_id] = true
  74. jobs = []
  75. for user_id, _ of users
  76. do (user_id) ->
  77. jobs.push (callback) ->
  78. WebApiManager.getUserInfo user_id, (error, userInfo) ->
  79. return callback(error) if error?
  80. users[user_id] = userInfo
  81. callback()
  82. async.series jobs, (error) ->
  83. return callback(error) if error?
  84. for update in updates
  85. user_id = update.meta.user_id
  86. delete update.meta.user_id
  87. if UpdatesManager._validUserId(user_id)
  88. update.meta.user = users[user_id]
  89. callback null, updates
  90. _validUserId: (user_id) ->
  91. if !user_id?
  92. return false
  93. else
  94. return !!user_id.match(/^[a-f0-9]{24}$/)