backend/src/utils/media/stream.util.ts

import { Request, Response } from 'express';
import { VideoInfo, Format } from '../../types/index.js';
import { sendEvent, sendBufferedEvent } from '../network/sse.util.js';
import { signProxyParams } from '../network/secrets.util.js';
import { buildPhoneMediaUrl } from '../../services/ytdlp/phone-media.js';
import { Transform } from 'node:stream';
import { pipeline } from 'node:stream/promises';

export const isDirect = (format: Format): boolean =>
  Boolean(
    format.url &&
    !format.url.includes('youtube.com/watch') &&
    !format.url.includes('youtu.be/') &&
    (!format.note ||
      (!format.note.includes('m3u8') && !format.note.includes('manifest'))) &&
    !format.url.includes('.m3u8')
  );

export const isAvc = (format: Format | null | undefined): boolean => {
  if (!format) return false;
  const vcodec = format.vcodec || '';
  return vcodec.startsWith('avc1') || vcodec.startsWith('h264');
};

export function selectVideoFormat(
  formats: Format[],
  formatId: string | undefined
): Format | null {
  const isMuxed = (targetFormat: Format) =>
    targetFormat.vcodec !== 'none' && targetFormat.acodec !== 'none';
  const videoFormats = formats.filter(
    (format) => format.vcodec && format.vcodec !== 'none'
  );

  if (videoFormats.length === 0) return null;

  const h264Formats = videoFormats.filter(
    (format) =>
      format.vcodec?.startsWith('avc1') || format.vcodec?.startsWith('h264')
  );

  const available = (h264Formats.length > 0 ? h264Formats : videoFormats).sort(
    (formatA, formatB) => {
      const heightA = formatA.height || 0;
      const heightB = formatB.height || 0;
      if (heightB !== heightA) return heightB - heightA;
      if (isMuxed(formatB) && !isMuxed(formatA)) return 1;
      return -1;
    }
  );

  const requested = videoFormats.find(
    (format) => String(format.formatId) === String(formatId)
  );
  if (requested) return requested;

  return (
    available.find((format) => (format.height || 0) <= 1080) || available[0]
  );
}

export function selectAudioFormat(
  formats: Format[],
  formatId: string | undefined,
  isAudioOnly: boolean,
  needsWebm: boolean,
  audioLang?: string
): Format | null {
  const availableAudioOnly = formats.filter(
    (format) =>
      format.acodec !== 'none' && (format.vcodec === 'none' || !format.isVideo)
  );

  const langPool = pickAudioLanguagePool(availableAudioOnly, audioLang);
  const audioAbr = (format: Format) =>
    Number(format.abr) || parseInt(format.quality || '0', 10) || 0;

  const m4aAudio = langPool
    .filter((format) => format.extension === 'm4a')
    .sort((formatA, formatB) => audioAbr(formatB) - audioAbr(formatA))[0];
  const webmAudio = langPool
    .filter((format) => format.extension === 'webm' || format.acodec === 'opus')
    .sort((formatA, formatB) => audioAbr(formatB) - audioAbr(formatA))[0];

  const requested = isAudioOnly
    ? formats.find((format) => String(format.formatId) === String(formatId))
    : null;

  return (
    (requested as Format) ||
    (needsWebm && webmAudio ? webmAudio : m4aAudio || webmAudio) ||
    langPool.find((format) => format.acodec !== 'none') ||
    formats.find((format) => format.acodec !== 'none') ||
    null
  );
}

export function pickAudioLanguagePool(
  pool: Format[],
  audioLang?: string
): Format[] {
  if (pool.length === 0) return pool;
  const hasLangMeta = pool.some((format) => format.language || format.isOriginal);
  if (!hasLangMeta) return pool;

  if (audioLang) {
    const wanted = audioLang.toLowerCase();
    const exact = pool.filter(
      (format) => (format.language || '').toLowerCase() === wanted
    );
    if (exact.length) return exact;
    // base-language fallback: "es" matches "es-419"
    const base = wanted.split('-')[0];
    const baseMatch = pool.filter(
      (format) => (format.language || '').split('-')[0].toLowerCase() === base
    );
    if (baseMatch.length) return baseMatch;
  }

  const original = pool.filter((format) => format.isOriginal);
  if (original.length) return original;

  return pool;
}

export function buildProxyUrl(
  req: Request,
  format: Format | null | undefined,
  targetUrl: string
): string | null {
  if (!format?.formatId) return null;

  const host = (req.headers['x-forwarded-host'] as string) || req.get('host');
  const protocol = (req.headers['x-forwarded-proto'] as string) || req.protocol;

  const rawUrl = isDirect(format) ? format.url : undefined;
  const formatId = String(format.formatId);
  const phoneUrl = buildPhoneMediaUrl(rawUrl, targetUrl);
  if (phoneUrl) {
    console.log(`[Download] fmt ${formatId}: via phone relay`);
    return phoneUrl;
  }
  console.log(`[Download] fmt ${formatId}: via server proxy`);

  const { exp, sig } = signProxyParams({ targetUrl, rawUrl, formatId });

  let proxyUrl = `${protocol}://${host}/proxy?targetUrl=${encodeURIComponent(targetUrl)}&formatId=${formatId}&ext=${format.extension || 'mp4'}`;
  if (rawUrl) {
    proxyUrl += `&rawUrl=${encodeURIComponent(rawUrl)}`;
  }
  return `${proxyUrl}&exp=${exp}&sig=${sig}`;
}

export function getOutputMetadata(
  isAudioOnly: boolean,
  emeExtension: string,
  info: VideoInfo
) {
  const mimeMap: Record<string, string> = {
    mp3: 'audio/mpeg',
    m4a: 'audio/mp4',
    mp4: 'video/mp4',
    webm: isAudioOnly ? 'audio/webm' : 'video/webm',
    jpg: 'image/jpeg',
    jpeg: 'image/jpeg',
    png: 'image/png',
  };

  const type =
    mimeMap[emeExtension] ||
    (isAudioOnly ? `audio/${emeExtension}` : 'video/mp4');

  return {
    type,
    metadata: {
      title: info.title,
      artist: info.uploader || info.artist,
    },
  };
}

export function setupStreamListeners(
  videoProcess: NodeJS.ReadableStream,
  res: Response,
  clientId: string | undefined,
  totalBytesSent: { value: number }
) {
  let lastReportedProgress = 30;

  if (clientId) {
    sendEvent(clientId, {
      status: 'downloading',
      progress: 30,
      subStatus: 'STREAMING: Initializing Handshake...',
    });
  }

  const readable = videoProcess as unknown as NodeJS.EventEmitter & {
    pipe: (res: Response) => void;
  };
  if (typeof readable.on === 'function') {
    readable.on('progress', (progress: number) => {
      if (clientId) {
        const scaledProgress = 30 + progress * 0.65;
        const newProgress = Math.min(95, Math.round(scaledProgress));

        if (newProgress > lastReportedProgress) {
          lastReportedProgress = newProgress;
          sendBufferedEvent(clientId, {
            status: 'downloading',
            progress: newProgress,
            subStatus: `STREAMING: ${progress.toFixed(1)}%`,
          });
        }
      }
    });
  }

  let heartbeatTimer: NodeJS.Timeout | null = null;
  const startHeartbeat = () => {
    // tcp keep-alive prevents body corruption
    if (res.socket && !res.socket.destroyed) {
      res.socket.setKeepAlive(true, 25000);
    }
  };

  const stopHeartbeat = () => {
    if (heartbeatTimer) {
      clearInterval(heartbeatTimer);
      heartbeatTimer = null;
    }
  };

  const progressObserver = new Transform({
    highWaterMark: 64 * 1024,
    transform(chunk, encoding, callback) {
      stopHeartbeat();
      totalBytesSent.value += chunk.length;

      if (totalBytesSent.value === chunk.length) {
        console.log(
          `[StreamUtil] First chunk received for client ${clientId} (${chunk.length} bytes)`
        );
        if (clientId) {
          sendEvent(clientId, {
            status: 'downloading',
            progress: lastReportedProgress,
            subStatus: 'TRANSMITTING: Streaming via Turbo',
          });
        }
      }

      startHeartbeat();
      callback(null, chunk);
    },
    flush(callback) {
      stopHeartbeat();
      callback();
    },
  });

  startHeartbeat();

  const onStreamComplete = () => {
    console.log(
      `[StreamUtil] Stream closed for client ${clientId} (Total: ${(totalBytesSent.value / (1024 * 1024)).toFixed(1)}MB)`
    );
    if (clientId) {
      sendEvent(clientId, {
        status: 'completed',
        progress: 100,
        subStatus: `STREAMING: Finalized (${(totalBytesSent.value / (1024 * 1024)).toFixed(1)}MB)`,
      });
    }
  };

  if (typeof readable.pipe === 'function') {
    // pipeline ends res; early end truncates tail
    pipeline(
      videoProcess as unknown as import('stream').Readable,
      progressObserver,
      res
    )
      .then(onStreamComplete)
      .catch((error) => {
        if (error.code !== 'ERR_STREAM_PREMATURE_CLOSE') {
          console.error('[Stream] Pipeline failed:', error.message);
        }
      });
  } else {
    videoProcess.on('close', () => {
      onStreamComplete();
      if (!res.writableEnded) res.end();
    });
  }

  videoProcess.on('error', (error: unknown) => {
    const typedError = error as NodeJS.ErrnoException;
    if (typedError.code === 'ERR_STREAM_WRITE_AFTER_END') return;
    console.error('[Convert] Stream Error:', typedError.message);
    if (!res.headersSent) {
      res.status(500).json({ error: 'Stream generation failed' });
    } else {
      if (!res.writableEnded) res.end();
    }
  });
}