FSPersistor.js 7.0 KB

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