Creator Node Track Upload Improvements

creator-node/package.json

@@ -32,7 +32,7 @@
  "ffmpeg-static": "^2.4.0",
  "ffprobe-static": "^3.0.0",
  "ioredis": "^4.9.3",
- "ipfs-api": "^26.1.2",
+ "ipfs-http-client": "^33.1.1",
  "multer": "^1.4.0",
  "pg": "^7.6.1",
  "rate-limit-redis": "^1.6.0",

creator-node/src/ffprobe.js

@@ -6,9 +6,10 @@ var spawn = require('child_process').spawn
 
 async function getTrackDuration (fileDir) {
  try {
- const info = await getInfo(fileDir, { path: ffprobeStatic.path })
- // TODO - data validation
- return info.streams[0].duration
+ const resp = await getInfo(fileDir, { path: ffprobeStatic.path })
+ const duration = Number(resp.streams[0].duration)
+ if (isNaN(duration)) throw new Error(`Invalid return value from FFProbe: ${duration}`)
+ return duration
  } catch (e) {
  // If the error is the text below, it means the segment doesn't have any
  // data. In that case, just return null so we skip adding the segment

creator-node/src/fileManager.js

@@ -15,26 +15,27 @@ const maxMemoryFileSize = parseInt(config.get('maxMemoryFileSizeBytes')) // Defa
 const ALLOWED_UPLOAD_FILE_EXTENSIONS = config.get('allowedUploadFileExtensions') // default set in config.json
 const AUDIO_MIME_TYPE_REGEX = /audio\/(.*)/
 
-/** (1) Add file to IPFS; (2) save file to disk;
- * (3) pin file via IPFS; (4) save file ref to DB
+/**
+ * (1) Add file to IPFS; (2) save file to disk;
+ * (3) pin file via IPFS; (4) save file ref to DB
+ * @dev - only call this function when file is not already stored to disk
+ * - if it is, then use saveFileToIPFSFromFS()
  */
-async function saveFile (req, buffer) {
+async function saveFileFromBuffer (req, buffer) {
  // make sure user has authenticated before saving file
  if (!req.userId) {
  throw new Error('User must be authenticated to save a file')
  }
 
  const ipfs = req.app.get('ipfsAPI')
 
- let multihash = await ipfs.files.add(buffer, { onlyHash: true })
- multihash = multihash[0].hash
+ const multihash = (await ipfs.add(buffer))[0].hash
 
- const fileLocation = path.join(req.app.get('storagePath'), '/' + multihash)
- await writeFile(fileLocation, buffer)
+ const dstPath = path.join(req.app.get('storagePath'), multihash)
 
-// TODO(roneilr): switch to using the IPFS filestore below to avoid duplicating content
- const filesAdded = await ipfs.files.add(buffer)
-assert.strictEqual(multihash, filesAdded[0].hash)
+await writeFile(dstPath, buffer)
+
+// TODO: switch to using the IPFS filestore below to avoid duplicating content
  await ipfs.pin.add(multihash)
 
  // add reference to file to database
@@ -43,7 +44,7 @@ async function saveFile (req, buffer) {
  cnodeUserUUID: req.userId,
  multihash: multihash,
  sourceFile: req.fileName,
- storagePath: fileLocation
+ storagePath: dstPath
  }
  })
 
@@ -53,6 +54,41 @@ async function saveFile (req, buffer) {
  return { multihash: multihash, fileUUID: file.fileUUID }
 }
 
+/**
+ * Save file to IPFS given file path.
+ * - Add and pin file to IPFS.
+ * - Re-save file to disk under multihash.
+ * - Save reference to file in DB.
+ */
+async function saveFileToIPFSFromFS (req, srcPath) {
+ // make sure user has authenticated before saving file
+ if (!req.userId) throw new Error('User must be authenticated to save a file')
+
+ const ipfs = req.app.get('ipfsAPI')
+
+ const multihash = (await ipfs.addFromFs(srcPath))[0].hash
+ const dstPath = path.join(req.app.get('storagePath'), multihash)
+
+ // store segment file copy under multihash for easy future retrieval
+ fs.copyFileSync(srcPath, dstPath)
+
+ // TODO: switch to using the IPFS filestore below to avoid duplicating content
+ await ipfs.pin.add(multihash)
+
+ // add reference to file to database
+ const file = (await models.File.findOrCreate({ where:
+ {
+ cnodeUserUUID: req.userId,
+ multihash: multihash,
+ sourceFile: req.fileName,
+ storagePath: dstPath
+ }
+ }))[0].dataValues
+
+ req.logger.info(`\nAdded file: ${multihash} for fileUUID ${file.fileUUID} from sourceFile ${req.fileName}`)
+ return { multihash: multihash, fileUUID: file.fileUUID }
+}
+
 /** Save file to disk given IPFS multihash, and ensure is pinned.
  * Steps:
  * - If file already stored on disk, return immediately.
@@ -168,7 +204,7 @@ const trackDiskStorage = multer.diskStorage({
  destination: function (req, file, cb) {
  // save file under randomly named folders to avoid collisions
  const randomFileName = getUuid()
- const fileDir = req.app.get('storagePath') + '/' + randomFileName
+ const fileDir = path.join(req.app.get('storagePath'), randomFileName)
 
  // create directories for original file and segments
  fs.mkdirSync(fileDir)
@@ -207,4 +243,4 @@ function getFileExtension (fileName) {
  return (fileName.lastIndexOf('.') >= 0) ? fileName.substr(fileName.lastIndexOf('.')) : ''
 }
 
-module.exports = { saveFile, saveFileForMultihash, removeTrackFolder, upload, trackFileUpload }
+module.exports = { saveFileFromBuffer, saveFileToIPFSFromFS, saveFileForMultihash, removeTrackFolder, upload, trackFileUpload }

creator-node/src/index.js

@@ -2,7 +2,7 @@
 
 const S3 = require('aws-sdk').S3
 const ON_DEATH = require('death')
-const ipfsAPI = require('ipfs-api')
+const ipfsClient = require('ipfs-http-client')
 const path = require('path')
 
 const initializeApp = require('./app')
@@ -53,7 +53,7 @@ if (!ipfsAddr) {
  logger.error('Must set ipfsAddr')
  process.exit(1)
 }
-let ipfs = ipfsAPI(ipfsAddr, config.get('ipfsPort'))
+let ipfs = ipfsClient(ipfsAddr, config.get('ipfsPort'))
 
 // run all migrations
 logger.info('Executing database migrations...')

creator-node/src/redis.js

@@ -28,7 +28,6 @@ module.exports.lock = RedisLock
 
 /** Ensure resource write access */
 async function nodeSyncMiddleware (req, res, next) {
- req.logger.info('before nodeysnc middleware')
  if (req.session && req.session.wallet) {
  const redisKey = getNodeSyncRedisKey(req.session.wallet)
  const lockHeld = await RedisLock.getLock(redisKey)
@@ -38,7 +37,6 @@ async function nodeSyncMiddleware (req, res, next) {
  ))
  }
  }
- req.logger.info('after nodeysnc middleware')
  next()
 }
 

creator-node/src/routes/audiusUsers.js

@@ -1,19 +1,19 @@
+const { Buffer } = require('ipfs-http-client')
+
 const models = require('../models')
 const authMiddleware = require('../authMiddleware')
 const nodeSyncMiddleware = require('../redis').nodeSyncMiddleware
-const { saveFile } = require('../fileManager')
+const { saveFileFromBuffer } = require('../fileManager')
 const { handleResponse, successResponse, errorResponseBadRequest } = require('../apiHelpers')
 
 module.exports = function (app) {
  // create AudiusUser from provided metadata, and make metadata available to network
  app.post('/audius_users', authMiddleware, nodeSyncMiddleware, handleResponse(async (req, res) => {
- const ipfs = req.app.get('ipfsAPI')
-
- // TODO(roneilr): do some validation on metadata given
+ // TODO: do some validation on metadata given
  const metadataJSON = req.body
 
- const metadataBuffer = ipfs.types.Buffer.from(JSON.stringify(metadataJSON))
- const { multihash, fileUUID } = await saveFile(req, metadataBuffer)
+ const metadataBuffer = Buffer.from(JSON.stringify(metadataJSON))
+ const { multihash, fileUUID } = await saveFileFromBuffer(req, metadataBuffer)
 
  const audiusUserObj = {
  cnodeUserUUID: req.userId,
@@ -58,7 +58,6 @@ module.exports = function (app) {
 
  // update a AudiusUser
  app.put('/audius_users/:blockchainId', authMiddleware, nodeSyncMiddleware, handleResponse(async (req, res) => {
- const ipfs = req.app.get('ipfsAPI')
  const blockchainId = req.params.blockchainId
  const audiusUser = await models.AudiusUser.findOne({ where: { blockchainId, cnodeUserUUID: req.userId } })
 
@@ -67,13 +66,12 @@ module.exports = function (app) {
  return errorResponseBadRequest(`Audius User doesn't exist for that blockchainId`)
  }
 
- // TODO(roneilr, dmanjunath): do some validation on metadata given
+ // TODO: do some validation on metadata given
  const metadataJSON = req.body
-
- const metadataBuffer = ipfs.types.Buffer.from(JSON.stringify(metadataJSON))
+ const metadataBuffer = Buffer.from(JSON.stringify(metadataJSON))
 
  // write to a new file so there's still a record of the old file
- const { multihash, fileUUID } = await saveFile(req, metadataBuffer)
+ const { multihash, fileUUID } = await saveFileFromBuffer(req, metadataBuffer)
 
  // Update the file to the new fileId and write the metadata blob in the json field
  let updateObj = {

creator-node/src/routes/files.js

@@ -1,4 +1,6 @@
-const { saveFile, upload } = require('../fileManager')
+const { Buffer } = require('ipfs-http-client')
+
+const { saveFileFromBuffer, upload } = require('../fileManager')
 const { handleResponse, sendResponse, successResponse, errorResponseBadRequest, errorResponseServerError, errorResponseNotFound } = require('../apiHelpers')
 
 const models = require('../models')
@@ -10,19 +12,19 @@ let Redis = require('ioredis')
 let client = new Redis(config.get('redisPort'), config.get('redisHost'))
 
 module.exports = function (app) {
- // upload image file and make avail
- // TODO(ss) - input validation
+ /** Store image on disk + DB and make available via IPFS */
  app.post('/image_upload', authMiddleware, nodeSyncMiddleware, upload.single('file'), handleResponse(async (req, res) => {
- const { multihash } = await saveFile(req, req.file.buffer)
+ // TODO: input validation
+ // TODO: switch to saveFileToIPFSFromFS
+ const { multihash } = await saveFileFromBuffer(req, req.file.buffer)
  return successResponse({ 'image_file_multihash': multihash })
  }))
 
- // upload metadata to IPFS and save in Files table
+ /** upload metadata to IPFS and save in Files table */
  app.post('/metadata', authMiddleware, nodeSyncMiddleware, handleResponse(async (req, res) => {
- const ipfs = req.app.get('ipfsAPI')
  const metadataJSON = req.body
- const metadataBuffer = ipfs.types.Buffer.from(JSON.stringify(metadataJSON))
- const { multihash } = await saveFile(req, metadataBuffer)
+ const metadataBuffer = Buffer.from(JSON.stringify(metadataJSON))
+ const { multihash } = await saveFileFromBuffer(req, metadataBuffer)
  return successResponse({ 'metadataMultihash': multihash })
  }))
 

creator-node/src/routes/tracks.js

@@ -1,22 +1,25 @@
-const fs = require('fs')
+const path = require('path')
+const { Buffer } = require('ipfs-http-client')
+
 const ffmpeg = require('../ffmpeg')
 const ffprobe = require('../ffprobe')
 
 const models = require('../models')
 const authMiddleware = require('../authMiddleware')
 const nodeSyncMiddleware = require('../redis').nodeSyncMiddleware
-const { saveFile, removeTrackFolder, trackFileUpload } = require('../fileManager')
+const { saveFileFromBuffer, saveFileToIPFSFromFS, removeTrackFolder, trackFileUpload } = require('../fileManager')
 const { handleResponse, successResponse, errorResponseBadRequest, errorResponseServerError } = require('../apiHelpers')
 
 module.exports = function (app) {
- // upload track segment files and make avail - will later be associated with Audius track
+ /**
+ * upload track segment files and make avail - will later be associated with Audius track
+ * @dev - currently stores each segment twice, once under random file UUID & once under IPFS multihash
+ * - this should be addressed eventually
+ */
  app.post('/track_content', authMiddleware, nodeSyncMiddleware, trackFileUpload.single('file'), handleResponse(async (req, res) => {
- if (req.fileFilterError) {
- // POST body is not a valid file type
- return errorResponseBadRequest(req.fileFilterError)
- }
+ if (req.fileFilterError) return errorResponseBadRequest(req.fileFilterError)
 
- // create and save segments to disk
+ // create and save track file segments to disk
  let segmentFilePaths
  try {
  segmentFilePaths = await ffmpeg.segmentFile(req, req.fileDir, req.fileName)
@@ -25,26 +28,35 @@ module.exports = function (app) {
  return errorResponseServerError(err)
  }
 
- // for each path, read file into buffer and pass to saveFile
- const files = []
- for (let path of segmentFilePaths) {
- let absolutePath = req.fileDir + '/segments/' + path
- let fileBuffer = fs.readFileSync(absolutePath)
- let { multihash } = await saveFile(req, fileBuffer)
- const duration = await ffprobe.getTrackDuration(absolutePath)
- if (duration) files.push({ 'multihash': multihash, duration: duration })
+ // for each path, call saveFile and get back multihash; return multihash + segment duration
+ // run all async ops in parallel as they are not independent
+ let saveFileProms = []
+ let durationProms = []
+ for (let filePath of segmentFilePaths) {
+ const absolutePath = path.join(req.fileDir, 'segments', filePath)
+ const saveFileProm = saveFileToIPFSFromFS(req, absolutePath)
+ const durationProm = ffprobe.getTrackDuration(absolutePath)
+ saveFileProms.push(saveFileProm)
+ durationProms.push(durationProm)
  }
+ // Resolve all promises + process responses
+ const [saveFilePromResps, durationPromResps] = await Promise.all(
+ [saveFileProms, durationProms].map(promiseArray => Promise.all(promiseArray))
+ )
+ let trackSegments = saveFilePromResps.map((saveFileResp, i) => {
+ return { 'multihash': saveFileResp.multihash, 'duration': durationPromResps[i] }
+ })
+ // exclude 0-length segments that are sometimes outputted by ffmpeg segmentation
+ trackSegments = trackSegments.filter(trackSegment => trackSegment.duration)
 
- return successResponse({ 'track_segments': files })
+ return successResponse({ 'track_segments': trackSegments })
  }))
 
  /** given track metadata object, create track and share track metadata with network
  * - return on success: temporary ID of track
  * - return on failure: error if linked segments have not already been created via POST /track_content
  */
  app.post('/tracks', authMiddleware, nodeSyncMiddleware, handleResponse(async (req, res) => {
- const ipfs = req.app.get('ipfsAPI')
-
  // TODO - input validation
  const metadataJSON = req.body
 
@@ -78,8 +90,8 @@ module.exports = function (app) {
  }
 
  // store metadata multihash
- const metadataBuffer = ipfs.types.Buffer.from(JSON.stringify(metadataJSON))
- const { multihash, fileUUID } = await saveFile(req, metadataBuffer)
+ const metadataBuffer = Buffer.from(JSON.stringify(metadataJSON))
+ const { multihash, fileUUID } = await saveFileFromBuffer(req, metadataBuffer)
 
  // build track object for db storage
  const trackObj = {
@@ -131,7 +143,7 @@ module.exports = function (app) {
  return errorResponseBadRequest('Invalid track ID')
  }
 
- // TODO(roneilr): validate that provided blockchain ID is indeed associated with
+ // TODO: validate that provided blockchain ID is indeed associated with
  // user wallet and metadata CID
  await track.update({
  blockchainId: blockchainId
@@ -152,21 +164,19 @@ module.exports = function (app) {
 
  // update a track
  app.put('/tracks/:blockchainId', authMiddleware, nodeSyncMiddleware, handleResponse(async (req, res) => {
- const ipfs = req.app.get('ipfsAPI')
  const blockchainId = req.params.blockchainId
  const cnodeUserUUID = req.userId
 
  const track = await models.Track.findOne({ where: { blockchainId, cnodeUserUUID } })
 
  if (!track) return errorResponseBadRequest(`Could not find track with id ${blockchainId} owned by calling user`)
 
- // TODO(roneilr, dmanjunath): do some validation on metadata given
+ // TODO: do some validation on metadata given
  const metadataJSON = req.body
-
- const metadataBuffer = ipfs.types.Buffer.from(JSON.stringify(metadataJSON))
+ const metadataBuffer = Buffer.from(JSON.stringify(metadataJSON))
 
  // write to a new file so there's still a record of the old file
- const { multihash, fileUUID } = await saveFile(req, metadataBuffer)
+ const { multihash, fileUUID } = await saveFileFromBuffer(req, metadataBuffer)
 
  const coverArtFileMultihash = metadataJSON.cover_art
  let coverArtFileUUID = null

creator-node/test/audiusUsers.js

@@ -21,7 +21,7 @@ describe('test AudiusUsers', function () {
  const metadata = {
  test: 'field1'
  }
- ipfsMock.files.add.twice().withArgs(Buffer.from(JSON.stringify(metadata)))
+ ipfsMock.add.twice().withArgs(Buffer.from(JSON.stringify(metadata)))
  ipfsMock.pin.add.once().withArgs('testCIDLink')
 
  request(app)
@@ -44,7 +44,7 @@ describe('test AudiusUsers', function () {
  const metadata = {
  test: 'field1'
  }
- ipfsMock.files.add.twice().withArgs(Buffer.from(JSON.stringify(metadata)))
+ ipfsMock.add.twice().withArgs(Buffer.from(JSON.stringify(metadata)))
  ipfsMock.pin.add.once().withArgs('testCIDLink')
 
  request(app)