fix(channel): bind NOTIFY payloads via pg_notify

modules/core/shared/src/main/scala/Channel.scala

@@ -10,9 +10,10 @@ import cats.effect.{ Resource, MonadCancel }
 import cats.effect.kernel.MonadCancelThrow
 import cats.syntax.all._
 import fs2.{ Pipe, Stream }
+import skunk.codec.all._
 import skunk.data.{ Identifier, Notification }
 import skunk.net.Protocol
-import skunk.util.Origin
+import skunk.util.{ Origin, Typer }
 
 /**
  * A '''channel''' that can be used for inter-process communication, implemented in terms of
@@ -140,9 +141,14 @@ object Channel {
       } yield stream.filter(_.channel === name)
 
 
+    val notifyQuery =
+      Query("SELECT pg_notify($1, $2) IS NULL", Origin.unknown, text ~ text, bool)
+
     def notify(message: String): F[Unit] =
-      // TODO: escape the message
-      proto.execute(Command(s"NOTIFY ${name.sql}, '$message'", Origin.unknown, Void.codec)).void
+      for {
+        pq <- proto.prepare(notifyQuery, Typer.Static)
+        _  <- pq.executeSized((name.value, message), Origin.unknown, 1).void
+      } yield ()
 
   }
 

modules/tests/shared/src/test/scala/ChannelTest.scala

@@ -56,6 +56,19 @@ class ChannelTest extends SkunkTest {
     }
   }
 
+  sessionTest("channel notify payload is passed as a bound parameter") { s =>
+    val payload = "it's a 'test\"message"
+    val ch = s.channel(ident"channel_test")
+
+    ch.listenR(42).use { r =>
+      for {
+        _ <- ch.notify(payload)
+        d <- r.map(_.value).take(1).compile.toList
+        _ <- assert(s"channel payload $d", d == List(payload))
+      } yield "ok"
+    }
+  }
+
   sessionTest("channel with quoted identifier round-trips through LISTEN/NOTIFY/UNLISTEN") { s =>
     val data = List("foo", "bar", "baz")
     val ch = s.channel(ident"q_my_queue.INSERT")