fix(api): forward parsed ClickHouse proxy bodies (#3383)

## Summary

Fix request bodies being dropped by the ClickHouse proxy after Express consumes them, including charset-suffixed JSON and URL-encoded forms. Capture the parsers' inflated bytes before decoding or parsing, then forward those bytes with matching framing. This preserves large integers, duplicate JSON keys, empty bodies, form escapes and text encodings. Unparsed multipart streams still pass through untouched.

### How to reproduce

**Starting point:** the API integration stack from CONTRIBUTING.md. Install with `yarn install --immutable`, then run `make dev-int-build`. The frontend demo and Vercel local mode have no API and cannot reproduce this path.

**On `main`:** see #2942. The parser consumes charset-suffixed JSON and URL-encoded forms, then the proxy tries to write the parsed object and never forwards its bytes. The charset-JSON regression case times out.

**On the earlier PR commit `9a88bbf0`:** reconstruction rounds integers such as `9007199254740993`, collapses duplicate JSON keys and changes an empty body into `{}`. The new fidelity regression cases reproduce those changes.

**On this branch:** run `JEST_RETRY_TIMES=0 make dev-int FILE=clickhouseProxy`. The suite uses the authenticated API and a real upstream echo server. It checks exact JSON and form bytes, UTF-16 charset preservation, gzip inflation, chunked framing, empty bodies, duplicate charset parameters, invalid JSON rejection and binary multipart passthrough.

The focused unit suite is `JEST_RETRY_TIMES=0 yarn workspace @hyperdx/api exec jest --runInBand --ci --coverage=false src/utils/__tests__/proxyBody.test.ts`.

Validation: 24 unit tests and 16 real-proxy integration tests passed with retries disabled. The four new fidelity regression cases fail against the earlier PR commit `9a88bbf0`. `yarn lint:fix`, API TypeScript checks and focused ESLint passed. This run used isolated ports and the published ClickStack collector image with the repository's MongoDB/ClickHouse services. Upstream code workflows still need repository approval, and Vercel deployment needs HyperDX team authorization.

### How to test on Vercel preview

N/A, API-only change.

### References

Closes #2942. This follows the deferred work from #2932 and the closed, unmerged attempt in #2978.

Author: adityareddy-dev

.changeset/proxy-parsed-bodies.md

@@ -0,0 +1,7 @@
+---
+'@hyperdx/api': patch
+---
+
+Forward the bytes consumed by the ClickHouse proxy's body parsers without
+changing JSON values, form fields or text encodings. Keep request framing in
+sync with inflated bytes and preserve unparsed multipart streams.

packages/api/src/api-app.ts

@@ -22,6 +22,7 @@ import usageStats from './tasks/usageStats';
 import { registerAITelemetry } from './utils/aiTelemetry';
 import logger, { expressLogger } from './utils/logger';
 import passport from './utils/passport';
+import { captureProxyBody } from './utils/proxyBody';
 
 registerAITelemetry();
 
@@ -56,9 +57,15 @@ if (!config.IS_CI && config.FRONTEND_URL) {
 
 app.disable('x-powered-by');
 app.use(compression());
-app.use(express.json({ limit: '32mb' }));
-app.use(express.text({ limit: '32mb' }));
-app.use(express.urlencoded({ extended: false, limit: '32mb' }));
+app.use(express.json({ limit: '32mb', verify: captureProxyBody }));
+app.use(express.text({ limit: '32mb', verify: captureProxyBody }));
+app.use(
+  express.urlencoded({
+    extended: false,
+    limit: '32mb',
+    verify: captureProxyBody,
+  }),
+);
 app.use(session(sess));
 
 if (!config.IS_LOCAL_APP_MODE) {

packages/api/src/routers/api/__tests__/clickhouseProxy.int.test.ts

@@ -1,8 +1,10 @@
 import { buildLogComment } from '@hyperdx/common-utils/dist/clickhouse';
 import http from 'http';
+import { gzipSync } from 'zlib';
 
 import { getLoggedInAgent, getServer } from '@/fixtures';
 import Connection from '@/models/connection';
+import logger from '@/utils/logger';
 
 // Use the real proxy instead of the blanket mock in jest.setup.ts; the int
 // jest config transpiles it (see jest.int.config.js).
@@ -16,14 +18,11 @@ type CapturedRequest = {
 };
 
 /**
- * Pins what /clickhouse-proxy forwards upstream: text/plain re-injection, the
- * raw-stream passthrough of multipart/form-data bodies, which
+ * Pins what /clickhouse-proxy forwards upstream: captured parser bytes with
+ * valid framing, the raw-stream passthrough of multipart/form-data bodies, which
  * `@clickhouse/client-web` emits when query params exceed its URL budget
  * (regression context: ClickHouse support-escalation #8482), and the
  * decode/re-encode the query string goes through in `pathRewrite`.
- *
- * Known-latent cases (charset-suffixed JSON, urlencoded) are not covered;
- * tracked in https://github.com/hyperdxio/hyperdx/issues/2942.
  */
 describe('clickhouse-proxy forwarding', () => {
   const server = getServer();
@@ -66,6 +65,7 @@ describe('clickhouse-proxy forwarding', () => {
   });
 
   afterAll(async () => {
+    upstream.closeAllConnections();
     await new Promise<void>((resolve, reject) =>
       upstream.close(err => (err ? reject(err) : resolve())),
     );
@@ -101,6 +101,217 @@ describe('clickhouse-proxy forwarding', () => {
     expect(captured[0].headers['x-clickhouse-user']).toBe('default');
   });
 
+  it.each(['application/json', 'application/json; charset=utf-8'])(
+    'preserves %s bytes and Content-Length',
+    async contentType => {
+      const { agent, team } = await getLoggedInAgent(server);
+      const connectionId = await createConnection(team._id.toString());
+      const body = {
+        query: "SELECT '東京 🍰' FORMAT JSON",
+        param_name: 'café',
+      };
+      const original = JSON.stringify(body, null, 2);
+      expect(Buffer.byteLength(original)).toBeGreaterThan(original.length);
+
+      await agent
+        .post('/clickhouse-proxy/?query_id=test-json')
+        .set('x-hyperdx-connection-id', connectionId)
+        .set('Content-Type', contentType)
+        .send(original)
+        .timeout({ response: 5000, deadline: 10000 })
+        .expect(200);
+
+      expect(captured).toHaveLength(1);
+      expect(captured[0].body).toEqual(Buffer.from(original));
+      expect(captured[0].headers['content-type']).toBe(contentType);
+      expect(captured[0].headers['content-length']).toBe(
+        String(Buffer.byteLength(original)),
+      );
+      expect(captured[0].headers['transfer-encoding']).toBeUndefined();
+    },
+  );
+
+  it('preserves parsed UTF-16LE JSON bytes and charset', async () => {
+    const { agent, team } = await getLoggedInAgent(server);
+    const connectionId = await createConnection(team._id.toString());
+    const body = { query: "SELECT '東京 🍰' FORMAT JSON" };
+    const original = Buffer.from(JSON.stringify(body, null, 2), 'utf16le');
+
+    const proxyRequest = agent
+      .post('/clickhouse-proxy/?query_id=test-utf16-json')
+      .set('x-hyperdx-connection-id', connectionId)
+      .set('Content-Type', 'application/json; charset=utf-16le')
+      .set('Content-Length', String(original.length))
+      .timeout({ response: 5000, deadline: 10000 });
+    proxyRequest.write(original);
+    await proxyRequest.expect(200);
+
+    expect(captured).toHaveLength(1);
+    expect(captured[0].body).toEqual(original);
+    expect(captured[0].headers['content-type']).toBe(
+      'application/json; charset=utf-16le',
+    );
+    expect(captured[0].headers['content-length']).toBe(String(original.length));
+    expect(captured[0].headers['content-encoding']).toBeUndefined();
+  });
+
+  it('forwards parsed gzip JSON without the original content encoding', async () => {
+    const { agent, team } = await getLoggedInAgent(server);
+    const connectionId = await createConnection(team._id.toString());
+    const body = { query: "SELECT 'café' FORMAT JSON" };
+    const serialized = Buffer.from(JSON.stringify(body, null, 2));
+    const original = gzipSync(serialized);
+
+    const proxyRequest = agent
+      .post('/clickhouse-proxy/?query_id=test-gzip-json')
+      .set('x-hyperdx-connection-id', connectionId)
+      .set('Content-Type', 'application/json; charset=utf-8')
+      .set('Content-Encoding', 'gzip')
+      .set('Content-Length', String(original.length))
+      .timeout({ response: 5000, deadline: 10000 });
+    proxyRequest.write(original);
+    await proxyRequest.expect(200);
+
+    expect(captured).toHaveLength(1);
+    expect(captured[0].body).toEqual(serialized);
+    expect(captured[0].headers['content-type']).toBe(
+      'application/json; charset=utf-8',
+    );
+    expect(captured[0].headers['content-length']).toBe(
+      String(serialized.length),
+    );
+    expect(captured[0].headers['content-encoding']).toBeUndefined();
+  });
+
+  it.each([
+    'application/x-www-form-urlencoded',
+    'application/x-www-form-urlencoded; charset=utf-8',
+  ])('preserves %s bytes and repeated form keys', async contentType => {
+    const { agent, team } = await getLoggedInAgent(server);
+    const connectionId = await createConnection(team._id.toString());
+    const original =
+      'query=SELECT%20%27caf%C3%A9%27&param=first&param=second%20value&empty=&literal=%2B%26%3D';
+    const entries = [
+      ['query', "SELECT 'café'"],
+      ['param', 'first'],
+      ['param', 'second value'],
+      ['empty', ''],
+      ['literal', '+&='],
+    ];
+
+    await agent
+      .post('/clickhouse-proxy/?query_id=test-urlencoded')
+      .set('x-hyperdx-connection-id', connectionId)
+      .set('Content-Type', contentType)
+      .send(original)
+      .timeout({ response: 5000, deadline: 10000 })
+      .expect(200);
+
+    expect(captured).toHaveLength(1);
+    expect(captured[0].body.toString()).toBe(original);
+    expect([...new URLSearchParams(captured[0].body.toString())]).toEqual(
+      entries,
+    );
+    expect(captured[0].headers['content-type']).toBe(contentType);
+    expect(captured[0].headers['content-length']).toBe(
+      String(Buffer.byteLength(original)),
+    );
+  });
+
+  it('keeps chunked framing when forwarding captured JSON bytes', async () => {
+    const { agent, team } = await getLoggedInAgent(server);
+    const connectionId = await createConnection(team._id.toString());
+    const body = { query: "SELECT 'café' FORMAT JSON" };
+    const original = JSON.stringify(body, null, 2);
+    const proxyRequest = agent
+      .post('/clickhouse-proxy/?query_id=test-chunked-json')
+      .set('x-hyperdx-connection-id', connectionId)
+      .set('Content-Type', 'application/json; charset=utf-8')
+      .set('Transfer-Encoding', 'chunked')
+      .timeout({ response: 5000, deadline: 10000 });
+
+    proxyRequest.write(original.slice(0, 10));
+    proxyRequest.write(original.slice(10));
+    await proxyRequest.expect(200);
+
+    expect(captured).toHaveLength(1);
+    expect(captured[0].body.toString()).toBe(original);
+    expect(captured[0].headers['transfer-encoding']).toBe('chunked');
+    expect(captured[0].headers['content-length']).toBeUndefined();
+  });
+
+  it.each(['/clickhouse-proxy/', '/CLICKHOUSE-PROXY/'])(
+    'preserves JSON number literals and duplicate keys on %s',
+    async route => {
+      const { agent, team } = await getLoggedInAgent(server);
+      const connectionId = await createConnection(team._id.toString());
+      const original =
+        '{"id":9007199254740993,"value":1.0,"value":1e2,"query":"SELECT 1"}';
+
+      await agent
+        .post(`${route}?query_id=test-json-fidelity`)
+        .set('x-hyperdx-connection-id', connectionId)
+        .set('Content-Type', 'application/json; charset=utf-8')
+        .send(original)
+        .timeout({ response: 5000, deadline: 10000 })
+        .expect(200);
+
+      expect(captured).toHaveLength(1);
+      expect(captured[0].body).toEqual(Buffer.from(original));
+    },
+  );
+
+  it('preserves an empty JSON body for query-string SQL', async () => {
+    const { agent, team } = await getLoggedInAgent(server);
+    const connectionId = await createConnection(team._id.toString());
+
+    await agent
+      .post('/clickhouse-proxy/?query=SELECT+1')
+      .set('x-hyperdx-connection-id', connectionId)
+      .set('Content-Type', 'application/json')
+      .set('Content-Length', '0')
+      .timeout({ response: 5000, deadline: 10000 })
+      .expect(200);
+
+    expect(captured).toHaveLength(1);
+    expect(captured[0].body).toEqual(Buffer.alloc(0));
+    expect(captured[0].headers['content-length']).toBe('0');
+  });
+
+  it('preserves bytes under duplicate charset parameters', async () => {
+    const { agent, team } = await getLoggedInAgent(server);
+    const connectionId = await createConnection(team._id.toString());
+    const contentType = 'application/json; charset=utf-8; charset=utf-16le';
+    const original = Buffer.from('{"query":"SELECT 1"}', 'utf16le');
+    const proxyRequest = agent
+      .post('/clickhouse-proxy/?query_id=test-duplicate-charset')
+      .set('x-hyperdx-connection-id', connectionId)
+      .set('Content-Type', contentType)
+      .set('Content-Length', String(original.length))
+      .timeout({ response: 5000, deadline: 10000 });
+    proxyRequest.write(original);
+    await proxyRequest.expect(200);
+
+    expect(captured).toHaveLength(1);
+    expect(captured[0].body).toEqual(original);
+    expect(captured[0].headers['content-type']).toBe(contentType);
+    expect(captured[0].headers['content-length']).toBe(String(original.length));
+  });
+
+  it('still rejects invalid JSON before forwarding', async () => {
+    const { agent, team } = await getLoggedInAgent(server);
+    const connectionId = await createConnection(team._id.toString());
+
+    await agent
+      .post('/clickhouse-proxy/')
+      .set('x-hyperdx-connection-id', connectionId)
+      .set('Content-Type', 'application/json')
+      .send('{invalid')
+      .expect(400);
+
+    expect(captured).toHaveLength(0);
+  });
+
   it('forwards a large multipart/form-data body byte-for-byte (unparsed stream passthrough)', async () => {
     const { agent, team } = await getLoggedInAgent(server);
     const connectionId = await createConnection(team._id.toString());
@@ -120,25 +331,40 @@ describe('clickhouse-proxy forwarding', () => {
           `some.rather.long.map.attribute.key.number.${i}\r\n`,
       );
     }
-    parts.push(`--${boundary}--\r\n`);
-    const body = parts.join('');
+    const body = Buffer.concat([
+      Buffer.from(parts.join('')),
+      Buffer.from(
+        `--${boundary}\r\n` +
+          `Content-Disposition: form-data; name="file"; filename="bytes.bin"\r\n` +
+          `Content-Type: application/octet-stream\r\n\r\n`,
+      ),
+      Buffer.from([0x00, 0xff, 0x80, 0x0d, 0x0a]),
+      Buffer.from(`\r\n--${boundary}--\r\n`),
+    ]);
     expect(Buffer.byteLength(body)).toBeGreaterThan(4096);
 
-    await agent
-      .post('/clickhouse-proxy/?query_id=test-multipart')
-      .set('x-hyperdx-connection-id', connectionId)
-      .set('Content-Type', `multipart/form-data; boundary=${boundary}`)
-      .send(body)
-      .expect(200);
+    const loggerError = jest.spyOn(logger, 'error');
+    try {
+      await agent
+        .post('/clickhouse-proxy/?query_id=test-multipart')
+        .set('x-hyperdx-connection-id', connectionId)
+        .set('Content-Type', `multipart/form-data; boundary=${boundary}`)
+        .send(body)
+        .timeout({ response: 5000, deadline: 10000 })
+        .expect(200);
 
-    expect(captured).toHaveLength(1);
-    expect(captured[0].headers['content-type']).toBe(
-      `multipart/form-data; boundary=${boundary}`,
-    );
-    expect(captured[0].body.toString()).toBe(body);
-    expect(captured[0].headers['content-length']).toBe(
-      String(Buffer.byteLength(body)),
-    );
+      expect(captured).toHaveLength(1);
+      expect(captured[0].headers['content-type']).toBe(
+        `multipart/form-data; boundary=${boundary}`,
+      );
+      expect(captured[0].body).toEqual(body);
+      expect(captured[0].headers['content-length']).toBe(
+        String(Buffer.byteLength(body)),
+      );
+      expect(loggerError).not.toHaveBeenCalled();
+    } finally {
+      loggerError.mockRestore();
+    }
   });
 
   // The browser client ships `log_comment` as a URL query param, and

packages/api/src/routers/api/clickhouseProxy.ts

@@ -11,6 +11,7 @@ import { getNonNullUserWithTeam } from '@/middleware/auth';
 import { validateRequestHeaders } from '@/middleware/validation';
 import { recordOperationOutcome } from '@/utils/instrumentation';
 import logger from '@/utils/logger';
+import { writeProxyBody } from '@/utils/proxyBody';
 import { IPV6_BRACKET_RE, isPrivateIp } from '@/utils/validators';
 import { objectIdSchema } from '@/utils/zod';
 
@@ -261,22 +262,7 @@ const proxyMiddleware: RequestHandler =
           return res.sendStatus(405);
         }
 
-        let body = _req.body;
-        if (_req.headers['content-type'] === 'application/json') {
-          try {
-            body = JSON.stringify(body);
-          } catch (e) {
-            console.error(e);
-          }
-        }
-
-        try {
-          proxyReq.write(body);
-        } catch {
-          console.error(
-            `clickhouseProxy error writing body, body is type ${typeof body}`,
-          );
-        }
+        writeProxyBody(proxyReq, _req);
       },
       proxyRes: (proxyRes, _req, res) => {
         const startedAt = (res as Response).locals?.hdxProxyStartedAt;