feat(flow-php/postgresql): libpq quoting, session settings and bundle connection options

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

@@ -98,6 +98,33 @@ when@test:
         dbname_suffix: '_test%env(default::TEST_TOKEN)%'
 ```
 
+### Connection Options and Session Settings
+
+```yaml
+flow_postgresql:
+  connections:
+    default:
+      dsn: '%env(DATABASE_URL)%'
+      options:
+        connect_timeout: '%app.timings.database.connect_timeout%'
+        keepalives_idle: 10
+        tcp_user_timeout: 30000
+      session_settings:
+        statement_timeout: 30s
+        lock_timeout: 10s
+        idle_in_transaction_session_timeout: 60s
+```
+
+`options` takes the keys of `ConnectionParameters::OPTION_KEYS`. Another key fails at container compilation:
+
+```
+Connection "default" has unknown libpq keys under "options": conect_timeout. Allowed keys: application_name, ...
+```
+
+Values are strings or integers, and `%env()%` works. A key in `options` wins over the same key in the DSN, and the
+other DSN options stay. `session_settings` go through `withSessionSettings()`, after a value of `options.options`.
+Behind PgBouncer, read the [PgBouncer note](/documentation/components/libs/postgresql/client-connection.md#pgbouncer).
+
 ### Telemetry
 
 To get distributed tracing, query logging, and metrics for a connection, enable telemetry on that connection.

documentation/components/libs/postgresql/client-connection.md

@@ -32,6 +32,16 @@ $client = pgsql_client(
 );
 ```
 
+Put a value with spaces, quotes or backslashes in single quotes:
+
+```php
+<?php
+
+use function Flow\PostgreSql\DSL\pgsql_connection;
+
+$params = pgsql_connection("host=localhost dbname=mydb password='my secret'");
+```
+
 ## DSN URL
 
 Parse the standard PostgreSQL DSN format that environment variables often use:
@@ -135,6 +145,33 @@ $params = (new DsnParser())->parse('postgresql://user:pass@localhost:5432/mydb')
 With `withDatabase()`, apply the override first. The suffix is then appended to the overridden name
 (`withDatabase('other')->withDatabaseSuffix('_test')` ⇒ `other_test`).
 
+## Session Settings
+
+```php
+<?php
+
+use function Flow\PostgreSql\DSL\{pgsql_client, pgsql_connection_dsn};
+
+$client = pgsql_client(
+    pgsql_connection_dsn('postgresql://user:pass@localhost:5432/mydb')
+        ->withSessionSettings([
+            'statement_timeout' => '30s',
+            'lock_timeout' => '10s',
+            'idle_in_transaction_session_timeout' => '60s',
+        ])
+);
+```
+
+Each pair goes to the server as `-c name=value` in the libpq `options` key, after a value that is already there. A
+name holds letters, digits, `_` and `.`. The client escapes spaces and backslashes in the value.
+
+### PgBouncer
+
+PgBouncer rejects a connection that sends startup parameters that it does not track. With
+[`ignore_startup_parameters`](https://www.pgbouncer.org/config.html#ignore_startup_parameters) it drops them, so the
+server does not apply the settings. Behind PgBouncer, set them with `SET` after you connect, or on the role with
+`ALTER ROLE ... SET`.
+
 ## Connection Lifecycle
 
 ### Checking Connection Status
@@ -172,5 +209,7 @@ For long-running applications, think about connection pooling at the infrastruct
 | `connect_timeout`     | Connection timeout in seconds                                      | (system default) |
 | `application_name`    | Application name for pg_stat_activity                              | (none)           |
 
+`ConnectionParameters::OPTION_KEYS` lists the libpq keys that a DSN query string and the Symfony bundle accept.
+
 For the full list of libpq connection parameters, read
 the [PostgreSQL documentation](https://www.postgresql.org/docs/current/libpq-connect.html#LIBPQ-PARAMKEYWORDS).

documentation/upgrading.md

@@ -240,6 +240,28 @@ before it returns it fails with PHP's `Cannot rewind a generator that was alread
 | `mv()` of a missing source throws `AzureException` (404)           | throws `Flow\Filesystem\Exception\RuntimeException`                                                           |
 | `Options::fromArray()` takes options without `copy_part_size`      | the options must hold `copy_part_size` (int)                                                                  |
 
+### 20) `flow-php/postgresql` - connection string values are quoted
+
+`ConnectionParameters::toString()` quotes a value that is empty or holds whitespace, `'` or `\`. A libpq
+`options` value that escapes spaces with a backslash no longer works. `ConnectionParameters::fromString()` and
+`pgsql_connection()` now read the string by the libpq rules: a backslash escapes the next character, and a keyword
+without `=` or an unterminated quote throws `InvalidArgumentException`.
+
+Before:
+
+```
+postgresql://user:pass@localhost/app?options=-cstatement_timeout%3D30s%5C%20-clock_timeout%3D10s
+```
+
+After:
+
+```
+postgresql://user:pass@localhost/app?options=-c%20statement_timeout%3D30s%20-c%20lock_timeout%3D10s
+```
+
+Or `withSessionSettings(['statement_timeout' => '30s', 'lock_timeout' => '10s'])`, or `session_settings` in the
+Symfony bundle.
+
 ---
 
 ## Upgrading from 0.46.x to 0.47.x

src/bridge/symfony/postgresql-bundle/src/Flow/Bridge/Symfony/PostgreSqlBundle/Connection/ConnectionParametersFactory.php

@@ -6,11 +6,20 @@
 
 use Flow\PostgreSql\Client\ConnectionParameters;
 use Flow\PostgreSql\Client\DsnParser;
+use InvalidArgumentException;
 
+use function get_debug_type;
+use function is_int;
+use function is_string;
+use function sprintf;
+
+/**
+ * @type ConnectionOverrides = array{dbname?: ?string, host?: ?string, port?: ?int, user?: ?string, password?: ?string, dbname_suffix?: string, options?: array<string, bool|float|int|string|null>, session_settings?: array<string, bool|float|int|string|null>}
+ */
 final class ConnectionParametersFactory
 {
     /**
-     * @param array{dbname?: ?string, host?: ?string, port?: ?int, user?: ?string, password?: ?string, dbname_suffix?: string} $overrides
+     * @param ConnectionOverrides $overrides
      */
     public static function create(DsnParser $parser, string $dsn, array $overrides = []): ConnectionParameters
     {
@@ -36,6 +45,32 @@ public static function create(DsnParser $parser, string $dsn, array $overrides =
             $params = $params->withPassword($overrides['password']);
         }
 
-        return $params->withDatabaseSuffix($overrides['dbname_suffix'] ?? '');
+        foreach ($overrides['options'] ?? [] as $key => $value) {
+            if (!is_string($value) && !is_int($value)) {
+                throw new InvalidArgumentException(sprintf(
+                    'Connection option "%s" must be a string or an integer, %s given.',
+                    $key,
+                    get_debug_type($value),
+                ));
+            }
+
+            $params = $params->withOption($key, (string) $value);
+        }
+
+        $sessionSettings = [];
+
+        foreach ($overrides['session_settings'] ?? [] as $name => $value) {
+            if (!is_string($value) && !is_int($value)) {
+                throw new InvalidArgumentException(sprintf(
+                    'Session setting "%s" must be a string or an integer, %s given.',
+                    $name,
+                    get_debug_type($value),
+                ));
+            }
+
+            $sessionSettings[$name] = (string) $value;
+        }
+
+        return $params->withSessionSettings($sessionSettings)->withDatabaseSuffix($overrides['dbname_suffix'] ?? '');
     }
 }

src/bridge/symfony/postgresql-bundle/src/Flow/Bridge/Symfony/PostgreSqlBundle/FlowPostgreSqlBundle.php

@@ -66,6 +66,7 @@
 use Twig\Environment;
 use Twig\Loader\FilesystemLoader;
 
+use function array_diff;
 use function array_key_exists;
 use function array_keys;
 use function class_exists;
@@ -155,6 +156,20 @@ public function configure(DefinitionConfigurator $definition): void
             ->defaultValue('')
             ->info('Adds the given suffix to the configured database name.')
             ->end()
+            ->arrayNode('options')
+            ->info(
+                'libpq connection options applied over the DSN, for example connect_timeout or keepalives_idle. Values are strings or integers.',
+            )
+            ->useAttributeAsKey('name')
+            ->scalarPrototype()->end()
+            ->end()
+            ->arrayNode('session_settings')
+            ->info(
+                'Server settings sent at connection start as "-c name=value" in the libpq "options" key, for example statement_timeout.',
+            )
+            ->useAttributeAsKey('name')
+            ->scalarPrototype()->end()
+            ->end()
             ->booleanNode('test_transaction_rollback')
             ->defaultFalse()
             ->info(
@@ -513,7 +528,7 @@ public function configure(DefinitionConfigurator $definition): void
     }
 
     /**
-     * @param array{connections: array<string, array{dsn: string, dbname: ?string, host: ?string, port: ?int, user: ?string, password: ?string, dbname_suffix: string, test_transaction_rollback: bool, lazy: bool, context?: array<string, mixed>, telemetry?: array{service_id: string, clock_service_id: ?string, trace_queries: bool, transaction_spans: 'grouped'|'per_operation'|'off', collect_metrics: bool, log_queries: bool, max_query_length: int, include_parameters: bool, max_parameters: int, max_parameter_length: int}, profiler?: bool}>, messenger: array{enabled: bool, table_name: string, schema: string}, cache: array{pools?: array<string, array{connection: ?string, table_name: string, schema: string, id_col: string, data_col: string, lifetime_col: string, time_col: string, namespace: string, default_lifetime: int, marshaller_service_id: ?string, share_connection: bool}>}, session: array{enabled: bool, connection: ?string, table_name: string, schema: string, id_col: string, data_col: string, lifetime_col: string, time_col: string, lock_mode: string, ttl: ?int, share_connection: bool}, migrations: array{enabled: bool, connection: ?string, directory: string, namespace: string, table_name: string, table_schema: string, migration_file_name: string, rollback_file_name: string, all_or_nothing: bool, generate_rollback: bool, drop_if_exists: bool, context?: array<string, mixed>, exclude?: list<array{schema: ?string, table: ?string, exact: ?string, starts_with: ?string, ends_with: ?string, pattern: ?string, policy_id: ?string, type: ?string, for_schema: ?string}>}, catalog_providers: list<array{catalog_provider_id: ?string, catalog: ?array<string, mixed>}>, profiler?: array{enabled?: bool|null, include_parameters?: bool, max_queries?: int, max_retained_parameters?: int, max_query_length?: int, migrations?: bool}} $config
+     * @param array{connections: array<string, array{dsn: string, dbname: ?string, host: ?string, port: ?int, user: ?string, password: ?string, dbname_suffix: string, options: array<string, bool|float|int|string|null>, session_settings: array<string, bool|float|int|string|null>, test_transaction_rollback: bool, lazy: bool, context?: array<string, mixed>, telemetry?: array{service_id: string, clock_service_id: ?string, trace_queries: bool, transaction_spans: 'grouped'|'per_operation'|'off', collect_metrics: bool, log_queries: bool, max_query_length: int, include_parameters: bool, max_parameters: int, max_parameter_length: int}, profiler?: bool}>, messenger: array{enabled: bool, table_name: string, schema: string}, cache: array{pools?: array<string, array{connection: ?string, table_name: string, schema: string, id_col: string, data_col: string, lifetime_col: string, time_col: string, namespace: string, default_lifetime: int, marshaller_service_id: ?string, share_connection: bool}>}, session: array{enabled: bool, connection: ?string, table_name: string, schema: string, id_col: string, data_col: string, lifetime_col: string, time_col: string, lock_mode: string, ttl: ?int, share_connection: bool}, migrations: array{enabled: bool, connection: ?string, directory: string, namespace: string, table_name: string, table_schema: string, migration_file_name: string, rollback_file_name: string, all_or_nothing: bool, generate_rollback: bool, drop_if_exists: bool, context?: array<string, mixed>, exclude?: list<array{schema: ?string, table: ?string, exact: ?string, starts_with: ?string, ends_with: ?string, pattern: ?string, policy_id: ?string, type: ?string, for_schema: ?string}>}, catalog_providers: list<array{catalog_provider_id: ?string, catalog: ?array<string, mixed>}>, profiler?: array{enabled?: bool|null, include_parameters?: bool, max_queries?: int, max_retained_parameters?: int, max_query_length?: int, migrations?: bool}} $config
      */
     #[Override]
     public function loadExtension(array $config, ContainerConfigurator $configurator, ContainerBuilder $container): void
@@ -784,14 +799,25 @@ private function registerCatalogProviders(array $catalogProviders, ContainerBuil
     }
 
     /**
-     * @param array{dsn: string, dbname: ?string, host: ?string, port: ?int, user: ?string, password: ?string, dbname_suffix: string, test_transaction_rollback: bool, lazy: bool, context?: array<string, mixed>, telemetry?: array{service_id: string, clock_service_id: ?string, trace_queries: bool, transaction_spans: 'grouped'|'per_operation'|'off', collect_metrics: bool, log_queries: bool, max_query_length: int, include_parameters: bool, max_parameters: int, max_parameter_length: int}, profiler?: bool} $connectionConfig
+     * @param array{dsn: string, dbname: ?string, host: ?string, port: ?int, user: ?string, password: ?string, dbname_suffix: string, options: array<string, bool|float|int|string|null>, session_settings: array<string, bool|float|int|string|null>, test_transaction_rollback: bool, lazy: bool, context?: array<string, mixed>, telemetry?: array{service_id: string, clock_service_id: ?string, trace_queries: bool, transaction_spans: 'grouped'|'per_operation'|'off', collect_metrics: bool, log_queries: bool, max_query_length: int, include_parameters: bool, max_parameters: int, max_parameter_length: int}, profiler?: bool} $connectionConfig
      */
     private function registerConnection(
         string $name,
         array $connectionConfig,
         ContainerBuilder $container,
         bool $isFirst,
     ): void {
+        $unknownOptions = array_diff(array_keys($connectionConfig['options']), ConnectionParameters::OPTION_KEYS);
+
+        if ($unknownOptions !== []) {
+            throw new LogicException(sprintf(
+                'Connection "%s" has unknown libpq keys under "options": %s. Allowed keys: %s.',
+                $name,
+                implode(', ', $unknownOptions),
+                implode(', ', ConnectionParameters::OPTION_KEYS),
+            ));
+        }
+
         $parserDef = new Definition(DsnParser::class);
         $container->setDefinition("flow.postgresql.{$name}.dsn_parser", $parserDef);
 
@@ -807,6 +833,8 @@ private function registerConnection(
                 'user' => $connectionConfig['user'] ?? null,
                 'password' => $connectionConfig['password'] ?? null,
                 'dbname_suffix' => $connectionConfig['dbname_suffix'] ?? '',
+                'options' => $connectionConfig['options'],
+                'session_settings' => $connectionConfig['session_settings'],
             ],
         ]);
         $container->setDefinition("flow.postgresql.{$name}.connection_parameters", $paramsDef);

src/bridge/symfony/postgresql-bundle/tests/Flow/Bridge/Symfony/PostgreSqlBundle/Tests/Integration/FlowPostgreSqlExtensionTest.php

@@ -51,6 +51,11 @@
 use Symfony\Component\DependencyInjection\Definition;
 use Symfony\Component\DependencyInjection\Reference;
 
+use function Flow\PostgreSql\DSL\func;
+use function Flow\PostgreSql\DSL\literal;
+use function Flow\PostgreSql\DSL\select;
+use function getenv;
+
 #[CoversClass(FlowPostgreSqlBundle::class)]
 final class FlowPostgreSqlExtensionTest extends KernelTestCase
 {
@@ -796,6 +801,111 @@ public function test_connection_parameters_apply_dsn_part_overrides(): void
         static::assertSame('pw', $params->password());
     }
 
+    public function test_connection_parameters_apply_options_and_session_settings(): void
+    {
+        $previousTcpUserTimeout = $_ENV['FLOW_TEST_TCP_USER_TIMEOUT'] ?? null;
+        $_ENV['FLOW_TEST_TCP_USER_TIMEOUT'] = '30000';
+
+        try {
+            $this->bootKernel([
+                'config' => static function (TestKernel $kernel): void {
+                    $kernel->addTestContainerConfigurator(static function (ContainerBuilder $container): void {
+                        $container->setParameter('app.database.connect_timeout', 5);
+                        $container->setParameter('app.database.statement_timeout', '30s');
+                    });
+                    $kernel->addTestExtensionConfig('flow_postgresql', [
+                        'connections' => [
+                            'default' => [
+                                'dsn' => 'postgresql://postgres:postgres@localhost:5432/app?sslmode=prefer',
+                                'options' => [
+                                    'connect_timeout' => '%app.database.connect_timeout%',
+                                    'tcp_user_timeout' => '%env(int:FLOW_TEST_TCP_USER_TIMEOUT)%',
+                                ],
+                                'session_settings' => ['statement_timeout' => '%app.database.statement_timeout%'],
+                            ],
+                        ],
+                    ]);
+                },
+            ]);
+
+            static::assertSame(
+                [
+                    'sslmode' => 'prefer',
+                    'connect_timeout' => '5',
+                    'tcp_user_timeout' => '30000',
+                    'options' => '-c statement_timeout=30s',
+                ],
+                $this->getContainer()->get('flow.postgresql.default.connection_parameters')->options(),
+            );
+        } finally {
+            if ($previousTcpUserTimeout === null) {
+                unset($_ENV['FLOW_TEST_TCP_USER_TIMEOUT']);
+            } else {
+                $_ENV['FLOW_TEST_TCP_USER_TIMEOUT'] = $previousTcpUserTimeout;
+            }
+        }
+    }
+
+    public function test_connection_rejects_unknown_libpq_option(): void
+    {
+        $this->expectException(LogicException::class);
+        $this->expectExceptionMessage(
+            'Connection "default" has unknown libpq keys under "options": conect_timeout. Allowed keys: application_name,',
+        );
+
+        $this->bootKernel([
+            'config' => static function (TestKernel $kernel): void {
+                $kernel->addTestExtensionConfig('flow_postgresql', [
+                    'connections' => [
+                        'default' => [
+                            'dsn' => 'postgresql://postgres:postgres@localhost:5432/app',
+                            'options' => ['conect_timeout' => 5],
+                        ],
+                    ],
+                ]);
+            },
+        ]);
+    }
+
+    public function test_connection_session_settings_reach_the_server(): void
+    {
+        $this->bootKernel([
+            'config' => static function (TestKernel $kernel): void {
+                $kernel->addTestExtensionConfig('flow_postgresql', [
+                    'connections' => [
+                        'default' => [
+                            'dsn' => getenv('PGSQL_DATABASE_URL')
+                                ?: 'postgresql://postgres:postgres@127.0.0.1:5452/postgres',
+                            'options' => ['connect_timeout' => 5],
+                            'session_settings' => [
+                                'statement_timeout' => '30s',
+                                'lock_timeout' => '10s',
+                                'idle_in_transaction_session_timeout' => '60s',
+                            ],
+                        ],
+                    ],
+                ]);
+            },
+        ]);
+
+        $client = $this->getContainer()->get('flow.postgresql.default.client');
+
+        static::assertSame(
+            '30s',
+            $client->fetchScalarString(select(func('current_setting', [literal('statement_timeout')]))),
+        );
+        static::assertSame(
+            '10s',
+            $client->fetchScalarString(select(func('current_setting', [literal('lock_timeout')]))),
+        );
+        static::assertSame(
+            '1min',
+            $client->fetchScalarString(select(func('current_setting', [literal(
+                'idle_in_transaction_session_timeout',
+            )]))),
+        );
+    }
+
     public function test_connection_parameters_apply_dbname_suffix(): void
     {
         $this->bootKernel([