Implement FIFO queues

README.md

@@ -48,6 +48,13 @@ And only the consumer accepts `Amp\Postgres\PostgresConnection`, because it itse
  - [Test routing](#test-routing)
  - [Validate routing key](#validate-routing-key)
  - [Validate topic pattern](#validate-topic-pattern)
+ - [Read grouped](#read-grouped)
+ - [Read grouped round-robin](#read-grouped-round-robin)
+ - [Read grouped head](#read-grouped-head)
+ - [Read grouped with poll](#read-grouped-with-poll)
+ - [Read grouped round-robin with poll](#read-grouped-round-robin-with-poll)
+ - [Create FIFO index](#create-fifo-index)
+ - [Create FIFO index for all queues](#create-fifo-index-for-all-queues)
  - [Consume messages](#consume-messages)
 
 ### Create queue
@@ -509,6 +516,113 @@ $pg = new Postgres\PostgresConnectionPool(Postgres\PostgresConfig::fromString(''
 Pgmq\validateTopicPattern($pg, 'events.*'); // throws PostgresQueryError if invalid
 ```
 
+### Read grouped
+
+Read messages respecting FIFO ordering within groups. Messages are grouped by the `x-pgmq-group` header. Only the oldest unprocessed message from each group is returned.
+
+```php
+use Thesis\Pgmq;
+use Amp\Postgres;
+use Thesis\Time\TimeSpan;
+
+$pg = new Postgres\PostgresConnectionPool(Postgres\PostgresConfig::fromString(''));
+
+$queue = Pgmq\createQueue($pg, 'orders');
+$queue->send(new Pgmq\SendMessage('{"action": "created"}', '{"x-pgmq-group": "order-1"}'));
+$queue->send(new Pgmq\SendMessage('{"action": "paid"}', '{"x-pgmq-group": "order-1"}'));
+$queue->send(new Pgmq\SendMessage('{"action": "created"}', '{"x-pgmq-group": "order-2"}'));
+
+$messages = $queue->readGrouped(10, TimeSpan::fromSeconds(30));
+```
+
+### Read grouped round-robin
+
+Read messages with round-robin distribution across groups.
+
+```php
+use Thesis\Pgmq;
+use Amp\Postgres;
+
+$pg = new Postgres\PostgresConnectionPool(Postgres\PostgresConfig::fromString(''));
+
+$queue = Pgmq\createQueue($pg, 'orders');
+$messages = $queue->readGroupedRR(10);
+```
+
+### Read grouped head
+
+Read exactly one message per FIFO group — the head (oldest, lowest msg_id) message in each group — across up to qty groups in a single operation.
+Only groups with a visible, unlocked head message are included.
+
+```php
+use Thesis\Pgmq;
+use Amp\Postgres;
+
+$pg = new Postgres\PostgresConnectionPool(Postgres\PostgresConfig::fromString(''));
+
+$queue = Pgmq\createQueue($pg, 'orders');
+$messages = $queue->readGroupedHead(10);
+```
+
+### Read grouped with poll
+
+```php
+use Thesis\Pgmq;
+use Amp\Postgres;
+use Thesis\Time\TimeSpan;
+
+$pg = new Postgres\PostgresConnectionPool(Postgres\PostgresConfig::fromString(''));
+
+$queue = Pgmq\createQueue($pg, 'orders');
+$messages = $queue->readGroupedWithPoll(
+    count: 10,
+    maxPoll: TimeSpan::fromSeconds(5),
+    pollInterval: TimeSpan::fromMilliseconds(250),
+);
+```
+
+### Read grouped round-robin with poll
+
+```php
+use Thesis\Pgmq;
+use Amp\Postgres;
+use Thesis\Time\TimeSpan;
+
+$pg = new Postgres\PostgresConnectionPool(Postgres\PostgresConfig::fromString(''));
+
+$queue = Pgmq\createQueue($pg, 'orders');
+$messages = $queue->readGroupedRRWithPoll(
+    count: 10,
+    maxPoll: TimeSpan::fromSeconds(5),
+    pollInterval: TimeSpan::fromMilliseconds(250),
+);
+```
+
+### Create FIFO index
+
+Create a GIN index on the headers column for FIFO queue performance optimization. This is required before using grouped read functions.
+
+```php
+use Thesis\Pgmq;
+use Amp\Postgres;
+
+$pg = new Postgres\PostgresConnectionPool(Postgres\PostgresConfig::fromString(''));
+
+$queue = Pgmq\createQueue($pg, 'orders');
+Pgmq\createFifoIndex($pg, 'orders');
+```
+
+### Create FIFO index for all queues
+
+```php
+use Thesis\Pgmq;
+use Amp\Postgres;
+
+$pg = new Postgres\PostgresConnectionPool(Postgres\PostgresConfig::fromString(''));
+
+Pgmq\createFifoIndexAll($pg);
+```
+
 ### Consume messages
 
 This functionality is not a standard feature of the **pgmq** extension, but is provided by the library as an add-on for reliable and correct processing of message batches from the queue, with the ability to `ack`, `nack` (with delay) and archive (`term`) messages from the queue.

examples/fifo.php

@@ -0,0 +1,30 @@
+<?php
+
+declare(strict_types=1);
+
+require_once __DIR__ . '/../vendor/autoload.php';
+
+use Amp\Postgres;
+use Thesis\Pgmq;
+
+$pg = new Postgres\PostgresConnectionPool(Postgres\PostgresConfig::fromString('host=pgmq user=postgres password=postgres'));
+
+Pgmq\createExtension($pg);
+
+$queue = Pgmq\createQueue($pg, 'orders');
+$queue->createFifoIndex();
+
+// Send messages with group headers for FIFO ordering
+$queue->send(new Pgmq\SendMessage('{"order": 1, "action": "created"}', '{"x-pgmq-group": "customer-1"}'));
+$queue->send(new Pgmq\SendMessage('{"order": 1, "action": "paid"}', '{"x-pgmq-group": "customer-1"}'));
+$queue->send(new Pgmq\SendMessage('{"order": 2, "action": "created"}', '{"x-pgmq-group": "customer-2"}'));
+$queue->send(new Pgmq\SendMessage('{"order": 2, "action": "paid"}', '{"x-pgmq-group": "customer-2"}'));
+
+// Read one message per group (FIFO within each group)
+$messages = [...$queue->readGroupedRR(10)];
+
+dump(array_map(static fn(Pgmq\Message $m): string => $m->value, $messages));
+
+$queue->deleteBatch(array_map(static fn(Pgmq\Message $message): int => $message->id, $messages));
+
+$queue->drop();

src/Queue.php

@@ -116,7 +116,7 @@ public function read(?TimeSpan $visibilityTimeout = null): ?Message
 
     /**
      * @param positive-int $count
-     * @return iterable<Message>
+     * @return iterable<int, Message>
      */
     public function readBatch(int $count, ?TimeSpan $visibilityTimeout = null): iterable
     {
@@ -128,6 +128,96 @@ public function readBatch(int $count, ?TimeSpan $visibilityTimeout = null): iter
         );
     }
 
+    /**
+     * @param positive-int $count
+     * @return iterable<int, Message>
+     */
+    public function readGroupedRR(int $count, ?TimeSpan $visibilityTimeout = null): iterable
+    {
+        return readGroupedRR(
+            pg: $this->pg,
+            queue: $this->name,
+            count: $count,
+            visibilityTimeout: $visibilityTimeout,
+        );
+    }
+
+    /**
+     * @param positive-int $count
+     * @return iterable<int, Message>
+     */
+    public function readGroupedRRWithPoll(
+        int $count,
+        ?TimeSpan $visibilityTimeout = null,
+        ?TimeSpan $maxPoll = null,
+        ?TimeSpan $pollInterval = null,
+    ): iterable {
+        return readGroupedRRWithPoll(
+            pg: $this->pg,
+            queue: $this->name,
+            count: $count,
+            visibilityTimeout: $visibilityTimeout,
+            maxPoll: $maxPoll,
+            pollInterval: $pollInterval,
+        );
+    }
+
+    /**
+     * @param positive-int $count
+     * @return iterable<int, Message>
+     */
+    public function readGrouped(int $count, ?TimeSpan $visibilityTimeout = null): iterable
+    {
+        return readGrouped(
+            pg: $this->pg,
+            queue: $this->name,
+            count: $count,
+            visibilityTimeout: $visibilityTimeout,
+        );
+    }
+
+    /**
+     * @param positive-int $count
+     * @return iterable<int, Message>
+     */
+    public function readGroupedWithPoll(
+        int $count,
+        ?TimeSpan $visibilityTimeout = null,
+        ?TimeSpan $maxPoll = null,
+        ?TimeSpan $pollInterval = null,
+    ): iterable {
+        return readGroupedWithPoll(
+            pg: $this->pg,
+            queue: $this->name,
+            count: $count,
+            visibilityTimeout: $visibilityTimeout,
+            maxPoll: $maxPoll,
+            pollInterval: $pollInterval,
+        );
+    }
+
+    /**
+     * @param positive-int $count
+     * @return iterable<int, Message>
+     */
+    public function readGroupedHead(int $count, ?TimeSpan $visibilityTimeout = null): iterable
+    {
+        return readGroupedHead(
+            pg: $this->pg,
+            queue: $this->name,
+            count: $count,
+            visibilityTimeout: $visibilityTimeout,
+        );
+    }
+
+    public function createFifoIndex(): void
+    {
+        createFifoIndex(
+            pg: $this->pg,
+            queue: $this->name,
+        );
+    }
+
     public function pop(): ?Message
     {
         return pop(

src/extension.php

@@ -0,0 +1,15 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Thesis\Pgmq;
+
+use Amp\Postgres\PostgresLink;
+
+/**
+ * @api
+ */
+function createExtension(PostgresLink $pg): void
+{
+    $pg->execute('CREATE EXTENSION IF NOT EXISTS pgmq CASCADE;');
+}