// helpers/hlsManager.js const fs = require('fs'); const path = require('path'); const { spawn, execSync } = require('child_process'); const fetch = require('node-fetch'); // node-fetch@2 const config = require('../api_v1/config'); const HLS_BASE = path.resolve(process.env.HLS_BASE || path.join(process.cwd(), 'hls_output')); const jobs = new Map(); // jobId -> job object function hasRkmpp() { try { const out = execSync('ffmpeg -encoders', { stdio: ['ignore', 'pipe', 'ignore'] }).toString(); return out.includes('h264_rkmpp'); } catch (e) { return false; } } function buildFfmpegArgs(inputPath, jobDir, useRkmpp) { const master = path.join(jobDir, 'master.m3u8'); const segPattern = path.join(jobDir, 'segment_%05d.ts'); if (useRkmpp) { return [ '-hide_banner', '-y', '-probesize', '5000000', '-analyzeduration', '10000000', '-hwaccel', 'rkmpp', '-i', inputPath, '-map', '0:v:0', '-map', '0:a:0', '-c:v', 'h264_rkmpp', '-b:v', '5000k', '-maxrate', '6000k', '-bufsize', '12000k', '-r', '30', '-g', '90', '-keyint_min', '90', '-pix_fmt', 'nv12', '-c:a', 'aac', '-b:a', '160k', '-ac', '2', '-ar', '48000', '-f', 'hls', '-hls_time', '3', '-hls_list_size', '0', '-hls_playlist_type', 'vod', '-hls_segment_type', 'mpegts', '-hls_segment_filename', segPattern, master ]; } else { return [ '-hide_banner', '-y', '-probesize', '5000000', '-analyzeduration', '10000000', '-i', inputPath, '-map', '0:v:0', '-map', '0:a:0', '-c:v', 'libx264', '-profile:v', 'high', '-level', '4.1', '-preset', 'veryfast', '-tune', 'zerolatency', '-b:v', '5000k', '-maxrate', '6000k', '-bufsize', '12000k', '-r', '30', '-g', '90', '-keyint_min', '90', '-sc_threshold', '0', '-pix_fmt', 'yuv420p', '-c:a', 'aac', '-b:a', '160k', '-ac', '2', '-ar', '48000', '-f', 'hls', '-hls_time', '3', '-hls_list_size', '0', '-hls_playlist_type', 'vod', '-hls_segment_type', 'mpegts', '-hls_segment_filename', segPattern, master ]; } } async function sendCallback(job, payload) { if (!job.callbackUrl) return; try { const resp = await fetch(job.callbackUrl, { method: 'POST', headers: { 'Content-Type': 'application/json' }, body: JSON.stringify(payload), timeout: 10000 }); console.log(`[job ${job.jobId}] callback ${resp.status}`); } catch (err) { console.error(`[job ${job.jobId}] callback failed: ${err.message}`); } } /** * startJob(options) * options: * - inputPath: absolute path to source file * - outDir: parent directory where job subfolder will be created (jobDir = outDir/) * - segmentsNeeded: number of .ts segments required before notifying callback * - callbackUrl: URL to POST readiness notifications * - publicBaseUrl: optional base URL to build public master.m3u8 URL (fallback to config.BASE_URL) * * Returns jobId (string) */ function startJob({ inputPath, outDir, segmentsNeeded = 3, callbackUrl, publicBaseUrl = null }) { const jobId = Date.now().toString(); const jobDir = path.join(outDir, jobId); fs.mkdirSync(jobDir, { recursive: true }); const useRkmpp = hasRkmpp(); const ffArgs = buildFfmpegArgs(inputPath, jobDir, useRkmpp); const proc = spawn('ffmpeg', ffArgs, { stdio: ['ignore', 'pipe', 'pipe'] }); const job = { jobId, inputPath, outDir: jobDir, callbackUrl, segmentsNeeded: Number(segmentsNeeded), proc, notified: false, startTime: Date.now(), useRkmpp, publicBaseUrl }; jobs.set(jobId, job); proc.stderr.setEncoding('utf8'); proc.stderr.on('data', d => console.log(`[ffmpeg ${jobId}] ${d.toString().trim()}`)); proc.stdout.setEncoding('utf8'); proc.stdout.on('data', d => console.log(`[ffmpeg ${jobId}] ${d.toString().trim()}`)); const masterPath = path.join(jobDir, 'master.m3u8'); const pollInterval = 500; const timeoutMs = 5 * 60 * 1000; // 5 minutes const poller = setInterval(async () => { try { if (!job.notified && fs.existsSync(masterPath)) { const files = fs.readdirSync(jobDir); const tsCount = files.filter(f => f.endsWith('.ts')).length; if (tsCount >= job.segmentsNeeded) { job.notified = true; // costruisci publicUrl relativo a HLS_BASE const base = job.publicBaseUrl || config.BASE_URL || null; const rel = path.relative(HLS_BASE, masterPath).split(path.sep).join('/'); const publicUrl = base ? `${base.replace(/\/$/, '')}/hls/${rel}` : `/${rel}`; await sendCallback(job, { jobId: job.jobId, master_m3u8: publicUrl, ready_segments: tsCount, status: 'ready' }); } } if (!job.notified && (Date.now() - job.startTime) > timeoutMs) { job.notified = true; await sendCallback(job, { jobId: job.jobId, out_dir: job.outDir, status: 'timeout' }); } } catch (err) { console.error(`[job ${jobId}] poller error: ${err.message}`); } }, pollInterval); job.poller = poller; proc.on('exit', async (code) => { clearInterval(poller); try { const files = fs.existsSync(job.outDir) ? fs.readdirSync(job.outDir) : []; const tsCount = files.filter(f => f.endsWith('.ts')).length; if (!job.notified) { const base = job.publicBaseUrl || config.BASE_URL || null; const rel = path.relative(HLS_BASE, masterPath).split(path.sep).join('/'); const publicUrl = (base && fs.existsSync(masterPath)) ? `${base.replace(/\/$/, '')}/hls/${rel}` : (fs.existsSync(masterPath) ? `/${rel}` : null); await sendCallback(job, { jobId: job.jobId, master_m3u8: publicUrl, ready_segments: tsCount, status: code === 0 ? 'finished' : 'error', returncode: code }); } } catch (err) { console.error(`[job ${jobId}] on-exit callback error: ${err.message}`); } finally { jobs.delete(jobId); } }); return jobId; } function getJob(jobId) { return jobs.get(jobId) || null; } module.exports = { startJob, getJob };