fix: Chat works when running on separate webhook server (#33724)

packages/cli/src/webhooks/__tests__/webhook-server.test.ts

@@ -0,0 +1,54 @@
+import { Logger } from '@n8n/backend-common';
+import { mockInstance } from '@n8n/backend-test-utils';
+import { GlobalConfig } from '@n8n/config';
+import { DbConnection } from '@n8n/db';
+import { Container } from '@n8n/di';
+import type express from 'express';
+import type * as http from 'node:http';
+import { mock } from 'vitest-mock-extended';
+
+import { ChatServer } from '@/chat/chat-server';
+import { WebhookServer } from '@/webhooks/webhook-server';
+
+const mockApp = mock<express.Application>();
+vi.mock('express', async () => ({ __esModule: true, default: () => mockApp }));
+
+describe('WebhookServer', () => {
+	it('should mount the chat WebSocket server', () => {
+		mockInstance(Logger);
+		Container.set(DbConnection, mock<DbConnection>());
+		Container.set(
+			GlobalConfig,
+			mock<GlobalConfig>({
+				path: '/',
+				protocol: 'http',
+				port: 5678,
+				listen_address: '0.0.0.0',
+				proxy_hops: 0,
+				ssl_key: '',
+				ssl_cert: '',
+				endpoints: {
+					rest: 'rest',
+					form: 'form',
+					formTest: 'form-test',
+					formWaiting: 'form-waiting',
+					webhook: 'webhook',
+					webhookTest: 'webhook-test',
+					webhookWaiting: 'webhook-waiting',
+					mcp: 'mcp',
+					mcpTest: 'mcp-test',
+					health: '/healthz',
+				},
+			}),
+		);
+		const chatServer = mockInstance(ChatServer);
+
+		const webhookServer = new WebhookServer();
+		const httpServer = mock<http.Server>();
+		Object.assign(webhookServer, { server: httpServer });
+
+		webhookServer['setupPushServer']();
+
+		expect(chatServer.setup).toHaveBeenCalledWith(httpServer, mockApp);
+	});
+});

packages/cli/src/webhooks/webhook-server.ts

@@ -1,6 +1,7 @@
 import { Container, Service } from '@n8n/di';
 
 import { AbstractServer } from '@/abstract-server';
+import { ChatServer } from '@/chat/chat-server';
 
 @Service()
 export class WebhookServer extends AbstractServer {
@@ -11,4 +12,9 @@ export class WebhookServer extends AbstractServer {
 			Container.get(PrometheusMetricsService).init(this.app);
 		}
 	}
+
+	/** The chat widget opens its WebSocket against the same origin that served the chat webhook */
+	protected setupPushServer() {
+		Container.get(ChatServer).setup(this.server, this.app);
+	}
 }

packages/testing/containers/services/load-balancer.ts

@@ -20,8 +20,11 @@ export interface LoadBalancerMeta {
 export type LoadBalancerResult = ServiceResult<LoadBalancerMeta>;
 
 // Production paths the `n8n webhook` proc serves. Test/waiting/form-test paths
-// stay on main, per `packages/cli/src/commands/webhook.ts`.
-const WEBHOOK_PROC_PATHS = ['/webhook/*', '/form/*'] as const;
+// stay on main, per `packages/cli/src/commands/webhook.ts`. `/chat` is the chat
+// widget's WebSocket endpoint: the widget always connects to the origin that
+// served the chat webhook, so it must route to the webhook procs too — same as
+// production setups where WEBHOOK_URL points at the dedicated webhook server.
+const WEBHOOK_PROC_PATHS = ['/webhook/*', '/form/*', '/chat'] as const;
 
 function buildCaddyConfig(
 	mainUpstreams: string[],

packages/testing/playwright/tests/infrastructure/webhook-proc/chat-send-and-wait.spec.ts

@@ -0,0 +1,138 @@
+import type { IWorkflowBase } from 'n8n-workflow';
+import { nanoid } from 'nanoid';
+
+import { test, expect } from '../../../fixtures/base';
+
+const HEARTBEAT = 'n8n|heartbeat';
+const HEARTBEAT_ACK = 'n8n|heartbeat-ack';
+const QUESTION = 'What is your name?';
+
+test.use({ capability: { webhooks: 1, workers: 1 } });
+
+function withChatWebhookId(webhookId: string) {
+	return (workflow: Partial<IWorkflowBase>) => {
+		workflow.nodes?.forEach((node) => {
+			if (node.type === '@n8n/n8n-nodes-langchain.chatTrigger') {
+				node.webhookId = webhookId;
+			}
+		});
+		return workflow;
+	};
+}
+
+function buildChatWsUrl(
+	backendUrl: string,
+	params: { sessionId: string; executionId: string; resumeToken?: string },
+) {
+	const url = new URL('/chat', backendUrl);
+	url.protocol = url.protocol.replace('http', 'ws');
+	url.searchParams.set('sessionId', params.sessionId);
+	url.searchParams.set('executionId', params.executionId);
+	url.searchParams.set('isPublic', 'true');
+	if (params.resumeToken) url.searchParams.set('token', params.resumeToken);
+	return url.toString();
+}
+
+function openChatSocket(wsUrl: string) {
+	const socket = new WebSocket(wsUrl);
+	const received: string[] = [];
+	const waiters: Array<{
+		predicate: (message: string) => boolean;
+		resolve: (message: string) => void;
+	}> = [];
+
+	socket.addEventListener('message', (event) => {
+		const message = String(event.data);
+		if (message === HEARTBEAT) {
+			socket.send(HEARTBEAT_ACK);
+			return;
+		}
+		received.push(message);
+		const index = waiters.findIndex(({ predicate }) => predicate(message));
+		if (index !== -1) waiters.splice(index, 1)[0].resolve(message);
+	});
+
+	const opened = new Promise<void>((resolve, reject) => {
+		socket.addEventListener('open', () => resolve());
+		socket.addEventListener('error', () =>
+			reject(new Error(`WebSocket connection to ${wsUrl} failed`)),
+		);
+	});
+
+	const waitForMessage = async (predicate: (message: string) => boolean, timeoutMs = 30_000) => {
+		const existing = received.find(predicate);
+		if (existing) return existing;
+
+		return await new Promise<string>((resolve, reject) => {
+			const waiter = {
+				predicate,
+				resolve: (message: string) => {
+					clearTimeout(timeout);
+					resolve(message);
+				},
+			};
+			const timeout = setTimeout(() => {
+				waiters.splice(waiters.indexOf(waiter), 1);
+				reject(
+					new Error(
+						`No matching chat message within ${timeoutMs}ms. Received: ${JSON.stringify(received)}`,
+					),
+				);
+			}, timeoutMs);
+			waiters.push(waiter);
+		});
+	};
+
+	return { socket, opened, waitForMessage };
+}
+
+test.describe(
+	'Chat sendAndWait with dedicated webhook process @mode:queue',
+	{ annotation: [{ type: 'owner', description: 'NODES' }] },
+	() => {
+		test('should deliver the sendAndWait message over WebSocket and resume the execution on reply', async ({
+			api,
+			backendUrl,
+		}) => {
+			const webhookId = nanoid();
+			const { workflowId } = await api.workflows.importWorkflowFromFile('chat-send-and-wait.json', {
+				transform: withChatWebhookId(webhookId),
+			});
+
+			const sessionId = nanoid();
+			const triggerResponse = await api.webhooks.trigger(`/webhook/${webhookId}/chat`, {
+				method: 'POST',
+				data: { action: 'sendMessage', sessionId, chatInput: 'hello' },
+			});
+			expect(triggerResponse.ok()).toBe(true);
+
+			const { executionId, resumeToken } = (await triggerResponse.json()) as {
+				executionId: string;
+				resumeToken?: string;
+			};
+			expect(executionId).toBeTruthy();
+
+			await api.workflows.waitForWorkflowStatus(workflowId, 'waiting', 30_000);
+
+			const chat = openChatSocket(
+				buildChatWsUrl(backendUrl, { sessionId, executionId, resumeToken }),
+			);
+
+			try {
+				await chat.opened;
+
+				const question = await chat.waitForMessage((message) => message.includes(QUESTION));
+				expect(question).toContain(QUESTION);
+
+				chat.socket.send(
+					JSON.stringify({ action: 'sendMessage', sessionId, chatInput: 'E2E reply' }),
+				);
+
+				const execution = await api.workflows.waitForWorkflowStatus(workflowId, 'success', 30_000);
+				expect(execution.status).toBe('success');
+			} finally {
+				chat.socket.close();
+			}
+		});
+	},
+);

packages/testing/playwright/workflows/chat-send-and-wait.json

@@ -0,0 +1,74 @@
+{
+	"name": "Chat sendAndWait",
+	"active": true,
+	"nodes": [
+		{
+			"parameters": {
+				"public": true,
+				"mode": "hostedChat",
+				"authentication": "none",
+				"initialMessages": "",
+				"options": {
+					"responseMode": "responseNodes"
+				}
+			},
+			"id": "d3a7f2b1-5c4e-4b8a-9f01-2e6d8c7a1b23",
+			"name": "When chat message received",
+			"type": "@n8n/n8n-nodes-langchain.chatTrigger",
+			"typeVersion": 1.4,
+			"position": [0, 0],
+			"webhookId": "chat-send-and-wait-trigger"
+		},
+		{
+			"parameters": {
+				"operation": "sendAndWait",
+				"message": "What is your name?",
+				"responseType": "freeTextChat",
+				"options": {}
+			},
+			"id": "a1b2c3d4-e5f6-4a7b-8c9d-0e1f2a3b4c5d",
+			"name": "Ask question",
+			"type": "@n8n/n8n-nodes-langchain.chat",
+			"typeVersion": 1.3,
+			"position": [220, 0],
+			"webhookId": "f0e9d8c7-b6a5-4433-9211-0f1e2d3c4b5a"
+		},
+		{
+			"parameters": {
+				"mode": "raw",
+				"jsonOutput": "={ \"reply\": \"{{ $json.chatInput }}\" }",
+				"options": {}
+			},
+			"id": "b2c3d4e5-f6a7-4b8c-9d0e-1f2a3b4c5d6e",
+			"name": "Capture Result",
+			"type": "n8n-nodes-base.set",
+			"typeVersion": 3.4,
+			"position": [440, 0]
+		}
+	],
+	"connections": {
+		"When chat message received": {
+			"main": [
+				[
+					{
+						"node": "Ask question",
+						"type": "main",
+						"index": 0
+					}
+				]
+			]
+		},
+		"Ask question": {
+			"main": [
+				[
+					{
+						"node": "Capture Result",
+						"type": "main",
+						"index": 0
+					}
+				]
+			]
+		}
+	},
+	"pinData": {}
+}