ltbxd-actorle/src/MessageHandler/ProcessImportMessageHandler.php

<?php

declare(strict_types=1);

namespace App\MessageHandler;

use App\Entity\Import;
use App\Gateway\LtbxdGateway;
use App\Message\ImportFilmsBatchMessage;
use App\Message\ProcessImportMessage;
use Doctrine\ORM\EntityManagerInterface;
use League\Flysystem\FilesystemOperator;
use Psr\Log\LoggerInterface;
use Symfony\Component\Messenger\Attribute\AsMessageHandler;
use Symfony\Component\Messenger\MessageBusInterface;

#[AsMessageHandler]
readonly class ProcessImportMessageHandler
{
    private const int BATCH_SIZE = 50;

    public function __construct(
        private EntityManagerInterface $em,
        private FilesystemOperator $defaultStorage,
        private LtbxdGateway $ltbxdGateway,
        private MessageBusInterface $bus,
        private LoggerInterface $logger,
    ) {}

    public function __invoke(ProcessImportMessage $message): void
    {
        $import = $this->em->getRepository(Import::class)->find($message->importId);
        if (!$import) {
            $this->logger->error('Import not found', ['importId' => $message->importId]);

            return;
        }

        try {
            $csvContent = $this->defaultStorage->read($import->getFilePath());

            $tmpFile = tempnam(sys_get_temp_dir(), 'import_');
            file_put_contents($tmpFile, $csvContent);

            try {
                $ltbxdMovies = $this->ltbxdGateway->parseFileFromPath($tmpFile);
            } finally {
                unlink($tmpFile);
            }

            $totalFilms = count($ltbxdMovies);

            $import->setTotalFilms($totalFilms);
            $import->setStatus(Import::STATUS_PROCESSING);
            $this->em->flush();

            $batches = array_chunk($ltbxdMovies, self::BATCH_SIZE);
            foreach ($batches as $batch) {
                $films = array_map(fn ($movie) => [
                    'name' => $movie->getName(),
                    'year' => $movie->getYear(),
                    'ltbxdUri' => $movie->getLtbxdUri(),
                    'date' => $movie->getDate()->format('Y-m-d'),
                ], $batch);

                $this->bus->dispatch(new ImportFilmsBatchMessage(
                    importId: $import->getId(),
                    films: $films,
                ));
            }
        } catch (\Throwable $e) {
            $this->logger->error('Import processing failed', [
                'importId' => $import->getId(),
                'error' => $e->getMessage(),
            ]);

            $import->setStatus(Import::STATUS_FAILED);
            $this->em->flush();
        }
    }
}