Web app -> API -> Job queue -> Worker -> FFmpeg/etc -> temp storage -> download * * The web tier never processes media inline: controllers only enqueue and * poll. Workers call Pipeline::process() from the CLI. */ final class Pipeline { /** @var array */ private array $drivers; public function __construct() { $this->drivers = [ 'ffmpeg' => new Ffmpeg(), 'image' => new Images(), 'pdf' => new Pdf(), 'youtube' => new Youtube(), 'qr' => new Qr(), ]; } /** * Enqueue a job after full validation. Returns the job id. * * @param array $tool registry row * @param array $params structured, allow-listed parameters */ public function enqueue(array $tool, array $params = [], string $source = 'web'): string { $site = config('Site'); $jobs = model(JobModel::class); $session = service('analytics')->sessionHash(); $ipHash = service('analytics')->ipHash(); // ---- abuse guards ------------------------------------------------- if ($jobs->countQueued() >= $site->queueMaxLength) { throw new \RuntimeException('The processing queue is currently full. Please try again shortly.'); } $hourAgo = gmdate('Y-m-d H:i:s', time() - 3600); $perSessionIp = $jobs->db->table('jobs') ->where('created_at >=', $hourAgo) ->whereIn('status', ['queued', 'processing', 'completed']) ->groupStart()->where('session_id', $session)->orWhere('ip_hash', $ipHash)->groupEnd() ->countAllResults(); $limit = min($site->maxJobsPerHourSession + ($source === 'api' ? 60 : 0), max($site->maxJobsPerHourIp, $site->maxJobsPerHourSession)); if ($perSessionIp >= $site->maxJobsPerHourIp && $source === 'web') { throw new \RuntimeException('Hourly limit reached. Please wait before starting more jobs.'); } $concurrent = $jobs->where('session_id', $session)->whereIn('status', ['queued', 'processing'])->countAllResults(); if ($concurrent >= $site->maxConcurrentSession) { throw new \RuntimeException('You already have the maximum number of jobs running. Please wait for them to finish.'); } // ---- capability check ---------------------------------------------- foreach (($tool['requires_binaries'] ?? []) as $bin) { if (! in_array($bin, ['yt-dlp'], true)) { // yt-dlp handled by its driver try { Process::binary((string) $bin, ['-version'], 10); } catch (\RuntimeException) { throw new \RuntimeException("This tool is temporarily unavailable on this server."); } } } if (in_array('yt-dlp', $tool['requires_binaries'] ?? [], true) && ! $this->youtubeEnabled()) { throw new \RuntimeException('This tool is disabled on this server.'); } return $jobs->create([ 'session_id' => $session, 'ip_hash' => $ipHash, 'tool_id' => $tool['id'], 'operation' => $params['_operation'] ?? ($tool['operation'] ?: 'convert'), 'params' => json_encode($params, JSON_UNESCAPED_UNICODE), 'input_file' => $params['_input_file'] ?? null, 'input_name' => $params['_input_name'] ?? null, 'input_size' => (int) ($params['_input_size'] ?? 0), 'input_mime' => $params['_input_mime'] ?? null, 'status' => 'queued', 'priority' => $source === 'api' ? 3 : 5, 'source' => $source, ]); } public function youtubeEnabled(): bool { if (! config('Site')->ytDlpEnabled) { return false; } $path = config('Site')->binaries['yt-dlp']; if (! is_executable($path)) { return false; } // feature flag may also be flipped at runtime from admin settings $setting = model(\App\Models\SettingModel::class)->get('ytdlp_enabled'); if ($setting !== null) { return $setting === '1'; } return true; } /** Executed ONLY by workers. Runs the driver for a job row. */ public function process(array $job): void { $jobs = model(JobModel::class); $tool = []; if ($job['tool_id'] !== null) { $toolRow = $jobs->db->table('tools')->where('id', $job['tool_id'])->get()->getFirstRow('array'); if ($toolRow !== null) { Finder::decodeTool($toolRow); $tool = $toolRow; } } $params = json_decode($job['params'] ?? '{}', true) ?: []; $driver = match (true) { str_starts_with($job['operation'], 'yt_') => $this->drivers['youtube'], $job['operation'] === 'qr_generate' => $this->drivers['qr'], isset($tool['requires_binaries']) && in_array('qpdf', $tool['requires_binaries'] ?? [], true) => $this->drivers['pdf'], isset($tool['requires_binaries']) && in_array('pdftoppm', $tool['requires_binaries'] ?? [], true) => $this->drivers['pdf'], isset($tool['requires_binaries']) && in_array('ffmpeg', $tool['requires_binaries'] ?? [], true) => $this->drivers['ffmpeg'], default => $this->pickByOperation($job['operation'], $tool), }; try { $result = $driver->handle($job, $tool, function (int $progress, string $stage) use ($jobs, $job): void { $jobs->update($job['id'], ['progress' => min(99, max(1, $progress)), 'stage' => $stage]); }); $jobs->update($job['id'], [ 'status' => 'completed', 'progress' => 100, 'stage' => 'complete', 'output_file' => $result['file'], 'output_name' => $result['name'], 'output_size' => filesize(WRITEPATH . 'media/' . $result['file']) ?: 0, 'completed_at' => gmdate('Y-m-d H:i:s'), ]); service('analytics')->track('tool_success', ['tool' => $tool['slug'] ?? '', 'format' => $result['ext'] ?? '']); } catch (\Throwable $e) { log_message('error', 'job {id} failed: {m}', ['id' => $job['id'], 'm' => $e->getMessage()]); $jobs->update($job['id'], [ 'status' => 'failed', 'stage' => 'failed', 'error' => mb_substr($e->getMessage(), 0, 500), 'completed_at' => gmdate('Y-m-d H:i:s'), ]); service('analytics')->track('tool_failure', ['tool' => $tool['slug'] ?? '']); } } private function pickByOperation(string $operation, array $tool = []): DriverInterface { // explicit PDF operations $pdfOps = ['pdf_merge', 'pdf_split', 'pdf_extract_pages', 'pdf_compress', 'pdf_remove_metadata', 'pdf_to_jpg', 'pdf_to_png', 'pdf_to_text', 'images_to_pdf']; if (in_array($operation, $pdfOps, true)) { return $this->drivers['pdf']; } // route by format families of the registered tool $formats = array_filter([ $tool['primary_input_format'] ?? null, $tool['primary_output_format'] ?? null, ]); $families = array_map(static fn ($f) => FormatCatalog::family((string) $f), $formats); if ($formats !== []) { if (in_array('document', $families, true)) { return $this->drivers['pdf']; } if (! in_array('video', $families, true) && ! in_array('audio', $families, true)) { return $this->drivers['image']; } } return $this->drivers['ffmpeg']; } /** Mark a job failed with a safe message (used by worker + reaper). */ public static function failJob(string $jobId, string $reason): void { model(\App\Models\JobModel::class)->update($jobId, [ 'status' => 'failed', 'stage' => 'failed', 'error' => mb_substr($reason, 0, 500), 'completed_at' => gmdate('Y-m-d H:i:s'), ]); } /** Temp storage root (never web-accessible directly). */ public static function storageDir(): string { $dir = WRITEPATH . 'media'; is_dir($dir) || mkdir($dir, 0750, true); return $dir; } public static function incomingDir(): string { $dir = WRITEPATH . 'incoming'; is_dir($dir) || mkdir($dir, 0750, true); return $dir; } /** Safe filename inside storage: uuid + extension only. */ public static function safeName(string $uuid, string $ext): string { return preg_match('/^[a-f0-9-]{36}$/', $uuid) === 1 ? $uuid . '.' . preg_replace('/[^a-z0-9]/i', '', strtolower($ext)) : throw new \InvalidArgumentException('bad name'); } }