feat: media processing pipeline, spatial indexing, canvas-based video rendering

- Background media worker: polls media_jobs table, runs ffprobe+ffmpeg with concurrency limit, generates video posters, emits socket events - Non-blocking video upload: stores file + enqueues job, returns immediately - Poster hydration on board load: GET /boards/:id injects poster/dimensions from DB into canvas_state video objects - SpatialGrid: fixed-cell (512px) spatial hash for O(nearby) culling instead of O(all) item scanning, eliminates setTimeout violations - Canvas-based video rendering: draws video frames to offscreen canvas then uploads to GPU, completely eliminates GL_INVALID_OPERATION errors from PixiJS VideoSource auto-update mechanism - Server poster upgrade path: culling ticker and applyProcessedMedia() both upgrade client-captured posters to server posters when available - Pause restores server poster (paused video behaves like an image) - Selection drag-end persistence: onObjectDragEnd broadcasts + saves + undo - Live media:job:update socket handler patches scene data + VideoSprite dimensions without broadcast fanout
2026-03-10 14:30:01 +05:30
parent fdcee3536f
commit de418b2ba5
12 changed files with 761 additions and 110 deletions
@@ -0,0 +1,176 @@
+/**
+ * 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' });
+      emitJobUpdate(boardId, jobId, imageId, 'failed');
+      return;
+    }
+
+    const videoBuffer = await fetchFromMinio(image.minio_path);
+    if (!videoBuffer) {
+      updateMediaJob(jobId, { status: 'failed', error: 'Failed to fetch video from storage' });
+      emitJobUpdate(boardId, jobId, imageId, 'failed');
+      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);
+
+    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 });
+      emitJobUpdate(boardId, jobId, imageId, 'failed');
+    }
+  }
+}
+
+/**
+ * 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 || {}),
+  });
+}
+
+module.exports = { startMediaWorker, stopMediaWorker };