refactor(core): Extract the webhook Buffer body codec into n8n-core (no-changelog) (#39301)

Co-authored-by: Claude Fable 5.1 <noreply@anthropic.com>

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, {

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<JobQueue>({
 	client: { ping: vi.fn() },

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);
 	});
 });

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 },

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<T>(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' &&

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';
 

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);
+	});
+});

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<T extends IN8nHttpFullResponse>(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<T>(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;
+}

packages/core/src/webhook-response/index.ts

@@ -0,0 +1 @@
+export { ENCODED_BUFFER_KEY, encodeBufferBody, decodeBufferBody } from './buffer-body-codec';