refactor(Postgres Node): Backport connection pooling to postgres v1 (#…

…12484)

packages/@n8n/nodes-langchain/nodes/memory/MemoryPostgresChat/MemoryPostgresChat.node.ts

@@ -1,9 +1,9 @@
 /* eslint-disable n8n-nodes-base/node-dirname-against-convention */
 import { PostgresChatMessageHistory } from '@langchain/community/stores/message/postgres';
 import { BufferMemory, BufferWindowMemory } from 'langchain/memory';
+import { configurePostgres } from 'n8n-nodes-base/dist/nodes/Postgres/transport';
 import type { PostgresNodeCredentials } from 'n8n-nodes-base/dist/nodes/Postgres/v2/helpers/interfaces';
 import { postgresConnectionTest } from 'n8n-nodes-base/dist/nodes/Postgres/v2/methods/credentialTest';
-import { configurePostgres } from 'n8n-nodes-base/dist/nodes/Postgres/v2/transport';
 import type {
 	ISupplyDataFunctions,
 	INodeType,

packages/@n8n/nodes-langchain/nodes/vector_store/VectorStorePGVector/VectorStorePGVector.node.ts

@@ -4,8 +4,8 @@ import {
 	type PGVectorStoreArgs,
 } from '@langchain/community/vectorstores/pgvector';
 import type { EmbeddingsInterface } from '@langchain/core/embeddings';
+import { configurePostgres } from 'n8n-nodes-base/dist/nodes/Postgres/transport';
 import type { PostgresNodeCredentials } from 'n8n-nodes-base/dist/nodes/Postgres/v2/helpers/interfaces';
-import { configurePostgres } from 'n8n-nodes-base/dist/nodes/Postgres/v2/transport';
 import type { INodeProperties } from 'n8n-workflow';
 import type pg from 'pg';
 

packages/nodes-base/credentials/Postgres.credentials.ts

@@ -37,6 +37,14 @@ export class Postgres implements ICredentialType {
 			},
 			default: '',
 		},
+		{
+			displayName: 'Maximum Number of Connections',
+			name: 'maxConnections',
+			type: 'number',
+			default: 100,
+			description:
+				'Make sure this value times the number of workers you have is lower than the maximum number of connections your postgres instance allows.',
+		},
 		{
 			displayName: 'Ignore SSL Issues (Insecure)',
 			name: 'allowUnauthorizedCerts',

packages/nodes-base/nodes/Postgres/PostgresTrigger.functions.ts

@@ -7,8 +7,8 @@ import type {
 	INodeListSearchItems,
 } from 'n8n-workflow';
 
+import { configurePostgres } from './transport';
 import type { PgpDatabase, PostgresNodeCredentials } from './v2/helpers/interfaces';
-import { configurePostgres } from './v2/transport';
 
 export function prepareNames(id: string, mode: string, additionalFields: IDataObject) {
 	let suffix = id.replace(/-/g, '_');

packages/nodes-base/nodes/Postgres/v2/transport/index.ts → packages/nodes-base/nodes/Postgres/transport/index.ts

@@ -16,7 +16,7 @@ import type {
 	PgpConnectionParameters,
 	PostgresNodeCredentials,
 	PostgresNodeOptions,
-} from '../helpers/interfaces';
+} from '../v2/helpers/interfaces';
 
 const getPostgresConfig = (
 	credentials: PostgresNodeCredentials,
@@ -29,6 +29,7 @@ const getPostgresConfig = (
 		user: credentials.user,
 		password: credentials.password,
 		keepAlive: true,
+		max: credentials.maxConnections,
 	};
 
 	if (options.connectionTimeout) {

packages/nodes-base/nodes/Postgres/v1/PostgresV1.node.ts

@@ -1,7 +1,6 @@
 import type {
 	ICredentialsDecrypted,
 	ICredentialTestFunctions,
-	IDataObject,
 	IExecuteFunctions,
 	INodeCredentialTestResult,
 	INodeExecutionData,
@@ -10,11 +9,12 @@ import type {
 	INodeTypeDescription,
 } from 'n8n-workflow';
 import { NodeConnectionType, NodeOperationError } from 'n8n-workflow';
-import pgPromise from 'pg-promise';
 
 import { oldVersionNotice } from '@utils/descriptions';
 
 import { pgInsertV2, pgQueryV2, pgUpdate, wrapData } from './genericFunctions';
+import { configurePostgres } from '../transport';
+import type { PgpConnection, PostgresNodeCredentials } from '../v2/helpers/interfaces';
 
 const versionDescription: INodeTypeDescription = {
 	displayName: 'Postgres',
@@ -298,33 +298,27 @@ export class PostgresV1 implements INodeType {
 				this: ICredentialTestFunctions,
 				credential: ICredentialsDecrypted,
 			): Promise<INodeCredentialTestResult> {
-				const credentials = credential.data as IDataObject;
-				try {
-					const pgp = pgPromise();
-					const config: IDataObject = {
-						host: credentials.host as string,
-						port: credentials.port as number,
-						database: credentials.database as string,
-						user: credentials.user as string,
-						password: credentials.password as string,
-					};
+				const credentials = credential.data as PostgresNodeCredentials;
 
-					if (credentials.allowUnauthorizedCerts === true) {
-						config.ssl = {
-							rejectUnauthorized: false,
-						};
-					} else {
-						config.ssl = !['disable', undefined].includes(credentials.ssl as string | undefined);
-						config.sslmode = (credentials.ssl as string) || 'disable';
-					}
+				let connection: PgpConnection | undefined;
 
-					const db = pgp(config);
-					await db.connect();
+				try {
+					const { db } = await configurePostgres.call(this, credentials, {});
+
+					// Acquires a new connection that can be used to to run multiple
+					// queries on the same connection and must be released again
+					// manually.
+					connection = await db.connect();
 				} catch (error) {
 					return {
 						status: 'Error',
 						message: error.message,
 					};
+				} finally {
+					if (connection) {
+						// release connection
+						await connection.done();
+					}
 				}
 				return {
 					status: 'OK',
@@ -335,42 +329,19 @@ export class PostgresV1 implements INodeType {
 	};
 
 	async execute(this: IExecuteFunctions): Promise<INodeExecutionData[][]> {
-		const credentials = await this.getCredentials('postgres');
+		const credentials = await this.getCredentials<PostgresNodeCredentials>('postgres');
 		const largeNumbersOutput = this.getNodeParameter(
 			'additionalFields.largeNumbersOutput',
 			0,
 			'',
 		) as string;
 
-		const pgp = pgPromise();
-
-		if (largeNumbersOutput === 'numbers') {
-			pgp.pg.types.setTypeParser(20, (value: string) => {
-				return parseInt(value, 10);
-			});
-			pgp.pg.types.setTypeParser(1700, (value: string) => {
-				return parseFloat(value);
-			});
-		}
-
-		const config: IDataObject = {
-			host: credentials.host as string,
-			port: credentials.port as number,
-			database: credentials.database as string,
-			user: credentials.user as string,
-			password: credentials.password as string,
-		};
-
-		if (credentials.allowUnauthorizedCerts === true) {
-			config.ssl = {
-				rejectUnauthorized: false,
-			};
-		} else {
-			config.ssl = !['disable', undefined].includes(credentials.ssl as string | undefined);
-			config.sslmode = (credentials.ssl as string) || 'disable';
-		}
-
-		const db = pgp(config);
+		const { db, pgp } = await configurePostgres.call(this, credentials, {
+			largeNumbersOutput:
+				largeNumbersOutput === 'numbers' || largeNumbersOutput === 'text'
+					? largeNumbersOutput
+					: undefined,
+		});
 
 		let returnItems: INodeExecutionData[] = [];
 

packages/nodes-base/nodes/Postgres/v2/actions/router.ts

@@ -3,9 +3,9 @@ import { NodeExecutionOutput, NodeOperationError } from 'n8n-workflow';
 
 import * as database from './database/Database.resource';
 import type { PostgresType } from './node.type';
+import { configurePostgres } from '../../transport';
 import type { PostgresNodeCredentials, PostgresNodeOptions } from '../helpers/interfaces';
 import { configureQueryRunner } from '../helpers/utils';
-import { configurePostgres } from '../transport';
 
 export async function router(this: IExecuteFunctions): Promise<INodeExecutionData[][]> {
 	let returnData: INodeExecutionData[] = [];

packages/nodes-base/nodes/Postgres/v2/helpers/interfaces.ts

@@ -28,6 +28,7 @@ export type EnumInfo = {
 export type PgpClient = pgPromise.IMain<{}, pg.IClient>;
 export type PgpDatabase = pgPromise.IDatabase<{}, pg.IClient>;
 export type PgpConnectionParameters = pg.IConnectionParameters<pg.IClient>;
+export type PgpConnection = pgPromise.IConnected<{}, pg.IClient>;
 export type ConnectionsData = { db: PgpDatabase; pgp: PgpClient };
 
 export type QueriesRunner = (
@@ -57,6 +58,7 @@ export type PostgresNodeCredentials = {
 	database: string;
 	user: string;
 	password: string;
+	maxConnections: number;
 	allowUnauthorizedCerts?: boolean;
 	ssl?: 'disable' | 'allow' | 'require' | 'verify' | 'verify-full';
 } & (

packages/nodes-base/nodes/Postgres/v2/methods/credentialTest.ts

@@ -4,23 +4,21 @@ import type {
 	INodeCredentialTestResult,
 } from 'n8n-workflow';
 
-import type { PgpClient, PostgresNodeCredentials } from '../helpers/interfaces';
-import { configurePostgres } from '../transport';
+import { configurePostgres } from '../../transport';
+import type { PgpConnection, PostgresNodeCredentials } from '../helpers/interfaces';
 
 export async function postgresConnectionTest(
 	this: ICredentialTestFunctions,
 	credential: ICredentialsDecrypted,
 ): Promise<INodeCredentialTestResult> {
 	const credentials = credential.data as PostgresNodeCredentials;
 
-	let pgpClientCreated: PgpClient | undefined;
+	let connection: PgpConnection | undefined;
 
 	try {
-		const { db, pgp } = await configurePostgres.call(this, credentials, {});
+		const { db } = await configurePostgres.call(this, credentials, {});
 
-		pgpClientCreated = pgp;
-
-		await db.connect();
+		connection = await db.connect();
 	} catch (error) {
 		let message = error.message as string;
 
@@ -41,8 +39,8 @@ export async function postgresConnectionTest(
 			message,
 		};
 	} finally {
-		if (pgpClientCreated) {
-			pgpClientCreated.end();
+		if (connection) {
+			await connection.done();
 		}
 	}
 	return {

packages/nodes-base/nodes/Postgres/v2/methods/listSearch.ts

@@ -1,7 +1,7 @@
 import type { ILoadOptionsFunctions, INodeListSearchResult } from 'n8n-workflow';
 
+import { configurePostgres } from '../../transport';
 import type { PostgresNodeCredentials } from '../helpers/interfaces';
-import { configurePostgres } from '../transport';
 
 export async function schemaSearch(this: ILoadOptionsFunctions): Promise<INodeListSearchResult> {
 	const credentials = await this.getCredentials<PostgresNodeCredentials>('postgres');

packages/nodes-base/nodes/Postgres/v2/methods/loadOptions.ts

@@ -1,8 +1,8 @@
 import type { ILoadOptionsFunctions, INodePropertyOptions } from 'n8n-workflow';
 
+import { configurePostgres } from '../../transport';
 import type { PostgresNodeCredentials } from '../helpers/interfaces';
 import { getTableSchema } from '../helpers/utils';
-import { configurePostgres } from '../transport';
 
 export async function getColumns(this: ILoadOptionsFunctions): Promise<INodePropertyOptions[]> {
 	const credentials = await this.getCredentials<PostgresNodeCredentials>('postgres');

packages/nodes-base/nodes/Postgres/v2/methods/resourceMapping.ts

@@ -1,8 +1,8 @@
 import type { ILoadOptionsFunctions, ResourceMapperFields, FieldType } from 'n8n-workflow';
 
+import { configurePostgres } from '../../transport';
 import type { PostgresNodeCredentials } from '../helpers/interfaces';
 import { getEnumValues, getEnums, getTableSchema, uniqueColumns } from '../helpers/utils';
-import { configurePostgres } from '../transport';
 
 const fieldTypeMapping: Partial<Record<FieldType, string[]>> = {
 	string: ['text', 'varchar', 'character varying', 'character', 'char'],