import { Box, Button, Card, Divider, Group, Loader, Stack, Text, Typography } from '@tidbcloud/uikit'
import { FormErrorMessage, FormSelect, FormTextInput } from '@tidbcloud/uikit/biz'
import { dayjs } from '@tidbcloud/uikit/utils'
import { capitalize, omit, toUpper } from 'lodash-es'
import { Fragment, ReactNode, useMemo } from 'react'
import { FormProvider, useForm } from 'react-hook-form'
import { useQuery } from 'react-query'
import { useHistory } from 'react-router-dom'

import { ChangefeedNames, showTidbExtension, useIsEditingChangefeed } from 'dbaas/features/Changefeed'
import { getConnectionMethodDisplayName } from 'dbaas/features/Changefeed/connection'
import {
  AuthTypeOptionMap,
  BigIntOptionMap,
  DecimalOptionMap,
  getTopicDistributionModeOptions,
  KafkaVersionOptions,
  partitionDistributionOptions,
  PulsarAuthTypeOptions,
  PulsarCompressionOptions,
  PulsarDestinationProtocolOptions
} from 'dbaas/features/Changefeed/constants'
import { canConfigureEventSplitting } from 'dbaas/features/Changefeed/helpers'
import { SharedCluster } from 'dbaas/features/ClusterManagement/cluster'
import { CloudProviderConst } from 'dbaas/features/Project'
import { EventFilterProperties } from 'dbaas/screens/Changefeed/ChangefeedDetail/TableFilterProperties'
import { getProtocolDisplayName } from 'dbaas/screens/Changefeed/CreateOrEditChangefeed/getProtocolDisplayName'
import { renderDescriptiveSelectOption } from 'dbaas/screens/Changefeed/CreateOrEditChangefeed/renderDescriptiveSelectOption'
import type { FormData } from 'dbaas/screens/Changefeed/CreateOrEditChangefeed/types'
import HeadLabel from 'dbaas/screens/ClustersV2/new/components/HeadLabel'
import { useServerlessServiceGetCluster } from 'dbaas/services/api-clients/v2023/serverless/v1beta1'
import {
  calChangefeedFee,
  useCreateConnector,
  useEditConnector,
  useGetSinkEndpoint,
  useListChangefeedInstanceTypes
} from 'dbaas/services/dataflow'
import {
  DataflowConnectorType,
  DataflowDataFormat,
  DataflowDispatchType,
  DataflowKafkaAuthType,
  DataflowProtocol,
  DataflowS3Dataformat,
  DataflowS3Protocol,
  DataflowStartMode,
  DataflowStartPosition,
  DataflowTableConfig
} from 'dbaas/services/dataflow/interface'
import { getErrorMessage } from 'dbaas/services/errorCodes'
import useStores from 'dbaas/stores/useStores'
import { eventTracking } from 'dbaas/utils/tracking'

function ReviewProperty({ label, value }: { label: ReactNode; value: ReactNode }) {
  return (
    <Group align="flex-start" wrap="nowrap">
      <Typography variant="body-lg" w={240} c="carbon.7" sx={{ flexShrink: 0 }}>
        {label}
      </Typography>
      <Typography c="carbon.9" sx={{ wordBreak: 'break-all', flexGrow: 1 }}>
        {value}
      </Typography>
    </Group>
  )
}

interface ConfigureSpecForm {
  name: string
  changefeed_instance_type: string
}

function Fee({ amount, deprecated }: { amount?: number; deprecated?: boolean }) {
  return (
    <Text
      span
      fz={deprecated ? 14 : 16}
      c={deprecated ? 'carbon.6' : 'carbon.7'}
      sx={
        deprecated
          ? {
              textDecorationLine: 'line-through',
              textDecorationThickness: '1px',
              textDecorationColor: 'var(--mantine-color-carbon-6)',
              textDecorationSkipInk: 'none'
            }
          : undefined
      }
    >
      <Text span fw={700} c={deprecated ? 'carbon.6' : 'carbon.9'} inherit>
        {`$${amount?.toFixed(4) ?? 0}`}
      </Text>
      {!deprecated && (
        <Text span c="carbon.7">
          /hour
        </Text>
      )}
      {deprecated && (
        <Text span c="carbon.6" fw={400} fz={12} inherit>
          /hour
        </Text>
      )}
    </Text>
  )
}

export const Review = ({
  onGoPrev,
  onCancel,
  formData,
  cluster
}: {
  cluster: SharedCluster
  formData: FormData
  onGoPrev: () => void
  onCancel: () => void
}) => {
  const {
    store: { tenantData, activeProjectId, currentCluster }
  } = useStores()
  const isEditing = useIsEditingChangefeed()
  const duplicateFromId = new URLSearchParams(window.location.search).get('from_id') as string
  const history = useHistory()
  const createConnector = useCreateConnector({
    request: { timeout: 60 * 1000 * 15 }
  })
  const editConnector = useEditConnector({
    request: { timeout: 60 * 1000 * 15 }
  })
  const methods = useForm<ConfigureSpecForm>({
    defaultValues: {
      name: formData.name ?? `changefeed-${formData.connector_type?.toLowerCase()}-${dayjs().format('YYYYMMDD')}`,
      changefeed_instance_type: formData.changefeed_instance_type ?? ''
    }
  })
  const { handleSubmit, watch, setValue } = methods
  const instanceType = watch('changefeed_instance_type')
  const error = isEditing ? editConnector.error : createConnector.error

  const { data: targetClusterData } = useServerlessServiceGetCluster(
    formData.serverless_tidb_info?.cluster_id!,
    {
      orgId: tenantData.id,
      projectId: activeProjectId
    },
    {
      query: {
        enabled: !!formData.serverless_tidb_info?.cluster_id
      }
    }
  )
  const { data, isFetching } = useQuery(
    ['/cal_changefeed_fee', instanceType],
    () =>
      calChangefeedFee(tenantData.id, activeProjectId, {
        cloud_provider_name: currentCluster?.provider,
        region_name: currentCluster?.region,
        changefeed_instance_type: instanceType,
        duration: '60'
      }),
    {
      enabled: Boolean(instanceType),
      refetchOnWindowFocus: false
    }
  )
  const shouldShowOriginalFee = !!data && data.raw_fee !== data.fee

  const brokerEndpoints = formData.kafka_info?.broker?.broker_endpoints
  const tlsEncryption = formData.kafka_info?.broker?.tls_enable
  const insecureSkipVerify = formData.kafka_info?.broker?.insecure_skip_verify
  const authType = formData.kafka_info?.authentication?.auth_type
  const privateLinkServiceName = formData.network_info?.private_link_service_name
  const ports = formData.network_info?.ports
  const privateEndpointId = formData.network_info?.sink_endpoint_id
  const isGcp = cluster.provider === CloudProviderConst.GCP
  const isAzure = cluster.provider === CloudProviderConst.AZURE
  const { data: privateLinkData } = useGetSinkEndpoint(tenantData.id, activeProjectId, privateEndpointId!, {
    query: {
      refetchOnWindowFocus: false,
      enabled: !!privateEndpointId
    }
  })

  const advertisedPatternGroup =
    privateLinkData?.advertised_listener?.records?.map((i) => [i.zone, i.subdomain_name]) ?? []

  const { data: instanceTypesData, isLoading: isLoadingInstanceTypes } = useListChangefeedInstanceTypes(
    {
      connector_type: formData.connector_type,
      cluster_id: currentCluster?.id
    },
    {
      query: {
        onSuccess(data) {
          if (!instanceType && data.items.length > 0) {
            setValue('changefeed_instance_type', data.items[0].name!)
          }
        }
      }
    }
  )

  const options = useMemo(() => {
    return (
      instanceTypesData?.items.map((i) => ({
        label: `${i.rcu_num} RCUs`,
        description: `The maximum replication performance is ${i.rps_limit} rows/s`,
        value: i.name!
      })) ?? []
    )
  }, [instanceTypesData])

  const onCreate = async (data: ConfigureSpecForm) => {
    const s3AccessType =
      formData.connector_type === DataflowConnectorType.S3 ? formData.s3_info?.endpoint?.auth_type : undefined
    const eventTrackingProps = {
      'Cluster Type': cluster.servicePlan,
      'Changefeed Type': formData.connector_type,
      ConnectionMethod: formData.network_info?.network_type,
      s3AccessType
    }

    try {
      // Skip actual submission in test environment
      const isTestEnv =
        process.env.NODE_ENV === 'test' || window.location.search.includes('e2e-test') || window.E2E_Test
      if (isTestEnv) {
        const payload = {
          ...omit(formData, 'password'),
          connector_type: formData.connector_type!,
          name: data.name,
          changefeed_instance_type: data.changefeed_instance_type
        }
        console.log(payload)
        console.log('[E2E Test]-[CreateChangefeed]-[Step3]-[Payload]-' + JSON.stringify(payload))
        return
      }

      if (isEditing) {
        // network_info is not allowed to be updated
        const { network_info, ...rest } = formData

        await editConnector.mutateAsync({
          orgId: tenantData.id,
          projectId: activeProjectId,
          clusterId: cluster.id,
          data: {
            ...rest,
            connector_id: formData.connector_id!,
            connector_type: formData.connector_type!,
            name: data.name
          }
        })
        eventTracking(`Update changefeed successful`, eventTrackingProps)
      } else {
        await createConnector.mutateAsync({
          orgId: tenantData.id,
          projectId: activeProjectId,
          clusterId: cluster.id,
          data: {
            ...omit(formData, 'password'),
            connector_type: formData.connector_type!,
            name: data.name,
            changefeed_instance_type: data.changefeed_instance_type
          }
        })
        eventTracking(`Create changefeed successful`, eventTrackingProps)
      }

      history.push(`/clusters/${cluster.id}/changefeed`)
    } catch (_e) {
      eventTracking('Failed when create kafka changefeed', eventTrackingProps)
    }
  }

  return (
    <FormProvider {...methods}>
      <Card p={0}>
        <form onSubmit={handleSubmit(onCreate)}>
          <Stack m={24}>
            <HeadLabel label="Specification and Name" />

            <Group wrap="nowrap" align="flex-start">
              <Box sx={{ flex: 1 }}>
                <FormSelect
                  rules={{ required: 'Required' }}
                  name="changefeed_instance_type"
                  data={options}
                  placeholder="Select the number of RCU"
                  renderOption={renderDescriptiveSelectOption}
                  disabled={isEditing || isLoadingInstanceTypes}
                  label="Specify the number of RCU used by this changefeed"
                  mb={16}
                />

                <FormTextInput
                  name="name"
                  label="Give this changefeed a name to make it easy to find"
                  placeholder="Optional, enter the name of the changefeed"
                  disabled={isEditing}
                />
              </Box>

              <Stack gap={4}>
                <Typography fw={500}>Cost</Typography>
                <Card
                  withBorder={false}
                  h={shouldShowOriginalFee ? 60 : 40}
                  bg="carbon.3"
                  px={12}
                  py={0}
                  miw={180}
                  sx={{ display: 'flex', alignItems: 'center', justifyContent: 'center' }}
                >
                  {isFetching ? (
                    <Loader size="xs" />
                  ) : (
                    <Stack gap={0} align="center">
                      {data ? <Fee amount={data.fee} /> : '-'}
                      {shouldShowOriginalFee && <Fee amount={data.raw_fee} deprecated />}
                    </Stack>
                  )}
                </Card>
              </Stack>
            </Group>
          </Stack>

          <Divider />

          <Stack m={24}>
            <HeadLabel label="Replication Destination" />

            <Stack gap={8}>
              <ReviewProperty label="Destination" value={ChangefeedNames[formData.connector_type!]} />

              {formData.connector_type === DataflowConnectorType.KAFKA && (
                <>
                  <ReviewProperty
                    label="Connectivity Method"
                    value={getConnectionMethodDisplayName(formData.network_info?.network_type!)}
                  />
                  <ReviewProperty
                    label="Kafka Version"
                    value={
                      KafkaVersionOptions.find((i) => i.value === formData.kafka_info?.broker?.kafka_version)?.label
                    }
                  />
                  <ReviewProperty label="Compress" value={formData.kafka_info?.broker?.compression?.toLowerCase()} />
                  {brokerEndpoints && <ReviewProperty label="Brokers Endpoints" value={brokerEndpoints} />}
                  {privateEndpointId && (
                    <ReviewProperty
                      label="Kafka Advertised Listener Pattern"
                      value={
                        advertisedPatternGroup.length > 0 ? (
                          <Stack gap={4}>
                            {advertisedPatternGroup.map(([key, value]) => {
                              if (advertisedPatternGroup.length === 1) {
                                return <span>{value}</span>
                              }
                              return <span>{`${key}: ${value}`}</span>
                            })}
                          </Stack>
                        ) : (
                          '-'
                        )
                      }
                    />
                  )}
                  {privateLinkServiceName && (
                    <ReviewProperty
                      label={(() => {
                        if (isGcp) {
                          return `Service Attachment`
                        }
                        if (isAzure) {
                          return `Alias of Private Link Service`
                        }
                        return `Endpoint Service Name`
                      })()}
                      value={privateLinkServiceName}
                    />
                  )}
                  {ports && <ReviewProperty label="Bootstrap Ports" value={ports.join(', ')} />}
                  <ReviewProperty label="TLS Encryption" value={tlsEncryption ? 'Enable' : 'Disable'} />
                  {tlsEncryption && (
                    <ReviewProperty label="Insecure Skip Verify" value={insecureSkipVerify ? 'Enable' : 'Disable'} />
                  )}
                  <ReviewProperty label="Authentication" value={AuthTypeOptionMap[authType!]} />

                  {authType !== DataflowKafkaAuthType.DISABLE && (
                    <ReviewProperty label="SASL User Name" value={formData.kafka_info?.authentication?.userName} />
                  )}
                </>
              )}

              {formData.connector_type === DataflowConnectorType.MYSQL && (
                <>
                  {formData.mysql_info?.connection?.endpoint && (
                    <ReviewProperty label="MySQL Endpoint" value={formData.mysql_info?.connection?.endpoint} />
                  )}

                  {formData.network_info?.sink_endpoint_id && (
                    <ReviewProperty label="Private Endpoint" value={formData.network_info?.sink_endpoint_id} />
                  )}

                  {formData.network_info?.ports?.[0] && (
                    <ReviewProperty label="Port" value={formData.network_info?.ports?.[0]} />
                  )}

                  <ReviewProperty label="MySQL Username" value={formData.mysql_info?.connection?.username ?? '-'} />
                </>
              )}

              {formData.connector_type === DataflowConnectorType.S3 && (
                <>
                  <ReviewProperty label="S3 URI" value={formData.s3_info?.endpoint?.s3Uri} />
                </>
              )}

              {formData.connector_type === DataflowConnectorType.AZURE_BLOB_STORAGE && (
                <>
                  <ReviewProperty label="Blob URL" value={formData.azure_blob_storage_info?.endpoint?.blob_uri} />
                </>
              )}

              {formData.connector_type === DataflowConnectorType.PULSAR && (
                <>
                  <ReviewProperty
                    label="Protocol"
                    value={
                      PulsarDestinationProtocolOptions.find(
                        (i) => i.value === formData.pulsar_info?.destination_protocol
                      )?.label
                    }
                  />
                  <ReviewProperty
                    label="Connectivity Method"
                    value={getConnectionMethodDisplayName(formData.network_info?.network_type!)}
                  />
                  <ReviewProperty label="Broker Endpoint" value={formData.pulsar_info?.endpoint} />
                  <ReviewProperty
                    label="Authentication"
                    value={
                      PulsarAuthTypeOptions.find((i) => i.value === formData.pulsar_info?.authentication?.auth_type)
                        ?.label
                    }
                  />
                  {formData.pulsar_info?.authentication?.user_name && (
                    <ReviewProperty label="User Name" value={formData.pulsar_info?.authentication?.user_name} />
                  )}
                  <ReviewProperty
                    label="Compression"
                    value={
                      PulsarCompressionOptions.find((i) => i.value === formData.pulsar_info?.compression_type)?.label
                    }
                  />
                </>
              )}
            </Stack>
          </Stack>

          <Divider />

          <Stack m={24}>
            <HeadLabel label="Replication Settings" />
            <Stack gap={8}>
              {formData.connector_type === DataflowConnectorType.KAFKA && (
                <>
                  <TableFilterProperties data={formData.kafka_info?.table_config} />

                  {!!formData.kafka_info?.column_selectors?.length && (
                    <ReviewProperty
                      label="Column Selector"
                      value={
                        <Stack gap={4}>
                          {formData.kafka_info?.column_selectors.map((i, idx) => (
                            <span key={idx}>
                              {i.matcher?.join(', ')}: {i.columns?.join(', ')}
                            </span>
                          ))}
                        </Stack>
                      }
                    />
                  )}

                  <StartPositionProperties data={formData.kafka_info?.start_position} />

                  <DataFormatProperties data={formData.kafka_info?.data_format} />

                  <ReviewProperty
                    label="Topic Distribution Mode"
                    value={
                      getTopicDistributionModeOptions({
                        target: 'Kafka'
                      }).find((i) => i.value === formData.kafka_info?.topic_partition_config?.dispatch_type)?.label
                    }
                  />

                  <ReviewProperty
                    label="Topic Name"
                    value={formData.kafka_info?.topic_partition_config?.default_topic!}
                  />

                  {formData.kafka_info?.topic_partition_config?.dispatch_type !==
                    DataflowDispatchType.DISPATCH_TYPE_ONE_TOPIC && (
                    <>
                      {formData.kafka_info?.topic_partition_config?.topic_prefix && (
                        <ReviewProperty
                          label="Topic Prefix"
                          value={formData.kafka_info?.topic_partition_config?.topic_prefix}
                        />
                      )}
                      {formData.kafka_info?.topic_partition_config?.separator &&
                        formData.kafka_info?.topic_partition_config?.dispatch_type ===
                          DataflowDispatchType.DISPATCH_TYPE_BY_TABLE && (
                          <ReviewProperty
                            label="Topic Separator"
                            value={formData.kafka_info?.topic_partition_config?.separator}
                          />
                        )}
                      {formData.kafka_info?.topic_partition_config?.topic_suffix && (
                        <ReviewProperty
                          label="Topic Suffix"
                          value={formData.kafka_info?.topic_partition_config?.topic_suffix}
                        />
                      )}
                    </>
                  )}

                  {formData.kafka_info?.topic_partition_config?.partition_dispatchers && (
                    <ReviewProperty
                      label="Partition Dispatcher"
                      value={
                        <Box sx={{ display: 'grid', gridTemplateColumns: 'fit-content(50%) 1fr', gap: 4 }}>
                          {formData.kafka_info?.topic_partition_config?.partition_dispatchers?.map((i, index) => (
                            <Fragment key={index}>
                              <Typography>
                                {i.matcher?.join(',')}
                                {' :'}
                              </Typography>

                              <Typography>
                                {[
                                  partitionDistributionOptions.find((j) => j.value === i.partition_type)?.text,
                                  i.index_name,
                                  i.columns?.join(',')
                                ]
                                  .filter(Boolean)
                                  .join(', ')}
                              </Typography>
                            </Fragment>
                          ))}
                        </Box>
                      }
                    />
                  )}

                  <ReviewProperty
                    label="Replication Factor"
                    value={formData.kafka_info?.topic_partition_config?.replication_factor}
                  />

                  <ReviewProperty
                    label="Partition Number"
                    value={formData.kafka_info?.topic_partition_config?.partition_num}
                  />

                  {canConfigureEventSplitting(cluster.version) && (
                    <EventSplittingProperty value={formData.kafka_info?.data_format?.output_raw_change_event} />
                  )}
                </>
              )}

              {formData.connector_type === DataflowConnectorType.MYSQL && (
                <>
                  <TableFilterProperties data={formData.mysql_info?.table_config} />

                  <StartPositionProperties data={formData?.mysql_info?.start_position} />
                </>
              )}

              {formData.connector_type === DataflowConnectorType.SERVERLESS_TIDB && (
                <>
                  <ReviewProperty
                    label="Cluster"
                    value={targetClusterData?.displayName ?? formData.serverless_tidb_info?.cluster_id}
                  />
                  <ReviewProperty
                    label="User Name"
                    value={formData.serverless_tidb_info?.mysql_info?.connection?.username}
                  />

                  <ReviewProperty
                    label="Start Position"
                    value={formData.serverless_tidb_info?.mysql_info?.start_position?.tso}
                  />

                  <TableFilterProperties data={formData.serverless_tidb_info?.mysql_info?.table_config} />
                </>
              )}

              {formData.connector_type === DataflowConnectorType.S3 && (
                <>
                  <TableFilterProperties data={formData.s3_info?.table_config} />
                  <StartPositionProperties data={formData.s3_info?.start_position} />
                  <DataFormatProperties data={formData.s3_info?.data_format} />
                  {canConfigureEventSplitting(cluster.version) && (
                    <EventSplittingProperty value={formData.s3_info?.data_format?.output_raw_change_event} />
                  )}
                </>
              )}

              {formData.connector_type === DataflowConnectorType.GCS && (
                <>
                  <TableFilterProperties data={formData.gcs_info?.table_config} />
                  <StartPositionProperties data={formData.gcs_info?.start_position} />
                  <DataFormatProperties data={formData.gcs_info?.data_format} />
                  {canConfigureEventSplitting(cluster.version) && (
                    <EventSplittingProperty value={formData.gcs_info?.data_format?.output_raw_change_event} />
                  )}
                </>
              )}

              {formData.connector_type === DataflowConnectorType.AZURE_BLOB_STORAGE && (
                <>
                  <TableFilterProperties data={formData.azure_blob_storage_info?.table_config} />
                  <StartPositionProperties data={formData.azure_blob_storage_info?.start_position} />
                  <DataFormatProperties data={formData.azure_blob_storage_info?.data_format} />
                  {canConfigureEventSplitting(cluster.version) && (
                    <EventSplittingProperty
                      value={formData.azure_blob_storage_info?.data_format?.output_raw_change_event}
                    />
                  )}
                </>
              )}

              {formData.connector_type === DataflowConnectorType.PULSAR && (
                <>
                  <TableFilterProperties data={formData.pulsar_info?.table_config} />
                  <StartPositionProperties data={formData.pulsar_info?.start_position} />
                  <DataFormatProperties data={formData.pulsar_info?.data_format} />

                  <ReviewProperty
                    label="Topic Distribution Mode"
                    value={
                      getTopicDistributionModeOptions({
                        target: 'Pulsar'
                      }).find((i) => i.value === formData.pulsar_info?.topic_distribution?.dispatch_type)?.label
                    }
                  />

                  <ReviewProperty label="Topic Name" value={formData.pulsar_info?.topic_distribution?.default_topic} />
                  {formData.pulsar_info?.topic_distribution?.dispatch_type !==
                    DataflowDispatchType.DISPATCH_TYPE_ONE_TOPIC && (
                    <>
                      {formData.pulsar_info?.topic_distribution?.topic_prefix && (
                        <ReviewProperty
                          label="Topic Prefix"
                          value={formData.pulsar_info?.topic_distribution?.topic_prefix}
                        />
                      )}
                      {formData.pulsar_info?.topic_distribution?.separator &&
                        formData.pulsar_info?.topic_distribution?.dispatch_type ===
                          DataflowDispatchType.DISPATCH_TYPE_BY_TABLE && (
                          <ReviewProperty
                            label="Topic Separator"
                            value={formData.pulsar_info?.topic_distribution?.separator}
                          />
                        )}
                      {formData.pulsar_info?.topic_distribution?.topic_suffix && (
                        <ReviewProperty
                          label="Topic Suffix"
                          value={formData.pulsar_info?.topic_distribution?.topic_suffix}
                        />
                      )}
                    </>
                  )}

                  <ReviewProperty
                    label="Pulsar Tenant"
                    value={formData.pulsar_info?.topic_distribution?.pulsar_tenant}
                  />
                  <ReviewProperty
                    label="Pulsar Namespace"
                    value={formData.pulsar_info?.topic_distribution?.pulsar_namespace}
                  />

                  {formData.pulsar_info?.partition_dispatchers?.length && (
                    <ReviewProperty
                      label="Partition Dispatcher"
                      value={
                        <Box sx={{ display: 'grid', gridTemplateColumns: 'fit-content(50%) 1fr', gap: 4 }}>
                          {formData.pulsar_info?.partition_dispatchers?.map((i, index) => (
                            <Fragment key={index}>
                              <Typography>
                                {i.matcher?.join(',')}
                                {' :'}
                              </Typography>

                              <Typography>
                                {[
                                  partitionDistributionOptions.find((j) => j.value === i.partition_type)?.text,
                                  i.index_name,
                                  i.columns?.join(',')
                                ]
                                  .filter(Boolean)
                                  .join(', ')}
                              </Typography>
                            </Fragment>
                          ))}
                        </Box>
                      }
                    />
                  )}
                </>
              )}
            </Stack>
          </Stack>

          <Divider />

          {error && <FormErrorMessage message={getErrorMessage(error)} m={24} />}

          <Group m={24} justify="flex-end">
            <Button
              variant="default"
              type="button"
              data-mp-event={`Create ${ChangefeedNames[formData.connector_type!]} changefeed Cancel Button Clicked`}
              onClick={onCancel}
            >
              Cancel
            </Button>

            <Button
              variant="outline"
              type="button"
              data-mp-event={`Create ${ChangefeedNames[formData.connector_type!]} changefeed Previous Button Clicked`}
              onClick={onGoPrev}
            >
              Previous
            </Button>

            <Button
              type="submit"
              loading={createConnector.isLoading || editConnector.isLoading}
              data-mp-event={`Create ${ChangefeedNames[formData.connector_type!]} changefeed Create Button Clicked`}
              data-mp-duplicate-from-id={duplicateFromId || undefined}
              data-mp-private-endpoint-id={formData.network_info?.sink_endpoint_id || 'N/A'}
            >
              Submit
            </Button>
          </Group>
        </form>
      </Card>
    </FormProvider>
  )
}

function EventSplittingProperty({ value }: { value: boolean | undefined }) {
  return <ReviewProperty label="Output Raw Events" value={value ? 'Yes' : 'No'} />
}

function TableFilterProperties({ data }: { data?: DataflowTableConfig }) {
  return (
    <>
      <ReviewProperty label="Table Filter" value={data?.filter_rule?.join(', ') || '-'} />
      {data?.case_sensitive !== undefined && (
        <ReviewProperty label={<Box pl={8}>• Case Sensitive</Box>} value={data?.case_sensitive ? 'Yes' : 'No'} />
      )}
      {!!data?.event_filter_rule?.length && (
        <ReviewProperty label="Event Filter" value={<EventFilterProperties data={data} />} />
      )}
    </>
  )
}

function StartPositionProperties({ data }: { data?: DataflowStartPosition }) {
  return (
    <ReviewProperty
      label="Start Position"
      value={
        <Group>
          <span>{toUpper(data?.mode?.split('_').join(' '))}</span>
          {data?.mode === DataflowStartMode.FROM_TSO && data.tso && <span>{data.tso}</span>}
          {data?.mode === DataflowStartMode.FROM_UTC && data.utc && <span>{data.utc}</span>}
        </Group>
      }
    />
  )
}

function DataFormatProperties({ data }: { data?: DataflowS3Dataformat | DataflowDataFormat }) {
  return (
    <>
      <ReviewProperty label="Data Format" value={getProtocolDisplayName(data?.protocol!)} />

      {data && 'date_separator' in data && (
        <ReviewProperty label="Date Separator" value={capitalize(data?.date_separator)} />
      )}

      {data?.protocol === DataflowS3Protocol.S3_PROTOCOL_CSV && (
        <>
          <ReviewProperty label="Binary Encoding Method" value={capitalize(data.binary_encoding_method)} />
          <ReviewProperty label="Delimiter" value={data.delimiter} />
          <ReviewProperty label="Quote" value={data.quote} />
          <ReviewProperty label="Null" value={data.null} />
          <ReviewProperty label="Include Commit Ts" value={data.include_commit_ts ? 'True' : 'False'} />
        </>
      )}

      {data && showTidbExtension(data.protocol!) && (
        <>
          <ReviewProperty label="TiDB Extension" value={data.enable_tidb_extension ? 'Enable' : 'Disable'} />
        </>
      )}

      {data?.protocol === DataflowProtocol.DEBEZIUM && (
        <>
          <ReviewProperty
            label="Include Old Values in Output"
            value={data?.debezium_config?.output_old_value ? 'Yes' : 'No'}
          />
          <ReviewProperty
            label="Include Schema Description in Output"
            value={data?.debezium_config?.disable_schema ? 'No' : 'Yes'}
          />
        </>
      )}

      {data?.protocol === DataflowProtocol.AVRO && (
        <>
          <ReviewProperty label="Decimal Format" value={DecimalOptionMap[data?.decimal_handling_mode!]} />
          <ReviewProperty
            label="Unsigned BigInt Format"
            value={BigIntOptionMap[data?.bigint_unsigned_handling_mode!]}
          />
          <ReviewProperty
            label="Schema Registry"
            value={data?.protocol === DataflowProtocol.AVRO ? 'Enable' : 'Disable'}
          />
          <ReviewProperty label="Schema Registry Endpoints" value={data.schema_registry?.schema_registry_endpoints} />
          <ReviewProperty
            label="HTTP Authentication"
            value={data.schema_registry?.enableHttpAuth ? 'Enable' : 'Disable'}
          />
          {data.schema_registry?.enableHttpAuth && (
            <ReviewProperty label="Schema Registry Username" value={data.schema_registry?.userName} />
          )}
        </>
      )}
    </>
  )
}