write.js 5.0 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184
  1. 'use strict'
  2. const util = require('util')
  3. const contentPath = require('./path')
  4. const fixOwner = require('../util/fix-owner')
  5. const fs = require('fs')
  6. const moveFile = require('../util/move-file')
  7. const Minipass = require('minipass')
  8. const Pipeline = require('minipass-pipeline')
  9. const Flush = require('minipass-flush')
  10. const path = require('path')
  11. const rimraf = util.promisify(require('rimraf'))
  12. const ssri = require('ssri')
  13. const uniqueFilename = require('unique-filename')
  14. const { disposer } = require('./../util/disposer')
  15. const fsm = require('fs-minipass')
  16. const writeFile = util.promisify(fs.writeFile)
  17. module.exports = write
  18. function write (cache, data, opts = {}) {
  19. const { algorithms, size, integrity } = opts
  20. if (algorithms && algorithms.length > 1) {
  21. throw new Error('opts.algorithms only supports a single algorithm for now')
  22. }
  23. if (typeof size === 'number' && data.length !== size) {
  24. return Promise.reject(sizeError(size, data.length))
  25. }
  26. const sri = ssri.fromData(data, algorithms ? { algorithms } : {})
  27. if (integrity && !ssri.checkData(data, integrity, opts)) {
  28. return Promise.reject(checksumError(integrity, sri))
  29. }
  30. return disposer(makeTmp(cache, opts), makeTmpDisposer,
  31. (tmp) => {
  32. return writeFile(tmp.target, data, { flag: 'wx' })
  33. .then(() => moveToDestination(tmp, cache, sri, opts))
  34. })
  35. .then(() => ({ integrity: sri, size: data.length }))
  36. }
  37. module.exports.stream = writeStream
  38. // writes proxied to the 'inputStream' that is passed to the Promise
  39. // 'end' is deferred until content is handled.
  40. class CacacheWriteStream extends Flush {
  41. constructor (cache, opts) {
  42. super()
  43. this.opts = opts
  44. this.cache = cache
  45. this.inputStream = new Minipass()
  46. this.inputStream.on('error', er => this.emit('error', er))
  47. this.inputStream.on('drain', () => this.emit('drain'))
  48. this.handleContentP = null
  49. }
  50. write (chunk, encoding, cb) {
  51. if (!this.handleContentP) {
  52. this.handleContentP = handleContent(
  53. this.inputStream,
  54. this.cache,
  55. this.opts
  56. )
  57. }
  58. return this.inputStream.write(chunk, encoding, cb)
  59. }
  60. flush (cb) {
  61. this.inputStream.end(() => {
  62. if (!this.handleContentP) {
  63. const e = new Error('Cache input stream was empty')
  64. e.code = 'ENODATA'
  65. // empty streams are probably emitting end right away.
  66. // defer this one tick by rejecting a promise on it.
  67. return Promise.reject(e).catch(cb)
  68. }
  69. this.handleContentP.then(
  70. (res) => {
  71. res.integrity && this.emit('integrity', res.integrity)
  72. res.size !== null && this.emit('size', res.size)
  73. cb()
  74. },
  75. (er) => cb(er)
  76. )
  77. })
  78. }
  79. }
  80. function writeStream (cache, opts = {}) {
  81. return new CacacheWriteStream(cache, opts)
  82. }
  83. function handleContent (inputStream, cache, opts) {
  84. return disposer(makeTmp(cache, opts), makeTmpDisposer, (tmp) => {
  85. return pipeToTmp(inputStream, cache, tmp.target, opts)
  86. .then((res) => {
  87. return moveToDestination(
  88. tmp,
  89. cache,
  90. res.integrity,
  91. opts
  92. ).then(() => res)
  93. })
  94. })
  95. }
  96. function pipeToTmp (inputStream, cache, tmpTarget, opts) {
  97. let integrity
  98. let size
  99. const hashStream = ssri.integrityStream({
  100. integrity: opts.integrity,
  101. algorithms: opts.algorithms,
  102. size: opts.size
  103. })
  104. hashStream.on('integrity', i => { integrity = i })
  105. hashStream.on('size', s => { size = s })
  106. const outStream = new fsm.WriteStream(tmpTarget, {
  107. flags: 'wx'
  108. })
  109. // NB: this can throw if the hashStream has a problem with
  110. // it, and the data is fully written. but pipeToTmp is only
  111. // called in promisory contexts where that is handled.
  112. const pipeline = new Pipeline(
  113. inputStream,
  114. hashStream,
  115. outStream
  116. )
  117. return pipeline.promise()
  118. .then(() => ({ integrity, size }))
  119. .catch(er => rimraf(tmpTarget).then(() => { throw er }))
  120. }
  121. function makeTmp (cache, opts) {
  122. const tmpTarget = uniqueFilename(path.join(cache, 'tmp'), opts.tmpPrefix)
  123. return fixOwner.mkdirfix(cache, path.dirname(tmpTarget)).then(() => ({
  124. target: tmpTarget,
  125. moved: false
  126. }))
  127. }
  128. function makeTmpDisposer (tmp) {
  129. if (tmp.moved) {
  130. return Promise.resolve()
  131. }
  132. return rimraf(tmp.target)
  133. }
  134. function moveToDestination (tmp, cache, sri, opts) {
  135. const destination = contentPath(cache, sri)
  136. const destDir = path.dirname(destination)
  137. return fixOwner
  138. .mkdirfix(cache, destDir)
  139. .then(() => {
  140. return moveFile(tmp.target, destination)
  141. })
  142. .then(() => {
  143. tmp.moved = true
  144. return fixOwner.chownr(cache, destination)
  145. })
  146. }
  147. function sizeError (expected, found) {
  148. const err = new Error(`Bad data size: expected inserted data to be ${expected} bytes, but got ${found} instead`)
  149. err.expected = expected
  150. err.found = found
  151. err.code = 'EBADSIZE'
  152. return err
  153. }
  154. function checksumError (expected, found) {
  155. const err = new Error(`Integrity check failed:
  156. Wanted: ${expected}
  157. Found: ${found}`)
  158. err.code = 'EINTEGRITY'
  159. err.expected = expected
  160. err.found = found
  161. return err
  162. }