feat: clustered ffmpeg transcoding

.env.development

@@ -82,6 +82,11 @@ IMGPROXY_ENABLE_VIDEO_THUMBNAILS=1
 # IMGPROXY_DEVELOPMENT_ERRORS_MODE=1
 # IMGPROXY_ENABLE_DEBUG_HEADERS=true
 
+# ffmpeg
+NEXT_PUBLIC_FFMPEG_URL=http://localhost:3002
+FFMPEG_PORT=3002
+FFMPEG_WORKERS=4
+
 NEXT_PUBLIC_AWS_UPLOAD_BUCKET=uploads
 NEXT_PUBLIC_MEDIA_DOMAIN=localhost:4566
 NEXT_PUBLIC_MEDIA_URL=http://localhost:4566/uploads

components/file-upload.js

@@ -28,6 +28,7 @@ export const FileUpload = forwardRef(({ children, className, onSelect, onUpload,
     return new Promise((resolve, reject) => {
       async function onload () {
         onUpload?.(file)
+        file = await transcodeFFMPEG(file) // will transcode if video
         let data
         const variables = {
           avatar,
@@ -126,6 +127,23 @@ export const FileUpload = forwardRef(({ children, className, onSelect, onUpload,
   )
 })
 
+const transcodeFFMPEG = async (file) => {
+  if (!file || !file.type.startsWith('video/')) return file
+  const formData = new FormData()
+  formData.append('file', file)
+  const res = await fetch(process.env.NEXT_PUBLIC_FFMPEG_URL + '/api/transcode', {
+    method: 'POST',
+    body: formData
+  })
+
+  if (!res.ok) {
+    throw new Error('Failed to contact FFMPEG service')
+  }
+
+  const processedFile = await res.blob()
+  return new File([processedFile], file.name, { type: file.type })
+}
+
 // from https://stackoverflow.com/a/77472484
 const removeExifData = async (file) => {
   if (!file || !file.type.startsWith('image/')) return file

docker-compose.yml

@@ -653,6 +653,17 @@ services:
       CONNECT: "localhost:${LNBITS_WEB_PORT}"
       TORDIR: "/app/.tor"
     cpu_shares: "${CPU_SHARES_LOW}"
+  ffmpeg:
+    build:
+      context: ./docker/ffmpeg
+    container_name: ffmpeg
+    env_file: *env_file
+    profiles:
+      - images
+    volumes:
+      - uploads:/app/uploads
+    ports:
+      - "3002:3002"
 volumes:
   db:
   os:
@@ -664,3 +675,4 @@ volumes:
   nwc_send:
   nwc_recv:
   tordata:
+  uploads:

docker/ffmpeg/Dockerfile

@@ -0,0 +1,15 @@
+FROM node:18.20.4-bullseye
+
+# Install ffmpeg
+RUN apt-get update -y && apt-get install --no-install-recommends -y ffmpeg && apt-get clean
+
+WORKDIR /app
+
+COPY package*.json ./
+RUN npm install
+
+COPY . .
+
+EXPOSE 3002
+
+CMD ["node", "server.js"]

docker/ffmpeg/package.json

@@ -0,0 +1,12 @@
+{
+    "name": "ffmpeg-service",
+    "version": "1.0.0",
+    "main": "server.js",
+    "scripts": {
+      "start": "node server.js"
+    },
+    "dependencies": {
+      "express": "^4.21.1",
+      "multer": "^1.4.5-lts.1"
+    }
+  }

docker/ffmpeg/server.js

@@ -0,0 +1,168 @@
+// TODO add comments
+const cluster = require('cluster')
+const express = require('express')
+const multer = require('multer')
+const { exec } = require('child_process')
+const { promisify } = require('util')
+const path = require('path')
+const fs = require('fs')
+
+const execAsync = promisify(exec)
+const unlinkAsync = promisify(fs.unlink)
+const WORKERS = process.env.FFMPEG_WORKERS || 2
+
+const FFMPEG_PRESET = 'slow'
+const DEFAULT_CRF = 25
+const DEFAULT_AUDIO_BITRATE = '96k'
+
+if (cluster.isMaster) {
+  console.log(`Server ${process.pid} is listening`)
+
+  const jobQueue = []
+  const workersStatus = {}
+  const jobResponses = {}
+
+  // spawn workers
+  for (let i = 0; i < WORKERS; i++) {
+    const worker = cluster.fork()
+    workersStatus[worker.id] = 'IDLE' // signal that worker is available
+  }
+
+  // replace dead workers
+  cluster.on('exit', (worker) => {
+    console.log(`Worker ${worker.process.pid} is dead`)
+    const newWorker = cluster.fork()
+    workersStatus[newWorker.id] = 'IDLE'
+  })
+
+  // assign job to worker
+  const assignJob = () => {
+    const idleWorkerId = Object.keys(workersStatus).find((id) => workersStatus[id] === 'IDLE')
+    if (idleWorkerId && jobQueue.length > 0) {
+      const job = jobQueue.shift()
+      workersStatus[idleWorkerId] = 'BUSY'
+      cluster.workers[idleWorkerId].send(job)
+    }
+  }
+
+  // initialize express http server
+  const app = express()
+  const storage = multer.diskStorage({ // set storage options
+    destination: 'uploads/', // set working path
+    filename: (req, file, cb) => { // sanitize file name
+      const sanitizedFileName = file.originalname.replace(/[^a-z0-9_\-.]/gi, '_')
+      cb(null, sanitizedFileName)
+    }
+  })
+  const upload = multer({
+    storage,
+    fileFilter: (req, file, cb) => {
+      if (!file.mimetype.startsWith('video/')) { // double check if video
+        return cb(new Error('Invalid file type'))
+      }
+      cb(null, true)
+    },
+    limits: { fileSize: 50 * 1024 * 1024 } // enforce 50 MB
+  })
+
+  // transcoding stage
+
+  app.post('/api/transcode', upload.single('file'), async (req, res) => {
+    try {
+      console.log(`Received file: ${req.file.originalname}`) // log file to be processed
+      const inputFilePath = req.file.path
+      const outputFilePath = path.join('uploads', `${req.file.filename}.mp4`)
+      const jobId = `${Date.now()}-${Math.random()}` // generate unique job id
+
+      jobResponses[jobId] = res
+
+      const message = {
+        jobId,
+        inputFilePath,
+        outputFilePath
+      }
+
+      jobQueue.push(message)
+      assignJob()
+    } catch (error) {
+      console.error('Error handling transcode request:', error)
+      res.status(500).send('Internal Server Error')
+    }
+  })
+
+  cluster.on('message', (worker, message) => {
+    if (message.jobId && jobResponses[message.jobId]) { // check if job id exists
+      const res = jobResponses[message.jobId] // formulate response
+      if (message.error) {
+        res.status(500).send(message.error)
+      } else {
+        res.sendFile(message.outputFilePath, async (err) => { // send transcoded video
+          if (err) {
+            console.error(`Error: ${err.message}`)
+            if (!res.headersSent) {
+              return res.status(500).send('Error sending transcoded video')
+            }
+          } else {
+            console.log(`Video successfully transcoded: ${message.outputFilePath}`)
+          }
+          try { // delete video files in any case
+            if (message.inputFilePath) {
+              await unlinkAsync(path.resolve(message.inputFilePath))
+            }
+            if (message.outputFilePath) {
+              await unlinkAsync(path.resolve(message.outputFilePath))
+            }
+            console.log(`Deleted video files: ${message.inputFilePath}, ${message.outputFilePath}`)
+          } catch (unlinkError) {
+            console.error(`Error deleting files: ${unlinkError.message}`)
+          }
+        })
+      }
+      delete jobResponses[message.jobId] // job is done, remove from responses
+      workersStatus[worker.id] = 'IDLE' // worker is available
+      assignJob()
+    }
+  })
+
+  const PORT = process.env.FFMPEG_PORT
+  app.listen(PORT, () => {
+    console.log(`FFmpeg cluster is listening on port ${PORT}`)
+  })
+} else { // engage worker transcoding stage
+  process.on('message', async (message) => {
+    const { jobId, inputFilePath, outputFilePath } = message
+
+    console.log(`FFmpeg worker PID ${process.pid} received task to transcode ${inputFilePath} to ${outputFilePath}`)
+    const command = ffmpegCommand(inputFilePath, outputFilePath)
+
+    try {
+      // TODO use spawn instead of exec
+      const { stdout, stderr } = await execAsync(command)
+      console.log(stdout)
+      console.error(stderr)
+
+      const absoluteOutputFilePath = path.resolve(outputFilePath)
+
+      process.send({ jobId, inputFilePath, outputFilePath: absoluteOutputFilePath })
+    } catch (error) {
+      console.error(`Error during transcoding job ${jobId}: ${error}`)
+      process.send({ jobId, error })
+    }
+  })
+}
+
+function ffmpegCommand (input, output) {
+  const params = [
+    '-i', input,
+    '-c:v', 'libx264',
+    '-preset', FFMPEG_PRESET,
+    '-crf', DEFAULT_CRF,
+    '-c:a', 'aac',
+    '-b:a', DEFAULT_AUDIO_BITRATE,
+    '-movflags', '+faststart',
+    '-pix_fmt', 'yuv420p',
+    output
+  ]
+
+  return `ffmpeg ${params.join(' ')}`
+}

lib/constants.js

@@ -26,6 +26,7 @@ export const UPLOAD_TYPES_ALLOW = [
   'image/png',
   'image/jpeg',
   'image/webp',
+  'video/quicktime',
   'video/mp4',
   'video/mpeg',
   'video/webm'

next.config.js

@@ -210,6 +210,7 @@ module.exports = withPlausibleProxy()({
             'process.env.MEDIA_URL_DOCKER': JSON.stringify(process.env.MEDIA_URL_DOCKER),
             'process.env.NEXT_PUBLIC_MEDIA_URL': JSON.stringify(process.env.NEXT_PUBLIC_MEDIA_URL),
             'process.env.NEXT_PUBLIC_MEDIA_DOMAIN': JSON.stringify(process.env.NEXT_PUBLIC_MEDIA_DOMAIN),
+            'process.env.NEXT_PUBLIC_FFMPEG_URL': JSON.stringify(process.env.NEXT_PUBLIC_FFMPEG_URL),
             'process.env.NEXT_PUBLIC_URL': JSON.stringify(process.env.NEXT_PUBLIC_URL),
             'process.env.NEXT_PUBLIC_FAST_POLL_INTERVAL': JSON.stringify(process.env.NEXT_PUBLIC_FAST_POLL_INTERVAL),
             'process.env.NEXT_PUBLIC_NORMAL_POLL_INTERVAL': JSON.stringify(process.env.NEXT_PUBLIC_NORMAL_POLL_INTERVAL),