test: connections failing auth are not added to the connection pool

.changeset/fast-gifts-jam.md

@@ -0,0 +1,5 @@
+---
+"slonik": minor
+---
+
+give pool instance EventEmitter prototype

README.md

@@ -65,6 +65,7 @@ Note: Using this project does not require TypeScript. It is a regular ES6 module
         * [API](#api)
         * [Default configuration](#default-configuration)
         * [Checking out a client from the connection pool](#checking-out-a-client-from-the-connection-pool)
+        * [Events](#events)
     * [How are they different?](#how-are-they-different)
         * [`pg` vs `slonik`](#pg-vs-slonik)
         * [`pg-promise` vs `slonik`](#pg-promise-vs-slonik)
@@ -747,6 +748,24 @@ Connection is released back to the pool after the promise produced by the functi
 
 Read: [Protecting against unsafe connection handling](#protecting-against-unsafe-connection-handling).
 
+### Events
+
+The `DatabasePool` extends `DatabasePoolEventEmitter` and exposes the following events:
+
+- `error`: `(error: SlonikError) => void` – emitted for all errors that happen within the pool.
+
+```ts
+import {
+  createPool,
+} from 'slonik';
+
+const pool = await createPool('postgres://localhost');
+
+pool.on('error', (error) => {
+  console.error(error);
+});
+```
+
 ## How are they different?
 
 ### <code>pg</code> vs <code>slonik</code>

packages/slonik/package.json

@@ -14,7 +14,8 @@
     "iso8601-duration": "^1.3.0",
     "postgres-interval": "^4.0.2",
     "roarr": "^7.21.1",
-    "serialize-error": "^8.0.0"
+    "serialize-error": "^8.0.0",
+    "strict-event-emitter-types": "^2.0.0"
   },
   "description": "A Node.js PostgreSQL client with strict types, detailed logging and assertions.",
   "devDependencies": {

packages/slonik/src/binders/bindPool.ts

@@ -4,15 +4,17 @@ import { type ConnectionPool } from '../factories/createConnectionPool';
 import {
   type ClientConfiguration,
   type DatabasePool,
+  type DatabasePoolEventEmitter,
   type Logger,
 } from '../types';
 
 export const bindPool = (
+  events: DatabasePoolEventEmitter,
   parentLog: Logger,
   pool: ConnectionPool,
   clientConfiguration: ClientConfiguration,
 ): DatabasePool => {
-  return {
+  const boundPool = {
     any: async (query) => {
       return await createConnection(
         parentLog,
@@ -219,5 +221,10 @@ export const bindPool = (
         },
       );
     },
+    ...events,
   };
+
+  Object.setPrototypeOf(boundPool, events);
+
+  return boundPool;
 };

packages/slonik/src/connectionMethods/query.ts

@@ -6,6 +6,7 @@ import {
   type QueryResultRow,
 } from '../types';
 import { type DriverNotice, type DriverQueryResult } from '@slonik/driver';
+import { SlonikError } from '@slonik/errors';
 
 const executionRoutine: ExecutionRoutine = async (
   finalConnection,
@@ -47,12 +48,20 @@ export const query: InternalQueryMethod = async (
   slonikSql,
   inheritedQueryId,
 ) => {
-  return await executeQuery(
-    connectionLogger,
-    connection,
-    clientConfiguration,
-    slonikSql,
-    inheritedQueryId,
-    executionRoutine,
-  );
+  try {
+    await executeQuery(
+      connectionLogger,
+      connection,
+      clientConfiguration,
+      slonikSql,
+      inheritedQueryId,
+      executionRoutine,
+    );
+  } catch (error) {
+    if (error instanceof SlonikError) {
+      connection.events.emit('error', error);
+    }
+
+    throw error;
+  }
 };

packages/slonik/src/connectionMethods/stream.ts

@@ -9,6 +9,7 @@ import {
   type StreamHandler,
 } from '../types';
 import { type DriverStreamResult } from '@slonik/driver';
+import { SlonikError } from '@slonik/errors';
 import { Transform } from 'node:stream';
 import { pipeline } from 'node:stream/promises';
 
@@ -96,19 +97,29 @@ export const stream: InternalStreamFunction = async (
   onStream,
   uid,
 ) => {
-  const result = await executeQuery(
-    connectionLogger,
-    connection,
-    clientConfiguration,
-    slonikSql,
-    uid,
-    createExecutionRoutine(clientConfiguration, onStream),
-    true,
-  );
-
-  if (result.type === 'QueryResult') {
-    throw new Error('Query result cannot be returned in a streaming context.');
-  }
+  try {
+    const result = await executeQuery(
+      connectionLogger,
+      connection,
+      clientConfiguration,
+      slonikSql,
+      uid,
+      createExecutionRoutine(clientConfiguration, onStream),
+      true,
+    );
+
+    if (result.type === 'QueryResult') {
+      throw new Error(
+        'Query result cannot be returned in a streaming context.',
+      );
+    }
+
+    return result;
+  } catch (error) {
+    if (error instanceof SlonikError) {
+      connection.events.emit('error', error);
+    }
 
-  return result;
+    throw error;
+  }
 };

packages/slonik/src/factories/createConnectionPool.ts

@@ -1,4 +1,5 @@
 import { Logger } from '../Logger';
+import { type DatabasePoolEventEmitter } from '../types';
 import {
   type Driver,
   type DriverClientEventEmitter,
@@ -18,6 +19,7 @@ const logger = Logger.child({
 export type ConnectionPoolClient = {
   acquire: () => void;
   destroy: () => Promise<void>;
+  events: DatabasePoolEventEmitter;
   id: () => string;
   off: DriverClientEventEmitter['off'];
   on: DriverClientEventEmitter['on'];
@@ -58,10 +60,12 @@ export type ConnectionPool = {
 
 export const createConnectionPool = ({
   driver,
+  events,
   maximumPoolSize,
   minimumPoolSize,
 }: {
   driver: Driver;
+  events: DatabasePoolEventEmitter;
   idleTimeout: number;
   maximumPoolSize: number;
   minimumPoolSize: number;
@@ -127,7 +131,15 @@ export const createConnectionPool = ({
     }
 
     const addConnection = async () => {
-      const pendingConnection = driver.createClient();
+      const pendingConnection = driver
+        .createClient()
+        // eslint-disable-next-line promise/prefer-await-to-then
+        .then((resolvedConnection) => {
+          return {
+            ...resolvedConnection,
+            events,
+          };
+        });
 
       pendingConnections.push(pendingConnection);
 

packages/slonik/src/factories/createPool.ts

@@ -1,11 +1,16 @@
 import { bindPool } from '../binders/bindPool';
 import { Logger } from '../Logger';
-import { type ClientConfigurationInput, type DatabasePool } from '../types';
+import {
+  type ClientConfigurationInput,
+  type DatabasePool,
+  type DatabasePoolEventEmitter,
+} from '../types';
 import { createClientConfiguration } from './createClientConfiguration';
 import { createConnectionPool } from './createConnectionPool';
 import { createPoolConfiguration } from './createPoolConfiguration';
 import { type DriverFactory } from '@slonik/driver';
 import { createPgDriverFactory } from '@slonik/pg-driver';
+import EventEmitter from 'node:events';
 
 /**
  * @param connectionUri PostgreSQL [Connection URI](https://www.postgresql.org/docs/current/libpq-connect.html#LIBPQ-CONNSTRING).
@@ -27,12 +32,16 @@ export const createPool = async (
     driverConfiguration: clientConfiguration,
   });
 
+  const events = new EventEmitter() as DatabasePoolEventEmitter;
+
   const pool = createConnectionPool({
     driver,
+    events,
     ...createPoolConfiguration(clientConfiguration),
   });
 
   return bindPool(
+    events,
     Logger.child({
       poolId: pool.id(),
     }),

packages/slonik/src/helpers.test/createIntegrationTests.ts

@@ -230,6 +230,21 @@ export const createIntegrationTests = (
     await pool.end();
   });
 
+  test('emits thrown errors', async (t) => {
+    const pool = await createPool(t.context.dsn, {
+      driverFactory,
+    });
+
+    const onError = sinon.spy();
+
+    pool.on('error', onError);
+
+    await t.throwsAsync(pool.any(sql.unsafe`SELECT WHERE`));
+
+    t.is(onError.callCount, 1);
+    t.true(onError.firstCall.args[0] instanceof InputSyntaxError);
+  });
+
   test('retrieves correct infinity values (with timezone)', async (t) => {
     const pool = await createPool(t.context.dsn, {
       driverFactory,

packages/slonik/src/index.ts

@@ -21,6 +21,7 @@ export type {
   DatabaseConnection,
   DatabasePool,
   DatabasePoolConnection,
+  DatabasePoolEventEmitter,
   DatabaseTransactionConnection,
   Field,
   IdentifierNormalizer,

packages/slonik/src/types.ts

@@ -11,8 +11,10 @@ import {
   type QuerySqlToken,
   type SqlToken,
 } from '@slonik/sql-tag';
+import type EventEmitter from 'node:events';
 import { type ConnectionOptions as TlsConnectionOptions } from 'node:tls';
 import { type Logger } from 'roarr';
+import { type StrictEventEmitter } from 'strict-event-emitter-types';
 import { type z, type ZodTypeAny } from 'zod';
 
 export type StreamHandler<T> = (stream: DriverStream<T>) => void;
@@ -204,12 +206,20 @@ type PoolState = {
   readonly waitingClients: number;
 };
 
+export type DatabasePoolEventEmitter = StrictEventEmitter<
+  EventEmitter,
+  {
+    error: (error: SlonikError) => void;
+  }
+>;
+
 export type DatabasePool = {
   readonly configuration: ClientConfiguration;
   readonly connect: <T>(connectionRoutine: ConnectionRoutine<T>) => Promise<T>;
   readonly end: () => Promise<void>;
   readonly state: () => PoolState;
-} & CommonQueryMethods;
+} & CommonQueryMethods &
+  DatabasePoolEventEmitter;
 
 export type DatabaseConnection = DatabasePool | DatabasePoolConnection;