Implement AMQP like topics

.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.8.0
+            ghcr.io/pgmq/pg18-pgmq:v1.11.1
       - uses: shivammathur/setup-php@v2
         with:
           php-version: ${{ matrix.php }}

README.md

@@ -41,6 +41,13 @@ And only the consumer accepts `Amp\Postgres\PostgresConnection`, because it itse
  - [Delete batch](#delete-batch)
  - [Enable notify insert](#enable-notify-insert)
  - [Disable notify insert](#disable-notify-insert)
+ - [Bind topic](#bind-topic)
+ - [Unbind topic](#unbind-topic)
+ - [Send topic](#send-topic)
+ - [Send topic with delay](#send-topic-with-delay)
+ - [Test routing](#test-routing)
+ - [Validate routing key](#validate-routing-key)
+ - [Validate topic pattern](#validate-topic-pattern)
  - [Consume messages](#consume-messages)
 
 ### Create queue
@@ -408,6 +415,100 @@ $queue = Pgmq\createQueue($pg, 'events');
 $queue->disableNotifyInsert();
 ```
 
+### Bind topic
+
+Bind a queue to a topic pattern. Messages sent with a routing key matching the pattern will be delivered to the queue.
+
+```php
+use Thesis\Pgmq;
+use Amp\Postgres;
+
+$pg = new Postgres\PostgresConnectionPool(Postgres\PostgresConfig::fromString(''));
+
+$queue = Pgmq\createQueue($pg, 'emails');
+Pgmq\bindTopic($pg, 'notifications.*', $queue->name);
+```
+
+### Unbind topic
+
+Remove a queue binding from a topic pattern.
+
+```php
+use Thesis\Pgmq;
+use Amp\Postgres;
+
+$pg = new Postgres\PostgresConnectionPool(Postgres\PostgresConfig::fromString(''));
+
+Pgmq\unbindTopic($pg, 'notifications.*', 'emails');
+```
+
+### Send topic
+
+Send a message to all queues bound to patterns matching the given routing key.
+
+```php
+use Thesis\Pgmq;
+use Amp\Postgres;
+
+$pg = new Postgres\PostgresConnectionPool(Postgres\PostgresConfig::fromString(''));
+
+$matched = Pgmq\sendTopic($pg, 'notifications.email', new Pgmq\SendMessage('{"user": 1}'));
+```
+
+### Send topic with delay
+
+```php
+use Thesis\Pgmq;
+use Amp\Postgres;
+use Thesis\Time\TimeSpan;
+
+$pg = new Postgres\PostgresConnectionPool(Postgres\PostgresConfig::fromString(''));
+
+$matched = Pgmq\sendTopic(
+    $pg,
+    'notifications.email',
+    new Pgmq\SendMessage('{"user": 1}'),
+    TimeSpan::fromSeconds(5),
+);
+```
+
+### Test routing
+
+Test which queues would receive a message for a given routing key without actually sending a message.
+
+```php
+use Thesis\Pgmq;
+use Amp\Postgres;
+
+$pg = new Postgres\PostgresConnectionPool(Postgres\PostgresConfig::fromString(''));
+
+foreach (Pgmq\testRouting($pg, 'notifications.email') as $route) {
+    var_dump($route->pattern, $route->queue, $route->compiledRegex);
+}
+```
+
+### Validate routing key
+
+```php
+use Thesis\Pgmq;
+use Amp\Postgres;
+
+$pg = new Postgres\PostgresConnectionPool(Postgres\PostgresConfig::fromString(''));
+
+Pgmq\validateRoutingKey($pg, 'events.created'); // throws PostgresQueryError if invalid
+```
+
+### Validate topic pattern
+
+```php
+use Thesis\Pgmq;
+use Amp\Postgres;
+
+$pg = new Postgres\PostgresConnectionPool(Postgres\PostgresConfig::fromString(''));
+
+Pgmq\validateTopicPattern($pg, 'events.*'); // throws PostgresQueryError if invalid
+```
+
 ### 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.

docker-compose.yaml

@@ -7,7 +7,7 @@ services:
     command: sh -c 'trap "exit 0" TERM; tail -f /dev/null & wait'
 
   pgmq:
-    image: ghcr.io/pgmq/pg18-pgmq:v1.8.0
+    image: ghcr.io/pgmq/pg18-pgmq:v1.11.1
     environment:
       - POSTGRES_PASSWORD=postgres
     ports:

examples/topic.php

@@ -0,0 +1,36 @@
+<?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);
+
+$emailQueue = Pgmq\createQueue($pg, 'emails');
+$smsQueue = Pgmq\createQueue($pg, 'sms');
+
+Pgmq\bindTopic($pg, 'notifications.*', $emailQueue->name);
+Pgmq\bindTopic($pg, 'notifications.urgent', $smsQueue->name);
+
+// Route to all queues matching the pattern
+Pgmq\sendTopic($pg, 'notifications.urgent', new Pgmq\SendMessage('{"user": 1, "text": "Server is down"}'));
+
+dump($emailQueue->pop());
+dump($smsQueue->pop());
+
+// Route only to email queue
+Pgmq\sendTopic($pg, 'notifications.welcome', new Pgmq\SendMessage('{"user": 2, "text": "Welcome!"}'));
+
+dump($emailQueue->pop());
+dump($smsQueue->pop());
+
+Pgmq\unbindTopic($pg, 'notifications.*', $emailQueue->name);
+Pgmq\unbindTopic($pg, 'notifications.urgent', $smsQueue->name);
+
+$emailQueue->drop();
+$smsQueue->drop();

src/TopicRoute.php

@@ -0,0 +1,41 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Thesis\Pgmq;
+
+/**
+ * @api
+ * @phpstan-type RawRoute = array{
+ *      pattern: non-empty-string,
+ *      queue_name: non-empty-string,
+ *      compiled_regex: non-empty-string,
+ *  }
+ */
+final readonly class TopicRoute
+{
+    /**
+     * @internal
+     * @param array<array-key, mixed> $row
+     */
+    public static function fromArray(array $row): self
+    {
+        /** @var RawRoute $row */
+        return new self(
+            pattern: $row['pattern'],
+            queue: $row['queue_name'],
+            compiledRegex: $row['compiled_regex'],
+        );
+    }
+
+    /**
+     * @param non-empty-string $pattern
+     * @param non-empty-string $queue
+     * @param non-empty-string $compiledRegex
+     */
+    public function __construct(
+        public string $pattern,
+        public string $queue,
+        public string $compiledRegex,
+    ) {}
+}

src/pgmq.php

@@ -194,6 +194,114 @@ function queueMetadata(
     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>