initial commit

.dockerignore

@@ -0,0 +1,5 @@
+.git
+.vscode
+**/node_modules
+.DS_Store
+.data

.env.prod

@@ -0,0 +1,3 @@
+ENV=prod
+REDIS_HOST=redis
+REDIS_PORT=6379

.gitignore

@@ -0,0 +1,24 @@
+# See https://help.github.com/ignore-files/ for more about ignoring files.
+
+# dependencies
+/node_modules
+
+# testing
+/coverage
+
+# production
+/build
+
+# misc
+.DS_Store
+.env
+.env.production
+.env.local
+.env.development.local
+.env.test.local
+.env.production.local
+
+npm-debug.log*
+yarn-debug.log*
+yarn-error.log*
+*.log

.vscode/launch.json

@@ -0,0 +1,23 @@
+{
+  // Use IntelliSense to learn about possible attributes.
+  // Hover to view descriptions of existing attributes.
+  // For more information, visit: https://go.microsoft.com/fwlink/?linkid=830387
+  "version": "0.2.0",
+  "configurations": [
+    {
+      "type": "node",
+      "request": "attach",
+      "name": "Attach to Remote",
+      "address": "localhost",
+      "port": 9229,
+      "protocol": "inspector",
+      "localRoot": "${workspaceFolder}",
+      "remoteRoot": "/usr/app",
+      "restart": true,
+      "skipFiles": [
+        "<node_internals>/**"
+      ]
+    }
+
+  ]
+}

LICENSE

@@ -0,0 +1,24 @@
+This is free and unencumbered software released into the public domain.
+
+Anyone is free to copy, modify, publish, use, compile, sell, or
+distribute this software, either in source code form or as a compiled
+binary, for any purpose, commercial or non-commercial, and by any
+means.
+
+In jurisdictions that recognize copyright laws, the author or authors
+of this software dedicate any and all copyright interest in the
+software to the public domain. We make this dedication for the benefit
+of the public at large and to the detriment of our heirs and
+successors. We intend this dedication to be an overt act of
+relinquishment in perpetuity of all present and future rights to this
+software under copyright law.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
+EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
+MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.
+IN NO EVENT SHALL THE AUTHORS BE LIABLE FOR ANY CLAIM, DAMAGES OR
+OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE,
+ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR
+OTHER DEALINGS IN THE SOFTWARE.
+
+For more information, please refer to <http://unlicense.org/>

README.md

@@ -0,0 +1,7 @@
+# palomitas-transcoder
+
+This project uses Docker and docker-compose to manage the parts of the application.
+
+Run `docker-compose build` once to build the docker images and then run `docker-compose up` to start the redis container and the node container.
+
+You can start a development version that listens for changes in the files with the command `docker-compose -f docker-compose.dev.yml`. In order for this to work, a `.env` file must be located in the root of the project with at least the content of the `.env.prod` file.

api/Dockerfile

@@ -0,0 +1,12 @@
+FROM node:10-alpine
+
+RUN apk --no-cache add ffmpeg git python make gcc g++
+WORKDIR /usr/app
+
+ENV PATH /usr/app/node_modules/.bin:$PATH
+ENV NODE_ENV production
+COPY package*.json /usr/app/
+RUN npm ci
+COPY . /usr/app
+
+CMD [ "npm", "start" ]

api/errorHandler.js

@@ -0,0 +1,33 @@
+// global error handler
+const errorHandler = (err, req, res, next) => {
+  if (typeof err === 'string') {
+    err = { message: err }
+  }
+  if (process.env.ENV === 'dev') {
+    console.error(err)
+  } else {
+    console.error(err.message)
+    err.message = "Internal server error"
+  }
+  const status = isNaN(err.code) ? 500 : err.code
+  res.status(status).json({ code: err.code, error: err.message })
+}
+
+function wrapAsync (fn) {
+  return async function wrappedAsyncHandler (req, res, next) {
+    try {
+      await fn(req, res, next)
+    } catch (err) {
+      errorHandler(err, req, res, next)
+    }
+  }
+}
+
+class ErrorWithCode extends Error {
+  constructor (code, message) {
+    super(message)
+    this.code = code
+  }
+}
+
+module.exports = { errorHandler, wrapAsync, ErrorWithCode }

api/index.js

@@ -0,0 +1,163 @@
+require('dotenv').config()
+const express = require('express')
+const cors = require('cors')
+const helmet = require('helmet')
+const logger = require('morgan')
+const bodyParser = require('body-parser')
+const pkg = require('./package.json')
+const { wrapAsync, ErrorWithCode } = require('./errorHandler')
+const Queue = require('bull')
+
+const app = express()
+
+app.set('json spaces', 2)
+app.use(cors())
+app.use(helmet())
+app.use(logger('tiny'))
+app.use(bodyParser.json())
+
+const videoQueue = new Queue('video transcoding', {
+  redis: {
+    port: process.env.REDIS_PORT,
+    host: process.env.REDIS_HOST
+  }
+})
+
+// STEPS:
+// 1. Connect to the video url and create a pipe that handles the downloading
+// 2. Plug that pipe into the ffmpeg command
+// 3. Get ffmpeg progress, error and complete events and send them to the queue
+// 4. Save result of transcoding into a file
+const ffmpeg = require('fluent-ffmpeg')
+const axios = require('axios')
+const tempy = require('tempy')
+const del = require('del')
+const fs = require('fs')
+const ffprobe = require('node-ffprobe')
+
+videoQueue.process(async (job, done) => {
+  const url = job.data.url
+  job.log(`starting job for resource ${url}`)
+
+  try {
+    const res = await axios({ url, method: 'get', responseType: 'stream' })
+    const tempPath = tempy.file()
+    job.log(`[tempy] Created temporary file in ${tempPath}`)
+    res.data.pipe(fs.createWriteStream(tempPath))
+    ffmpegCommand(tempPath, job, done)
+  } catch (err) {
+    console.error('error processing video', err)
+    done(err)
+  }
+})
+
+function parseDuration (timeStr) {
+  const [h, m, s] = timeStr.split(':').map(Number)
+  return (h * 60 * 60) + (m * 60) + s
+}
+
+function ffmpegCommand (path, job, done) {
+  let duration = 0
+  ffmpeg(path)
+    .videoCodec('libvpx')
+    .audioCodec('libvorbis')
+    .format('webm')
+    // .audioBitrate(128)
+    // .videoBitrate(1024)
+    .outputOptions([
+      '-pix_fmt yuv420p',
+      '-color_primaries 1',
+      '-color_trc 1',
+      '-colorspace 1',
+      '-movflags +faststart',
+      '-quality good',
+      '-crf 32'
+    ])
+    .on('start', cmd => {
+      job.log(`[ffmpeg] ${cmd}`)
+    })
+    .on('error', err => {
+      job.log(`[ffmpeg] ${err}`)
+      done(err)
+    })
+    .on('codecData', data => {
+      job.log(`[codec-data] ${JSON.stringify(data)}`)
+      duration = parseDuration(data.duration)
+    })
+    .on('progress', (progress) => {
+      job.log(`[ffmpeg] progress: ${JSON.stringify(progress)}`)
+      const time = parseDuration(progress.timemark)
+      const percent = duration && (time / duration) * 100
+      job.progress(percent)
+    })
+    .on('end', async () => {
+      const deleted = await del([path])
+      job.log(`[del] Deleted temporary file in ${deleted}`)
+      done()
+    })
+    .saveToFile('./output.webm')
+}
+
+app.get('/', (req, res) => {
+  res.json({
+    name: pkg.name,
+    version: pkg.version,
+    descriptipn: pkg.descriptipn,
+    endpoints: [
+      'GET /counts',
+      'GET /jobs?status=&start=&end=&asc=',
+      'GET /jobs/:id',
+      'GET /jobs/:id/logs?start=&end=',
+      'POST /job { url }',
+      'DELETE /jobs?grace=1000&status=&limit='
+    ]
+  })
+})  
+
+app.get('/counts', wrapAsync(async (req, res) => {
+  const counts = await videoQueue.getJobCounts()
+  res.json(counts)
+}))
+
+app.get('/jobs', wrapAsync(async (req, res) => {
+  const statuses = (req.query.status || '').split(',')
+  const start = req.query.start // first index to fetch for every state
+  const end = req.query.end     // last index to fetch for every state
+  const asc = req.query.asc     // sort ascending or descending
+  const jobs = await videoQueue.getJobs(statuses, start, end, asc)
+  res.json(jobs)
+}))
+
+app.get('/jobs/:id', wrapAsync(async (req, res) => {
+  const job = await videoQueue.getJob(req.params.id)
+  res.json({ job })
+}))
+
+app.get('/jobs/:id/logs', wrapAsync(async (req, res) => {
+  const start = req.query.start // first index to fetch
+  const end = req.query.end     // last index to fetch
+  const logs = await videoQueue.getJobLogs(req.params.id, start, end)
+  res.json(logs)
+}))
+
+app.post('/jobs', wrapAsync(async (req, res) => {
+  const url = req.body.url
+  if (!url) {
+    throw new ErrorWithCode(400, 'Failed to create job. Invalid URL param')
+  }
+  const job = await videoQueue.add({ time: Date.now(), url: req.body.url })
+  res.json({ message: 'job added', job })
+}))
+
+app.delete('/jobs', wrapAsync(async (req, res) => {
+  const gracePeriod = req.query.grace || 1000
+  const status = req.query.status
+  const limit = req.query.limit
+  const ids = await videoQueue.clean(gracePeriod, status, limit)
+  res.json({ message: `cleaned ${ids.length} jobs`, deleted_ids: ids })
+}))
+
+const port = process.env.PORT || 4000
+app.listen(port, () => {
+  console.log(`> app is listening at port ${port}`)
+})