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 hands the file to add() 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 received into `$upload`, which is moved and gone afterwards. */ public function add(string $folder, string $hash, string $upload, string $name): void { if (!chmod($upload, AtomicFile::HIDDEN) || !rename($upload, $this->source($folder, $hash))) { throw new RuntimeException('Cannot write uploaded video'); } $this->write($folder, $hash, [ 'folder' => $folder, 'hash' => $hash, 'name' => $name, 'state' => self::QUEUED, 'progress' => 0, 'error' => '', 'createdAt' => gmdate('c'), ]); } /** @return array|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> 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 $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; } public function remove(string $folder, string $hash): bool { $removed = false; foreach ([$this->source($folder, $hash), $this->jobPath($folder, $hash)] as $path) { if (is_file($path)) { $removed = unlink($path) || $removed; } } return $removed; } /** Drops the upload once it is converted or cannot be; the job itself says what happened. */ public function dropSource(string $folder, string $hash): void { $path = $this->source($folder, $hash); if (is_file($path)) { unlink($path); } } public function source(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 $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'); } }