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; } $converted = self::probe($tmpVideo); $details = VideoFormat::details($converted); $this->run(VideoFormat::posterArgs($tmpVideo, $tmpPoster, $details['duration'], VideoFormat::colour($converted)), 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 */ 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 $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 $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; } }