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

import {
  Changefeed,
  KafkaAuthTypeEnum,
  KafkaCompressionTypeEnum,
  KafkaNetworkTypeEnum,
  KafkaProtocolEnum,
  KafkaVersionEnum,
  TopicDispatchTypeEnum
} from 'dbaas/services/api-clients/legacy/serverless/interface'

interface KafkaPropertiesEssentialProps {
  data: Changefeed
  cloudProvider: string
}

const AuthTypeOptionMap: Record<KafkaAuthTypeEnum, string> = {
  [KafkaAuthTypeEnum.DISABLE]: 'Disabled',
  [KafkaAuthTypeEnum.SASL_PLAIN]: 'SASL/PLAIN',
  [KafkaAuthTypeEnum.SASL_SCRAM_SHA_256]: 'SASL/SCRAM-SHA-256',
  [KafkaAuthTypeEnum.SASL_SCRAM_SHA_512]: 'SASL/SCRAM-SHA-512'
}

const NetworkTypeMap: Record<KafkaNetworkTypeEnum, string> = {
  [KafkaNetworkTypeEnum.PUBLIC]: 'Public',
  [KafkaNetworkTypeEnum.PRIVATE_LINK]: 'Private Link'
}

const ProtocolMap: Record<KafkaProtocolEnum, string> = {
  [KafkaProtocolEnum.CANAL_JSON]: 'Canal-JSON',
  [KafkaProtocolEnum.OPEN_PROTOCOL]: 'Open Protocol',
  [KafkaProtocolEnum.AVRO]: 'Avro',
  [KafkaProtocolEnum.DEBEZIUM]: 'Debezium'
}

const TopicDispatchTypeMap: Record<TopicDispatchTypeEnum, string> = {
  [TopicDispatchTypeEnum.ONE_TOPIC]: 'Single Topic',
  [TopicDispatchTypeEnum.BY_TABLE]: 'By Table',
  [TopicDispatchTypeEnum.BY_DATABASE]: 'By Database'
}

const KafkaVersionMap: Record<KafkaVersionEnum, string> = {
  [KafkaVersionEnum.VERSION_2XX]: 'Kafka v2',
  [KafkaVersionEnum.VERSION_3XX]: 'Kafka v3'
}

const KafkaCompressionMap: Record<KafkaCompressionTypeEnum, string> = {
  [KafkaCompressionTypeEnum.NONE]: 'None',
  [KafkaCompressionTypeEnum.GZIP]: 'Gzip',
  [KafkaCompressionTypeEnum.SNAPPY]: 'Snappy',
  [KafkaCompressionTypeEnum.LZ4]: 'Lz4',
  [KafkaCompressionTypeEnum.ZSTD]: 'Zstd'
}

export const KafkaPropertiesEssential = ({ data }: KafkaPropertiesEssentialProps) => {
  const kafka = data.sink?.kafka
  const networkType = kafka?.network?.network_type
  const brokerEndpoints = kafka?.network?.public_endpoints

  return (
    <>
      <PropertyCard.Divider />

      <PropertyCard.Item label="Connection Method">{networkType ? NetworkTypeMap[networkType] : '-'}</PropertyCard.Item>

      <PropertyCard.Item label="Kafka Version">
        {kafka?.broker?.kafka_version ? KafkaVersionMap[kafka.broker.kafka_version] : '-'}
      </PropertyCard.Item>

      <PropertyCard.Item label="Compress">
        {kafka?.broker?.compression ? KafkaCompressionMap[kafka.broker.compression] : '-'}
      </PropertyCard.Item>

      {networkType === KafkaNetworkTypeEnum.PUBLIC && (
        <PropertyCard.Item label="Broker Endpoints">{brokerEndpoints || '-'}</PropertyCard.Item>
      )}

      {networkType === KafkaNetworkTypeEnum.PRIVATE_LINK && kafka?.network?.private_link && (
        <>
          <PropertyCard.Item label="Private Link Connection ID">
            {kafka.network.private_link.private_link_connection_id || '-'}
          </PropertyCard.Item>
          {kafka.network.private_link.port != null && (
            <PropertyCard.Item label="Port">{kafka.network.private_link.port}</PropertyCard.Item>
          )}
        </>
      )}

      <PropertyCard.Item label="TLS Encryption">
        {kafka?.authentication?.enable_tls ? 'Enabled' : 'Disabled'}
      </PropertyCard.Item>

      <PropertyCard.Item label="Authentication">
        {kafka?.authentication?.auth_type ? AuthTypeOptionMap[kafka.authentication.auth_type] : '-'}
      </PropertyCard.Item>

      {kafka?.authentication?.username && (
        <PropertyCard.Item label="User Name">{kafka.authentication.username}</PropertyCard.Item>
      )}

      <PropertyCard.Divider />

      {/* Table Filter */}
      <PropertyCard.Item label="Table Filter">
        {data.filter?.filter_rule?.length ? (
          <Stack gap={4}>
            {data.filter.filter_rule.map((rule, idx) => (
              <Typography key={idx} variant="body-lg">
                {rule}
              </Typography>
            ))}
          </Stack>
        ) : (
          '-'
        )}
      </PropertyCard.Item>

      {/* Data Format */}
      <PropertyCard.Item label="Data Format">
        {kafka?.data_format?.protocol ? ProtocolMap[kafka.data_format.protocol] : '-'}
      </PropertyCard.Item>

      {/* Topic Distribution */}
      <PropertyCard.Item label="Topic Distribution Mode">
        {kafka?.topic_partition_config?.dispatch_type
          ? TopicDispatchTypeMap[kafka.topic_partition_config.dispatch_type]
          : '-'}
      </PropertyCard.Item>

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

      {kafka?.topic_partition_config?.dispatch_type !== TopicDispatchTypeEnum.ONE_TOPIC && (
        <>
          {kafka?.topic_partition_config?.topic_prefix && (
            <PropertyCard.Item label="Topic Prefix">{kafka.topic_partition_config.topic_prefix}</PropertyCard.Item>
          )}
          {kafka?.topic_partition_config?.separator && (
            <PropertyCard.Item label="Topic Separator">{kafka.topic_partition_config.separator}</PropertyCard.Item>
          )}
          {kafka?.topic_partition_config?.topic_suffix && (
            <PropertyCard.Item label="Topic Suffix">{kafka.topic_partition_config.topic_suffix}</PropertyCard.Item>
          )}
        </>
      )}

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

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

      {/* Start Position */}
      <PropertyCard.Item label="Start Position">{data.start_position?.mode || '-'}</PropertyCard.Item>
    </>
  )
}