feuerwerk/api/lib/VideoWorker.php

<?php
declare(strict_types=1);

/**
 * Converts what waits in the VideoQueue, one video at a time (#402). Runs in the container
 * `feuerwerk-video` through `tools/video-worker.php`, and in CI beside the PHP server.
 *
 * Each job goes: probe the upload, convert it into a temp file beside its target, read the result
 * back for its details, cut a poster, then move all three into place and drop the job. Whatever
 * fails on the way fails the job with ffmpeg's own words, and the temp files go.
 */
final class VideoWorker
{
    /** How often a running conversion writes its progress, in seconds. */
    private const PROGRESS_EVERY = 0.5;

    public function __construct(private readonly VideoStore $store, private readonly string $videos)
    {
    }

    /** Converts the oldest waiting video; false when there was none. */
    public function runOnce(): bool
    {
        $job = $this->store->queue()->next();
        if ($job === null) {
            return false;
        }
        $this->convert((string) $job['folder'], (string) $job['hash']);
        return true;
    }

    private function convert(string $folder, string $hash): void
    {
        $queue = $this->store->queue();
        if (!$queue->update($folder, $hash, ['state' => VideoQueue::CONVERTING, 'progress' => 0, 'error' => ''])) {
            return;
        }
        $dir = "{$this->videos}/{$folder}";
        $tag = bin2hex(random_bytes(4));
        $tmpVideo = "{$dir}/{$hash}.{$tag}.tmp.mp4";
        $tmpPoster = "{$dir}/{$hash}.{$tag}.tmp.jpg";
        try {
            $plan = VideoFormat::plan(self::probe($queue->source($folder, $hash)));
            $finished = $this->run(
                VideoFormat::convertArgs($queue->source($folder, $hash), $tmpVideo, $plan),
                $plan['duration'],
                static fn(float $p): bool => $queue->update($folder, $hash, ['progress' => round($p, 3)]),
            );
            if (!$finished) {
                return;
            }
            $details = VideoFormat::details(self::probe($tmpVideo));
            $this->run(VideoFormat::posterArgs($tmpVideo, $tmpPoster, $details['duration']), 0, null);
            if ($queue->job($folder, $hash) === null) {
                return;
            }
            chmod($tmpVideo, AtomicFile::SERVED);
            chmod($tmpPoster, AtomicFile::SERVED);
            VideoStore::writeDetails($dir, $hash, $details);
            rename($tmpPoster, "{$dir}/{$hash}.jpg");
            rename($tmpVideo, "{$dir}/{$hash}.mp4");
            $queue->remove($folder, $hash);
        } catch (Throwable $e) {
            $queue->dropSource($folder, $hash);
            $queue->update($folder, $hash, ['state' => VideoQueue::FAILED, 'error' => self::shorten($e->getMessage())]);
        } finally {
            foreach ([$tmpVideo, $tmpPoster] as $path) {
                if (is_file($path)) {
                    unlink($path);
                }
            }
        }
    }

    /** @return array<string,mixed> */
    private static function probe(string $path): array
    {
        [$code, $out, $err] = self::capture(VideoFormat::probeArgs($path));
        $probe = json_decode($out, true);
        if ($code !== 0 || !is_array($probe)) {
            throw new RuntimeException('ffmpeg cannot read this file' . ($err === '' ? '' : ": {$err}"));
        }
        return $probe;
    }

    /**
     * Runs ffmpeg, reporting progress from its `-progress pipe:1` lines to `$report`. Returns false
     * when `$report` says the job is gone, after stopping ffmpeg.
     *
     * @param list<string> $args
     * @param (callable(float):bool)|null $report
     */
    private function run(array $args, float $duration, ?callable $report): bool
    {
        $err = tempnam(sys_get_temp_dir(), 'video-err-');
        $process = proc_open($args, [0 => ['file', '/dev/null', 'r'], 1 => ['pipe', 'w'], 2 => ['file', $err, 'w']], $pipes);
        if (!is_resource($process)) {
            throw new RuntimeException('Cannot start ffmpeg');
        }
        $last = 0.0;
        $alive = true;
        while (($line = fgets($pipes[1])) !== false) {
            $share = $report === null ? null : VideoFormat::progress($line, $duration);
            if ($share !== null && microtime(true) - $last >= self::PROGRESS_EVERY) {
                $last = microtime(true);
                if (!$report($share)) {
                    $alive = false;
                    proc_terminate($process);
                    break;
                }
            }
        }
        fclose($pipes[1]);
        $code = proc_close($process);
        $message = trim((string) file_get_contents($err));
        unlink($err);
        if (!$alive) {
            return false;
        }
        if ($code !== 0) {
            throw new RuntimeException($message === '' ? "ffmpeg stopped with code {$code}" : $message);
        }
        return true;
    }

    /**
     * @param list<string> $args
     * @return array{0:int,1:string,2:string}
     */
    private static function capture(array $args): array
    {
        $process = proc_open($args, [0 => ['file', '/dev/null', 'r'], 1 => ['pipe', 'w'], 2 => ['pipe', 'w']], $pipes);
        if (!is_resource($process)) {
            throw new RuntimeException('Cannot start ffprobe');
        }
        $out = (string) stream_get_contents($pipes[1]);
        $err = trim((string) stream_get_contents($pipes[2]));
        fclose($pipes[1]);
        fclose($pipes[2]);
        return [proc_close($process), $out, $err];
    }

    /** The last lines of what ffmpeg said: enough to tell why, not a log. */
    private static function shorten(string $message): string
    {
        $lines = array_values(array_filter(array_map('trim', explode("\n", $message)), static fn(string $l): bool => $l !== ''));
        $text = implode(' · ', array_slice($lines, -2));
        return mb_strlen($text) > 300 ? mb_substr($text, 0, 300) . '…' : $text;
    }
}