import { randomUUID } from 'node:crypto'; import { mkdir, readdir, rm } from 'node:fs/promises'; import { join } from 'node:path'; import type { FastifyBaseLogger } from 'fastify'; import { type DownloadMode, buildDownloadArgs, expectedExtension, parseProgress, } from './downloadPlan.js'; import { runYtDlp } from './ytDlpRunner.js'; /** État d'avancement d'un élément de la file d'attente. */ export type QueueItemStatus = 'pending' | 'running' | 'completed' | 'failed' | 'cancelled'; /** Vue publique (immuable) d'un élément de la file d'attente. */ export interface QueueItem { /** Identifiant unique de l'élément. */ readonly id: string; /** URL de la vidéo source. */ readonly url: string; /** Titre de la vidéo, si connu au moment de l'ajout (purement informatif, pour l'affichage). */ readonly title: string | null; /** Mode de téléchargement demandé. */ readonly mode: DownloadMode; /** État d'avancement courant. */ readonly status: QueueItemStatus; /** Progression estimée (0-100), au mieux selon la sortie yt-dlp. */ readonly progress: number; /** Message d'erreur si `status` vaut `"failed"`, sinon `null`. */ readonly error: string | null; /** Nom du fichier produit une fois `status` à `"completed"`, sinon `null`. */ readonly fileName: string | null; /** Date de création de l'élément, au format ISO 8601. */ readonly createdAt: string; } /** Délai maximal accordé à un téléchargement avant arrêt forcé (30 minutes). */ const DOWNLOAD_TIMEOUT_MS = 30 * 60_000; interface InternalQueueItem { id: string; url: string; title: string | null; mode: DownloadMode; status: QueueItemStatus; progress: number; error: string | null; fileName: string | null; createdAt: string; outputDir: string; abortController: AbortController | null; } /** * Recherche, dans un dossier, le premier fichier dont l'extension * correspond à celle attendue pour un mode de téléchargement donné. * * @param dir - Dossier à inspecter. * @param expectedExt - Extension attendue (sans le point). * @returns Le nom du fichier trouvé, ou `null` si aucun ne correspond. */ async function findOutputFile(dir: string, expectedExt: string): Promise { const entries = await readdir(dir); const suffix = `.${expectedExt.toLowerCase()}`; return entries.find((name) => name.toLowerCase().endsWith(suffix)) ?? null; } /** * File d'attente de téléchargements : accepte des éléments, les exécute * avec une concurrence bornée via yt-dlp, et expose leur progression. * * Chaque élément obtient son propre sous-dossier sous `baseDir`, supprimé * après échec/annulation ou après livraison du fichier au client (voir * {@link QueueService.consumeCompletedFile}). */ export class QueueService { private readonly logger: FastifyBaseLogger; private readonly baseDir: string; private readonly maxConcurrent: number; private readonly items = new Map(); private readonly pendingIds: string[] = []; private runningCount = 0; /** * @param logger - Logger utilisé pour tracer le cycle de vie des éléments. * @param baseDir - Dossier racine des sorties de téléchargement (un sous-dossier par élément). * @param maxConcurrent - Nombre maximal de téléchargements exécutés simultanément. */ public constructor(logger: FastifyBaseLogger, baseDir: string, maxConcurrent: number) { this.logger = logger; this.baseDir = baseDir; this.maxConcurrent = maxConcurrent; } /** * Ajoute un élément à la file d'attente et déclenche son exécution dès * qu'un emplacement de concurrence se libère. * * @param url - URL de la vidéo à télécharger (déjà validée par l'appelant). * @param mode - Mode de téléchargement demandé. * @param title - Titre de la vidéo, si déjà connu (ex. depuis une recherche de chaîne), pour l'affichage uniquement. * @returns L'élément créé, à l'état `"pending"`. */ public enqueue(url: string, mode: DownloadMode, title: string | null = null): QueueItem { const id = randomUUID(); const item: InternalQueueItem = { id, url, title, mode, status: 'pending', progress: 0, error: null, fileName: null, createdAt: new Date().toISOString(), outputDir: join(this.baseDir, id), abortController: null, }; this.items.set(id, item); this.pendingIds.push(id); this.logger.info({ id, url, mode: mode.kind }, "File d'attente : élément ajouté"); this.pump(); return this.toPublic(item); } /** * @returns Tous les éléments de la file, dans leur ordre d'ajout. */ public list(): QueueItem[] { return [...this.items.values()].map((item) => this.toPublic(item)); } /** * @param id - Identifiant de l'élément recherché. * @returns L'élément correspondant, ou `null` s'il n'existe pas. */ public get(id: string): QueueItem | null { const item = this.items.get(id); return item ? this.toPublic(item) : null; } /** * Retire un élément de la file, quel que soit son état : s'il est en * attente ou en cours, il est d'abord arrêté (le processus yt-dlp est * annulé le cas échéant) ; dans tous les cas l'entrée est supprimée de * la liste et son dossier de sortie éventuel nettoyé. * * @param id - Identifiant de l'élément à retirer. * @returns `true` si l'élément existait, `false` sinon. */ public remove(id: string): boolean { const item = this.items.get(id); if (!item) { return false; } if (item.status === 'pending') { const index = this.pendingIds.indexOf(id); if (index !== -1) { this.pendingIds.splice(index, 1); } } else if (item.status === 'running') { item.abortController?.abort(); } this.items.delete(id); this.cleanupDir(item).catch((error: unknown) => { this.logger.error( { err: error, id }, "File d'attente : échec du nettoyage lors de la suppression", ); }); this.logger.info({ id, status: item.status }, "File d'attente : élément supprimé"); return true; } /** * Construit la livraison d'un fichier terminé : son chemin absolu, et un * `onSent` qui nettoie son dossier de sortie une fois appelé (le fichier * a été livré, ou l'envoi a échoué). Le serveur ne conserve donc jamais * durablement les fichiers : ils sont livrés au navigateur puis * nettoyés. * * @param item - Élément terminé dont le fichier doit être livré. * @param fileName - Nom du fichier à livrer (déjà vérifié non nul par l'appelant). * @returns Le chemin absolu du fichier, son nom, et le callback de nettoyage post-livraison. */ private buildDelivery( item: InternalQueueItem, fileName: string, ): { path: string; fileName: string; onSent: () => Promise } { return { path: join(item.outputDir, fileName), fileName, onSent: () => this.cleanupDir(item) .catch((error: unknown) => { this.logger.error( { err: error, id: item.id }, "File d'attente : échec du nettoyage après livraison", ); }) .finally(() => { // Le fichier vient d'être envoyé (et son dossier nettoyé, au // mieux) : `fileName` ne doit plus désigner un fichier // consommable, sous peine qu'un second appel tente de lire un // fichier qui n'existe plus. item.fileName = null; }), }; } /** * Retourne le chemin absolu du fichier produit par un élément terminé, * et programme la suppression de son dossier de sortie une fois `onSent` * résolu. * * @param id - Identifiant de l'élément terminé. * @returns Le chemin absolu du fichier et son nom, ou `null` si l'élément n'est pas terminé. */ public consumeCompletedFile( id: string, ): { path: string; fileName: string; onSent: () => Promise } | null { const item = this.items.get(id); if (!item || item.status !== 'completed' || !item.fileName) { return null; } return this.buildDelivery(item, item.fileName); } /** * Retourne tous les éléments terminés dont le fichier est encore * disponible, chacun avec son propre `onSent` (mêmes règles de nettoyage * que {@link consumeCompletedFile}, appliquées indépendamment par * fichier) - destiné au téléchargement groupé en zip. * * @returns La liste des fichiers disponibles au téléchargement, dans l'ordre d'ajout. */ public consumeAllCompletedFiles(): Array<{ path: string; fileName: string; onSent: () => Promise; }> { const deliveries: Array<{ path: string; fileName: string; onSent: () => Promise }> = []; for (const item of this.items.values()) { if (item.status === 'completed' && item.fileName) { deliveries.push(this.buildDelivery(item, item.fileName)); } } return deliveries; } /** * Lance autant d'éléments en attente que la concurrence disponible le * permet. */ private pump(): void { while (this.runningCount < this.maxConcurrent && this.pendingIds.length > 0) { const id = this.pendingIds.shift(); if (!id) { break; } const item = this.items.get(id); if (!item) { continue; } this.runningCount += 1; this.runItem(item) .catch((error: unknown) => { this.logger.error( { err: error, id }, "File d'attente : erreur inattendue durant l'exécution", ); }) .finally(() => { this.runningCount -= 1; this.pump(); }); } } /** * Exécute le téléchargement d'un élément et met à jour son état au fil * de l'avancement. * * @param item - Élément interne à exécuter. */ private async runItem(item: InternalQueueItem): Promise { item.status = 'running'; item.abortController = new AbortController(); this.logger.info( { id: item.id, url: item.url }, 'File d’attente : démarrage du téléchargement', ); try { await mkdir(item.outputDir, { recursive: true }); const args = buildDownloadArgs(item.url, item.mode, item.outputDir); await runYtDlp(args, { timeoutMs: DOWNLOAD_TIMEOUT_MS, signal: item.abortController.signal, onStdoutLine: (line) => { const progress = parseProgress(line); if (progress !== null) { item.progress = progress; } }, }); const fileName = await findOutputFile(item.outputDir, expectedExtension(item.mode)); if (!fileName) { throw new Error('Fichier de sortie introuvable après le téléchargement'); } item.fileName = fileName; item.progress = 100; item.status = 'completed'; this.logger.info({ id: item.id, fileName }, 'File d’attente : téléchargement terminé'); } catch (error) { if (error instanceof Error && error.name === 'AbortError') { item.status = 'cancelled'; this.logger.info({ id: item.id }, 'File d’attente : téléchargement annulé'); } else { item.status = 'failed'; item.error = error instanceof Error ? error.message : String(error); this.logger.error({ err: error, id: item.id }, 'File d’attente : échec du téléchargement'); } try { await this.cleanupDir(item); } catch (cleanupError) { this.logger.error( { err: cleanupError, id: item.id }, "File d'attente : échec du nettoyage après erreur", ); } } finally { item.abortController = null; } } /** * Supprime le dossier de sortie d'un élément. * * @param item - Élément dont le dossier de sortie doit être supprimé. */ private cleanupDir(item: InternalQueueItem): Promise { return rm(item.outputDir, { recursive: true, force: true }); } /** * Convertit un élément interne en vue publique immuable. * * @param item - Élément interne à exposer. * @returns Une copie publique de l'élément. */ private toPublic(item: InternalQueueItem): QueueItem { return { id: item.id, url: item.url, title: item.title, mode: item.mode, status: item.status, progress: item.progress, error: item.error, fileName: item.fileName, createdAt: item.createdAt, }; } }