Added:

- activity on menu
- starting activity page

src/Controller/ActivityController.php

@@ -0,0 +1,113 @@
+<?php
+
+namespace App\Controller;
+
+use App\Message\DownloadChapter;
+use App\Repository\ChapterRepository;
+use Doctrine\DBAL\Connection;
+use Symfony\Bundle\FrameworkBundle\Controller\AbstractController;
+use Symfony\Component\HttpFoundation\JsonResponse;
+use Symfony\Component\HttpFoundation\Response;
+use Symfony\Component\Messenger\Envelope;
+use Symfony\Component\Routing\Annotation\Route;
+
+class ActivityController extends AbstractController
+{
+    public function __construct(private Connection $connection, private readonly ChapterRepository $chapterRepository)
+    {
+
+    }
+
+    #[Route('/activity', name: 'app_activity')]
+    public function index(): Response
+    {
+        $queueStatus = $this->getQueueStatus();
+        $decodedPending = $this->decodeMessages($queueStatus['pending']);
+        $decodedProcessing = $this->decodeMessages($queueStatus['processing']);
+
+        $status = array_merge(
+            $this->buildStatusActivity($decodedPending),
+            $this->buildStatusActivity($decodedProcessing)
+        );
+
+        return $this->render('activity/index.html.twig', [
+            'controller_name' => 'ActivityController',
+            'status' => $status,
+        ]);
+    }
+
+    #[Route('/activity/status', name: 'app_activity_status', methods: ['GET'])]
+    public function getStatus(): JsonResponse
+    {
+        $queueStatus = $this->getQueueStatus();
+        $decodedPending = $this->decodeMessages($queueStatus['pending']);
+        $decodedProcessing = $this->decodeMessages($queueStatus['processing']);
+        $status = array_merge(
+            $this->buildStatusActivity($decodedPending),
+            $this->buildStatusActivity($decodedProcessing)
+        );
+
+        return new JsonResponse($status);
+    }
+
+//    TODO refactorer ce code avec celui du QueueStatusSubscriber
+    private function getQueueStatus(): array
+    {
+        // Requête pour récupérer les messages en attente
+        $sqlPending = 'SELECT * FROM messenger_messages WHERE queue_name = :queue AND available_at IS NULL';
+        $pending = $this->connection->fetchAllAssociative($sqlPending, ['queue' => 'default']);
+
+        // Requête pour récupérer les messages en cours de traitement
+        $sqlProcessing = 'SELECT * FROM messenger_messages WHERE queue_name = :queue AND available_at IS NOT NULL';
+        $processing = $this->connection->fetchAllAssociative($sqlProcessing, ['queue' => 'default']);
+
+        return [
+            'pending' => $pending,
+            'processing' => $processing
+        ];
+    }
+
+    private function buildStatusActivity(array $activity): array
+    {
+        $status = [];
+        foreach ($activity as $envelope) {
+            $envelope = $envelope['body'];
+            if ($envelope instanceof Envelope) {
+                if (!$envelope->getMessage() instanceof DownloadChapter) {
+                    continue;
+                }
+
+                $chapter = $this->chapterRepository->find($envelope->getMessage()->getChapterId());
+                $manga = $chapter->getManga();
+                $status[] = [
+                    'manga' => $manga->getTitle(),
+                    'volume' => $chapter->getVolume(),
+                    'chapter' => $chapter->getNumber(),
+                    'title' => $chapter->getTitle(),
+                ];
+            }
+        }
+
+        return $status;
+    }
+
+    private function decodeMessages(array $messages): array
+    {
+        $decodedMessages = [];
+
+        foreach ($messages as $message) {
+            $decodedMessages[] = [
+                'id' => $message['id'],
+                'body' => $this->decodeMessageBody($message['body']),
+                'headers' => json_decode($message['headers'], true),
+            ];
+        }
+
+        return $decodedMessages;
+    }
+
+    private function decodeMessageBody(string $body)
+    {
+        return unserialize(stripcslashes($body));
+    }
+}

src/Controller/TestController.php

@@ -5,26 +5,86 @@ namespace App\Controller;
 use App\Entity\Chapter;
 use App\Entity\ContentSource;
 use App\Entity\Manga;
+use App\Message\DownloadChapter;
+use App\Repository\ChapterRepository;
 use App\Repository\MangaRepository;
 use App\Service\MangadexProvider;
 use App\Service\MangaScraperService;
 use App\Service\MangaUpdatesMetadataProvider;
 use App\Service\SushiScanProviderService;
+use Doctrine\DBAL\Connection;
 use Symfony\Bundle\FrameworkBundle\Controller\AbstractController;
 use Symfony\Component\HttpFoundation\Response;
+use Symfony\Component\Messenger\Envelope;
+use Symfony\Component\Messenger\MessageBusInterface;
 use Symfony\Component\Routing\Attribute\Route;
+use Symfony\Component\Serializer\SerializerInterface;
 
 class TestController extends AbstractController
 {
-    public function __construct(private MangadexProvider $mangadexProvider, private MangaRepository $mangaRepository)
+    public function __construct(
+        private MangadexProvider $mangadexProvider,
+        private MangaRepository $mangaRepository,
+        private MessageBusInterface $bus,
+        private Connection $connection,
+        private SerializerInterface $serializer,
+        private readonly ChapterRepository $chapterRepository
+    )
     {
     }
 
     #[Route('/test', name: 'test')]
     public function test(): Response
     {
-        $manga = $this->mangaRepository->find(8);
+        $sqlPending = 'SELECT * FROM messenger_messages WHERE queue_name = :queue';
+        $pending = $this->connection->fetchAllAssociative($sqlPending, ['queue' => 'default']);
 
-        dd($this->mangadexProvider->getFeed($manga));
+//        // Requête pour récupérer les messages en cours de traitement
+//        $sqlProcessing = 'SELECT * FROM messenger_messages WHERE queue_name = :queue AND available_at IS NOT NULL';
+//        $processing = $this->connection->fetchAllAssociative($sqlProcessing, ['queue' => 'default']);
+
+//        dd($pending);
+        $decoded = $this->decodeMessages($pending);
+
+        $status = [];
+        foreach($decoded as $message) {
+            $message = $message['body'];
+            if($message instanceof Envelope) {
+                $chapter = $this->chapterRepository->find($message->getMessage()->getChapterId());
+                $manga = $chapter->getManga();
+                $status[] = [
+                    'manga' => $manga->getTitle(),
+                    'volume' => $chapter->getVolume(),
+                    'chapter' => $chapter->getNumber(),
+                    'title' => $chapter->getTitle(),
+                ];
+            }
+        }
+
+//        $this->bus->dispatch(new DownloadChapter(1));
+
+        dd($status);
     }
+
+    private function decodeMessages(array $messages): array
+    {
+        $decodedMessages = [];
+
+        foreach ($messages as $message) {
+            $decodedMessages[] = [
+                'id' => $message['id'],
+                'body' => $this->decodeMessageBody($message['body']),
+                'headers' => json_decode($message['headers'], true),
+            ];
+        }
+
+        return $decodedMessages;
+    }
+
+    private function decodeMessageBody(string $body)
+    {
+        return unserialize(stripcslashes($body));
+    }
+
+
 }

src/Entity/Chapter.php

@@ -116,19 +116,19 @@ class Chapter
         return $this;
     }
 
-	public function getPageByNumber(int $number): ?Page
-                                    	{
-                                    		/**
-                                    		 * @var Page $page
-                                    		 */
-                                    		foreach ($this->pagesLink as $page) {
-                                    			if ($page->getNumber() === $number) {
-                                    				return $page;
-                                    			}
-                                    		}
-                                    
-                                    		return null;
-                                    	}
+    public function getPageByNumber(int $number): ?Page
+    {
+        /**
+         * @var Page $page
+         */
+        foreach ($this->pagesLink as $page) {
+            if ($page->getNumber() === $number) {
+                return $page;
+            }
+        }
+
+        return null;
+    }
 
     public function getVolume(): ?int
     {

src/Event/PageScrappingProgressEvent.php

@@ -0,0 +1,36 @@
+<?php
+
+namespace App\Event;
+
+use Symfony\Contracts\EventDispatcher\Event;
+
+class PageScrappingProgressEvent extends Event
+{
+    public const NAME = 'page.scrapping.progress';
+
+    private int $chapterId;
+    private int $pageIndex;
+    private int $totalPages;
+
+    public function __construct(int $chapterId, int $pageIndex, int $totalPages)
+    {
+        $this->chapterId = $chapterId;
+        $this->pageIndex = $pageIndex;
+        $this->totalPages = $totalPages;
+    }
+
+    public function getChapterId(): int
+    {
+        return $this->chapterId;
+    }
+
+    public function getPageIndex(): int
+    {
+        return $this->pageIndex;
+    }
+
+    public function getTotalPages(): int
+    {
+        return $this->totalPages;
+    }
+}

src/EventSubscriber/QueueStatusSubscriber.php

@@ -0,0 +1,146 @@
+<?php
+
+namespace App\EventSubscriber;
+
+use App\Event\PageScrappingProgressEvent;
+use App\Message\DownloadChapter;
+use App\Repository\ChapterRepository;
+use App\Service\ActivityService;
+use Doctrine\DBAL\Connection;
+use Symfony\Component\EventDispatcher\EventSubscriberInterface;
+use Symfony\Component\Messenger\Envelope;
+use Symfony\Component\Messenger\Event\WorkerMessageFailedEvent;
+use Symfony\Component\Messenger\Event\WorkerMessageHandledEvent;
+use Symfony\Component\Messenger\Event\WorkerMessageReceivedEvent;
+
+class QueueStatusSubscriber implements EventSubscriberInterface
+{
+    public function __construct(
+        private ActivityService   $activityService,
+        private Connection        $connection,
+        private ChapterRepository $chapterRepository
+    )
+    {
+    }
+
+    public static function getSubscribedEvents(): array
+    {
+        return [
+            WorkerMessageReceivedEvent::class => 'onMessageReceived',
+            WorkerMessageHandledEvent::class => 'onMessageHandled',
+            WorkerMessageFailedEvent::class => 'onMessageFailed',
+            PageScrappingProgressEvent::NAME => 'onPageScrapingProgress',
+        ];
+    }
+
+    public function onMessageReceived(WorkerMessageReceivedEvent $event): void
+    {
+        $envelope = $event->getEnvelope();
+        $message = $envelope->getMessage();
+
+        if ($message instanceof DownloadChapter) {
+            $this->activityService->sendUpdate($this->getActivity());
+        }
+    }
+
+    public function onMessageHandled(WorkerMessageHandledEvent $event): void
+    {
+        $envelope = $event->getEnvelope();
+        $message = $envelope->getMessage();
+
+        if ($message instanceof DownloadChapter) {
+            $this->activityService->sendUpdate($this->getActivity());
+        }
+    }
+
+    public function onMessageFailed(WorkerMessageFailedEvent $event): void
+    {
+        $envelope = $event->getEnvelope();
+        $message = $envelope->getMessage();
+
+        if ($message instanceof DownloadChapter) {
+            $this->activityService->sendUpdate($this->getActivity());
+        }
+    }
+
+    public function onPageScrapingProgress(PageScrappingProgressEvent $event): void
+    {
+        $data = [
+            'status' => 'Page scraping progress',
+            'chapterId' => $event->getChapterId(),
+            'pageIndex' => $event->getPageIndex(),
+            'totalPages' => $event->getTotalPages(),
+        ];
+        $this->activityService->sendUpdate($data);
+    }
+
+    private function getActivity(): array
+    {
+        $queueStatus = $this->getQueueStatus();
+        return [
+            'processing' => $this->buildStatusActivity($this->decodeMessages($queueStatus['processing'])),
+            'pending' => $this->buildStatusActivity($this->decodeMessages($queueStatus['pending']))
+        ];
+    }
+
+    //TODO refactorer ce code avec celui du ActivityController
+    private function buildStatusActivity(array $activity): array
+    {
+        $status = [];
+        foreach ($activity as $envelope) {
+            $envelope = $envelope['body'];
+            if ($envelope instanceof Envelope) {
+                if (!$envelope->getMessage() instanceof DownloadChapter) {
+                    continue;
+                }
+
+                $chapter = $this->chapterRepository->find($envelope->getMessage()->getChapterId());
+                $manga = $chapter->getManga();
+                $status[] = [
+                    'manga' => $manga->getTitle(),
+                    'volume' => $chapter->getVolume(),
+                    'chapter' => $chapter->getNumber(),
+                    'title' => $chapter->getTitle(),
+                ];
+            }
+        }
+
+        return $status;
+    }
+
+    private function getQueueStatus(): array
+    {
+        // Requête pour récupérer les messages en attente
+        $sqlPending = 'SELECT * FROM messenger_messages WHERE queue_name = :queue AND available_at IS NULL';
+        $pending = $this->connection->fetchAllAssociative($sqlPending, ['queue' => 'default']);
+
+        // Requête pour récupérer les messages en cours de traitement
+        $sqlProcessing = 'SELECT * FROM messenger_messages WHERE queue_name = :queue AND available_at IS NOT NULL';
+        $processing = $this->connection->fetchAllAssociative($sqlProcessing, ['queue' => 'default']);
+
+        return [
+            'pending' => $pending,
+            'processing' => $processing
+        ];
+    }
+
+    private function decodeMessages(array $messages): array
+    {
+        $decodedMessages = [];
+
+        foreach ($messages as $message) {
+            $decodedMessages[] = [
+                'id' => $message['id'],
+                'body' => $this->decodeMessageBody($message['body']),
+                'headers' => json_decode($message['headers'], true),
+            ];
+        }
+
+        return $decodedMessages;
+    }
+
+    private function decodeMessageBody(string $body)
+    {
+        return unserialize(stripcslashes($body));
+    }
+}

src/MessageHandler/DownloadChapterHandler.php

@@ -33,10 +33,10 @@ readonly class DownloadChapterHandler
     {
         $chapter = $this->chapterRepository->find($message->getChapterId());
         if (!$chapter) {
-            $this->notificationService->sendUpdate('notification', ['status' => 'error', 'message' => 'Chapter not found.']);
+            $this->notificationService->sendUpdate(['status' => 'error', 'message' => 'Chapter not found.']);
             throw new BadRequestHttpException('Chapter not found');
         } elseif ($chapter->getLocalPath() !== null) {
-            $this->notificationService->sendUpdate('notification', ['status' => 'error', 'message' => 'Chapter already scraped.']);
+            $this->notificationService->sendUpdate(['status' => 'error', 'message' => 'Chapter already scraped.']);
             throw new BadRequestHttpException('Chapter already downloaded');
         }
 
@@ -64,7 +64,7 @@ readonly class DownloadChapterHandler
                 $scrapedSuccessfully = true;
                 break;
             } catch (Exception $e) {
-                $this->notificationService->sendUpdate('notification', [
+                $this->notificationService->sendUpdate([
                     'status' => 'warning',
                     'message' => 'An error occurred while scraping with source: ' . $source->getBaseUrl() . '. Trying next source...'
                 ]);
@@ -74,13 +74,13 @@ readonly class DownloadChapterHandler
         }
 
         if (!$scrapedSuccessfully) {
-            $this->notificationService->sendUpdate('notification', [
+            $this->notificationService->sendUpdate([
                 'status' => 'error',
                 'message' => 'All sources failed to scrape the chapter ' . $chapter->getManga()->getTitle() . ' ' . $chapter->getNumber() . '.'
             ]);
             throw new Exception('All sources failed to scrape the chapter ' . $chapter->getManga()->getTitle() . ' ' . $chapter->getNumber() . '.');
         }
 
-        $this->notificationService->sendUpdate('notification', ['status' => 'success', 'message' => 'Chapter scraped successfully.']);
+        $this->notificationService->sendUpdate(['status' => 'success', 'message' => 'Chapter scraped successfully.']);
     }
 }

src/Service/ActivityService.php

@@ -0,0 +1,20 @@
+<?php
+
+namespace App\Service;
+
+use Symfony\Component\Mercure\HubInterface;
+use Symfony\Component\Mercure\Update;
+
+class ActivityService
+{
+    public function __construct(private HubInterface $hub)
+    {
+
+    }
+
+    public function sendUpdate(mixed $data): void
+    {
+        $update = new Update('activity', json_encode($data));
+        $this->hub->publish($update);
+    }
+}

src/Service/MangaScraperService.php

@@ -5,6 +5,7 @@ namespace App\Service;
 use App\Entity\Chapter;
 use App\Entity\Manga;
 use App\Entity\ContentSource;
+use App\Event\PageScrappingProgressEvent;
 use App\EventSubscriber\MangaScrapedEvent;
 use Exception;
 use GuzzleHttp\Client;
@@ -143,6 +144,9 @@ class MangaScraperService
 
                 $this->downloadAndSaveImage($pageUrl, $imagePath);
 
+                $event = new PageScrappingProgressEvent($chapter->getId(), count($pageData) + 1, count($results['chapter']['dataSaver']));
+                $this->eventDispatcher->dispatch($event, PageScrappingProgressEvent::NAME);
+
                 $pageData[] = [
                     'image_url' => $pageUrl,
                     'local_image_url' => sprintf('/manga-images/%s/%s/%s', $mangaTitle, $chapterNumber, $imageName),
@@ -215,6 +219,9 @@ class MangaScraperService
 
             $this->downloadAndSaveImage($page['image_url'], $imagePath);
 
+            $event = new PageScrappingProgressEvent($chapter->getId(), count($pageData) + 1, 0);
+            $this->eventDispatcher->dispatch($event, PageScrappingProgressEvent::NAME);
+
             $pageData[] = [
                 'image_url' => $page['image_url'],
                 'local_image_url' => sprintf('/manga-images/%s/%s/%s', $mangaTitle, $chapterNumber, $imageName),

src/Service/MangadexProvider.php

@@ -12,7 +12,7 @@ use Symfony\Component\String\Slugger\SluggerInterface;
 
 readonly class MangadexProvider implements MetadataProviderInterface
 {
-    public function __construct(private ClientInterface $client, private SluggerInterface $slugger)
+    public function __construct(private ClientInterface $client, private SluggerInterface $slugger, private NotificationService $notificationService)
     {
     }
 
@@ -22,12 +22,17 @@ readonly class MangadexProvider implements MetadataProviderInterface
             return new ArrayCollection();
         }
 
-        $results = $this->client->get('/manga', [
-            'title' => $title,
-            'contentRating' => ['safe', 'suggestive'],
-            'includes' => ['cover_art', 'author'],
-            'limit' => 25
-        ]);
+        try{
+            $results = $this->client->get('/manga', [
+                'title' => $title,
+                'contentRating' => ['safe', 'suggestive'],
+                'includes' => ['cover_art', 'author'],
+                'limit' => 25
+            ]);
+        }catch(\Exception $e){
+            $this->notificationService->sendUpdate('notification', ['status' => 'error', 'message' => 'An error occurred while fetching data from Mangadex.']);
+            return new ArrayCollection();
+        }
 
         $mangas = [];
         foreach ($results['data'] as $result) {

src/Service/NotificationService.php

@@ -12,9 +12,9 @@ class NotificationService
 
     }
 
-    public function sendUpdate(string $topic, mixed $data): void
+    public function sendUpdate(mixed $data): void
     {
-        $update = new Update($topic, json_encode($data));
+        $update = new Update('notification', json_encode($data));
         $this->hub->publish($update);
     }
 }