MongoManager.coffee 6.8 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175
  1. {db, ObjectId} = require "./mongojs"
  2. PackManager = require "./PackManager"
  3. async = require "async"
  4. _ = require "underscore"
  5. module.exports = MongoManager =
  6. getLastCompressedUpdate: (doc_id, callback = (error, update) ->) ->
  7. db.docHistory
  8. .find(doc_id: ObjectId(doc_id.toString()), {pack: {$slice:-1}}) # only return the last entry in a pack
  9. .sort( v: -1 )
  10. .limit(1)
  11. .toArray (error, compressedUpdates) ->
  12. return callback(error) if error?
  13. callback null, compressedUpdates[0] or null
  14. peekLastCompressedUpdate: (doc_id, callback = (error, update, version) ->) ->
  15. # under normal use we pass back the last update as
  16. # callback(null,update,version).
  17. #
  18. # when we have an existing last update but want to force a new one
  19. # to start, we pass it back as callback(null,null,version), just
  20. # giving the version so we can check consistency.
  21. MongoManager.getLastCompressedUpdate doc_id, (error, update) ->
  22. return callback(error) if error?
  23. if update?
  24. if update.broken
  25. # the update is marked as broken so we will force a new op
  26. return callback null, null
  27. else if update.pack?
  28. return callback null, update, update.pack[0]?.v
  29. else
  30. return callback null, update, update.v
  31. else
  32. MongoManager.getArchivedDocStatus doc_id, (error, status) ->
  33. return callback(error) if error?
  34. return callback(null, null, status.lastVersion) if status?.inS3? and status?.lastVersion?
  35. callback null, null
  36. insertCompressedUpdates: (project_id, doc_id, updates, temporary, callback = (error) ->) ->
  37. jobs = []
  38. for update in updates
  39. do (update) ->
  40. jobs.push (callback) -> MongoManager.insertCompressedUpdate project_id, doc_id, update, temporary, callback
  41. async.series jobs, (err, results) ->
  42. if not temporary
  43. # keep track of updates to be packed
  44. db.docHistoryStats.update {doc_id:ObjectId(doc_id)}, {
  45. $inc:{update_count:updates.length},
  46. $currentDate:{last_update:true}
  47. }, {upsert:true}, () ->
  48. callback(err,results)
  49. else
  50. callback(err,results)
  51. modifyCompressedUpdate: (lastUpdate, newUpdate, callback = (error) ->) ->
  52. return callback() if not newUpdate?
  53. db.docHistory.findAndModify
  54. query: lastUpdate,
  55. update:
  56. $set :
  57. op: newUpdate.op
  58. meta: newUpdate.meta
  59. v: newUpdate.v
  60. new: true
  61. , (err, result, lastErrorObject) ->
  62. return callback(error) if error?
  63. return new Error("could not modify existing op") if not result?
  64. callback(err, result)
  65. insertCompressedUpdate: (project_id, doc_id, update, temporary, callback = (error) ->) ->
  66. update = {
  67. doc_id: ObjectId(doc_id.toString())
  68. project_id: ObjectId(project_id.toString())
  69. op: update.op
  70. meta: update.meta
  71. v: update.v
  72. }
  73. if temporary
  74. seconds = 1000
  75. minutes = 60 * seconds
  76. hours = 60 * minutes
  77. days = 24 * hours
  78. update.expiresAt = new Date(Date.now() + 7 * days)
  79. # may need to roll over a pack here if we are inserting packs
  80. db.docHistory.insert update, callback
  81. getDocUpdates:(doc_id, options = {}, callback = (error, updates) ->) ->
  82. query =
  83. doc_id: ObjectId(doc_id.toString())
  84. if options.from?
  85. query["v"] ||= {}
  86. query["v"]["$gte"] = options.from
  87. if options.to?
  88. query["v"] ||= {}
  89. query["v"]["$lte"] = options.to
  90. PackManager.findDocResults(db.docHistory, query, options.limit, callback)
  91. getProjectUpdates: (project_id, options = {}, callback = (error, updates) ->) ->
  92. query =
  93. project_id: ObjectId(project_id.toString())
  94. if options.before?
  95. query["meta.end_ts"] = { $lt: options.before }
  96. PackManager.findProjectResults(db.docHistory, query, options.limit, callback)
  97. backportProjectId: (project_id, doc_id, callback = (error) ->) ->
  98. db.docHistory.update {
  99. doc_id: ObjectId(doc_id.toString())
  100. project_id: { $exists: false }
  101. }, {
  102. $set: { project_id: ObjectId(project_id.toString()) }
  103. }, {
  104. multi: true
  105. }, callback
  106. getProjectMetaData: (project_id, callback = (error, metadata) ->) ->
  107. db.projectHistoryMetaData.find {
  108. project_id: ObjectId(project_id.toString())
  109. }, (error, results) ->
  110. return callback(error) if error?
  111. callback null, results[0]
  112. setProjectMetaData: (project_id, metadata, callback = (error) ->) ->
  113. db.projectHistoryMetaData.update {
  114. project_id: ObjectId(project_id)
  115. }, {
  116. $set: metadata
  117. }, {
  118. upsert: true
  119. }, callback
  120. ensureIndices: () ->
  121. # For finding all updates that go into a diff for a doc
  122. db.docHistory.ensureIndex { doc_id: 1, v: 1 }, { background: true }
  123. # For finding all updates that affect a project
  124. db.docHistory.ensureIndex { project_id: 1, "meta.end_ts": 1 }, { background: true }
  125. # For finding all packs that affect a project (use a sparse index so only packs are included)
  126. db.docHistory.ensureIndex { project_id: 1, "pack.0.meta.end_ts": 1, "meta.end_ts": 1}, { background: true, sparse: true }
  127. # For finding updates that don't yet have a project_id and need it inserting
  128. db.docHistory.ensureIndex { doc_id: 1, project_id: 1 }, { background: true }
  129. # For finding project meta-data
  130. db.projectHistoryMetaData.ensureIndex { project_id: 1 }, { background: true }
  131. # TTL index for auto deleting week old temporary ops
  132. db.docHistory.ensureIndex { expiresAt: 1 }, { expireAfterSeconds: 0, background: true }
  133. # For finding documents which need packing
  134. db.docHistoryStats.ensureIndex { doc_id: 1 }, { background: true }
  135. db.docHistoryStats.ensureIndex { updates: -1, doc_id: 1 }, { background: true }
  136. getArchivedDocStatus: (doc_id, callback)->
  137. db.docHistoryStats.findOne {doc_id: ObjectId(doc_id.toString()), inS3: {$exists:true}}, {inS3: true, lastVersion: true}, callback
  138. getDocChangesCount: (doc_id, callback)->
  139. db.docHistory.count { doc_id : ObjectId(doc_id.toString())}, callback
  140. markDocHistoryAsArchiveInProgress: (doc_id, lastVersion, callback) ->
  141. db.docHistoryStats.update {doc_id: ObjectId(doc_id.toString())}, {$set : {inS3: false, lastVersion: lastVersion}}, {upsert:true}, callback
  142. clearDocHistoryAsArchiveInProgress: (doc_id, update, callback) ->
  143. db.docHistoryStats.update {doc_id: ObjectId(doc_id.toString())}, {$unset : {inS3: true, lastVersion: true}}, callback
  144. markDocHistoryAsArchived: (doc_id, lastVersion, callback)->
  145. db.docHistoryStats.update {doc_id: ObjectId(doc_id.toString())}, {$set : {inS3: true}}, {upsert:true}, (error)->
  146. return callback(error) if error?
  147. # clear the archived entries from the docHistory now we have finally succeeded
  148. db.docHistory.remove { doc_id : ObjectId(doc_id.toString()), v: {$lte : lastVersion}, expiresAt: {$exists : false} }, (error)->
  149. return callback(error) if error?
  150. callback(error)
  151. markDocHistoryAsUnarchived: (doc_id, callback)->
  152. # note this removes any inS3 field, regardless of its value (true/false/null)
  153. db.docHistoryStats.update {doc_id: ObjectId(doc_id.toString())}, { $unset : { inS3: true, lastVersion: true} }, (error)->
  154. callback(error)