video_socket-server/src/recording/composer.ts

import { spawn } from 'child_process';
import { v4 as uuid } from 'uuid';
import {
  RecordingPerson,
  ServerTrackRecordingFile,
  ServerTrackRecordingTarget,
  createComposedRecordingTarget,
  deleteServerTrackRecordingFiles,
  listServerTrackRecordingFiles,
  writeComposedRecordingMetadata
} from './storage';

export type RecordingCompositionStatus = 'queued' | 'running' | 'completed' | 'failed';

export type RecordingCompositionJob = {
  id: string;
  recordingId: string;
  meetingId: string;
  status: RecordingCompositionStatus;
  layout: string;
  format: string;
  createdAt: string;
  updatedAt: string;
  startedAt?: string;
  completedAt?: string;
  failedAt?: string;
  error?: string;
  inputFiles: string[];
  host?: RecordingPerson;
  participants?: RecordingPerson[];
  deletedInputFiles?: string[];
  output?: {
    meetingId: string;
    filename: string;
    filePath: string;
    metadataPath: string;
    downloadUrl: string;
    streamUrl: string;
  };
};

type StartCompositionInput = {
  meetingId: string;
  recordingId: string;
  layout?: string;
  format?: string;
  host?: RecordingPerson;
  participants?: RecordingPerson[];
};

type CompositionInputSets = {
  videoInputs: ServerTrackRecordingFile[];
  audioInputs: ServerTrackRecordingFile[];
};

type VideoTimelineSegment = {
  startMs: number;
  endMs: number;
  activeInputs: ServerTrackRecordingFile[];
};

const jobs: Map<string, RecordingCompositionJob> = new Map<string, RecordingCompositionJob>();

function nowIso(): string {
  return new Date().toISOString();
}

function normalizeOption(value: unknown, fallback: string): string {
  if (typeof value !== 'string') {
    return fallback;
  }
  const trimmed = value.trim();
  return trimmed ? trimmed.slice(0, 40) : fallback;
}

function normalizeFormat(value: unknown): string {
  return normalizeOption(value, 'webm') === 'mp4' ? 'mp4' : 'webm';
}

function getFfmpegPath(): string {
  return process.env.FFMPEG_PATH || 'ffmpeg';
}

function sortInputs(files: ServerTrackRecordingFile[]): ServerTrackRecordingFile[] {
  return files.slice().sort((a, b) => {
    const participantCompare = a.participantId.localeCompare(b.participantId);
    if (participantCompare !== 0) {
      return participantCompare;
    }
    return Date.parse(a.uploadedAt) - Date.parse(b.uploadedAt);
  });
}

function getInputSets(input: StartCompositionInput): CompositionInputSets {
  const files = listServerTrackRecordingFiles({
    meetingId: input.meetingId,
    recordingId: input.recordingId
  });
  return {
    videoInputs: sortInputs(files.filter((file) => file.trackKind === 'video')),
    audioInputs: sortInputs(files.filter((file) => file.trackKind === 'audio'))
  };
}

function isHostInput(file: ServerTrackRecordingFile): boolean {
  if (file.metadata && file.metadata.role === 'host') {
    return true;
  }

  const firstParticipant = file.metadata && Array.isArray(file.metadata.participants)
    ? file.metadata.participants[0]
    : null;
  return Boolean(firstParticipant && firstParticipant.role === 'host');
}

function orderVideoInputsForComposition(files: ServerTrackRecordingFile[]): ServerTrackRecordingFile[] {
  const hostIndex = files.findIndex(isHostInput);
  if (hostIndex <= 0) {
    return files.slice();
  }

  return [
    files[hostIndex],
    ...files.slice(0, hostIndex),
    ...files.slice(hostIndex + 1)
  ];
}

function parseInputTimestamp(value: unknown): number | null {
  if (typeof value !== 'string' || !value.trim()) {
    return null;
  }

  const timestamp = Date.parse(value);
  return isFinite(timestamp) ? timestamp : null;
}

function getInputStartMs(file: ServerTrackRecordingFile): number | null {
  const metadata = file.metadata || {};
  return parseInputTimestamp(metadata.recordingStartedAt)
    || parseInputTimestamp(file.recordingStartedAt)
    || parseInputTimestamp(metadata.startedAt)
    || parseInputTimestamp(file.uploadedAt)
    || parseInputTimestamp(metadata.uploadedAt);
}

function getInputEndMs(file: ServerTrackRecordingFile): number | null {
  const metadata = file.metadata || {};
  return parseInputTimestamp(metadata.recordingEndedAt)
    || parseInputTimestamp(file.recordingEndedAt)
    || parseInputTimestamp(metadata.endedAt)
    || parseInputTimestamp(metadata.updatedAt);
}

function getTimelineOriginMs(files: ServerTrackRecordingFile[]): number | null {
  const starts = files
    .map(getInputStartMs)
    .filter((timestamp) => timestamp !== null) as number[];
  if (starts.length === 0) {
    return null;
  }

  return Math.min(...starts);
}

function getTimelineDurationSeconds(files: ServerTrackRecordingFile[], timelineOriginMs: number | null): number | null {
  if (timelineOriginMs === null) {
    return null;
  }

  const ends = files
    .map(getInputEndMs)
    .filter((timestamp) => timestamp !== null) as number[];
  if (ends.length === 0) {
    return null;
  }

  const durationSeconds = (Math.max(...ends) - timelineOriginMs) / 1000;
  return durationSeconds > 0 ? durationSeconds : null;
}

function getTimelineEndMs(files: ServerTrackRecordingFile[]): number | null {
  const ends = files
    .map(getInputEndMs)
    .filter((timestamp) => timestamp !== null) as number[];
  if (ends.length === 0) {
    return null;
  }

  return Math.max(...ends);
}

function getInputOffsetSeconds(file: ServerTrackRecordingFile, timelineOriginMs: number | null): number {
  const startMs = getInputStartMs(file);
  if (startMs === null || timelineOriginMs === null) {
    return 0;
  }

  return Math.max(0, (startMs - timelineOriginMs) / 1000);
}

function formatSeconds(value: number): string {
  if (value <= 0.001) {
    return '0';
  }

  return value.toFixed(3).replace(/\.?0+$/, '');
}

function getBottomTileWidth(index: number, inputCount: number, outputWidth: number): number {
  const sideCount = inputCount - 1;
  if (sideCount <= 1) {
    return outputWidth;
  }

  const rawWidth = Math.floor(outputWidth / sideCount);
  const tileWidth = rawWidth % 2 === 0 ? rawWidth : rawWidth - 1;
  return index === sideCount - 1 ? outputWidth - (tileWidth * index) : tileWidth;
}

function createHostBottomLayout(inputCount: number, outputWidth: number, hostHeight: number): string {
  const positions = ['0_0'];
  const sideCount = inputCount - 1;
  let x = 0;
  for (let sideIndex = 0; sideIndex < sideCount; sideIndex += 1) {
    positions.push(`${x}_${hostHeight}`);
    x += getBottomTileWidth(sideIndex, inputCount, outputWidth);
  }
  return positions.join('|');
}

function sortActiveVideoInputs(files: ServerTrackRecordingFile[]): ServerTrackRecordingFile[] {
  return orderVideoInputsForComposition(files).sort((a, b) => {
    const aIsHost = isHostInput(a);
    const bIsHost = isHostInput(b);
    if (aIsHost !== bIsHost) {
      return aIsHost ? -1 : 1;
    }
    return a.participantId.localeCompare(b.participantId);
  });
}

function getVideoTimelineSegments(
  files: ServerTrackRecordingFile[],
  timelineOriginMs: number | null,
  timelineEndMs: number | null
): VideoTimelineSegment[] {
  if (timelineOriginMs === null || timelineEndMs === null || timelineEndMs <= timelineOriginMs) {
    return [];
  }

  const pointsByMs: { [timestamp: string]: boolean } = {};
  pointsByMs[String(timelineOriginMs)] = true;
  pointsByMs[String(timelineEndMs)] = true;
  files.forEach((file) => {
    const startMs = getInputStartMs(file);
    const endMs = getInputEndMs(file);
    if (startMs !== null && startMs > timelineOriginMs && startMs < timelineEndMs) {
      pointsByMs[String(startMs)] = true;
    }
    if (endMs !== null && endMs > timelineOriginMs && endMs < timelineEndMs) {
      pointsByMs[String(endMs)] = true;
    }
  });

  const points = Object.keys(pointsByMs)
    .map((value) => Number(value))
    .sort((a, b) => a - b);
  const segments: VideoTimelineSegment[] = [];
  for (let index = 0; index < points.length - 1; index += 1) {
    const startMs = points[index];
    const endMs = points[index + 1];
    if (endMs <= startMs) {
      continue;
    }

    const activeInputs = sortActiveVideoInputs(files.filter((file) => {
      const inputStartMs = getInputStartMs(file) || timelineOriginMs;
      const inputEndMs = getInputEndMs(file) || timelineEndMs;
      return inputStartMs < endMs && inputEndMs > startMs;
    }));
    segments.push({ startMs, endMs, activeInputs });
  }

  return segments;
}

export function buildFfmpegCompositionArgs(input: {
  videoInputs: ServerTrackRecordingFile[];
  audioInputs: ServerTrackRecordingFile[];
  outputPath: string;
  format: string;
}): string[] {
  const outputWidth = 1280;
  const outputHeight = 720;
  const hostHeight = 540;
  const bottomHeight = outputHeight - hostHeight;
  const videoInputs = orderVideoInputsForComposition(input.videoInputs);
  const timelineOriginMs = getTimelineOriginMs(videoInputs.concat(input.audioInputs));
  const timelineEndMs = getTimelineEndMs(videoInputs.concat(input.audioInputs));
  const timelineDurationSeconds = getTimelineDurationSeconds(videoInputs.concat(input.audioInputs), timelineOriginMs);
  const videoSegments = getVideoTimelineSegments(videoInputs, timelineOriginMs, timelineEndMs);
  const args = ['-y'];
  const orderedInputs = videoInputs.concat(input.audioInputs);
  orderedInputs.forEach((file) => {
    args.push('-i', file.filePath);
  });

  const filters: string[] = [];
  const videoInputUseCounts = videoInputs.map((file) => videoSegments.filter((segment) => segment.activeInputs.indexOf(file) >= 0).length);
  const videoInputUsePositions = videoInputs.map(() => 0);
  videoInputUseCounts.forEach((useCount, inputIndex) => {
    if (useCount <= 1) {
      return;
    }

    const splitLabels = [];
    for (let splitIndex = 0; splitIndex < useCount; splitIndex += 1) {
      splitLabels.push(`[vin${inputIndex}_${splitIndex}]`);
    }
    filters.push(`[${inputIndex}:v]split=${useCount}${splitLabels.join('')}`);
  });

  videoSegments.forEach((segment, segmentIndex) => {
    const segmentDurationSeconds = (segment.endMs - segment.startMs) / 1000;
    if (segment.activeInputs.length === 0) {
      filters.push(`color=color=black:size=${outputWidth}x${outputHeight}:rate=30:duration=${formatSeconds(segmentDurationSeconds)},format=yuv420p[seg${segmentIndex}]`);
      return;
    }

    segment.activeInputs.forEach((file, activeIndex) => {
      const inputIndex = videoInputs.indexOf(file);
      const inputLabel = videoInputUseCounts[inputIndex] > 1
        ? `vin${inputIndex}_${videoInputUsePositions[inputIndex]++}`
        : `${inputIndex}:v`;
      const inputStartMs = getInputStartMs(file) || segment.startMs;
      const trimStartSeconds = Math.max(0, (segment.startMs - inputStartMs) / 1000);
      const width = segment.activeInputs.length === 1
        ? outputWidth
        : activeIndex === 0 ? outputWidth : getBottomTileWidth(activeIndex - 1, segment.activeInputs.length, outputWidth);
      const height = segment.activeInputs.length === 1
        ? outputHeight
        : activeIndex === 0 ? hostHeight : bottomHeight;
      filters.push(`[${inputLabel}]trim=start=${formatSeconds(trimStartSeconds)}:duration=${formatSeconds(segmentDurationSeconds)},setpts=PTS-STARTPTS,scale=${width}:${height}:force_original_aspect_ratio=decrease,pad=${width}:${height}:(ow-iw)/2:(oh-ih)/2:black,setsar=1[seg${segmentIndex}v${activeIndex}]`);
    });

    if (segment.activeInputs.length === 1) {
      filters.push(`[seg${segmentIndex}v0]fps=30,format=yuv420p[seg${segmentIndex}]`);
      return;
    }

    const segmentVideoLabels = segment.activeInputs.map((_file, activeIndex) => `[seg${segmentIndex}v${activeIndex}]`).join('');
    filters.push(`${segmentVideoLabels}xstack=inputs=${segment.activeInputs.length}:layout=${createHostBottomLayout(segment.activeInputs.length, outputWidth, hostHeight)}:fill=black,fps=30,format=yuv420p[seg${segmentIndex}]`);
  });

  if (videoSegments.length === 1) {
    filters.push('[seg0]null[vout]');
  } else {
    const videoLabels = videoSegments.map((_segment, index) => `[seg${index}]`).join('');
    filters.push(`${videoLabels}concat=n=${videoSegments.length}:v=1:a=0[vout]`);
  }

  if (input.audioInputs.length === 1) {
    const audioInputIndex = videoInputs.length;
    const offsetMs = Math.round(getInputOffsetSeconds(input.audioInputs[0], timelineOriginMs) * 1000);
    const offsetFilter = offsetMs > 1 ? `,adelay=${offsetMs}:all=1` : '';
    filters.push(`[${audioInputIndex}:a]aresample=async=1:first_pts=0${offsetFilter}[aout]`);
  } else if (input.audioInputs.length > 1) {
    const audioLabels = input.audioInputs.map((file, index) => {
      const audioInputIndex = videoInputs.length + index;
      const offsetMs = Math.round(getInputOffsetSeconds(file, timelineOriginMs) * 1000);
      const offsetFilter = offsetMs > 1 ? `,adelay=${offsetMs}:all=1` : '';
      filters.push(`[${audioInputIndex}:a]aresample=async=1:first_pts=0${offsetFilter}[a${index}]`);
      return `[a${index}]`;
    }).join('');
    filters.push(`${audioLabels}amix=inputs=${input.audioInputs.length}:duration=longest:dropout_transition=2[aout]`);
  }

  args.push('-filter_complex', filters.join(';'), '-map', '[vout]');
  if (input.audioInputs.length > 0) {
    args.push('-map', '[aout]');
  }

  if (input.format === 'mp4') {
    args.push('-c:v', 'libx264', '-preset', 'veryfast', '-pix_fmt', 'yuv420p');
    if (input.audioInputs.length > 0) {
      args.push('-c:a', 'aac');
    }
  } else {
    args.push('-c:v', 'libvpx-vp9', '-deadline', 'realtime', '-cpu-used', '4');
    if (input.audioInputs.length > 0) {
      args.push('-c:a', 'libopus');
    }
  }

  if (timelineDurationSeconds !== null) {
    args.push('-t', formatSeconds(timelineDurationSeconds));
  }
  args.push(input.outputPath);
  return args;
}

function runFfmpeg(args: string[]): Promise<void> {
  return new Promise((resolve, reject) => {
    const child = spawn(getFfmpegPath(), args, { windowsHide: true });
    let stderr = '';
    child.stderr.on('data', (chunk) => {
      stderr += chunk.toString();
    });
    child.on('error', (error: any) => {
      if (error && error.code === 'ENOENT') {
        reject(new Error('ffmpeg was not found. Install ffmpeg or set FFMPEG_PATH.'));
        return;
      }
      reject(error);
    });
    child.on('close', (code) => {
      if (code === 0) {
        resolve();
        return;
      }
      reject(new Error(stderr || `ffmpeg exited with code ${code}`));
    });
  });
}

function toOutput(job: RecordingCompositionJob, target: ServerTrackRecordingTarget): RecordingCompositionJob['output'] {
  return {
    meetingId: target.meetingId,
    filename: target.filename,
    filePath: target.filePath,
    metadataPath: target.metadataPath,
    downloadUrl: `/api/recordings/${encodeURIComponent(target.meetingId)}/${encodeURIComponent(target.filename)}/download`,
    streamUrl: `/api/recordings/${encodeURIComponent(target.meetingId)}/${encodeURIComponent(target.filename)}/stream`
  };
}

async function runRecordingCompositionJob(job: RecordingCompositionJob): Promise<RecordingCompositionJob> {
  const timestamp = nowIso();
  job.status = 'running';
  job.startedAt = timestamp;
  job.updatedAt = timestamp;

  try {
    const inputSets = getInputSets(job);
    if (inputSets.videoInputs.length === 0) {
      throw new Error('No server-side video track files are available for composition.');
    }

    const target = createComposedRecordingTarget({
      meetingId: job.meetingId,
      recordingId: job.recordingId,
      format: job.format
    });
    const compositionInputs = inputSets.videoInputs.concat(inputSets.audioInputs);
    const args = buildFfmpegCompositionArgs({
      ...inputSets,
      outputPath: target.filePath,
      format: job.format
    });
    await runFfmpeg(args);
    writeComposedRecordingMetadata({
      target,
      recordingId: job.recordingId,
      inputs: compositionInputs,
      layout: job.layout,
      format: job.format,
      host: job.host,
      participants: job.participants
    });
    const deletedInputFiles = deleteServerTrackRecordingFiles(compositionInputs);

    const completedAt = nowIso();
    job.status = 'completed';
    job.completedAt = completedAt;
    job.updatedAt = completedAt;
    job.inputFiles = compositionInputs.map((file) => file.filename);
    job.deletedInputFiles = deletedInputFiles;
    job.output = toOutput(job, target);
  } catch (error) {
    const failedAt = nowIso();
    job.status = 'failed';
    job.failedAt = failedAt;
    job.updatedAt = failedAt;
    job.error = error instanceof Error ? error.message : String(error);
  }

  return job;
}

export function startRecordingCompositionJob(input: StartCompositionInput): RecordingCompositionJob {
  const timestamp = nowIso();
  const inputSets = getInputSets(input);
  const job: RecordingCompositionJob = {
    id: uuid(),
    recordingId: normalizeOption(input.recordingId, ''),
    meetingId: normalizeOption(input.meetingId, ''),
    status: 'queued',
    layout: normalizeOption(input.layout, 'grid'),
    format: normalizeFormat(input.format),
    createdAt: timestamp,
    updatedAt: timestamp,
    inputFiles: inputSets.videoInputs.concat(inputSets.audioInputs).map((file) => file.filename),
    host: input.host,
    participants: input.participants
  };
  jobs.set(job.id, job);
  runRecordingCompositionJob(job);
  return job;
}

export function getRecordingCompositionJob(jobId: string): RecordingCompositionJob | null {
  return jobs.get(jobId) || null;
}

export function listRecordingCompositionJobs(meetingId?: string): RecordingCompositionJob[] {
  const allJobs = Array.from(jobs.values());
  return meetingId
    ? allJobs.filter((job) => job.meetingId === meetingId)
    : allJobs;
}

export function resetRecordingCompositionJobs(): void {
  jobs.clear();
}