Merge branch 'fix/supabase-session-mode-port' into 'main'

fix(monitoring): use session-mode port for Supabase pooler

Closes #342

See merge request postgres-ai/postgresai!403

Author: Sarumyan

cli/bin/postgres-ai.ts

@@ -64,6 +64,7 @@ import {
   buildClientConfig,
   sslOptionFromConnString,
   warnIfLaxSslmode,
+  warnIfTransactionPoolerPort,
 } from "../lib/instances";
 
 // Node.js version check - require Node 18+
@@ -3231,6 +3232,7 @@ mon
           let testClient: InstanceType<typeof Client> | null = null;
           try {
             warnIfLaxSslmode(connStr);
+            warnIfTransactionPoolerPort(connStr);
             testClient = new Client(buildClientConfig(connStr, { connectionTimeoutMillis: 10000 }));
             await testClient.connect();
             const result = await testClient.query("select version();");
@@ -3279,6 +3281,7 @@ mon
                 let testClient: InstanceType<typeof Client> | null = null;
                 try {
                   warnIfLaxSslmode(connStr);
+                  warnIfTransactionPoolerPort(connStr);
             testClient = new Client(buildClientConfig(connStr, { connectionTimeoutMillis: 10000 }));
                   await testClient.connect();
                   const result = await testClient.query("select version();");
@@ -4227,6 +4230,7 @@ targets
     console.log(`Testing connection to monitoring target '${name}'...`);
 
     warnIfLaxSslmode(instance.conn_str);
+    warnIfTransactionPoolerPort(instance.conn_str);
     const client = new Client(buildClientConfig(instance.conn_str, { connectionTimeoutMillis: 10000 }));
 
     try {

cli/lib/instances.ts

@@ -188,6 +188,49 @@ export function warnIfLaxSslmode(connStr: string): void {
   );
 }
 
+/**
+ * Supabase's pooler (Supavisor) serves transaction mode on 6543 and session
+ * mode on 5432. Under transaction mode, client sessions share a pool of server
+ * backends, so pgx's cached server-side prepared statements collide with
+ * `prepared statement "stmtcache_<hash>" already exists` (42P05) and metric
+ * collection stops without surfacing a hard failure.
+ *
+ * Only a pooler HOST on that exact port qualifies. A direct
+ * `db.<ref>.supabase.co` host runs no pooler, so 6543 there is merely an
+ * unusual port — warning about it would be wrong.
+ */
+const SUPABASE_POOLER_HOST_SUFFIX = "pooler.supabase.com";
+const SUPABASE_POOLER_TRANSACTION_PORT = "6543";
+const SUPABASE_POOLER_SESSION_PORT = "5432";
+
+export function isTransactionPoolerUrl(connStr: string): boolean {
+  try {
+    const u = new URL(connStr);
+    return (
+      u.hostname.toLowerCase().endsWith(SUPABASE_POOLER_HOST_SUFFIX) &&
+      u.port === SUPABASE_POOLER_TRANSACTION_PORT
+    );
+  } catch {
+    return false;
+  }
+}
+
+/**
+ * Print a stderr warning when the connection string points at the pooler's
+ * transaction-mode port. Sits alongside `warnIfLaxSslmode` at the same
+ * Client-construction sites so hand-written connection strings get the same
+ * treatment the Supabase provisioning path applies automatically.
+ */
+export function warnIfTransactionPoolerPort(connStr: string): void {
+  if (!isTransactionPoolerUrl(connStr)) return;
+  console.error(
+    `⚠ port ${SUPABASE_POOLER_TRANSACTION_PORT} is the pooler's TRANSACTION mode: ` +
+      `pooled backends share prepared statements, so collection fails with ` +
+      `42P05 (prepared statement already exists) and metrics stop silently. ` +
+      `Use port ${SUPABASE_POOLER_SESSION_PORT} on the same host for session mode.`,
+  );
+}
+
 /**
  * Build a `pg.Client` config from a connection string that ACTUALLY honors
  * libpq sslmode semantics.

cli/lib/supabase.ts

@@ -10,6 +10,44 @@
 
 const SUPABASE_API_BASE = "https://api.supabase.com";
 
+/**
+ * Supabase's connection pooler (Supavisor) listens on two ports:
+ *   6543 — transaction mode: client sessions are multiplexed onto a shared
+ *          pool of server backends, so anything with per-session server state
+ *          is unsafe.
+ *   5432 — session mode: one client gets one backend for the life of the
+ *          connection, so server-side state behaves as on a direct connection.
+ *
+ * pgwatch connects with pgx, which caches server-side PREPARED STATEMENTS by
+ * default. In transaction mode those names collide across the shared backends:
+ *
+ *   ERROR: prepared statement "stmtcache_<hash>" already exists (SQLSTATE 42P05)
+ *
+ * Collection then fails and the series just stop, with no hard failure
+ * anywhere the operator would look. Monitoring must therefore use session
+ * mode. The pooler host is still the right target: the direct
+ * `db.<ref>.supabase.co` host is IPv6-only, while the pooler is reachable over
+ * IPv4 and serves session mode on 5432.
+ */
+const SUPABASE_POOLER_TRANSACTION_PORT = 6543;
+const SUPABASE_POOLER_SESSION_PORT = 5432;
+const SUPABASE_POOLER_HOST_SUFFIX = "pooler.supabase.com";
+
+/**
+ * Map a pooler endpoint onto its session-mode port.
+ *
+ * Only the exact transaction-mode port on a pooler host is rewritten. A direct
+ * host runs no pooler, so its port is authoritative and passes through
+ * verbatim; so does any other port, which we have no basis to second-guess.
+ */
+function sessionModePort(host: string, port: number | string): number | string {
+  const isPoolerHost = host.toLowerCase().endsWith(SUPABASE_POOLER_HOST_SUFFIX);
+  if (isPoolerHost && Number(port) === SUPABASE_POOLER_TRANSACTION_PORT) {
+    return SUPABASE_POOLER_SESSION_PORT;
+  }
+  return port;
+}
+
 export type SupabaseConfig = {
   /** Supabase project reference (e.g., "abc123xyz") */
   projectRef: string;
@@ -340,6 +378,10 @@ export class SupabaseClient {
  * Note: The username will be automatically suffixed with `.<projectRef>` if not
  * already present, as required by Supabase pooler connections.
  *
+ * The API reports the pooler's transaction-mode port; this returns the
+ * session-mode port instead, because pgwatch's pgx prepared statements are
+ * unsafe under transaction pooling. See `sessionModePort` above.
+ *
  * @param config Supabase configuration with projectRef and accessToken
  * @param username Username to include in the URL (e.g., monitoring user).
  *                 Will be transformed to `<username>.<projectRef>` format.
@@ -384,14 +426,17 @@ export async function fetchPoolerDatabaseUrl(
       const pooler = data[0];
       // Build URL from components if available
       if (pooler.db_host && pooler.db_port && pooler.db_name) {
-        return `postgresql://${encodedUsername}@${pooler.db_host}:${pooler.db_port}/${pooler.db_name}`;
+        const port = sessionModePort(pooler.db_host, pooler.db_port);
+        return `postgresql://${encodedUsername}@${pooler.db_host}:${port}/${pooler.db_name}`;
       }
       // Fallback: try to extract from connection_string if present
       if (typeof pooler.connection_string === "string") {
         try {
           const connUrl = new URL(pooler.connection_string);
           // Use provided username; handle empty port for default ports (e.g., 5432)
-          const portPart = connUrl.port ? `:${connUrl.port}` : "";
+          const portPart = connUrl.port
+            ? `:${sessionModePort(connUrl.hostname, connUrl.port)}`
+            : "";
           return `postgresql://${encodedUsername}@${connUrl.hostname}${portPart}${connUrl.pathname}`;
         } catch {
           return null;

cli/test/monitoring.test.ts

@@ -12,6 +12,8 @@ import {
   buildClientConfig,
   sslOptionFromConnString,
   warnIfLaxSslmode,
+  warnIfTransactionPoolerPort,
+  isTransactionPoolerUrl,
   isLaxSslmode,
   extractSslmode,
   InstancesParseError,
@@ -706,6 +708,81 @@ describe("warnIfLaxSslmode — UX warning for lax sslmode", () => {
   });
 });
 
+describe("isTransactionPoolerUrl — detects Supabase transaction-mode pooler", () => {
+  test("true for a pooler host on the transaction port", () => {
+    expect(
+      isTransactionPoolerUrl(
+        "postgresql://u.ref:p@aws-1-eu-west-1.pooler.supabase.com:6543/postgres",
+      ),
+    ).toBe(true);
+  });
+
+  test("false for the same pooler host on the session port", () => {
+    expect(
+      isTransactionPoolerUrl(
+        "postgresql://u.ref:p@aws-1-eu-west-1.pooler.supabase.com:5432/postgres",
+      ),
+    ).toBe(false);
+  });
+
+  test("false for a direct host, even on 6543", () => {
+    // A direct host runs no pooler; 6543 there is just an unusual port and
+    // rewriting or warning about it would be wrong.
+    expect(
+      isTransactionPoolerUrl(
+        "postgresql://postgres:p@db.abcdefghij.supabase.co:6543/postgres",
+      ),
+    ).toBe(false);
+  });
+
+  test("false for an unrelated host on 6543", () => {
+    expect(
+      isTransactionPoolerUrl("postgresql://u:p@db.example.com:6543/postgres"),
+    ).toBe(false);
+  });
+
+  test("false for an unparseable connection string", () => {
+    expect(isTransactionPoolerUrl("not-a-url")).toBe(false);
+  });
+});
+
+describe("warnIfTransactionPoolerPort — UX warning for transaction pooling", () => {
+  let stderrSpy: ReturnType<typeof spyOn>;
+
+  beforeEach(() => {
+    stderrSpy = spyOn(console, "error").mockImplementation(() => {});
+  });
+
+  afterEach(() => {
+    stderrSpy.mockRestore();
+  });
+
+  test("warns, naming both ports and the failure it causes", () => {
+    warnIfTransactionPoolerPort(
+      "postgresql://u.ref:p@aws-1-eu-west-1.pooler.supabase.com:6543/postgres",
+    );
+    expect(stderrSpy).toHaveBeenCalledTimes(1);
+    const msg = String(stderrSpy.mock.calls[0][0]);
+    expect(msg).toContain("6543");
+    expect(msg).toContain("5432");
+    expect(msg).toContain("42P05");
+  });
+
+  test("does NOT warn on the session-mode port", () => {
+    warnIfTransactionPoolerPort(
+      "postgresql://u.ref:p@aws-1-eu-west-1.pooler.supabase.com:5432/postgres",
+    );
+    expect(stderrSpy).not.toHaveBeenCalled();
+  });
+
+  test("does NOT warn on a direct host", () => {
+    warnIfTransactionPoolerPort(
+      "postgresql://postgres:p@db.abcdefghij.supabase.co:5432/postgres",
+    );
+    expect(stderrSpy).not.toHaveBeenCalled();
+  });
+});
+
 describe("buildClientConfig — silences pg-connection-string deprecation warning", () => {
   // pg-connection-string v2.x prints `process.emitWarning("SECURITY WARNING:
   // ... 'prefer'/'require'/'verify-ca' ...")` whenever a recognised lax

cli/test/supabase.test.ts

@@ -166,7 +166,7 @@ describe("Supabase module", () => {
         "postgres_ai_mon"
       );
       expect(url).toBe(
-        "postgresql://postgres_ai_mon.xhaqmsvczjkkvkgdyast@aws-1-eu-west-1.pooler.supabase.com:6543/postgres"
+        "postgresql://postgres_ai_mon.xhaqmsvczjkkvkgdyast@aws-1-eu-west-1.pooler.supabase.com:5432/postgres"
       );
     });
 
@@ -191,7 +191,7 @@ describe("Supabase module", () => {
         "postgres_ai_mon.xhaqmsvczjkkvkgdyast"
       );
       expect(url).toBe(
-        "postgresql://postgres_ai_mon.xhaqmsvczjkkvkgdyast@aws-1-eu-west-1.pooler.supabase.com:6543/postgres"
+        "postgresql://postgres_ai_mon.xhaqmsvczjkkvkgdyast@aws-1-eu-west-1.pooler.supabase.com:5432/postgres"
       );
     });
 
@@ -216,7 +216,7 @@ describe("Supabase module", () => {
         "postgres_ai_mon"
       );
       expect(url).toBe(
-        "postgresql://postgres_ai_mon.xhaqmsvczjkkvkgdyast@aws-1-eu-west-1.pooler.supabase.com:6543/postgres"
+        "postgresql://postgres_ai_mon.xhaqmsvczjkkvkgdyast@aws-1-eu-west-1.pooler.supabase.com:5432/postgres"
       );
     });
 
@@ -265,6 +265,86 @@ describe("Supabase module", () => {
       expect(url).toBeNull();
     });
 
+    test("rewrites the transaction-mode pooler port to session mode", async () => {
+      // The Supabase Management API reports the TRANSACTION-mode pooler port
+      // (6543). pgwatch connects with pgx, which uses server-side prepared
+      // statements; under transaction pooling those collide across the shared
+      // backends with `prepared statement "stmtcache_<hash>" already exists`
+      // (42P05) and metric collection silently stops.
+      globalThis.fetch = mock(() =>
+        Promise.resolve(
+          new Response(
+            JSON.stringify([
+              {
+                db_host: "aws-1-eu-west-1.pooler.supabase.com",
+                db_port: 6543,
+                db_name: "postgres",
+              },
+            ]),
+            { status: 200 }
+          )
+        )
+      ) as unknown as typeof fetch;
+
+      const url = await fetchPoolerDatabaseUrl(
+        { projectRef: "xhaqmsvczjkkvkgdyast", accessToken: "token" },
+        "postgres_ai_mon"
+      );
+      expect(new URL(url!).port).toBe("5432");
+      // Same host — the pooler serves session mode on 5432. Switching to the
+      // direct `db.<ref>.supabase.co` host would require IPv6.
+      expect(new URL(url!).hostname).toBe("aws-1-eu-west-1.pooler.supabase.com");
+    });
+
+    test("leaves a session-mode pooler port untouched", async () => {
+      globalThis.fetch = mock(() =>
+        Promise.resolve(
+          new Response(
+            JSON.stringify([
+              {
+                db_host: "aws-1-eu-west-1.pooler.supabase.com",
+                db_port: 5432,
+                db_name: "postgres",
+              },
+            ]),
+            { status: 200 }
+          )
+        )
+      ) as unknown as typeof fetch;
+
+      const url = await fetchPoolerDatabaseUrl(
+        { projectRef: "xhaqmsvczjkkvkgdyast", accessToken: "token" },
+        "postgres_ai_mon"
+      );
+      expect(new URL(url!).port).toBe("5432");
+    });
+
+    test("leaves a direct (non-pooler) host untouched", async () => {
+      // A direct host never runs a transaction pooler, so its port is
+      // authoritative and must be passed through verbatim.
+      globalThis.fetch = mock(() =>
+        Promise.resolve(
+          new Response(
+            JSON.stringify([
+              {
+                db_host: "db.xhaqmsvczjkkvkgdyast.supabase.co",
+                db_port: 6543,
+                db_name: "postgres",
+              },
+            ]),
+            { status: 200 }
+          )
+        )
+      ) as unknown as typeof fetch;
+
+      const url = await fetchPoolerDatabaseUrl(
+        { projectRef: "xhaqmsvczjkkvkgdyast", accessToken: "token" },
+        "postgres_ai_mon"
+      );
+      expect(new URL(url!).hostname).toBe("db.xhaqmsvczjkkvkgdyast.supabase.co");
+      expect(new URL(url!).port).toBe("6543");
+    });
+
     test("returns null when fetch throws network error", async () => {
       globalThis.fetch = mock(() =>
         Promise.reject(new Error("Network error"))