feature: postgresql symfony messenger bridge (#2298)

- added postgresql symfony messenger bridge
- added listen/unlisen to postgresql client
- added postgresql symfony messenger bridge to postgersql symfony bundle

Author: norberttech

.codecov.yaml

@@ -108,6 +108,10 @@ component_management:
       name: bridge-symfony-postgresql-bundle
       paths:
         - src/bridge/symfony/postgresql-bundle/**
+    - component_id: bridge-symfony-postgresql-messenger
+      name: bridge-symfony-postgresql-messenger
+      paths:
+        - src/bridge/symfony/postgresql-messenger/**
     - component_id: bridge-symfony-telemetry-bundle
       name: bridge-symfony-telemetry-bundle
       paths:

.github/workflows/monorepo-split.yml

@@ -98,6 +98,8 @@ jobs:
             split_repository: 'symfony-http-foundation-telemetry-bridge'
           - local_path: 'src/bridge/symfony/postgresql-bundle'
             split_repository: 'symfony-postgresql-bundle'
+          - local_path: 'src/bridge/symfony/postgresql-messenger'
+            split_repository: 'symfony-postgresql-messenger-bridge'
           - local_path: 'src/bridge/symfony/telemetry-bundle'
             split_repository: 'symfony-telemetry-bundle'
           - local_path: 'src/bridge/telemetry/otlp'

.nix/php/lib/php.ini.dist

@@ -8,3 +8,5 @@ upload_max_filesize = 150M
 file_uploads = On
 max_file_uploads = 20
 short_open_tag = off
+opcache.enable=1
+opcache.enable_cli=0

composer.json

@@ -111,6 +111,7 @@
         "flow-php/symfony-http-foundation-bridge": "self.version",
         "flow-php/symfony-http-foundation-telemetry-bridge": "self.version",
         "flow-php/symfony-postgresql-bundle": "self.version",
+        "flow-php/symfony-postgresql-messenger-bridge": "self.version",
         "flow-php/symfony-telemetry-bundle": "self.version",
         "flow-php/telemetry": "self.version",
         "flow-php/telemetry-otlp-bridge": "self.version",
@@ -148,6 +149,7 @@
                 "src/bridge/symfony/http-foundation-telemetry/src/Flow",
                 "src/bridge/symfony/http-foundation/src/Flow",
                 "src/bridge/symfony/postgresql-bundle/src/Flow",
+                "src/bridge/symfony/postgresql-messenger/src/Flow",
                 "src/bridge/symfony/telemetry-bundle/src/Flow",
                 "src/bridge/telemetry/otlp/src/Flow",
                 "src/bridge/phpunit/telemetry/src/Flow",
@@ -248,6 +250,7 @@
                 "src/bridge/symfony/http-foundation-telemetry/tests/Flow",
                 "src/bridge/symfony/http-foundation/tests/Flow",
                 "src/bridge/symfony/postgresql-bundle/tests/Flow",
+                "src/bridge/symfony/postgresql-messenger/tests/Flow",
                 "src/bridge/symfony/telemetry-bundle/tests/Flow",
                 "src/bridge/telemetry/otlp/tests/Flow",
                 "src/bridge/phpunit/telemetry/tests/Flow",
@@ -325,6 +328,7 @@
             "@test:bridge:symfony-http-foundation",
             "@test:bridge:symfony-http-foundation-telemetry",
             "@test:bridge:symfony-postgresql-bundle",
+            "@test:bridge:symfony-postgresql-messenger",
             "@test:bridge:symfony-telemetry-bundle",
             "@test:bridge:telemetry-otlp"
         ],
@@ -430,6 +434,10 @@
             "tools/phpunit/vendor/bin/phpunit --testsuite=bridge-symfony-postgresql-bundle-unit --log-junit ./var/phpunit/logs/bridge-symfony-postgresql-bundle-unit.junit.xml --coverage-clover=./var/phpunit/coverage/clover/bridge-symfony-postgresql-bundle-unit.coverage.xml",
             "tools/phpunit/vendor/bin/phpunit --testsuite=bridge-symfony-postgresql-bundle-integration --log-junit ./var/phpunit/logs/bridge-symfony-postgresql-bundle-integration.junit.xml --coverage-clover=./var/phpunit/coverage/clover/bridge-symfony-postgresql-bundle-integration.coverage.xml"
         ],
+        "test:bridge:symfony-postgresql-messenger": [
+            "tools/phpunit/vendor/bin/phpunit --testsuite=bridge-symfony-postgresql-messenger-unit --log-junit ./var/phpunit/logs/bridge-symfony-postgresql-messenger-unit.junit.xml --coverage-clover=./var/phpunit/coverage/clover/bridge-symfony-postgresql-messenger-unit.coverage.xml",
+            "tools/phpunit/vendor/bin/phpunit --testsuite=bridge-symfony-postgresql-messenger-integration --log-junit ./var/phpunit/logs/bridge-symfony-postgresql-messenger-integration.junit.xml --coverage-clover=./var/phpunit/coverage/clover/bridge-symfony-postgresql-messenger-integration.coverage.xml"
+        ],
         "test:bridge:symfony-telemetry-bundle": [
             "tools/phpunit/vendor/bin/phpunit --testsuite=bridge-symfony-telemetry-bundle-unit --log-junit ./var/phpunit/logs/bridge-symfony-telemetry-bundle-unit.junit.xml --coverage-clover=./var/phpunit/coverage/clover/bridge-symfony-telemetry-bundle-unit.coverage.xml",
             "tools/phpunit/vendor/bin/phpunit --testsuite=bridge-symfony-telemetry-bundle-integration --log-junit ./var/phpunit/logs/bridge-symfony-telemetry-bundle-integration.junit.xml --coverage-clover=./var/phpunit/coverage/clover/bridge-symfony-telemetry-bundle-integration.coverage.xml"

documentation/components/bridges/symfony-postgresql-bundle.md

@@ -299,6 +299,65 @@ Target a specific connection with any command:
 php bin/console flow:migrations:migrate --connection=reporting
 ```
 
+## Symfony Messenger Integration
+
+The bundle integrates with [flow-php/symfony-postgresql-messenger-bridge](/documentation/components/bridges/symfony-postgresql-messenger-bridge.md) to provide a Symfony Messenger transport backed by Flow's native PostgreSQL client — no Doctrine DBAL required.
+
+### Setup
+
+1. Install the messenger bridge:
+
+```bash
+composer require flow-php/symfony-postgresql-messenger-bridge:~--FLOW_PHP_VERSION--
+```
+
+2. Enable messenger:
+
+```yaml
+flow_postgresql:
+  connections:
+    default:
+      dsn: '%env(DATABASE_URL)%'
+
+  messenger:
+    enabled: true
+    table_name: messenger_messages  # default
+    schema: public                  # default
+```
+
+3. Configure the Symfony Messenger transport:
+
+```yaml
+# config/packages/messenger.yaml
+framework:
+  messenger:
+    transports:
+      async:
+        dsn: 'flow-pgsql://default'
+    routing:
+      App\Message\MyMessage: async
+```
+
+4. Generate and run the migration to create the messenger table:
+
+```bash
+php bin/console flow:migrations:diff
+php bin/console flow:migrations:migrate
+```
+
+The `MessengerCatalogProvider` is registered automatically when `messenger.enabled: true`, so the `messenger_messages` table appears in schema diffs alongside your other catalog-managed tables.
+
+### Configuration Options
+
+| Option | Default | Location | Description |
+|--------|---------|----------|-------------|
+| `table_name` | `messenger_messages` | `flow_postgresql.messenger` | Table name in the database |
+| `schema` | `public` | `flow_postgresql.messenger` | Schema owning the table |
+| `queue_name` | `default` | `framework.messenger.transports.*.options` | Queue name for message routing |
+| `redeliver_timeout` | `3600` | `framework.messenger.transports.*.options` | Seconds before unacknowledged messages are redelivered |
+
+For full documentation, see the [Symfony PostgreSQL Messenger Bridge](/documentation/components/bridges/symfony-postgresql-messenger-bridge.md).
+
 ## Complete Example
 
 ```yaml
@@ -316,6 +375,9 @@ flow_postgresql:
         log_queries: false
         max_query_length: 1000
 
+  messenger:
+    enabled: true
+
   migrations:
     enabled: true
     directory: "%kernel.project_dir%/migrations"

documentation/components/bridges/symfony-postgresql-messenger-bridge.md

@@ -0,0 +1,136 @@
+# Symfony PostgreSQL Messenger Bridge
+
+A Symfony Messenger transport backed by Flow PHP's native PostgreSQL library. Replaces `symfony/doctrine-messenger` without requiring Doctrine DBAL — messages are stored directly in PostgreSQL using Flow's query builder and client.
+
+- [Back](/documentation/introduction.md)
+- [Packagist](https://packagist.org/packages/flow-php/symfony-postgresql-messenger-bridge)
+- [GitHub](https://github.com/flow-php/symfony-postgresql-messenger-bridge)
+- [API Reference](/documentation/api/bridge/symfony-postgresql-messenger)
+
+[TOC]
+
+## Installation
+
+```bash
+composer require flow-php/symfony-postgresql-messenger-bridge:~--FLOW_PHP_VERSION--
+```
+
+This package is used together with the [Symfony PostgreSQL Bundle](/documentation/components/bridges/symfony-postgresql-bundle.md), which registers the transport factory and catalog provider automatically.
+
+## How It Works
+
+The bridge provides a Symfony Messenger transport that stores messages in a PostgreSQL table using Flow's native `Client`. No Doctrine DBAL is involved.
+
+- **Sending** inserts a row with `body`, `headers`, `queue_name`, `created_at`, and `available_at`
+- **Receiving** uses `SELECT ... FOR UPDATE SKIP LOCKED` inside a transaction for safe concurrent consumption
+- **Delayed messages** offset `available_at` by the `DelayStamp` value (in milliseconds)
+- **Redelivery** of timed-out messages is handled automatically based on `redeliver_timeout`
+- **Keepalive** refreshes `delivered_at` to prevent redelivery during long-running handlers
+
+## Configuration
+
+### 1. Enable Messenger
+
+In your `flow_postgresql` configuration, enable messenger at the root level:
+
+```yaml
+# config/packages/flow_postgresql.yaml
+flow_postgresql:
+  connections:
+    default:
+      dsn: '%env(DATABASE_URL)%'
+
+  messenger:
+    enabled: true
+    table_name: messenger_messages  # default
+    schema: public                  # default
+```
+
+This registers a `MessengerCatalogProvider` and a transport factory that can resolve any configured connection by name. The `messenger_messages` table will be included in schema diffs and migrations automatically.
+
+### 2. Configure the Messenger Transport
+
+```yaml
+# config/packages/messenger.yaml
+framework:
+  messenger:
+    transports:
+      async:
+        dsn: 'flow-pgsql://default'
+        options:
+          queue_name: default           # default
+          redeliver_timeout: 3600       # seconds, default
+    routing:
+      App\Message\MyMessage: async
+```
+
+The DSN format is `flow-pgsql://<connection_name>` where `<connection_name>` matches a connection defined under `flow_postgresql.connections`. Both `flow-pgsql://` and `flow-postgresql://` schemes are supported.
+
+### 3. Create the Table via Migrations
+
+Because the catalog provider is registered automatically, the messenger table appears in migration diffs:
+
+```bash
+php bin/console flow:migrations:diff
+php bin/console flow:migrations:migrate
+```
+
+This is the recommended approach. The bridge does **not** implement `SetupableTransportInterface` — there is no `messenger:setup-transports` support. Use migrations for schema management.
+
+## DSN Options
+
+Options can be set in the DSN query string or in the `options` array under the transport configuration. The `options` array takes precedence over DSN query parameters.
+
+| Option | Default | Description |
+|--------|---------|-------------|
+| `queue_name` | `default` | Queue name for message routing |
+| `table_name` | `messenger_messages` | Table storing messages (must match `flow_postgresql.messenger` config) |
+| `schema` | `public` | Schema owning the table (must match `flow_postgresql.messenger` config) |
+| `redeliver_timeout` | `3600` | Seconds before a delivered-but-unacknowledged message becomes eligible for redelivery |
+
+Example with DSN options:
+
+```yaml
+framework:
+  messenger:
+    transports:
+      high_priority:
+        dsn: 'flow-pgsql://default?queue_name=high&redeliver_timeout=1800'
+```
+
+## Multiple Queues
+
+Multiple transports can share the same table but use different queue names:
+
+```yaml
+framework:
+  messenger:
+    transports:
+      async:
+        dsn: 'flow-pgsql://default'
+        options:
+          queue_name: default
+      high_priority:
+        dsn: 'flow-pgsql://default'
+        options:
+          queue_name: high
+    routing:
+      App\Message\ImportantMessage: high_priority
+      App\Message\BackgroundTask: async
+```
+
+## Table Schema
+
+The messenger table has the following structure:
+
+| Column | Type | Description |
+|--------|------|-------------|
+| `id` | `bigint GENERATED ALWAYS AS IDENTITY` | Primary key |
+| `body` | `text` | Serialized message body |
+| `headers` | `text` | JSON-encoded message headers (stamps, type info) |
+| `queue_name` | `varchar(190)` | Queue name for message routing |
+| `created_at` | `timestamptz` | When the message was dispatched |
+| `available_at` | `timestamptz` | When the message becomes available for consumption |
+| `delivered_at` | `timestamptz` | When a worker picked up the message (null = not yet delivered) |
+
+Three indexes are created: on `queue_name`, `available_at`, and `delivered_at`.

documentation/installation/packages/symfony-postgresql-bundle.md

@@ -36,4 +36,5 @@ composer require flow-php/symfony-postgresql-bundle:~--FLOW_PHP_VERSION--
 
 ## Suggested Dependencies
 
+- [flow-php/symfony-postgresql-messenger-bridge](/documentation/installation/packages/symfony-postgresql-messenger-bridge.md) — for Symfony Messenger PostgreSQL transport support (replaces `symfony/doctrine-messenger`)
 - [flow-php/symfony-telemetry-bundle](/documentation/installation/packages/symfony-telemetry-bundle.md) — for telemetry integration (distributed tracing, metrics, logging)

documentation/installation/packages/symfony-postgresql-messenger-bridge.md

@@ -0,0 +1,32 @@
+---
+seo_title: "Installing Symfony PostgreSQL Messenger Bridge"
+seo_description: >
+  How to install flow-php/symfony-postgresql-messenger-bridge in your PHP project using Composer.
+---
+
+# Symfony PostgreSQL Messenger Bridge
+
+- [Back](/documentation/installation.md)
+- [Documentation](/documentation/components/bridges/symfony-postgresql-messenger-bridge.md)
+- [Packagist](https://packagist.org/packages/flow-php/symfony-postgresql-messenger-bridge)
+
+[TOC]
+
+## Composer
+
+```bash
+composer require flow-php/symfony-postgresql-messenger-bridge:~--FLOW_PHP_VERSION--
+```
+
+## Core Dependencies
+
+- [flow-php/postgresql](/documentation/installation/packages/postgresql.md)
+- [symfony/messenger](https://packagist.org/packages/symfony/messenger)
+
+## Recommended Packages
+
+- [flow-php/symfony-postgresql-bundle](/documentation/installation/packages/symfony-postgresql-bundle.md) — registers the transport factory and catalog provider automatically via Symfony bundle configuration
+
+## Recommended Extensions
+
+- `ext-pgsql` — required for database connections and query execution

manifest.json

@@ -191,6 +191,11 @@
       "path": "src/bridge/symfony/postgresql-bundle",
       "type": "bridge"
     },
+    {
+      "name": "flow-php/symfony-postgresql-messenger-bridge",
+      "path": "src/bridge/symfony/postgresql-messenger",
+      "type": "bridge"
+    },
     {
       "name": "flow-php/symfony-telemetry-bundle",
       "path": "src/bridge/symfony/telemetry-bundle",

phpdoc/bridge.symfony.postgresql-messenger.xml

@@ -0,0 +1,24 @@
+<?xml version="1.0" encoding="UTF-8" ?>
+<phpdocumentor
+        configVersion="3"
+        xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+        xmlns="https://www.phpdoc.org"
+>
+    <title>Flow PHP</title>
+    <paths>
+        <output>./../web/landing/build/documentation/api/bridge/symfony/postgresql-messenger</output>
+        <cache>./../var/phpdocumentor/cache/bridge/symfony/postgresql-messenger</cache>
+    </paths>
+    <version number="1.x">
+        <api format="php">
+            <source dsn="./../">
+                <path>src/bridge/symfony/postgresql-messenger/src</path>
+            </source>
+            <output>symfony-postgresql-messenger</output>
+            <default-package-name>Symfony PostgreSQL Messenger</default-package-name>
+            <visibility>public</visibility>
+            <include-source>false</include-source>
+        </api>
+    </version>
+    <setting name="template.color" value="orange"/>
+</phpdocumentor>