import 'dart:async'; import 'dart:convert'; import 'dart:io'; import '../utils/log_redactor.dart'; /// One running FFmpeg transcoding session (live, VOD or recording playback). class FfmpegSession { final String id; final Process process; final Directory dir; final bool isLive; final DateTime startedAt = DateTime.now(); DateTime lastAccess = DateTime.now(); bool exited = false; int? exitCode; /// Last stderr lines, kept for fast-fail diagnostics. final List recentStderr = []; FfmpegSession({ required this.id, required this.process, required this.dir, required this.isLive, }); void touch() => lastAccess = DateTime.now(); } /// Registry of FFmpeg transcoding processes. /// /// Responsibilities: /// - one process per session id (`live_{id}_{quality}`, `vod_{id}`, ...) /// - reaper kills sessions idle beyond a TTL (no viewer fetching segments) /// - startup wipe of the HLS temp dir (orphan dirs after a crash) /// - SIGTERM/SIGINT hook so `docker stop` leaves no orphan ffmpeg class FfmpegSessionManager { final Directory baseDir; final Map _sessions = {}; Timer? _reaper; static const liveIdleTimeout = Duration(minutes: 4); static const vodIdleTimeout = Duration(minutes: 15); /// Live playlist not rewritten for this long => FFmpeg is wedged. static const liveStallTimeout = Duration(seconds: 45); FfmpegSessionManager(this.baseDir); /// Wipe orphan session dirs and start the reaper. Call once at startup. Future init() async { if (baseDir.existsSync()) { try { baseDir.deleteSync(recursive: true); } catch (e) { print('[FFmpegManager] Could not wipe temp dir: $e'); } } baseDir.createSync(recursive: true); _reaper = Timer.periodic(const Duration(seconds: 60), (_) => _reap()); // L'arrêt propre (docker stop / Ctrl+C) est orchestré par server.dart : // il clôture d'abord les enregistrements puis appelle killAll(). Un // handler local qui ferait exit(0) immédiatement court-circuiterait // cette clôture. } FfmpegSession? get(String id) => _sessions[id]; /// Marks a session as recently used (call from playlist AND segment routes). void touch(String id) => _sessions[id]?.touch(); bool contains(String id) => _sessions.containsKey(id); /// Vrai si [getOrStart] lancerait un nouveau processus FFmpeg pour [id] — /// donc une nouvelle connexion au fournisseur. bool needsStart(String id, {required bool isLive}) { final existing = _sessions[id]; if (existing == null) return true; if (!existing.exited) return false; // A VOD/recording transcode that finished cleanly is still fully // playable from its segments — reuse it instead of re-transcoding. return isLive || existing.exitCode != 0 || !_playlistComplete(existing); } /// Returns the existing healthy session or starts a new FFmpeg process. /// /// [argsBuilder] receives the session working directory and returns the /// FFmpeg argument list. The session directory is recreated for new /// sessions. Future getOrStart({ required String id, required bool isLive, required String ffmpegPath, required List Function(Directory dir) argsBuilder, }) async { final existing = _sessions[id]; if (existing != null && !needsStart(id, isLive: isLive)) { existing.touch(); return existing; } if (existing != null) { // Process died: clean up before restarting killSession(id); } final dir = Directory('${baseDir.path}/$id'); if (dir.existsSync()) dir.deleteSync(recursive: true); dir.createSync(recursive: true); final args = argsBuilder(dir); final process = await Process.start( ffmpegPath, args, workingDirectory: dir.path, ); final session = FfmpegSession( id: id, process: process, dir: dir, isLive: isLive, ); _sessions[id] = session; process.stderr.transform(utf8.decoder).listen((data) { session.recentStderr.add(data); if (session.recentStderr.length > 20) session.recentStderr.removeAt(0); // FFmpeg rappelle l'URL d'entrée (« Input #0 … from 'http://…' ») : // masquer les identifiants Xtream avant de journaliser. print('[FFmpeg $id] ${LogRedactor.redactUrl(data)}'); }); process.exitCode.then((code) { session.exited = true; session.exitCode = code; print('[FFmpegManager] Session $id exited with code $code'); }); return session; } bool _playlistComplete(FfmpegSession session) { final playlistFile = File('${session.dir.path}/playlist.m3u8'); if (!playlistFile.existsSync()) return false; try { return playlistFile.readAsStringSync().contains('#EXT-X-ENDLIST'); } catch (_) { return false; } } /// Waits until the session's playlist references at least [minSegments] /// segments. Fails fast when the process dies before producing output, /// returning the recent stderr for diagnostics. /// /// [minSegments] > 1 donne au lecteur une avance de démarrage : servir la /// playlist dès le premier segment le fait partir avec 4 s de marge sur un /// encodeur qui n'a pas fini — la moindre hésitation coupe la lecture. La /// playlist déjà close (`#EXT-X-ENDLIST`) est renvoyée telle quelle, quel /// que soit son nombre de segments : rien de plus n'arrivera. Future<({bool ready, String? error})> waitForPlaylist( FfmpegSession session, { Duration timeout = const Duration(seconds: 30), int minSegments = 1, }) async { final playlistFile = File('${session.dir.path}/playlist.m3u8'); final deadline = DateTime.now().add(timeout); while (DateTime.now().isBefore(deadline)) { if (playlistFile.existsSync()) { final content = playlistFile.readAsStringSync(); final segments = '.ts'.allMatches(content).length; if (segments >= minSegments || (segments > 0 && content.contains('#EXT-X-ENDLIST'))) { return (ready: true, error: null); } } if (session.exited) { // Un encodeur qui s'est terminé proprement en ayant produit moins de // segments que demandé a simplement fini : c'est un succès. if (session.exitCode == 0 && playlistFile.existsSync() && playlistFile.readAsStringSync().contains('.ts')) { return (ready: true, error: null); } return ( ready: false, error: 'FFmpeg exited (${session.exitCode}): ' '${session.recentStderr.join().trim()}' ); } // 100 ms : à 500 ms, on ajoutait en moyenne un quart de seconde de // latence pure entre la disponibilité de la playlist et sa réponse. await Future.delayed(const Duration(milliseconds: 100)); } // Le délai est écoulé : mieux vaut servir ce qui existe que rien. if (playlistFile.existsSync() && playlistFile.readAsStringSync().contains('.ts')) { return (ready: true, error: null); } return (ready: false, error: 'Timeout waiting for transcoder'); } void killSession(String id) { final session = _sessions.remove(id); if (session == null) return; try { session.process.kill(ProcessSignal.sigterm); } catch (_) {} try { if (session.dir.existsSync()) session.dir.deleteSync(recursive: true); } catch (e) { print('[FFmpegManager] Could not delete dir for $id: $e'); } } /// Tue les sessions frères d'un même contenu (`rec__t0`, `rec__t600`…) /// que plus personne ne consomme depuis [idleFor]. /// /// Chaque saut hors de la zone déjà transcodée démarre un FFmpeg à un /// nouvel offset ; sans ce ménage, une poignée d'allers-retours dans la /// barre de progression laisserait autant de processus vivants jusqu'au /// TTL de 15 minutes. Le délai d'inactivité protège un second spectateur /// du même enregistrement : sa session est « touchée » à chaque segment. void killIdleSiblings( String prefix, { required String keep, Duration idleFor = const Duration(seconds: 30), }) { final now = DateTime.now(); for (final session in _sessions.values.toList()) { if (session.id == keep || !session.id.startsWith(prefix)) continue; if (now.difference(session.lastAccess) < idleFor) continue; print('[FFmpegManager] Dropping stale sibling ${session.id}'); killSession(session.id); } } void killAll() { print('[FFmpegManager] Killing ${_sessions.length} session(s)'); for (final id in _sessions.keys.toList()) { killSession(id); } _reaper?.cancel(); } void _reap() { final now = DateTime.now(); for (final session in _sessions.values.toList()) { final timeout = session.isLive ? liveIdleTimeout : vodIdleTimeout; final idle = now.difference(session.lastAccess); // Exited sessions keep their segments until the idle timeout: a // finished VOD transcode is usually still being watched, and killing // it here would delete the segments mid-playback. if (idle > timeout) { print( '[FFmpegManager] Reaping idle session ${session.id} ' '(idle ${idle.inSeconds}s)', ); killSession(session.id); continue; } // Watchdog: a live FFmpeg that is running but has not updated its // playlist recently is wedged on a stalled upstream. Kill it so the // player's next playlist poll restarts the transcoder. if (session.isLive && !session.exited && _isStalled(session, now)) { print('[FFmpegManager] Restarting stalled live session ${session.id}'); killSession(session.id); } } } /// A live session is stalled when its playlist exists but has not been /// rewritten for [liveStallTimeout] (FFmpeg rewrites it on every segment). bool _isStalled(FfmpegSession session, DateTime now) { final playlistFile = File('${session.dir.path}/playlist.m3u8'); if (!playlistFile.existsSync()) { // Never produced output: give it until liveStallTimeout after start. return now.difference(session.startedAt) > liveStallTimeout; } try { return now.difference(playlistFile.lastModifiedSync()) > liveStallTimeout; } catch (_) { return false; } } }