diff --git a/README.md b/README.md
index e3f3b9f..e48ba22 100644
--- a/README.md
+++ b/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.
diff --git a/examples/fifo.php b/examples/fifo.php
new file mode 100644
index 0000000..1bb918c
--- /dev/null
+++ b/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();
diff --git a/src/Queue.php b/src/Queue.php
index ad464f0..fba37af 100644
--- a/src/Queue.php
+++ b/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(
diff --git a/src/extension.php b/src/extension.php
new file mode 100644
index 0000000..82552a2
--- /dev/null
+++ b/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;');
+}
diff --git a/src/fifo.php b/src/fifo.php
new file mode 100644
index 0000000..982db97
--- /dev/null
+++ b/src/fifo.php
@@ -0,0 +1,152 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Thesis\Pgmq;
+
+use Amp\Postgres\PostgresLink;
+use Thesis\Time\TimeSpan;
+
+/**
+ * @api
+ * @param non-empty-string $queue
+ * @param positive-int $count
+ * @return iterable<int, Message>
+ */
+function readGroupedRR(
+    PostgresLink $pg,
+    string $queue,
+    int $count,
+    ?TimeSpan $visibilityTimeout = null,
+): iterable {
+    $result = $pg->execute('SELECT * FROM pgmq.read_grouped_rr(:queue_name, :vt, :qty)', [
+        'queue_name' => $queue,
+        'qty' => $count,
+        'vt' => ($visibilityTimeout ?? TimeSpan::fromSeconds(30))->toSeconds(),
+    ]);
+
+    foreach ($result as $row) {
+        yield Message::fromArray($row);
+    }
+}
+
+/**
+ * @api
+ * @param non-empty-string $queue
+ * @param positive-int $count
+ * @return iterable<int, Message>
+ */
+function readGroupedRRWithPoll(
+    PostgresLink $pg,
+    string $queue,
+    int $count,
+    ?TimeSpan $visibilityTimeout = null,
+    ?TimeSpan $maxPoll = null,
+    ?TimeSpan $pollInterval = null,
+): iterable {
+    $result = $pg->execute('SELECT * FROM pgmq.read_grouped_rr_with_poll(:queue_name, :vt, :qty, :max_poll_seconds, :poll_interval_ms)', [
+        'queue_name' => $queue,
+        'qty' => $count,
+        'vt' => ($visibilityTimeout ?? TimeSpan::fromSeconds(30))->toSeconds(),
+        'max_poll_seconds' => ($maxPoll ?? TimeSpan::fromSeconds(5))->toSeconds(),
+        'poll_interval_ms' => ($pollInterval ?? TimeSpan::fromMilliseconds(250))->toMilliseconds(),
+    ]);
+
+    foreach ($result as $row) {
+        yield Message::fromArray($row);
+    }
+}
+
+/**
+ * @api
+ * @param non-empty-string $queue
+ * @param positive-int $count
+ * @return iterable<int, Message>
+ */
+function readGrouped(
+    PostgresLink $pg,
+    string $queue,
+    int $count,
+    ?TimeSpan $visibilityTimeout = null,
+): iterable {
+    $result = $pg->execute('SELECT * FROM pgmq.read_grouped(:queue_name, :vt, :qty)', [
+        'queue_name' => $queue,
+        'qty' => $count,
+        'vt' => ($visibilityTimeout ?? TimeSpan::fromSeconds(30))->toSeconds(),
+    ]);
+
+    foreach ($result as $row) {
+        yield Message::fromArray($row);
+    }
+}
+
+/**
+ * @api
+ * @param non-empty-string $queue
+ * @param positive-int $count
+ * @return iterable<int, Message>
+ */
+function readGroupedWithPoll(
+    PostgresLink $pg,
+    string $queue,
+    int $count,
+    ?TimeSpan $visibilityTimeout = null,
+    ?TimeSpan $maxPoll = null,
+    ?TimeSpan $pollInterval = null,
+): iterable {
+    $result = $pg->execute('SELECT * FROM pgmq.read_grouped_with_poll(:queue_name, :vt, :qty, :max_poll_seconds, :poll_interval_ms)', [
+        'queue_name' => $queue,
+        'qty' => $count,
+        'vt' => ($visibilityTimeout ?? TimeSpan::fromSeconds(30))->toSeconds(),
+        'max_poll_seconds' => ($maxPoll ?? TimeSpan::fromSeconds(5))->toSeconds(),
+        'poll_interval_ms' => ($pollInterval ?? TimeSpan::fromMilliseconds(250))->toMilliseconds(),
+    ]);
+
+    foreach ($result as $row) {
+        yield Message::fromArray($row);
+    }
+}
+
+/**
+ * @api
+ * @param non-empty-string $queue
+ * @param positive-int $count
+ * @return iterable<int, Message>
+ */
+function readGroupedHead(
+    PostgresLink $pg,
+    string $queue,
+    int $count,
+    ?TimeSpan $visibilityTimeout = null,
+): iterable {
+    $result = $pg->execute('SELECT * FROM pgmq.read_grouped_head(:queue_name, :vt, :qty)', [
+        'queue_name' => $queue,
+        'qty' => $count,
+        'vt' => ($visibilityTimeout ?? TimeSpan::fromSeconds(30))->toSeconds(),
+    ]);
+
+    foreach ($result as $row) {
+        yield Message::fromArray($row);
+    }
+}
+
+/**
+ * @api
+ * @param non-empty-string $queue
+ */
+function createFifoIndex(
+    PostgresLink $pg,
+    string $queue,
+): void {
+    $pg->execute('SELECT * FROM pgmq.create_fifo_index(:queue_name)', [
+        'queue_name' => $queue,
+    ]);
+}
+
+/**
+ * @api
+ */
+function createFifoIndexAll(PostgresLink $pg): void
+{
+    $pg->execute('SELECT * FROM pgmq.create_fifo_indexes_all()');
+}
diff --git a/src/pgmq.php b/src/pgmq.php
index 674caf0..c07ec81 100644
--- a/src/pgmq.php
+++ b/src/pgmq.php
@@ -4,613 +4,7 @@
 
 namespace Thesis\Pgmq;
 
-use Amp\Postgres\PostgresConnection;
-use Amp\Postgres\PostgresLink;
-use Amp\Postgres\PostgresQueryError;
-use Thesis\Time\TimeSpan;
-
-/**
- * @api
- */
-function createExtension(PostgresLink $pg): void
-{
-    $pg->execute('CREATE EXTENSION IF NOT EXISTS pgmq CASCADE;');
-}
-
-/**
- * @api
- * @param non-empty-string $queue
- * @throws PostgresQueryError if queue name is invalid
- */
-function validateQueueName(
-    PostgresLink $pg,
-    string $queue,
-): void {
-    $pg->execute('SELECT pgmq.validate_queue_name(:queue_name);', [
-        'queue_name' => $queue,
-    ]);
-}
-
-/**
- * @api
- * @param non-empty-string $queue
- */
-function createQueue(
-    PostgresLink $pg,
-    string $queue,
-): Queue {
-    $pg->execute('SELECT pgmq.create(:queue_name)', [
-        'queue_name' => $queue,
-    ]);
-
-    return new Queue($queue, $pg);
-}
-
-/**
- * @api
- * @param non-empty-string $queue
- */
-function createUnloggedQueue(
-    PostgresLink $pg,
-    string $queue,
-): Queue {
-    $pg->execute('SELECT pgmq.create_unlogged(:queue_name)', [
-        'queue_name' => $queue,
-    ]);
-
-    return new Queue($queue, $pg);
-}
-
-/**
- * @param non-empty-string $queue
- * @throws QueueNotFound
- */
-function findQueue(
-    PostgresLink $pg,
-    string $queue,
-): Queue {
-    $md = queueMetadata($pg, $queue);
-
-    return new Queue($md->name, $pg);
-}
-
-/**
- * @api
- * @param non-empty-string $queue
- * @param non-negative-int|non-empty-string $partitionInterval this can be either any valid Postgres Duration supported by pg_partman, or an integer value
- * @param non-negative-int|non-empty-string $retentionInterval this can be either any valid Postgres Duration supported by pg_partman, or an integer value
- */
-function createPartitionedQueue(
-    PostgresLink $pg,
-    string $queue,
-    int|string $partitionInterval = 10_000,
-    int|string $retentionInterval = 100_000,
-): Queue {
-    $pg->execute('SELECT pgmq.create(:queue_name, :partition_interval, :retention_interval)', [
-        'queue_name' => $queue,
-        'partition_interval' => (string) $partitionInterval,
-        'retention_interval' => (string) $retentionInterval,
-    ]);
-
-    return new Queue($queue, $pg);
-}
-
-/**
- * @api
- * @return iterable<QueueMetadata>
- */
-function listQueueMetadata(PostgresLink $pg): iterable
-{
-    $result = $pg->query('SELECT queue_name, created_at, is_partitioned, is_unlogged FROM pgmq.list_queues()');
-
-    foreach ($result as $row) {
-        yield QueueMetadata::fromArray($row);
-    }
-}
-
-/**
- * @api
- * @return iterable<Queue>
- */
-function listQueues(PostgresLink $pg): iterable
-{
-    $result = $pg->query('SELECT queue_name FROM pgmq.list_queues()');
-
-    /** @var array{queue_name: non-empty-string} $row */
-    foreach ($result as $row) {
-        yield new Queue($row['queue_name'], $pg);
-    }
-}
-
-/**
- * @api
- * @param non-empty-string $queue
- */
-function dropQueue(
-    PostgresLink $pg,
-    string $queue,
-): bool {
-    /** @var array{drop_queue?: bool} $result */
-    $result = $pg
-        ->execute('SELECT pgmq.drop_queue(:queue_name)', [
-            'queue_name' => $queue,
-        ])
-        ->fetchRow() ?? [];
-
-    return $result['drop_queue'] ?? false;
-}
-
-/**
- * @api
- * @param non-empty-string $queue
- */
-function purgeQueue(
-    PostgresLink $pg,
-    string $queue,
-): int {
-    /** @var array{purge_queue?: non-negative-int} $result */
-    $result = $pg
-        ->execute('SELECT pgmq.purge_queue(:queue_name)', [
-            'queue_name' => $queue,
-        ])
-        ->fetchRow() ?? [];
-
-    return $result['purge_queue'] ?? 0;
-}
-
-/**
- * @api
- * @param non-empty-string $queue
- * @throws QueueNotFound
- */
-function queueMetrics(
-    PostgresLink $pg,
-    string $queue,
-): QueueMetric {
-    $result = $pg
-        ->execute('SELECT * FROM pgmq.metrics(:queue_name)', [
-            'queue_name' => $queue,
-        ])
-        ->fetchRow() ?? throw new QueueNotFound();
-
-    return QueueMetric::fromArray($result);
-}
-
-/**
- * @api
- * @param non-empty-string $queue
- * @throws QueueNotFound
- */
-function queueMetadata(
-    PostgresLink $pg,
-    string $queue,
-): QueueMetadata {
-    $result = $pg
-        ->execute('SELECT queue_name, created_at, is_partitioned, is_unlogged FROM pgmq.list_queues() WHERE queue_name = :queue_name', [
-            'queue_name' => $queue,
-        ])
-        ->fetchRow() ?? throw new QueueNotFound();
-
-    return QueueMetadata::fromArray($result);
-}
-
-/**
- * @api
- * @param non-empty-string $pattern
- * @param non-empty-string $queue
- */
-function bindTopic(
-    PostgresLink $pg,
-    string $pattern,
-    string $queue,
-): void {
-    $pg->execute('SELECT pgmq.bind_topic(:pattern, :queue_name);', [
-        'pattern' => $pattern,
-        'queue_name' => $queue,
-    ]);
-}
-
-/**
- * @api
- * @param non-empty-string $pattern
- * @param non-empty-string $queue
- */
-function unbindTopic(
-    PostgresLink $pg,
-    string $pattern,
-    string $queue,
-): void {
-    $pg->execute('SELECT pgmq.unbind_topic(:pattern, :queue_name);', [
-        'pattern' => $pattern,
-        'queue_name' => $queue,
-    ]);
-}
-
-/**
- * @api
- * @param non-empty-string $routingKey
- * @return non-negative-int
- */
-function sendTopic(
-    PostgresLink $pg,
-    string $routingKey,
-    SendMessage $message,
-    null|TimeSpan|\DateTimeImmutable $delay = null,
-): int {
-    $delay ??= TimeSpan::fromSeconds(0);
-
-    if ($delay instanceof \DateTimeImmutable) {
-        $delay = TimeSpan::diff($delay, new \DateTimeImmutable());
-    }
-
-    /** @var array{send_topic: non-negative-int} $result */
-    $result = $pg
-        ->execute('SELECT * FROM pgmq.send_topic(:routing_key, :msg, :headers, :delay::int)', [
-            'routing_key' => $routingKey,
-            'msg' => $message->valueJson,
-            'headers' => $message->headerJson,
-            'delay' => max((int) $delay->toSeconds(), 0),
-        ])
-        ->fetchRow() ?? throw new \RuntimeException("Failed to send message using routingKey {$routingKey}.");
-
-    return $result['send_topic'];
-}
-
-/**
- * @api
- * @param non-empty-string $routingKey
- * @return iterable<int, TopicRoute>
- */
-function testRouting(
-    PostgresLink $pg,
-    string $routingKey,
-): iterable {
-    $result = $pg->execute('SELECT * FROM pgmq.test_routing(:routing_key)', [
-        'routing_key' => $routingKey,
-    ]);
-
-    foreach ($result as $row) {
-        yield TopicRoute::fromArray($row);
-    }
-}
-
-/**
- * @api
- * @param non-empty-string $routingKey
- * @throws PostgresQueryError if routingKey is invalid
- */
-function validateRoutingKey(
-    PostgresLink $pg,
-    string $routingKey,
-): void {
-    $pg->execute('SELECT * FROM pgmq.validate_routing_key(:routing_key)', [
-        'routing_key' => $routingKey,
-    ]);
-}
-
-/**
- * @api
- * @param non-empty-string $pattern
- * @throws PostgresQueryError if topic pattern is invalid
- */
-function validateTopicPattern(
-    PostgresLink $pg,
-    string $pattern,
-): void {
-    $pg->execute('SELECT * FROM pgmq.validate_topic_pattern(:pattern)', [
-        'pattern' => $pattern,
-    ]);
-}
-
-/**
- * @api
- * @return iterable<QueueMetric>
- */
-function metrics(PostgresLink $pg): iterable
-{
-    $result = $pg->query('SELECT * FROM pgmq.metrics_all();');
-
-    foreach ($result as $row) {
-        yield QueueMetric::fromArray($row);
-    }
-}
-
-/**
- * @api
- * @param non-empty-string $queue
- * @return int the message id, unique to the queue, is returned
- */
-function send(
-    PostgresLink $pg,
-    string $queue,
-    SendMessage $message,
-    null|TimeSpan|\DateTimeImmutable $delay = null,
-): int {
-    $delay ??= TimeSpan::fromSeconds(0);
-
-    $sql = match (true) {
-        $delay instanceof TimeSpan => 'SELECT * FROM pgmq.send(:queue_name, :msg, :headers, :delay::int)',
-        default => 'SELECT * FROM pgmq.send(:queue_name, :msg, :headers, :delay::timestamptz)',
-    };
-
-    /** @var array{send: int} $result */
-    $result = $pg
-        ->execute($sql, [
-            'queue_name' => $queue,
-            'msg' => $message->valueJson,
-            'headers' => $message->headerJson,
-            'delay' => $delay instanceof TimeSpan ? (int) $delay->toSeconds() : $delay->format(\DateTimeInterface::RFC3339),
-        ])
-        ->fetchRow() ?? throw new \RuntimeException("Failed to send message to the queue {$queue}.");
-
-    return $result['send'];
-}
-
-/**
- * @api
- * @param non-empty-string $queue
- * @param non-empty-list<SendMessage> $messages
- * @return list<int>
- */
-function sendBatch(
-    PostgresLink $pg,
-    string $queue,
-    array $messages,
-    null|TimeSpan|\DateTimeImmutable $delay = null,
-): array {
-    $delay ??= TimeSpan::fromSeconds(0);
-
-    $sql = match (true) {
-        $delay instanceof TimeSpan => 'SELECT * FROM pgmq.send_batch(:queue_name, :msgs::jsonb[], :headers::jsonb[], :delay::int)',
-        default => 'SELECT * FROM pgmq.send_batch(:queue_name, :msgs::jsonb[], :headers::jsonb[], :delay::timestamptz)',
-    };
-
-    $result = $pg->execute($sql, [
-        'queue_name' => $queue,
-        'msgs' => array_map(static fn(SendMessage $message): string => $message->valueJson, $messages),
-        'headers' => array_map(static fn(SendMessage $message): ?string => $message->headerJson, $messages),
-        'delay' => $delay instanceof TimeSpan ? $delay->toSeconds() : $delay->format(\DateTimeInterface::RFC3339),
-    ]);
-
-    $messageIds = [];
-
-    /** @var array{send_batch: int} $row */
-    foreach ($result as $row) {
-        $messageIds[] = $row['send_batch'];
-    }
-
-    return $messageIds;
-}
-
-/**
- * @api
- * @param non-empty-string $queue
- * @param positive-int $batch
- * @return iterable<Message>
- */
-function readPoll(
-    PostgresLink $pg,
-    string $queue,
-    int $batch = 1,
-    ?TimeSpan $visibilityTimeout = null,
-    ?TimeSpan $maxPoll = null,
-    ?TimeSpan $pollInterval = null,
-): iterable {
-    $result = $pg->execute('SELECT * FROM pgmq.read_with_poll(:queue_name, :vt, :limit, :poll_timeout_s, :poll_interval_ms);', [
-        'queue_name' => $queue,
-        'vt' => ($visibilityTimeout ?? TimeSpan::fromSeconds(30))->toSeconds(),
-        'limit' => $batch,
-        'poll_timeout_s' => ($maxPoll ?? TimeSpan::fromSeconds(5))->toSeconds(),
-        'poll_interval_ms' => ($pollInterval ?? TimeSpan::fromMilliseconds(250))->toMilliseconds(),
-    ]);
-
-    foreach ($result as $row) {
-        yield Message::fromArray($row);
-    }
-}
-
-/**
- * @api
- * @param non-empty-string $queue
- */
-function read(
-    PostgresLink $pg,
-    string $queue,
-    ?TimeSpan $visibilityTimeout = null,
-): ?Message {
-    foreach (readBatch($pg, $queue, 1, $visibilityTimeout) as $message) {
-        return $message;
-    }
-
-    return null;
-}
-
-/**
- * @api
- * @param non-empty-string $queue
- * @param positive-int $count
- * @return iterable<int, Message>
- */
-function readBatch(
-    PostgresLink $pg,
-    string $queue,
-    int $count,
-    ?TimeSpan $visibilityTimeout = null,
-): iterable {
-    $visibilityTimeout ??= TimeSpan::fromSeconds(30);
-
-    $result = $pg->execute('SELECT * FROM pgmq.read(:queue_name, :vt, :limit)', [
-        'queue_name' => $queue,
-        'vt' => $visibilityTimeout->toSeconds(),
-        'limit' => $count,
-    ]);
-
-    foreach ($result as $row) {
-        yield Message::fromArray($row);
-    }
-}
-
-/**
- * @api
- * @param non-empty-string $queue
- */
-function pop(
-    PostgresLink $pg,
-    string $queue,
-): ?Message {
-    $row = $pg
-        ->execute('SELECT * FROM pgmq.pop(:queue_name)', [
-            'queue_name' => $queue,
-        ])
-        ->fetchRow();
-
-    return $row !== null ? Message::fromArray($row) : null;
-}
-
-/**
- * @api
- * @param non-empty-string $queue
- */
-function archive(
-    PostgresLink $pg,
-    string $queue,
-    int $messageId,
-): bool {
-    return \in_array($messageId, archiveBatch($pg, $queue, [$messageId]), true);
-}
-
-/**
- * @api
- * @param non-empty-string $queue
- * @param list<int> $messageIds
- * @return list<int>
- */
-function archiveBatch(
-    PostgresLink $pg,
-    string $queue,
-    array $messageIds,
-): array {
-    $result = $pg->execute('SELECT * FROM pgmq.archive(:queue_name, :msg_ids::bigint[])', [
-        'queue_name' => $queue,
-        'msg_ids' => $messageIds,
-    ]);
-
-    $archive = [];
-
-    /** @var array{archive: int} $row */
-    foreach ($result as $row) {
-        $archive[] = $row['archive'];
-    }
-
-    return $archive;
-}
-
-/**
- * @api
- * @param non-empty-string $queue
- */
-function delete(
-    PostgresLink $pg,
-    string $queue,
-    int $messageId,
-): bool {
-    return \in_array($messageId, deleteBatch($pg, $queue, [$messageId]), true);
-}
-
-/**
- * @api
- * @param non-empty-string $queue
- * @param list<int> $messageIds
- * @return list<int>
- */
-function deleteBatch(
-    PostgresLink $pg,
-    string $queue,
-    array $messageIds,
-): array {
-    $result = $pg->execute('SELECT pgmq.delete(:queue_name, :msg_ids::bigint[])', [
-        'queue_name' => $queue,
-        'msg_ids' => $messageIds,
-    ]);
-
-    $deleted = [];
-
-    /** @var array{delete: int} $row */
-    foreach ($result as $row) {
-        $deleted[] = $row['delete'];
-    }
-
-    return $deleted;
-}
-
-/**
- * @api
- * @param non-empty-string $queue
- * @param list<int> $messageIds
- */
-function setVisibilityTimeout(
-    PostgresLink $pg,
-    string $queue,
-    array $messageIds,
-    TimeSpan $visibilityTimeout,
-): ?Message {
-    $row = $pg
-        ->execute('SELECT * FROM pgmq.set_vt(:queue_name, :msg_ids::bigint[], :vt::int)', [
-            'queue_name' => $queue,
-            'msg_ids' => $messageIds,
-            'vt' => $visibilityTimeout->toSeconds(),
-        ])
-        ->fetchRow();
-
-    return $row !== null ? Message::fromArray($row) : null;
-}
-
-/**
- * @api
- * @param non-empty-string $queue
- * @return non-empty-string
- */
-function enableNotifyInsert(
-    PostgresLink $pg,
-    string $queue,
-    ?TimeSpan $throttleInterval = null,
-): string {
-    $pg->execute('SELECT pgmq.enable_notify_insert(:queue_name, :throttle_interval_ms)', [
-        'queue_name' => $queue,
-        'throttle_interval_ms' => ($throttleInterval ?? TimeSpan::fromMilliseconds(30))->toMilliseconds(),
-    ]);
-
-    return channelName($queue);
-}
-
-/**
- * @api
- * @param non-empty-string $queue
- */
-function disableNotifyInsert(
-    PostgresLink $pg,
-    string $queue,
-): void {
-    $pg->execute('SELECT pgmq.disable_notify_insert(:queue_name)', [
-        'queue_name' => $queue,
-    ]);
-}
-
-/**
- * @api
- * @param non-empty-string $queue
- * @return non-empty-string
- */
-function channelName(string $queue): string
-{
-    return "pgmq.q_{$queue}.INSERT";
-}
-
-/**
- * @api
- */
-function createConsumer(
-    PostgresConnection $pg,
-): Consumer {
-    return new Consumer($pg);
-}
+require_once __DIR__ . '/extension.php';
+require_once __DIR__ . '/queue.php';
+require_once __DIR__ . '/topic.php';
+require_once __DIR__ . '/fifo.php';
diff --git a/src/queue.php b/src/queue.php
new file mode 100644
index 0000000..27bc88a
--- /dev/null
+++ b/src/queue.php
@@ -0,0 +1,500 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Thesis\Pgmq;
+
+use Amp\Postgres\PostgresConnection;
+use Amp\Postgres\PostgresLink;
+use Amp\Postgres\PostgresQueryError;
+use Thesis\Time\TimeSpan;
+
+/**
+ * @api
+ * @param non-empty-string $queue
+ * @throws PostgresQueryError if queue name is invalid
+ */
+function validateQueueName(
+    PostgresLink $pg,
+    string $queue,
+): void {
+    $pg->execute('SELECT pgmq.validate_queue_name(:queue_name);', [
+        'queue_name' => $queue,
+    ]);
+}
+
+/**
+ * @api
+ * @param non-empty-string $queue
+ */
+function createQueue(
+    PostgresLink $pg,
+    string $queue,
+): Queue {
+    $pg->execute('SELECT pgmq.create(:queue_name)', [
+        'queue_name' => $queue,
+    ]);
+
+    return new Queue($queue, $pg);
+}
+
+/**
+ * @api
+ * @param non-empty-string $queue
+ */
+function createUnloggedQueue(
+    PostgresLink $pg,
+    string $queue,
+): Queue {
+    $pg->execute('SELECT pgmq.create_unlogged(:queue_name)', [
+        'queue_name' => $queue,
+    ]);
+
+    return new Queue($queue, $pg);
+}
+
+/**
+ * @param non-empty-string $queue
+ * @throws QueueNotFound
+ */
+function findQueue(
+    PostgresLink $pg,
+    string $queue,
+): Queue {
+    $md = queueMetadata($pg, $queue);
+
+    return new Queue($md->name, $pg);
+}
+
+/**
+ * @api
+ * @param non-empty-string $queue
+ * @param non-negative-int|non-empty-string $partitionInterval this can be either any valid Postgres Duration supported by pg_partman, or an integer value
+ * @param non-negative-int|non-empty-string $retentionInterval this can be either any valid Postgres Duration supported by pg_partman, or an integer value
+ */
+function createPartitionedQueue(
+    PostgresLink $pg,
+    string $queue,
+    int|string $partitionInterval = 10_000,
+    int|string $retentionInterval = 100_000,
+): Queue {
+    $pg->execute('SELECT pgmq.create(:queue_name, :partition_interval, :retention_interval)', [
+        'queue_name' => $queue,
+        'partition_interval' => (string) $partitionInterval,
+        'retention_interval' => (string) $retentionInterval,
+    ]);
+
+    return new Queue($queue, $pg);
+}
+
+/**
+ * @api
+ * @return iterable<QueueMetadata>
+ */
+function listQueueMetadata(PostgresLink $pg): iterable
+{
+    $result = $pg->query('SELECT queue_name, created_at, is_partitioned, is_unlogged FROM pgmq.list_queues()');
+
+    foreach ($result as $row) {
+        yield QueueMetadata::fromArray($row);
+    }
+}
+
+/**
+ * @api
+ * @return iterable<Queue>
+ */
+function listQueues(PostgresLink $pg): iterable
+{
+    $result = $pg->query('SELECT queue_name FROM pgmq.list_queues()');
+
+    /** @var array{queue_name: non-empty-string} $row */
+    foreach ($result as $row) {
+        yield new Queue($row['queue_name'], $pg);
+    }
+}
+
+/**
+ * @api
+ * @param non-empty-string $queue
+ */
+function dropQueue(
+    PostgresLink $pg,
+    string $queue,
+): bool {
+    /** @var array{drop_queue?: bool} $result */
+    $result = $pg
+        ->execute('SELECT pgmq.drop_queue(:queue_name)', [
+            'queue_name' => $queue,
+        ])
+        ->fetchRow() ?? [];
+
+    return $result['drop_queue'] ?? false;
+}
+
+/**
+ * @api
+ * @param non-empty-string $queue
+ */
+function purgeQueue(
+    PostgresLink $pg,
+    string $queue,
+): int {
+    /** @var array{purge_queue?: non-negative-int} $result */
+    $result = $pg
+        ->execute('SELECT pgmq.purge_queue(:queue_name)', [
+            'queue_name' => $queue,
+        ])
+        ->fetchRow() ?? [];
+
+    return $result['purge_queue'] ?? 0;
+}
+
+/**
+ * @api
+ * @param non-empty-string $queue
+ * @throws QueueNotFound
+ */
+function queueMetrics(
+    PostgresLink $pg,
+    string $queue,
+): QueueMetric {
+    $result = $pg
+        ->execute('SELECT * FROM pgmq.metrics(:queue_name)', [
+            'queue_name' => $queue,
+        ])
+        ->fetchRow() ?? throw new QueueNotFound();
+
+    return QueueMetric::fromArray($result);
+}
+
+/**
+ * @api
+ * @param non-empty-string $queue
+ * @throws QueueNotFound
+ */
+function queueMetadata(
+    PostgresLink $pg,
+    string $queue,
+): QueueMetadata {
+    $result = $pg
+        ->execute('SELECT queue_name, created_at, is_partitioned, is_unlogged FROM pgmq.list_queues() WHERE queue_name = :queue_name', [
+            'queue_name' => $queue,
+        ])
+        ->fetchRow() ?? throw new QueueNotFound();
+
+    return QueueMetadata::fromArray($result);
+}
+
+/**
+ * @api
+ * @return iterable<QueueMetric>
+ */
+function metrics(PostgresLink $pg): iterable
+{
+    $result = $pg->query('SELECT * FROM pgmq.metrics_all();');
+
+    foreach ($result as $row) {
+        yield QueueMetric::fromArray($row);
+    }
+}
+
+/**
+ * @api
+ * @param non-empty-string $queue
+ * @return int the message id, unique to the queue, is returned
+ */
+function send(
+    PostgresLink $pg,
+    string $queue,
+    SendMessage $message,
+    null|TimeSpan|\DateTimeImmutable $delay = null,
+): int {
+    $delay ??= TimeSpan::fromSeconds(0);
+
+    $sql = match (true) {
+        $delay instanceof TimeSpan => 'SELECT * FROM pgmq.send(:queue_name, :msg, :headers, :delay::int)',
+        default => 'SELECT * FROM pgmq.send(:queue_name, :msg, :headers, :delay::timestamptz)',
+    };
+
+    /** @var array{send: int} $result */
+    $result = $pg
+        ->execute($sql, [
+            'queue_name' => $queue,
+            'msg' => $message->valueJson,
+            'headers' => $message->headerJson,
+            'delay' => $delay instanceof TimeSpan ? (int) $delay->toSeconds() : $delay->format(\DateTimeInterface::RFC3339),
+        ])
+        ->fetchRow() ?? throw new \RuntimeException("Failed to send message to the queue {$queue}.");
+
+    return $result['send'];
+}
+
+/**
+ * @api
+ * @param non-empty-string $queue
+ * @param non-empty-list<SendMessage> $messages
+ * @return list<int>
+ */
+function sendBatch(
+    PostgresLink $pg,
+    string $queue,
+    array $messages,
+    null|TimeSpan|\DateTimeImmutable $delay = null,
+): array {
+    $delay ??= TimeSpan::fromSeconds(0);
+
+    $sql = match (true) {
+        $delay instanceof TimeSpan => 'SELECT * FROM pgmq.send_batch(:queue_name, :msgs::jsonb[], :headers::jsonb[], :delay::int)',
+        default => 'SELECT * FROM pgmq.send_batch(:queue_name, :msgs::jsonb[], :headers::jsonb[], :delay::timestamptz)',
+    };
+
+    $result = $pg->execute($sql, [
+        'queue_name' => $queue,
+        'msgs' => array_map(static fn(SendMessage $message): string => $message->valueJson, $messages),
+        'headers' => array_map(static fn(SendMessage $message): ?string => $message->headerJson, $messages),
+        'delay' => $delay instanceof TimeSpan ? $delay->toSeconds() : $delay->format(\DateTimeInterface::RFC3339),
+    ]);
+
+    $messageIds = [];
+
+    /** @var array{send_batch: int} $row */
+    foreach ($result as $row) {
+        $messageIds[] = $row['send_batch'];
+    }
+
+    return $messageIds;
+}
+
+/**
+ * @api
+ * @param non-empty-string $queue
+ * @param positive-int $batch
+ * @return iterable<Message>
+ */
+function readPoll(
+    PostgresLink $pg,
+    string $queue,
+    int $batch = 1,
+    ?TimeSpan $visibilityTimeout = null,
+    ?TimeSpan $maxPoll = null,
+    ?TimeSpan $pollInterval = null,
+): iterable {
+    $result = $pg->execute('SELECT * FROM pgmq.read_with_poll(:queue_name, :vt, :limit, :poll_timeout_s, :poll_interval_ms);', [
+        'queue_name' => $queue,
+        'vt' => ($visibilityTimeout ?? TimeSpan::fromSeconds(30))->toSeconds(),
+        'limit' => $batch,
+        'poll_timeout_s' => ($maxPoll ?? TimeSpan::fromSeconds(5))->toSeconds(),
+        'poll_interval_ms' => ($pollInterval ?? TimeSpan::fromMilliseconds(250))->toMilliseconds(),
+    ]);
+
+    foreach ($result as $row) {
+        yield Message::fromArray($row);
+    }
+}
+
+/**
+ * @api
+ * @param non-empty-string $queue
+ */
+function read(
+    PostgresLink $pg,
+    string $queue,
+    ?TimeSpan $visibilityTimeout = null,
+): ?Message {
+    foreach (readBatch($pg, $queue, 1, $visibilityTimeout) as $message) {
+        return $message;
+    }
+
+    return null;
+}
+
+/**
+ * @api
+ * @param non-empty-string $queue
+ * @param positive-int $count
+ * @return iterable<int, Message>
+ */
+function readBatch(
+    PostgresLink $pg,
+    string $queue,
+    int $count,
+    ?TimeSpan $visibilityTimeout = null,
+): iterable {
+    $visibilityTimeout ??= TimeSpan::fromSeconds(30);
+
+    $result = $pg->execute('SELECT * FROM pgmq.read(:queue_name, :vt, :limit)', [
+        'queue_name' => $queue,
+        'vt' => $visibilityTimeout->toSeconds(),
+        'limit' => $count,
+    ]);
+
+    foreach ($result as $row) {
+        yield Message::fromArray($row);
+    }
+}
+
+/**
+ * @api
+ * @param non-empty-string $queue
+ */
+function pop(
+    PostgresLink $pg,
+    string $queue,
+): ?Message {
+    $row = $pg
+        ->execute('SELECT * FROM pgmq.pop(:queue_name)', [
+            'queue_name' => $queue,
+        ])
+        ->fetchRow();
+
+    return $row !== null ? Message::fromArray($row) : null;
+}
+
+/**
+ * @api
+ * @param non-empty-string $queue
+ */
+function archive(
+    PostgresLink $pg,
+    string $queue,
+    int $messageId,
+): bool {
+    return \in_array($messageId, archiveBatch($pg, $queue, [$messageId]), true);
+}
+
+/**
+ * @api
+ * @param non-empty-string $queue
+ * @param list<int> $messageIds
+ * @return list<int>
+ */
+function archiveBatch(
+    PostgresLink $pg,
+    string $queue,
+    array $messageIds,
+): array {
+    $result = $pg->execute('SELECT * FROM pgmq.archive(:queue_name, :msg_ids::bigint[])', [
+        'queue_name' => $queue,
+        'msg_ids' => $messageIds,
+    ]);
+
+    $archive = [];
+
+    /** @var array{archive: int} $row */
+    foreach ($result as $row) {
+        $archive[] = $row['archive'];
+    }
+
+    return $archive;
+}
+
+/**
+ * @api
+ * @param non-empty-string $queue
+ */
+function delete(
+    PostgresLink $pg,
+    string $queue,
+    int $messageId,
+): bool {
+    return \in_array($messageId, deleteBatch($pg, $queue, [$messageId]), true);
+}
+
+/**
+ * @api
+ * @param non-empty-string $queue
+ * @param list<int> $messageIds
+ * @return list<int>
+ */
+function deleteBatch(
+    PostgresLink $pg,
+    string $queue,
+    array $messageIds,
+): array {
+    $result = $pg->execute('SELECT pgmq.delete(:queue_name, :msg_ids::bigint[])', [
+        'queue_name' => $queue,
+        'msg_ids' => $messageIds,
+    ]);
+
+    $deleted = [];
+
+    /** @var array{delete: int} $row */
+    foreach ($result as $row) {
+        $deleted[] = $row['delete'];
+    }
+
+    return $deleted;
+}
+
+/**
+ * @api
+ * @param non-empty-string $queue
+ * @param list<int> $messageIds
+ */
+function setVisibilityTimeout(
+    PostgresLink $pg,
+    string $queue,
+    array $messageIds,
+    TimeSpan $visibilityTimeout,
+): ?Message {
+    $row = $pg
+        ->execute('SELECT * FROM pgmq.set_vt(:queue_name, :msg_ids::bigint[], :vt::int)', [
+            'queue_name' => $queue,
+            'msg_ids' => $messageIds,
+            'vt' => $visibilityTimeout->toSeconds(),
+        ])
+        ->fetchRow();
+
+    return $row !== null ? Message::fromArray($row) : null;
+}
+
+/**
+ * @api
+ * @param non-empty-string $queue
+ * @return non-empty-string
+ */
+function enableNotifyInsert(
+    PostgresLink $pg,
+    string $queue,
+    ?TimeSpan $throttleInterval = null,
+): string {
+    $pg->execute('SELECT pgmq.enable_notify_insert(:queue_name, :throttle_interval_ms)', [
+        'queue_name' => $queue,
+        'throttle_interval_ms' => ($throttleInterval ?? TimeSpan::fromMilliseconds(30))->toMilliseconds(),
+    ]);
+
+    return channelName($queue);
+}
+
+/**
+ * @api
+ * @param non-empty-string $queue
+ */
+function disableNotifyInsert(
+    PostgresLink $pg,
+    string $queue,
+): void {
+    $pg->execute('SELECT pgmq.disable_notify_insert(:queue_name)', [
+        'queue_name' => $queue,
+    ]);
+}
+
+/**
+ * @api
+ * @param non-empty-string $queue
+ * @return non-empty-string
+ */
+function channelName(string $queue): string
+{
+    return "pgmq.q_{$queue}.INSERT";
+}
+
+/**
+ * @api
+ */
+function createConsumer(
+    PostgresConnection $pg,
+): Consumer {
+    return new Consumer($pg);
+}
diff --git a/src/topic.php b/src/topic.php
new file mode 100644
index 0000000..206b303
--- /dev/null
+++ b/src/topic.php
@@ -0,0 +1,117 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Thesis\Pgmq;
+
+use Amp\Postgres\PostgresLink;
+use Amp\Postgres\PostgresQueryError;
+use Thesis\Time\TimeSpan;
+
+/**
+ * @api
+ * @param non-empty-string $pattern
+ * @param non-empty-string $queue
+ */
+function bindTopic(
+    PostgresLink $pg,
+    string $pattern,
+    string $queue,
+): void {
+    $pg->execute('SELECT pgmq.bind_topic(:pattern, :queue_name);', [
+        'pattern' => $pattern,
+        'queue_name' => $queue,
+    ]);
+}
+
+/**
+ * @api
+ * @param non-empty-string $pattern
+ * @param non-empty-string $queue
+ */
+function unbindTopic(
+    PostgresLink $pg,
+    string $pattern,
+    string $queue,
+): void {
+    $pg->execute('SELECT pgmq.unbind_topic(:pattern, :queue_name);', [
+        'pattern' => $pattern,
+        'queue_name' => $queue,
+    ]);
+}
+
+/**
+ * @api
+ * @param non-empty-string $routingKey
+ * @return non-negative-int
+ */
+function sendTopic(
+    PostgresLink $pg,
+    string $routingKey,
+    SendMessage $message,
+    null|TimeSpan|\DateTimeImmutable $delay = null,
+): int {
+    $delay ??= TimeSpan::fromSeconds(0);
+
+    if ($delay instanceof \DateTimeImmutable) {
+        $delay = TimeSpan::diff($delay, new \DateTimeImmutable());
+    }
+
+    /** @var array{send_topic: non-negative-int} $result */
+    $result = $pg
+        ->execute('SELECT * FROM pgmq.send_topic(:routing_key, :msg, :headers, :delay::int)', [
+            'routing_key' => $routingKey,
+            'msg' => $message->valueJson,
+            'headers' => $message->headerJson,
+            'delay' => max((int) $delay->toSeconds(), 0),
+        ])
+        ->fetchRow() ?? throw new \RuntimeException("Failed to send message using routingKey {$routingKey}.");
+
+    return $result['send_topic'];
+}
+
+/**
+ * @api
+ * @param non-empty-string $routingKey
+ * @return iterable<int, TopicRoute>
+ */
+function testRouting(
+    PostgresLink $pg,
+    string $routingKey,
+): iterable {
+    $result = $pg->execute('SELECT * FROM pgmq.test_routing(:routing_key)', [
+        'routing_key' => $routingKey,
+    ]);
+
+    foreach ($result as $row) {
+        yield TopicRoute::fromArray($row);
+    }
+}
+
+/**
+ * @api
+ * @param non-empty-string $routingKey
+ * @throws PostgresQueryError if routingKey is invalid
+ */
+function validateRoutingKey(
+    PostgresLink $pg,
+    string $routingKey,
+): void {
+    $pg->execute('SELECT * FROM pgmq.validate_routing_key(:routing_key)', [
+        'routing_key' => $routingKey,
+    ]);
+}
+
+/**
+ * @api
+ * @param non-empty-string $pattern
+ * @throws PostgresQueryError if topic pattern is invalid
+ */
+function validateTopicPattern(
+    PostgresLink $pg,
+    string $pattern,
+): void {
+    $pg->execute('SELECT * FROM pgmq.validate_topic_pattern(:pattern)', [
+        'pattern' => $pattern,
+    ]);
+}
diff --git a/tests/PgmqTest.php b/tests/PgmqTest.php
index a88ec63..e1461d7 100644
--- a/tests/PgmqTest.php
+++ b/tests/PgmqTest.php
@@ -501,6 +501,123 @@ public function testTestRouting(): void
         }
     }
 
+    public function testReadGrouped(): void
+    {
+        $queue = createQueue($this->pg, $this->randomQueueName());
+        $queue->createFifoIndex();
+
+        $queue->send(new SendMessage(self::TESTING_MESSAGE, '{"x-pgmq-group": "a"}'));
+        $queue->send(new SendMessage(self::TESTING_MESSAGE, '{"x-pgmq-group": "a"}'));
+        $queue->send(new SendMessage(self::TESTING_MESSAGE, '{"x-pgmq-group": "b"}'));
+
+        $messages = [...$queue->readGrouped(10)];
+
+        self::assertCount(3, $messages);
+
+        $groups = array_map(
+            static fn(Message $message) => self::findHeader($message, 'x-pgmq-group', \strval(...)),
+            $messages,
+        );
+
+        self::assertContains('a', $groups);
+        self::assertContains('b', $groups);
+    }
+
+    public function testReadGroupedRR(): void
+    {
+        $queue = createQueue($this->pg, $this->randomQueueName());
+        $queue->createFifoIndex();
+
+        $queue->send(new SendMessage('{"order": 1}', '{"x-pgmq-group": "a"}'));
+        $queue->send(new SendMessage('{"order": 2}', '{"x-pgmq-group": "a"}'));
+        $queue->send(new SendMessage('{"order": 3}', '{"x-pgmq-group": "b"}'));
+        $queue->send(new SendMessage('{"order": 4}', '{"x-pgmq-group": "b"}'));
+
+        $messages = [...$queue->readGroupedRR(10)];
+
+        self::assertCount(4, $messages);
+
+        $groups = array_map(
+            static fn(Message $message) => self::findHeader($message, 'x-pgmq-group', \strval(...)),
+            $messages,
+        );
+
+        self::assertContains('a', $groups);
+        self::assertContains('b', $groups);
+    }
+
+    public function testReadGroupedHead(): void
+    {
+        $queue = createQueue($this->pg, $this->randomQueueName());
+        $queue->createFifoIndex();
+
+        $queue->send(new SendMessage('{"order": 1}', '{"x-pgmq-group": "a"}'));
+        $queue->send(new SendMessage('{"order": 2}', '{"x-pgmq-group": "a"}'));
+        $queue->send(new SendMessage('{"order": 3}', '{"x-pgmq-group": "b"}'));
+
+        $messages = [...$queue->readGroupedHead(10)];
+
+        self::assertCount(2, $messages);
+
+        $groups = array_map(
+            static fn(Message $message) => self::findHeader($message, 'x-pgmq-group', \strval(...)),
+            $messages,
+        );
+
+        self::assertContains('a', $groups);
+        self::assertContains('b', $groups);
+    }
+
+    public function testReadGroupedWithPoll(): void
+    {
+        $queue = createQueue($this->pg, $this->randomQueueName());
+        $queue->createFifoIndex();
+
+        $queue->send(new SendMessage(self::TESTING_MESSAGE, '{"x-pgmq-group": "a"}'));
+        $queue->send(new SendMessage(self::TESTING_MESSAGE, '{"x-pgmq-group": "b"}'));
+
+        $messages = [...$queue->readGroupedWithPoll(
+            count: 10,
+            maxPoll: TimeSpan::fromSeconds(1),
+            pollInterval: TimeSpan::fromMilliseconds(100),
+        )];
+
+        self::assertCount(2, $messages);
+    }
+
+    public function testReadGroupedRRWithPoll(): void
+    {
+        $queue = createQueue($this->pg, $this->randomQueueName());
+        $queue->createFifoIndex();
+
+        $queue->send(new SendMessage(self::TESTING_MESSAGE, '{"x-pgmq-group": "a"}'));
+        $queue->send(new SendMessage(self::TESTING_MESSAGE, '{"x-pgmq-group": "b"}'));
+
+        $messages = [...$queue->readGroupedRRWithPoll(
+            count: 10,
+            maxPoll: TimeSpan::fromSeconds(1),
+            pollInterval: TimeSpan::fromMilliseconds(100),
+        )];
+
+        self::assertCount(2, $messages);
+    }
+
+    #[DoesNotPerformAssertions]
+    public function testCreateFifoIndex(): void
+    {
+        $queue = createQueue($this->pg, $this->randomQueueName());
+        $queue->createFifoIndex();
+    }
+
+    #[DoesNotPerformAssertions]
+    public function testCreateFifoIndexAll(): void
+    {
+        createQueue($this->pg, $this->randomQueueName());
+        createQueue($this->pg, $this->randomQueueName());
+
+        createFifoIndexAll($this->pg);
+    }
+
     public function testValidateRoutingKey(): void
     {
         validateRoutingKey($this->pg, 'events.created');
@@ -527,4 +644,27 @@ private function randomQueueName(): string
         /** @var non-empty-string */
         return substr(bin2hex(random_bytes(30)), 0, length: 30);
     }
+
+    /**
+     * @template T
+     * @param non-empty-string $header
+     * @param callable(scalar): T $coerce
+     * @param T $default
+     * @return ($default is null ? (?T) : T)
+     */
+    private static function findHeader(
+        Message $message,
+        string $header,
+        callable $coerce,
+        mixed $default = null,
+    ): mixed {
+        /** @var array<non-empty-string, scalar> $headers */
+        $headers = json_decode($message->headers ?? '{}', true);
+
+        if (isset($headers[$header])) {
+            return $coerce($headers[$header]);
+        }
+
+        return $default;
+    }
 }