Implement batch visibility timeout

.github/workflows/test.yml

@@ -19,7 +19,7 @@ jobs:
             --name pgmq-postgres \
             -p 5432:5432 \
             -e POSTGRES_PASSWORD=postgres \
-            ghcr.io/pgmq/pg18-pgmq:v1.7.0
+            ghcr.io/pgmq/pg18-pgmq:v1.8.0
       - uses: shivammathur/setup-php@v2
         with:
           php-version: ${{ matrix.php }}

README.md

@@ -621,7 +621,7 @@ trapSignal([\SIGINT, \SIGTERM])
 $consumer->stop();
 $context->awaitCompletion();
 ```
-\
+
 ## License
 
 The MIT License (MIT). Please see [License File](LICENSE) for more information.

src/ConsumeController.php

@@ -4,10 +4,8 @@
 
 namespace Thesis\Pgmq;
 
-use Amp\Future;
 use Amp\Postgres\PostgresTransaction;
 use Thesis\Time\TimeSpan;
-use function Amp\async;
 
 /**
  * @api
@@ -41,17 +39,13 @@ public function ack(array $messages): void
      */
     public function nack(array $messages, TimeSpan $delay): void
     {
-        $futures = [];
-
-        foreach ($messages as $message) {
-            $futures[] = async(
-                $this->queue->setVisibilityTimeout(...),
-                $message->id,
-                $delay,
-            );
-        }
-
-        Future\awaitAll($futures);
+        $this->queue->setVisibilityTimeout(
+            array_map(
+                static fn(Message $message): int => $message->id,
+                $messages,
+            ),
+            $delay,
+        );
     }
 
     /**

src/Queue.php

@@ -180,12 +180,15 @@ public function deleteBatch(array $messageIds): array
         );
     }
 
-    public function setVisibilityTimeout(int $messageId, TimeSpan $visibilityTimeout): ?Message
+    /**
+     * @param list<int> $messageIds
+     */
+    public function setVisibilityTimeout(array $messageIds, TimeSpan $visibilityTimeout): ?Message
     {
         return setVisibilityTimeout(
             pg: $this->pg,
             queue: $this->name,
-            messageId: $messageId,
+            messageIds: $messageIds,
             visibilityTimeout: $visibilityTimeout,
         );
     }

src/pgmq.php

@@ -438,17 +438,18 @@ function deleteBatch(
 /**
  * @api
  * @param non-empty-string $queue
+ * @param list<int> $messageIds
  */
 function setVisibilityTimeout(
     PostgresLink $pg,
     string $queue,
-    int $messageId,
+    array $messageIds,
     TimeSpan $visibilityTimeout,
 ): ?Message {
     $row = $pg
-        ->execute('SELECT * FROM pgmq.set_vt(:queue_name, :msg_id::bigint, :vt::int)', [
+        ->execute('SELECT * FROM pgmq.set_vt(:queue_name, :msg_ids::bigint[], :vt::int)', [
             'queue_name' => $queue,
-            'msg_id' => $messageId,
+            'msg_ids' => $messageIds,
             'vt' => $visibilityTimeout->toSeconds(),
         ])
         ->fetchRow();

tests/PgmqTest.php

@@ -324,6 +324,49 @@ static function (array $messages, ConsumeController $ctrl) use (&$consumed): voi
         self::assertSame(0, $queue->metrics()->length);
     }
 
+    public function testNackBatch(): void
+    {
+        $queue = createQueue($this->pg, $this->randomQueueName());
+        $messageIds = $queue->sendBatch([
+            new SendMessage(self::TESTING_MESSAGE),
+            new SendMessage(self::TESTING_MESSAGE),
+        ]);
+
+        self::assertCount(2, $messageIds);
+        self::assertSame(2, $queue->metrics()->length);
+
+        $count = 0;
+
+        /** @var array<non-negative-int, list<non-empty-string>> $consumed */
+        $consumed = [];
+
+        $consumer = createConsumer($this->pg);
+        $context = $consumer->consume(
+            static function (array $messages, ConsumeController $ctrl) use (&$consumed, &$count): void {
+                /** @var Message $message */
+                foreach ($messages as $message) {
+                    $consumed[$message->id][] = $message->value;
+                    ++$count;
+                }
+
+                if ($count === 2) {
+                    $ctrl->nack($messages, TimeSpan::fromSeconds(1));
+                } elseif ($count > 2) {
+                    $ctrl->ack($messages);
+                    $ctrl->stop();
+                }
+            },
+            new ConsumeConfig($queue->name, pollInterval: TimeSpan::fromMilliseconds(500)),
+        );
+
+        $context->awaitCompletion();
+
+        self::assertCount(2, $consumed);
+        self::assertEquals($messageIds, array_keys($consumed));
+        self::assertEquals([[self::TESTING_MESSAGE, self::TESTING_MESSAGE], [self::TESTING_MESSAGE, self::TESTING_MESSAGE]], array_values($consumed));
+        self::assertSame(0, $queue->metrics()->length);
+    }
+
     public function testStopConsumeOnUnhandledException(): void
     {
         $queue = createQueue($this->pg, $this->randomQueueName());