chore(studio): reuse key-value field array for log drains (#44060)

## What kind of change does this PR introduce? Chore that resolves FE-2785. ## What is the current behavior? The Log Drains headers section still uses a bespoke add-header mini form and stores headers directly as a record in form state. That makes it inconsistent with the shared `KeyValueFieldArray` pattern already adopted elsewhere in Studio. ## What is the new behavior? - reuses the shared `KeyValueFieldArray` in the Log Drains destination sheet - keeps the external submit contract unchanged by converting between: - form-only `headerEntries: { key, value }[]` - submitted `headers: Record<string, string>` - preserves type-specific defaults: - webhook starts with `Content-Type: application/json` - OTLP starts with `Content-Type: application/x-protobuf` - moves header validation into standard form errors for: - max 20 headers - duplicate header names - partially filled rows, while still allowing fully empty draft rows - adds focused utility and sheet tests for the new adapter and validation behaviour | Before | After | | --- | --- | | <img width="1728" height="997" alt="Log Drains Settings Mallet Toolshed Supabase-027E4669-8B02-4C43-8771-794E13799FA3" src="https://github.com/user-attachments/assets/43ed6334-28ef-4d47-9747-dfb3221462ec" /> | <img width="1728" height="997" alt="Log Drains Settings Mallet Toolshed Supabase-68477EA1-6F56-4BE2-9355-C121896F11E4" src="https://github.com/user-attachments/assets/9391bccd-3a50-4468-91e7-05059d41543c" /> | ## Additional context This is PR 4 of the DEPR-394 field-array stack.  ## Summary by CodeRabbit ## Release Notes * **Bug Fixes** * Enhanced header validation with specific error messages for duplicate headers and missing values. * Improved form behavior for edge cases in header entry management. * **New Features** * Added destination-specific default headers for clearer initial configuration. * **Tests** * Added comprehensive test suite validating header management and form submission behavior.
2026-05-06 22:18:00 +08:00 · 2026-04-02 11:14:34 +11:00
parent 73692b0a4d
commit 16308ad286
4 changed files with 678 additions and 294 deletions
--- a/apps/studio/components/interfaces/LogDrains/LogDrainDestinationSheetForm.test.tsx
+++ b/apps/studio/components/interfaces/LogDrains/LogDrainDestinationSheetForm.test.tsx
@@ -0,0 +1,233 @@
+import { fireEvent, screen, waitFor } from '@testing-library/react'
+import userEvent from '@testing-library/user-event'
+import type { ComponentProps } from 'react'
+import { render } from 'tests/helpers'
+import { beforeEach, describe, expect, it, vi } from 'vitest'
+
+import { LogDrainDestinationSheetForm } from './LogDrainDestinationSheetForm'
+
+const { trackMock, useFlagMock, useLogDrainsQueryMock, useParamsMock, useTrackMock } = vi.hoisted(
+  () => ({
+    trackMock: vi.fn(),
+    useFlagMock: vi.fn(),
+    useLogDrainsQueryMock: vi.fn(),
+    useParamsMock: vi.fn(),
+    useTrackMock: vi.fn(),
+  })
+)
+
+vi.mock(import('common'), async (importOriginal) => {
+  const actual = await importOriginal()
+
+  return {
+    ...actual,
+    IS_PLATFORM: false,
+    useFlag: useFlagMock,
+    useParams: useParamsMock,
+  }
+})
+
+vi.mock(import('data/log-drains/log-drains-query'), async (importOriginal) => {
+  const actual = await importOriginal()
+
+  return {
+    ...actual,
+    useLogDrainsQuery: useLogDrainsQueryMock,
+  }
+})
+
+vi.mock('lib/telemetry/track', () => ({
+  useTrack: useTrackMock,
+}))
+
+vi.mock('sonner', () => ({
+  toast: {
+    error: vi.fn(),
+    success: vi.fn(),
+  },
+}))
+
+const renderForm = (props?: Partial<ComponentProps<typeof LogDrainDestinationSheetForm>>) => {
+  const onOpenChange = vi.fn()
+  const onSubmit = vi.fn()
+
+  render(
+    <LogDrainDestinationSheetForm
+      open
+      mode="create"
+      isLoading={false}
+      onOpenChange={onOpenChange}
+      onSubmit={onSubmit}
+      defaultValues={{ type: 'webhook' }}
+      {...props}
+    />
+  )
+
+  return { onOpenChange, onSubmit }
+}
+
+const submitForm = () =>
+  fireEvent.submit(document.getElementById('log-drain-destination-form') as HTMLFormElement)
+
+describe('LogDrainDestinationSheetForm', () => {
+  beforeEach(() => {
+    vi.clearAllMocks()
+
+    useFlagMock.mockReturnValue(true)
+    useParamsMock.mockReturnValue({ ref: 'project-ref' })
+    useLogDrainsQueryMock.mockReturnValue({ data: [] })
+    useTrackMock.mockReturnValue(trackMock)
+  })
+
+  it('shows the JSON content type header for webhook create mode', async () => {
+    renderForm()
+
+    await screen.findByRole('dialog')
+
+    expect(screen.getByDisplayValue('Content-Type')).toBeInTheDocument()
+    expect(screen.getByDisplayValue('application/json')).toBeInTheDocument()
+  })
+
+  it('shows the protobuf content type header for OTLP create mode', async () => {
+    renderForm({
+      defaultValues: { type: 'otlp' },
+    })
+
+    await screen.findByRole('dialog')
+
+    expect(screen.getByDisplayValue('Content-Type')).toBeInTheDocument()
+    expect(screen.getByDisplayValue('application/x-protobuf')).toBeInTheDocument()
+  })
+
+  it('submits headers as a record without leaking the internal headerEntries field', async () => {
+    const user = userEvent.setup()
+    const { onSubmit } = renderForm()
+
+    await screen.findByRole('dialog')
+
+    await user.type(screen.getByPlaceholderText('My Destination'), 'Webhook sink')
+    await user.type(
+      screen.getByPlaceholderText('https://example.com/log-drain'),
+      'https://logs.example.com/ingest'
+    )
+    await user.click(screen.getByRole('button', { name: 'Add a new header' }))
+
+    const headerNameInputs = screen.getAllByPlaceholderText('Header name')
+    const headerValueInputs = screen.getAllByPlaceholderText('Header value')
+
+    await user.type(headerNameInputs[1], 'X-API-Key')
+    await user.type(headerValueInputs[1], 'secret-key')
+
+    submitForm()
+
+    await waitFor(() => expect(onSubmit).toHaveBeenCalledTimes(1))
+    expect(onSubmit).toHaveBeenCalledWith(
+      expect.objectContaining({
+        type: 'webhook',
+        headers: {
+          'Content-Type': 'application/json',
+          'X-API-Key': 'secret-key',
+        },
+      })
+    )
+    expect(onSubmit.mock.calls[0][0]).not.toHaveProperty('headerEntries')
+  })
+
+  it('rejects duplicate header names with an inline field error', async () => {
+    const user = userEvent.setup()
+    const { onSubmit } = renderForm()
+
+    await screen.findByRole('dialog')
+
+    await user.type(screen.getByPlaceholderText('My Destination'), 'Webhook sink')
+    await user.type(
+      screen.getByPlaceholderText('https://example.com/log-drain'),
+      'https://logs.example.com/ingest'
+    )
+    await user.click(screen.getByRole('button', { name: 'Add a new header' }))
+
+    const headerNameInputs = screen.getAllByPlaceholderText('Header name')
+    const headerValueInputs = screen.getAllByPlaceholderText('Header value')
+
+    await user.type(headerNameInputs[1], 'Content-Type')
+    await user.type(headerValueInputs[1], 'application/custom')
+
+    submitForm()
+
+    expect((await screen.findAllByText('Header name already exists')).length).toBeGreaterThan(0)
+    expect(onSubmit).not.toHaveBeenCalled()
+  })
+
+  it('allows fully empty header rows without blocking submit', async () => {
+    const user = userEvent.setup()
+    const { onSubmit } = renderForm()
+
+    await screen.findByRole('dialog')
+
+    await user.type(screen.getByPlaceholderText('My Destination'), 'Webhook sink')
+    await user.type(
+      screen.getByPlaceholderText('https://example.com/log-drain'),
+      'https://logs.example.com/ingest'
+    )
+    await user.click(screen.getByRole('button', { name: 'Add a new header' }))
+
+    submitForm()
+
+    await waitFor(() => expect(onSubmit).toHaveBeenCalledTimes(1))
+    expect(onSubmit).toHaveBeenCalledWith(
+      expect.objectContaining({
+        type: 'webhook',
+        headers: {
+          'Content-Type': 'application/json',
+        },
+      })
+    )
+    expect(screen.queryByText('undefined')).not.toBeInTheDocument()
+  })
+
+  it('shows an inline value error when a header key is entered without a value', async () => {
+    const user = userEvent.setup()
+    const { onSubmit } = renderForm()
+
+    await screen.findByRole('dialog')
+
+    await user.type(screen.getByPlaceholderText('My Destination'), 'Webhook sink')
+    await user.type(
+      screen.getByPlaceholderText('https://example.com/log-drain'),
+      'https://logs.example.com/ingest'
+    )
+    await user.click(screen.getByRole('button', { name: 'Add a new header' }))
+
+    const headerNameInputs = screen.getAllByPlaceholderText('Header name')
+    await user.type(headerNameInputs[1], 'X-Draft-Only')
+
+    submitForm()
+
+    expect(await screen.findByText('Header value is required')).toBeInTheDocument()
+    expect(onSubmit).not.toHaveBeenCalled()
+    expect(screen.queryByText('undefined')).not.toBeInTheDocument()
+  })
+
+  it('shows an inline key error when a header value is entered without a key', async () => {
+    const user = userEvent.setup()
+    const { onSubmit } = renderForm()
+
+    await screen.findByRole('dialog')
+
+    await user.type(screen.getByPlaceholderText('My Destination'), 'Webhook sink')
+    await user.type(
+      screen.getByPlaceholderText('https://example.com/log-drain'),
+      'https://logs.example.com/ingest'
+    )
+    await user.click(screen.getByRole('button', { name: 'Add a new header' }))
+
+    const headerValueInputs = screen.getAllByPlaceholderText('Header value')
+    await user.type(headerValueInputs[1], 'draft-value')
+
+    submitForm()
+
+    expect(await screen.findByText('Header name is required')).toBeInTheDocument()
+    expect(onSubmit).not.toHaveBeenCalled()
+    expect(screen.queryByText('undefined')).not.toBeInTheDocument()
+  })
+})
--- a/apps/studio/components/interfaces/LogDrains/LogDrainDestinationSheetForm.tsx
+++ b/apps/studio/components/interfaces/LogDrains/LogDrainDestinationSheetForm.tsx
@@ -1,8 +1,7 @@
 import { zodResolver } from '@hookform/resolvers/zod'
 import { IS_PLATFORM, useFlag, useParams } from 'common'
-import { TrashIcon } from 'lucide-react'
 import Link from 'next/link'
-import { ReactNode, useEffect, useState } from 'react'
+import { ReactNode, useEffect } from 'react'
 import { useForm } from 'react-hook-form'
 import { toast } from 'sonner'
 import {
@@ -33,6 +32,7 @@ import {
  Switch,
 } from 'ui'
 import { FormItemLayout } from 'ui-patterns/form/FormItemLayout/FormItemLayout'
+import { KeyValueFieldArray } from 'ui-patterns/form/KeyValueFieldArray/KeyValueFieldArray'
 import { InfoTooltip } from 'ui-patterns/info-tooltip'
 import { z } from 'zod'

@@ -44,8 +44,12 @@ import {
  OTLP_PROTOCOLS,
 } from './LogDrains.constants'
 import {
+  getDefaultHeadersByType,
  getHeadersSectionDescription as getHeadersDescription,
-  validateNewHeader,
+  headerRecordToRows,
+  headerRowsToRecord,
+  logDrainHeaderEntriesSchema,
+  type LogDrainHeaderRow,
 } from './LogDrains.utils'
 import { LogDrainData, useLogDrainsQuery } from '@/data/log-drains/log-drains-query'
 import { DOCS_URL } from '@/lib/constants'
@@ -54,88 +58,161 @@ import { httpEndpointUrlSchema } from '@/lib/validation/http-url'

 const FORM_ID = 'log-drain-destination-form'

+const headerRecordSchema = z.record(z.string(), z.string())
+
+const webhookFields = {
+  type: z.literal('webhook'),
+  url: httpEndpointUrlSchema({
+    requiredMessage: 'Endpoint URL is required',
+    invalidMessage: 'Endpoint URL must be a valid URL',
+    prefixMessage: 'Endpoint URL must start with http:// or https://',
+  }),
+  http: z.enum(['http1', 'http2']),
+  gzip: z.boolean(),
+}
+
+const webhookFormSchema = z.object({
+  ...webhookFields,
+  headerEntries: logDrainHeaderEntriesSchema.optional(),
+})
+
+const webhookSubmitSchema = z.object({
+  ...webhookFields,
+  headers: headerRecordSchema.optional(),
+})
+
+const datadogSchema = z.object({
+  type: z.literal('datadog'),
+  api_key: z.string().min(1, { message: 'API key is required' }),
+  region: z.string().min(1, { message: 'Region is required' }),
+})
+
+const lokiFields = {
+  type: z.literal('loki'),
+  url: httpEndpointUrlSchema({
+    requiredMessage: 'Loki URL is required',
+    invalidMessage: 'Loki URL must be a valid URL',
+    prefixMessage: 'Loki URL must start with http:// or https://',
+  }),
+  username: z.string().optional(),
+  password: z.string().optional(),
+}
+
+const lokiFormSchema = z.object({
+  ...lokiFields,
+  headerEntries: logDrainHeaderEntriesSchema.optional(),
+})
+
+const lokiSubmitSchema = z.object({
+  ...lokiFields,
+  headers: headerRecordSchema,
+})
+
+const elasticSchema = z.object({
+  type: z.literal('elastic'),
+})
+
+const postgresSchema = z.object({
+  type: z.literal('postgres'),
+})
+
+const bigquerySchema = z.object({
+  type: z.literal('bigquery'),
+})
+
+const clickhouseSchema = z.object({
+  type: z.literal('clickhouse'),
+})
+
+const s3Schema = z.object({
+  type: z.literal('s3'),
+  s3_bucket: z.string().min(1, { message: 'Bucket name is required' }),
+  storage_region: z.string().min(1, { message: 'Region is required' }),
+  access_key_id: z.string().min(1, { message: 'Access Key ID is required' }),
+  secret_access_key: z.string().min(1, { message: 'Secret Access Key is required' }),
+  batch_timeout: z.coerce
+    .number()
+    .int({ message: 'Batch timeout must be an integer' })
+    .min(1, { message: 'Batch timeout must be a positive integer' }),
+})
+
+const sentrySchema = z.object({
+  type: z.literal('sentry'),
+  dsn: z
+    .string()
+    .min(1, { message: 'Sentry DSN is required' })
+    .refine((dsn) => dsn.startsWith('https://'), 'Sentry DSN must start with https://'),
+})
+
+const axiomSchema = z.object({
+  type: z.literal('axiom'),
+  api_token: z.string().min(1, { message: 'API token is required' }),
+  dataset_name: z.string().min(1, { message: 'Dataset name is required' }),
+})
+
+const last9Schema = z.object({
+  type: z.literal('last9'),
+  region: z.string().min(1, { message: 'Region is required' }),
+  username: z.string().min(1, { message: 'Username is required' }),
+  password: z.string().min(1, { message: 'Password is required' }),
+})
+
+const otlpFields = {
+  type: z.literal('otlp'),
+  endpoint: httpEndpointUrlSchema({
+    requiredMessage: 'OTLP endpoint is required',
+    invalidMessage: 'OTLP endpoint must be a valid URL',
+    prefixMessage: 'OTLP endpoint must start with http:// or https://',
+  }),
+  protocol: z.string().optional().default('http/protobuf'),
+  gzip: z.boolean().optional().default(true),
+}
+
+const otlpFormSchema = z.object({
+  ...otlpFields,
+  headerEntries: logDrainHeaderEntriesSchema.optional(),
+})
+
+const otlpSubmitSchema = z.object({
+  ...otlpFields,
+  headers: headerRecordSchema.optional(),
+})
+
+const syslogSchema = z.object({
+  type: z.literal('syslog'),
+})
+
 const formUnion = z.discriminatedUnion('type', [
-  z.object({
-    type: z.literal('webhook'),
-    url: httpEndpointUrlSchema({
-      requiredMessage: 'Endpoint URL is required',
-      invalidMessage: 'Endpoint URL must be a valid URL',
-      prefixMessage: 'Endpoint URL must start with http:// or https://',
-    }),
-    http: z.enum(['http1', 'http2']),
-    gzip: z.boolean(),
-    headers: z.record(z.string(), z.string()).optional(),
-  }),
-  z.object({
-    type: z.literal('datadog'),
-    api_key: z.string().min(1, { message: 'API key is required' }),
-    region: z.string().min(1, { message: 'Region is required' }),
-  }),
-  z.object({
-    type: z.literal('loki'),
-    url: httpEndpointUrlSchema({
-      requiredMessage: 'Loki URL is required',
-      invalidMessage: 'Loki URL must be a valid URL',
-      prefixMessage: 'Loki URL must start with http:// or https://',
-    }),
-    headers: z.record(z.string(), z.string()),
-    username: z.string().optional(),
-    password: z.string().optional(),
-  }),
+  webhookFormSchema,
+  datadogSchema,
+  lokiFormSchema,
  // [Joshen] To fix API types, not supported in the UI
-  z.object({
-    type: z.literal('elastic'),
-  }),
-  z.object({
-    type: z.literal('postgres'),
-  }),
-  z.object({
-    type: z.literal('bigquery'),
-  }),
-  z.object({
-    type: z.literal('clickhouse'),
-  }),
-  z.object({
-    type: z.literal('s3'),
-    s3_bucket: z.string().min(1, { message: 'Bucket name is required' }),
-    storage_region: z.string().min(1, { message: 'Region is required' }),
-    access_key_id: z.string().min(1, { message: 'Access Key ID is required' }),
-    secret_access_key: z.string().min(1, { message: 'Secret Access Key is required' }),
-    batch_timeout: z.coerce
-      .number()
-      .int({ message: 'Batch timeout must be an integer' })
-      .min(1, { message: 'Batch timeout must be a positive integer' }),
-  }),
-  z.object({
-    type: z.literal('sentry'),
-    dsn: z
-      .string()
-      .min(1, { message: 'Sentry DSN is required' })
-      .refine((dsn) => dsn.startsWith('https://'), 'Sentry DSN must start with https://'),
-  }),
-  z.object({
-    type: z.literal('axiom'),
-    api_token: z.string().min(1, { message: 'API token is required' }),
-    dataset_name: z.string().min(1, { message: 'Dataset name is required' }),
-  }),
-  z.object({
-    type: z.literal('last9'),
-    region: z.string().min(1, { message: 'Region is required' }),
-    username: z.string().min(1, { message: 'Username is required' }),
-    password: z.string().min(1, { message: 'Password is required' }),
-  }),
-  z.object({
-    type: z.literal('otlp'),
-    endpoint: httpEndpointUrlSchema({
-      requiredMessage: 'OTLP endpoint is required',
-      invalidMessage: 'OTLP endpoint must be a valid URL',
-      prefixMessage: 'OTLP endpoint must start with http:// or https://',
-    }),
-    protocol: z.string().optional().default('http/protobuf'),
-    gzip: z.boolean().optional().default(true),
-    headers: z.record(z.string(), z.string()).optional(),
-  }),
-  z.object({ type: z.literal('syslog') }),
+  elasticSchema,
+  postgresSchema,
+  bigquerySchema,
+  clickhouseSchema,
+  s3Schema,
+  sentrySchema,
+  axiomSchema,
+  last9Schema,
+  otlpFormSchema,
+  syslogSchema,
+])
+
+const submitUnion = z.discriminatedUnion('type', [
+  webhookSubmitSchema,
+  datadogSchema,
+  lokiSubmitSchema,
+  elasticSchema,
+  postgresSchema,
+  bigquerySchema,
+  clickhouseSchema,
+  s3Schema,
+  sentrySchema,
+  axiomSchema,
+  last9Schema,
+  otlpSubmitSchema,
+  syslogSchema,
 ])

 const formSchema = z
@@ -147,6 +224,39 @@ const formSchema = z
  })
  .and(formUnion)

+const submitSchema = z
+  .object({
+    name: z.string().min(1, {
+      message: 'Destination name is required',
+    }),
+    description: z.string().optional(),
+  })
+  .and(submitUnion)
+
+type LogDrainDestinationFormValues = z.infer<typeof formSchema>
+type LogDrainDestinationSubmitValues = z.infer<typeof submitSchema>
+
+const HEADER_ENABLED_TYPES = ['webhook', 'loki', 'otlp'] as const
+
+function toSubmitValues(values: LogDrainDestinationFormValues): LogDrainDestinationSubmitValues {
+  if (!HEADER_ENABLED_TYPES.includes(values.type as (typeof HEADER_ENABLED_TYPES)[number])) {
+    return submitSchema.parse(values)
+  }
+
+  const { headerEntries = [], ...rest } = values as LogDrainDestinationFormValues & {
+    headerEntries?: LogDrainHeaderRow[]
+  }
+  const headers = headerRowsToRecord(headerEntries)
+  const transformedValues =
+    rest.type === 'loki'
+      ? { ...rest, headers }
+      : Object.keys(headers).length > 0
+        ? { ...rest, headers }
+        : rest
+
+  return submitSchema.parse(transformedValues)
+}
+
 function LogDrainFormItem({
  value,
  label,
@@ -191,7 +301,7 @@ export function LogDrainDestinationSheetForm({
  onOpenChange: (v: boolean) => void
  defaultValues?: DefaultValues
  isLoading?: boolean
-  onSubmit: (values: z.infer<typeof formSchema>) => void
+  onSubmit: (values: LogDrainDestinationSubmitValues) => void
  mode: 'create' | 'update'
 }) {
  // NOTE(kamil): This used to be `any` for a long long time, but after moving to Zod,
@@ -200,14 +310,9 @@ export function LogDrainDestinationSheetForm({
  // out of it, therefore for an ease of use now, we bail to `any` until the better time come.
  const defaultConfig = (defaultValues?.config || {}) as any
  const defaultType = defaultValues?.type || 'webhook'
-  const CREATE_DEFAULT_HEADERS_BY_TYPE: Partial<Record<LogDrainType, Record<string, string>>> = {
-    webhook: { 'Content-Type': 'application/json' },
-    otlp: { 'Content-Type': 'application/x-protobuf' },
-  }
-  const DEFAULT_HEADERS =
-    mode === 'create'
-      ? (CREATE_DEFAULT_HEADERS_BY_TYPE[defaultType] ?? {})
-      : defaultConfig?.headers || {}
+  const defaultHeaderEntries = headerRecordToRows(
+    mode === 'create' ? getDefaultHeadersByType(defaultType) : defaultConfig?.headers || {}
+  )

  const sentryEnabled = useFlag('SentryLogDrain')
  const s3Enabled = useFlag('S3logdrain')
@@ -220,10 +325,9 @@ export function LogDrainDestinationSheetForm({
    ref,
  })

-  const [newCustomHeader, setNewCustomHeader] = useState({ name: '', value: '' })
  const track = useTrack()

-  const form = useForm<z.infer<typeof formSchema>>({
+  const form = useForm<LogDrainDestinationFormValues>({
    resolver: zodResolver(formSchema),
    values: {
      name: defaultValues?.name || '',
@@ -231,7 +335,7 @@ export function LogDrainDestinationSheetForm({
      type: defaultType,
      http: defaultConfig?.http || 'http2',
      gzip: mode === 'create' ? true : defaultConfig?.gzip || false,
-      headers: DEFAULT_HEADERS,
+      headerEntries: defaultHeaderEntries,
      url: defaultConfig?.url || '',
      api_key: defaultConfig?.api_key || '',
      region: defaultConfig?.region || '',
@@ -250,47 +354,24 @@ export function LogDrainDestinationSheetForm({
    },
  })

-  const headers = form.watch('headers')
  const type = form.watch('type')

-  function removeHeader(key: string) {
-    const newHeaders = {
-      ...headers,
-    }
-    delete newHeaders[key]
-    form.setValue('headers', newHeaders)
-  }
-
-  function addHeader() {
-    const formHeaders = form.getValues('headers')
-    if (!formHeaders) return
-
-    const validation = validateNewHeader(formHeaders, newCustomHeader)
-    if (!validation.valid) {
-      toast.error(validation.error)
-      return
-    }
-
-    form.setValue('headers', { ...formHeaders, [newCustomHeader.name]: newCustomHeader.value })
-    setNewCustomHeader({ name: '', value: '' })
-  }
-
-  const hasHeaders = Object.keys(headers || {})?.length > 0
-
  useEffect(() => {
    if (mode === 'create' && !open) {
      form.reset()
    }
  }, [mode, open, form])

+  useEffect(() => {
+    if (!open || mode !== 'create') return
+
+    form.setValue('headerEntries', headerRecordToRows(getDefaultHeadersByType(type)), {
+      shouldValidate: true,
+    })
+  }, [form, mode, open, type])
+
  return (
-    <Sheet
-      open={open}
-      onOpenChange={(v) => {
-        setNewCustomHeader({ name: '', value: '' })
-        onOpenChange(v)
-      }}
-    >
+    <Sheet open={open} onOpenChange={onOpenChange}>
      <SheetContent
        tabIndex={undefined}
        showClose={false}
@@ -316,8 +397,10 @@ export function LogDrainDestinationSheetForm({
                  return
                }

-                form.handleSubmit(onSubmit)(e)
-                track('log_drain_save_button_clicked', { destination: form.getValues('type') })
+                form.handleSubmit((values) => onSubmit(toSubmitValues(values)))(e)
+                track('log_drain_save_button_clicked', {
+                  destination: form.getValues('type'),
+                })
              }}
            >
              <div className="space-y-8 px-content">
@@ -723,78 +806,37 @@ export function LogDrainDestinationSheetForm({
                    />
                  </div>
                )}
-                <FormMessage_Shadcn_ />
+                {HEADER_ENABLED_TYPES.includes(type as (typeof HEADER_ENABLED_TYPES)[number]) && (
+                  <div className="px-content">
+                    <FormField_Shadcn_
+                      control={form.control}
+                      name="headerEntries"
+                      render={({ fieldState }) => (
+                        <FormItemLayout
+                          layout="horizontal"
+                          label="Custom Headers"
+                          description={getHeadersDescription(type)}
+                          hideMessage={!fieldState.error?.message}
+                        >
+                          <KeyValueFieldArray
+                            control={form.control}
+                            name="headerEntries"
+                            keyFieldName="key"
+                            valueFieldName="value"
+                            createEmptyRow={() => ({ key: '', value: '' })}
+                            keyPlaceholder="Header name"
+                            valuePlaceholder="Header value"
+                            addLabel="Add a new header"
+                            removeLabel="Remove header"
+                          />
+                        </FormItemLayout>
+                      )}
+                    />
+                  </div>
+                )}
              </div>
            </form>
          </Form_Shadcn_>
-
-          {/* This form needs to be outside the <Form_Shadcn_> */}
-          {(type === 'webhook' || type === 'loki' || type === 'otlp') && (
-            <>
-              <div className="border-t mt-4">
-                <div className="px-content pt-2 pb-3 border-b bg-background-alternative-200">
-                  <h2 className="text-sm">Custom Headers</h2>
-                  <p className="text-xs text-foreground-lighter">{getHeadersDescription(type)}</p>
-                </div>
-                <div className="divide-y">
-                  {hasHeaders &&
-                    Object.keys(headers || {})?.map((headerKey) => (
-                      <div
-                        className="flex text-sm px-content text-foreground items-center font-mono py-1.5 group"
-                        key={headerKey}
-                      >
-                        <div className="w-full">{headerKey}</div>
-                        <div className="w-full truncate" title={headers?.[headerKey]}>
-                          {headers?.[headerKey]}
-                        </div>
-                        <Button
-                          className="justify-self-end opacity-0 group-hover:opacity-100 w-7"
-                          type="text"
-                          title="Remove"
-                          icon={<TrashIcon />}
-                          onClick={() => removeHeader(headerKey)}
-                        ></Button>
-                      </div>
-                    ))}
-                </div>
-              </div>
-              <form
-                onSubmit={(e) => {
-                  e.preventDefault()
-                  e.stopPropagation()
-                  addHeader()
-                }}
-                className="flex border-t py-4 gap-4 items-center px-content"
-              >
-                <label className="sr-only" htmlFor="header-name">
-                  Header name
-                </label>
-                <Input_Shadcn_
-                  id="header-name"
-                  type="text"
-                  placeholder="x-header-name"
-                  value={newCustomHeader.name}
-                  onChange={(e) => setNewCustomHeader({ ...newCustomHeader, name: e.target.value })}
-                />
-                <label className="sr-only" htmlFor="header-value">
-                  Header value
-                </label>
-                <Input_Shadcn_
-                  id="header-value"
-                  type="text"
-                  placeholder="Header value"
-                  value={newCustomHeader.value}
-                  onChange={(e) =>
-                    setNewCustomHeader({ ...newCustomHeader, value: e.target.value })
-                  }
-                />
-
-                <Button htmlType="submit" type="outline">
-                  Add
-                </Button>
-              </form>
-            </>
-          )}
        </SheetSection>

        <div className="mt-auto">
--- a/apps/studio/components/interfaces/LogDrains/LogDrains.utils.test.ts
+++ b/apps/studio/components/interfaces/LogDrains/LogDrains.utils.test.ts
@@ -1,10 +1,13 @@
 import { describe, expect, it } from 'vitest'

 import {
+  getDefaultHeadersByType,
  getHeadersSectionDescription,
  HEADER_VALIDATION_ERRORS,
+  headerRecordToRows,
+  headerRowsToRecord,
+  logDrainHeaderEntriesSchema,
  otlpConfigSchema,
-  validateNewHeader,
 } from './LogDrains.utils'

 describe('getHeadersSectionDescription', () => {
@@ -32,88 +35,133 @@ describe('getHeadersSectionDescription', () => {
  })
 })

-describe('validateNewHeader', () => {
-  describe('valid cases', () => {
-    it('accepts valid header with empty existing headers', () => {
-      const result = validateNewHeader({}, { name: 'Authorization', value: 'Bearer token' })
-      expect(result.valid).toBe(true)
-      expect(result.error).toBeUndefined()
-    })
-
-    it('accepts valid header with existing headers', () => {
-      const existingHeaders = {
-        'Content-Type': 'application/json',
-        'X-Custom': 'value',
-      }
-      const result = validateNewHeader(existingHeaders, {
-        name: 'Authorization',
-        value: 'Bearer token',
-      })
-      expect(result.valid).toBe(true)
-      expect(result.error).toBeUndefined()
+describe('getDefaultHeadersByType', () => {
+  it('returns the JSON content type header for webhook destinations', () => {
+    expect(getDefaultHeadersByType('webhook')).toEqual({
+      'Content-Type': 'application/json',
    })
  })

-  describe('validation errors', () => {
-    it('rejects when 20 headers already exist', () => {
-      const existingHeaders = Object.fromEntries(
-        Array.from({ length: 20 }, (_, i) => [`Header-${i}`, `value-${i}`])
-      )
-      const result = validateNewHeader(existingHeaders, { name: 'New-Header', value: 'value' })
-      expect(result.valid).toBe(false)
-      expect(result.error).toBe(HEADER_VALIDATION_ERRORS.MAX_LIMIT)
-    })
-
-    it('rejects duplicate header names', () => {
-      const existingHeaders = {
-        'Content-Type': 'application/json',
-        Authorization: 'Bearer old-token',
-      }
-      const result = validateNewHeader(existingHeaders, {
-        name: 'Authorization',
-        value: 'Bearer new-token',
-      })
-      expect(result.valid).toBe(false)
-      expect(result.error).toBe(HEADER_VALIDATION_ERRORS.DUPLICATE)
-    })
-
-    it('rejects header with empty name', () => {
-      const result = validateNewHeader({}, { name: '', value: 'some-value' })
-      expect(result.valid).toBe(false)
-      expect(result.error).toBe(HEADER_VALIDATION_ERRORS.REQUIRED)
-    })
-
-    it('rejects header with empty value', () => {
-      const result = validateNewHeader({}, { name: 'Some-Header', value: '' })
-      expect(result.valid).toBe(false)
-      expect(result.error).toBe(HEADER_VALIDATION_ERRORS.REQUIRED)
-    })
-
-    it('rejects header with both empty name and value', () => {
-      const result = validateNewHeader({}, { name: '', value: '' })
-      expect(result.valid).toBe(false)
-      expect(result.error).toBe(HEADER_VALIDATION_ERRORS.REQUIRED)
+  it('returns the protobuf content type header for OTLP destinations', () => {
+    expect(getDefaultHeadersByType('otlp')).toEqual({
+      'Content-Type': 'application/x-protobuf',
    })
  })

-  describe('edge cases', () => {
-    it('allows exactly 19 existing headers', () => {
-      const existingHeaders = Object.fromEntries(
-        Array.from({ length: 19 }, (_, i) => [`Header-${i}`, `value-${i}`])
-      )
-      const result = validateNewHeader(existingHeaders, { name: 'New-Header', value: 'value' })
-      expect(result.valid).toBe(true)
-    })
+  it('returns an empty object for destinations without default headers', () => {
+    expect(getDefaultHeadersByType('loki')).toEqual({})
+    expect(getDefaultHeadersByType('datadog')).toEqual({})
+  })
+})

-    it('is case-sensitive for duplicate checking', () => {
-      const existingHeaders = { authorization: 'bearer token' }
-      const result = validateNewHeader(existingHeaders, {
-        name: 'Authorization',
-        value: 'Bearer token',
+describe('headerRecordToRows', () => {
+  it('converts a header record into key/value rows', () => {
+    expect(
+      headerRecordToRows({
+        Authorization: 'Bearer token',
+        'Content-Type': 'application/json',
      })
-      expect(result.valid).toBe(true)
+    ).toEqual([
+      { key: 'Authorization', value: 'Bearer token' },
+      { key: 'Content-Type', value: 'application/json' },
+    ])
+  })
+
+  it('returns an empty array for empty header records', () => {
+    expect(headerRecordToRows({})).toEqual([])
+  })
+})
+
+describe('headerRowsToRecord', () => {
+  it('converts key/value rows back into a header record', () => {
+    expect(
+      headerRowsToRecord([
+        { key: 'Authorization', value: 'Bearer token' },
+        { key: 'Content-Type', value: 'application/json' },
+      ])
+    ).toEqual({
+      Authorization: 'Bearer token',
+      'Content-Type': 'application/json',
    })
  })
+
+  it('trims row values and skips incomplete rows', () => {
+    expect(
+      headerRowsToRecord([
+        { key: ' Authorization ', value: ' Bearer token ' },
+        { key: '', value: 'missing-key' },
+        { key: 'X-Skip', value: '' },
+      ])
+    ).toEqual({
+      Authorization: 'Bearer token',
+    })
+  })
+})
+
+describe('logDrainHeaderEntriesSchema', () => {
+  it('accepts fully empty draft rows', () => {
+    const result = logDrainHeaderEntriesSchema.safeParse([
+      { key: 'Content-Type', value: 'application/json' },
+      { key: '', value: '' },
+    ])
+
+    expect(result.success).toBe(true)
+  })
+
+  it('rejects rows with a key but no value', () => {
+    const result = logDrainHeaderEntriesSchema.safeParse([{ key: 'X-Draft-Only', value: '' }])
+
+    expect(result.success).toBe(false)
+    if (!result.success) {
+      expect(result.error.issues).toEqual(
+        expect.arrayContaining([
+          expect.objectContaining({
+            message: HEADER_VALIDATION_ERRORS.VALUE_REQUIRED,
+            path: [0, 'value'],
+          }),
+        ])
+      )
+    }
+  })
+
+  it('rejects rows with a value but no key', () => {
+    const result = logDrainHeaderEntriesSchema.safeParse([{ key: '', value: 'draft-value' }])
+
+    expect(result.success).toBe(false)
+    if (!result.success) {
+      expect(result.error.issues).toEqual(
+        expect.arrayContaining([
+          expect.objectContaining({
+            message: HEADER_VALIDATION_ERRORS.KEY_REQUIRED,
+            path: [0, 'key'],
+          }),
+        ])
+      )
+    }
+  })
+
+  it('still rejects duplicate completed header names', () => {
+    const result = logDrainHeaderEntriesSchema.safeParse([
+      { key: 'Content-Type', value: 'application/json' },
+      { key: 'Content-Type', value: 'application/custom' },
+    ])
+
+    expect(result.success).toBe(false)
+    if (!result.success) {
+      expect(result.error.issues).toEqual(
+        expect.arrayContaining([
+          expect.objectContaining({
+            message: 'Header name already exists',
+            path: [0, 'key'],
+          }),
+          expect.objectContaining({
+            message: 'Header name already exists',
+            path: [1, 'key'],
+          }),
+        ])
+      )
+    }
+  })
 })

 describe('otlpConfigSchema', () => {
--- a/apps/studio/components/interfaces/LogDrains/LogDrains.utils.ts
+++ b/apps/studio/components/interfaces/LogDrains/LogDrains.utils.ts
@@ -8,6 +8,11 @@ import { z } from 'zod'
 import { LogDrainType } from './LogDrains.constants'
 import { httpEndpointUrlSchema } from '@/lib/validation/http-url'

+export type LogDrainHeaderRow = {
+  key: string
+  value: string
+}
+
 /**
 * Get the description text for the custom headers section based on log drain type
 */
@@ -30,33 +35,89 @@ export function getHeadersSectionDescription(type: LogDrainType): string {
 export const HEADER_VALIDATION_ERRORS = {
  MAX_LIMIT: 'You can only have 20 custom headers',
  DUPLICATE: 'Header name already exists',
-  REQUIRED: 'Header name and value are required',
+  KEY_REQUIRED: 'Header name is required',
+  VALUE_REQUIRED: 'Header value is required',
 } as const

-/**
- * Validates if a new header can be added to the existing headers
- */
-export function validateNewHeader(
-  existingHeaders: Record<string, string>,
-  newHeader: { name: string; value: string }
-): { valid: boolean; error?: string } {
-  const headerKeys = Object.keys(existingHeaders)
-
-  if (headerKeys.length >= 20) {
-    return { valid: false, error: HEADER_VALIDATION_ERRORS.MAX_LIMIT }
-  }
-
-  if (headerKeys.includes(newHeader.name)) {
-    return { valid: false, error: HEADER_VALIDATION_ERRORS.DUPLICATE }
-  }
-
-  if (!newHeader.name || !newHeader.value) {
-    return { valid: false, error: HEADER_VALIDATION_ERRORS.REQUIRED }
-  }
-
-  return { valid: true }
+const DEFAULT_HEADERS_BY_TYPE: Partial<Record<LogDrainType, Record<string, string>>> = {
+  webhook: { 'Content-Type': 'application/json' },
+  otlp: { 'Content-Type': 'application/x-protobuf' },
 }

+export function getDefaultHeadersByType(type: LogDrainType): Record<string, string> {
+  return DEFAULT_HEADERS_BY_TYPE[type] ?? {}
+}
+
+export function headerRecordToRows(headers: Record<string, string> = {}): LogDrainHeaderRow[] {
+  return Object.entries(headers).map(([key, value]) => ({ key, value }))
+}
+
+export function headerRowsToRecord(rows: LogDrainHeaderRow[] = []): Record<string, string> {
+  return rows.reduce<Record<string, string>>((acc, row) => {
+    const key = row.key.trim()
+    const value = row.value.trim()
+
+    if (key && value) {
+      acc[key] = value
+    }
+
+    return acc
+  }, {})
+}
+
+export const logDrainHeaderEntriesSchema = z
+  .array(
+    z.object({
+      key: z.string().trim(),
+      value: z.string().trim(),
+    })
+  )
+  .max(20, HEADER_VALIDATION_ERRORS.MAX_LIMIT)
+  .superRefine((rows, ctx) => {
+    const rowIndexesByKey = new Map<string, number[]>()
+
+    rows.forEach((row, index) => {
+      const key = row.key.trim()
+      const value = row.value.trim()
+
+      if (!key && !value) return
+
+      if (key && !value) {
+        ctx.addIssue({
+          code: z.ZodIssueCode.custom,
+          message: HEADER_VALIDATION_ERRORS.VALUE_REQUIRED,
+          path: [index, 'value'],
+        })
+        return
+      }
+
+      if (!key && value) {
+        ctx.addIssue({
+          code: z.ZodIssueCode.custom,
+          message: HEADER_VALIDATION_ERRORS.KEY_REQUIRED,
+          path: [index, 'key'],
+        })
+        return
+      }
+
+      const existingIndexes = rowIndexesByKey.get(key) ?? []
+      existingIndexes.push(index)
+      rowIndexesByKey.set(key, existingIndexes)
+    })
+
+    rowIndexesByKey.forEach((indexes) => {
+      if (indexes.length < 2) return
+
+      indexes.forEach((index) => {
+        ctx.addIssue({
+          code: z.ZodIssueCode.custom,
+          message: HEADER_VALIDATION_ERRORS.DUPLICATE,
+          path: [index, 'key'],
+        })
+      })
+    })
+  })
+
 /**
 * Zod schema for OTLP log drain configuration
 * Extracted for testing purposes