chore: add test filtered pg changes to multiple subs

packages/tests/src/test_suites/const.ts

@@ -1,3 +1,5 @@
 export const BROADCAST_CONFIG = { config: { broadcast: { self: true } } }
 export const LOAD_MESSAGES = 20
 export const LOAD_DELIVERY_SLO = 99
+export const FANOUT_SUBSCRIBERS = 15
+export const FANOUT_DELIVERY_SLO = 100

packages/tests/src/test_suites/load-postgres-changes.ts

@@ -10,7 +10,13 @@ import {
   waitForPostgresChannel,
 } from '../helpers'
 import type { TestSuite } from '../types'
-import { BROADCAST_CONFIG, LOAD_DELIVERY_SLO, LOAD_MESSAGES } from './const'
+import {
+  BROADCAST_CONFIG,
+  FANOUT_DELIVERY_SLO,
+  FANOUT_SUBSCRIBERS,
+  LOAD_DELIVERY_SLO,
+  LOAD_MESSAGES,
+} from './const'
 
 export default {
   'load-postgres-changes': [
@@ -136,5 +142,33 @@ export default {
         return measureThroughput(latencies, LOAD_MESSAGES, LOAD_DELIVERY_SLO)
       },
     },
+    {
+      name: 'postgres changes fan-out to many concurrent filtered subscriptions on one table',
+      body: async (supabase, { email, password }) => {
+        await signInUser(supabase, email, password)
+        const value = randomId()
+        const latencies: number[] = []
+        let sentAt = 0
+
+        const channels = Array.from({ length: FANOUT_SUBSCRIBERS }, () =>
+          supabase
+            .channel(`topic:${randomId()}`, BROADCAST_CONFIG)
+            .on(
+              'postgres_changes',
+              { event: 'INSERT', schema: 'public', table: 'pg_changes', filter: `value=eq.${value}` },
+              () => latencies.push(now() - sentAt),
+            )
+            .subscribe(),
+        )
+
+        await Promise.all(channels.map((channel) => waitForPostgresChannel(channel)))
+
+        sentAt = now()
+        const { error } = await supabase.from('pg_changes').insert([{ value }])
+        if (error) throw new Error(`Error inserting data: ${error.message}`)
+
+        return measureThroughput(latencies, FANOUT_SUBSCRIBERS, FANOUT_DELIVERY_SLO)
+      },
+    },
   ],
 } satisfies TestSuite