FSPersistor.js 7.1 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278
  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, sourceMd5) {
  39. const fsPath = await this._writeStream(sourceStream)
  40. if (!sourceMd5) {
  41. sourceMd5 = await FSPersistor._getFileMd5HashForPath(fsPath)
  42. }
  43. try {
  44. await this.sendFile(location, target, fsPath)
  45. const destMd5 = await this.getObjectMd5Hash(location, target)
  46. if (sourceMd5 !== destMd5) {
  47. await this._deleteFile(`${location}/${filterName(target)}`)
  48. throw new WriteError({
  49. message: 'md5 hash mismatch',
  50. info: { sourceMd5, destMd5, location, 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 fsOpen(`${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 fsStat(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. message: 'unable to get md5 hash from file',
  97. info: { location, filename }
  98. }).withCause(err)
  99. }
  100. }
  101. async copyObject(location, fromName, toName) {
  102. const filteredFromName = filterName(fromName)
  103. const filteredToName = filterName(toName)
  104. try {
  105. const sourceStream = fs.createReadStream(
  106. `${location}/${filteredFromName}`
  107. )
  108. const targetStream = fs.createWriteStream(`${location}/${filteredToName}`)
  109. await pipeline(sourceStream, targetStream)
  110. } catch (err) {
  111. throw PersistorHelper.wrapError(
  112. err,
  113. 'failed to copy file',
  114. { location, filteredFromName, filteredToName },
  115. WriteError
  116. )
  117. }
  118. }
  119. async deleteObject(location, name) {
  120. const filteredName = filterName(name)
  121. try {
  122. await fsUnlink(`${location}/${filteredName}`)
  123. } catch (err) {
  124. const wrappedError = PersistorHelper.wrapError(
  125. err,
  126. 'failed to delete file',
  127. { location, filteredName },
  128. WriteError
  129. )
  130. if (!(wrappedError instanceof NotFoundError)) {
  131. // S3 doesn't give us a 404 when a file wasn't there to be deleted, so we
  132. // should be consistent here as well
  133. throw wrappedError
  134. }
  135. }
  136. }
  137. async deleteDirectory(location, name) {
  138. const filteredName = filterName(name.replace(/\/$/, ''))
  139. try {
  140. await Promise.all(
  141. (await fsGlob(`${location}/${filteredName}_*`)).map((file) =>
  142. fsUnlink(file)
  143. )
  144. )
  145. } catch (err) {
  146. throw PersistorHelper.wrapError(
  147. err,
  148. 'failed to delete directory',
  149. { location, filteredName },
  150. WriteError
  151. )
  152. }
  153. }
  154. async checkIfObjectExists(location, name) {
  155. const filteredName = filterName(name)
  156. try {
  157. const stat = await fsStat(`${location}/${filteredName}`)
  158. return !!stat
  159. } catch (err) {
  160. if (err.code === 'ENOENT') {
  161. return false
  162. }
  163. throw PersistorHelper.wrapError(
  164. err,
  165. 'failed to stat file',
  166. { location, filteredName },
  167. ReadError
  168. )
  169. }
  170. }
  171. // note, does not recurse into subdirectories, as we use a flattened directory structure
  172. async directorySize(location, name) {
  173. const filteredName = filterName(name.replace(/\/$/, ''))
  174. let size = 0
  175. try {
  176. const files = await fsGlob(`${location}/${filteredName}_*`)
  177. for (const file of files) {
  178. try {
  179. const stat = await fsStat(file)
  180. if (stat.isFile()) {
  181. size += stat.size
  182. }
  183. } catch (err) {
  184. // ignore files that may have just been deleted
  185. if (err.code !== 'ENOENT') {
  186. throw err
  187. }
  188. }
  189. }
  190. } catch (err) {
  191. throw PersistorHelper.wrapError(
  192. err,
  193. 'failed to get directory size',
  194. { location, name },
  195. ReadError
  196. )
  197. }
  198. return size
  199. }
  200. _getPath(key) {
  201. if (key == null) {
  202. key = uuid.v1()
  203. }
  204. key = key.replace(/\//g, '-')
  205. return path.join(this.settings.paths.uploadFolder, key)
  206. }
  207. async _writeStream(stream, key) {
  208. let timer
  209. if (this.settings.Metrics) {
  210. timer = new this.settings.Metrics.Timer('writingFile')
  211. }
  212. const fsPath = this._getPath(key)
  213. const writeStream = fs.createWriteStream(fsPath)
  214. try {
  215. await pipeline(stream, writeStream)
  216. if (timer) {
  217. timer.done()
  218. }
  219. return fsPath
  220. } catch (err) {
  221. await this._deleteFile(fsPath)
  222. throw new WriteError({
  223. message: 'problem writing file locally',
  224. info: { err, fsPath }
  225. }).withCause(err)
  226. }
  227. }
  228. async _deleteFile(fsPath) {
  229. if (!fsPath) {
  230. return
  231. }
  232. try {
  233. await fsUnlink(fsPath)
  234. } catch (err) {
  235. if (err.code !== 'ENOENT') {
  236. throw new WriteError({
  237. message: 'failed to delete file',
  238. info: { fsPath }
  239. }).withCause(err)
  240. }
  241. }
  242. }
  243. static async _getFileMd5HashForPath(fullPath) {
  244. const stream = fs.createReadStream(fullPath)
  245. return PersistorHelper.calculateStreamMd5(stream)
  246. }
  247. }