feat: add JSON body mode for engine API endpoints (#294)

* feat: restore JSON body config mode without breaking CLI paths

Reintroduce application/json config bodies for the engine API while keeping
legacy header and NDJSON flows working. Fix OpenAPIHono and the OpenAPI-based
CLI generation so typed header schemas, streaming endpoints, and connector
loading all continue to work with the new spec.

Made-with: Cursor
Committed-By-Agent: cursor

* refactor: centralize mode-specific pipeline access

Move JSON-vs-header requiredness into shared helpers so handlers stop
merging two conditional code paths into nullable locals.

Made-with: Cursor
Committed-By-Agent: cursor

* fix: tighten JSON content-type validation

Keep JSON validation strict for JSON routes while avoiding false positives for NDJSON and JSON-header alternatives.

Made-with: Cursor
Committed-By-Agent: cursor

* fix: align isJsonBody with middleware content-type logic

Export isApplicationJsonContentType from hono-zod-openapi and reuse it
in app.ts so handler and middleware agree on what counts as JSON body
mode. Fixes mixed-case and json-seq false positives/negatives.

Made-with: Cursor
Committed-By-Agent: cursor

Author: tonyxiao

apps/engine/src/__generated__/openapi.d.ts

@@ -867,6 +867,230 @@ describe('POST /source_discover', () => {
   })
 })
 
+// ---------------------------------------------------------------------------
+// JSON body mode (Content-Type: application/json)
+// ---------------------------------------------------------------------------
+
+const syncParamsObj = JSON.parse(syncParams)
+
+describe('JSON body mode', () => {
+  it('POST /pipeline_check accepts pipeline in JSON body', async () => {
+    const app = await createApp(resolver)
+    const res = await app.request('/pipeline_check', {
+      method: 'POST',
+      headers: { 'Content-Type': 'application/json' },
+      body: JSON.stringify({ pipeline: syncParamsObj }),
+    })
+    expect(res.status).toBe(200)
+    expect(res.headers.get('Content-Type')).toBe('application/x-ndjson')
+    const events = await readNdjson<Record<string, unknown>>(res)
+    const statuses = events.filter((e) => e.type === 'connection_status')
+    expect(statuses).toHaveLength(2)
+  })
+
+  it('POST /pipeline_setup accepts pipeline in JSON body', async () => {
+    const app = await createApp(resolver)
+    const res = await app.request('/pipeline_setup', {
+      method: 'POST',
+      headers: { 'Content-Type': 'application/json' },
+      body: JSON.stringify({ pipeline: syncParamsObj }),
+    })
+    expect(res.status).toBe(200)
+    expect(res.headers.get('Content-Type')).toBe('application/x-ndjson')
+  })
+
+  it('POST /pipeline_teardown accepts pipeline in JSON body', async () => {
+    const app = await createApp(resolver)
+    const res = await app.request('/pipeline_teardown', {
+      method: 'POST',
+      headers: { 'Content-Type': 'application/json' },
+      body: JSON.stringify({ pipeline: syncParamsObj }),
+    })
+    expect(res.status).toBe(200)
+    expect(res.headers.get('Content-Type')).toBe('application/x-ndjson')
+  })
+
+  it('POST /source_discover accepts source in JSON body', async () => {
+    const app = await createApp(resolver)
+    const res = await app.request('/source_discover', {
+      method: 'POST',
+      headers: { 'Content-Type': 'application/json' },
+      body: JSON.stringify({
+        source: { type: 'test', test: { streams: { customers: {} } } },
+      }),
+    })
+    expect(res.status).toBe(200)
+    const events = await readNdjson<Record<string, unknown>>(res)
+    expect(events.some((e) => e.type === 'catalog')).toBe(true)
+  })
+
+  it('POST /pipeline_read accepts pipeline + state + body array in JSON body', async () => {
+    const app = await createApp(resolver)
+    const res = await app.request('/pipeline_read', {
+      method: 'POST',
+      headers: { 'Content-Type': 'application/json' },
+      body: JSON.stringify({
+        pipeline: syncParamsObj,
+        body: [
+          {
+            type: 'record',
+            record: {
+              stream: 'customers',
+              data: { id: 'cus_1', name: 'Alice' },
+              emitted_at: new Date().toISOString(),
+            },
+          },
+          {
+            type: 'source_state',
+            source_state: { stream: 'customers', data: { status: 'complete' } },
+          },
+        ],
+      }),
+    })
+    expect(res.status).toBe(200)
+    const events = await readNdjson<Message>(res)
+    expect(events).toHaveLength(3)
+    expect(events[0]!.type).toBe('record')
+    expect(events[1]!.type).toBe('source_state')
+    expect(events[2]).toMatchObject({ type: 'eof', eof: { reason: 'complete' } })
+  })
+
+  it('POST /pipeline_read accepts pipeline in JSON body without input', async () => {
+    const app = await createApp(resolver)
+    const res = await app.request('/pipeline_read', {
+      method: 'POST',
+      headers: { 'Content-Type': 'application/json' },
+      body: JSON.stringify({ pipeline: syncParamsObj }),
+    })
+    expect(res.status).toBe(200)
+    const events = await readNdjson<Message>(res)
+    expect(events.some((e) => e.type === 'eof')).toBe(true)
+  })
+
+  it('POST /pipeline_write accepts pipeline + body array in JSON body', async () => {
+    const app = await createApp(resolver)
+    const res = await app.request('/pipeline_write', {
+      method: 'POST',
+      headers: { 'Content-Type': 'application/json' },
+      body: JSON.stringify({
+        pipeline: syncParamsObj,
+        body: [
+          {
+            type: 'record',
+            record: {
+              stream: 'customers',
+              data: { id: 'cus_1' },
+              emitted_at: '2024-01-01T00:00:00.000Z',
+            },
+          },
+          {
+            type: 'source_state',
+            source_state: { stream: 'customers', data: { cursor: 'cus_1' } },
+          },
+        ],
+      }),
+    })
+    expect(res.status).toBe(200)
+    const events = await readNdjson<Record<string, unknown>>(res)
+    expect(events.some((e) => e.type === 'source_state')).toBe(true)
+  })
+
+  it('POST /pipeline_sync accepts pipeline + state + body array in JSON body', async () => {
+    const app = await createApp(resolver)
+    const res = await app.request('/pipeline_sync', {
+      method: 'POST',
+      headers: { 'Content-Type': 'application/json' },
+      body: JSON.stringify({
+        pipeline: syncParamsObj,
+        body: [
+          {
+            type: 'record',
+            record: {
+              stream: 'customers',
+              data: { id: 'cus_1', name: 'Alice' },
+              emitted_at: new Date().toISOString(),
+            },
+          },
+          {
+            type: 'source_state',
+            source_state: { stream: 'customers', data: { status: 'complete' } },
+          },
+        ],
+      }),
+    })
+    expect(res.status).toBe(200)
+    const events = await readNdjson<Record<string, unknown>>(res)
+    expect(events.some((e) => e.type === 'source_state')).toBe(true)
+    expect(events.some((e) => e.type === 'eof')).toBe(true)
+  })
+
+  it('POST /pipeline_sync without body array runs backfill mode', async () => {
+    const app = await createApp(resolver)
+    const res = await app.request('/pipeline_sync', {
+      method: 'POST',
+      headers: { 'Content-Type': 'application/json' },
+      body: JSON.stringify({ pipeline: syncParamsObj }),
+    })
+    expect(res.status).toBe(200)
+    const events = await readNdjson<Record<string, unknown>>(res)
+    expect(events.some((e) => e.type === 'eof')).toBe(true)
+  })
+
+  it('returns 400 when JSON body is missing pipeline', async () => {
+    const app = await createApp(resolver)
+    const res = await app.request('/pipeline_check', {
+      method: 'POST',
+      headers: { 'Content-Type': 'application/json' },
+      body: JSON.stringify({}),
+    })
+    expect(res.status).toBe(400)
+  })
+
+  it('NDJSON content-type uses header mode even with JSON-like body', async () => {
+    const app = await createApp(resolver)
+    const res = await app.request('/pipeline_check', {
+      method: 'POST',
+      headers: {
+        'Content-Type': 'application/x-ndjson',
+        'X-Pipeline': syncParams,
+      },
+    })
+    expect(res.status).toBe(200)
+  })
+
+  it('no content-type defaults to header mode', async () => {
+    const app = await createApp(resolver)
+    const res = await app.request('/pipeline_check', {
+      method: 'POST',
+      headers: { 'X-Pipeline': syncParams },
+    })
+    expect(res.status).toBe(200)
+  })
+
+  it('mixed-case Content-Type is accepted as JSON body mode', async () => {
+    const app = await createApp(resolver)
+    const res = await app.request('/pipeline_check', {
+      method: 'POST',
+      headers: { 'Content-Type': 'Application/JSON; charset=utf-8' },
+      body: JSON.stringify({ pipeline: syncParamsObj }),
+    })
+    expect(res.status).toBe(200)
+    expect(res.headers.get('Content-Type')).toBe('application/x-ndjson')
+  })
+
+  it('application/json-seq falls back to header mode, not JSON body', async () => {
+    const app = await createApp(resolver)
+    const res = await app.request('/pipeline_check', {
+      method: 'POST',
+      headers: {
+        'Content-Type': 'application/json-seq',
+        'X-Pipeline': syncParams,
+      },
+    })
+    expect(res.status).toBe(200)
+  })
+})
+
 // ---------------------------------------------------------------------------
 // POST /internal/query
 // ---------------------------------------------------------------------------