Batch SQS message processing in jobs to improve efficiency. Add error filtering to process monitor.

Author: rbruhn

app/Jobs/TesseractOcrProcessJob.php

@@ -88,20 +88,35 @@ public function handle(SqsClient $sqs): void
             ->where('processed', 0)
             ->orderBy('id')
             ->chunkById(1000, function ($files) use ($sqs, $queueUrl, $updatesQueueUrl, &$sentCount) {
+                $batch = [];
                 foreach ($files as $file) {
-                    $payload = [
-                        'ocrQueueFileId' => $file->id,
-                        'subjectId' => $file->subject_id,
-                        'access_uri' => $file->access_uri,
-                        'updatesQueueUrl' => $updatesQueueUrl,
+                    $batch[] = [
+                        'Id' => (string) $file->id,
+                        'MessageBody' => json_encode([
+                            'ocrQueueFileId' => $file->id,
+                            'subjectId' => $file->subject_id,
+                            'access_uri' => $file->access_uri,
+                            'updatesQueueUrl' => $updatesQueueUrl,
+                        ]),
                     ];
 
-                    $sqs->sendMessage([
+                    if (count($batch) === 10) {
+                        $sqs->sendMessageBatch([
+                            'QueueUrl' => $queueUrl,
+                            'Entries' => $batch,
+                        ]);
+                        $sentCount += 10;
+                        $batch = [];
+                    }
+                }
+
+                // Send remaining messages
+                if (! empty($batch)) {
+                    $sqs->sendMessageBatch([
                         'QueueUrl' => $queueUrl,
-                        'MessageBody' => json_encode($payload),
+                        'Entries' => $batch,
                     ]);
-
-                    $sentCount++;
+                    $sentCount += count($batch);
                 }
             }, 'id');
 

app/Jobs/ZooniverseExportProcessImagesJob.php

@@ -87,22 +87,38 @@ public function handle(SqsClient $sqs): void
             ->where('processed', 0)
             ->orderBy('id')
             ->chunkById(1000, function ($files) use ($sqs, $queueUrl, $updatesQueueUrl, $processDir, &$sentCount) {
-                foreach ($files as $file) {
-                    $payload = [
-                        'processDir' => $processDir,
-                        'accessURI' => $file->access_uri,
-                        'subjectId' => $file->subject_id,
-                        's3Bucket' => config('filesystems.disks.s3.bucket'),
-                        'updatesQueueUrl' => $updatesQueueUrl,
-                        'queueId' => $this->exportQueue->id,
+                $batch = [];
+                foreach ($files as $index => $file) {
+                    $batch[] = [
+                        'Id' => (string) $file->id, // Required: unique string for the batch
+                        'MessageBody' => json_encode([
+                            'processDir' => $processDir,
+                            'accessURI' => $file->access_uri,
+                            'subjectId' => $file->subject_id,
+                            's3Bucket' => config('filesystems.disks.s3.bucket'),
+                            'updatesQueueUrl' => $updatesQueueUrl,
+                            'queueId' => $this->exportQueue->id,
+                        ]),
                     ];
 
-                    $sqs->sendMessage([
+                    // SQS Batch limit is 10
+                    if (count($batch) === 10) {
+                        $sqs->sendMessageBatch([
+                            'QueueUrl' => $queueUrl,
+                            'Entries' => $batch,
+                        ]);
+                        $sentCount += count($batch);
+                        $batch = [];
+                    }
+                }
+
+                // Send remaining messages (less than 10)
+                if (count($batch) > 0) {
+                    $sqs->sendMessageBatch([
                         'QueueUrl' => $queueUrl,
-                        'MessageBody' => json_encode($payload),
+                        'Entries' => $batch,
                     ]);
-
-                    $sentCount++;
+                    $sentCount += count($batch);
                 }
             }, 'id');
 

app/Livewire/ProcessMonitor.php

@@ -59,6 +59,7 @@ private function getOcrQueues(): Collection
             ->select(['id', 'expedition_id', 'project_id', 'total', 'stage'])
             ->with(['expedition:id,title', 'project:id,title'])
             ->withCount(['files as processed_files' => fn ($q) => $q->where('processed', 1)])
+            ->where('error', 0)
             ->orderBy('id')
             ->get();
     }