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

import {
  ChangefeedNames,
  getChangefeedDestinationName,
  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, getEndpointServiceNameLabel } from 'dbaas/features/Changefeed/helpers'
import { getChangefeedClusterMetadata } from 'dbaas/features/Changefeed/types'
import { CloudProviderConst } from 'dbaas/features/Project'
import { EventFilterProperties } from 'dbaas/screens/Changefeed/ChangefeedDetail/TableFilterProperties'
import { getProtocolDisplayName } from 'dbaas/screens/Changefeed/CreateOrEditChangefeed/getProtocolDisplayName'
import HeadLabel from 'dbaas/screens/ClustersV2/new/components/HeadLabel'
import { SharedTiDB } from 'dbaas/screens/NextGen/shared/model/shared-tidb-state'
import {
  ngCalChangefeedFee,
  useListChangefeedSpecsX,
  useNgCreateChangefeed,
  useNgEditChangefeed,
  useNgGetSinkEndpoint
} from 'dbaas/services/dataflow'
import {
  ConsoleapiDataFormat,
  ConsoleapiPulsarDataFormat,
  ConsoleapiProtocol,
  ConsoleapiS3Dataformat,
  ConsoleapiStartPosition,
  DataflowConnectorType,
  DataflowDataFormat,
  DataflowDispatchType,
  DataflowKafkaAuthType,
  DataflowProtocol,
  DataflowS3Dataformat,
  DataflowS3Protocol,
  DataflowStartMode,
  DataflowTableConfig
} from 'dbaas/services/dataflow/interface'
import { getErrorMessage } from 'dbaas/services/errorCodes'
import { eventTracking } from 'dbaas/utils/tracking'

import { renderDescriptiveSelectOption } from '../../CreateOrEditChangefeed/renderDescriptiveSelectOption'
import { getStartModeLabel, isPrivateConnect, isTestEnv } from '../helper'
import type { CreateOrEditChangefeedFormData } from '../types'

import { buildCreateChangefeedPayload, buildEditChangefeedPayload } from './payloadBuilders'

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>
  )
}

const displayStorageUri = (uri?: string, isAliCloudOSS?: boolean) => {
  if (!uri || !isAliCloudOSS) {
    return uri
  }

  return uri.replace(/^s3:\/\//, 'oss://')
}

export const Review = ({
  onGoPrev,
  onCancel,
  cluster
}: {
  cluster: SharedTiDB
  onGoPrev: () => void
  onCancel: () => void
}) => {
  const isEditing = useIsEditingChangefeed()
  const duplicateFromId = new URLSearchParams(window.location.search).get('from_id') as string
  const history = useHistory()
  const createChangefeed = useNgCreateChangefeed()
  const editChangefeed = useNgEditChangefeed()
  const { id: clusterId, version } = useMemo(() => getChangefeedClusterMetadata(cluster), [cluster])
  const isAliCloud = cluster.cloudProvider === CloudProviderConst.ALICLOUD
  const methods = useFormContext<CreateOrEditChangefeedFormData>()
  const { handleSubmit, setValue, watch } = methods
  const selectedSpec = watch('changefeed_spec')
  const error = isEditing ? editChangefeed.error : createChangefeed.error

  const { data: changefeedFeeData, isFetching } = useQuery(
    ['ngCalChangefeedFee', clusterId, selectedSpec],
    () =>
      ngCalChangefeedFee(clusterId, {
        changefeed_spec: selectedSpec,
        duration: '60'
      }),
    {
      enabled: Boolean(clusterId && selectedSpec),
      refetchOnWindowFocus: false
    }
  )

  const formData = methods.getValues()
  const dataFormat =
    formData.changefeed_type === DataflowConnectorType.KAFKA ? formData.kafka_info?.data_format?.protocol : undefined
  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 ports = formData.network_info?.ports
  const sinkEndpointId = formData.network_info?.sink_endpoint_id

  const { data: sinkEndpointData } = useNgGetSinkEndpoint(clusterId, sinkEndpointId ?? '', {
    query: {
      enabled: !!sinkEndpointId,
      refetchOnWindowFocus: false
    }
  })
  const privateLinkServiceName = sinkEndpointData?.endpoint_service_name

  const advertisedPatternGroup =
    sinkEndpointData?.advertised_listener?.records?.map((record) => [
      record.zone ?? 'default',
      record.subdomain_name
    ]) ?? []

  const { data: changefeedSpecsData, isLoading: isLoadingSpecs } = useListChangefeedSpecsX(
    clusterId,
    {
      changefeed_type: formData.changefeed_type
    },
    {
      query: {
        enabled: !!formData.changefeed_type,
        onSuccess(data) {
          const items = data.items ?? []
          if (!selectedSpec && items.length > 0) {
            setValue('changefeed_spec', items[0]?.name ?? '')
          }
        }
      }
    }
  )

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

  const onCreate = async (data: CreateOrEditChangefeedFormData) => {
    const payload = isEditing ? buildEditChangefeedPayload(data) : buildCreateChangefeedPayload(data)

    try {
      // Skip actual submission in test environment
      if (isTestEnv()) {
        console.log(payload)
        return
      }

      if (isEditing) {
        await editChangefeed.mutateAsync({
          clusterId,
          changefeedId: data.changefeed_id!,
          data: payload
        })
        eventTracking(`Update changefeed successful`, {
          'Cluster Type': cluster.computed.clusterType
        })
      } else {
        await createChangefeed.mutateAsync({
          clusterId,
          data: payload
        })
        eventTracking(`Create changefeed successful`, {
          'Cluster Type': cluster.computed.clusterType,
          'Changefeed Type': data.changefeed_type
        })
      }

      history.push(`/tidbs/${clusterId}/changefeed`)
    } catch (_e) {
      eventTracking('Failed when create kafka changefeed', {
        'Cluster Type': cluster.computed.clusterType
      })
    }
  }

  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_spec"
                  data={options}
                  placeholder="Select the number of CCU"
                  renderOption={renderDescriptiveSelectOption}
                  disabled={isEditing || isLoadingSpecs}
                  label="Specify the number of changefeed capacity unit 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} data-testid="changefeed-review-cost">
                <Typography fw={500}>Cost</Typography>
                <Card
                  withBorder={false}
                  h={40}
                  bg="carbon.3"
                  px={12}
                  py={0}
                  miw={180}
                  sx={{ display: 'flex', alignItems: 'center', justifyContent: 'center' }}
                >
                  {isFetching ? (
                    <Loader size="xs" />
                  ) : (
                    <Group wrap="nowrap" gap={1} align="center" justify="flex-start" h={40}>
                      <Typography fz={20} fw={700} c="carbon.9">
                        {changefeedFeeData?.fee != null ? `$${changefeedFeeData.fee.toFixed(4)}` : '-'}
                      </Typography>
                      {changefeedFeeData?.fee != null && <Text c="carbon.7">/hour</Text>}
                    </Group>
                  )}
                </Card>
              </Stack>
            </Group>
          </Stack>

          <Divider />

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

            <Stack gap={8}>
              <ReviewProperty
                label="Destination"
                value={getChangefeedDestinationName(formData, cluster.cloudProvider)}
              />

              {formData.changefeed_type === DataflowConnectorType.KAFKA && (
                <>
                  <ReviewProperty
                    label="Connectivity Method"
                    value={getConnectionMethodDisplayName(formData.network_info?.network_type!)}
                  />
                  {isPrivateConnect(formData.network_info?.network_type) &&
                    formData.network_info?.sink_endpoint_id &&
                    sinkEndpointData && (
                      <ReviewProperty
                        label="Private Endpoint"
                        value={sinkEndpointData.endpoint_name ?? sinkEndpointData.endpoint_id ?? '-'}
                      />
                    )}
                  <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} />}
                  {sinkEndpointData && (
                    <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={getEndpointServiceNameLabel(cluster.cloudProvider)}
                      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.changefeed_type === DataflowConnectorType.MYSQL && (
                <>
                  {formData.mysql_info?.connection?.endpoint && (
                    <ReviewProperty label="MySQL Endpoint" value={formData.mysql_info?.connection?.endpoint} />
                  )}
                  {isPrivateConnect(formData.network_info?.network_type) &&
                    formData.network_info?.sink_endpoint_id &&
                    sinkEndpointData && (
                      <ReviewProperty
                        label="Private Endpoint"
                        value={sinkEndpointData.endpoint_name ?? sinkEndpointData.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.changefeed_type === DataflowConnectorType.S3 && (
                <>
                  <ReviewProperty
                    label={isAliCloud ? 'OSS URI' : 'S3 URI'}
                    value={displayStorageUri(formData.s3_info?.endpoint?.s3Uri, isAliCloud)}
                  />
                </>
              )}

              {formData.changefeed_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.changefeed_type === DataflowConnectorType.KAFKA && (
                <>
                  <TableFilterProperties data={formData.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(version) && (
                    <EventSplittingProperty value={formData.kafka_info?.data_format?.output_raw_change_event} />
                  )}
                </>
              )}

              {formData.changefeed_type === DataflowConnectorType.MYSQL && (
                <>
                  <TableFilterProperties data={formData.table_config} />

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

              {formData.changefeed_type === DataflowConnectorType.S3 && (
                <>
                  <TableFilterProperties data={formData?.table_config} />
                  <StartPositionProperties data={formData.s3_info?.start_position} />
                  <DataFormatProperties data={formData.s3_info?.data_format} />
                  {canConfigureEventSplitting(version) && (
                    // @ts-expect-error
                    <EventSplittingProperty value={formData.s3_info?.data_format?.output_raw_change_event} />
                  )}
                </>
              )}

              {formData.changefeed_type === DataflowConnectorType.GCS && (
                <>
                  <TableFilterProperties data={formData?.table_config} />
                  <StartPositionProperties data={formData.gcs_info?.start_position} />
                  <DataFormatProperties data={formData.gcs_info?.data_format} />
                  {canConfigureEventSplitting(version) && (
                    // @ts-expect-error
                    <EventSplittingProperty value={formData.gcs_info?.data_format?.output_raw_change_event} />
                  )}
                </>
              )}

              {formData.changefeed_type === DataflowConnectorType.PULSAR && (
                <>
                  <TableFilterProperties data={formData?.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.changefeed_type!]} changefeed Cancel Button Clicked`}
              onClick={onCancel}
            >
              Cancel
            </Button>

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

            <Button
              type="submit"
              loading={createChangefeed.isLoading || editChangefeed.isLoading}
              data-mp-event={`Create ${ChangefeedNames[formData.changefeed_type!]} changefeed Create Button Clicked`}
              data-mp-duplicate-from-id={duplicateFromId || undefined}
              data-mp-format={dataFormat}
              data-mp-plan="Premium"
            >
              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?: ConsoleapiStartPosition }) {
  const modeLabel = getStartModeLabel(data?.mode)
  const positionValue =
    data?.mode === DataflowStartMode.FROM_TSO ? data.tso : data?.mode === DataflowStartMode.FROM_UTC ? data.utc : ''

  return (
    <ReviewProperty
      label="Start Position"
      value={positionValue ? `${modeLabel}: ${positionValue}` : modeLabel || '-'}
    />
  )
}

type ReviewDataFormat =
  | ConsoleapiDataFormat
  | ConsoleapiPulsarDataFormat
  | ConsoleapiS3Dataformat
  | DataflowDataFormat
  | DataflowS3Dataformat

type ReviewStructuredDataFormat = ConsoleapiDataFormat | DataflowDataFormat

function isStructuredDataFormat(data?: ReviewDataFormat): data is ReviewStructuredDataFormat {
  return (
    data?.protocol === DataflowProtocol.AVRO ||
    data?.protocol === DataflowProtocol.DEBEZIUM ||
    data?.protocol === ConsoleapiProtocol.DEBEZIUM_AVRO
  )
}

function DataFormatProperties({ data }: { data?: ReviewDataFormat }) {
  const isDebeziumProtocol = isStructuredDataFormat(data) && data.protocol === DataflowProtocol.DEBEZIUM
  const isAvroProtocol =
    isStructuredDataFormat(data) &&
    (data.protocol === DataflowProtocol.AVRO || data.protocol === ConsoleapiProtocol.DEBEZIUM_AVRO)

  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'} />
        </>
      )}

      {isDebeziumProtocol && (
        <>
          <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'}
          />
        </>
      )}

      {isAvroProtocol && (
        <>
          <ReviewProperty label="Decimal Handling" value={DecimalOptionMap[data?.decimal_handling_mode!]} />
          <ReviewProperty
            label="Unsigned BigInt Handling"
            value={BigIntOptionMap[data?.bigint_unsigned_handling_mode!]}
          />
          <ReviewProperty label="Schema Registry" value={isAvroProtocol ? '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} />
          )}
        </>
      )}
    </>
  )
}