feat: add ProcessImportMessageHandler

src/Gateway/LtbxdGateway.php

@@ -20,13 +20,13 @@ readonly class LtbxdGateway
      * @return LtbxdMovie[]
      * @throws GatewayException
      */
-    public function parseFile(): array
+    public function parseFileFromPath(string $path): array
     {
-        if (!file_exists($this->fileDir)) {
+        if (!file_exists($path)) {
-            throw new GatewayException(sprintf('Could not find file %s', $this->fileDir));
+            throw new GatewayException(sprintf('Could not find file %s', $path));
         }
 
-        $fileContent = file_get_contents($this->fileDir);
+        $fileContent = file_get_contents($path);
 
         try {
             return $this->serializer->deserialize($fileContent, LtbxdMovie::class.'[]', 'csv');
@@ -34,4 +34,13 @@ readonly class LtbxdGateway
             throw new GatewayException('Error while deserializing Letterboxd data', previous: $e);
         }
     }
+
+    /**
+     * @return LtbxdMovie[]
+     * @throws GatewayException
+     */
+    public function parseFile(): array
+    {
+        return $this->parseFileFromPath($this->fileDir);
+    }
 }

src/MessageHandler/ProcessImportMessageHandler.php

@@ -0,0 +1,83 @@
+<?php
+
+declare(strict_types=1);
+
+namespace App\MessageHandler;
+
+use App\Entity\Import;
+use App\Entity\Notification;
+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);
+            $totalBatches = (int) ceil($totalFilms / self::BATCH_SIZE);
+
+            $import->setTotalFilms($totalFilms);
+            $import->setTotalBatches($totalBatches);
+            $import->setStatus(Import::STATUS_PROCESSING);
+            $this->em->flush();
+
+            for ($i = 0; $i < $totalBatches; $i++) {
+                $this->bus->dispatch(new ImportFilmsBatchMessage(
+                    importId: $import->getId(),
+                    offset: $i * self::BATCH_SIZE,
+                    limit: self::BATCH_SIZE,
+                ));
+            }
+        } catch (\Throwable $e) {
+            $this->logger->error('Import processing failed', [
+                'importId' => $import->getId(),
+                'error' => $e->getMessage(),
+            ]);
+
+            $import->setStatus(Import::STATUS_FAILED);
+            $this->em->flush();
+
+            $notification = new Notification();
+            $notification->setUser($import->getUser());
+            $notification->setMessage('L\'import a échoué.');
+            $this->em->persist($notification);
+            $this->em->flush();
+        }
+    }
+}