MongoAWS.js 6.1 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207
  1. /* eslint-disable
  2. camelcase,
  3. no-return-assign,
  4. no-unused-vars,
  5. */
  6. // TODO: This file was created by bulk-decaffeinate.
  7. // Fix any style issues and re-enable lint.
  8. /*
  9. * decaffeinate suggestions:
  10. * DS101: Remove unnecessary use of Array.from
  11. * DS102: Remove unnecessary code created because of implicit returns
  12. * DS207: Consider shorter variations of null checks
  13. * Full docs: https://github.com/decaffeinate/decaffeinate/blob/master/docs/suggestions.md
  14. */
  15. let MongoAWS
  16. const settings = require('@overleaf/settings')
  17. const logger = require('@overleaf/logger')
  18. const AWS = require('aws-sdk')
  19. const S3S = require('s3-streams')
  20. const { db, ObjectId } = require('./mongodb')
  21. const JSONStream = require('JSONStream')
  22. const ReadlineStream = require('byline')
  23. const zlib = require('zlib')
  24. const Metrics = require('@overleaf/metrics')
  25. const DAYS = 24 * 3600 * 1000 // one day in milliseconds
  26. const createStream = function (streamConstructor, project_id, doc_id, pack_id) {
  27. const AWS_CONFIG = {
  28. accessKeyId: settings.trackchanges.s3.key,
  29. secretAccessKey: settings.trackchanges.s3.secret,
  30. endpoint: settings.trackchanges.s3.endpoint,
  31. s3ForcePathStyle: settings.trackchanges.s3.pathStyle,
  32. }
  33. return streamConstructor(new AWS.S3(AWS_CONFIG), {
  34. Bucket: settings.trackchanges.stores.doc_history,
  35. Key: project_id + '/changes-' + doc_id + '/pack-' + pack_id,
  36. })
  37. }
  38. module.exports = MongoAWS = {
  39. archivePack(project_id, doc_id, pack_id, _callback) {
  40. if (_callback == null) {
  41. _callback = function () {}
  42. }
  43. const callback = function (...args) {
  44. _callback(...Array.from(args || []))
  45. return (_callback = function () {})
  46. }
  47. const query = {
  48. _id: ObjectId(pack_id),
  49. doc_id: ObjectId(doc_id),
  50. }
  51. if (project_id == null) {
  52. return callback(new Error('invalid project id'))
  53. }
  54. if (doc_id == null) {
  55. return callback(new Error('invalid doc id'))
  56. }
  57. if (pack_id == null) {
  58. return callback(new Error('invalid pack id'))
  59. }
  60. logger.debug({ project_id, doc_id, pack_id }, 'uploading data to s3')
  61. const upload = createStream(S3S.WriteStream, project_id, doc_id, pack_id)
  62. return db.docHistory.findOne(query, function (err, result) {
  63. if (err != null) {
  64. return callback(err)
  65. }
  66. if (result == null) {
  67. return callback(new Error('cannot find pack to send to s3'))
  68. }
  69. if (result.expiresAt != null) {
  70. return callback(new Error('refusing to send pack with TTL to s3'))
  71. }
  72. const uncompressedData = JSON.stringify(result)
  73. if (uncompressedData.indexOf('\u0000') !== -1) {
  74. const error = new Error('null bytes found in upload')
  75. logger.error({ err: error, project_id, doc_id, pack_id }, error.message)
  76. return callback(error)
  77. }
  78. return zlib.gzip(uncompressedData, function (err, buf) {
  79. logger.debug(
  80. {
  81. project_id,
  82. doc_id,
  83. pack_id,
  84. origSize: uncompressedData.length,
  85. newSize: buf.length,
  86. },
  87. 'compressed pack'
  88. )
  89. if (err != null) {
  90. return callback(err)
  91. }
  92. upload.on('error', err => callback(err))
  93. upload.on('finish', function () {
  94. Metrics.inc('archive-pack')
  95. logger.debug(
  96. { project_id, doc_id, pack_id },
  97. 'upload to s3 completed'
  98. )
  99. return callback(null)
  100. })
  101. upload.write(buf)
  102. return upload.end()
  103. })
  104. })
  105. },
  106. readArchivedPack(project_id, doc_id, pack_id, _callback) {
  107. if (_callback == null) {
  108. _callback = function () {}
  109. }
  110. const callback = function (...args) {
  111. _callback(...Array.from(args || []))
  112. return (_callback = function () {})
  113. }
  114. if (project_id == null) {
  115. return callback(new Error('invalid project id'))
  116. }
  117. if (doc_id == null) {
  118. return callback(new Error('invalid doc id'))
  119. }
  120. if (pack_id == null) {
  121. return callback(new Error('invalid pack id'))
  122. }
  123. logger.debug({ project_id, doc_id, pack_id }, 'downloading data from s3')
  124. const download = createStream(S3S.ReadStream, project_id, doc_id, pack_id)
  125. const inputStream = download
  126. .on('open', obj => 1)
  127. .on('error', err => callback(err))
  128. const gunzip = zlib.createGunzip()
  129. gunzip.setEncoding('utf8')
  130. gunzip.on('error', function (err) {
  131. logger.debug(
  132. { project_id, doc_id, pack_id, err },
  133. 'error uncompressing gzip stream'
  134. )
  135. return callback(err)
  136. })
  137. const outputStream = inputStream.pipe(gunzip)
  138. const parts = []
  139. outputStream.on('error', err => callback(err))
  140. outputStream.on('end', function () {
  141. let object
  142. logger.debug(
  143. { project_id, doc_id, pack_id },
  144. 'download from s3 completed'
  145. )
  146. try {
  147. object = JSON.parse(parts.join(''))
  148. } catch (e) {
  149. return callback(e)
  150. }
  151. object._id = ObjectId(object._id)
  152. object.doc_id = ObjectId(object.doc_id)
  153. object.project_id = ObjectId(object.project_id)
  154. for (const op of Array.from(object.pack)) {
  155. if (op._id != null) {
  156. op._id = ObjectId(op._id)
  157. }
  158. }
  159. return callback(null, object)
  160. })
  161. return outputStream.on('data', data => parts.push(data))
  162. },
  163. unArchivePack(project_id, doc_id, pack_id, callback) {
  164. if (callback == null) {
  165. callback = function () {}
  166. }
  167. return MongoAWS.readArchivedPack(
  168. project_id,
  169. doc_id,
  170. pack_id,
  171. function (err, object) {
  172. if (err != null) {
  173. return callback(err)
  174. }
  175. Metrics.inc('unarchive-pack')
  176. // allow the object to expire, we can always retrieve it again
  177. object.expiresAt = new Date(Date.now() + 7 * DAYS)
  178. logger.debug(
  179. { project_id, doc_id, pack_id },
  180. 'inserting object from s3'
  181. )
  182. return db.docHistory.insertOne(object, (err, confirmation) => {
  183. if (err) return callback(err)
  184. object._id = confirmation.insertedId
  185. callback(null, object)
  186. })
  187. }
  188. )
  189. },
  190. }