MongoManager.coffee 11 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330
  1. {db, ObjectId} = require "./mongojs"
  2. async = require "async"
  3. _ = require "underscore"
  4. module.exports = MongoManager =
  5. # only used in this module
  6. getLastCompressedUpdate: (doc_id, callback = (error, update) ->) ->
  7. db.docHistory
  8. .find(doc_id: ObjectId(doc_id.toString()))
  9. .sort( v: -1 )
  10. .limit(1)
  11. .toArray (error, compressedUpdates) ->
  12. return callback(error) if error?
  13. return callback null, null if compressedUpdates[0]?.pack? # cannot pop from a pack
  14. return callback null, compressedUpdates[0] or null
  15. # only used in this module
  16. deleteCompressedUpdate: (id, callback = (error) ->) ->
  17. db.docHistory.remove({ _id: ObjectId(id.toString()) }, callback)
  18. # used in UpdatesManager
  19. popLastCompressedUpdate: (doc_id, callback = (error, update) ->) ->
  20. MongoManager.getLastCompressedUpdate doc_id, (error, update) ->
  21. return callback(error) if error?
  22. if update?
  23. MongoManager.deleteCompressedUpdate update._id, (error) ->
  24. return callback(error) if error?
  25. callback null, update
  26. else
  27. callback null, null
  28. # used in UpdatesManager
  29. insertCompressedUpdates: (project_id, doc_id, updates, permanent, callback = (error) ->) ->
  30. jobs = []
  31. for update in updates
  32. do (update) ->
  33. jobs.push (callback) -> MongoManager.insertCompressedUpdate project_id, doc_id, update, permanent, callback
  34. async.series jobs, callback
  35. insertCompressedUpdate: (project_id, doc_id, update, temporary, callback = (error) ->) ->
  36. update = {
  37. doc_id: ObjectId(doc_id.toString())
  38. project_id: ObjectId(project_id.toString())
  39. op: update.op
  40. meta: update.meta
  41. v: update.v
  42. }
  43. if temporary
  44. seconds = 1000
  45. minutes = 60 * seconds
  46. hours = 60 * minutes
  47. days = 24 * hours
  48. update.expiresAt = new Date(Date.now() + 7 * days)
  49. # may need to roll over a pack here if we are inserting packs
  50. db.docHistory.insert update, callback
  51. # The following function implements a method like a mongo find, but
  52. # which expands any documents containing a 'pack' field into
  53. # multiple values
  54. #
  55. # e.g. a single update looks like
  56. #
  57. # {
  58. # "doc_id" : 549dae9e0a2a615c0c7f0c98,
  59. # "project_id" : 549dae9c0a2a615c0c7f0c8c,
  60. # "op" : [ {"p" : 6981, "d" : "?" } ],
  61. # "meta" : { "user_id" : 52933..., "start_ts" : 1422310693931, "end_ts" : 1422310693931 },
  62. # "v" : 17082
  63. # }
  64. #
  65. # and a pack looks like this
  66. #
  67. # {
  68. # "doc_id" : 549dae9e0a2a615c0c7f0c98,
  69. # "project_id" : 549dae9c0a2a615c0c7f0c8c,
  70. # "pack" : [ D1, D2, D3, ....],
  71. # "meta" : { "user_id" : 52933..., "start_ts" : 1422310693931, "end_ts" : 1422310693931 },
  72. # "v" : 17082
  73. # }
  74. #
  75. # where D1, D2, D3, .... are single updates stripped of their
  76. # doc_id and project_id fields (which are the same for all the
  77. # updates in the pack). The meta and v fields of the pack itself
  78. # are those of the first entry in the pack D1 (this makes it
  79. # possible to treat packs and single updates in the same way).
  80. _findDocResults: (query, limit, callback) ->
  81. # query - the mongo query selector, includes both the doc_id/project_id and
  82. # the range on v
  83. # limit - the mongo limit, we need to apply it after unpacking any
  84. # packs
  85. sort = {}
  86. sort['v'] = -1;
  87. cursor = db.docHistory
  88. .find( query )
  89. .sort( sort )
  90. # if we have packs, we will trim the results more later after expanding them
  91. if limit?
  92. cursor.limit(limit)
  93. # take the part of the query which selects the range over the parameter
  94. rangeQuery = query['v']
  95. # helper function to check if an item from a pack is inside the
  96. # desired range
  97. filterFn = (item) ->
  98. return false if rangeQuery?['$gte']? && item['v'] < rangeQuery['$gte']
  99. return false if rangeQuery?['$lte']? && item['v'] > rangeQuery['$lte']
  100. return false if rangeQuery?['$lt']? && item['v'] >= rangeQuery['$lt']
  101. return false if rangeQuery?['$gt']? && item['v'] <= rangeQuery['$gt']
  102. return true
  103. # create a query which can be used to select the entries BEFORE
  104. # the range because we sometimes need to find extra ones (when the
  105. # boundary falls in the middle of a pack)
  106. extraQuery = _.clone(query)
  107. # The pack uses its first entry for its metadata and v, so the
  108. # only queries where we might not get all the packs are those for
  109. # $gt and $gte (i.e. we need to find packs which start before our
  110. # range but end in it)
  111. if rangeQuery?['$gte']?
  112. extraQuery['v'] = {'$lt' : rangeQuery['$gte']}
  113. else if rangeQuery?['$gt']
  114. extraQuery['v'] = {'$lte' : rangeQuery['$gt']}
  115. else
  116. delete extraQuery['v']
  117. needMore = false # keep track of whether we need to load more data
  118. updates = [] # used to accumulate the set of results
  119. cursor.toArray (err, result) ->
  120. unpackedSet = MongoManager._unpackResults(result)
  121. updates = MongoManager._filterAndLimit(updates, unpackedSet, filterFn, limit)
  122. # check if we need to retrieve more data, because there is a
  123. # pack that crosses into our range
  124. last = if unpackedSet.length then unpackedSet[unpackedSet.length-1] else null
  125. if limit? && updates.length == limit
  126. needMore = false
  127. else if extraQuery['v']? && last? && filterFn(last)
  128. needMore = true
  129. else if extraQuery['v']? && updates.length == 0
  130. needMore = true
  131. if needMore
  132. # we do need an extra result set
  133. extra = db.docHistory
  134. .find(extraQuery)
  135. .sort(sort)
  136. .limit(1)
  137. extra.toArray (err, result2) ->
  138. if err?
  139. return callback err, updates
  140. else
  141. extraSet = MongoManager._unpackResults(result2)
  142. updates = MongoManager._filterAndLimit(updates, extraSet, filterFn, limit)
  143. callback err, updates
  144. return
  145. if err?
  146. callback err, result
  147. else
  148. callback err, updates
  149. _findProjectResults: (query, limit, callback) ->
  150. # query - the mongo query selector, includes both the doc_id/project_id and
  151. # the range on v or meta.end_ts
  152. # limit - the mongo limit, we need to apply it after unpacking any
  153. # packs
  154. sort = {}
  155. sort['meta.end_ts'] = -1;
  156. cursor = db.docHistory
  157. .find( query, {"op":false, "pack.op": false} ) # no need to return the op only need version info
  158. .sort( sort )
  159. # if we have packs, we will trim the results more later after expanding them
  160. if limit?
  161. cursor.limit(limit)
  162. # take the part of the query which selects the range over the parameter
  163. before = query['meta.end_ts']?['$lt'] # may be null
  164. updates = [] # used to accumulate the set of results
  165. extraQuery = _.clone(query)
  166. cursor.toArray (err, result) ->
  167. if err?
  168. return callback err, result
  169. if result.length == 0 && not before? # no results and no time range specified
  170. return callback err, result
  171. unpackedSet = MongoManager._unpackResults(result)
  172. if limit?
  173. unpackedSet = unpackedSet.slice(0, limit)
  174. # find the end time of the last result, we will take all the
  175. # results up to this, and then all the changes at that time
  176. # (without imposing a limit)
  177. cutoff = if unpackedSet.length then unpackedSet[unpackedSet.length-1].meta.end_ts else before
  178. console.log 'before is', before
  179. console.log 'cutoff is', cutoff
  180. console.log 'limit is', limit
  181. filterFn = (item) ->
  182. ts = item?.meta?.end_ts
  183. return false if before? && ts >= before
  184. return false if cutoff? && ts < cutoff
  185. return true
  186. updates = MongoManager._filterAndLimit(updates, unpackedSet, filterFn, limit)
  187. console.log 'initial updates are', updates
  188. # now find any extra entries which are exactly at the last time (cutoff)
  189. # or in packs that overlap it
  190. extraQuery['meta.end_ts'] = {"$gte": +cutoff}
  191. extraQuery['meta.start_ts'] = {"$lte": +cutoff}
  192. # we don't specify a limit here, as there could be any number of
  193. # docs at that timestamp
  194. #
  195. # NB. need to catch items in original query and followup query for duplicates
  196. console.log 'extraQuery is', extraQuery
  197. extra = db.docHistory
  198. .find(extraQuery, {"op": false, "pack.op": false})
  199. .sort(sort)
  200. extra.toArray (err, result2) ->
  201. console.log 'got extra result', err, result
  202. if err?
  203. return callback err, updates
  204. else
  205. extraSet = MongoManager._unpackResults(result2)
  206. # note: final argument is null, no limit applied because we
  207. # need all the updates at the final time to avoid breaking
  208. # the changeset into parts
  209. updates = MongoManager._filterAndLimit(updates, extraSet, filterFn, null)
  210. # remove duplicates
  211. seen = {}
  212. updates = updates.filter (item) ->
  213. key = item.doc_id + ' ' + item.v
  214. console.log 'key is', key
  215. if seen[key]
  216. return false
  217. else
  218. seen[key] = true
  219. console.log 'extra updates are', updates
  220. callback err, updates
  221. _unpackResults: (updates) ->
  222. # iterate over the updates, if there's a pack, expand it into ops and
  223. # insert it into the array at that point
  224. result = []
  225. updates.forEach (item) ->
  226. if item.pack?
  227. all = MongoManager._explodePackToOps item
  228. result = result.concat all
  229. else
  230. result.push item
  231. return result
  232. _explodePackToOps: (packObj) ->
  233. # convert a pack into an array of ops
  234. doc_id = packObj.doc_id
  235. project_id = packObj.project_id
  236. result = packObj.pack.map (item) ->
  237. item.doc_id = doc_id
  238. item.project_id = project_id
  239. item
  240. return result.reverse()
  241. _filterAndLimit: (results, extra, filterFn, limit) ->
  242. # update results with extra docs, after filtering and limiting
  243. filtered = extra.filter(filterFn)
  244. newResults = results.concat filtered
  245. newResults.slice(0, limit) if limit?
  246. return newResults
  247. getDocUpdates:(doc_id, options = {}, callback = (error, updates) ->) ->
  248. query =
  249. doc_id: ObjectId(doc_id.toString())
  250. if options.from?
  251. query["v"] ||= {}
  252. query["v"]["$gte"] = options.from
  253. if options.to?
  254. query["v"] ||= {}
  255. query["v"]["$lte"] = options.to
  256. MongoManager._findDocResults(query, options.limit, callback)
  257. getProjectUpdates: (project_id, options = {}, callback = (error, updates) ->) ->
  258. query =
  259. project_id: ObjectId(project_id.toString())
  260. if options.before?
  261. query["meta.end_ts"] = { $lt: options.before }
  262. MongoManager._findProjectResults(query, options.limit, callback)
  263. backportProjectId: (project_id, doc_id, callback = (error) ->) ->
  264. db.docHistory.update {
  265. doc_id: ObjectId(doc_id.toString())
  266. project_id: { $exists: false }
  267. }, {
  268. $set: { project_id: ObjectId(project_id.toString()) }
  269. }, {
  270. multi: true
  271. }, callback
  272. getProjectMetaData: (project_id, callback = (error, metadata) ->) ->
  273. db.projectHistoryMetaData.find {
  274. project_id: ObjectId(project_id.toString())
  275. }, (error, results) ->
  276. return callback(error) if error?
  277. callback null, results[0]
  278. setProjectMetaData: (project_id, metadata, callback = (error) ->) ->
  279. db.projectHistoryMetaData.update {
  280. project_id: ObjectId(project_id)
  281. }, {
  282. $set: metadata
  283. }, {
  284. upsert: true
  285. }, callback
  286. ensureIndices: () ->
  287. # For finding all updates that go into a diff for a doc (getLastCompressedUpdate, getDocUpdates v > from && v < to)
  288. db.docHistory.ensureIndex { doc_id: 1, v: 1 }, { background: true }
  289. # For finding all updates that affect a project (getProjectUpdates meta.end_ts < before
  290. db.docHistory.ensureIndex { project_id: 1, "meta.end_ts": 1 }, { background: true }
  291. # For finding updates that don't yet have a project_id and need it inserting
  292. db.docHistory.ensureIndex { doc_id: 1, project_id: 1 }, { background: true }
  293. # For finding project meta-data
  294. db.projectHistoryMetaData.ensureIndex { project_id: 1 }, { background: true }
  295. # TTL index for auto deleting week old temporary ops
  296. db.docHistory.ensureIndex { expiresAt: 1 }, { expireAfterSeconds: 0, background: true }