MeshVault/src/Indexer/ImportWorker.php

<?php

declare(strict_types=1);

namespace MeshVault\Indexer;

use MeshVault\Exception\PathSafetyException;
use MeshVault\Indexer\Filesystem\PathSafety;
use MeshVault\Indexer\SpaceChecker;
use MeshVault\Catalog\CatalogServiceInterface;
use MeshVault\Repository\ImportJobRepositoryInterface;
use Psr\Log\LoggerInterface;
use Throwable;
use DomainException;

/**
 * Serializes every managed-library mutation behind one advisory lock file and
 * drives a claimed job item by item.
 *
 * Durability contract: the planned managed path plus the source digest are
 * committed BEFORE any byte is published. After a crash the recovered worker
 * re-reads that plan, verifies the expected destination, and completes the
 * catalog instead of copying again, so restarts never orphan or duplicate
 * files. Items fail independently; one failure never abandons the remaining
 * work, and the job reports Failed only once every item is terminal.
 * A graceful-stop callback is consulted before each item and leaves the job
 * Pending (resumable) rather than terminal.
 */
final readonly class ImportWorker
{
    public const LOCK_FILE = '.meshvault-import.lock';

    /** @param null|callable(): bool $shouldStop Consulted before each item. */
    public function __construct(
        private ImportJobRepositoryInterface $jobs,
        private CatalogServiceInterface $catalog,
        private StreamingHasher $hasher,
        private VerifiedCopier $copier,
        private StlValidator $validator,
        private PathSafety $pathSafety,
        private SpaceChecker $spaceChecker,
        private string $libraryRoot,
        private LoggerInterface $logger,
        private mixed $shouldStop = null,
    ) {
    }

    /**
     * Claim and drain at most one job. Returns the job id, or null when idle.
     *
     * @throws ImportException LOCK_UNAVAILABLE when another worker holds the library.
     */
    public function runOnce(): ?int
    {
        $lock = $this->lock();
        try {
            // Recovery runs under the lock, so no live worker can be reset.
            $this->jobs->resumeInterrupted();
            $job = $this->jobs->claimNext();
            if ($job === null) {
                return null;
            }
            $jobId = (int) $job['id'];
            $this->drain($jobId);
            return $jobId;
        } finally {
            flock($lock, LOCK_UN);
            fclose($lock);
        }
    }

    /** @return resource */
    private function lock()
    {
        $path = $this->pathSafety->ensureDestSafe($this->libraryRoot . DIRECTORY_SEPARATOR . self::LOCK_FILE);
        $handle = @fopen($path, 'c');
        if ($handle === false) {
            throw new ImportException('LOCK_UNAVAILABLE', 'Cannot open the library lock file.');
        }
        if (!flock($handle, LOCK_EX | LOCK_NB)) {
            fclose($handle);
            throw new ImportException('LOCK_UNAVAILABLE', 'Another import worker holds the library lock.');
        }
        return $handle;
    }

    private function drain(int $jobId): void
    {
        foreach ($this->jobs->items($jobId) as $item) {
            if (!in_array($item['status'], ['Pending', 'Copying'], true)) {
                continue;
            }
            if ($this->shouldStop !== null && ($this->shouldStop)()) {
                // Graceful stop: return the job and its unfinished items to Pending.
                $this->jobs->release($jobId);
                $this->logger->info('Import worker stopped gracefully.', ['job_id' => $jobId]);
                return;
            }
            $this->process($jobId, $item);
        }
    }

    /** @param array<string, mixed> $item */
    private function process(int $jobId, array $item): void
    {
        $itemId = (int) $item['id'];
        try {
            $plan = $this->plan($item);
            $destination = $this->pathSafety->ensureDestSafe(
                $this->libraryRoot . DIRECTORY_SEPARATOR
                . str_replace('/', DIRECTORY_SEPARATOR, $plan['managed_path'])
            );
            $published = $this->publish($plan, $destination);
            $relative = $this->relative($published);
            // Catalog and progress commit together, so a crash here simply replays.
            $this->catalog->atomic(function () use ($item, $itemId, $plan, $relative): void {
                // Catalog locks precede item/job locks on the shared PDO connection.
                try {
                    $this->catalog->registerFile(
                        $item['model_id'] === null ? null : (int) $item['model_id'],
                        $relative,
                        $plan['hash'],
                        $plan['size'],
                        null,
                        'Present'
                    );
                } catch (DomainException $error) {
                    throw new ImportException('CATALOG_CONFLICT', $error->getMessage(), $error);
                }
                $this->jobs->updateManagedPath($itemId, $relative);
                $this->jobs->finishItem($itemId, 'Completed');
            });
        } catch (Throwable $error) {
            $this->failItem($jobId, $itemId, $error);
        }
    }

    /**
     * Persist the immutable plan (destination plus digest) before publication.
     *
     * @param array<string, mixed> $item
     * @return array{source: string, managed_path: string, hash: string, size: int}
     */
    private function plan(array $item): array
    {
        $source = (string) $item['source_path'];
        if (strtolower(pathinfo($source, PATHINFO_EXTENSION)) === 'stl') {
            $this->validator->validate($source);
        }
        if ($item['managed_path'] !== null && $item['hash'] !== null) {
            // Recovered plan: trust the committed intent, never choose a new name.
            return [
                'source' => $source,
                'managed_path' => (string) $item['managed_path'],
                'hash' => (string) $item['hash'],
                'size' => (int) $item['size'],
            ];
        }
        $hash = $this->hasher->hash($source);
        $size = filesize($source);
        if ($size === false) {
            throw new ImportException('SCAN_READ_ERROR', 'Cannot determine the source size.');
        }
        $managed = $this->managedPath($item, $source);
        $this->jobs->planItem((int) $item['id'], $managed, $hash, $size);
        return ['source' => $source, 'managed_path' => $managed, 'hash' => $hash, 'size' => $size];
    }

    /** @param array<string, mixed> $item */
    private function managedPath(array $item, string $source): string
    {
        $folder = 'unfiled';
        if ($item['model_id'] !== null) {
            $aggregate = $this->catalog->aggregate((int) $item['model_id']);
            if ($aggregate !== null) {
                $folder = trim((string) $aggregate['folder']['path'], '/');
            }
        }
        return ($folder === '' ? '' : $folder . '/') . basename(str_replace('\\', '/', $source));
    }

    /**
     * Publish, or adopt an already-published destination left by a crashed run.
     *
     * @param array{source: string, managed_path: string, hash: string, size: int} $plan
     * @return string The actual published path (may have collision suffix)
     */
    private function publish(array $plan, string $destination): string
    {
        $directory = dirname($destination);
        if (!is_dir($directory) && !@mkdir($directory, 0775, true) && !is_dir($directory)) {
            throw new ImportException('COPY_FAILED', 'Cannot create the managed destination directory.');
        }
        // Re-validate the destination with PathSafety and reject links before adopting.
        $destination = $this->pathSafety->ensureDestSafe($destination);
        clearstatcache(true, $destination);
        if (is_file($destination)) {
            // Idempotent restart: the planned destination already holds our bytes.
            if (hash_file('sha256', $destination) === $plan['hash']) {
                return $destination;
            }
            // Different content at the planned destination: let VerifiedCopier resolve the collision
            // by appending a _001 suffix. This keeps restart idempotent (we don't re-copy the same
            // file) while still allowing two same-basename sources with different content to coexist.
        } else {
            // New file being copied: check free space before copying
            $this->spaceChecker->check($directory, $plan['size']);
        }
        return $this->copier->copy($plan['source'], $destination, $plan['size'], $plan['hash']);
    }

    private function relative(string $published): string
    {
        // Use the canonical (realpath-resolved) root from PathSafety so prefix comparison
        // is reliable even when the configured library root contains symlinks or case differences.
        $prefix = rtrim(str_replace('\\', '/', $this->pathSafety->libraryRoot()), '/') . '/';
        $path = str_replace('\\', '/', $published);
        if (!str_starts_with($path, $prefix)) {
            throw new ImportException('PATH_TRAVERSAL', 'The published file escaped the managed library.');
        }
        return substr($path, strlen($prefix));
    }

    private function failItem(int $jobId, int $itemId, Throwable $error): void
    {
        $reason = match (true) {
            $error instanceof ImportException => $error->reason,
            $error instanceof PathSafetyException => $error->reason,
            default => 'COPY_FAILED',
        };
        $this->logger->error('Import item failed.', [
            'job_id' => $jobId,
            'item_id' => $itemId,
            'reason' => $reason,
        ]);
        // Failures are per item: the loop continues with the remaining work.
        $this->jobs->finishItem($itemId, 'Failed', $reason . ': ' . $error->getMessage());
    }
}