TrackChangesClient.coffee 4.8 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144
  1. async = require 'async'
  2. zlib = require 'zlib'
  3. request = require "request"
  4. Settings = require "settings-sharelatex"
  5. rclient = require("redis-sharelatex").createClient(Settings.redis.history) # Only works locally for now
  6. Keys = Settings.redis.history.key_schema
  7. {db, ObjectId} = require "../../../../app/js/mongojs"
  8. aws = require "aws-sdk"
  9. s3 = new aws.S3(
  10. accessKeyId: Settings.trackchanges.s3.key
  11. secretAccessKey: Settings.trackchanges.s3.secret
  12. endpoint: Settings.trackchanges.s3.endpoint
  13. s3ForcePathStyle: Settings.trackchanges.s3.pathStyle
  14. )
  15. S3_BUCKET = Settings.trackchanges.stores.doc_history
  16. module.exports = TrackChangesClient =
  17. flushAndGetCompressedUpdates: (project_id, doc_id, callback = (error, updates) ->) ->
  18. TrackChangesClient.flushDoc project_id, doc_id, (error) ->
  19. return callback(error) if error?
  20. TrackChangesClient.getCompressedUpdates doc_id, callback
  21. flushDoc: (project_id, doc_id, callback = (error) ->) ->
  22. request.post {
  23. url: "http://localhost:3015/project/#{project_id}/doc/#{doc_id}/flush"
  24. }, (error, response, body) =>
  25. response.statusCode.should.equal 204
  26. callback(error)
  27. flushProject: (project_id, callback = (error) ->) ->
  28. request.post {
  29. url: "http://localhost:3015/project/#{project_id}/flush"
  30. }, (error, response, body) =>
  31. response.statusCode.should.equal 204
  32. callback(error)
  33. getCompressedUpdates: (doc_id, callback = (error, updates) ->) ->
  34. db.docHistory
  35. .find(doc_id: ObjectId(doc_id))
  36. .sort("meta.end_ts": 1)
  37. .toArray callback
  38. getProjectMetaData: (project_id, callback = (error, updates) ->) ->
  39. db.projectHistoryMetaData
  40. .find {
  41. project_id: ObjectId(project_id)
  42. },
  43. (error, projects) ->
  44. callback error, projects[0]
  45. setPreserveHistoryForProject: (project_id, callback = (error) ->) ->
  46. db.projectHistoryMetaData.update {
  47. project_id: ObjectId(project_id)
  48. }, {
  49. $set: { preserveHistory: true }
  50. }, {
  51. upsert: true
  52. }, callback
  53. pushRawUpdates: (project_id, doc_id, updates, callback = (error) ->) ->
  54. rclient.sadd Keys.docsWithHistoryOps({project_id}), doc_id, (error) ->
  55. return callback(error) if error?
  56. rclient.rpush Keys.uncompressedHistoryOps({doc_id}), (JSON.stringify(u) for u in updates)..., callback
  57. getDiff: (project_id, doc_id, from, to, callback = (error, diff) ->) ->
  58. request.get {
  59. url: "http://localhost:3015/project/#{project_id}/doc/#{doc_id}/diff?from=#{from}&to=#{to}"
  60. }, (error, response, body) =>
  61. response.statusCode.should.equal 200
  62. callback null, JSON.parse(body)
  63. getUpdates: (project_id, options, callback = (error, body) ->) ->
  64. request.get {
  65. url: "http://localhost:3015/project/#{project_id}/updates?before=#{options.before}&min_count=#{options.min_count}"
  66. }, (error, response, body) =>
  67. response.statusCode.should.equal 200
  68. callback null, JSON.parse(body)
  69. restoreDoc: (project_id, doc_id, version, user_id, callback = (error) ->) ->
  70. request.post {
  71. url: "http://localhost:3015/project/#{project_id}/doc/#{doc_id}/version/#{version}/restore"
  72. headers:
  73. "X-User-Id": user_id
  74. }, (error, response, body) =>
  75. response.statusCode.should.equal 204
  76. callback null
  77. pushDocHistory: (project_id, doc_id, callback = (error) ->) ->
  78. request.post {
  79. url: "http://localhost:3015/project/#{project_id}/doc/#{doc_id}/push"
  80. }, (error, response, body) =>
  81. response.statusCode.should.equal 204
  82. callback(error)
  83. pullDocHistory: (project_id, doc_id, callback = (error) ->) ->
  84. request.post {
  85. url: "http://localhost:3015/project/#{project_id}/doc/#{doc_id}/pull"
  86. }, (error, response, body) =>
  87. response.statusCode.should.equal 204
  88. callback(error)
  89. waitForS3: (done, retries=42) ->
  90. if !Settings.trackchanges.s3.endpoint
  91. return done()
  92. request.get "#{Settings.trackchanges.s3.endpoint}/", (err, res) ->
  93. if res && res.statusCode < 500
  94. return done()
  95. if retries == 0
  96. return done(err or new Error("s3 returned #{res.statusCode}"))
  97. setTimeout () ->
  98. TrackChangesClient.waitForS3(done, --retries)
  99. , 1000
  100. getS3Doc: (project_id, doc_id, pack_id, callback = (error, body) ->) ->
  101. params =
  102. Bucket: S3_BUCKET
  103. Key: "#{project_id}/changes-#{doc_id}/pack-#{pack_id}"
  104. s3.getObject params, (error, data) ->
  105. return callback(error) if error?
  106. body = data.Body
  107. return callback(new Error("empty response from s3")) if not body?
  108. zlib.gunzip body, (err, result) ->
  109. return callback(err) if err?
  110. callback(null, JSON.parse(result.toString()))
  111. removeS3Doc: (project_id, doc_id, callback = (error, res, body) ->) ->
  112. params =
  113. Bucket: S3_BUCKET
  114. Prefix: "#{project_id}/changes-#{doc_id}"
  115. s3.listObjects params, (error, data) ->
  116. return callback(error) if error?
  117. params =
  118. Bucket: S3_BUCKET
  119. Delete:
  120. Objects: data.Contents.map((s3object) -> {Key: s3object.Key})
  121. s3.deleteObjects params, callback