<?php

declare(strict_types=1);

class TranscodeWorkerScript
{
    public static function generate(): string
    {
        return <<<'WORKER'
<?php
/**
 * TubePress Transcode Worker
 *
 * Install on your transcoding server:
 *   1. Save this file as transcode-worker.php
 *   2. Install FFmpeg: sudo apt install ffmpeg
 *   3. Run: php transcode-worker.php --port=8090 --key=YOUR_API_KEY
 *   4. For production: put behind Nginx with HTTPS
 */

$opts = getopt('', ['port:', 'key:', 'help']);
if (isset($opts['help']) || !isset($opts['key'])) {
    echo "TubePress Transcode Worker\n\n";
    echo "Usage: php transcode-worker.php --key=YOUR_API_KEY [--port=8090]\n\n";
    echo "  --key    API key from TubePress admin (required)\n";
    echo "  --port   Port to listen on (default: 8090)\n";
    exit(0);
}

$API_KEY = $opts['key'];
$PORT = (int)($opts['port'] ?? 8090);
$MAX_JOBS = 2;
$WORK_DIR = __DIR__ . '/transcode_tmp';

if (!is_dir($WORK_DIR)) mkdir($WORK_DIR, 0755, true);

$ffmpeg = trim(shell_exec('which ffmpeg 2>/dev/null') ?? '');
if ($ffmpeg === '') { echo "ERROR: FFmpeg not found.\n"; exit(1); }
$ffprobe = trim(shell_exec('which ffprobe 2>/dev/null') ?? '');

// Detect best hardware encoder
$HW_ENCODER = detectEncoder($ffmpeg);

foreach (glob("{$WORK_DIR}/*") as $tmp) {
    if (is_file($tmp) && filemtime($tmp) < time() - 3600) @unlink($tmp);
}

pcntl_signal(SIGCHLD, function() {
    while (pcntl_waitpid(-1, $status, WNOHANG) > 0) {}
});

echo "TubePress Transcode Worker\n";
echo "FFmpeg:  {$ffmpeg}\n";
echo "Encoder: {$HW_ENCODER['name']} ({$HW_ENCODER['type']})\n";
echo "Listening on port {$PORT}...\n\n";

$socket = stream_socket_server("tcp://0.0.0.0:{$PORT}", $errno, $errstr);
if (!$socket) { echo "Cannot bind: {$errstr}\n"; exit(1); }

$CHILD_PIDS = [];
$_lastCleanup = time();
while (true) {
    $conn = @stream_socket_accept($socket, 5);
    if ($conn === false) {
        pcntl_signal_dispatch();
        // Periodic cleanup: remove cached files older than 30 minutes
        if (time() - $_lastCleanup > 300) {
            $_lastCleanup = time();
            foreach (glob("{$WORK_DIR}/*_cached.*") as $cf) {
                if (is_file($cf) && filemtime($cf) < time() - 1800) @unlink($cf);
            }
        }
        continue;
    }
    pcntl_signal_dispatch();
    // Read headers with timeout to prevent slowloris
    stream_set_timeout($conn, 10);
    $request = '';
    $headerBytes = 0;
    while ($line = fgets($conn, 4096)) {
        $request .= $line;
        $headerBytes += strlen($line);
        if ($headerBytes > 32768) break; // Max 32KB headers
        if (trim($line) === '') break;
        $info = stream_get_meta_data($conn);
        if ($info['timed_out']) break;
    }

    preg_match('/Content-Length:\s*(\d+)/i', $request, $clMatch);
    $contentLength = (int)($clMatch[1] ?? 0);
    $body = '';
    if ($contentLength > 0) {
        $remaining = $contentLength;
        while ($remaining > 0) {
            $chunk = fread($conn, min(8192, $remaining));
            if ($chunk === false || $chunk === '') break;
            $body .= $chunk;
            $remaining -= strlen($chunk);
        }
    }

    preg_match('/^(GET|POST)\s+(\S+)/', $request, $mm);
    $method = $mm[1] ?? 'GET'; $path = $mm[2] ?? '/';
    preg_match('/X-API-Key:\s*(\S+)/i', $request, $km);
    $requestKey = trim($km[1] ?? '');

    if ($requestKey !== $API_KEY && $path !== '/') { respond($conn, 401, ['error' => 'Unauthorized']); continue; }

    if ($path === '/health') {
        $load = sys_getloadavg();
        $cpuPct = getCpuUsage();
        $gpuPct = getGpuUsage();
        respond($conn, 200, [
            'status'=>'ok',
            'ffmpeg'=>trim(shell_exec("{$ffmpeg} -version 2>/dev/null | head -1") ?? ''),
            'encoder'=>$HW_ENCODER['name'].' ('.$HW_ENCODER['type'].')',
            'cpu'=>$cpuPct,
            'gpu'=>$gpuPct,
            'load'=>$load?round($load[0],2):0,
        ]);
        continue;
    }

    if ($method === 'POST' && $path === '/transcode') {
        foreach ($CHILD_PIDS as $k => $cpid) { if (pcntl_waitpid($cpid, $st, WNOHANG) !== 0) unset($CHILD_PIDS[$k]); }
        $data = json_decode($body, true);
        if (!$data) { respond($conn, 400, ['error' => 'Invalid JSON']); continue; }
        if (isset($data['max_jobs']) && (int)$data['max_jobs'] > 0) $MAX_JOBS = (int)$data['max_jobs'];
        if (count($CHILD_PIDS) >= $MAX_JOBS) { respond($conn, 503, ['error' => 'Busy (' . count($CHILD_PIDS) . '/' . $MAX_JOBS . ' jobs)']); continue; }
        respond($conn, 202, ['accepted' => true]);
        $pid = pcntl_fork();
        if ($pid === 0) { fclose($socket); processJob($data, $ffmpeg, $ffprobe, $WORK_DIR); exit(0); }
        if ($pid > 0) { $CHILD_PIDS[] = $pid; }
        if ($pid < 0) { echo "Fork failed\n"; }
        continue;
    }
    respond($conn, 404, ['error' => 'Not found']);
}

$_cpuPrev = null;
function getCpuUsage(): int {
    global $_cpuPrev;
    // Linux: compare /proc/stat snapshots between health calls (no sleep)
    if (is_readable('/proc/stat')) {
        $raw = file_get_contents('/proc/stat');
        if (preg_match('/^cpu\s+(.+)$/m', $raw, $m)) {
            $cur = array_map('intval', preg_split('/\s+/', trim($m[1])));
            if ($_cpuPrev !== null) {
                $idle = ($cur[3] - $_cpuPrev[3]) + (($cur[4] ?? 0) - ($_cpuPrev[4] ?? 0));
                $total = array_sum($cur) - array_sum($_cpuPrev);
                $_cpuPrev = $cur;
                if ($total > 0) return max(0, min(100, (int)round(($total - $idle) / $total * 100)));
            }
            $_cpuPrev = $cur;
        }
    }
    // macOS: vm_stat + sysctl (fast, no blocking)
    $idle = @shell_exec('sysctl -n kern.cp_time 2>/dev/null');
    if (!$idle) {
        $top = @shell_exec('ps -A -o %cpu 2>/dev/null');
        if ($top) {
            $sum = 0;
            foreach (explode("\n", $top) as $line) { $v = trim($line); if (is_numeric($v)) $sum += (float)$v; }
            $cores = (int)shell_exec('sysctl -n hw.ncpu 2>/dev/null') ?: 1;
            return max(0, min(100, (int)round($sum / $cores)));
        }
    }
    // Fallback: load average
    $load = sys_getloadavg();
    $cores = (int)shell_exec('nproc 2>/dev/null || sysctl -n hw.ncpu 2>/dev/null') ?: 1;
    return min(100, (int)round(($load ? $load[0] : 0) / $cores * 100));
}

function getGpuUsage(): int {
    global $HW_ENCODER;
    if ($HW_ENCODER['name'] === 'libx264') return -1;
    // NVIDIA
    $nv = @shell_exec('nvidia-smi --query-gpu=utilization.gpu --format=csv,noheader,nounits 2>/dev/null');
    if ($nv !== null && trim($nv) !== '') return max(0, min(100, (int)trim($nv)));
    // Intel
    $intel = @shell_exec('intel_gpu_top -s 100 -l 1 2>/dev/null');
    if ($intel && preg_match('/(\d+\.\d+)%\s+render/', $intel, $m)) return (int)round((float)$m[1]);
    // macOS: try powermetrics (may need sudo)
    $pm = @shell_exec('sudo -n powermetrics -n 1 -i 500 --samplers gpu_power 2>/dev/null');
    if ($pm && preg_match('/GPU Active:\s*([\d.]+)%/', $pm, $m)) return (int)round((float)$m[1]);
    // macOS fallback: ioreg GPU busy percentage
    $ioreg = @shell_exec('ioreg -r -d 1 -c IOAccelerator 2>/dev/null');
    if ($ioreg && preg_match('/"Device Utilization %"=(\d+)/', $ioreg, $m)) return (int)$m[1];
    if ($ioreg && preg_match('/"GPU Activity\(%\)"=(\d+)/', $ioreg, $m)) return (int)$m[1];
    // Last resort: check active ffmpeg processes using GPU
    $procs = (int)trim(shell_exec('pgrep -f "' . $HW_ENCODER['name'] . '" 2>/dev/null | wc -l') ?? '0');
    if ($procs > 0) return min(100, $procs * 30);
    return 0;
}

function detectEncoder(string $ffmpeg): array {
    $encoders = trim(shell_exec($ffmpeg . ' -encoders 2>/dev/null') ?? '');
    $cpu = ['name' => 'libx264', 'type' => 'CPU', 'opts' => '-c:v libx264 -preset medium -crf 20'];
    $candidates = [
        'h264_nvenc'        => ['opts' => '-c:v h264_nvenc -preset p4 -cq 23',                              'type' => 'GPU/NVIDIA'],
        'h264_amf'          => ['opts' => '-c:v h264_amf -quality balanced -rc cqp -qp_i 23 -qp_p 23',     'type' => 'GPU/AMD'],
        'h264_qsv'          => ['opts' => '-c:v h264_qsv -preset medium -global_quality 23',                'type' => 'GPU/Intel'],
        'h264_videotoolbox' => ['opts' => '-c:v h264_videotoolbox -q:v 53 -allow_sw 1',                     'type' => 'GPU/Apple'],
    ];
    foreach ($candidates as $name => $cfg) {
        if (!str_contains($encoders, $name)) continue;
        $test = shell_exec($ffmpeg . ' -f lavfi -i nullsrc=s=64x64:d=0.1 ' . $cfg['opts'] . ' -frames:v 1 -f null - 2>&1') ?? '';
        $testCheck = str_replace('muxing overhead: unknown', '', strtolower($test));
        if (!str_contains($testCheck, 'error') && !str_contains($testCheck, 'cannot') && !str_contains($testCheck, 'failed') && !str_contains($testCheck, 'not found') && !str_contains($testCheck, 'not supported') && !str_contains($testCheck, 'no such') && !str_contains($testCheck, 'unknown')) {
            return ['name' => $name, 'type' => $cfg['type'], 'opts' => $cfg['opts']];
        }
        echo "  Note: {$name} listed but not functional, skipping.\n";
    }
    return $cpu;
}

function hwEncoderOpts(?string $baseOpts, array $hwEncoder): string {
    $baseOpts = (string)$baseOpts;
    // Extract CRF from original opts to map quality level
    $crf = 23; // default
    if (preg_match('/-crf\s+(\d+)/', $baseOpts, $m)) $crf = (int)$m[1];

    // Strip video-codec-specific options, keep audio/container AND maxrate/bufsize (bitrate control works with all encoders)
    $stripped = preg_replace('/-(?:c:v|vcodec)\s+\S+/', '', $baseOpts);
    $stripped = preg_replace('/-(?:preset|crf|cq|qp_i|qp_p|global_quality|q:v|quality|rc|tune|profile:v|level:v|level|x264-params|x264opts|bf|refs|g|keyint_min|pix_fmt)\s+\S+/', '', $stripped);
    $stripped = trim(preg_replace('/\s+/', ' ', $stripped));

    // Map CRF to GPU-specific quality (lower CRF = higher quality)
    $name = $hwEncoder['name'];
    if ($name === 'h264_nvenc') {
        $gpuOpts = "-c:v h264_nvenc -preset p4 -cq {$crf}";
    } elseif ($name === 'h264_amf') {
        $gpuOpts = "-c:v h264_amf -quality balanced -rc cqp -qp_i {$crf} -qp_p {$crf}";
    } elseif ($name === 'h264_qsv') {
        $gpuOpts = "-c:v h264_qsv -preset medium -global_quality {$crf}";
    } elseif ($name === 'h264_videotoolbox') {
        // VideoToolbox q:v scale: ~CRF 20→50, CRF 23→58, CRF 25→62, CRF 28→68
        $vtQ = max(30, min(80, (int)round(40 + ($crf - 18) * 2.5)));
        $gpuOpts = "-c:v h264_videotoolbox -q:v {$vtQ} -allow_sw 1";
    } else {
        $gpuOpts = "-c:v libx264 -preset medium -crf {$crf}";
    }

    return $gpuOpts . ($stripped !== '' ? ' ' . $stripped : '');
}

function respond($conn, int $code, array $data): void {
    $st = match($code){200=>'OK',202=>'Accepted',401=>'Unauthorized',503=>'Service Unavailable',default=>'Error'};
    $r = json_encode($data);
    fwrite($conn, "HTTP/1.1 {$code} {$st}\r\nContent-Type: application/json\r\nContent-Length: ".strlen($r)."\r\nConnection: close\r\n\r\n{$r}");
    fclose($conn);
}

function jobStillValid(string $callbackUrl, string $apiKey, int $jobId): bool {
    if ($callbackUrl === '' || $jobId <= 0) return true;
    $checkUrl = str_replace('/transcode-callback', '/transcode-job-check', $callbackUrl) . '?job_id=' . $jobId;
    $ch = curl_init($checkUrl);
    curl_setopt_array($ch, [CURLOPT_HTTPHEADER => ['X-API-Key: ' . $apiKey], CURLOPT_RETURNTRANSFER => true, CURLOPT_TIMEOUT => 5, CURLOPT_CONNECTTIMEOUT => 3]);
    $body = curl_exec($ch);
    $code = (int)curl_getinfo($ch, CURLINFO_HTTP_CODE);
    curl_close($ch);
    if ($code === 0) return true; // CMS unreachable — assume valid, don't block
    if ($code >= 500) return false; // CMS error — don't waste resources
    if ($code !== 200) return true;
    $json = json_decode($body, true);
    return !isset($json['valid']) || $json['valid'] === true;
}

function processJob(array $data, string $ffmpeg, string $ffprobe, string $workDir): void {
    global $HW_ENCODER;
    $jobId = (int)($data['job_id'] ?? 0);
    $sourceUrl = $data['source_url'] ?? '';
    $task = $data['task'] ?? '';
    $callbackUrl = $data['callback_url'] ?? '';
    $uploadUrl = $data['upload_url'] ?? '';
    $apiKey = $data['api_key'] ?? '';
    $folder = $data['video_folder'] ?? '';
    $deleteSource = $data['delete_source'] ?? '';
    $trimStart = (int)($data['trim_start'] ?? 0);
    $format = $data['format'] ?? null;
    $watermark = $data['watermark'] ?? null;
    $timelineCfg = $data['timeline'] ?? null;

    echo "\n[".date('H:i:s')."] Job #{$jobId}: {$task} for {$folder}\n";

    if (!jobStillValid($callbackUrl, $apiKey, $jobId)) {
        echo "[".date('H:i:s')."]   Skipped — job no longer valid on CMS.\n";
        return;
    }

    $cachedSrc = ($folder !== '') ? "{$workDir}/{$folder}_cached.mp4" : '';
    $cacheLock = ($folder !== '') ? "{$workDir}/{$folder}_cached.lock" : '';
    $srcFile = "{$workDir}/{$jobId}_source.mp4";
    $wmFile = null;
    $progressUrl = str_replace('/transcode-upload', '/transcode-progress', $uploadUrl ?: ($callbackUrl ? str_replace('/transcode-callback', '/transcode-progress', $callbackUrl) : ''));

    $usedCache = false;
    if ($cachedSrc !== '' && file_exists($cachedSrc) && !file_exists($cacheLock) && filesize($cachedSrc) > 1000) {
        echo "[".date('H:i:s')."]   Using cached source (" . round(filesize($cachedSrc)/1048576, 1) . " MB)\n";
        copy($cachedSrc, $srcFile);
        $usedCache = true;
    }
    if (!$usedCache) {
        if ($cacheLock !== '') @file_put_contents($cacheLock, (string)$jobId);
        reportStage($progressUrl, $apiKey, $jobId, 'downloading', 0);
        $dlErr = downloadFileWithProgress($sourceUrl, $srcFile, $progressUrl, $apiKey, $jobId);
        if ($dlErr !== '') {
            if ($cacheLock !== '') @unlink($cacheLock);
            notifyCallback($callbackUrl, $apiKey, $jobId, 'failed', 'Source download: ' . $dlErr);
            @unlink($srcFile); return;
        }
        if ($cachedSrc !== '') { @copy($srcFile, $cachedSrc); @unlink($cacheLock); }
        echo "[".date('H:i:s')."]   Downloaded " . round(filesize($srcFile)/1048576, 1) . " MB\n";
    }

    if ($watermark && !empty($watermark['image_url'])) {
        $wmFile = "{$workDir}/{$jobId}_watermark.png";
        if (downloadFile($watermark['image_url'], $wmFile) !== '') { $wmFile = null; $watermark = null; }
    }

    $outFiles = [];
    $duration = null;

    if ($format && $task !== 'timeline') {
        $outSuffix = $task === 'source' ? 'reencoded' : $task;
        $outFile = "{$workDir}/{$jobId}_{$outSuffix}.mp4";
        $outFilename = $format['file'] ?? "video_{$task}.mp4";

        $vf = buildVideoFilter($format, $watermark, $wmFile);
        $isComplex = str_contains($vf, '[') && str_contains($vf, ']');

        $cmd = $ffmpeg;
        if ($trimStart > 0) $cmd .= ' -ss ' . $trimStart;
        $cmd .= ' -i ' . escapeshellarg($srcFile);
        if ($wmFile && $watermark) $cmd .= ' -i ' . escapeshellarg($wmFile);
        $cmd .= ($isComplex ? ' -filter_complex ' : ' -vf ') . escapeshellarg($vf);

        $ffmpegOpts = trim((string)($format['ffmpeg_options'] ?? ''));
        // Strip shell metacharacters: ffmpeg_options is concatenated into a shell
        // command (shell_exec / proc_open) downstream, so allow plain flags/values only.
        // str_replace (not a regex) so it can never fail to compile and null out the opts.
        $ffmpegOpts = str_replace([';', '|', '&', '$', '`', '(', ')', '<', '>', "\n", "\r", "\\"], '', $ffmpegOpts);
        if ($ffmpegOpts !== '') {
            $cmd .= ' ' . hwEncoderOpts($ffmpegOpts, $HW_ENCODER);
        } else {
            $cmd .= ' ' . $HW_ENCODER['opts'] . ' -c:a aac -b:a 128k -ar 44100 -movflags +faststart';
        }
        $cmd .= ' -y ' . escapeshellarg($outFile);

        $srcDurationForProgress = getVideoDuration($ffprobe, $srcFile) ?: (int)($data['duration'] ?? 0);
        reportStage($progressUrl, $apiKey, $jobId, 'transcoding', 0);
        echo "[".date('H:i:s')."]   Transcoding {$task} ({$HW_ENCODER['name']})...\n";
        $output = runFfmpegWithProgress($cmd, $srcDurationForProgress, $progressUrl, $apiKey, $jobId);

        if ($output === '__CANCELLED__') {
            @unlink($outFile);
            cleanupAll($srcFile, $wmFile, $outFiles, $workDir, $folder); return;
        }

        if ((!file_exists($outFile) || filesize($outFile) < 1000) && $HW_ENCODER['name'] !== 'libx264') {
            echo "[".date('H:i:s')."]   GPU failed, retrying with CPU...\n";
            $cmd = $ffmpeg;
            if ($trimStart > 0) $cmd .= ' -ss ' . $trimStart;
            $cmd .= ' -i ' . escapeshellarg($srcFile);
            if ($wmFile && $watermark) $cmd .= ' -i ' . escapeshellarg($wmFile);
            $cmd .= ($isComplex ? ' -filter_complex ' : ' -vf ') . escapeshellarg($vf);
            $cmd .= ' -c:v libx264 -preset medium -crf 20 -c:a aac -b:a 128k -ar 44100 -movflags +faststart -y ' . escapeshellarg($outFile);
            $output = runFfmpegWithProgress($cmd, $srcDurationForProgress, $progressUrl, $apiKey, $jobId);
        }

        if ($output === '__CANCELLED__') {
            @unlink($outFile);
            cleanupAll($srcFile, $wmFile, $outFiles, $workDir, $folder); return;
        }

        if (!file_exists($outFile) || filesize($outFile) < 1000) {
            notifyCallback($callbackUrl, $apiKey, $jobId, 'failed', 'FFmpeg failed: ' . substr($output ?? '', -200));
            cleanupAll($srcFile, $wmFile, $outFiles, $workDir, $folder); return;
        }
        echo "[".date('H:i:s')."]   Transcode OK: {$outFilename} (" . round(filesize($outFile)/1048576, 1) . " MB)\n";
        $intros = $data['intros'] ?? null;
        if ($intros && ($format['width'] ?? 0) > 0) {
            concatIntrosWorker($ffmpeg, $outFile, (int)$format['width'], (int)$format['height'], $intros, $workDir, $jobId);
        }

        $outFiles[] = ['local' => $outFile, 'name' => $outFilename];
        $duration = getVideoDuration($ffprobe, $outFile);

    } elseif ($task === 'timeline') {
        $srcDuration = getVideoDuration($ffprobe, $srcFile) ?: 60;
        $timelineStartOffset = max(0, (int)($timelineCfg['start'] ?? 0));
        $totalSeek = $trimStart + $timelineStartOffset;
        if ($totalSeek >= $srcDuration - 5) $totalSeek = $trimStart;
        $effectiveDuration = max(1, $srcDuration - $totalSeek);
        $count = max(5, min(60, (int)($timelineCfg['count'] ?? 10)));
        $interval = $effectiveDuration / $count;
        $tw = 160; $th = 90; $cols = 10;
        $rows = (int)ceil($count / $cols);
        $fps = 1 / max(0.5, $interval);
        $ssArg = $totalSeek > 0 ? ' -ss ' . $totalSeek : '';

        $spriteFile = "{$workDir}/{$jobId}_timeline.jpg";
        $cmd = $ffmpeg . $ssArg
            . ' -i ' . escapeshellarg($srcFile)
            . ' -vf ' . escapeshellarg("fps={$fps},scale={$tw}:{$th}:force_original_aspect_ratio=decrease,pad={$tw}:{$th}:(ow-iw)/2:(oh-ih)/2,tile={$cols}x{$rows}")
            . ' -frames:v 1 -q:v 5 -y ' . escapeshellarg($spriteFile) . ' 2>&1';
        shell_exec($cmd);

        if (!file_exists($spriteFile) || filesize($spriteFile) < 500) {
            notifyCallback($callbackUrl, $apiKey, $jobId, 'failed', 'Timeline generation failed');
            cleanupAll($srcFile, $wmFile, $outFiles, $workDir, $folder); return;
        }

        $vtt = "WEBVTT\n\n";
        for ($i = 0; $i < $count; $i++) {
            $s = $i * $interval; $e = min(($i+1)*$interval, $effectiveDuration);
            $x = ($i % $cols) * $tw; $y = (int)floor($i / $cols) * $th;
            $vtt .= fmtVtt($s) . ' --> ' . fmtVtt($e) . "\ntimeline.jpg#xywh={$x},{$y},{$tw},{$th}\n\n";
        }
        $vttFile = "{$workDir}/{$jobId}_timeline.vtt";
        file_put_contents($vttFile, $vtt);

        $outFiles[] = ['local' => $spriteFile, 'name' => 'timeline.jpg'];
        $outFiles[] = ['local' => $vttFile, 'name' => 'timeline.vtt'];

    } else {
        notifyCallback($callbackUrl, $apiKey, $jobId, 'failed', "Unknown task: {$task}");
        cleanupAll($srcFile, $wmFile, $outFiles, $workDir, $folder); return;
    }

    if ($uploadUrl === '') {
        notifyCallback($callbackUrl, $apiKey, $jobId, 'failed', 'No upload URL — check site_url in CMS settings');
        cleanupAll($srcFile, $wmFile, $outFiles, $workDir, $folder); return;
    }

    if (!jobStillValid($callbackUrl, $apiKey, $jobId)) {
        echo "[".date('H:i:s')."]   Skipped upload — job no longer valid on CMS.\n";
        cleanupAll($srcFile, $wmFile, $outFiles, $workDir, $folder); return;
    }

    reportStage($progressUrl, $apiKey, $jobId, 'uploading', 0);
    echo "[".date('H:i:s')."]   Uploading to CMS...\n";
    foreach ($outFiles as $of) {
        echo "[".date('H:i:s')."]     {$of['name']} (" . round(filesize($of['local'])/1048576, 1) . " MB)... ";
        $upErr = uploadToCMS($uploadUrl, $apiKey, $jobId, $of['local'], $of['name'], $deleteSource);
        if ($upErr !== '') {
            echo "FAILED: {$upErr}\n";
            notifyCallback($callbackUrl, $apiKey, $jobId, 'failed', 'Upload failed (' . $of['name'] . '): ' . $upErr);
            cleanupAll($srcFile, $wmFile, $outFiles, $workDir, $folder); return;
        }
        echo "OK\n";
    }

    if (($data['is_source_reencode'] ?? false) && $cachedSrc !== '' && !empty($outFiles)) {
        $tmpCache = $cachedSrc . '.tmp';
        if (@copy($outFiles[0]['local'], $tmpCache) && filesize($tmpCache) > 1000) {
            @unlink($cachedSrc);
            @rename($tmpCache, $cachedSrc);
        } else {
            @unlink($tmpCache);
        }
    }

    notifyCallback($callbackUrl, $apiKey, $jobId, 'completed', '', ['duration' => $duration ?? 0]);
    cleanupAll($srcFile, $wmFile, $outFiles, $workDir, $folder);
    echo "[".date('H:i:s')."] Job #{$jobId} done.\n";
}

function buildVideoFilter(array $format, ?array $wm, ?string $wmFile): string {
    $w = (int)($format['width'] ?? 1280); $h = (int)($format['height'] ?? 720);
    $keepAspect = $format['keep_aspect'] ?? true;
    $scale = $keepAspect
        ? "scale={$w}:{$h}:force_original_aspect_ratio=decrease,pad={$w}:{$h}:(ow-iw)/2:(oh-ih)/2"
        : "scale={$w}:{$h}";

    if ($wm && $wmFile && file_exists($wmFile)) {
        $pos = $wm['position'] ?? 'bottom-right';
        $opacity = max(10, min(100, (int)($wm['opacity'] ?? 80))) / 100;
        $maxW = max(5, min(50, (int)($wm['max_width'] ?? 15)));
        $paddingPct = max(0, min(15, (int)($wm['padding'] ?? 2)));
        $wmScaleW = (int)round($w * $maxW / 100);
        $padX = (int)round($w * $paddingPct / 100);
        $padY = (int)round($h * $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}",
        };
        return "[0:v]{$scale}[scaled];[1:v]scale={$wmScaleW}:-1,format=rgba,colorchannelmixer=aa={$opacity}[wm];[scaled][wm]overlay={$overlayPos}";
    }
    return $scale;
}

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

    $proc = @proc_open($cmd . ' 2>&1', [0 => ['file', '/dev/null', 'r'], 1 => ['pipe', 'w']], $pipes);
    if (!is_resource($proc)) return shell_exec($cmd . ' 2>&1') ?? '';

    stream_set_blocking($pipes[1], false);
    $output = '';
    $lastReport = 0;
    $lastPct = -1;
    $cancelled = false;

    while (true) {
        $chunk = @fread($pipes[1], 8192);
        if ($chunk !== false && $chunk !== '') {
            $output .= $chunk;
        } else {
            $status = proc_get_status($proc);
            if (!$status['running']) break;
            usleep(500000);
            continue;
        }

        if (preg_match_all('/time=(\d+):(\d+):(\d+\.\d+)/', $output, $matches, PREG_SET_ORDER)) {
            $last = end($matches);
            $currentTime = (int)$last[1] * 3600 + (int)$last[2] * 60 + (float)$last[3];
            $pct = min(99, max(0, (int)round($currentTime / $totalDuration * 100)));
            if ($pct !== $lastPct && (time() - $lastReport) >= 3) {
                $lastPct = $pct;
                $lastReport = time();
                echo "\r  Progress: {$pct}%";
                $ch = curl_init($progressUrl);
                curl_setopt_array($ch, [CURLOPT_POST=>true, CURLOPT_POSTFIELDS=>json_encode(['job_id'=>$jobId,'progress'=>$pct,'stage'=>'transcoding']), CURLOPT_HTTPHEADER=>['Content-Type: application/json','X-API-Key: '.$apiKey], CURLOPT_RETURNTRANSFER=>true, CURLOPT_TIMEOUT=>3]);
                $resp = @curl_exec($ch); unset($ch);
                if ($resp) {
                    $r = json_decode($resp, true);
                    if (isset($r['cancel']) && $r['cancel']) {
                        echo "\n[".date('H:i:s')."]   Cancelled by CMS.\n";
                        $cancelled = true;
                        $pid = proc_get_status($proc)['pid'] ?? 0;
                        if ($pid > 0) @posix_kill($pid, 15);
                        break;
                    }
                }
            }
        }
    }

    fclose($pipes[1]);
    proc_close($proc);
    echo "\n";
    return $cancelled ? '__CANCELLED__' : $output;
}

function fmtVtt(float $s): string {
    return sprintf('%02d:%02d:%02d.%03d', (int)($s/3600), (int)($s%3600/60), (int)($s%60), (int)(($s-floor($s))*1000));
}

function getVideoDuration(string $ffprobe, string $file): int {
    if ($ffprobe === '' || !file_exists($file)) return 0;
    $json = shell_exec($ffprobe.' -v quiet -print_format json -show_format '.escapeshellarg($file).' 2>/dev/null');
    if (!$json) return 0;
    $data = json_decode($json, true);
    return (int)(float)($data['format']['duration'] ?? 0);
}

function concatIntrosWorker(string $ffmpeg, string $videoFile, int $w, int $h, array $intros, string $workDir, int $jobId): void {
    $parts = [];
    $tmpFiles = [];

    // Download + re-encode pre-roll
    if (!empty($intros['preroll_url'])) {
        $preSrc = "{$workDir}/{$jobId}_preroll_src.mp4";
        $preTmp = "{$workDir}/{$jobId}_preroll.mp4";
        $err = downloadFile($intros['preroll_url'], $preSrc);
        if ($err === '') {
            $cmd = $ffmpeg . ' -i ' . escapeshellarg($preSrc)
                . ' -vf ' . escapeshellarg("scale={$w}:{$h}:force_original_aspect_ratio=decrease,pad={$w}:{$h}:(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;
        }
        $tmpFiles[] = $preSrc;
        $tmpFiles[] = $preTmp;
    }

    $parts[] = $videoFile;

    // Download + re-encode post-roll
    if (!empty($intros['postroll_url'])) {
        $postSrc = "{$workDir}/{$jobId}_postroll_src.mp4";
        $postTmp = "{$workDir}/{$jobId}_postroll.mp4";
        $err = downloadFile($intros['postroll_url'], $postSrc);
        if ($err === '') {
            $cmd = $ffmpeg . ' -i ' . escapeshellarg($postSrc)
                . ' -vf ' . escapeshellarg("scale={$w}:{$h}:force_original_aspect_ratio=decrease,pad={$w}:{$h}:(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;
        }
        $tmpFiles[] = $postSrc;
        $tmpFiles[] = $postTmp;
    }

    if (count($parts) > 1) {
        $listFile = "{$workDir}/{$jobId}_concat.txt";
        $listContent = '';
        foreach ($parts as $p) $listContent .= "file " . escapeshellarg($p) . "\n";
        file_put_contents($listFile, $listContent);

        $concatOut = "{$workDir}/{$jobId}_concat.mp4";
        $cmd = $ffmpeg . ' -f concat -safe 0 -i ' . escapeshellarg($listFile)
            . ' -c copy -movflags +faststart -y ' . escapeshellarg($concatOut) . ' 2>&1';
        shell_exec($cmd);

        if (file_exists($concatOut) && filesize($concatOut) > 1000) {
            @unlink($videoFile);
            rename($concatOut, $videoFile);
            echo "  Intro/outro concatenated.\n";
        }
        @unlink($listFile);
        @unlink($concatOut);
    }

    foreach ($tmpFiles as $f) { if (file_exists($f)) @unlink($f); }
}

function reportStage(string $progressUrl, string $apiKey, int $jobId, string $stage, int $pct): void {
    if ($progressUrl === '' || $jobId <= 0) return;
    $ch = curl_init($progressUrl);
    curl_setopt_array($ch, [CURLOPT_POST=>true, CURLOPT_POSTFIELDS=>json_encode(['job_id'=>$jobId,'progress'=>$pct,'stage'=>$stage]), CURLOPT_HTTPHEADER=>['Content-Type: application/json','X-API-Key: '.$apiKey], CURLOPT_RETURNTRANSFER=>true, CURLOPT_TIMEOUT=>3]);
    @curl_exec($ch); unset($ch);
}

function _curlDiagnose(string $err, int $errno, int $code, array $info, string $url): string {
    $parsed = parse_url($url);
    $host = $parsed['host'] ?? '?';
    $scheme = $parsed['scheme'] ?? 'http';
    $path = $parsed['path'] ?? '/';
    $shortUrl = $scheme . '://' . $host . $path;
    if (strlen($shortUrl) > 120) $shortUrl = substr($shortUrl, 0, 117) . '...';

    $dns = round(($info['namelookup_time'] ?? 0) * 1000);
    $conn = round(($info['connect_time'] ?? 0) * 1000);
    $total = round(($info['total_time'] ?? 0) * 1000);
    $down = (int)($info['size_download'] ?? 0);
    $ip = $info['primary_ip'] ?? '';
    $diag = "DNS {$dns}ms > Connect {$conn}ms > Total {$total}ms";
    if ($ip !== '') $diag .= " | IP: {$ip}";
    if ($down > 0) $diag .= " | Downloaded: " . round($down / 1048576, 1) . "MB";

    if ($err !== '') {
        $type = match($errno) {
            6  => "DNS failed for \"{$host}\"",
            7  => "Connection refused by " . ($ip ?: $host) . " on port " . ($parsed['port'] ?? ($scheme === 'https' ? 443 : 80)),
            28 => $down > 0
                    ? 'Download stalled after ' . round($down / 1048576, 1) . 'MB'
                    : ($conn > 0 ? 'Server connected but never sent data' : "Cannot reach \"{$host}\" (timeout)"),
            35 => "SSL handshake failed for \"{$host}\"",
            51 => "SSL certificate invalid for \"{$host}\"",
            56 => 'Connection reset by server' . ($down > 0 ? ' after ' . round($down / 1048576, 1) . 'MB' : ''),
            default => $err,
        };
        return "{$type} | URL: {$shortUrl} | {$diag}";
    }

    if ($code === 0) return "No response (timeout or DNS) | URL: {$shortUrl} | {$diag}";
    if ($code === 403) return "HTTP 403 Forbidden | URL: {$shortUrl} | {$diag}";
    if ($code === 404) return "HTTP 404 Not Found | URL: {$shortUrl} | {$diag}";
    if ($code >= 400) return "HTTP {$code} | URL: {$shortUrl} | {$diag}";

    return '';
}

function downloadFileWithProgress(string $url, string $dest, string $progressUrl, string $apiKey, int $jobId): string {
    if ($url === '') return 'No source URL provided';
    $ch = curl_init($url);
    $fh = fopen($dest, 'wb');
    curl_setopt_array($ch, [CURLOPT_FILE=>$fh, CURLOPT_FOLLOWLOCATION=>true, CURLOPT_TIMEOUT=>600, CURLOPT_CONNECTTIMEOUT=>30, CURLOPT_NOPROGRESS=>false,
        CURLOPT_PROGRESSFUNCTION=>function($resource, $dlTotal, $dlNow) use ($progressUrl, $apiKey, $jobId) {
            static $lastReport = 0;
            if ($dlTotal > 0 && (time() - $lastReport) >= 3) {
                $pct = min(99, (int)round($dlNow / $dlTotal * 100));
                reportStage($progressUrl, $apiKey, $jobId, 'downloading', $pct);
                $lastReport = time();
            }
            return 0;
        }
    ]);
    curl_exec($ch);
    $code = (int)curl_getinfo($ch, CURLINFO_HTTP_CODE);
    $err = curl_error($ch);
    $errno = curl_errno($ch);
    $info = curl_getinfo($ch);
    $contentType = curl_getinfo($ch, CURLINFO_CONTENT_TYPE) ?: '';
    $expectedSize = (int)curl_getinfo($ch, CURLINFO_CONTENT_LENGTH_DOWNLOAD);
    unset($ch); fclose($fh);
    $diag = _curlDiagnose($err, $errno, $code, $info, $url);
    if ($diag !== '') return $diag;
    if (!file_exists($dest) || filesize($dest) < 1000) return 'Downloaded file is empty (' . @filesize($dest) . ' bytes) | URL: ' . $url;
    if ($contentType !== '' && str_contains($contentType, 'text/html')) return 'Server returned HTML instead of video (nginx error page?) | URL: ' . $url;
    if ($expectedSize > 0 && filesize($dest) < $expectedSize * 0.95) return 'Download truncated: got ' . round(filesize($dest)/1048576,1) . 'MB of ' . round($expectedSize/1048576,1) . 'MB | URL: ' . $url;
    return '';
}

function downloadFile(string $url, string $dest): string {
    if ($url === '') return 'No source URL provided';
    $ch = curl_init($url); $fh = fopen($dest, 'wb');
    curl_setopt_array($ch, [CURLOPT_FILE=>$fh, CURLOPT_FOLLOWLOCATION=>true, CURLOPT_TIMEOUT=>600, CURLOPT_CONNECTTIMEOUT=>30]);
    curl_exec($ch);
    $code = (int)curl_getinfo($ch, CURLINFO_HTTP_CODE);
    $err = curl_error($ch);
    $errno = curl_errno($ch);
    $info = curl_getinfo($ch);
    unset($ch); fclose($fh);
    $diag = _curlDiagnose($err, $errno, $code, $info, $url);
    if ($diag !== '') return $diag;
    if (!file_exists($dest) || filesize($dest) < 100) return 'Downloaded file is empty or corrupted (' . (file_exists($dest) ? filesize($dest) . ' bytes)' : 'missing)') . ' | URL: ' . $url;
    return '';
}

function uploadToCMSChunked(string $url, string $apiKey, int $jobId, string $filePath, string $filename, string $deleteSource): string {
    // Native Cloudflare 100 MB bypass: POST the rendition in <100 MB raw-body chunks
    // (metadata in the query string), reassembled server-side at each offset.
    $total = filesize($filePath);
    $chunkUrl = $url . '-chunk';
    $chunkSize = 64 * 1024 * 1024; // < Cloudflare 100 MB; streamed so RAM stays flat
    $fp = fopen($filePath, 'rb');
    if (!$fp) return 'Cannot open rendition for chunked upload';
    $offset = 0;
    while ($offset < $total) {
        $thisLen = (int)min($chunkSize, $total - $offset);
        if ($thisLen <= 0) break;
        $isLast = ($offset + $thisLen) >= $total;
        fseek($fp, $offset);
        $sent = 0;
        $q = http_build_query(['job_id'=>$jobId,'filename'=>$filename,'offset'=>$offset,'total'=>$total]);
        if ($isLast && $deleteSource !== '') $q .= '&delete_source=' . rawurlencode($deleteSource);
        $ch = curl_init($chunkUrl . '?' . $q);
        curl_setopt_array($ch, [
            CURLOPT_UPLOAD => true,
            CURLOPT_CUSTOMREQUEST => 'POST',
            CURLOPT_INFILESIZE => $thisLen,
            CURLOPT_READFUNCTION => function ($c, $fd, $maxLen) use ($fp, &$sent, $thisLen) {
                if ($sent >= $thisLen) return '';
                $buf = fread($fp, (int)min($maxLen, $thisLen - $sent));
                if ($buf === false) return '';
                $sent += strlen($buf);
                return $buf;
            },
            CURLOPT_HTTPHEADER => ['X-API-Key: '.$apiKey, 'Content-Type: application/octet-stream', 'Expect:'],
            CURLOPT_RETURNTRANSFER => true,
            CURLOPT_TIMEOUT => 1800,
        ]);
        $body = curl_exec($ch);
        $code = (int)curl_getinfo($ch, CURLINFO_HTTP_CODE);
        $err = curl_error($ch);
        unset($ch);
        if ($err !== '') { fclose($fp); return "Chunk connection failed at {$offset}: {$err}"; }
        if ($code === 413) { fclose($fp); return "Chunk rejected (HTTP 413) at offset {$offset} - chunk {$thisLen}B exceeds proxy cap"; }
        if ($code < 200 || $code >= 300) { fclose($fp); return "Chunk HTTP {$code} at offset {$offset}: " . substr((string)$body, 0, 200); }
        $json = json_decode((string)$body, true);
        if (isset($json['error'])) { fclose($fp); return 'CMS rejected chunk: ' . $json['error']; }
        $offset += $thisLen;
    }
    fclose($fp);
    return '';
}

function uploadToCMS(string $url, string $apiKey, int $jobId, string $filePath, string $filename, string $deleteSource = ''): string {
    $fileSize = filesize($filePath);
    if ($fileSize > 95 * 1024 * 1024) {
        return uploadToCMSChunked($url, $apiKey, $jobId, $filePath, $filename, $deleteSource);
    }
    $fields = ['job_id'=>$jobId,'filename'=>$filename,'file'=>new CURLFile($filePath,'',$filename)];
    if ($deleteSource !== '') $fields['delete_source'] = $deleteSource;
    $ch = curl_init($url);
    curl_setopt_array($ch, [CURLOPT_POST=>true, CURLOPT_POSTFIELDS=>$fields, CURLOPT_HTTPHEADER=>['X-API-Key: '.$apiKey], CURLOPT_RETURNTRANSFER=>true, CURLOPT_TIMEOUT=>600]);
    $body = curl_exec($ch);
    $code = (int)curl_getinfo($ch, CURLINFO_HTTP_CODE);
    $err = curl_error($ch);
    unset($ch);
    if ($err !== '') return "Connection failed: {$err}";
    if ($code === 0) return 'Could not connect to CMS (timeout or DNS failure)';
    if ($code === 413) return "File too large for CMS ({$fileSize} bytes) — increase nginx client_max_body_size and PHP upload_max_filesize";
    if ($code === 401) return 'Invalid API key (HTTP 401)';
    if ($code === 403) return 'Access denied (HTTP 403) — CSRF or auth error';
    if ($code >= 500) return "CMS server error (HTTP {$code}): " . substr($body ?? '', 0, 200);
    if ($code < 200 || $code >= 300) return "Unexpected HTTP {$code}: " . substr($body ?? '', 0, 200);
    $json = json_decode($body, true);
    if (isset($json['error'])) return 'CMS rejected: ' . $json['error'];
    return '';
}

function notifyCallback(string $url, string $apiKey, int $jobId, string $status, string $error='', array $extra=[]): void {
    if ($url === '') return;
    $payload = array_merge(['job_id' => $jobId, 'status' => $status, 'error' => $error], $extra);
    for ($attempt = 1; $attempt <= 3; $attempt++) {
        $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: ' . $apiKey], CURLOPT_RETURNTRANSFER => true, CURLOPT_TIMEOUT => 10]);
        $body = curl_exec($ch);
        $code = (int)curl_getinfo($ch, CURLINFO_HTTP_CODE);
        unset($ch);
        if ($code >= 200 && $code < 300) return;
        echo "[".date('H:i:s')."]   Callback {$attempt}/3 failed: HTTP {$code}" . ($body ? " — " . substr($body, 0, 100) : "") . "\n";
        if ($attempt < 3) sleep(3);
    }
}

function cleanupAll(string $src, ?string $wm, array $outFiles, string $workDir = '', string $folder = ''): void {
    @unlink($src); if ($wm) @unlink($wm);
    foreach ($outFiles as $of) { if (isset($of['local']) && file_exists($of['local'])) @unlink($of['local']); }
    // Note: cached source ({folder}_cached.mp4) is NOT deleted here.
    // It stays on disk so the next job for this video can reuse it.
    // Old caches are cleaned at worker startup (files > 1 hour old).
}
WORKER;
    }
}
