From 6e3fa90cc0c8dc85f4a254c0b7e51dd60404ab39 Mon Sep 17 00:00:00 2001
From: Fabiana Romagnoli <fromagnoli@webgriffe.com>
Date: Thu, 7 Nov 2024 15:12:32 +0100
Subject: [PATCH] Fix errored job removal from batch

---
 src/ProducerInstance.php                        |  8 --------
 src/Service/QueueManager.php                    | 11 ++++++++++-
 tests/Integration/ElasticSearchIndexingTest.php | 10 +++++++++-
 3 files changed, 19 insertions(+), 10 deletions(-)

diff --git a/src/ProducerInstance.php b/src/ProducerInstance.php
index fa97a17..77fd584 100644
--- a/src/ProducerInstance.php
+++ b/src/ProducerInstance.php
@@ -173,14 +173,6 @@ public function produceAndQueueJobs($data = null): Promise
                     $job = $jobs->getCurrent();
                     $job->addEvent(new ProducedJobEvent(new \DateTime(), \get_class($this->producer)));
                     $jobsCount += yield $this->queueManager->enqueue($job);
-                    $this->logger->info(
-                        'Successfully produced a new Job',
-                        [
-                            'producer' => \get_class($this->producer),
-                            'job_uuid' => $job->getUuid(),
-                            'payload_data' => NonUtf8Cleaner::clean($job->getPayloadData())
-                        ]
-                    );
                 }
 
                 $jobsCount += yield $this->queueManager->flush();
diff --git a/src/Service/QueueManager.php b/src/Service/QueueManager.php
index b77f778..5d8c01b 100644
--- a/src/Service/QueueManager.php
+++ b/src/Service/QueueManager.php
@@ -13,6 +13,7 @@
 use Webgriffe\Esb\Model\FlowConfig;
 use Webgriffe\Esb\Model\Job;
 use Webgriffe\Esb\Model\JobInterface;
+use Webgriffe\Esb\NonUtf8Cleaner;
 
 final class QueueManager implements ProducerQueueManagerInterface, WorkerQueueManagerInterface
 {
@@ -96,7 +97,7 @@ public function enqueue(JobInterface $job): Promise
                     )
                 );
             }
-            $this->batch[] = $job;
+            $this->batch[$job->getUuid()] = $job;
 
             $count = count($this->batch);
             if ($count < $this->batchSize) {
@@ -248,6 +249,14 @@ private function processBatch(): \Generator
                 $singleJob->getDelay(),
                 $singleJob->getPriority()
             );
+            $this->logger->info(
+                'Successfully enqueued a new Job',
+                [
+                    'flow_name' => $this->flowConfig->getName(),
+                    'job_uuid' => $singleJob->getUuid(),
+                    'payload_data' => NonUtf8Cleaner::clean($singleJob->getPayloadData())
+                ]
+            );
         }
 
         $this->batch = [];
diff --git a/tests/Integration/ElasticSearchIndexingTest.php b/tests/Integration/ElasticSearchIndexingTest.php
index 21ebded..b81660e 100644
--- a/tests/Integration/ElasticSearchIndexingTest.php
+++ b/tests/Integration/ElasticSearchIndexingTest.php
@@ -175,7 +175,7 @@ function (Job $job) {
     /**
      * @test
      */
-    public function itLogsAndSkipsJobsThatCouldNotBeIndexedOntoElasticSearchWithAllEvents()
+    public function itLogsAndSkipsJobsThatCouldNotBeIndexedOntoElasticSearch()
     {
         $producerDir = vfsStream::url('root/producer_dir');
         $workerFile = vfsStream::url('root/worker.data');
@@ -216,6 +216,14 @@ function ($log) {
         $search = Promise\wait($this->esClient->uriSearchOneIndex(self::FLOW_CODE, ''));
         $this->assertCount(1, $search['hits']['hits']);
         $this->assertTrue($this->logHandler()->hasErrorThatContains('Job could not be indexed in ElasticSearch'));
+        $logRecords = $this->logHandler()->getRecords();
+        $successfullyIndexedLog = array_filter(
+            $logRecords,
+            function ($log) {
+                return strpos($log['message'], 'Successfully enqueued a new Job') !== false;
+            }
+        );
+        $this->assertCount(1, $successfullyIndexedLog);
     }
 
     private function assertForEachJob(callable $callable, array $jobsData)