Support database poolers: clean up LISTEN/NOTIFY state when opening a connection

This should resolve the following issue:

  encountered unexpected or invalid data: execute: unexpected message: NotificationResponse (sqlx_postgres::connection::executor:361)

Author: sandhose

crates/cli/src/util.rs

@@ -23,7 +23,7 @@ use mas_storage::RepositoryAccess;
 use mas_storage_pg::PgRepository;
 use mas_templates::{SiteConfigExt, TemplateLoadingError, Templates};
 use sqlx::{
-    ConnectOptions, PgConnection, PgPool,
+    ConnectOptions, Executor, PgConnection, PgPool,
     postgres::{PgConnectOptions, PgPoolOptions},
 };
 use tokio_util::{sync::CancellationToken, task::TaskTracker};
@@ -339,6 +339,15 @@ pub async fn database_pool_from_config(config: &DatabaseConfig) -> Result<PgPool
         .acquire_timeout(config.connect_timeout)
         .idle_timeout(config.idle_timeout)
         .max_lifetime(config.max_lifetime)
+        .after_connect(|conn, _meta| {
+            Box::pin(async move {
+                // Unlisten from all channels, as we might be connected via a connection pooler
+                // that doesn't clean up LISTEN/NOTIFY state when reusing connections.
+                conn.execute("UNLISTEN *;").await?;
+
+                Ok(())
+            })
+        })
         .connect_with(options)
         .await
         .context("could not connect to the database")