import { Box, Stack, Typography } from '@tidbcloud/uikit'
import { PropertyCard } from '@tidbcloud/uikit/biz'
import { Fragment } from 'react'

import { showTidbExtension } from 'dbaas/features/Changefeed'
import { getConnectionMethodDisplayName } from 'dbaas/features/Changefeed/connection'
import {
  AuthTypeOptionMap,
  getTopicDistributionModeOptions,
  KafkaCompressionOptions,
  KafkaVersionOptions,
  partitionDistributionOptions
} from 'dbaas/features/Changefeed/constants'
import { canConfigureEventSplitting, getEndpointServiceNameLabel } from 'dbaas/features/Changefeed/helpers'
import {
  DataflowConnector,
  DataflowDispatchType,
  DataflowNetworkType,
  DataflowSinkEndpoint
} from 'dbaas/services/dataflow/interface'

import { DataFormatProperties } from './DataFormatProperties'
import { StartPositionProperty } from './StartPositionDisplay'
import { TableConfigProperties } from './TableFilterProperties'

interface KafkaPropertiesProps {
  data: DataflowConnector
  sinkEndpointData?: DataflowSinkEndpoint
  clusterVersion: string
  cloudProvider: string
}

export const KafkaProperties = ({ data, sinkEndpointData, clusterVersion, cloudProvider }: KafkaPropertiesProps) => {
  const networkType = data.request?.network_info?.network_type
  const patternGroup = data.request?.network_info?.advertised_pattern_group ?? {}
  const patternGroupEntries = Object.entries(patternGroup)

  return (
    <>
      <PropertyCard.Divider />
      <PropertyCard.Item label="Connection Method">{getConnectionMethodDisplayName(networkType!)}</PropertyCard.Item>
      <PropertyCard.Item label="Kafka Version">
        {KafkaVersionOptions.find((i) => i.value === data.request?.kafka_info?.broker?.kafka_version)?.label || '-'}
      </PropertyCard.Item>
      <PropertyCard.Item label="Compress">
        {KafkaCompressionOptions.find((i) => i.value === data.request?.kafka_info?.broker?.compression)?.label || '-'}
      </PropertyCard.Item>
      {(networkType === DataflowNetworkType.NetworkType_Public ||
        networkType === DataflowNetworkType.NetworkType_VPC_Peering) && (
        <PropertyCard.Item label="Broker Endpoints">
          {data.request?.kafka_info?.broker?.broker_endpoints || '-'}
        </PropertyCard.Item>
      )}
      {sinkEndpointData && (
        <>
          {sinkEndpointData.endpoint_name && (
            <PropertyCard.Item label="Private Endpoint Name">{sinkEndpointData.endpoint_name}</PropertyCard.Item>
          )}
          {sinkEndpointData.endpoint_service_name && (
            <PropertyCard.Item label={getEndpointServiceNameLabel(cloudProvider)}>
              {sinkEndpointData.endpoint_service_name}
            </PropertyCard.Item>
          )}
          <PropertyCard.Item label="Kafka Advertised Listener Pattern" align="start">
            {patternGroupEntries.length > 0 ? (
              <Stack gap={4}>
                {patternGroupEntries.map(([key, value]) => (
                  <div key={key}>{patternGroupEntries.length > 1 ? `${key}: ${value}` : value}</div>
                ))}
              </Stack>
            ) : (
              '-'
            )}
          </PropertyCard.Item>
          <PropertyCard.Item label="Bootstrap Ports">
            {data.request?.network_info?.ports?.join(', ') || '-'}
          </PropertyCard.Item>
        </>
      )}
      <PropertyCard.Item label="TLS Encryption">
        {data.request?.kafka_info?.broker?.tls_enable ? 'Enabled' : 'Disabled'}
      </PropertyCard.Item>
      {data.request?.kafka_info?.broker?.tls_enable && (
        <PropertyCard.Item label="Insecure Skip Verify">
          {data.request?.kafka_info?.broker?.insecure_skip_verify ? 'Enabled' : 'Disabled'}
        </PropertyCard.Item>
      )}
      <PropertyCard.Item label="Authentication">
        {AuthTypeOptionMap[data.request?.kafka_info?.authentication?.auth_type!] || '-'}
      </PropertyCard.Item>
      {data.request?.kafka_info?.authentication?.userName && (
        <PropertyCard.Item label="User Name">{data.request.kafka_info.authentication.userName}</PropertyCard.Item>
      )}

      <PropertyCard.Divider />

      <TableConfigProperties data={data.request?.kafka_info?.table_config} />

      {data.request?.kafka_info?.column_selectors && data.request?.kafka_info?.column_selectors.length > 0 && (
        <PropertyCard.Item label="Column Selector">
          <Stack gap={4}>
            {data.request.kafka_info.column_selectors.map((i, idx) => (
              <Typography variant="body-lg" key={idx} sx={{ wordBreak: 'break-all' }}>
                <b>{i.matcher?.join(', ')}</b>: {i.columns?.join(', ')}
              </Typography>
            ))}
          </Stack>
        </PropertyCard.Item>
      )}

      <StartPositionProperty data={data.request?.kafka_info?.start_position} />

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

      {data.request?.kafka_info?.data_format?.protocol &&
        showTidbExtension(data.request.kafka_info.data_format.protocol) && (
          <PropertyCard.Item label="TiDB Extension">
            {data.request?.kafka_info?.data_format?.enable_tidb_extension ? 'Enable' : 'Disable'}
          </PropertyCard.Item>
        )}

      <PropertyCard.Item label="Topic Distribution Mode">
        {getTopicDistributionModeOptions({ target: 'Kafka' }).find(
          (i) => i.value === data.request?.kafka_info?.topic_partition_config?.dispatch_type
        )?.label || '-'}
      </PropertyCard.Item>

      <PropertyCard.Item label="Topic Name">
        {data.request?.kafka_info?.topic_partition_config?.default_topic || '-'}
      </PropertyCard.Item>

      {data.request?.kafka_info?.topic_partition_config?.dispatch_type !==
        DataflowDispatchType.DISPATCH_TYPE_ONE_TOPIC && (
        <>
          {data.request?.kafka_info?.topic_partition_config?.topic_prefix && (
            <PropertyCard.Item label="Topic Prefix">
              {data.request.kafka_info.topic_partition_config.topic_prefix}
            </PropertyCard.Item>
          )}
          {data.request?.kafka_info?.topic_partition_config?.separator &&
            data.request?.kafka_info?.topic_partition_config?.dispatch_type ===
              DataflowDispatchType.DISPATCH_TYPE_BY_TABLE && (
              <PropertyCard.Item label="Topic Separator">
                {data.request.kafka_info.topic_partition_config.separator}
              </PropertyCard.Item>
            )}
          {data.request?.kafka_info?.topic_partition_config?.topic_suffix && (
            <PropertyCard.Item label="Topic Suffix">
              {data.request.kafka_info.topic_partition_config.topic_suffix}
            </PropertyCard.Item>
          )}
        </>
      )}

      {data.request?.kafka_info?.topic_partition_config?.partition_dispatchers && (
        <PropertyCard.Item label="Partition Dispatcher">
          <Box sx={{ display: 'grid', gridTemplateColumns: 'fit-content(50%) 1fr', gap: 4 }}>
            {data.request.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>
        </PropertyCard.Item>
      )}

      <PropertyCard.Item label="Replication Factor">
        {data.request?.kafka_info?.topic_partition_config?.replication_factor || '-'}
      </PropertyCard.Item>

      <PropertyCard.Item label="Partition Number">
        {data.request?.kafka_info?.topic_partition_config?.partition_num || '-'}
      </PropertyCard.Item>

      {canConfigureEventSplitting(clusterVersion) && (
        <PropertyCard.Item label="Output Raw Events">
          {data.request?.kafka_info?.data_format?.output_raw_change_event ? 'Yes' : 'No'}
        </PropertyCard.Item>
      )}
    </>
  )
}