ltbxd-actorle/src/MessageHandler/ImportFilmsBatchMessageHandler.php

<?php

declare(strict_types=1);

namespace App\MessageHandler;

use App\Entity\Import;
use App\Entity\UserMovie;
use App\Gateway\LtbxdGateway;
use App\Message\ImportFilmsBatchMessage;
use App\Repository\ImportRepository;
use App\Service\ActorSyncer;
use App\Service\AwardImporter;
use App\Service\FilmImporter;
use Doctrine\ORM\EntityManagerInterface;
use League\Flysystem\FilesystemOperator;
use Psr\Log\LoggerInterface;
use Symfony\Component\Messenger\Attribute\AsMessageHandler;

#[AsMessageHandler]
readonly class ImportFilmsBatchMessageHandler
{
    public function __construct(
        private EntityManagerInterface $em,
        private FilesystemOperator $defaultStorage,
        private LtbxdGateway $ltbxdGateway,
        private FilmImporter $filmImporter,
        private ActorSyncer $actorSyncer,
        private ImportRepository $importRepository,
        private AwardImporter $awardImporter,
        private LoggerInterface $logger,
    ) {}

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

            return;
        }

        $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);
        }

        $batch = array_slice($ltbxdMovies, $message->offset, $message->limit);
        $userId = $import->getUser()->getId();
        $importId = $import->getId();

        foreach ($batch as $ltbxdMovie) {
            try {
                $movie = $this->filmImporter->importFromLtbxdMovie($ltbxdMovie);
                if (!$movie) {
                    $this->importRepository->incrementFailedFilms($import);
                    continue;
                }

                $this->actorSyncer->syncActorsForMovie($movie);

                // Import awards for actors of this movie
                foreach ($movie->getActors() as $role) {
                    $this->awardImporter->importForActor($role->getActor());
                }

                $user = $this->em->getReference(\App\Entity\User::class, $userId);
                $existingLink = $this->em->getRepository(UserMovie::class)->findOneBy([
                    'user' => $user,
                    'movie' => $movie,
                ]);
                if (!$existingLink) {
                    $userMovie = new UserMovie();
                    $userMovie->setUser($user);
                    $userMovie->setMovie($movie);
                    $this->em->persist($userMovie);
                }

                $this->em->flush();
            } catch (\Throwable $e) {
                $this->logger->warning('Failed to import film', [
                    'film' => $ltbxdMovie->getName(),
                    'importId' => $importId,
                    'error' => $e->getMessage(),
                ]);
                $this->importRepository->incrementFailedFilms($import);
            }

            $this->em->clear();
            $import = $this->em->getRepository(Import::class)->find($importId);
        }

        $processedBatches = $this->importRepository->incrementProcessedBatches($import);

        if ($processedBatches >= $import->getTotalBatches()) {
            // Refresh the entity to get updated failedFilms from DB
            $this->em->refresh($import);

            $import->setStatus(Import::STATUS_COMPLETED);
            $import->setCompletedAt(new \DateTimeImmutable());
            $this->em->flush();
        }
    }
}