feat: add ImportFilmsBatchMessageHandler

src/MessageHandler/ImportFilmsBatchMessageHandler.php

@@ -0,0 +1,110 @@
+<?php
+
+declare(strict_types=1);
+
+namespace App\MessageHandler;
+
+use App\Entity\Import;
+use App\Entity\Notification;
+use App\Entity\UserMovie;
+use App\Exception\GatewayException;
+use App\Gateway\LtbxdGateway;
+use App\Message\ImportFilmsBatchMessage;
+use App\Repository\ImportRepository;
+use App\Service\ActorSyncer;
+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 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);
+        $user = $import->getUser();
+
+        foreach ($batch as $ltbxdMovie) {
+            try {
+                $movie = $this->filmImporter->importFromLtbxdMovie($ltbxdMovie);
+                if (!$movie) {
+                    $this->importRepository->incrementFailedFilms($import);
+                    continue;
+                }
+
+                $this->actorSyncer->syncActorsForMovie($movie);
+
+                $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' => $import->getId(),
+                    'error' => $e->getMessage(),
+                ]);
+                $this->importRepository->incrementFailedFilms($import);
+            }
+        }
+
+        $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();
+
+            $imported = $import->getTotalFilms() - $import->getFailedFilms();
+            $notification = new Notification();
+            $notification->setUser($user);
+            $notification->setMessage(sprintf(
+                'Import terminé : %d/%d films importés.',
+                $imported,
+                $import->getTotalFilms()
+            ));
+            $this->em->persist($notification);
+            $this->em->flush();
+        }
+    }
+}