FSPersistor.js 7.0 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273
  1. const fs = require('fs')
  2. const fsPromises = require('fs/promises')
  3. const globCallbacks = require('glob')
  4. const uuid = require('node-uuid')
  5. const path = require('path')
  6. const { pipeline } = require('stream/promises')
  7. const { promisify } = require('util')
  8. const AbstractPersistor = require('./AbstractPersistor')
  9. const { NotFoundError, ReadError, WriteError } = require('./Errors')
  10. const PersistorHelper = require('./PersistorHelper')
  11. const glob = promisify(globCallbacks)
  12. const filterName = key => key.replace(/\//g, '_')
  13. module.exports = class FSPersistor extends AbstractPersistor {
  14. constructor(settings) {
  15. super()
  16. this.settings = settings
  17. }
  18. async sendFile(location, target, source) {
  19. const filteredTarget = filterName(target)
  20. // actually copy the file (instead of moving it) to maintain consistent behaviour
  21. // between the different implementations
  22. try {
  23. const sourceStream = fs.createReadStream(source)
  24. const targetStream = fs.createWriteStream(`${location}/${filteredTarget}`)
  25. await pipeline(sourceStream, targetStream)
  26. } catch (err) {
  27. throw PersistorHelper.wrapError(
  28. err,
  29. 'failed to copy the specified file',
  30. { location, target, source },
  31. WriteError
  32. )
  33. }
  34. }
  35. async sendStream(location, target, sourceStream, opts = {}) {
  36. const fsPath = await this._writeStream(sourceStream)
  37. let sourceMd5 = opts.sourceMd5
  38. if (!sourceMd5) {
  39. sourceMd5 = await FSPersistor._getFileMd5HashForPath(fsPath)
  40. }
  41. try {
  42. await this.sendFile(location, target, fsPath)
  43. const destMd5 = await this.getObjectMd5Hash(location, target)
  44. if (sourceMd5 !== destMd5) {
  45. await this._deleteFile(`${location}/${filterName(target)}`)
  46. throw new WriteError('md5 hash mismatch', {
  47. sourceMd5,
  48. destMd5,
  49. location,
  50. target,
  51. })
  52. }
  53. } finally {
  54. await this._deleteFile(fsPath)
  55. }
  56. }
  57. // opts may be {start: Number, end: Number}
  58. async getObjectStream(location, name, opts) {
  59. const filteredName = filterName(name)
  60. try {
  61. opts.fd = await fsPromises.open(`${location}/${filteredName}`, 'r')
  62. } catch (err) {
  63. throw PersistorHelper.wrapError(
  64. err,
  65. 'failed to open file for streaming',
  66. { location, filteredName, opts },
  67. ReadError
  68. )
  69. }
  70. return fs.createReadStream(null, opts)
  71. }
  72. async getRedirectUrl() {
  73. // not implemented
  74. return null
  75. }
  76. async getObjectSize(location, filename) {
  77. const fullPath = path.join(location, filterName(filename))
  78. try {
  79. const stat = await fsPromises.stat(fullPath)
  80. return stat.size
  81. } catch (err) {
  82. throw PersistorHelper.wrapError(
  83. err,
  84. 'failed to stat file',
  85. { location, filename },
  86. ReadError
  87. )
  88. }
  89. }
  90. async getObjectMd5Hash(location, filename) {
  91. const fullPath = path.join(location, filterName(filename))
  92. try {
  93. return await FSPersistor._getFileMd5HashForPath(fullPath)
  94. } catch (err) {
  95. throw new ReadError(
  96. 'unable to get md5 hash from file',
  97. { location, filename },
  98. err
  99. )
  100. }
  101. }
  102. async copyObject(location, fromName, toName) {
  103. const filteredFromName = filterName(fromName)
  104. const filteredToName = filterName(toName)
  105. try {
  106. const sourceStream = fs.createReadStream(
  107. `${location}/${filteredFromName}`
  108. )
  109. const targetStream = fs.createWriteStream(`${location}/${filteredToName}`)
  110. await pipeline(sourceStream, targetStream)
  111. } catch (err) {
  112. throw PersistorHelper.wrapError(
  113. err,
  114. 'failed to copy file',
  115. { location, filteredFromName, filteredToName },
  116. WriteError
  117. )
  118. }
  119. }
  120. async deleteObject(location, name) {
  121. const filteredName = filterName(name)
  122. try {
  123. await fsPromises.unlink(`${location}/${filteredName}`)
  124. } catch (err) {
  125. const wrappedError = PersistorHelper.wrapError(
  126. err,
  127. 'failed to delete file',
  128. { location, filteredName },
  129. WriteError
  130. )
  131. if (!(wrappedError instanceof NotFoundError)) {
  132. // S3 doesn't give us a 404 when a file wasn't there to be deleted, so we
  133. // should be consistent here as well
  134. throw wrappedError
  135. }
  136. }
  137. }
  138. async deleteDirectory(location, name) {
  139. const filteredName = filterName(name.replace(/\/$/, ''))
  140. try {
  141. await Promise.all(
  142. (
  143. await glob(`${location}/${filteredName}_*`)
  144. ).map(file => fsPromises.unlink(file))
  145. )
  146. } catch (err) {
  147. throw PersistorHelper.wrapError(
  148. err,
  149. 'failed to delete directory',
  150. { location, filteredName },
  151. WriteError
  152. )
  153. }
  154. }
  155. async checkIfObjectExists(location, name) {
  156. const filteredName = filterName(name)
  157. try {
  158. const stat = await fsPromises.stat(`${location}/${filteredName}`)
  159. return !!stat
  160. } catch (err) {
  161. if (err.code === 'ENOENT') {
  162. return false
  163. }
  164. throw PersistorHelper.wrapError(
  165. err,
  166. 'failed to stat file',
  167. { location, filteredName },
  168. ReadError
  169. )
  170. }
  171. }
  172. // note, does not recurse into subdirectories, as we use a flattened directory structure
  173. async directorySize(location, name) {
  174. const filteredName = filterName(name.replace(/\/$/, ''))
  175. let size = 0
  176. try {
  177. const files = await glob(`${location}/${filteredName}_*`)
  178. for (const file of files) {
  179. try {
  180. const stat = await fsPromises.stat(file)
  181. if (stat.isFile()) {
  182. size += stat.size
  183. }
  184. } catch (err) {
  185. // ignore files that may have just been deleted
  186. if (err.code !== 'ENOENT') {
  187. throw err
  188. }
  189. }
  190. }
  191. } catch (err) {
  192. throw PersistorHelper.wrapError(
  193. err,
  194. 'failed to get directory size',
  195. { location, name },
  196. ReadError
  197. )
  198. }
  199. return size
  200. }
  201. _getPath(key) {
  202. if (key == null) {
  203. key = uuid.v1()
  204. }
  205. key = key.replace(/\//g, '-')
  206. return path.join(this.settings.paths.uploadFolder, key)
  207. }
  208. async _writeStream(stream, key) {
  209. let timer
  210. if (this.settings.Metrics) {
  211. timer = new this.settings.Metrics.Timer('writingFile')
  212. }
  213. const fsPath = this._getPath(key)
  214. const writeStream = fs.createWriteStream(fsPath)
  215. try {
  216. await pipeline(stream, writeStream)
  217. if (timer) {
  218. timer.done()
  219. }
  220. return fsPath
  221. } catch (err) {
  222. await this._deleteFile(fsPath)
  223. throw new WriteError('problem writing file locally', { err, fsPath }, err)
  224. }
  225. }
  226. async _deleteFile(fsPath) {
  227. if (!fsPath) {
  228. return
  229. }
  230. try {
  231. await fsPromises.unlink(fsPath)
  232. } catch (err) {
  233. if (err.code !== 'ENOENT') {
  234. throw new WriteError('failed to delete file', { fsPath }, err)
  235. }
  236. }
  237. }
  238. static async _getFileMd5HashForPath(fullPath) {
  239. const stream = fs.createReadStream(fullPath)
  240. return PersistorHelper.calculateStreamMd5(stream)
  241. }
  242. }