fix(db): pin the tenant scope to one connection (coreci-chat#14)

packages/db/src/withTenant.ts

@@ -6,7 +6,9 @@
  * outside withTenant() returns zero rows (the RLS policy evaluates NULL = X → false).
  *
  * In PGlite (single-connection), BEGIN/COMMIT work on the single connection.
- * In pg (pool), withTenant acquires a dedicated client for the transaction.
+ * In pg (pool), withTenant runs the transaction on a dedicated client that the
+ * client pins for the whole transaction (`DbClient.transaction`), so the tenant
+ * scope and every statement of the transaction share one connection.
  *
  * For simplicity in M1, this implementation holds a single client (set via
  * `setDbClient`). The control plane's API gateway will call setDbClient at boot.
@@ -47,23 +49,44 @@ export async function withTenant<T>(
   fn: (client: ScopedClient) => Promise<T>,
 ): Promise<T> {
   const c = client();
-  const scoped: ScopedClient = {
-    async query(text, params) {
-      return c.query(text, params);
-    },
-  };
   try {
-    await c.query("BEGIN");
-    await c.query("SELECT set_config('app.tenant_id', $1, true)", [tenantId]);
-    const result = await fn(scoped);
-    await c.query("COMMIT");
-    return result;
-  } catch (err) {
-    try {
-      await c.query("ROLLBACK");
-    } catch {
-      /* ignore rollback failure; original error is more important */
+    if (typeof c.transaction === "function") {
+      // The client can pin ONE connection for the whole transaction, so the
+      // tenant scope (SET LOCAL) and every statement in `fn` share it. A pool
+      // otherwise hands out a different connection per statement, and the
+      // scope is then set on one connection and read on another
+      // (coreci-chat#14).
+      return await c.transaction(async (tx) => {
+        await tx.query("SELECT set_config('app.tenant_id', $1, true)", [tenantId]);
+        return fn({
+          async query(text, params) {
+            return tx.query(text, params);
+          },
+        });
+      });
     }
+    // Fallback for a client that cannot pin a connection: the scope and the
+    // statements still share the client's own connection.
+    const scoped: ScopedClient = {
+      async query(text, params) {
+        return c.query(text, params);
+      },
+    };
+    try {
+      await c.query("BEGIN");
+      await c.query("SELECT set_config('app.tenant_id', $1, true)", [tenantId]);
+      const result = await fn(scoped);
+      await c.query("COMMIT");
+      return result;
+    } catch (err) {
+      try {
+        await c.query("ROLLBACK");
+      } catch {
+        /* ignore rollback failure; original error is more important */
+      }
+      throw err;
+    }
+  } catch (err) {
     throw new TenantContextError(
       `withTenant(${tenantId}) failed: ${err instanceof Error ? err.message : String(err)}`,
       err,