storage.js 12 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426
  1. /*
  2. * This file is part of the storage node for the Joystream project.
  3. * Copyright (C) 2019 Joystream Contributors
  4. *
  5. * This program is free software: you can redistribute it and/or modify
  6. * it under the terms of the GNU General Public License as published by
  7. * the Free Software Foundation, either version 3 of the License, or
  8. * (at your option) any later version.
  9. *
  10. * This program is distributed in the hope that it will be useful,
  11. * but WITHOUT ANY WARRANTY; without even the implied warranty of
  12. * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
  13. * GNU General Public License for more details.
  14. *
  15. * You should have received a copy of the GNU General Public License
  16. * along with this program. If not, see <https://www.gnu.org/licenses/>.
  17. */
  18. 'use strict'
  19. const { Transform } = require('stream')
  20. const fs = require('fs')
  21. const debug = require('debug')('joystream:storage:storage')
  22. const Promise = require('bluebird')
  23. const Hash = require('ipfs-only-hash')
  24. Promise.config({
  25. cancellation: true,
  26. })
  27. const fileType = require('file-type')
  28. const ipfsClient = require('ipfs-http-client')
  29. const temp = require('temp').track()
  30. const _ = require('lodash')
  31. // Default request timeout; imposed on top of the IPFS client, because the
  32. // client doesn't seem to care.
  33. const DEFAULT_TIMEOUT = 30 * 1000
  34. // Default/dummy resolution implementation.
  35. const DEFAULT_RESOLVE_CONTENT_ID = async (original) => {
  36. debug('Warning: Default resolution returns original CID', original)
  37. return original
  38. }
  39. // Default file info if nothing could be detected.
  40. const DEFAULT_FILE_INFO = {
  41. mimeType: 'application/octet-stream',
  42. ext: 'bin',
  43. }
  44. /*
  45. * fileType is a weird name, because we're really looking at MIME types.
  46. * Also, the type field includes extension info, so we're going to call
  47. * it fileInfo { mimeType, ext } instead.
  48. * Nitpicking, but it also means we can add our default type if things
  49. * go wrong.
  50. */
  51. function fixFileInfo(info) {
  52. if (!info) {
  53. info = DEFAULT_FILE_INFO
  54. } else {
  55. info.mimeType = info.mime
  56. delete info.mime
  57. }
  58. return info
  59. }
  60. function fixFileInfoOnStream(stream) {
  61. const info = fixFileInfo(stream.fileType)
  62. delete stream.fileType
  63. stream.fileInfo = info
  64. return stream
  65. }
  66. /*
  67. * Internal Transform stream for helping write to a temporary location, adding
  68. * MIME type detection, and a commit() function.
  69. */
  70. class StorageWriteStream extends Transform {
  71. constructor(storage, options) {
  72. options = _.clone(options || {})
  73. super(options)
  74. this.storage = storage
  75. // Create temp target.
  76. this.temp = temp.createWriteStream()
  77. this.buf = Buffer.alloc(0)
  78. }
  79. _transform(chunk, encoding, callback) {
  80. // Deal with buffers only
  81. if (typeof chunk === 'string') {
  82. chunk = Buffer.from(chunk)
  83. }
  84. this.temp.write(chunk, (err) => {
  85. // Try to detect file type during streaming.
  86. if (!this.fileInfo && this.buf.byteLength <= fileType.minimumBytes) {
  87. this.buf = Buffer.concat([this.buf, chunk])
  88. if (this.buf >= fileType.minimumBytes) {
  89. const info = fileType(this.buf)
  90. // No info? We will try again at the end of the stream.
  91. if (info) {
  92. this.fileInfo = fixFileInfo(info)
  93. this.emit('fileInfo', this.fileInfo)
  94. }
  95. }
  96. }
  97. callback(err)
  98. })
  99. }
  100. _flush(callback) {
  101. debug('Flushing temporary stream:', this.temp.path)
  102. this.temp.end(() => {
  103. callback(null)
  104. })
  105. }
  106. /*
  107. * Get file info
  108. */
  109. async info() {
  110. if (!this.temp) {
  111. throw new Error('Cannot get info on temporary stream that does not exist. Did you call cleanup()?')
  112. }
  113. if (!this.fileInfo) {
  114. const read = fs.createReadStream(this.temp.path)
  115. const stream = await fileType.stream(read)
  116. this.fileInfo = fixFileInfoOnStream(stream).fileInfo
  117. }
  118. if (!this.hash) {
  119. const read = fs.createReadStream(this.temp.path)
  120. this.hash = await Hash.of(read)
  121. }
  122. this.emit('info', this.fileInfo, this.hash)
  123. return {
  124. info: this.fileInfo,
  125. hash: this.hash,
  126. }
  127. }
  128. /*
  129. * Commit this stream to the IPFS backend.
  130. */
  131. commit() {
  132. if (!this.temp) {
  133. throw new Error('Cannot commit a temporary stream that does not exist. Did you call cleanup()?')
  134. }
  135. debug('Committing temporary stream: ', this.temp.path)
  136. this.storage.ipfs
  137. .addFromFs(this.temp.path)
  138. .then(async (result) => {
  139. const hash = result[0].hash
  140. debug('Stream committed as', hash)
  141. this.emit('committed', hash)
  142. await this.storage.ipfs.pin.add(hash)
  143. this.cleanup()
  144. })
  145. .catch((err) => {
  146. debug('Error committing stream', err)
  147. this.emit('error', err)
  148. this.cleanup()
  149. })
  150. }
  151. /*
  152. * Clean up temporary data.
  153. */
  154. cleanup() {
  155. // Make it safe to call cleanup more than once
  156. if (!this.temp) return
  157. debug('Cleaning up temporary file: ', this.temp.path)
  158. fs.unlink(this.temp.path, () => {
  159. /* Ignore errors. */
  160. })
  161. delete this.temp
  162. }
  163. }
  164. /*
  165. * Manages the storage backend interaction. This provides a Promise-based API.
  166. *
  167. * Usage:
  168. *
  169. * const store = await Storage.create({ ... });
  170. * store.open(...);
  171. */
  172. class Storage {
  173. /*
  174. * Create a Storage instance. Options include:
  175. *
  176. * - an `ipfs` property, which is itself a hash containing
  177. * - `connect_options` to be passed to the IPFS client library for
  178. * connecting to an IPFS node.
  179. * - a `resolve_content_id` function, which translates Joystream
  180. * content IDs to IPFS content IDs or vice versa. The default is to
  181. * not perform any translation, which is not practical for a production
  182. * system, but serves its function during development and testing. The
  183. * function must be asynchronous.
  184. * - a `timeout` parameter, defaulting to DEFAULT_TIMEOUT. After this time,
  185. * requests to the IPFS backend time out.
  186. *
  187. * Functions in this class accept an optional timeout parameter. If the
  188. * timeout is given, it is used - otherwise, the `option.timeout` value
  189. * above is used.
  190. */
  191. static create(options) {
  192. const storage = new Storage()
  193. storage._init(options)
  194. return storage
  195. }
  196. _init(options) {
  197. this.options = _.clone(options || {})
  198. this.options.ipfs = this.options.ipfs || {}
  199. this._timeout = this.options.timeout || DEFAULT_TIMEOUT
  200. this._resolve_content_id = this.options.resolve_content_id || DEFAULT_RESOLVE_CONTENT_ID
  201. this.ipfs = ipfsClient(this.options.ipfsHost || 'localhost', '5001', { protocol: 'http' })
  202. this.pinned = {}
  203. this.pinning = {}
  204. this.ipfs.id((err, identity) => {
  205. if (err) {
  206. debug(`Warning IPFS daemon not running: ${err.message}`)
  207. } else {
  208. debug(`IPFS node is up with identity: ${identity.id}`)
  209. // TODO: wait for IPFS daemon to be online for this to be effective..?
  210. // set the IPFS HTTP Gateway config we desire.. operator might need
  211. // to restart their daemon if the config was changed.
  212. this.ipfs.config.set('Gateway.PublicGateways', { 'localhost': null })
  213. }
  214. })
  215. }
  216. /*
  217. * Uses bluebird's timeout mechanism to return a Promise that times out after
  218. * the given timeout interval, and tries to execute the given operation within
  219. * that time.
  220. */
  221. async withSpecifiedTimeout(timeout, operation) {
  222. // TODO: rewrite this method to async-await style
  223. // eslint-disable-next-line no-async-promise-executor
  224. return new Promise(async (resolve, reject) => {
  225. try {
  226. resolve(await new Promise(operation))
  227. } catch (err) {
  228. reject(err)
  229. }
  230. }).timeout(timeout || this._timeout)
  231. }
  232. /*
  233. * Resolve content ID with timeout.
  234. */
  235. async resolveContentIdWithTimeout(timeout, contentId) {
  236. return await this.withSpecifiedTimeout(timeout, async (resolve, reject) => {
  237. try {
  238. resolve(await this._resolve_content_id(contentId))
  239. } catch (err) {
  240. reject(err)
  241. }
  242. })
  243. }
  244. /*
  245. * Stat a content ID.
  246. */
  247. async stat(contentId, timeout) {
  248. const resolved = await this.resolveContentIdWithTimeout(timeout, contentId)
  249. return await this.withSpecifiedTimeout(timeout, (resolve, reject) => {
  250. this.ipfs.files.stat(`/ipfs/${resolved}`, { withLocal: true }, (err, res) => {
  251. if (err) {
  252. reject(err)
  253. return
  254. }
  255. resolve(res)
  256. })
  257. })
  258. }
  259. /*
  260. * Return the size of a content ID.
  261. */
  262. async size(contentId, timeout) {
  263. const stat = await this.stat(contentId, timeout)
  264. return stat.size
  265. }
  266. /*
  267. * Opens the specified content in read or write mode, and returns a Promise
  268. * with the stream.
  269. *
  270. * Read streams will contain a fileInfo property, with:
  271. * - a `mimeType` field providing the file's MIME type, or a default.
  272. * - an `ext` property, providing a file extension suggestion, or a default.
  273. *
  274. * Write streams have a slightly different flow, in order to allow for MIME
  275. * type detection and potential filtering. First off, they are written to a
  276. * temporary location, and only committed to the backend once their
  277. * `commit()` function is called.
  278. *
  279. * When the commit has finished, a `committed` event is emitted, which
  280. * contains the IPFS backend's content ID.
  281. *
  282. * Write streams also emit a `fileInfo` event during writing. It is passed
  283. * the `fileInfo` field as described above. Event listeners may now opt to
  284. * abort the write or continue and eventually `commit()` the file. There is
  285. * an explicit `cleanup()` function that removes temporary files as well,
  286. * in case comitting is not desired.
  287. */
  288. async open(contentId, mode, timeout) {
  289. if (mode !== 'r' && mode !== 'w') {
  290. throw Error('The only supported modes are "r", "w" and "a".')
  291. }
  292. // Write stream
  293. if (mode === 'w') {
  294. return await this.createWriteStream(contentId, timeout)
  295. }
  296. // Read stream - with file type detection
  297. return await this.createReadStream(contentId, timeout)
  298. }
  299. async createWriteStream() {
  300. // IPFS wants us to just dump a stream into its storage, then returns a
  301. // content ID (of its own).
  302. // We need to instead return a stream immediately, that we eventually
  303. // decorate with the content ID when that's available.
  304. return new Promise((resolve) => {
  305. const stream = new StorageWriteStream(this)
  306. resolve(stream)
  307. })
  308. }
  309. async createReadStream(contentId, timeout) {
  310. const resolved = await this.resolveContentIdWithTimeout(timeout, contentId)
  311. let found = false
  312. return await this.withSpecifiedTimeout(timeout, (resolve, reject) => {
  313. const ls = this.ipfs.getReadableStream(resolved)
  314. ls.on('data', async (result) => {
  315. if (result.path === resolved) {
  316. found = true
  317. const ftStream = await fileType.stream(result.content)
  318. resolve(fixFileInfoOnStream(ftStream))
  319. }
  320. })
  321. ls.on('error', (err) => {
  322. ls.end()
  323. debug(err)
  324. reject(err)
  325. })
  326. ls.on('end', () => {
  327. if (!found) {
  328. const err = new Error('No matching content found for', contentId)
  329. debug(err)
  330. reject(err)
  331. }
  332. })
  333. ls.resume()
  334. })
  335. }
  336. /*
  337. * Synchronize the given content ID
  338. */
  339. async synchronize(contentId, callback) {
  340. const resolved = await this.resolveContentIdWithTimeout(this._timeout, contentId)
  341. // TODO: validate resolved id is proper ipfs_cid, not null or empty string
  342. if (!this.pinning[resolved] && !this.pinned[resolved]) {
  343. debug(`Pinning hash: ${resolved} content-id: ${contentId}`)
  344. this.pinning[resolved] = true
  345. // Callback passed to add() will be called on error or when the entire file
  346. // is retrieved. So on success we consider the content synced.
  347. this.ipfs.pin.add(resolved, { quiet: true, pin: true }, (err) => {
  348. delete this.pinning[resolved]
  349. if (err) {
  350. debug(`Error Pinning: ${resolved}`)
  351. callback && callback(err)
  352. } else {
  353. debug(`Pinned ${resolved}`)
  354. this.pinned[resolved] = true
  355. callback && callback(null, this.syncStatus(resolved))
  356. }
  357. })
  358. } else {
  359. callback && callback(null, this.syncStatus(resolved))
  360. }
  361. }
  362. syncStatus(ipfsHash) {
  363. return {
  364. syncing: this.pinning[ipfsHash] === true,
  365. synced: this.pinned[ipfsHash] === true,
  366. }
  367. }
  368. }
  369. module.exports = {
  370. Storage,
  371. }