<?php

declare(strict_types=1);

class TranscodeManager
{
    private static ?string $_ffmpegPath = null;
    private static ?string $_ffprobePath = null;
    private static ?array $_formats = null;

    public static function getFormats(): array
    {
        if (self::$_formats !== null) return self::$_formats;
        try {
            $rows = Database::fetchAll("SELECT * FROM video_formats WHERE is_active = 1 ORDER BY sort_order, id");
        } catch (\Throwable) {
            $rows = [];
        }
        $formats = [];
        foreach ($rows as $row) {
            $task = trim(pathinfo($row['postfix'], PATHINFO_FILENAME), '_');
            if ($task === '') continue;
            $formats[$task] = [
                'w' => (int)$row['width'],
                'h' => (int)$row['height'],
                'file' => 'video' . $row['postfix'],
                'keep_aspect' => (bool)$row['keep_aspect'],
                'ffmpeg_options' => $row['ffmpeg_options'] ?? '',
                'is_source' => (bool)$row['is_source'],
            ];
        }
        if (empty($formats)) {
            $formats = [
                '720p' => ['w' => 1280, 'h' => 720, 'file' => 'video_720p.mp4', 'keep_aspect' => true, 'ffmpeg_options' => '', 'is_source' => false],
                '480p' => ['w' => 854, 'h' => 480, 'file' => 'video_480p.mp4', 'keep_aspect' => true, 'ffmpeg_options' => '', 'is_source' => false],
                '360p' => ['w' => 640, 'h' => 360, 'file' => 'video_360p.mp4', 'keep_aspect' => true, 'ffmpeg_options' => '', 'is_source' => false],
            ];
        }
        self::$_formats = $formats;
        return $formats;
    }

    public static function findSource(string $dir): ?string
    {
         
        $best = null;
        $bestH = -1;
        foreach (glob($dir . '/video_*p.mp4') as $path) {
            if (preg_match('/video_(\d+)p\.mp4$/', $path, $m) && filesize($path) > 1000) {
                $h = (int)$m[1];
                if ($h > $bestH) { $best = $path; $bestH = $h; }
            }
        }
         
        if (!$best) {
            foreach (['video.mp4', 'video.webm', 'video.ogg'] as $f) {
                $path = $dir . '/' . $f;
                if (file_exists($path) && filesize($path) > 1000) { $best = $path; break; }
            }
        }
        return $best;
    }

    public static function isEnabled(): bool
    {
        return Setting::get('transcode_enabled', '0') === '1';
    }

    public static function canProcess(): bool
    {
        if (self::isAvailable()) return true;
        $servers = Database::fetchOne("SELECT id FROM transcode_servers WHERE is_active = 1 LIMIT 1");
        return (bool)$servers;
    }

    public static function needsSourceReencode(): bool
    {
        $hasWatermark = Setting::get('watermark_enabled', '0') === '1' && Setting::get('watermark_image', '') !== '';
        $hasTrim = (int)Setting::get('transcode_trim_start', '0') > 0;
        return $hasWatermark || $hasTrim;
    }

    public static function ffmpegPath(): string
    {
        if (self::$_ffmpegPath === null) {
            $configured = Setting::get('transcode_ffmpeg_path', '');
            if ($configured !== '' && is_executable($configured)) {
                self::$_ffmpegPath = $configured;
            } else {
                self::$_ffmpegPath = self::detectBinary('ffmpeg');
            }
        }
        return self::$_ffmpegPath;
    }

    public static function ffprobePath(): string
    {
        if (self::$_ffprobePath === null) {
            $configured = Setting::get('transcode_ffprobe_path', '');
            if ($configured !== '' && is_executable($configured)) {
                self::$_ffprobePath = $configured;
            } else {
                self::$_ffprobePath = self::detectBinary('ffprobe');
            }
        }
        return self::$_ffprobePath;
    }

    public static function isAvailable(): bool
    {
        return self::ffmpegPath() !== '';
    }

    private static function detectBinary(string $name): string
    {
        $paths = ["/usr/bin/{$name}", "/usr/local/bin/{$name}", "/opt/homebrew/bin/{$name}"];
        foreach ($paths as $p) {
            if (is_executable($p)) return $p;
        }
        $which = @shell_exec("which {$name} 2>/dev/null");
        if ($which) {
            $path = trim($which);
            if (is_executable($path)) return $path;
        }
        return '';
    }

    public static function getVideoInfo(string $filePath): ?array
    {
        $probe = self::ffprobePath();
        if ($probe === '' || !file_exists($filePath)) return null;

        $cmd = escapeshellcmd($probe) . ' -v quiet -print_format json -show_format -show_streams ' . escapeshellarg($filePath);
        $json = @shell_exec($cmd . ' 2>/dev/null');
        if (!$json) return null;

        $data = json_decode($json, true);
        if (!$data) return null;

        $videoStream = null;
        foreach (($data['streams'] ?? []) as $s) {
            if (($s['codec_type'] ?? '') === 'video') { $videoStream = $s; break; }
        }
        if (!$videoStream) return null;

        return [
            'width' => (int)($videoStream['width'] ?? 0),
            'height' => (int)($videoStream['height'] ?? 0),
            'duration' => (int)(float)($data['format']['duration'] ?? $videoStream['duration'] ?? 0),
            'codec' => $videoStream['codec_name'] ?? '',
            'bitrate' => (int)($data['format']['bit_rate'] ?? 0),
        ];
    }

    public static function queue(int $videoId, ?array $onlyTasks = null, bool $keepSource = true): int
    {
        if (!self::isEnabled()) return 0;

        $video = Database::fetchOne('SELECT id, video_folder, storage_id, embed_code FROM videos WHERE id = ?', [$videoId]);
        if (!$video || empty($video['video_folder'])) return 0;
        if (!empty($video['embed_code'])) return 0;

        $folder = $video['video_folder'];
        $storageId = $video['storage_id'] ? (int)$video['storage_id'] : null;
        $dir = VideoAsset::dir($folder);
        $source = self::findSource($dir);
        $isRemote = $storageId && MediaStorage::isRemote($storageId);

        if (!$source && !$isRemote) return 0;

        $sourceHeight = 1080;
        if ($source) {
            $info = self::getVideoInfo($source);
            $sourceHeight = $info ? $info['height'] : 1080;
            if ($info && $info['height'] > 0) {
                Database::query('UPDATE videos SET source_height = ? WHERE id = ? AND source_height IS NULL', [$info['height'], $videoId]);
            }
        } else {
            $dbHeight = (int)(Database::fetchOne('SELECT source_height FROM videos WHERE id = ?', [$videoId])['source_height'] ?? 0);
            if ($dbHeight > 0) $sourceHeight = $dbHeight;
        }

        $formats = self::getFormats();
        $generateTimeline = Setting::get('transcode_timeline', '0') === '1';

        $existing = Database::fetchAll('SELECT task FROM transcode_jobs WHERE video_id = ?', [$videoId]);
        $existingTasks = array_column($existing, 'task');

        $hasWatermark = Setting::get('watermark_enabled', '0') === '1' && Setting::get('watermark_image', '') !== '';
        $hasTrim = (int)Setting::get('transcode_trim_start', '0') > 0;
        $needsReencode = $hasWatermark || $hasTrim;

         
        $jobBase = ['video_id' => $videoId, 'video_folder' => $folder, 'storage_id' => $storageId];

        $queued = 0;

         
         
         
         
         
        if ($needsReencode && $keepSource && !in_array('source', $existingTasks, true)) {
            Database::insert('transcode_jobs', array_merge($jobBase, ['task' => 'source']));
            $queued++;
        }

        foreach ($formats as $task => $fmt) {
            if ($onlyTasks !== null && !in_array($task, $onlyTasks, true)) continue;
            if (in_array($task, $existingTasks, true)) continue;
            if ($fmt['h'] >= $sourceHeight) continue;
            if (file_exists($dir . '/' . $fmt['file']) && !$needsReencode) continue;

            Database::insert('transcode_jobs', array_merge($jobBase, ['task' => $task]));
            $queued++;
        }

        if ($generateTimeline && !in_array('timeline', $existingTasks, true) && !file_exists($dir . '/timeline.vtt')) {
            if ($onlyTasks === null || in_array('timeline', $onlyTasks, true)) {
                Database::insert('transcode_jobs', array_merge($jobBase, ['task' => 'timeline']));
                $queued++;
            }
        }

        if ($queued > 0) {
            self::dispatchPending();
            self::asyncPing();
        }

        return $queued;
    }

    public static function dispatchPending(): void
    {
        $jobs = Database::fetchAll(
            "SELECT j.* FROM transcode_jobs j
             JOIN videos v ON v.id = j.video_id
             WHERE j.status = 'pending' AND (j.started_at IS NULL OR j.started_at <= NOW())
               AND (j.task = 'source' OR NOT EXISTS (
                   SELECT 1 FROM transcode_jobs js WHERE js.video_id = j.video_id AND js.task = 'source' AND js.status != 'completed' AND NOT (js.status = 'failed' AND js.retry_count >= 3)
               ))
             ORDER BY (j.task = 'source') DESC, CAST(REGEXP_REPLACE(j.task, '[^0-9]', '') AS UNSIGNED) DESC
             LIMIT 20"
        );
        foreach ($jobs as $job) {
            $avoid = $job['last_worker'] ?? null;
            $server = self::pickServer($avoid);
            if ($server) {
                self::dispatchToServer($server, $job);
            } else {
                break;
            }
        }
    }

    private static function asyncPing(): void
    {
        $url = rtrim(Setting::get('site_url', ''), '/') . '/';
        if ($url === '/') return;
        $ch = curl_init($url);
        curl_setopt_array($ch, [
            CURLOPT_RETURNTRANSFER => true,
            CURLOPT_TIMEOUT_MS => 500,
            CURLOPT_CONNECTTIMEOUT_MS => 500,
            CURLOPT_SSL_VERIFYPEER => true,
            CURLOPT_USERAGENT => 'TubePress-Cron/1.0',
        ]);
        @curl_exec($ch);
        curl_close($ch);
    }

    public static function processNext(): bool
    {
        if (!self::isAvailable()) return false;

         
        $job = Database::fetchOne(
            "SELECT j.*, v.video_folder, v.storage_id
             FROM transcode_jobs j
             JOIN videos v ON v.id = j.video_id
             WHERE j.status = 'pending' AND (j.started_at IS NULL OR j.started_at <= NOW())
               AND (j.task = 'source' OR NOT EXISTS (
                   SELECT 1 FROM transcode_jobs js
                   WHERE js.video_id = j.video_id AND js.task = 'source' AND js.status != 'completed' AND NOT (js.status = 'failed' AND js.retry_count >= 3)
               ))
             ORDER BY (j.task = 'source') DESC, (j.task = 'timeline') ASC, CAST(REGEXP_REPLACE(j.task, '[^0-9]', '') AS UNSIGNED) DESC, j.created_at ASC
             LIMIT 1"
        );
        if (!$job) return false;

        $jobId = (int)$job['id'];
        $affected = Database::query(
            "UPDATE transcode_jobs SET status = 'processing', started_at = NOW(), worker = ? WHERE id = ? AND status = 'pending'",
            [gethostname(), $jobId]
        )->rowCount();
        if ($affected === 0) return false;

        try {
            $dir = VideoAsset::dir($job['video_folder']);
            $source = self::findSource($dir);

            if (!$source) {
                self::failJob($jobId, 'Source video file not found.');
                return true;
            }

            $task = $job['task'];
            $storageId = $job['storage_id'] ? (int)$job['storage_id'] : null;
            $formats = self::getFormats();

            if ($task === 'source') {
                self::reencodeSource($jobId, $source, $dir, $job['video_folder'], $storageId);
            } elseif (isset($formats[$task])) {
                self::transcodeResolution($jobId, $source, $dir, $job['video_folder'], $task, $storageId, $formats[$task]);
            } elseif ($task === 'timeline') {
                self::generateTimeline($jobId, $source, $dir, $job['video_folder'], $storageId);
            }

            return true;
        } catch (\Throwable $e) {
            self::failJob($jobId, $e->getMessage());
            return true;
        }
    }

    

    private static function absoluteSourceUrl(string $folder, ?int $storageId, string $siteUrl): string
    {
        $url = self::directVideoUrl($folder, $storageId);
        if ($url !== '' && !str_starts_with($url, 'http')) {
            $url = $siteUrl . '/' . ltrim($url, '/');
        }
        return $url;
    }

    

    private static function directVideoUrl(string $folder, ?int $storageId): string
    {
        $dir = VideoAsset::dir($folder);
        $base = MediaStorage::mediaUrl($storageId);
        $pathPrefix = 'uploads/videos/' . $folder . '/';
        $isRemote = $storageId && MediaStorage::isRemote($storageId);

         
        $wkSuffix = '';
        if ($isRemote) {
            $storage = MediaStorage::get($storageId);
            $cfg = $storage ? (is_array($storage['config']) ? $storage['config'] : json_decode($storage['config'] ?? '{}', true)) : [];
            $wk = $cfg['monitor_key'] ?? '';
            if ($wk !== '') $wkSuffix = '?wk=' . $wk;
        }

         
        $localFiles = glob($dir . '/video_*p.mp4') ?: [];
        usort($localFiles, fn($a, $b) => (int)filter_var(basename($b), FILTER_SANITIZE_NUMBER_INT) - (int)filter_var(basename($a), FILTER_SANITIZE_NUMBER_INT));
        foreach ($localFiles as $file) {
            if (filesize($file) > 1000) {
                return $base . '/' . $pathPrefix . basename($file) . $wkSuffix;
            }
        }

         
        foreach (['video.mp4', 'video.webm', 'video.ogg'] as $srcFile) {
            if (file_exists($dir . '/' . $srcFile) && filesize($dir . '/' . $srcFile) > 1000) {
                return $base . '/' . $pathPrefix . $srcFile . $wkSuffix;
            }
        }

         
        if ($isRemote) {
            return $base . '/' . $pathPrefix . 'video.mp4' . $wkSuffix;
        }

        return '';
    }

    

    private static function findTranscodedSourceUrl(string $folder, ?int $storageId, string $siteUrl): string
    {
        $dir = VideoAsset::dir($folder);
        $base = MediaStorage::mediaUrl($storageId);
        $pathPrefix = 'uploads/videos/' . $folder . '/';
        $isRemote = $storageId && MediaStorage::isRemote($storageId);

         
        $wkSuffix = '';
        if ($isRemote) {
            $storage = MediaStorage::get($storageId);
            $cfg = $storage ? (is_array($storage['config']) ? $storage['config'] : json_decode($storage['config'] ?? '{}', true)) : [];
            $wk = $cfg['monitor_key'] ?? '';
            if ($wk !== '') $wkSuffix = '?wk=' . $wk;
        }

         
        if ($isRemote) {
            $video = Database::fetchOne("SELECT source_height FROM videos WHERE video_folder = ? LIMIT 1", [$folder]);
            $srcH = (int)($video['source_height'] ?? 0);
            if ($srcH > 0) {
                $url = $base . '/' . $pathPrefix . 'video_' . $srcH . 'p.mp4' . $wkSuffix;
                if (!str_starts_with($url, 'http')) {
                    $url = $siteUrl . '/' . ltrim($url, '/');
                }
                return $url;
            }
            return '';
        }

         
        foreach (['2160', '1440', '1080', '720', '480', '360'] as $h) {
            $file = "video_{$h}p.mp4";
            if (file_exists($dir . '/' . $file) && filesize($dir . '/' . $file) > 1000) {
                $url = $base . '/' . $pathPrefix . $file;
                if (!str_starts_with($url, 'http')) {
                    $url = $siteUrl . '/' . ltrim($url, '/');
                }
                return $url;
            }
        }

        return '';
    }

    private static function runFfmpegWithProgress(string $cmd, int $jobId, int $totalDuration): string
    {
        $cmd .= ' 2>&1';
        if ($totalDuration <= 0) return shell_exec($cmd) ?? '';

        $proc = @proc_open($cmd, [1 => ['pipe', 'w'], 2 => ['pipe', 'w']], $pipes);
        if (!is_resource($proc)) return shell_exec($cmd) ?? '';

        stream_set_blocking($pipes[1], false);
        stream_set_blocking($pipes[2], false);

        $output = '';
        $lastUpdate = 0;

        while (true) {
            $read = [$pipes[1], $pipes[2]];
            $w = null; $e = null;
            @stream_select($read, $w, $e, 2);
            foreach ($read as $pipe) {
                $chunk = fread($pipe, 8192);
                if ($chunk !== false && $chunk !== '') $output .= $chunk;
            }
            if (preg_match_all('/time=(\d+):(\d+):(\d+\.\d+)/', $output, $m, PREG_SET_ORDER)) {
                $last = end($m);
                $sec = (int)$last[1] * 3600 + (int)$last[2] * 60 + (float)$last[3];
                $pct = min(99, max(0, (int)round($sec / $totalDuration * 100)));
                if ((time() - $lastUpdate) >= 5) {
                    Database::query('UPDATE transcode_jobs SET progress = ? WHERE id = ?', [$pct, $jobId]);
                    $lastUpdate = time();
                }
            }
            $status = proc_get_status($proc);
            if (!$status['running']) break;
        }

        fclose($pipes[1]); fclose($pipes[2]);
        proc_close($proc);
        return $output;
    }

    private static function concatIntros(string $videoFile, int $width, int $height): bool
    {
        $preEnabled = Setting::get('transcode_preroll_enabled', '0') === '1';
        $postEnabled = Setting::get('transcode_postroll_enabled', '0') === '1';
        $preFile = $preEnabled ? Setting::get('transcode_preroll_file', '') : '';
        $postFile = $postEnabled ? Setting::get('transcode_postroll_file', '') : '';

        $prePath = $preFile !== '' ? APP_ROOT . '/public/uploads/intros/' . $preFile : '';
        $postPath = $postFile !== '' ? APP_ROOT . '/public/uploads/intros/' . $postFile : '';

        if (($prePath === '' || !file_exists($prePath)) && ($postPath === '' || !file_exists($postPath))) return true;

        $ffmpeg = self::ffmpegPath();
        $dir = dirname($videoFile);
        $parts = [];

         
        if ($prePath !== '' && file_exists($prePath)) {
            $preTmp = $dir . '/intro_pre_tmp.mp4';
            $cmd = escapeshellcmd($ffmpeg) . ' -i ' . escapeshellarg($prePath)
                . ' -vf ' . escapeshellarg("scale={$width}:{$height}:force_original_aspect_ratio=decrease,pad={$width}:{$height}:(ow-iw)/2:(oh-ih)/2")
                . ' -c:v libx264 -preset fast -crf 23 -c:a aac -b:a 128k -ar 44100 -movflags +faststart -y ' . escapeshellarg($preTmp) . ' 2>&1';
            shell_exec($cmd);
            if (file_exists($preTmp) && filesize($preTmp) > 1000) $parts[] = $preTmp;
        }

        $parts[] = $videoFile;

         
        if ($postPath !== '' && file_exists($postPath)) {
            $postTmp = $dir . '/intro_post_tmp.mp4';
            $cmd = escapeshellcmd($ffmpeg) . ' -i ' . escapeshellarg($postPath)
                . ' -vf ' . escapeshellarg("scale={$width}:{$height}:force_original_aspect_ratio=decrease,pad={$width}:{$height}:(ow-iw)/2:(oh-ih)/2")
                . ' -c:v libx264 -preset fast -crf 23 -c:a aac -b:a 128k -ar 44100 -movflags +faststart -y ' . escapeshellarg($postTmp) . ' 2>&1';
            shell_exec($cmd);
            if (file_exists($postTmp) && filesize($postTmp) > 1000) $parts[] = $postTmp;
        }

        if (count($parts) <= 1) {
            @unlink($dir . '/intro_pre_tmp.mp4');
            @unlink($dir . '/intro_post_tmp.mp4');
            return true;
        }

         
        $listFile = $dir . '/concat_list.txt';
        $listContent = '';
        foreach ($parts as $p) $listContent .= "file " . escapeshellarg($p) . "\n";
        file_put_contents($listFile, $listContent);

        $concatOutput = $dir . '/concat_output.mp4';
        $cmd = escapeshellcmd($ffmpeg) . ' -f concat -safe 0 -i ' . escapeshellarg($listFile)
            . ' -c copy -movflags +faststart -y ' . escapeshellarg($concatOutput) . ' 2>&1';
        shell_exec($cmd);

         
        @unlink($listFile);
        @unlink($dir . '/intro_pre_tmp.mp4');
        @unlink($dir . '/intro_post_tmp.mp4');

        if (file_exists($concatOutput) && filesize($concatOutput) > 1000) {
            @unlink($videoFile);
            rename($concatOutput, $videoFile);
            return true;
        }

        @unlink($concatOutput);
        return false;
    }

    private static function ffmpegEscapePath(string $path): string
    {
        return str_replace(["\\", "'", ":", ";", "[", "]"], ["\\\\", "\\'", "\\:", "\\;", "\\[", "\\]"], $path);
    }

    private static function buildFilterChain(int $width, int $height, bool $keepAspect): string
    {
        $filters = [];

        if ($keepAspect) {
            $filters[] = "scale={$width}:{$height}:force_original_aspect_ratio=decrease,pad={$width}:{$height}:(ow-iw)/2:(oh-ih)/2";
        } else {
            $filters[] = "scale={$width}:{$height}";
        }

        $wmEnabled = Setting::get('watermark_enabled', '0') === '1';
        $wmImage = Setting::get('watermark_image', '');
        if ($wmEnabled && $wmImage !== '') {
            $wmPath = APP_ROOT . '/public/uploads/watermark/' . $wmImage;
            if (file_exists($wmPath)) {
                $pos = Setting::get('watermark_position', 'bottom-right');
                $opacity = max(10, min(100, (int)Setting::get('watermark_opacity', '80'))) / 100;
                $maxW = max(5, min(50, (int)Setting::get('watermark_max_width', '15')));
                $paddingPct = max(0, min(15, (int)Setting::get('watermark_padding', '2')));
                $wmScaleW = (int)round($width * $maxW / 100);
                $padX = (int)round($width * $paddingPct / 100);
                $padY = (int)round($height * $paddingPct / 100);
                $p2x = $padX * 2;
                $p2y = $padY * 2;

                $overlayPos = match ($pos) {
                    'top-left' => "{$padX}:{$padY}",
                    'top-right' => "main_w-overlay_w-{$padX}:{$padY}",
                    'bottom-left' => "{$padX}:main_h-overlay_h-{$padY}",
                    'center' => '(main_w-overlay_w)/2:(main_h-overlay_h)/2',
                    'random' => "{$padX}+mod(n\\,main_w-overlay_w-{$p2x}):{$padY}+mod(n*7\\,main_h-overlay_h-{$p2y})",
                    default => "main_w-overlay_w-{$padX}:main_h-overlay_h-{$padY}",
                };

                $escapedPath = self::ffmpegEscapePath($wmPath);
                $filters = ["[0:v]" . implode(',', $filters) . "[scaled];movie={$escapedPath},scale={$wmScaleW}:-1,format=rgba,colorchannelmixer=aa={$opacity}[wm];[scaled][wm]overlay={$overlayPos}"];
            }
        }

        return implode(',', $filters);
    }

    private static function transcodeResolution(int $jobId, string $source, string $dir, string $folder, string $res, ?int $storageId, array $fmt): void
    {
        $output = $dir . '/' . $fmt['file'];
        $ffmpeg = self::ffmpegPath();

         
        $videoId = (int)Database::fetchOne('SELECT video_id FROM transcode_jobs WHERE id = ?', [$jobId])['video_id'];
        $sourceReencoded = (bool)Database::fetchOne(
            "SELECT 1 FROM transcode_jobs WHERE video_id = ? AND task = 'source' AND status = 'completed'", [$videoId]
        );

        if ($sourceReencoded) {
            $vf = $fmt['keep_aspect']
                ? "scale={$fmt['w']}:{$fmt['h']}:force_original_aspect_ratio=decrease,pad={$fmt['w']}:{$fmt['h']}:(ow-iw)/2:(oh-ih)/2"
                : "scale={$fmt['w']}:{$fmt['h']}";
        } else {
            $vf = self::buildFilterChain($fmt['w'], $fmt['h'], $fmt['keep_aspect']);
        }

        $isComplex = str_contains($vf, '[') && str_contains($vf, ']');
        $trimStart = $sourceReencoded ? 0 : (int)Setting::get('transcode_trim_start', '0');

        $cmd = escapeshellcmd($ffmpeg);
        if ($trimStart > 0) $cmd .= ' -ss ' . $trimStart;
        $cmd .= ' -i ' . escapeshellarg($source)
            . ($isComplex ? ' -filter_complex ' : ' -vf ') . escapeshellarg($vf);

        $ffmpegOpts = trim($fmt['ffmpeg_options']);
        if ($ffmpegOpts !== '') {
            $cmd .= ' ' . $ffmpegOpts;
        } else {
            $cmd .= ' -c:v libx264 -preset medium -crf 23 -c:a aac -b:a 128k -ar 44100 -movflags +faststart';
        }

        $cmd .= ' -y ' . escapeshellarg($output);

        $srcInfo = self::getVideoInfo($source);
        $out = self::runFfmpegWithProgress($cmd, $jobId, $srcInfo ? $srcInfo['duration'] : 0);
        if (!file_exists($output) || filesize($output) < 1000) {
            self::failJob($jobId, 'FFmpeg output missing or too small. ' . substr($out ?? '', -200));
            return;
        }

         
        if (!$sourceReencoded) {
            self::concatIntros($output, $fmt['w'], $fmt['h']);
        }

        if ($storageId && MediaStorage::isRemote($storageId)) {
             
             
            if (!MediaStorage::put($output, 'uploads/videos/' . $folder . '/' . $fmt['file'], $storageId)) {
                self::failJob($jobId, 'Upload of ' . $fmt['file'] . ' to remote storage failed.');
                return;
            }
        }

        if ($trimStart > 0) {
            $newInfo = self::getVideoInfo($output);
            if ($newInfo && $newInfo['duration'] > 0) {
                $videoId = (int)Database::fetchOne('SELECT video_id FROM transcode_jobs WHERE id = ?', [$jobId])['video_id'];
                Database::query('UPDATE videos SET duration = ? WHERE id = ? AND duration > ?', [$newInfo['duration'], $videoId, $newInfo['duration']]);
            }
        }

        self::completeJob($jobId);
    }

    private static function reencodeSource(int $jobId, string $source, string $dir, string $folder, ?int $storageId): void
    {
        $info = self::getVideoInfo($source);
        if (!$info || $info['height'] < 100 || $info['width'] < 100) {
            self::failJob($jobId, 'Cannot read source video info or invalid resolution (' . ($info ? $info['width'] . 'x' . $info['height'] : 'null') . ').');
            return;
        }

        $w = $info['width'];
        $h = $info['height'];
        $ffmpeg = self::ffmpegPath();
        $trimStart = (int)Setting::get('transcode_trim_start', '0');

        $vf = self::buildFilterChain($w, $h, true);
        $isComplex = str_contains($vf, '[') && str_contains($vf, ']');

        $finalName = 'video_' . $h . 'p.mp4';
        $tmpOutput = $dir . '/video_source_tmp.mp4';

        $cmd = escapeshellcmd($ffmpeg);
        if ($trimStart > 0) $cmd .= ' -ss ' . $trimStart;
        $cmd .= ' -i ' . escapeshellarg($source)
            . ($isComplex ? ' -filter_complex ' : ' -vf ') . escapeshellarg($vf)
            . ' -c:v libx264 -profile:v high -level:v 4.0 -preset medium -crf 21 -maxrate 6000k -bufsize 12000k -pix_fmt yuv420p -c:a aac -b:a 192k -ar 44100 -movflags +faststart'
            . ' -y ' . escapeshellarg($tmpOutput);

        $out = self::runFfmpegWithProgress($cmd, $jobId, $info['duration'] ?? 0);
        if (!file_exists($tmpOutput) || filesize($tmpOutput) < 1000) {
            self::failJob($jobId, 'Source re-encode failed. ' . substr($out ?? '', -200));
            @unlink($tmpOutput);
            return;
        }

        if (!rename($tmpOutput, $dir . '/' . $finalName)) {
            self::failJob($jobId, 'Failed to move re-encoded file.');
            @unlink($tmpOutput);
            return;
        }
        self::concatIntros($dir . '/' . $finalName, $w, $h);

        if ($storageId && MediaStorage::isRemote($storageId)) {
            if (!MediaStorage::put($dir . '/' . $finalName, 'uploads/videos/' . $folder . '/' . $finalName, $storageId)) {
                 
                 
                if (basename($source) !== $finalName && is_file($source)) {
                    @unlink($dir . '/' . $finalName);
                }
                self::failJob($jobId, 'Upload of ' . $finalName . ' to remote storage failed.');
                return;
            }
        }

         
        if (basename($source) !== $finalName) {
            @unlink($source);
        }

        $videoId = (int)Database::fetchOne('SELECT video_id FROM transcode_jobs WHERE id = ?', [$jobId])['video_id'];

        $newInfo = self::getVideoInfo($dir . '/' . $finalName);
        $updates = ['source_height' => $h];
        if ($newInfo && $trimStart > 0 && $newInfo['duration'] > 0) {
            $updates['duration'] = $newInfo['duration'];
        }
        Database::update('videos', $updates, 'id = ?', [$videoId]);

        Database::query("UPDATE videos SET status = 'published', published_at = COALESCE(published_at, NOW()) WHERE id = ? AND status = 'processing'", [$videoId]);

        self::completeJob($jobId);
    }

    private static function generateTimeline(int $jobId, string $source, string $dir, string $folder, ?int $storageId): void
    {
        $info = self::getVideoInfo($source);
        $duration = $info ? $info['duration'] : 0;
        if ($duration < 5) {
            self::failJob($jobId, 'Video too short for timeline.');
            return;
        }

        $videoId = (int)Database::fetchOne('SELECT video_id FROM transcode_jobs WHERE id = ?', [$jobId])['video_id'];
        $sourceReencoded = (bool)Database::fetchOne(
            "SELECT 1 FROM transcode_jobs WHERE video_id = ? AND task = 'source' AND status = 'completed'", [$videoId]
        );
        $trimStart = $sourceReencoded ? 0 : (int)Setting::get('transcode_trim_start', '0');
        $timelineStart = max(0, (int)Setting::get('transcode_timeline_start', '0'));

         
        $totalSeek = $trimStart + $timelineStart;
         
        if ($totalSeek >= $duration - 5) {
            $totalSeek = $trimStart;  
        }

        $effectiveDuration = max(1, $duration - $totalSeek);
        $count = max(5, min(60, (int)Setting::get('transcode_timeline_count', '10')));
        $interval = $effectiveDuration / $count;
        $thumbW = 160;
        $thumbH = 90;
        $cols = 10;
        $rows = (int)ceil($count / $cols);

        $ffmpeg = self::ffmpegPath();
        $spritePath = $dir . '/timeline.jpg';
        $vttPath = $dir . '/timeline.vtt';

        $fps = 1 / max(0.5, $interval);
        $ssArg = $totalSeek > 0 ? ' -ss ' . $totalSeek : '';
        $cmd = escapeshellcmd($ffmpeg) . $ssArg
            . ' -i ' . escapeshellarg($source)
            . ' -vf ' . escapeshellarg("fps={$fps},scale={$thumbW}:{$thumbH}:force_original_aspect_ratio=decrease,pad={$thumbW}:{$thumbH}:(ow-iw)/2:(oh-ih)/2,tile={$cols}x{$rows}")
            . ' -frames:v 1 -q:v 5'
            . ' -y ' . escapeshellarg($spritePath)
            . ' 2>&1';

        shell_exec($cmd);
        if (!file_exists($spritePath) || filesize($spritePath) < 500) {
            self::failJob($jobId, 'Timeline sprite generation failed.');
            return;
        }

        $vtt = "WEBVTT\n\n";
        for ($i = 0; $i < $count; $i++) {
            $start = $i * $interval;
            $end = min(($i + 1) * $interval, $effectiveDuration);
            $x = ($i % $cols) * $thumbW;
            $y = (int)floor($i / $cols) * $thumbH;
            $vtt .= self::formatVttTime($start) . ' --> ' . self::formatVttTime($end) . "\n";
            $vtt .= "timeline.jpg#xywh={$x},{$y},{$thumbW},{$thumbH}\n\n";
        }
        file_put_contents($vttPath, $vtt);

        if ($storageId && MediaStorage::isRemote($storageId)) {
            if (!MediaStorage::put($spritePath, 'uploads/videos/' . $folder . '/timeline.jpg', $storageId)
                || !MediaStorage::put($vttPath, 'uploads/videos/' . $folder . '/timeline.vtt', $storageId)) {
                self::failJob($jobId, 'Upload of timeline to remote storage failed.');
                return;
            }
        }

        self::completeJob($jobId);
    }

    private static function formatVttTime(float $seconds): string
    {
        $h = (int)floor($seconds / 3600);
        $m = (int)floor(($seconds % 3600) / 60);
        $s = (int)floor($seconds % 60);
        $ms = (int)round(($seconds - floor($seconds)) * 1000);
        return sprintf('%02d:%02d:%02d.%03d', $h, $m, $s, $ms);
    }

    private static function completeJob(int $jobId): void
    {
        $job = Database::fetchOne("SELECT video_id, worker, task FROM transcode_jobs WHERE id = ?", [$jobId]);
        $worker = $job['worker'] ?? null;
        Database::query("UPDATE transcode_jobs SET status = 'completed', progress = 100, completed_at = NOW() WHERE id = ?", [$jobId]);
        if ($worker) {
            Database::query("UPDATE transcode_servers SET active_jobs = IF(active_jobs > 0, active_jobs - 1, 0) WHERE name = ?", [$worker]);
        }
         
         
        if ($job && ($job['task'] ?? '') === 'timeline') {
            Database::query("UPDATE videos SET timeline_ready = 1 WHERE id = ?", [(int)$job['video_id']]);
        }
        if ($job) self::checkAutoPublish((int)$job['video_id']);
    }

    

    public static function autoRetryStuck(): void
    {
        try {
             
             
            $reset = Database::query(
                "UPDATE transcode_jobs
                 SET status = 'pending', retry_count = 0, worker = NULL, started_at = NULL,
                     completed_at = NULL, progress = 0,
                     error_message = CONCAT('Auto-retry: ', LEFT(COALESCE(error_message, ''), 200))
                 WHERE status = 'failed'
                   AND (error_message IS NULL OR error_message NOT LIKE 'Skipped — source task%')
                   AND completed_at IS NOT NULL
                   AND completed_at < (NOW() - INTERVAL 60 MINUTE)"
            )->rowCount();

            if ($reset > 0) {
                 
                Database::query(
                    "UPDATE transcode_jobs SET status = 'pending', retry_count = 0, worker = NULL,
                            started_at = NULL, completed_at = NULL, progress = 0, error_message = NULL
                     WHERE status = 'failed' AND error_message LIKE 'Skipped — source task%'
                       AND video_id IN (SELECT vid FROM (SELECT DISTINCT video_id vid FROM transcode_jobs WHERE status = 'pending') t)"
                );
                 
                Database::query(
                    "UPDATE videos SET status = 'processing'
                     WHERE status = 'error'
                       AND id IN (SELECT DISTINCT video_id FROM transcode_jobs WHERE status = 'pending')"
                );
            }
        } catch (\Throwable $e) {
            error_log('TranscodeManager::autoRetryStuck: ' . $e->getMessage());
        }
    }

    private static function checkAutoPublish(int $videoId): void
    {
        $video = Database::fetchOne("SELECT status, video_folder, storage_id, cleanup_source, source_height FROM videos WHERE id = ?", [$videoId]);
        if (!$video || !in_array($video['status'], ['processing', 'error'], true)) return;

         
        $sourceJob = Database::fetchOne(
            "SELECT status, retry_count FROM transcode_jobs WHERE video_id = ? AND task = 'source' ORDER BY id LIMIT 1",
            [$videoId]
        );

        if ($sourceJob) {
            if ($sourceJob['status'] === 'completed') {
                 
                Database::query(
                    "UPDATE videos SET status = 'published', published_at = COALESCE(published_at, NOW()) WHERE id = ? AND status IN ('processing','error')",
                    [$videoId]
                );
            } elseif ($sourceJob['status'] === 'failed' && (int)$sourceJob['retry_count'] >= 3) {
                 
                Database::query("UPDATE videos SET status = 'error' WHERE id = ? AND status = 'processing'", [$videoId]);
            }
             
        } else {
            $totalJobs = (int)(Database::fetchOne(
                "SELECT COUNT(*) as c FROM transcode_jobs WHERE video_id = ?", [$videoId]
            )['c'] ?? 0);
            if ($totalJobs === 0) return;  

            $incomplete = (int)(Database::fetchOne(
                "SELECT COUNT(*) as c FROM transcode_jobs WHERE video_id = ? AND status NOT IN ('completed','failed','skipped')",
                [$videoId]
            )['c'] ?? 0);

            if ($incomplete === 0) {
                $anyCompleted = (bool)Database::fetchOne(
                    "SELECT 1 FROM transcode_jobs WHERE video_id = ? AND status = 'completed'",
                    [$videoId]
                );
                if ($anyCompleted) {
                    Database::query(
                        "UPDATE videos SET status = 'published', published_at = COALESCE(published_at, NOW()) WHERE id = ? AND status IN ('processing','error')",
                        [$videoId]
                    );
                } else {
                    Database::query("UPDATE videos SET status = 'error' WHERE id = ? AND status = 'processing'", [$videoId]);
                }
            }
        }

         
        if (!empty($video['cleanup_source']) && !empty($video['video_folder'])) {
            $stillRunning = (int)(Database::fetchOne(
                "SELECT COUNT(*) as c FROM transcode_jobs WHERE video_id = ? AND status NOT IN ('completed','failed','skipped')",
                [$videoId]
            )['c'] ?? 0);

            if ($stillRunning === 0) {
                 
                $anyOk = (bool)Database::fetchOne(
                    "SELECT 1 FROM transcode_jobs WHERE video_id = ? AND status = 'completed' LIMIT 1",
                    [$videoId]
                );
                if ($anyOk) {
                    self::cleanupSourceFiles($videoId, $video);
                } else {
                     
                    Database::query("UPDATE videos SET cleanup_source = 0 WHERE id = ?", [$videoId]);
                }
            }
        }

         
         
         
         
         
        try {
            $fin = Database::fetchOne("SELECT status, video_folder, storage_id FROM videos WHERE id = ?", [$videoId]);
            if ($fin && $fin['status'] === 'published' && !empty($fin['video_folder'])) {
                $finSid = $fin['storage_id'] ? (int) $fin['storage_id'] : null;
                if ($finSid && MediaStorage::isRemote($finSid)) {
                    $finPending = (int) (Database::fetchOne(
                        "SELECT COUNT(*) as c FROM transcode_jobs WHERE video_id = ? AND status NOT IN ('completed','failed','skipped')",
                        [$videoId]
                    )['c'] ?? 0);
                    if ($finPending === 0) {
                        VideoAsset::purgeLocalFolder($fin['video_folder']);
                    }
                }
            }
        } catch (\Throwable $e) {}
    }

    

    private static function cleanupSourceFiles(int $videoId, array $video): void
    {
        $folder = $video['video_folder'];
        $dir = VideoAsset::dir($folder);
        $storageId = $video['storage_id'] ? (int) $video['storage_id'] : null;
        $isRemote = $storageId && MediaStorage::isRemote($storageId);

         
        $completedTasks = Database::fetchAll(
            "SELECT task FROM transcode_jobs WHERE video_id = ? AND status = 'completed'",
            [$videoId]
        );
        $completedTaskNames = array_column($completedTasks, 'task');

         
        $keepFiles = [];
        $formats = self::getFormats();
        foreach ($formats as $task => $fmt) {
            if (in_array($task, $completedTaskNames, true)) {
                $keepFiles[] = $fmt['file'];  
            }
        }

         
         
         
        $sourceHeight = (int) ($video['source_height'] ?? 0);
        $sourceReencodeFile = $sourceHeight > 0 ? "video_{$sourceHeight}p.mp4" : null;

         
        $keptHeights = [];
        foreach ($formats as $task => $fmt) {
            if (in_array($task, $completedTaskNames, true)) {
                $keptHeights[] = $fmt['h'];
            }
        }

         
        $hasTranscodedFile = false;
        foreach ($keepFiles as $f) {
            if (file_exists($dir . '/' . $f) && filesize($dir . '/' . $f) > 1000) {
                $hasTranscodedFile = true;
                break;
            }
        }
         
        if (!$hasTranscodedFile && $isRemote && !empty($completedTaskNames)) {
            $hasTranscodedFile = true;
        }

        if (!$hasTranscodedFile) {
            return;  
        }

         
        foreach (['video.mp4', 'video.webm', 'video.ogg'] as $srcFile) {
            $path = $dir . '/' . $srcFile;
            if (file_exists($path)) {
                @unlink($path);
            }
            if ($isRemote) {
                try { MediaStorage::deleteFile('uploads/videos/' . $folder . '/' . $srcFile, $storageId); } catch (\Throwable $e) {}
            }
        }

         
        if ($sourceReencodeFile && !in_array($sourceHeight, $keptHeights, true)) {
            $path = $dir . '/' . $sourceReencodeFile;
            if (file_exists($path)) {
                @unlink($path);
            }
            if ($isRemote) {
                try { MediaStorage::deleteFile('uploads/videos/' . $folder . '/' . $sourceReencodeFile, $storageId); } catch (\Throwable $e) {}
            }
        }

         
        Database::query("UPDATE videos SET cleanup_source = 0 WHERE id = ?", [$videoId]);
    }

    

    public static function gcLocalMedia(int $maxFolders = 200): int
    {
        $base = APP_ROOT . '/public/uploads/videos';
        if (!is_dir($base)) return 0;

        $now = time();
        $reaped = 0;
        foreach (glob($base . '/*', GLOB_ONLYDIR) ?: [] as $dir) {
            if ($reaped >= $maxFolders) break;
            $folder = basename($dir);
            if (!preg_match('/^[a-f0-9]+$/i', $folder)) continue;
            if ($now - (int) @filemtime($dir) < 7200) continue;  

            $video = Database::fetchOne("SELECT id, status, storage_id FROM videos WHERE video_folder = ? LIMIT 1", [$folder]);
            if (!$video) {
                 
                if ($now - (int) @filemtime($dir) >= 86400) { VideoAsset::purgeLocalFolder($folder); $reaped++; }
                continue;
            }
            $sid = $video['storage_id'] ? (int) $video['storage_id'] : null;
            if (!$sid || !MediaStorage::isRemote($sid)) continue;           
            if (($video['status'] ?? '') !== 'published') continue;         
            $pending = (int) (Database::fetchOne(
                "SELECT COUNT(*) as c FROM transcode_jobs WHERE video_id = ? AND status NOT IN ('completed','failed','skipped')",
                [(int) $video['id']]
            )['c'] ?? 0);
            if ($pending > 0) continue;                                     
             
             
            if (!self::localFolderIsOnRemote($folder, $sid)) continue;

            VideoAsset::purgeLocalFolder($folder);
            $reaped++;
        }
        return $reaped;
    }

    

    private static function localFolderIsOnRemote(string $folder, int $storageId): bool
    {
        $dir = VideoAsset::dir($folder);
        foreach (glob($dir . '/*') ?: [] as $f) {
            if (!is_file($f)) continue;
            $name = basename($f);
            if (!preg_match('/\.(mp4|webm|ogg|jpg|jpeg|vtt)$/i', $name)) continue;
            if (!MediaStorage::exists('uploads/videos/' . $folder . '/' . $name, $storageId)) {
                return false;
            }
        }
        return true;
    }

    private static function failJob(int $jobId, string $error): void
    {
        $job = Database::fetchOne("SELECT retry_count, worker, task, video_id FROM transcode_jobs WHERE id = ?", [$jobId]);
        if (!$job) return;
        $retries = (int)$job['retry_count'];
        $worker = $job['worker'] ?? null;
        $maxRetries = 8;

        if ($retries < $maxRetries) {
             
             
             
            $delays = [30, 120, 300, 900, 1800, 3600, 7200, 14400];
            $delay = $delays[$retries] ?? 14400;
            $retryAfter = date('Y-m-d H:i:s', time() + $delay);
            Database::query(
                "UPDATE transcode_jobs SET status = 'pending', retry_count = retry_count + 1, last_worker = worker, worker = NULL, started_at = ?, error_message = ? WHERE id = ?",
                [$retryAfter, 'Retry ' . ($retries + 1) . '/' . $maxRetries . ' (in ' . $delay . 's): ' . mb_substr($error, 0, 4000), $jobId]
            );
        } else {
            Database::query(
                "UPDATE transcode_jobs SET status = 'failed', error_message = ?, completed_at = NOW() WHERE id = ?",
                [mb_substr($error, 0, 4000), $jobId]
            );

             
             
            if ($job['task'] === 'source') {
                $videoId = (int)$job['video_id'];
                Database::query(
                    "UPDATE transcode_jobs SET status = 'failed', error_message = 'Skipped — source task failed after 3 retries. Fix the source issue and retry.', completed_at = NOW()
                     WHERE video_id = ? AND task != 'source' AND status IN ('pending', 'processing')",
                    [$videoId]
                );
            }
        }

        if ($worker) {
            Database::query("UPDATE transcode_servers SET active_jobs = IF(active_jobs > 0, active_jobs - 1, 0) WHERE name = ?", [$worker]);
        }

         
        self::checkAutoPublish((int)$job['video_id']);
    }

    public static function getStats(): array
    {
        $rows = Database::fetchAll("SELECT status, COUNT(DISTINCT video_id) as c FROM transcode_jobs GROUP BY status");
        $byStatus = [];
        foreach ($rows as $r) $byStatus[$r['status']] = (int)$r['c'];

        $hasActive = ($byStatus['pending'] ?? 0) + ($byStatus['processing'] ?? 0) + ($byStatus['failed'] ?? 0);
        $totalVideos = (int)Database::fetchOne("SELECT COUNT(DISTINCT video_id) as c FROM transcode_jobs")['c'];

        return [
            'pending' => $byStatus['pending'] ?? 0,
            'processing' => $byStatus['processing'] ?? 0,
            'completed' => max(0, $totalVideos - $hasActive),
            'failed' => $byStatus['failed'] ?? 0,
        ];
    }

    public static function getVideoQueue(string $filter = '', int $page = 1, int $perPage = 50): array
    {
        $offset = ($page - 1) * $perPage;

         
        if ($filter === '' || $filter === 'pending' || $filter === 'processing' || $filter === 'failed') {
            $statusFilter = $filter !== '' ? "WHERE status = ?" : '';
            $params = $filter !== '' ? [$filter] : [];

            $total = (int)Database::fetchOne("SELECT COUNT(DISTINCT video_id) as c FROM transcode_jobs {$statusFilter}", $params)['c'];
            $videoIds = array_column(
                Database::fetchAll("SELECT DISTINCT video_id FROM transcode_jobs {$statusFilter} ORDER BY video_id DESC LIMIT {$perPage} OFFSET {$offset}", $params),
                'video_id'
            );
        } else {
             
            $total = (int)Database::fetchOne(
                "SELECT COUNT(*) as c FROM (SELECT video_id FROM transcode_jobs GROUP BY video_id HAVING MAX(status IN ('pending','processing','failed')) = 0) sub"
            )['c'];
            $videoIds = array_column(
                Database::fetchAll(
                    "SELECT video_id FROM transcode_jobs GROUP BY video_id HAVING MAX(status IN ('pending','processing','failed')) = 0 ORDER BY video_id DESC LIMIT {$perPage} OFFSET {$offset}"
                ),
                'video_id'
            );
        }

        if (empty($videoIds)) {
            return ['videos' => [], 'total' => $total, 'page' => $page, 'per_page' => $perPage, 'pages' => (int)ceil($total / $perPage)];
        }

         
        $ph = implode(',', array_fill(0, count($videoIds), '?'));
        $jobs = Database::fetchAll(
            "SELECT j.id, j.video_id, j.task, j.status, j.progress, j.stage, j.worker, j.error_message, j.retry_count, j.started_at, j.completed_at, v.title as video_title, v.source_height
             FROM transcode_jobs j
             LEFT JOIN videos v ON v.id = j.video_id
             WHERE j.video_id IN ({$ph})
             ORDER BY j.video_id DESC, j.id ASC",
            $videoIds
        );

         
        $grouped = [];
        foreach ($jobs as $j) {
            $vid = (int)$j['video_id'];
            if (!isset($grouped[$vid])) {
                $grouped[$vid] = [
                    'video_id' => $vid,
                    'video_title' => $j['video_title'],
                    'source_height' => (int)($j['source_height'] ?? 0),
                    'tasks' => [],
                    'cnt_pending' => 0,
                    'cnt_processing' => 0,
                    'cnt_completed' => 0,
                    'cnt_failed' => 0,
                    'cnt_total' => 0,
                ];
            }
            $grouped[$vid]['tasks'][] = [
                'id' => (int)$j['id'],
                'task' => $j['task'],
                'status' => $j['status'],
                'progress' => (int)$j['progress'],
                'stage' => $j['stage'] ?? null,
                'worker' => $j['worker'] ?? null,
                'error_message' => $j['error_message'] ?? null,
                'retry_count' => (int)($j['retry_count'] ?? 0),
                'started_at' => $j['started_at'] ?? null,
                'completed_at' => $j['completed_at'] ?? null,
            ];
            $grouped[$vid]['cnt_' . $j['status']]++;
            $grouped[$vid]['cnt_total']++;
        }

         
         
        $rank = static function (string $task): int {
            if ($task === 'source') return 100000;
            if ($task === 'timeline') return -1;
            return preg_match('/(\d+)/', $task, $m) ? (int) $m[1] : 0;
        };
        foreach ($grouped as &$g) {
            usort($g['tasks'], static fn($a, $b) => $rank($b['task']) <=> $rank($a['task']));
        }
        unset($g);

        return ['videos' => array_values($grouped), 'total' => $total, 'page' => $page, 'per_page' => $perPage, 'pages' => (int)ceil($total / $perPage)];
    }

    public static function retryFailed(): int
    {
         
        $count = Database::query(
            "UPDATE transcode_jobs SET status = 'pending', error_message = NULL, started_at = NULL, completed_at = NULL, retry_count = 0, progress = 0
             WHERE status = 'failed' AND (error_message IS NULL OR error_message NOT LIKE 'Skipped — source task%')"
        )->rowCount();

         
        if ($count > 0) {
            Database::query(
                "UPDATE transcode_jobs SET status = 'pending', error_message = NULL, started_at = NULL, completed_at = NULL, retry_count = 0, progress = 0
                 WHERE status = 'failed' AND error_message LIKE 'Skipped — source task%'
                 AND video_id IN (SELECT DISTINCT video_id FROM (SELECT video_id FROM transcode_jobs WHERE task = 'source' AND status = 'pending') src)"
            );

             
            Database::query(
                "UPDATE videos SET status = 'processing' WHERE status = 'error' AND id IN (SELECT DISTINCT video_id FROM transcode_jobs WHERE status = 'pending')"
            );

        }

        return $count;
    }

    public static function getServers(): array
    {
        return Database::fetchAll("SELECT * FROM transcode_servers WHERE is_active = 1 ORDER BY active_jobs ASC, id ASC");
    }

    private static function healthCheckServers(): void
    {
        $servers = Database::fetchAll("SELECT * FROM transcode_servers WHERE is_active = 1");
        foreach ($servers as $s) {
            $result = self::checkServer($s);
            $newStatus = $result['ok'] ? 'online' : 'error';
            $lastError = $result['ok'] ? null : ($result['error'] ?? 'Unreachable');
            Database::query(
                "UPDATE transcode_servers SET status = ?, last_error = ?, last_check = NOW() WHERE id = ?",
                [$newStatus, $lastError, (int)$s['id']]
            );
        }
    }

    public static function pickServer(?string $avoidWorker = null): ?array
    {
        $servers = self::getServers();
        $fallback = null;
        foreach ($servers as $s) {
            if ($s['status'] === 'error') continue;
            if ((int)$s['active_jobs'] >= (int)$s['max_jobs']) continue;
            if ($avoidWorker !== null && $s['name'] === $avoidWorker) {
                $fallback = $fallback ?? $s;
                continue;
            }
            return $s;
        }
        return $fallback;
    }

    public static function checkServer(array $server): array
    {
        $url = rtrim($server['url'], '/') . '/health';
        $ch = curl_init($url);
        curl_setopt_array($ch, [
            CURLOPT_RETURNTRANSFER => true,
            CURLOPT_TIMEOUT => 5,
            CURLOPT_CONNECTTIMEOUT => 3,
            CURLOPT_HTTPHEADER => ['X-API-Key: ' . $server['api_key']],
        ]);
        $body = curl_exec($ch);
        $code = (int)curl_getinfo($ch, CURLINFO_HTTP_CODE);
        $err = curl_error($ch);
        curl_close($ch);

        if ($code === 0) return ['ok' => false, 'error' => 'Cannot connect: ' . ($err ?: 'timeout')];
        if ($code === 401) return ['ok' => false, 'error' => 'Authentication failed (wrong API key)'];
        if ($code !== 200) return ['ok' => false, 'error' => "HTTP {$code}"];

        $data = json_decode($body, true);
        if (!$data || !isset($data['ffmpeg'])) return ['ok' => false, 'error' => 'Invalid response (FFmpeg not detected on server)'];

        return ['ok' => true, 'ffmpeg' => $data['ffmpeg'] ?? '', 'encoder' => $data['encoder'] ?? 'unknown', 'cpu' => $data['cpu'] ?? -1, 'gpu' => $data['gpu'] ?? -1, 'load' => $data['load'] ?? ''];
    }

    public static function dispatchToServer(array $server, array $job): bool
    {
         
        $folder = $job['video_folder'] ?? null;
        $storageId = $job['storage_id'] ? (int)$job['storage_id'] : null;
        if (!$folder) {
             
            $video = Database::fetchOne('SELECT video_folder, storage_id FROM videos WHERE id = ?', [(int)$job['video_id']]);
            if (!$video || !$video['video_folder']) {
                 
                 
                self::failJob((int)$job['id'], 'Video introuvable ou sans video_folder - source impossible a localiser.');
                return false;
            }
            $folder = $video['video_folder'];
            $storageId = $video['storage_id'] ? (int)$video['storage_id'] : null;
        }
        $storage = $storageId ? MediaStorage::get($storageId) : null;
        $task = $job['task'];
        $formats = self::getFormats();
        $fmt = $formats[$task] ?? null;
        $siteUrl = rtrim(Setting::get('site_url', ''), '/');

         
        $sourceReencoded = $task !== 'source' && (bool)Database::fetchOne(
            "SELECT 1 FROM transcode_jobs WHERE video_id = ? AND task = 'source' AND status = 'completed'", [(int)$job['video_id']]
        );
        $trimStart = $sourceReencoded ? 0 : (int)Setting::get('transcode_trim_start', '0');

        $videoDuration = (int)Database::fetchOne("SELECT duration FROM videos WHERE id = ?", [(int)$job['video_id']])['duration'];

         
         
         
        $sourceUrl = '';
        if ($sourceReencoded) {
             
            $sourceJob = Database::fetchOne(
                "SELECT 1 FROM transcode_jobs WHERE video_id = ? AND task = 'source' AND status = 'completed'",
                [(int)$job['video_id']]
            );
            if (!$sourceJob) {
                 
                return false;
            }
             
            $sourceUrl = self::findTranscodedSourceUrl($folder, $storageId, $siteUrl);
        }
        if ($sourceUrl === '') {
            $sourceUrl = self::absoluteSourceUrl($folder, $storageId, $siteUrl);
        }

        $payload = [
            'job_id' => (int)$job['id'],
            'video_folder' => $folder,
            'task' => $task,
            'callback_url' => $siteUrl . '/api/transcode-callback',
            'upload_url' => $siteUrl . '/api/transcode-upload',
            'api_key' => $server['api_key'],
            'source_url' => $sourceUrl,
            'trim_start' => $trimStart,
            'duration' => $videoDuration,
            'max_jobs' => (int)$server['max_jobs'],
        ];

        if (empty($payload['source_url'])) {
             
             
             
            $fresh = Database::fetchOne('SELECT video_folder, storage_id FROM videos WHERE id = ?', [(int)$job['video_id']]);
            $freshFolder = $fresh['video_folder'] ?? null;
            $freshStorage = ($fresh && $fresh['storage_id']) ? (int)$fresh['storage_id'] : null;
            if ($freshFolder && ($freshFolder !== $folder || $freshStorage !== $storageId)) {
                Database::query("UPDATE transcode_jobs SET video_folder = ?, storage_id = ? WHERE id = ?", [$freshFolder, $freshStorage, (int)$job['id']]);
                $folder = $freshFolder; $storageId = $freshStorage; $payload['video_folder'] = $folder;
                $payload['source_url'] = $sourceReencoded ? self::findTranscodedSourceUrl($folder, $storageId, $siteUrl) : self::absoluteSourceUrl($folder, $storageId, $siteUrl);
                if ($payload['source_url'] === '') $payload['source_url'] = self::absoluteSourceUrl($folder, $storageId, $siteUrl);
            }
             
             
             
            if (empty($payload['source_url'])) {
                self::failJob((int)$job['id'], 'Source introuvable: video_folder=' . $folder . ', storage_id=' . ($storageId ?? 'null') . ' (fichier source absent du stockage).');
                return false;
            }
        }

        if ($task === 'source') {
            $sourceFile = self::findSource(VideoAsset::dir($folder));
            $srcInfo = $sourceFile ? self::getVideoInfo($sourceFile) : null;
            $srcH = $srcInfo ? $srcInfo['height'] : 1080;
            $payload['format'] = [
                'width' => $srcInfo ? $srcInfo['width'] : 1920,
                'height' => $srcH,
                'keep_aspect' => true,
                'file' => 'video_' . $srcH . 'p.mp4',
                'ffmpeg_options' => '-c:v libx264 -profile:v high -level:v 4.0 -preset medium -crf 21 -maxrate 6000k -bufsize 12000k -pix_fmt yuv420p -c:a aac -b:a 192k -ar 44100 -movflags +faststart',
            ];
            $payload['delete_source'] = basename($sourceFile ?: 'video.mp4');
            $payload['is_source_reencode'] = true;
        } elseif ($fmt) {
            $payload['format'] = [
                'width' => $fmt['w'],
                'height' => $fmt['h'],
                'keep_aspect' => $fmt['keep_aspect'],
                'file' => $fmt['file'],
                'ffmpeg_options' => $fmt['ffmpeg_options'],
            ];
        }

        if ($task === 'timeline') {
            $payload['timeline'] = [
                'count' => max(5, min(60, (int)Setting::get('transcode_timeline_count', '10'))),
                'start' => max(0, min(300, (int)Setting::get('transcode_timeline_start', '0'))),
            ];
        }

        $wmEnabled = Setting::get('watermark_enabled', '0') === '1';
        $wmImage = Setting::get('watermark_image', '');
        if ($wmEnabled && $wmImage !== '' && !$sourceReencoded && ($fmt || $task === 'source')) {
            $payload['watermark'] = [
                'image_url' => $siteUrl . '/uploads/watermark/' . $wmImage,
                'position' => Setting::get('watermark_position', 'bottom-right'),
                'opacity' => (int)Setting::get('watermark_opacity', '80'),
                'max_width' => (int)Setting::get('watermark_max_width', '15'),
                'padding' => (int)Setting::get('watermark_padding', '10'),
            ];
        }

         
        $preEnabled = Setting::get('transcode_preroll_enabled', '0') === '1';
        $postEnabled = Setting::get('transcode_postroll_enabled', '0') === '1';
        $shouldConcat = !$sourceReencoded;
        if ($shouldConcat && ($preEnabled || $postEnabled)) {
            $intros = [];
            $preFile = $preEnabled ? Setting::get('transcode_preroll_file', '') : '';
            if ($preFile !== '') $intros['preroll_url'] = $siteUrl . '/uploads/intros/' . $preFile;
            $postFile = $postEnabled ? Setting::get('transcode_postroll_file', '') : '';
            if ($postFile !== '') $intros['postroll_url'] = $siteUrl . '/uploads/intros/' . $postFile;
            if (!empty($intros)) $payload['intros'] = $intros;
        }

         
         
         

         
        $claimed = Database::query(
            "UPDATE transcode_jobs SET status = 'processing', worker = ?, started_at = NOW() WHERE id = ? AND status = 'pending'",
            [$server['name'], (int)$job['id']]
        )->rowCount();
        if ($claimed === 0) {
            return false;  
        }
        Database::query("UPDATE transcode_servers SET active_jobs = active_jobs + 1 WHERE id = ?", [(int)$server['id']]);

        $url = rtrim($server['url'], '/') . '/transcode';
        $ch = curl_init($url);
        curl_setopt_array($ch, [
            CURLOPT_POST => true,
            CURLOPT_POSTFIELDS => json_encode($payload),
            CURLOPT_HTTPHEADER => ['Content-Type: application/json', 'X-API-Key: ' . $server['api_key']],
            CURLOPT_RETURNTRANSFER => true,
            CURLOPT_TIMEOUT => 15,
        ]);
        $body = curl_exec($ch);
        $code = (int)curl_getinfo($ch, CURLINFO_HTTP_CODE);
        curl_close($ch);

        if ($code >= 200 && $code < 300) {
            return true;
        }

         
        Database::query("UPDATE transcode_jobs SET status = 'pending', worker = NULL, started_at = NULL WHERE id = ?", [(int)$job['id']]);
        Database::query("UPDATE transcode_servers SET active_jobs = IF(active_jobs > 0, active_jobs - 1, 0) WHERE id = ?", [(int)$server['id']]);
        return false;
    }

    

    private static function expectedRenditionFile(string $task, int $videoId): ?string
    {
        if ($task === 'timeline') return null;
        if ($task === 'source') {
            $h = (int)(Database::fetchOne('SELECT source_height FROM videos WHERE id = ?', [$videoId])['source_height'] ?? 0);
            return $h > 0 ? "video_{$h}p.mp4" : null;
        }
        try {
            $formats = self::getFormats();
            if (!empty($formats[$task]['file'])) return $formats[$task]['file'];
        } catch (\Throwable) {}
        return null;
    }

     
    private static function renditionPresent(string $folder, string $file, ?int $storageId): bool
    {
        $path = 'uploads/videos/' . $folder . '/' . $file;
        if ($storageId && MediaStorage::isRemote($storageId)) {
            return MediaStorage::exists($path, $storageId);
        }
        return is_file(VideoAsset::dir($folder) . '/' . $file);
    }

    public static function handleCallback(array $input, string $apiKey): array
    {
        $server = Database::fetchOne('SELECT id, name FROM transcode_servers WHERE api_key = ?', [$apiKey]);
        if (!$server) return ['error' => 'Invalid API key', 'code' => 401];

        $jobId = (int)($input['job_id'] ?? 0);
        $status = ($input['status'] ?? '') === 'completed' ? 'completed' : 'failed';
        $error = $input['error'] ?? null;
        $duration = isset($input['duration']) ? (int)$input['duration'] : null;

        $job = Database::fetchOne('SELECT j.*, v.video_folder, v.storage_id FROM transcode_jobs j LEFT JOIN videos v ON v.id = j.video_id WHERE j.id = ?', [$jobId]);
        if (!$job) return ['error' => 'Job not found', 'code' => 404];

         
        if (!$job['video_folder']) {
            Database::query("DELETE FROM transcode_jobs WHERE id = ?", [$jobId]);
            Database::query("UPDATE transcode_servers SET active_jobs = IF(active_jobs > 0, active_jobs - 1, 0) WHERE name = ?", [$server['name']]);
            return ['ok' => true, 'code' => 200];
        }

        if ($job['worker'] && $job['worker'] !== $server['name']) {
            return ['error' => 'Job not owned by this server', 'code' => 403];
        }

        $videoId = (int)$job['video_id'];

        if ($status === 'failed') {
            self::failJob($jobId, $error ?? 'Remote server reported failure');
        } else {
             
             
             
             
            $expected = self::expectedRenditionFile($job['task'], $videoId);
            $sid = $job['storage_id'] ? (int)$job['storage_id'] : null;
            if ($expected !== null && !self::renditionPresent($job['video_folder'], $expected, $sid)) {
                self::failJob($jobId, 'Rendition ' . $expected . ' missing on storage after completion — re-queued');
                $status = 'failed';
            } else {
                Database::update('transcode_jobs', ['status' => 'completed', 'progress' => 100, 'error_message' => null, 'completed_at' => date('Y-m-d H:i:s')], 'id = ?', [$jobId]);
            }
        }

        if ($status === 'completed') {
            if ($duration && $duration > 0) {
                Database::query('UPDATE videos SET duration = LEAST(duration, ?) WHERE id = ? AND duration > 0', [$duration, $videoId]);
                Database::query('UPDATE videos SET duration = ? WHERE id = ? AND duration = 0', [$duration, $videoId]);
            }
        }

        Database::query("UPDATE transcode_servers SET active_jobs = IF(active_jobs > 0, active_jobs - 1, 0) WHERE name = ?", [$server['name']]);

        try {
            self::checkAutoPublish($videoId);
        } catch (\Throwable $e) {
            error_log("TranscodeManager::handleCallback error: " . $e->getMessage());
        }

        return ['ok' => true, 'code' => 200];
    }

    public static function handleUpload(string $apiKey): array
    {
        $server = Database::fetchOne('SELECT id FROM transcode_servers WHERE api_key = ?', [$apiKey]);
        if (!$server) return ['error' => 'Invalid API key', 'code' => 401];

         
        if (empty($_POST) && empty($_FILES) && (int)($_SERVER['CONTENT_LENGTH'] ?? 0) > 0) {
            return ['error' => 'Upload too large (' . round((int)$_SERVER['CONTENT_LENGTH'] / 1048576) . 'MB) — increase post_max_size (current: ' . ini_get('post_max_size') . ')', 'code' => 413];
        }

        $jobId = (int)($_POST['job_id'] ?? 0);
        if (!$jobId) return ['error' => 'Missing job_id — POST data may be empty (check post_max_size=' . ini_get('post_max_size') . ')', 'code' => 400];

        $job = Database::fetchOne('SELECT j.*, v.video_folder as v_folder, v.storage_id as v_storage FROM transcode_jobs j LEFT JOIN videos v ON v.id = j.video_id WHERE j.id = ?', [$jobId]);
        if (!$job) return ['error' => 'Job not found', 'code' => 404];

         
        $folder = $job['video_folder'] ?: ($job['v_folder'] ?? null);
        $jobStorageId = $job['storage_id'] ?: ($job['v_storage'] ?? null);
        if (!$folder) {
            Database::query("DELETE FROM transcode_jobs WHERE id = ?", [$jobId]);
            return ['error' => 'Video was deleted', 'code' => 410];
        }

        if (empty($_FILES['file']) || $_FILES['file']['error'] !== UPLOAD_ERR_OK) {
            $err = isset($_FILES['file']) ? 'Upload error code: ' . $_FILES['file']['error'] : 'No file in request';
            return ['error' => $err, 'code' => 400];
        }
        $dir = VideoAsset::dir($folder);
        if (!is_dir($dir)) mkdir($dir, 0755, true);

        $filename = basename($_POST['filename'] ?? $_FILES['file']['name']);
        if (!preg_match('/^[a-zA-Z0-9_.-]+$/', $filename)) {
            return ['error' => 'Invalid filename', 'code' => 400];
        }
         
        $uext = strtolower(pathinfo($filename, PATHINFO_EXTENSION));
        if (!in_array($uext, ['mp4', 'webm', 'ogg', 'jpg', 'jpeg', 'png', 'webp', 'vtt'], true)) {
            return ['error' => 'Disallowed file type', 'code' => 400];
        }

        $dest = $dir . '/' . $filename;
        if (!move_uploaded_file($_FILES['file']['tmp_name'], $dest)) {
            return ['error' => 'Failed to save file', 'code' => 500];
        }

        return self::finalizeRendition($folder, $dir, $filename, trim($_POST['delete_source'] ?? ''), $jobStorageId ? (int)$jobStorageId : null);
    }

    

    public static function handleUploadChunk(string $apiKey): array
    {
        $server = Database::fetchOne('SELECT id FROM transcode_servers WHERE api_key = ?', [$apiKey]);
        if (!$server) return ['error' => 'Invalid API key', 'code' => 401];

        $jobId    = (int) ($_GET['job_id'] ?? 0);
        $offset   = (int) ($_GET['offset'] ?? -1);
        $total    = (int) ($_GET['total'] ?? -1);
        $filename = basename((string) ($_GET['filename'] ?? ''));
        if (!$jobId || $offset < 0 || $total <= 0) return ['error' => 'Missing chunk params', 'code' => 400];
        if (!preg_match('/^[a-zA-Z0-9_.-]+$/', $filename)) return ['error' => 'Invalid filename', 'code' => 400];
        $uext = strtolower(pathinfo($filename, PATHINFO_EXTENSION));
        if (!in_array($uext, ['mp4', 'webm', 'ogg', 'jpg', 'jpeg', 'png', 'webp', 'vtt'], true)) return ['error' => 'Disallowed file type', 'code' => 400];

        $job = Database::fetchOne('SELECT j.*, v.video_folder as v_folder, v.storage_id as v_storage FROM transcode_jobs j LEFT JOIN videos v ON v.id = j.video_id WHERE j.id = ?', [$jobId]);
        if (!$job) return ['error' => 'Job not found', 'code' => 404];
        $folder = $job['video_folder'] ?: ($job['v_folder'] ?? null);
        if (!$folder) { Database::query("DELETE FROM transcode_jobs WHERE id = ?", [$jobId]); return ['error' => 'Video was deleted', 'code' => 410]; }
        $sid = $job['storage_id'] ?: ($job['v_storage'] ?? null);
        $storageId = $sid ? (int) $sid : null;

        $dir = VideoAsset::dir($folder);
        if (!is_dir($dir)) mkdir($dir, 0755, true);
        $dest = $dir . '/' . $filename;

         
        $in  = fopen('php://input', 'rb');
        $out = fopen($dest, 'cb');
        if (!$in || !$out) return ['error' => 'open failed', 'code' => 500];
        flock($out, LOCK_EX);
        if ($offset === 0) ftruncate($out, 0);
        fseek($out, $offset);
        $written = 0;
        while (!feof($in)) { $b = fread($in, 1 << 20); if ($b === false || $b === '') break; $written += fwrite($out, $b); }
        fflush($out); flock($out, LOCK_UN); fclose($out); fclose($in);

        if ($offset + $written < $total) {
            return ['ok' => true, 'done' => false, 'received' => $offset + $written, 'code' => 200];
        }
        clearstatcache(true, $dest);
        if (filesize($dest) !== $total) return ['error' => 'size mismatch — resend', 'code' => 409];

        return self::finalizeRendition($folder, $dir, $filename, trim((string) ($_GET['delete_source'] ?? '')), $storageId);
    }

    

    private static function finalizeRendition(?string $folder, string $dir, string $filename, string $deleteSource, ?int $storageId): array
    {
        $dest = $dir . '/' . $filename;

        $ext = strtolower(pathinfo($filename, PATHINFO_EXTENSION));
        if (in_array($ext, ['mp4', 'webm', 'ogg'], true) && (@filesize($dest) ?: 0) > 0) {
            $probe = self::ffprobePath();
            if ($probe !== '') {
                $check = @shell_exec(escapeshellcmd($probe) . ' -v error -show_entries format=duration ' . escapeshellarg($dest) . ' 2>&1');
                if ($check !== null && (str_contains($check, 'moov atom not found') || str_contains($check, 'Invalid data'))) {
                    @unlink($dest);
                    return ['error' => 'Uploaded file is corrupted (incomplete transcode)', 'code' => 400];
                }
            }
        }

        if ($deleteSource !== '' && $deleteSource !== $filename && preg_match('/^[a-zA-Z0-9_.-]+$/', $deleteSource)) {
            $oldFile = $dir . '/' . $deleteSource;
            if (file_exists($oldFile)) @unlink($oldFile);
            if ($storageId && MediaStorage::isRemote($storageId)) {
                MediaStorage::deleteFile('uploads/videos/' . $folder . '/' . $deleteSource, $storageId);
            }
        }

        if ($storageId && MediaStorage::isRemote($storageId)) {
            if (!MediaStorage::put($dest, 'uploads/videos/' . $folder . '/' . $filename, $storageId)) {
                return ['error' => 'Failed to upload ' . $filename . ' to remote storage (FTP/S3 error)', 'code' => 503];
            }
            @unlink($dest);
        }

        return ['ok' => true, 'code' => 200];
    }

    public static function recoverStalled(): void
    {
         
        Database::query(
            "DELETE j FROM transcode_jobs j LEFT JOIN videos v ON v.id = j.video_id WHERE v.id IS NULL"
        );

         
        Database::query(
            "UPDATE transcode_jobs SET status = 'pending', progress = 0, retry_count = retry_count + 1, last_worker = worker, worker = NULL, started_at = NULL,
                    error_message = CONCAT('Server stalled (retry ', retry_count + 1, '/3)')
             WHERE status = 'processing' AND started_at < DATE_SUB(NOW(), INTERVAL 30 MINUTE)
             AND worker != ? AND retry_count < 3",
            [gethostname()]
        );
         
        Database::query(
            "UPDATE transcode_jobs SET status = 'pending', progress = 0, retry_count = retry_count + 1, last_worker = worker, worker = NULL, started_at = NULL,
                    error_message = CONCAT('Local stalled (retry ', retry_count + 1, '/3)')
             WHERE status = 'processing' AND started_at < DATE_SUB(NOW(), INTERVAL 2 HOUR)
             AND worker = ? AND retry_count < 3",
            [gethostname()]
        );
         
        $permFailed = Database::query(
            "UPDATE transcode_jobs SET status = 'failed', error_message = 'Failed after 3 recovery attempts', completed_at = NOW()
             WHERE status = 'processing' AND started_at < DATE_SUB(NOW(), INTERVAL 2 HOUR) AND retry_count >= 3"
        )->rowCount();

         
        if ($permFailed > 0) {
            Database::query(
                "UPDATE transcode_jobs sub
                 SET sub.status = 'failed',
                     sub.error_message = 'Skipped — source task failed after 3 retries. Fix the source issue and retry.',
                     sub.completed_at = NOW()
                 WHERE sub.task != 'source' AND sub.status IN ('pending', 'processing')
                 AND EXISTS (
                     SELECT 1 FROM (SELECT video_id FROM transcode_jobs WHERE task = 'source' AND status = 'failed' AND retry_count >= 3) src
                     WHERE src.video_id = sub.video_id
                 )"
            );
        }

         
        Database::query(
            "UPDATE transcode_servers SET active_jobs = GREATEST(0, (
                SELECT COUNT(*) FROM transcode_jobs WHERE status = 'processing' AND worker = transcode_servers.name
            ))"
        );

         
        $stuckVideos = Database::fetchAll(
            "SELECT DISTINCT v.id FROM videos v
             WHERE v.status IN ('processing', 'error')
             AND NOT EXISTS (SELECT 1 FROM transcode_jobs j WHERE j.video_id = v.id AND j.status IN ('pending','processing'))"
        );
        foreach ($stuckVideos as $sv) {
            self::checkAutoPublish((int)$sv['id']);
        }
    }

    public static function registerCron(): void
    {
        if (!self::isEnabled()) return;
        $interval = (int)Setting::get('transcode_interval', '30');

        CronManager::register('transcode', max(10, $interval), function () {
            self::recoverStalled();
            self::healthCheckServers();
            self::autoRetryStuck();

            $maxPerTick = (int)Setting::get('transcode_max_per_tick', '1');

            $pendingJobs = Database::transaction(function () {
                return Database::fetchAll(
                    "SELECT j.* FROM transcode_jobs j
                     WHERE j.status = 'pending' AND (j.started_at IS NULL OR j.started_at <= NOW())
                       AND (j.task = 'source' OR NOT EXISTS (
                           SELECT 1 FROM transcode_jobs js
                           WHERE js.video_id = j.video_id AND js.task = 'source' AND js.status != 'completed' AND NOT (js.status = 'failed' AND js.retry_count >= 3)
                       ))
                     ORDER BY (j.task != 'source') DESC, j.created_at ASC
                     LIMIT 10 FOR UPDATE SKIP LOCKED"
                );
            });

             
            $serversWithPendingSubs = Database::fetchAll(
                "SELECT DISTINCT js.worker FROM transcode_jobs j
                 JOIN transcode_jobs js ON js.video_id = j.video_id AND js.task = 'source' AND js.status = 'completed' AND js.worker IS NOT NULL
                 WHERE j.task != 'source' AND j.status = 'pending'"
            );
            $reservedServers = array_column($serversWithPendingSubs, 'worker');

            foreach ($pendingJobs as $job) {
                $server = null;

                if ($job['task'] !== 'source') {
                    $sourceWorker = Database::fetchOne(
                        "SELECT worker FROM transcode_jobs WHERE video_id = ? AND task = 'source' AND status = 'completed' AND worker IS NOT NULL",
                        [(int)$job['video_id']]
                    );
                    if ($sourceWorker && $sourceWorker['worker']) {
                        $workerName = $sourceWorker['worker'];
                        $candidate = Database::fetchOne(
                            "SELECT * FROM transcode_servers WHERE name = ? AND is_active = 1",
                            [$workerName]
                        );
                        if ($candidate && $candidate['status'] !== 'error' && (int)$candidate['active_jobs'] < (int)$candidate['max_jobs']) {
                            $server = $candidate;
                        } elseif (!$candidate || $candidate['status'] === 'error') {
                            $lastCheck = $candidate ? strtotime($candidate['last_check'] ?? '0') : 0;
                            $downMinutes = $lastCheck > 0 ? (int)((time() - $lastCheck) / 60) : 999;
                            if ($downMinutes >= 10) {
                                $server = self::pickServer();
                            }
                        }
                    } else {
                        $server = self::pickServer($job['last_worker'] ?? null);
                    }
                } else {
                     
                    $server = self::pickServer($job['last_worker'] ?? null);
                    if ($server && in_array($server['name'], $reservedServers, true)) {
                        $server = null;
                    }
                }

                if (!$server) continue;
                self::dispatchToServer($server, $job);
            }

            $localEnabled = self::isAvailable() && $maxPerTick > 0;
            if ($localEnabled) {
                $hostname = gethostname();
                $currentlyProcessing = (int)Database::fetchOne(
                    "SELECT COUNT(*) as cnt FROM transcode_jobs WHERE status = 'processing' AND worker = ?",
                    [$hostname]
                )['cnt'];
                $slotsAvailable = max(0, $maxPerTick - $currentlyProcessing);
                for ($i = 0; $i < $slotsAvailable; $i++) {
                    if (!self::processNext()) break;
                }
            }
        });
    }
}
