feuerwerk/api/lib/VideoQueue.php

<?php
declare(strict_types=1);

/**
 * The videos waiting to be converted (#402): `data/videos/queue/{folder}-{hash}.json` says what
 * became of an upload, and its `source` where the upload lies – kept as it came beside the MP4,
 * `{folder}/{hash}.source.{ext}`, for downloading it. The API adds to it, the worker in the
 * container `feuerwerk-video` takes from it, one at a time, oldest first.
 *
 * A job is `queued`, `converting` (with `progress` from 0 to 1) or `failed` (with `error`). A done
 * one is no job any more: its MP4 lies in the account's folder and the job is gone. A failed one
 * stays, so the picker can say why, until the file is deleted or uploaded again.
 *
 * A job from before the uploads were kept has no `source`: its bytes wait hidden in the queue as
 * `{folder}-{hash}.src` and go once they are converted.
 */
final class VideoQueue
{
    public const QUEUED = 'queued';
    public const CONVERTING = 'converting';
    public const FAILED = 'failed';

    public function __construct(private readonly string $dir)
    {
        if (!is_dir($this->dir) && !mkdir($this->dir, 0775, true) && !is_dir($this->dir)) {
            throw new RuntimeException("Cannot create directory {$this->dir}");
        }
    }

    /**
     * Copies an upload into a temp file beside the jobs, never more than `$maxBytes` of it: a video
     * is too big to be held in memory. The caller keeps the file or deletes it.
     *
     * @param resource $in
     * @throws VideoTooLargeException when the stream has more than `$maxBytes`
     */
    public function receive($in, int $maxBytes): string
    {
        $tmp = "{$this->dir}/upload-" . bin2hex(random_bytes(8)) . '.tmp';
        $out = fopen($tmp, 'xb');
        if ($out === false) {
            throw new RuntimeException('Cannot write uploaded video');
        }
        $copied = stream_copy_to_stream($in, $out, $maxBytes + 1);
        fclose($out);
        if ($copied === false || $copied > $maxBytes) {
            unlink($tmp);
            throw $copied === false ? new RuntimeException('Cannot write uploaded video') : new VideoTooLargeException();
        }
        return $tmp;
    }

    /**
     * Queues the upload kept at `$source`, a path inside the videos' directory such as
     * `{folder}/{hash}.source.mkv`. The upload stays where it is.
     */
    public function add(string $folder, string $hash, string $source, string $name): void
    {
        $this->write($folder, $hash, [
            'folder' => $folder,
            'hash' => $hash,
            'source' => $source,
            'name' => $name,
            'state' => self::QUEUED,
            'progress' => 0,
            'error' => '',
            'createdAt' => gmdate('c'),
        ]);
    }

    /** @return array<string,mixed>|null */
    public function job(string $folder, string $hash): ?array
    {
        $path = $this->jobPath($folder, $hash);
        if (!is_file($path)) {
            return null;
        }
        $raw = @file_get_contents($path);
        $job = $raw === false ? null : json_decode($raw, true);
        return is_array($job) ? $job : null;
    }

    /** @return list<array<string,mixed>> every job, oldest first */
    public function jobs(): array
    {
        $jobs = [];
        foreach (glob($this->dir . '/*.json') ?: [] as $path) {
            if (!preg_match('/^(' . UserStore::FOLDER_PATTERN . ')-([a-f0-9]{64})\.json$/', basename($path), $m)) {
                continue;
            }
            $job = $this->job($m[1], $m[2]);
            if ($job !== null) {
                $jobs[] = $job;
            }
        }
        usort($jobs, static fn(array $a, array $b): int => strcmp((string) $a['createdAt'], (string) $b['createdAt']));
        return $jobs;
    }

    /** The oldest job nobody has started on, or null when there is none. */
    public function next(): ?array
    {
        foreach ($this->jobs() as $job) {
            if ($job['state'] === self::QUEUED && is_file($this->source($job['folder'], $job['hash']))) {
                return $job;
            }
        }
        return null;
    }

    /**
     * Changes a job, unless it was deleted in the meantime: then false, and the one converting it
     * knows to stop.
     *
     * @param array<string,mixed> $changes
     */
    public function update(string $folder, string $hash, array $changes): bool
    {
        $job = $this->job($folder, $hash);
        if ($job === null) {
            return false;
        }
        $this->write($folder, $hash, $changes + $job);
        return true;
    }

    /** A job that was being converted when the worker stopped is started again. */
    public function requeueInterrupted(): int
    {
        $count = 0;
        foreach ($this->jobs() as $job) {
            if ($job['state'] === self::CONVERTING) {
                $this->update($job['folder'], $job['hash'], ['state' => self::QUEUED, 'progress' => 0]);
                $count++;
            }
        }
        return $count;
    }

    /** Drops the job, and an old one's hidden upload; a kept upload is the store's to delete. */
    public function remove(string $folder, string $hash): bool
    {
        $removed = false;
        foreach ([$this->hiddenUpload($folder, $hash), $this->jobPath($folder, $hash)] as $path) {
            if (is_file($path)) {
                $removed = unlink($path) || $removed;
            }
        }
        return $removed;
    }

    /**
     * Drops an old job's hidden upload once it cannot be converted; the job itself says why. A kept
     * upload stays: it can still be downloaded as it came.
     */
    public function dropSource(string $folder, string $hash): void
    {
        $path = $this->hiddenUpload($folder, $hash);
        if (is_file($path)) {
            unlink($path);
        }
    }

    /** Where the bytes of a job lie, for ffmpeg to read. */
    public function source(string $folder, string $hash): string
    {
        $source = (string) ($this->job($folder, $hash)['source'] ?? '');
        $kept = '#^' . UserStore::FOLDER_PATTERN . '/[a-f0-9]{64}\.source\.[a-z0-9]+$#';
        return preg_match($kept, $source) ? dirname($this->dir) . "/{$source}" : $this->hiddenUpload($folder, $hash);
    }

    private function hiddenUpload(string $folder, string $hash): string
    {
        return "{$this->dir}/{$folder}-{$hash}.src";
    }

    private function jobPath(string $folder, string $hash): string
    {
        return "{$this->dir}/{$folder}-{$hash}.json";
    }

    /** @param array<string,mixed> $job */
    private function write(string $folder, string $hash, array $job): void
    {
        $json = json_encode($job, JSON_UNESCAPED_UNICODE | JSON_UNESCAPED_SLASHES);
        AtomicFile::write($this->jobPath($folder, $hash), (string) $json, AtomicFile::HIDDEN, 'video job');
    }
}