Mert

Client (React) ←WebSocket→ Node.js WS Server ←Bull Events→ Worker Process

// ws/server.js
import { WebSocketServer } from 'ws';
import { QueueEvents } from 'bullmq';
import { verifyJobOwnership } from '../lib/auth.js';

const redisConnection = {
  host: process.env.REDIS_HOST,
  port: parseInt(process.env.REDIS_PORT || '6379'),
  password: process.env.REDIS_PASSWORD
};

// Map of jobId → Set of WebSocket connections
const jobConnections = new Map();

export function setupWebSocketServer(httpServer) {
  const wss = new WebSocketServer({
    server: httpServer,
    path: '/ws'
  });

  // Listen for Bull queue events
  const queueEvents = new QueueEvents('video:encode', { connection: redisConnection });

  queueEvents.on('progress', ({ jobId, data }) => {
    const connections = jobConnections.get(jobId);
    if (!connections) return;

    const message = JSON.stringify({
      type: 'progress',
      jobId,
      stage: data.stage,
      pct: data.pct,
      message: data.message
    });

    for (const ws of connections) {
      if (ws.readyState === ws.OPEN) {
        ws.send(message);
      }
    }
  });

  queueEvents.on('completed', ({ jobId, returnvalue }) => {
    const connections = jobConnections.get(jobId);
    if (!connections) return;

    const message = JSON.stringify({
      type: 'completed',
      jobId,
      result: JSON.parse(returnvalue || '{}')
    });

    for (const ws of connections) {
      if (ws.readyState === ws.OPEN) {
        ws.send(message);
        ws.close(1000, 'Job complete');
      }
    }

    jobConnections.delete(jobId);
  });

  queueEvents.on('failed', ({ jobId, failedReason }) => {
    const connections = jobConnections.get(jobId);
    if (!connections) return;

    const message = JSON.stringify({
      type: 'failed',
      jobId,
      error: failedReason
    });

    for (const ws of connections) {
      if (ws.readyState === ws.OPEN) ws.send(message);
    }

    jobConnections.delete(jobId);
  });

  // Handle new WebSocket connections
  wss.on('connection', async (ws, req) => {
    // Parse jobId from URL: /ws?jobId=xxx&token=yyy
    const params = new URL(req.url, 'http://localhost').searchParams;
    const jobId = params.get('jobId');
    const token = params.get('token');

    if (!jobId || !token) {
      ws.close(4001, 'Missing jobId or token');
      return;
    }

    // Verify the user owns this job
    const authorized = await verifyJobOwnership(token, jobId);
    if (!authorized) {
      ws.close(4003, 'Unauthorized');
      return;
    }

    // Register connection
    if (!jobConnections.has(jobId)) {
      jobConnections.set(jobId, new Set());
    }
    jobConnections.get(jobId).add(ws);

    // Send current job state immediately
    const currentState = await getCurrentJobState(jobId);
    if (currentState) {
      ws.send(JSON.stringify({ type: 'state', ...currentState }));
    }

    ws.on('close', () => {
      const connections = jobConnections.get(jobId);
      if (connections) {
        connections.delete(ws);
        if (connections.size === 0) jobConnections.delete(jobId);
      }
    });

    // Heartbeat
    ws.isAlive = true;
    ws.on('pong', () => { ws.isAlive = true; });
  });

  // Heartbeat interval to clean up dead connections
  const heartbeat = setInterval(() => {
    wss.clients.forEach(ws => {
      if (!ws.isAlive) return ws.terminate();
      ws.isAlive = false;
      ws.ping();
    });
  }, 30_000);

  wss.on('close', () => clearInterval(heartbeat));

  return wss;
}

// workers/processors/encode.js
export async function encodeProcessor(job) {
  const { videoPath, clipSegments } = job.data;

  await job.updateProgress({ stage: 'starting', pct: 5, message: 'Preparing video...' });

  // Face detection
  let cropX;
  try {
    await job.updateProgress({ stage: 'analyzing', pct: 15, message: 'Detecting faces...' });
    cropX = await detectAndComputeCrop(videoPath);
  } catch {
    cropX = null; // fallback
  }

  await job.updateProgress({ stage: 'encoding', pct: 30, message: 'Encoding clips...' });

  const outputPaths = [];
  for (let i = 0; i < clipSegments.length; i++) {
    await encodeClip(videoPath, clipSegments[i], cropX);
    const pct = 30 + Math.floor(((i + 1) / clipSegments.length) * 50);
    await job.updateProgress({
      stage: 'encoding',
      pct,
      message: `Encoded clip ${i + 1} of ${clipSegments.length}`
    });
    outputPaths.push(/* output path */);
  }

  await job.updateProgress({ stage: 'uploading', pct: 85, message: 'Uploading clips...' });
  const urls = await uploadClips(outputPaths);

  await job.updateProgress({ stage: 'complete', pct: 100, message: 'Done!' });
  return { clipUrls: urls };
}

// hooks/useJobProgress.js
import { useEffect, useRef, useState, useCallback } from 'react';

const RECONNECT_DELAY = 2000;
const MAX_RECONNECTS = 5;

export function useJobProgress(jobId, authToken) {
  const [state, setState] = useState({ status: 'connecting', pct: 0, stage: '', message: '' });
  const wsRef = useRef(null);
  const reconnectCount = useRef(0);
  const reconnectTimer = useRef(null);

  const connect = useCallback(() => {
    if (!jobId || !authToken) return;

    const url = `${process.env.NEXT_PUBLIC_WS_URL}/ws?jobId=${jobId}&token=${authToken}`;
    const ws = new WebSocket(url);
    wsRef.current = ws;

    ws.onopen = () => {
      reconnectCount.current = 0;
      setState(prev => ({ ...prev, status: 'connected' }));
    };

    ws.onmessage = (event) => {
      const data = JSON.parse(event.data);

      switch (data.type) {
        case 'progress':
          setState({ status: 'processing', pct: data.pct, stage: data.stage, message: data.message });
          break;
        case 'completed':
          setState({ status: 'completed', pct: 100, stage: 'done', clips: data.result.clipUrls });
          break;
        case 'failed':
          setState({ status: 'failed', error: data.error, pct: 0, stage: '' });
          break;
        case 'state':
          setState({ status: data.status, pct: data.pct || 0, stage: data.stage || '' });
          break;
      }
    };

    ws.onclose = (event) => {
      if (event.code === 1000) return; // Normal close

      if (reconnectCount.current < MAX_RECONNECTS) {
        reconnectCount.current++;
        reconnectTimer.current = setTimeout(connect, RECONNECT_DELAY);
      } else {
        setState(prev => ({ ...prev, status: 'disconnected' }));
      }
    };

    ws.onerror = () => {
      ws.close();
    };
  }, [jobId, authToken]);

  useEffect(() => {
    connect();
    return () => {
      clearTimeout(reconnectTimer.current);
      wsRef.current?.close(1000, 'Component unmounted');
    };
  }, [connect]);

  return state;
}

// components/JobProgress.jsx
import { useJobProgress } from '../hooks/useJobProgress';

export function JobProgress({ jobId, token }) {
  const { status, pct, stage, message, clips, error } = useJobProgress(jobId, token);

  if (status === 'completed') {
    return (
      <div>
        <p>Processing complete! {clips?.length} clips ready.</p>
        {clips?.map((url, i) => (
          <a key={i} href={url} download>Download clip {i + 1}</a>
        ))}
      </div>
    );
  }

  if (status === 'failed') {
    return <p>Processing failed: {error}</p>;
  }

  return (
    <div>
      <div style={{ width: `${pct}%`, height: 4, background: '#4ade80', transition: 'width 0.3s' }} />
      <p>{message || stage} — {pct}%</p>
    </div>
  );
}

const delay = RECONNECT_DELAY * (1 + Math.random() * 0.5); // ±25% jitter
reconnectTimer.current = setTimeout(connect, delay);

// extract_audio.js
import { execa } from 'execa';
import path from 'path';

export async function extractAudio(videoPath) {
  const audioPath = videoPath.replace(/.(mp4|mov|webm)$/, '_audio.mp3');

  await execa('ffmpeg', [
    '-i', videoPath,
    '-vn',                    // no video
    '-acodec', 'mp3',
    '-ar', '16000',           // Whisper prefers 16kHz
    '-ac', '1',               // mono
    '-b:a', '64k',
    '-y',
    audioPath
  ]);

  return audioPath;
}

// transcribe.js
import OpenAI from 'openai';
import fs from 'fs';

const openai = new OpenAI({ apiKey: process.env.OPENAI_API_KEY });

export async function transcribeAudio(audioPath) {
  const response = await openai.audio.transcriptions.create({
    file: fs.createReadStream(audioPath),
    model: 'whisper-1',
    response_format: 'verbose_json',
    timestamp_granularities: ['word']  // word-level timestamps
  });

  return response.words; // [{ word, start, end }]
}

// caption_chunker.js
export function buildCaptionChunks(words, maxWords = 5, maxDuration = 2.5) {
  const chunks = [];
  let current = [];
  let chunkStart = null;

  for (const word of words) {
    if (current.length === 0) {
      chunkStart = word.start;
    }

    current.push(word.word);
    const chunkDuration = word.end - chunkStart;

    if (current.length >= maxWords || chunkDuration >= maxDuration) {
      chunks.push({
        text: current.join(' ').toUpperCase(), // caps look better in short-form
        start: chunkStart,
        end: word.end
      });
      current = [];
      chunkStart = null;
    }
  }

  // Flush remaining words
  if (current.length > 0) {
    chunks.push({
      text: current.join(' ').toUpperCase(),
      start: chunkStart,
      end: words[words.length - 1].end
    });
  }

  return chunks;
}

// ass_writer.js
import fs from 'fs';

export function writeAssFile(chunks, outputPath, videoWidth = 1080, videoHeight = 1920) {
  const header = `[Script Info]
ScriptType: v4.00+
PlayResX: ${videoWidth}
PlayResY: ${videoHeight}
ScaledBorderAndShadow: yes

[V4+ Styles]
Format: Name, Fontname, Fontsize, PrimaryColour, SecondaryColour, OutlineColour, BackColour, Bold, Italic, Underline, StrikeOut, ScaleX, ScaleY, Spacing, Angle, BorderStyle, Outline, Shadow, Alignment, MarginL, MarginR, MarginV, Encoding
Style: Default,Arial,${Math.floor(videoWidth * 0.055)},&H00FFFFFF,&H000000FF,&H00000000,&H80000000,-1,0,0,0,100,100,0,0,1,3,1,2,10,10,${Math.floor(videoHeight * 0.1)},1

[Events]
Format: Layer, Start, End, Style, Name, MarginL, MarginR, MarginV, Effect, Text
`;

  const events = chunks.map(chunk => {
    const start = formatAssTime(chunk.start);
    const end = formatAssTime(chunk.end);
    return `Dialogue: 0,${start},${end},Default,,0,0,0,,{\an5}${chunk.text}`;
  }).join('n');

  fs.writeFileSync(outputPath, header + events);
}

function formatAssTime(seconds) {
  const h = Math.floor(seconds / 3600);
  const m = Math.floor((seconds % 3600) / 60);
  const s = Math.floor(seconds % 60);
  const cs = Math.floor((seconds % 1) * 100);
  return `${h}:${String(m).padStart(2, '0')}:${String(s).padStart(2, '0')}.${String(cs).padStart(2, '0')}`;
}

// caption_burner.js
import { execa } from 'execa';

export async function burnCaptions(videoPath, assPath, outputPath) {
  await execa('ffmpeg', [
    '-i', videoPath,
    '-vf', `ass=${assPath}`,
    '-c:v', 'libx264',
    '-preset', 'fast',
    '-crf', '22',
    '-c:a', 'copy',       // don't re-encode audio
    '-y',
    outputPath
  ]);
}

// caption_worker.js
import { extractAudio } from './extract_audio.js';
import { transcribeAudio } from './transcribe.js';
import { buildCaptionChunks } from './caption_chunker.js';
import { writeAssFile } from './ass_writer.js';
import { burnCaptions } from './caption_burner.js';
import fs from 'fs';
import path from 'path';

export async function addCaptionsToClip(clipPath) {
  const tmp = path.dirname(clipPath);
  const base = path.basename(clipPath, '.mp4');
  const audioPath = path.join(tmp, `${base}_audio.mp3`);
  const assPath = path.join(tmp, `${base}.ass`);
  const outputPath = path.join(tmp, `${base}_captioned.mp4`);

  try {
    const audioFile = await extractAudio(clipPath);
    const words = await transcribeAudio(audioFile);
    const chunks = buildCaptionChunks(words);
    writeAssFile(chunks, assPath);
    await burnCaptions(clipPath, assPath, outputPath);
    return outputPath;
  } finally {
    // Cleanup temp files
    [audioPath, assPath].forEach(f => {
      try { fs.unlinkSync(f); } catch {}
    });
  }
}