refboard-ayon/backend/services/media-worker.js

/**
 * media-worker.js — Background queue worker for video processing.
 *
 * Polls the media_jobs table for pending jobs and runs ffprobe + ffmpeg
 * with a concurrency limit so uploads return instantly.
 *
 * Emits socket events so the frontend can upgrade placeholders in real-time.
 */

const {
  getPendingMediaJobs,
  updateMediaJob,
  updateImageMedia,
  getImage,
} = require('../db');
const { probeVideo, extractPoster } = require('../video-utils');
const { putBuffer, minioClient, MINIO_BUCKET } = require('../minio');

const POLL_INTERVAL_MS = 3000;
const MAX_CONCURRENCY = 2;

let io = null;
let pollTimer = null;
let activeCount = 0;
let stopping = false;

/**
 * Start the media worker. Pass the Socket.IO server instance for notifications.
 */
function startMediaWorker(ioInstance) {
  io = ioInstance;
  stopping = false;
  console.log('[media-worker] Started (poll=%dms, concurrency=%d)', POLL_INTERVAL_MS, MAX_CONCURRENCY);
  poll();
}

function stopMediaWorker() {
  stopping = true;
  if (pollTimer) {
    clearTimeout(pollTimer);
    pollTimer = null;
  }
  console.log('[media-worker] Stopped');
}

function schedulePoll() {
  if (stopping) return;
  pollTimer = setTimeout(poll, POLL_INTERVAL_MS);
}

async function poll() {
  if (stopping) return;

  const slotsAvailable = MAX_CONCURRENCY - activeCount;
  if (slotsAvailable <= 0) {
    schedulePoll();
    return;
  }

  try {
    const jobs = getPendingMediaJobs(slotsAvailable);
    for (const job of jobs) {
      activeCount++;
      processJob(job).finally(() => {
        activeCount--;
      });
    }
  } catch (err) {
    console.error('[media-worker] Poll error:', err.message);
  }

  schedulePoll();
}

async function processJob(job) {
  const jobId = job.id;
  const imageId = job.image_id;
  const boardId = job.board_id;

  try {
    updateMediaJob(jobId, { status: 'processing', startedAt: new Date().toISOString() });
    emitJobUpdate(boardId, jobId, imageId, 'processing');

    // Fetch the raw video from MinIO
    const image = getImage(imageId);
    if (!image) {
      updateMediaJob(jobId, { status: 'failed', error: 'Image record not found' });
      emitJobFailed(boardId, jobId, imageId, 'Image record not found');
      return;
    }

    const videoBuffer = await fetchFromMinio(image.minio_path);
    if (!videoBuffer) {
      updateMediaJob(jobId, { status: 'failed', error: 'Failed to fetch video from storage' });
      emitJobFailed(boardId, jobId, imageId, 'Failed to fetch video from storage');
      return;
    }

    // Run ffprobe + ffmpeg in parallel
    const [meta, posterBuf] = await Promise.all([
      probeVideo(videoBuffer),
      extractPoster(videoBuffer),
    ]);

    let posterAssetKey = null;
    if (posterBuf) {
      posterAssetKey = `boards/${boardId}/${imageId}_poster.jpg`;
      await putBuffer(posterAssetKey, posterBuf, 'image/jpeg');
    }

    const nativeWidth = meta?.width || null;
    const nativeHeight = meta?.height || null;
    const duration = meta?.duration || null;

    // Update the image record with processed media info
    updateImageMedia(imageId, { posterAssetKey, duration, nativeWidth, nativeHeight });

    updateMediaJob(jobId, {
      status: 'done',
      finishedAt: new Date().toISOString(),
    });

    emitJobUpdate(boardId, jobId, imageId, 'done', {
      posterAssetKey,
      nativeWidth,
      nativeHeight,
      duration,
    });

    console.log('[media-worker] Job %s done (image=%s, poster=%s)', jobId, imageId, !!posterAssetKey);
  } catch (err) {
    console.error('[media-worker] Job %s failed:', jobId, err.message);

    // Classify error for user-facing message
    let userError = 'Video processing failed';
    if (err.killed || err.signal === 'SIGTERM') {
      userError = 'Video processing timed out — file may be too large or corrupt';
    } else if (err.message?.includes('ENOMEM') || err.message?.includes('Cannot allocate')) {
      userError = 'Out of memory — video file is too large to process';
    } else if (err.message?.includes('Invalid data')) {
      userError = 'Invalid or corrupt video file';
    }

    const attempts = (job.attempts || 0) + 1;
    if (attempts < 3) {
      updateMediaJob(jobId, { status: 'retry', attempts, error: err.message });
      emitJobUpdate(boardId, jobId, imageId, 'retry');
    } else {
      updateMediaJob(jobId, { status: 'failed', attempts, error: err.message });
      emitJobFailed(boardId, jobId, imageId, userError);
    }
  }
}

/**
 * Fetch object from MinIO as a Buffer.
 */
async function fetchFromMinio(objectPath) {
  try {
    const stream = await minioClient.getObject(MINIO_BUCKET, objectPath);
    const chunks = [];
    for await (const chunk of stream) {
      chunks.push(chunk);
    }
    return Buffer.concat(chunks);
  } catch (err) {
    console.warn('[media-worker] MinIO fetch failed:', err.message);
    return null;
  }
}

/**
 * Emit a media job update to all sockets in the board room.
 */
function emitJobUpdate(boardId, jobId, imageId, status, result) {
  if (!io) return;
  const room = `board:${boardId}`;
  io.to(room).emit('media:job:update', {
    jobId,
    imageId,
    status,
    ...(result || {}),
  });
}

/**
 * Emit a job failure with error details to the board room.
 */
function emitJobFailed(boardId, jobId, imageId, error) {
  emitJobUpdate(boardId, jobId, imageId, 'failed', { error });
}

module.exports = { startMediaWorker, stopMediaWorker };