blastoise/ytdlp.ts

// MusicRoom - yt-dlp integration module
// Handles fetching audio from URLs via yt-dlp

import { spawn } from "child_process";
import { join } from "path";
import {
  saveSlowQueueItem,
  updateSlowQueueItem,
  loadSlowQueue,
  deleteSlowQueueItem,
  clearCompletedSlowQueue,
  addTracksToPlaylist,
  type SlowQueueRow
} from "./db";

export interface QueueItem {
  id: string;
  url: string;
  title: string;
  userId: number;
  status: "queued" | "downloading" | "complete" | "error" | "cancelled";
  progress: number;
  queueType: "fast" | "slow";
  error?: string;
  filename?: string;
  createdAt: number;
  completedAt?: number;
  playlistId?: string;
  playlistName?: string;
  position?: number;
  trackId?: string;  // Set after successful download
}

export interface YtdlpStatus {
  available: boolean;
  enabled: boolean;
  version: string | null;
  ffmpeg: boolean;
}

export interface PlaylistInfo {
  type: "playlist";
  title: string;
  count: number;
  items: { id: string; url: string; title: string }[];
  requiresConfirmation: true;
}

export interface SingleVideoInfo {
  type: "single";
  id: string;
  title: string;
  url: string;
}

type ProgressCallback = (item: QueueItem) => void;

// Configuration
let ytdlpCommand = "yt-dlp";
let ffmpegCommand = "ffmpeg";
let musicDir = "./music";
let fastQueueConcurrent = 2;
let slowQueueInterval = 180;
let allowPlaylists = true;

// Status
let ytdlpAvailable = false;
let ytdlpVersion: string | null = null;
let ffmpegAvailable = false;
let featureEnabled = false;

// Queues
const fastQueue: QueueItem[] = [];
const slowQueue: QueueItem[] = [];
let activeDownloads = 0;
let slowQueueTimer: ReturnType<typeof setTimeout> | null = null;
let lastSlowDownload = 0;

// Callbacks
let onProgress: ProgressCallback | null = null;
let onTrackReady: ((item: QueueItem) => void) | null = null;

// Generate unique ID
function generateId(): string {
  return Math.random().toString(36).substring(2, 10);
}

// Initialize ytdlp module
export async function initYtdlp(config: {
  enabled: boolean;
  command: string;
  ffmpegCommand: string;
  musicDir: string;
  fastQueueConcurrent: number;
  slowQueueInterval: number;
  allowPlaylists: boolean;
}): Promise<YtdlpStatus> {
  featureEnabled = config.enabled;
  ytdlpCommand = config.command;
  ffmpegCommand = config.ffmpegCommand;
  musicDir = config.musicDir;
  fastQueueConcurrent = config.fastQueueConcurrent;
  slowQueueInterval = config.slowQueueInterval;
  allowPlaylists = config.allowPlaylists;

  if (!featureEnabled) {
    console.log("[ytdlp] Feature disabled in config");
    return { available: false, enabled: false, version: null, ffmpeg: false };
  }

  // Check yt-dlp availability
  try {
    ytdlpVersion = await runCommand(ytdlpCommand, ["--version"]);
    ytdlpAvailable = true;
    console.log(`[ytdlp] Found yt-dlp version: ${ytdlpVersion.trim()}`);
  } catch (e) {
    console.error(`[ytdlp] yt-dlp not found (command: ${ytdlpCommand})`);
    ytdlpAvailable = false;
    featureEnabled = false;
  }

  // Check ffmpeg availability
  try {
    await runCommand(ffmpegCommand, ["-version"]);
    ffmpegAvailable = true;
    console.log("[ytdlp] ffmpeg available");
  } catch (e) {
    console.warn("[ytdlp] ffmpeg not found - audio extraction may fail");
    ffmpegAvailable = false;
  }

  // Load persisted slow queue from database
  if (featureEnabled) {
    const savedQueue = loadSlowQueue();
    for (const row of savedQueue) {
      slowQueue.push(rowToQueueItem(row));
    }
    if (savedQueue.length > 0) {
      console.log(`[ytdlp] Restored ${savedQueue.length} items from slow queue`);
    }
  }

  // Start slow queue processor
  if (featureEnabled) {
    startSlowQueueProcessor();
  }

  return getStatus();
}

// Convert database row to QueueItem
function rowToQueueItem(row: SlowQueueRow): QueueItem {
  return {
    id: row.id,
    url: row.url,
    title: row.title,
    userId: row.user_id,
    status: row.status as QueueItem["status"],
    progress: row.progress,
    queueType: "slow",
    error: row.error ?? undefined,
    createdAt: row.created_at * 1000,
    completedAt: row.completed_at ? row.completed_at * 1000 : undefined,
    playlistId: row.playlist_id ?? undefined,
    playlistName: row.playlist_name ?? undefined,
    position: row.position ?? undefined
  };
}

// Run a command and return stdout
function runCommand(cmd: string, args: string[]): Promise<string> {
  const fullCmd = `${cmd} ${args.join(" ")}`;
  console.log(`[ytdlp] Running: ${fullCmd}`);
  return new Promise((resolve, reject) => {
    const proc = spawn(cmd, args);
    let stdout = "";
    let stderr = "";
    proc.stdout.on("data", (data) => { stdout += data; });
    proc.stderr.on("data", (data) => { stderr += data; });
    proc.on("close", (code) => {
      console.log(`[ytdlp] Command exited with code ${code}`);
      if (code === 0) resolve(stdout);
      else reject(new Error(stderr || `Exit code ${code}`));
    });
    proc.on("error", reject);
  });
}

// Get current status
export function getStatus(): YtdlpStatus {
  return {
    available: ytdlpAvailable,
    enabled: featureEnabled,
    version: ytdlpVersion,
    ffmpeg: ffmpegAvailable
  };
}

// Check if feature is enabled and available
export function isAvailable(): boolean {
  return featureEnabled && ytdlpAvailable;
}

// Set progress callback
export function setProgressCallback(callback: ProgressCallback): void {
  onProgress = callback;
}

// Set track ready callback (called when download completes and needs playlist association)
export function setTrackReadyCallback(callback: (item: QueueItem) => void): void {
  onTrackReady = callback;
}

// Get all queue items
export function getQueues(): { fastQueue: QueueItem[]; slowQueue: QueueItem[]; slowQueueNextIn: number } {
  const now = Date.now();
  const nextIn = Math.max(0, Math.floor((lastSlowDownload + slowQueueInterval * 1000 - now) / 1000));
  return {
    fastQueue: [...fastQueue],
    slowQueue: [...slowQueue],
    slowQueueNextIn: nextIn
  };
}

// Get queue items for a specific user
export function getUserQueues(userId: number): { fastQueue: QueueItem[]; slowQueue: QueueItem[]; slowQueueNextIn: number } {
  const queues = getQueues();
  return {
    fastQueue: queues.fastQueue.filter(i => i.userId === userId),
    slowQueue: queues.slowQueue.filter(i => i.userId === userId),
    slowQueueNextIn: queues.slowQueueNextIn
  };
}

// Check URL and detect if it's a playlist
export async function checkUrl(url: string): Promise<PlaylistInfo | SingleVideoInfo> {
  const args = ["--flat-playlist", "--dump-json", "--no-warnings", url];
  const output = await runCommand(ytdlpCommand, args);
  
  // Parse JSON lines
  const lines = output.trim().split("\n").filter(l => l);
  
  if (lines.length === 0) {
    throw new Error("No video found");
  }
  
  if (lines.length === 1) {
    const data = JSON.parse(lines[0]);
    if (data._type === "playlist") {
      // It's a playlist with entries
      const items = (data.entries || []).map((e: any) => ({
        id: generateId(),
        url: e.url || e.webpage_url || `https://youtube.com/watch?v=${e.id}`,
        title: e.title || "Unknown"
      }));
      return {
        type: "playlist",
        title: data.title || "Playlist",
        count: items.length,
        items,
        requiresConfirmation: true
      };
    } else {
      // Single video
      return {
        type: "single",
        id: generateId(),
        title: data.title || "Unknown",
        url
      };
    }
  } else {
    // Multiple JSON lines = playlist
    const items = lines.map(line => {
      const data = JSON.parse(line);
      return {
        id: generateId(),
        url: data.url || data.webpage_url || url,
        title: data.title || "Unknown"
      };
    });
    return {
      type: "playlist",
      title: "Playlist",
      count: items.length,
      items,
      requiresConfirmation: true
    };
  }
}

// Add single video to fast queue
export function addToFastQueue(url: string, title: string, userId: number): QueueItem {
  const item: QueueItem = {
    id: generateId(),
    url,
    title,
    userId,
    status: "queued",
    progress: 0,
    queueType: "fast",
    createdAt: Date.now()
  };
  fastQueue.push(item);
  processNextFast();
  return item;
}

// Add items to slow queue (for playlists)
export function addToSlowQueue(
  items: { url: string; title: string }[],
  userId: number,
  playlist?: { id: string; name: string }
): QueueItem[] {
  const now = Date.now();
  const queueItems: QueueItem[] = items.map((item, index) => ({
    id: generateId(),
    url: item.url,
    title: item.title,
    userId,
    status: "queued" as const,
    progress: 0,
    queueType: "slow" as const,
    createdAt: now,
    playlistId: playlist?.id,
    playlistName: playlist?.name,
    position: playlist ? index : undefined
  }));
  
  // Persist to database
  for (const item of queueItems) {
    saveSlowQueueItem({
      id: item.id,
      url: item.url,
      title: item.title,
      userId: item.userId,
      status: item.status,
      progress: item.progress,
      playlistId: item.playlistId,
      playlistName: item.playlistName,
      position: item.position,
      createdAt: Math.floor(item.createdAt / 1000)
    });
  }
  
  slowQueue.push(...queueItems);
  return queueItems;
}

// Process next item in fast queue
function processNextFast(): void {
  if (activeDownloads >= fastQueueConcurrent) return;
  
  const item = fastQueue.find(i => i.status === "queued");
  if (!item) return;
  
  activeDownloads++;
  downloadItem(item).finally(() => {
    activeDownloads--;
    processNextFast();
  });
}

// Start slow queue processor
function startSlowQueueProcessor(): void {
  if (slowQueueTimer) return;
  
  const processNext = () => {
    const item = slowQueue.find(i => i.status === "queued");
    if (item) {
      lastSlowDownload = Date.now();
      downloadItem(item).finally(() => {
        slowQueueTimer = setTimeout(processNext, slowQueueInterval * 1000);
      });
    } else {
      slowQueueTimer = setTimeout(processNext, 5000); // Check again in 5s
    }
  };
  
  // Start immediately if there are items
  const hasQueued = slowQueue.some(i => i.status === "queued");
  if (hasQueued) {
    processNext();
  } else {
    slowQueueTimer = setTimeout(processNext, 5000);
  }
}

// Download a single item
async function downloadItem(item: QueueItem): Promise<void> {
  item.status = "downloading";
  item.progress = 0;
  notifyProgress(item);
  
  console.log(`[ytdlp] Starting download: ${item.title} (${item.url})`);
  
  try {
    const outputTemplate = join(musicDir, "%(title)s.%(ext)s");
    const args = [
      "-x",
      "--audio-format", "mp3",
      "-o", outputTemplate,
      "--progress",
      "--newline",
      "--no-warnings",
      item.url
    ];
    
    const fullCmd = `${ytdlpCommand} ${args.join(" ")}`;
    console.log(`[ytdlp] Running: ${fullCmd}`);
    
    await new Promise<void>((resolve, reject) => {
      const proc = spawn(ytdlpCommand, args);
      
      proc.stdout.on("data", (data) => {
        const line = data.toString();
        console.log(`[ytdlp] ${line.trim()}`);
        // Parse progress from yt-dlp output
        const match = line.match(/(\d+\.?\d*)%/);
        if (match) {
          item.progress = parseFloat(match[1]);
          notifyProgress(item);
        }
      });
      
      proc.stderr.on("data", (data) => {
        console.error(`[ytdlp] stderr: ${data}`);
      });
      
      proc.on("close", (code) => {
        console.log(`[ytdlp] Download finished with code ${code}`);
        if (code === 0) resolve();
        else reject(new Error(`yt-dlp exited with code ${code}`));
      });
      
      proc.on("error", reject);
    });
    
    console.log(`[ytdlp] Complete: ${item.title}`);
    item.status = "complete";
    item.progress = 100;
    item.completedAt = Date.now();
    
    // Update database
    if (item.queueType === "slow") {
      updateSlowQueueItem(item.id, {
        status: "complete",
        progress: 100,
        completedAt: Math.floor(item.completedAt / 1000)
      });
      
      // Register for playlist addition immediately - library will match when it scans
      if (item.playlistId && onTrackReady) {
        onTrackReady(item);
      }
    }
    
    notifyProgress(item);
    
    // Remove from queue after delay
    setTimeout(() => removeFromQueue(item), 5000);
    
  } catch (e: any) {
    item.status = "error";
    item.error = e.message || "Download failed";
    item.completedAt = Date.now();
    
    // Update database
    if (item.queueType === "slow") {
      updateSlowQueueItem(item.id, {
        status: "error",
        error: item.error,
        completedAt: Math.floor(item.completedAt / 1000)
      });
    }
    
    notifyProgress(item);
    
    // Remove from queue after delay
    setTimeout(() => removeFromQueue(item), 10000);
  }
}

// Remove item from queue
function removeFromQueue(item: QueueItem): void {
  if (item.queueType === "fast") {
    const idx = fastQueue.findIndex(i => i.id === item.id);
    if (idx !== -1) fastQueue.splice(idx, 1);
  } else {
    const idx = slowQueue.findIndex(i => i.id === item.id);
    if (idx !== -1) slowQueue.splice(idx, 1);
    // Remove from database
    deleteSlowQueueItem(item.id);
  }
}

// Cancel a slow queue item
export function cancelSlowQueueItem(id: string, userId: number): boolean {
  const item = slowQueue.find(i => i.id === id && i.userId === userId);
  if (!item || item.status === "downloading") {
    return false; // Can't cancel if not found, not owned, or already downloading
  }
  
  item.status = "cancelled";
  item.completedAt = Date.now();
  
  // Update database
  updateSlowQueueItem(id, {
    status: "cancelled",
    completedAt: Math.floor(item.completedAt / 1000)
  });
  
  notifyProgress(item);
  
  // Remove from queue after brief delay
  setTimeout(() => removeFromQueue(item), 1000);
  
  return true;
}

// Notify progress callback
function notifyProgress(item: QueueItem): void {
  if (onProgress) {
    onProgress(item);
  }
}

// Mark a slow queue item as skipped (already in library)
export function skipSlowQueueItem(id: string, trackId: string): QueueItem | null {
  const item = slowQueue.find(i => i.id === id && i.status === "queued");
  if (!item) return null;
  
  item.status = "complete";
  item.progress = 100;
  item.completedAt = Date.now();
  item.trackId = trackId;
  
  // Update database
  updateSlowQueueItem(id, {
    status: "complete",
    progress: 100,
    completedAt: Math.floor(item.completedAt / 1000)
  });
  
  notifyProgress(item);
  
  // Remove from queue after brief delay
  setTimeout(() => removeFromQueue(item), 1000);
  
  return item;
}

// Get queued items from slow queue (for prescan)
export function getQueuedSlowItems(): QueueItem[] {
  return slowQueue.filter(i => i.status === "queued");
}

// Cleanup old completed/failed items
export function cleanupOldItems(maxAge: number = 3600000): void {
  const now = Date.now();
  const cleanup = (queue: QueueItem[]) => {
    for (let i = queue.length - 1; i >= 0; i--) {
      const item = queue[i];
      if ((item.status === "complete" || item.status === "error" || item.status === "cancelled") && 
          now - item.createdAt > maxAge) {
        queue.splice(i, 1);
      }
    }
  };
  cleanup(fastQueue);
  cleanup(slowQueue);
  
  // Also cleanup database
  clearCompletedSlowQueue(Math.floor(maxAge / 1000));
}