From 4f680fdfcb1ddf9eb1666decdc9b6d4e8e90f0d4 Mon Sep 17 00:00:00 2001 From: Pierantonio Cangianiello <10360978+pcan@users.noreply.github.com> Date: Tue, 22 Sep 2026 15:09:39 +0000 Subject: [PATCH] refactor(core): Extract the webhook Buffer body codec into n8n-core (no-changelog) (#39301) Co-authored-by: Claude Fable 5.1 --- .../__tests__/job-processor.service.test.ts | 4 +- .../scaling/__tests__/scaling.service.test.ts | 4 +- .../__tests__/webhook-response-relay.test.ts | 19 ++-- packages/cli/src/scaling/scaling.service.ts | 8 +- .../cli/src/scaling/webhook-response-relay.ts | 52 ++--------- packages/core/src/index.ts | 1 + .../__tests__/buffer-body-codec.test.ts | 90 +++++++++++++++++++ .../src/webhook-response/buffer-body-codec.ts | 58 ++++++++++++ packages/core/src/webhook-response/index.ts | 1 + 9 files changed, 173 insertions(+), 64 deletions(-) create mode 100644 packages/core/src/webhook-response/__tests__/buffer-body-codec.test.ts create mode 100644 packages/core/src/webhook-response/buffer-body-codec.ts create mode 100644 packages/core/src/webhook-response/index.ts diff --git a/packages/cli/src/scaling/__tests__/job-processor.service.test.ts b/packages/cli/src/scaling/__tests__/job-processor.service.test.ts index 91035088d83..a9bdbdcb8d3 100644 --- a/packages/cli/src/scaling/__tests__/job-processor.service.test.ts +++ b/packages/cli/src/scaling/__tests__/job-processor.service.test.ts @@ -9,7 +9,7 @@ import type { BinaryDataService, InstanceSettings, } from 'n8n-core'; -import { ExternalSecretsProxy, StructuredToolkit } from 'n8n-core'; +import { ENCODED_BUFFER_KEY, ExternalSecretsProxy, StructuredToolkit } from 'n8n-core'; import { mockInstance } from 'n8n-core/test/utils'; import { type IPinData, @@ -52,7 +52,7 @@ import { WorkflowStaticDataService } from '@/workflows/workflow-static-data.serv import { JobProcessor } from '../job-processor'; import type { Job } from '../scaling.types'; -import { ENCODED_BUFFER_KEY, WebhookResponseRelay } from '../webhook-response-relay'; +import { WebhookResponseRelay } from '../webhook-response-relay'; mockInstance(WorkflowPublishHistoryRepository); mockInstance(VariablesService, { diff --git a/packages/cli/src/scaling/__tests__/scaling.service.test.ts b/packages/cli/src/scaling/__tests__/scaling.service.test.ts index c6f520b396f..d27c1b885f8 100644 --- a/packages/cli/src/scaling/__tests__/scaling.service.test.ts +++ b/packages/cli/src/scaling/__tests__/scaling.service.test.ts @@ -4,7 +4,7 @@ import { GlobalConfig, WorkerPoolConfig } from '@n8n/config'; import type { ExecutionRepository } from '@n8n/db'; import { Container } from '@n8n/di'; import * as BullModule from 'bull'; -import { InstanceSettings } from 'n8n-core'; +import { ENCODED_BUFFER_KEY, InstanceSettings } from 'n8n-core'; import type { ErrorReporter } from 'n8n-core'; import { UnexpectedError } from 'n8n-workflow'; import type { MockInstance } from 'vitest'; @@ -18,7 +18,7 @@ import { JOB_TYPE_NAME } from '../constants'; import type { JobProcessor } from '../job-processor'; import { ScalingService } from '../scaling.service'; import type { Job, JobData, JobId, JobQueue } from '../scaling.types'; -import { ENCODED_BUFFER_KEY, type WebhookResponseRelay } from '../webhook-response-relay'; +import type { WebhookResponseRelay } from '../webhook-response-relay'; const queue = mock({ client: { ping: vi.fn() }, diff --git a/packages/cli/src/scaling/__tests__/webhook-response-relay.test.ts b/packages/cli/src/scaling/__tests__/webhook-response-relay.test.ts index f32d3253547..a82a559243c 100644 --- a/packages/cli/src/scaling/__tests__/webhook-response-relay.test.ts +++ b/packages/cli/src/scaling/__tests__/webhook-response-relay.test.ts @@ -1,6 +1,6 @@ import type { Logger } from '@n8n/backend-common'; import type { ExecutionsConfig } from '@n8n/config'; -import { FileLocation, FileTooLargeError } from 'n8n-core'; +import { decodeBufferBody, ENCODED_BUFFER_KEY, FileLocation, FileTooLargeError } from 'n8n-core'; import type { BinaryDataConfig, BinaryDataService } from 'n8n-core'; import { OperationalError } from 'n8n-workflow'; import type { IBinaryData, IExecuteResponsePromiseData } from 'n8n-workflow'; @@ -9,12 +9,7 @@ import { mock } from 'vitest-mock-extended'; import { WebhookResponseTooLargeError } from '@/errors/webhook-response-too-large.error'; -import { - decodeRelayedWebhookResponse, - ENCODED_BUFFER_KEY, - OFFLOADED_BODY_KIND_KEY, - WebhookResponseRelay, -} from '../webhook-response-relay'; +import { OFFLOADED_BODY_KIND_KEY, WebhookResponseRelay } from '../webhook-response-relay'; const ONE_MIB = 1024 * 1024; const SIZE_MAX_IN_MIB = 2; @@ -751,11 +746,11 @@ describe('WebhookResponseRelay', () => { }); }); -describe('decodeRelayedWebhookResponse', () => { +describe('decodeBufferBody on a relayed response', () => { it('restores a base64 envelope to a Buffer', () => { const response = fullResponse({ [ENCODED_BUFFER_KEY]: 'aGVsbG8=' }); - const decoded = decodeRelayedWebhookResponse(response); + const decoded = decodeBufferBody(response); expect(decoded).toBe(response); expect(bodyOf(decoded)).toEqual(Buffer.from('hello')); @@ -765,7 +760,7 @@ describe('decodeRelayedWebhookResponse', () => { const { relay } = buildRelay(); const body = Buffer.from([0x00, 0xff, 0x10]); - const decoded = decodeRelayedWebhookResponse(await relay.prepare(fullResponse(body), ctx)); + const decoded = decodeBufferBody(await relay.prepare(fullResponse(body), ctx)); expect(bodyOf(decoded)).toEqual(body); }); @@ -780,7 +775,7 @@ describe('decodeRelayedWebhookResponse', () => { ], ['an envelope whose payload is not a string', { [ENCODED_BUFFER_KEY]: 42 }], ])('leaves %s untouched', (_label, body) => { - const decoded = decodeRelayedWebhookResponse(fullResponse(body)); + const decoded = decodeBufferBody(fullResponse(body)); expect(bodyOf(decoded)).toEqual(body); }); @@ -788,6 +783,6 @@ describe('decodeRelayedWebhookResponse', () => { it('leaves a payload that is not a full response untouched', () => { const payload = { toolResult: 'done' }; - expect(decodeRelayedWebhookResponse(payload)).toBe(payload); + expect(decodeBufferBody(payload)).toBe(payload); }); }); diff --git a/packages/cli/src/scaling/scaling.service.ts b/packages/cli/src/scaling/scaling.service.ts index b089355e7ae..6327757d028 100644 --- a/packages/cli/src/scaling/scaling.service.ts +++ b/packages/cli/src/scaling/scaling.service.ts @@ -4,7 +4,7 @@ import { Time } from '@n8n/constants'; import { ExecutionRepository } from '@n8n/db'; import { OnLeaderStepdown, OnLeaderTakeover, OnShutdown } from '@n8n/decorators'; import { Container, Service } from '@n8n/di'; -import { ErrorReporter, InstanceSettings } from 'n8n-core'; +import { decodeBufferBody, ErrorReporter, InstanceSettings } from 'n8n-core'; import { ensureError } from '@n8n/utils/errors/ensure-error'; import { sleep } from '@n8n/utils/sleep'; import { jsonStringify, UnexpectedError } from 'n8n-workflow'; @@ -33,7 +33,7 @@ import type { JobMessage, JobFailedMessage, } from './scaling.types'; -import { decodeRelayedWebhookResponse, WebhookResponseRelay } from './webhook-response-relay'; +import { WebhookResponseRelay } from './webhook-response-relay'; const DRAIN_POLL_INTERVAL_MS = 500; @@ -502,7 +502,7 @@ export class ScalingService { this.activeExecutions.sendChunk(msg.executionId, msg.chunkText); break; case 'respond-to-webhook': { - const decodedResponse = decodeRelayedWebhookResponse(msg.response); + const decodedResponse = decodeBufferBody(msg.response); this.activeExecutions.resolveResponsePromise(msg.executionId, decodedResponse); break; } @@ -642,7 +642,7 @@ export class ScalingService { // it. So the stored body is left in place, and execution pruning // reclaims it. Restoring under this guard only spares the mains that // would discard the body a read of the whole thing. - const decoded = decodeRelayedWebhookResponse(response); + const decoded = decodeBufferBody(response); const toolResult = await this.webhookResponseRelay.restoreOffloadedBody(decoded, { reclaim: false, context: { executionId }, diff --git a/packages/cli/src/scaling/webhook-response-relay.ts b/packages/cli/src/scaling/webhook-response-relay.ts index a9c929091ee..1c9443385a7 100644 --- a/packages/cli/src/scaling/webhook-response-relay.ts +++ b/packages/cli/src/scaling/webhook-response-relay.ts @@ -13,9 +13,15 @@ import { Logger } from '@n8n/backend-common'; import { ExecutionsConfig } from '@n8n/config'; import { Service } from '@n8n/di'; import { jsonSizeExceeds } from '@n8n/utils/json/json-size-exceeds'; -import { BinaryDataConfig, BinaryDataService, FileLocation, FileTooLargeError } from 'n8n-core'; +import { + BinaryDataConfig, + BinaryDataService, + encodeBufferBody, + FileLocation, + FileTooLargeError, +} from 'n8n-core'; import type { BinaryData } from 'n8n-core'; -import { BINARY_ENCODING, jsonParse, OperationalError } from 'n8n-workflow'; +import { jsonParse, OperationalError } from 'n8n-workflow'; import type { IBinaryData, IDataObject, @@ -28,9 +34,6 @@ import { WebhookResponseTooLargeError } from '@/errors/webhook-response-too-larg const MIB = 1024 * 1024; -/** Sentinel key marking a base64-encoded Buffer body relayed inline through the queue. */ -export const ENCODED_BUFFER_KEY = '__@N8nEncodedBuffer@__'; - /** * Sentinel key recording an offloaded body's original form, so it can be restored. * It sits on the relayed response, beside `body`, never inside it: @@ -405,27 +408,6 @@ export class WebhookResponseRelay { } } -/** - * Reverses the inline base64 envelope on main, restoring a Buffer body. Every - * other body passes through, an offloaded one included: main streams that from - * storage rather than materializing it. - * - * @param response Relayed response. Mutated and returned. - * @returns The same `response`. - */ -export function decodeRelayedWebhookResponse(response: T): T { - if (!hasResponseBody(response)) { - return response; - } - - const encoded = encodedBufferIn(response.body); - if (encoded !== undefined) { - response.body = Buffer.from(encoded, BINARY_ENCODING); - } - - return response; -} - function withInlineLimit(guidance: string, limitInMib: number): string { return `The limit is ${limitInMib} MiB. ${guidance}`; } @@ -443,14 +425,6 @@ function base64Size(byteLength: number): number { return Math.ceil(byteLength / 3) * 4; } -function encodeBufferBody(response: IN8nHttpFullResponse): IN8nHttpFullResponse { - if (Buffer.isBuffer(response.body)) { - response.body = { [ENCODED_BUFFER_KEY]: response.body.toString(BINARY_ENCODING) }; - } - - return response; -} - /** * Whether `payload` would serialize to more than `maxBytes` inside a queue * message. Everything is measured, offloadable or not — a binary-data-shaped @@ -548,16 +522,6 @@ function hasResponseBody(response: unknown): response is IN8nHttpFullResponse { return typeof response === 'object' && response !== null && 'body' in response; } -/** The base64 payload of an {@link ENCODED_BUFFER_KEY} envelope, if the body is one. */ -function encodedBufferIn(body: IN8nHttpFullResponse['body']): string | undefined { - if (typeof body !== 'object' || body === null || !(ENCODED_BUFFER_KEY in body)) { - return undefined; - } - - const encoded = body[ENCODED_BUFFER_KEY]; - return typeof encoded === 'string' ? encoded : undefined; -} - function isPlainJson(payload: unknown): payload is IDataObject { return ( typeof payload === 'object' && diff --git a/packages/core/src/index.ts b/packages/core/src/index.ts index 80d600141fc..e28779c1006 100644 --- a/packages/core/src/index.ts +++ b/packages/core/src/index.ts @@ -13,6 +13,7 @@ export * from './html-sandbox'; export * from './instance-settings'; export * from './nodes-loader'; export * from './utils'; +export * from './webhook-response'; export { WorkflowHasIssuesError } from './errors/workflow-has-issues.error'; export * from './observability'; diff --git a/packages/core/src/webhook-response/__tests__/buffer-body-codec.test.ts b/packages/core/src/webhook-response/__tests__/buffer-body-codec.test.ts new file mode 100644 index 00000000000..af0a2a0be04 --- /dev/null +++ b/packages/core/src/webhook-response/__tests__/buffer-body-codec.test.ts @@ -0,0 +1,90 @@ +import type { IN8nHttpFullResponse } from 'n8n-workflow'; +import { deepCopy } from 'n8n-workflow'; + +import { decodeBufferBody, ENCODED_BUFFER_KEY, encodeBufferBody } from '../buffer-body-codec'; + +const fullResponse = (body: unknown): IN8nHttpFullResponse => ({ + body: body as IN8nHttpFullResponse['body'], + headers: { 'content-type': 'application/octet-stream' }, + statusCode: 200, +}); + +describe('encodeBufferBody', () => { + it('replaces a Buffer body with a base64 envelope', () => { + const response = fullResponse(Buffer.from('hello')); + + const encoded = encodeBufferBody(response); + + expect(encoded).toBe(response); + expect(encoded.body).toEqual({ [ENCODED_BUFFER_KEY]: 'aGVsbG8=' }); + }); + + it('encodes an empty Buffer', () => { + const encoded = encodeBufferBody(fullResponse(Buffer.alloc(0))); + + expect(encoded.body).toEqual({ [ENCODED_BUFFER_KEY]: '' }); + }); + + it.each([ + ['a JSON body', { hello: 'world' }], + ['a string body', 'hello'], + ['a null body', null], + ['a binary-data reference', { binaryData: { id: 'filesystem-v2:abc' } }], + ])('leaves %s untouched', (_label, body) => { + const encoded = encodeBufferBody(fullResponse(body)); + + expect(encoded.body).toEqual(body); + }); + + it('keeps the headers and the status code', () => { + const encoded = encodeBufferBody(fullResponse(Buffer.from('hello'))); + + expect(encoded.headers).toEqual({ 'content-type': 'application/octet-stream' }); + expect(encoded.statusCode).toBe(200); + }); +}); + +describe('decodeBufferBody', () => { + it('restores a base64 envelope to a Buffer', () => { + const response = fullResponse({ [ENCODED_BUFFER_KEY]: 'aGVsbG8=' }); + + const decoded = decodeBufferBody(response); + + expect(decoded).toBe(response); + expect(decoded.body).toEqual(Buffer.from('hello')); + }); + + it('round-trips arbitrary bytes through JSON', () => { + const bytes = Buffer.from([0x00, 0xff, 0x10, 0x80]); + + // `deepCopy` is a JSON round trip, the same transform a JSON transport applies. + const relayed: unknown = deepCopy(encodeBufferBody(fullResponse(bytes))); + const decoded = decodeBufferBody(relayed) as IN8nHttpFullResponse; + + expect(decoded.body).toEqual(bytes); + }); + + it('round-trips an empty Buffer', () => { + const decoded = decodeBufferBody(encodeBufferBody(fullResponse(Buffer.alloc(0)))); + + expect(decoded.body).toEqual(Buffer.alloc(0)); + }); + + it.each([ + ['a JSON body', { hello: 'world' }], + ['a string body', 'hello'], + ['a null body', null], + ['a binary-data reference', { binaryData: { id: 'filesystem-v2:abc' } }], + ['an envelope whose payload is not a string', { [ENCODED_BUFFER_KEY]: 42 }], + ])('leaves %s untouched', (_label, body) => { + const decoded = decodeBufferBody(fullResponse(body)); + + expect(decoded.body).toEqual(body); + }); + + it('leaves a payload that is not a full response untouched', () => { + const payload = { toolResult: 'done' }; + + expect(decodeBufferBody(payload)).toBe(payload); + }); +}); diff --git a/packages/core/src/webhook-response/buffer-body-codec.ts b/packages/core/src/webhook-response/buffer-body-codec.ts new file mode 100644 index 00000000000..131991fcd5d --- /dev/null +++ b/packages/core/src/webhook-response/buffer-body-codec.ts @@ -0,0 +1,58 @@ +import type { IN8nHttpFullResponse } from 'n8n-workflow'; +import { BINARY_ENCODING } from 'n8n-workflow'; + +/** + * Sentinel key that marks a base64-encoded Buffer body sent inline through a + * JSON transport. `JSON.stringify` turns a bare Buffer into + * `{ type: 'Buffer', data: [...] }`, so a Buffer body has to be wrapped before + * it is serialized and unwrapped after it is parsed. + */ +export const ENCODED_BUFFER_KEY = '__@N8nEncodedBuffer@__'; + +/** + * Replaces a Buffer body with its base64 envelope. Any other body passes through. + * + * @param response Full response. Mutated and returned. + * @returns The same `response`. + */ +export function encodeBufferBody(response: T): T { + if (Buffer.isBuffer(response.body)) { + response.body = { [ENCODED_BUFFER_KEY]: response.body.toString(BINARY_ENCODING) }; + } + + return response; +} + +/** + * Reverses the base64 envelope, restoring a Buffer body. Every other value + * passes through, a payload that is not a full response included. + * + * @param response Parsed payload. Mutated and returned. + * @returns The same `response`. + */ +export function decodeBufferBody(response: T): T { + if (!hasResponseBody(response)) { + return response; + } + + const encoded = encodedBufferIn(response.body); + if (encoded !== undefined) { + response.body = Buffer.from(encoded, BINARY_ENCODING); + } + + return response; +} + +function hasResponseBody(response: unknown): response is IN8nHttpFullResponse { + return typeof response === 'object' && response !== null && 'body' in response; +} + +/** The base64 payload of an {@link ENCODED_BUFFER_KEY} envelope, if the body is one. */ +function encodedBufferIn(body: IN8nHttpFullResponse['body']): string | undefined { + if (typeof body !== 'object' || body === null || !(ENCODED_BUFFER_KEY in body)) { + return undefined; + } + + const encoded = body[ENCODED_BUFFER_KEY]; + return typeof encoded === 'string' ? encoded : undefined; +} diff --git a/packages/core/src/webhook-response/index.ts b/packages/core/src/webhook-response/index.ts new file mode 100644 index 00000000000..8fb3c3af30e --- /dev/null +++ b/packages/core/src/webhook-response/index.ts @@ -0,0 +1 @@ +export { ENCODED_BUFFER_KEY, encodeBufferBody, decodeBufferBody } from './buffer-body-codec';