toolvana/app/Libraries/Pipeline.php

<?php

declare(strict_types=1);

namespace App\Libraries;

use App\Libraries\Pipeline\DriverInterface;
use App\Libraries\Pipeline\Ffmpeg;
use App\Libraries\Pipeline\Images;
use App\Libraries\Pipeline\Pdf;
use App\Libraries\Pipeline\Qr;
use App\Libraries\Pipeline\Youtube;
use App\Models\JobModel;

/**
 * Media pipeline facade.
 *
 *   Browser -> 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<string, DriverInterface> */
    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');
    }
}