MongoAWS.js 5.9 KB

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