feat(load): gzip large bulk batches and pass them through to Salesforce

apps/api/src/app/controllers/sf-bulk-api.controller.ts

@@ -142,13 +142,15 @@ const closeOrAbortJob = createRoute(routeDefinition.closeOrAbortJob.validators,
 
 const addBatchToJob = createRoute(
   routeDefinition.addBatchToJob.validators,
-  async ({ body, params, query, jetstreamConn }, _, res, next) => {
+  async ({ body, params, query, jetstreamConn }, req, res, next) => {
     try {
       const jobId = params.jobId;
       const csv = body;
       const closeJob = query.closeJob;
+      // gzipBodyPassthrough left the body compressed - Salesforce decodes it, this server never does
+      const contentEncoding = req.get(HTTP.HEADERS.CONTENT_ENCODING)?.toLowerCase() === 'gzip' ? 'gzip' : undefined;
 
-      const results = await jetstreamConn.bulk.addBatchToJob(csv, jobId, closeJob);
+      const results = await jetstreamConn.bulk.addBatchToJob(csv, jobId, closeJob, HTTP.CONTENT_TYPE.CSV, contentEncoding);
 
       sendJson(res, results);
     } catch (ex) {

apps/api/src/app/utils/__tests__/gzip-body-passthrough.middleware.spec.ts

@@ -0,0 +1,91 @@
+import { HTTP } from '@jetstream/shared/constants';
+import { raw } from 'body-parser';
+import express from 'express';
+import type { AddressInfo } from 'node:net';
+import { gzipSync } from 'node:zlib';
+import { gzipBodyPassthrough } from '../gzip-body-passthrough.middleware';
+
+const CSV = 'Name\nAcme\nGlobex\n';
+
+/** Mirrors main.ts: passthrough first, then the global parser that rejects every Content-Encoding */
+function createServer(limitBytes = 1024) {
+  const app = express();
+  app.post('/api/bulk/:jobId', gzipBodyPassthrough({ limitBytes }));
+  app.use(raw({ limit: '30mb', type: ['text/csv'], inflate: false }));
+  app.post('/api/bulk/:jobId', (req, res) => {
+    res.json({
+      isBuffer: Buffer.isBuffer(req.body),
+      bodyBase64: Buffer.isBuffer(req.body) ? req.body.toString('base64') : null,
+    });
+  });
+  app.use((err: { status?: number; type?: string }, _req: express.Request, res: express.Response, _next: express.NextFunction) => {
+    res.status(err.status ?? 500).json({ type: err.type });
+  });
+  return new Promise<{ url: string; close: () => void }>((resolve) => {
+    const server = app.listen(0, () => {
+      const { port } = server.address() as AddressInfo;
+      resolve({ url: `http://127.0.0.1:${port}`, close: () => server.close() });
+    });
+  });
+}
+
+function post(url: string, body: BodyInit, headers: Record<string, string>) {
+  return fetch(`${url}/api/bulk/750000000000001`, { method: 'POST', body, headers });
+}
+
+describe('gzipBodyPassthrough', () => {
+  let server: Awaited<ReturnType<typeof createServer>>;
+
+  afterEach(() => {
+    server?.close();
+  });
+
+  test('passes gzipped csv through as the original compressed bytes', async () => {
+    server = await createServer();
+    const compressed = gzipSync(CSV);
+    const response = await post(server.url, compressed, {
+      [HTTP.HEADERS.CONTENT_TYPE]: HTTP.CONTENT_TYPE.CSV,
+      [HTTP.HEADERS.CONTENT_ENCODING]: 'gzip',
+    });
+    expect(response.status).toBe(200);
+    const result = await response.json();
+    expect(result.isBuffer).toBe(true);
+    expect(Buffer.from(result.bodyBase64, 'base64').equals(compressed)).toBe(true);
+  });
+
+  test('leaves uncompressed csv to the regular parser', async () => {
+    server = await createServer();
+    const response = await post(server.url, CSV, { [HTTP.HEADERS.CONTENT_TYPE]: HTTP.CONTENT_TYPE.CSV });
+    expect(response.status).toBe(200);
+    const result = await response.json();
+    expect(Buffer.from(result.bodyBase64, 'base64').toString('utf8')).toBe(CSV);
+  });
+
+  test('rejects a body that is not gzip', async () => {
+    server = await createServer();
+    const response = await post(server.url, CSV, {
+      [HTTP.HEADERS.CONTENT_TYPE]: HTTP.CONTENT_TYPE.CSV,
+      [HTTP.HEADERS.CONTENT_ENCODING]: 'gzip',
+    });
+    expect(response.status).toBe(400);
+    expect((await response.json()).type).toBe('encoding.invalid');
+  });
+
+  test('rejects a compressed body over the limit', async () => {
+    server = await createServer(16);
+    const response = await post(server.url, gzipSync('x'.repeat(10_000) + CSV), {
+      [HTTP.HEADERS.CONTENT_TYPE]: HTTP.CONTENT_TYPE.CSV,
+      [HTTP.HEADERS.CONTENT_ENCODING]: 'gzip',
+    });
+    expect(response.status).toBe(413);
+  });
+
+  test('leaves other encodings to the global parser, which rejects them', async () => {
+    server = await createServer();
+    const response = await post(server.url, CSV, {
+      [HTTP.HEADERS.CONTENT_TYPE]: HTTP.CONTENT_TYPE.CSV,
+      [HTTP.HEADERS.CONTENT_ENCODING]: 'br',
+    });
+    expect(response.status).toBe(415);
+  });
+});

apps/api/src/app/utils/gzip-body-passthrough.middleware.ts

@@ -0,0 +1,48 @@
+import { HTTP } from '@jetstream/shared/constants';
+import express from 'express';
+
+function createBodyError(status: number, type: string, message: string) {
+  return Object.assign(new Error(message), { status, type, expose: true });
+}
+
+/**
+ * Reads a gzip-encoded CSV body into a Buffer WITHOUT decompressing it, so it can be forwarded to Salesforce
+ * as-is (Bulk API decodes gzip request bodies itself).
+ *
+ * The global body parsers reject every Content-Encoding (inflate: false) to keep untrusted input away from the
+ * zlib/brotli decoders, so this must be mounted before them - they skip any request whose body was already read.
+ * An over-limit body is drained before responding (as body-parser does) so the client receives the 413.
+ */
+export function gzipBodyPassthrough({ limitBytes }: { limitBytes: number }) {
+  return (req: express.Request, _res: express.Response, next: express.NextFunction) => {
+    if (req.get(HTTP.HEADERS.CONTENT_ENCODING)?.toLowerCase() !== 'gzip' || !req.is('text/csv')) {
+      next();
+      return;
+    }
+
+    const chunks: Buffer[] = [];
+    let receivedBytes = 0;
+
+    req.on('data', (chunk: Buffer) => {
+      receivedBytes += chunk.length;
+      if (receivedBytes <= limitBytes) {
+        chunks.push(chunk);
+      }
+    });
+    req.on('error', next);
+    req.on('end', () => {
+      if (receivedBytes > limitBytes) {
+        next(createBodyError(413, 'entity.too.large', 'request entity too large'));
+        return;
+      }
+      const body = Buffer.concat(chunks);
+      // gzip magic bytes - reject anything else here rather than forwarding garbage to Salesforce
+      if (body[0] !== 0x1f || body[1] !== 0x8b) {
+        next(createBodyError(400, 'encoding.invalid', 'request body is not valid gzip'));
+        return;
+      }
+      req.body = body;
+      next();
+    });
+  };
+}

apps/api/src/main.ts

@@ -64,6 +64,7 @@ import {
   setCacheControlForApiRoutes,
   setPermissionPolicy,
 } from './app/routes/route.middleware';
+import { gzipBodyPassthrough } from './app/utils/gzip-body-passthrough.middleware';
 import { healthCheck, uncaughtErrorHandler } from './app/utils/response.handlers';
 import { buildAppCspDirectives, buildCspDirectives, buildHstsConfig } from './app/utils/security-headers';
 import { handleWorkerExit, primaryClusterInitSideEffects, shutdownPrimaryGracefully } from './app/utils/server-process.utils';
@@ -319,8 +320,11 @@ if (ENV.NODE_ENV === 'production' && !ENV.CI && cluster.isPrimary) {
   // handles the report payload instead of the 20mb global JSON parser, and responds 204.
   app.use('/api/csp-report', cspReportRoutes);
 
-  // inflate: false - no client sends compressed request bodies; rejecting Content-Encoding (415) keeps
-  // untrusted input away from zlib/brotli decoders (e.g. CVE-2026-111497)
+  // Gzipped bulk batches are read as raw bytes and forwarded to Salesforce, which decodes them
+  app.post('/api/bulk/:jobId', gzipBodyPassthrough({ limitBytes: 30 * 1024 * 1024 }));
+
+  // inflate: false - rejecting Content-Encoding (415) keeps untrusted input away from zlib/brotli decoders
+  // (e.g. CVE-2026-111497); the one compressed body we accept is passed through above, never decoded
   app.use(raw({ limit: '30mb', type: ['text/csv'], inflate: false }));
   app.use(raw({ limit: '30mb', type: ['application/zip'], inflate: false }));
   app.use(json({ limit: '20mb', type: ['json'], inflate: false }));

apps/jetstream-e2e/src/tests/api/bulk.api.spec.ts

@@ -1,5 +1,6 @@
 import { CreateJobRequest } from '@jetstream/api-types';
 import { BulkJob, BulkJobBatchInfo, BulkJobWithBatches } from '@jetstream/types';
+import { gzipSync } from 'node:zlib';
 import { expect, test } from '../../fixtures/fixtures';
 
 test.describe.configure({ mode: 'parallel' });
@@ -54,6 +55,43 @@ test.describe('API - Bulk', () => {
     expect(getJobResponse2.state).toEqual('Closed');
   });
 
+  test('Bulk Job - Add gzipped Batch', async ({ apiRequestUtils }) => {
+    const createJobResponse = await apiRequestUtils.makeRequest<BulkJob>('POST', `/api/bulk`, {
+      externalId: null,
+      serialMode: false,
+      sObject: 'LEAD',
+      type: 'INSERT',
+    } as CreateJobRequest);
+
+    const timestamp = new Date().getTime();
+    // The job uses Salesforce's default comma delimiter
+    const csv = [
+      `LastName,FirstName,Company,Email,Status`,
+      `Gzip-${timestamp},Kathy,TNR Corp. ${timestamp},ksynder@tnr.${timestamp}.net,Working - Contacted`,
+    ].join('\n');
+
+    const addBatchToJobResponse = await apiRequestUtils.makeRequest<BulkJobBatchInfo>(
+      'POST',
+      `/api/bulk/${createJobResponse.id}?${new URLSearchParams({ closeJob: 'true' })}`,
+      gzipSync(csv),
+      { 'Content-Type': 'text/csv; charset=UTF-8', 'Content-Encoding': 'gzip' },
+    );
+
+    expect(addBatchToJobResponse.id).toBeTruthy();
+
+    // Salesforce decodes the batch itself - a successful insert proves the bytes arrived intact
+    await expect
+      .poll(
+        async () => {
+          const { batches } = await apiRequestUtils.makeRequest<BulkJobWithBatches>('GET', `/api/bulk/${createJobResponse.id}`);
+          const [{ state, stateMessage, numberRecordsProcessed, numberRecordsFailed }] = batches;
+          return { state, stateMessage, numberRecordsProcessed, numberRecordsFailed };
+        },
+        { timeout: 90_000 },
+      )
+      .toMatchObject({ state: 'Completed', numberRecordsProcessed: 1, numberRecordsFailed: 0 });
+  });
+
   test('Bulk Job - Errors', async ({ apiRequestUtils }) => {
     const createJobResponse = await apiRequestUtils.makeRequestRaw('POST', `/api/bulk`, {
       externalId: null,

libs/salesforce-api/src/lib/api-bulk.ts

@@ -68,13 +68,19 @@ export class ApiBulk extends SalesforceApi {
     jobId: string,
     closeJob = false,
     contentType: typeof HTTP.CONTENT_TYPE.CSV | typeof HTTP.CONTENT_TYPE.ZIP_CSV = HTTP.CONTENT_TYPE.CSV,
+    /** Set when `body` is already gzip-compressed - Salesforce decodes it */
+    contentEncoding?: 'gzip',
   ) {
     const result = await this.apiRequest<{ batchInfo: BulkJobBatchInfoUntyped }>({
       sessionInfo: this.sessionInfo,
       url: this.getBulkApiUrl(`/job/${jobId}/batch`),
       method: 'POST',
       body,
-      headers: { [HTTP.HEADERS.CONTENT_TYPE]: contentType, Accept: HTTP.CONTENT_TYPE.XML },
+      headers: {
+        [HTTP.HEADERS.CONTENT_TYPE]: contentType,
+        Accept: HTTP.CONTENT_TYPE.XML,
+        ...(contentEncoding && { [HTTP.HEADERS.CONTENT_ENCODING]: contentEncoding }),
+      },
       outputType: 'xml',
       rawBody: true,
     }).then(({ batchInfo }) => bulkApiEnsureTyped(batchInfo));

libs/shared/constants/src/lib/shared-constants.ts

@@ -102,6 +102,7 @@ export const HTTP = {
     X_CLIENT_RUNTIME: 'X-Client-Runtime',
     X_CLIENT_INSTALL: 'X-Client-Install',
     CONTENT_TYPE: 'Content-Type',
+    CONTENT_ENCODING: 'Content-Encoding',
     X_MOCK_KEY: 'X-MOCK-KEY',
     X_FORWARDED_FOR: 'X-FORWARDED-FOR',
     CONTENT_DISPOSITION: 'Content-Disposition',

libs/shared/data/src/lib/client-data.ts

@@ -23,6 +23,7 @@ import type {
 import { logger } from '@jetstream/shared/client-logger';
 import { HTTP, MIME_TYPES } from '@jetstream/shared/constants';
 import {
+  gzipEncodeText,
   hasIncompleteSubqueries,
   isIncompleteSubqueryResult,
   isSubqueryResult,
@@ -99,7 +100,7 @@ import { parseISO } from 'date-fns/parseISO';
 import isFunction from 'lodash/isFunction';
 import isNil from 'lodash/isNil';
 import orderBy from 'lodash/orderBy';
-import { handleExternalRequest, handleRequest, transformListMetadataResponse } from './client-data-data-helper';
+import { AxiosAdapterConfig, handleExternalRequest, handleRequest, transformListMetadataResponse } from './client-data-data-helper';
 import { getStandardValueSetTypes } from './standardValueSet';
 
 export interface PaginationCursorParams {
@@ -1327,19 +1328,42 @@ export async function bulkApiAbortJob(org: SalesforceOrgUi, jobId: string): Prom
   return handleRequest({ method: 'DELETE', url: `/api/bulk/${jobId}/abort` }, { org }).then(unwrapResponseIgnoreCache);
 }
 
+const BULK_BATCH_GZIP_MIN_LENGTH = 64 * 1024;
+
+/**
+ * Gzip large batches when they travel over the network to the Jetstream server, which forwards the compressed
+ * bytes to Salesforce untouched. Desktop, extension and canvas set a custom adapter and call Salesforce
+ * in-process, so their batches are left as plain text.
+ */
+async function getBulkBatchRequestBody(csv: string): Promise<{ data: string | Uint8Array; contentEncoding?: 'gzip' }> {
+  if (AxiosAdapterConfig.adapter || csv.length < BULK_BATCH_GZIP_MIN_LENGTH || typeof CompressionStream === 'undefined') {
+    return { data: csv };
+  }
+  try {
+    return { data: await gzipEncodeText(csv), contentEncoding: 'gzip' };
+  } catch (ex) {
+    logger.warn('[BULK] Failed to gzip batch, sending uncompressed', ex);
+    return { data: csv };
+  }
+}
+
 export async function bulkApiAddBatchToJob(
   org: SalesforceOrgUi,
   jobId: string,
   csv: string,
   closeJob?: boolean,
 ): Promise<BulkJobBatchInfo> {
+  const { data, contentEncoding } = await getBulkBatchRequestBody(csv);
   return handleRequest<BulkJobBatchInfo>(
     {
       method: 'POST',
       url: `/api/bulk/${jobId}`,
-      data: csv,
+      data,
       params: { closeJob },
-      headers: { [HTTP.HEADERS.CONTENT_TYPE]: HTTP.CONTENT_TYPE.CSV },
+      headers: {
+        [HTTP.HEADERS.CONTENT_TYPE]: HTTP.CONTENT_TYPE.CSV,
+        ...(contentEncoding && { [HTTP.HEADERS.CONTENT_ENCODING]: contentEncoding }),
+      },
     },
     { org },
   )

libs/shared/utils/src/lib/compression.ts

@@ -2,8 +2,11 @@ const TEXT_ENCODER = new TextEncoder();
 const TEXT_DECODER = new TextDecoder();
 
 export async function gzipEncode(value: unknown): Promise<Uint8Array<ArrayBuffer>> {
-  const json = JSON.stringify(value);
-  const stream = new Blob([TEXT_ENCODER.encode(json) as BlobPart]).stream().pipeThrough(new CompressionStream('gzip'));
+  return gzipEncodeText(JSON.stringify(value));
+}
+
+export async function gzipEncodeText(value: string): Promise<Uint8Array<ArrayBuffer>> {
+  const stream = new Blob([TEXT_ENCODER.encode(value) as BlobPart]).stream().pipeThrough(new CompressionStream('gzip'));
   const buffer = await new Response(stream).arrayBuffer();
   return new Uint8Array(buffer);
 }