From 33e490d6c2bec274d7ef2ea348256eb226cbf852 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Dimitri=20Lavren=C3=BCk?= <20122620+dlavrenuek@users.noreply.github.com> Date: Wed, 8 Jul 2026 08:54:14 +0200 Subject: [PATCH] fix: Chat works when running on separate webhook server (#33724) --- .../webhooks/__tests__/webhook-server.test.ts | 54 +++++++ packages/cli/src/webhooks/webhook-server.ts | 6 + .../containers/services/load-balancer.ts | 7 +- .../webhook-proc/chat-send-and-wait.spec.ts | 138 ++++++++++++++++++ .../workflows/chat-send-and-wait.json | 74 ++++++++++ 5 files changed, 277 insertions(+), 2 deletions(-) create mode 100644 packages/cli/src/webhooks/__tests__/webhook-server.test.ts create mode 100644 packages/testing/playwright/tests/infrastructure/webhook-proc/chat-send-and-wait.spec.ts create mode 100644 packages/testing/playwright/workflows/chat-send-and-wait.json diff --git a/packages/cli/src/webhooks/__tests__/webhook-server.test.ts b/packages/cli/src/webhooks/__tests__/webhook-server.test.ts new file mode 100644 index 00000000000..f45efbb02a5 --- /dev/null +++ b/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(); +vi.mock('express', async () => ({ __esModule: true, default: () => mockApp })); + +describe('WebhookServer', () => { + it('should mount the chat WebSocket server', () => { + mockInstance(Logger); + Container.set(DbConnection, mock()); + Container.set( + GlobalConfig, + mock({ + 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(); + Object.assign(webhookServer, { server: httpServer }); + + webhookServer['setupPushServer'](); + + expect(chatServer.setup).toHaveBeenCalledWith(httpServer, mockApp); + }); +}); diff --git a/packages/cli/src/webhooks/webhook-server.ts b/packages/cli/src/webhooks/webhook-server.ts index 7f6ae983692..644450514a5 100644 --- a/packages/cli/src/webhooks/webhook-server.ts +++ b/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); + } } diff --git a/packages/testing/containers/services/load-balancer.ts b/packages/testing/containers/services/load-balancer.ts index 6f254470a81..a9da330cd04 100644 --- a/packages/testing/containers/services/load-balancer.ts +++ b/packages/testing/containers/services/load-balancer.ts @@ -20,8 +20,11 @@ export interface LoadBalancerMeta { export type LoadBalancerResult = ServiceResult; // 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[], diff --git a/packages/testing/playwright/tests/infrastructure/webhook-proc/chat-send-and-wait.spec.ts b/packages/testing/playwright/tests/infrastructure/webhook-proc/chat-send-and-wait.spec.ts new file mode 100644 index 00000000000..35c32a72088 --- /dev/null +++ b/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) => { + 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((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((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(); + } + }); + }, +); diff --git a/packages/testing/playwright/workflows/chat-send-and-wait.json b/packages/testing/playwright/workflows/chat-send-and-wait.json new file mode 100644 index 00000000000..905d65b2be3 --- /dev/null +++ b/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": {} +}