Dev/batched messeages

composer.json

@@ -120,6 +120,7 @@
         "symfony/serializer": "^6.4",
         "symfony/string": "^6.4",
         "symfony/translation": "^6.4",
+        "symfony/uid": "^6.4",
         "symfony/validator": "^6.4",
         "symfony/yaml": "^6.4",
         "twig/twig": "^3.9"

configs/db/tables/gems__batch.20.sql

@@ -0,0 +1,24 @@
+CREATE TABLE if not exists gems__batch (
+    gba_id                 char(36) not null,
+    gba_iteration          int unsigned not null,
+
+    gba_name               varchar(64) null,
+    gba_group              varchar(64) null,
+    gba_status             varchar(64) not null,
+    gba_synchronous        tinyint(1) not null default 0,
+
+    gba_message            JSON null,
+    gba_message_class      varchar(128) null,
+
+    gba_info                text null,
+
+    gba_started            timestamp null,
+    gba_finished           timestamp null,
+    gba_created            timestamp not null default current_timestamp,
+    gba_changed            timestamp not null default current_timestamp ON UPDATE current_timestamp,
+
+    PRIMARY KEY (gba_id, gba_iteration)
+)
+    ENGINE=InnoDB
+    AUTO_INCREMENT = 1000
+    CHARACTER SET 'utf8mb4' COLLATE 'utf8mb4_unicode_ci';

src/Command/ClearMessengerBatch.php

@@ -0,0 +1,70 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Gems\Command;
+
+use DateInterval;
+use Exception;
+use Gems\Messenger\Batch\BatchStoreInterface;
+use Psr\Container\ContainerInterface;
+use Symfony\Component\Console\Attribute\AsCommand;
+use Symfony\Component\Console\Command\Command;
+use Symfony\Component\Console\Input\InputArgument;
+use Symfony\Component\Console\Input\InputInterface;
+use Symfony\Component\Console\Output\OutputInterface;
+
+#[AsCommand(name: 'app:clear-messenger-batch', description: 'Clear old messenger batch data')]
+class ClearMessengerBatch extends Command
+{
+
+    public function __construct(
+        private readonly ContainerInterface $container,
+    )
+    {
+        parent::__construct();
+    }
+
+    protected function configure()
+    {
+        $this->addArgument(
+            name: 'days',
+            mode: InputArgument::OPTIONAL,
+            description: 'Age in days of items to remove. Can also be a DateInterval notation (e.g. P7D)',
+            default: '7',
+        );
+
+        $this->addArgument(
+            name: 'group',
+            mode: InputArgument::OPTIONAL,
+            description: 'Optional Group to clear',
+        );
+    }
+
+    protected function execute(InputInterface $input, OutputInterface $output): int
+    {
+        $removeInterval = $this->getInterval($input);
+
+        /** @var BatchStoreInterface $batchStore */
+        $batchStore = $this->container->get(BatchStoreInterface::class);
+
+        $batchStore->clearOldBatches($removeInterval);
+
+        return Command::SUCCESS;
+    }
+
+    protected function getInterval(InputInterface $input): DateInterval
+    {
+        $days = $input->getArgument('days');
+        if (ctype_digit($days)) {
+            return new DateInterval('P' . (int)$days . 'D');
+        }
+
+        try {
+            $interval = new DateInterval($days);
+            return $interval;
+        } catch (Exception $e) {
+            throw new Exception(sprintf('Input days (%s) is not an int or a DateInterval string', $days));
+        }
+    }
+}

src/Command/RunCommJob.php

@@ -3,10 +3,12 @@
 namespace Gems\Command;
 
 use Gems\Console\ConsoleSettings;
+use Gems\Messenger\Batch\MessengerBatchRepository;
 use Gems\Messenger\Message\CommJob;
 use Gems\Repository\CommJobRepository;
 use Gems\Util\Lock\CommJobLock;
 use Gems\Util\Lock\MaintenanceLock;
+use Psr\Container\ContainerInterface;
 use Symfony\Component\Console\Attribute\AsCommand;
 use Symfony\Component\Console\Command\Command;
 use Symfony\Component\Console\Input\InputArgument;
@@ -19,12 +21,7 @@
 class RunCommJob extends Command
 {
     public function __construct(
-        protected CommJobLock $commJobLock,
-        protected MaintenanceLock $maintenanceLock,
-        protected MessageBusInterface $messageBus,
-        protected CommJobRepository $commJobRepository,
-        protected ConsoleSettings $consoleSettings,
-        protected array $config,
+        private readonly ContainerInterface $container,
     )
     {
         parent::__construct();
@@ -37,22 +34,38 @@ protected function configure(): void
 
     protected function execute(InputInterface $input, OutputInterface $output)
     {
-        if ($this->maintenanceLock->isLocked()) {
+        /**
+         * @var MaintenanceLock $maintenanceLock
+         */
+        $maintenanceLock = $this->container->get(MaintenanceLock::class);
+
+        if ($maintenanceLock->isLocked()) {
             $output->writeln('<error>Cannot run cron job in maintenance mode.</error>');
             return static::FAILURE;
         }
-        if ($this->commJobLock->isLocked()) {
+
+        /**
+         * @var CommJobLock $commJobLock
+         */
+        $commJobLock = $this->container->get(CommJobLock::class);
+        if ($commJobLock->isLocked()) {
             $output->writeln('<error>Cron jobs turned off.</error>');
             return static::FAILURE;
         }
 
-        $this->consoleSettings->setConsoleUser();
+        /**
+         * @var ConsoleSettings $consoleSettings
+         */
+        $consoleSettings = $this->container->get(ConsoleSettings::class);
+        $consoleSettings->setConsoleUser();
 
-        $jobs = $this->commJobRepository->getAutomaticJobs();
+        /** @var CommJobRepository $commJobRepository */
+        $commJobRepository = $this->container->get(CommJobRepository::class);
+        $jobs = $commJobRepository->getAutomaticJobs();
 
         $id = $input->getArgument('id');
         if ($id !== null) {
-            $jobs = $this->commJobRepository->getActiveJobs();
+            $jobs = $commJobRepository->getActiveJobs();
             $jobs = array_filter($jobs, function ($job) use ($id) {
                 return $job['gcj_id_job'] == $id;
             });
@@ -61,18 +74,27 @@ protected function execute(InputInterface $input, OutputInterface $output)
             }
         }
 
-        $this->commJobRepository->clearTokenQueue();
+        $commJobRepository->clearTokenQueue();
+
+        /** @var MessengerBatchRepository $messengerBatchRepository */
+        $messengerBatchRepository = $this->container->get(MessengerBatchRepository::class);
+        $now = new \DateTimeImmutable();
+        $batch = $messengerBatchRepository->createBatch('CommJob ' . $now->format('Y-m-d H:i:s'), 'CommJob', true);
         foreach($jobs as $jobData) {
             $commJobMessage = new CommJob($jobData);
-            $envelope = $this->messageBus->dispatch($commJobMessage);
-            /**
-             * @var HandledStamp $stamp
-             */
-            $stamp = $envelope->last(HandledStamp::class);
-            $output->writeln($stamp->getResult());
+            $batch->addMessage($commJobMessage);
         }
 
-        $output->writeln(sprintf('<info>Executed %d jobs</info>', count($jobs)));
+        $messengerBatchRepository->dispatch($batch);
+
+        $output->writeln(sprintf('<info>Dispatched %d jobs</info>', count($jobs)));
+
+        /** @var MessengerBatchRepository $messengerBatchRepository */
+        $messengerBatchRepository = $this->container->get(MessengerBatchRepository::class);
+        $messages = $messengerBatchRepository->getBatchInfoList($batch->batchId);
+        foreach($messages as $message) {
+            $output->writeln(sprintf('<info>%s</info>', $message));
+        }
 
         return static::SUCCESS;
     }

src/ConfigProvider.php

@@ -44,6 +44,8 @@
 use Gems\Menu\RouteHelperFactory;
 use Gems\Messenger\MessengerFactory;
 use Gems\Messenger\TransportFactory;
+use Gems\Messenger\Batch\BatchStoreInterface;
+use Gems\Messenger\Batch\DatabaseBatchStore;
 use Gems\Middleware\FlashMessageMiddleware;
 use Gems\Model\Bridge\GemsFormBridge;
 use Gems\Model\Bridge\GemsValidatorBridge;
@@ -485,6 +487,7 @@ public function getDependencies(): array
 
                 // Messenger
                 'messenger.bus.default' => MessageBusInterface::class,
+                BatchStoreInterface::class => DatabaseBatchStore::class,
 
                 // Session
                 //SessionPersistenceInterface::class => CacheSessionPersistence::class,

src/Db/ResultFetcher.php

@@ -147,7 +147,7 @@ protected function do_query(Select|string $select, ?array $params = null)
         return $this->db->query($select, $params, $resultSet);
     }
 
-    public function insertIntoTable(string $tableName, array $values): int
+    public function insertIntoTable(string $tableName, array|Select $values): int
     {
         $table = new TableGateway($tableName, $this->getAdapter());
         $table->insert($values);

src/Messenger/Batch/Batch.php

@@ -0,0 +1,74 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Gems\Messenger\Batch;
+
+use DateTimeInterface;
+
+class Batch
+{
+    /*
+     * @var object[]
+     */
+    private array $messages = [];
+
+    public function __construct(
+        public readonly string $batchId,
+        public readonly DateTimeInterface $created,
+        public readonly string|null $name = null,
+        public readonly string|null $group = null,
+        public readonly bool $isChain = false,
+        public readonly DateTimeInterface|null $finished = null,
+        public readonly int|null $totalItems = null,
+        public readonly int|null $pending = null,
+        public readonly int|null $running = null,
+        public readonly int|null $success = null,
+        public readonly int|null $failed = null,
+
+
+
+    )
+    {
+    }
+
+    public function addMessage(object $message): void
+    {
+        $this->messages[] = $message;
+    }
+
+    /**
+     * @param object[] $messages
+     * @return void
+     */
+    public function addMessages(array $messages): void
+    {
+        foreach ($messages as $message) {
+            $this->addMessage($message);
+        }
+    }
+
+    /**
+     * @return object[]
+     */
+    public function getCurrentMessages(): array
+    {
+        return $this->messages;
+    }
+
+    public function getPercent(): float
+    {
+        if (!$this->totalItems) {
+            return 0;
+        }
+        if (!$this->success || $this->success === 0) {
+            return 0;
+        }
+        return $this->success / $this->totalItems;
+    }
+
+    public function clearMessages(): void
+    {
+        $this->messages = [];
+    }
+}

src/Messenger/Batch/BatchMiddleware.php

@@ -0,0 +1,79 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Gems\Messenger\Batch;
+
+use Psr\Container\ContainerInterface;
+use Symfony\Component\Messenger\Envelope;
+use Symfony\Component\Messenger\MessageBusInterface;
+use Symfony\Component\Messenger\Middleware\MiddlewareInterface;
+use Symfony\Component\Messenger\Middleware\StackInterface;
+use Symfony\Component\Messenger\Stamp\HandledStamp;
+
+class BatchMiddleware implements MiddlewareInterface
+{
+    public function __construct(
+        private readonly ContainerInterface $container,
+    )
+    {
+    }
+
+    public function handle(Envelope $envelope, StackInterface $stack): Envelope
+    {
+        /** @var BatchStamp|null $stamp */
+        $stamp = $envelope->last(BatchStamp::class);
+        if (!$stamp) {
+            return $stack->next()->handle($envelope, $stack);
+        }
+
+        /** @var MessengerBatchRepository $batchRepository */
+        $batchRepository = $this->container->get(MessengerBatchRepository::class);
+
+        try {
+            $batchRepository->setIterationStatus($stamp->batchId, $stamp->iteration, BatchStatus::RUNNING);
+
+            $envelope = $stack->next()->handle($envelope, $stack);
+            $handledStamp =$envelope->last(HandledStamp::class);
+            $result = $handledStamp?->getResult();
+
+            $message = null;
+            if (is_string($result)) {
+                $message = $result;
+            }
+
+            $batch = $batchRepository->getBatch($stamp->batchId);
+            $batchRepository->setIterationStatus($stamp->batchId, $stamp->iteration, BatchStatus::SUCCESS, $message);
+
+            if ($batch->isChain) {
+                $nextMessage = $batchRepository->getBatchIterationMessage($stamp->batchId, $stamp->iteration + 1);
+                if ($nextMessage) {
+                    /**
+                     * @var MessageBusInterface $messageBus
+                     */
+                    $messageBus = $this->container->get(MessageBusInterface::class);
+                    $messageBus->dispatch($nextMessage, [
+                        new BatchStamp($stamp->batchId, $stamp->iteration + 1),
+                    ]);
+                }
+            }
+
+
+        } catch (\Throwable $e) {
+            $batchRepository->setIterationStatus(
+                $stamp->batchId,
+                $stamp->iteration,
+                BatchStatus::FAILED,
+                $e->getMessage()
+            );
+            $batch = $batchRepository->getBatch($stamp->batchId);
+            if ($batch->isChain) {
+                $batchRepository->failChain($stamp->batchId, $stamp->iteration);
+            }
+
+            throw $e;
+        }
+
+        return $envelope;
+    }
+}